Compare commits
5
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
1c37e62014 | ||
|
|
137426f7ad | ||
|
|
4a63578003 | ||
|
|
c7a444eb4b | ||
|
|
8cac50b2e7 |
@@ -7,6 +7,19 @@ from typing import Any
|
||||
|
||||
PROTECTED_BRANCHES = frozenset({"master", "main", "dev"})
|
||||
|
||||
# Evidence / preservation branches must never be removed by cleanup tools
|
||||
# (e.g. chore/issue-681-preserve-review-session-wip).
|
||||
_PRESERVATION_MARKERS = ("preserve", "preservation", "evidence")
|
||||
|
||||
|
||||
def is_preservation_or_evidence_branch(branch: str | None) -> bool:
|
||||
"""Return True when *branch* is a preservation/evidence ref that must stay."""
|
||||
if not branch:
|
||||
return False
|
||||
name = str(branch).lower()
|
||||
return any(marker in name for marker in _PRESERVATION_MARKERS)
|
||||
|
||||
|
||||
_RAW_BRANCH_DELETE_PATTERNS = (
|
||||
re.compile(r"\bgit(?:\s+-C\s+\S+)?\s+branch\s+-[dD]\b[^\n\r]*", re.I),
|
||||
re.compile(r"\bgit(?:\s+-C\s+\S+)?\s+push\b[^\n\r]*\s--delete\b[^\n\r]*", re.I),
|
||||
@@ -72,6 +85,11 @@ def assess_merged_pr_branch_cleanup(
|
||||
reasons.append("PR head branch is missing")
|
||||
if head_branch in protected:
|
||||
reasons.append(f"branch '{head_branch}' is protected")
|
||||
if is_preservation_or_evidence_branch(head_branch):
|
||||
reasons.append(
|
||||
f"branch '{head_branch}' is a preservation/evidence branch and "
|
||||
"cannot be deleted through merged-PR cleanup"
|
||||
)
|
||||
if head_branch in open_pr_heads:
|
||||
reasons.append("an open PR still references this head branch")
|
||||
if head_on_target is False:
|
||||
@@ -96,3 +114,490 @@ def assess_merged_pr_branch_cleanup(
|
||||
"block_reasons": reasons,
|
||||
"recommended_action": "delete_remote_branch" if safe else "keep_remote_branch",
|
||||
}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# #687 remediation: post-delete readback + active ownership protection
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
READBACK_NOT_FOUND = "not_found"
|
||||
READBACK_EXISTS = "exists"
|
||||
READBACK_AUTHENTICATION = "authentication_error"
|
||||
READBACK_AUTHORIZATION = "authorization_error"
|
||||
READBACK_TRANSPORT = "transport_error"
|
||||
READBACK_UNEXPECTED = "unexpected_response"
|
||||
READBACK_AMBIGUOUS_404 = "ambiguous_not_found"
|
||||
|
||||
# Scope of a 404: only branch-scoped absence may set verified_absent.
|
||||
NOT_FOUND_SCOPE_BRANCH = "branch"
|
||||
NOT_FOUND_SCOPE_REPOSITORY = "repository"
|
||||
NOT_FOUND_SCOPE_HOST = "host"
|
||||
NOT_FOUND_SCOPE_UNKNOWN = "unknown"
|
||||
|
||||
ERROR_CLASS_AUTHENTICATION = "authentication"
|
||||
ERROR_CLASS_AUTHORIZATION = "authorization"
|
||||
ERROR_CLASS_TRANSPORT = "transport"
|
||||
ERROR_CLASS_UNEXPECTED = "unexpected"
|
||||
|
||||
OWNERSHIP_CATEGORY_AUTHOR_SESSION = "author_session"
|
||||
OWNERSHIP_CATEGORY_AUTHOR_LEASE = "author_lease"
|
||||
OWNERSHIP_CATEGORY_REVIEWER_LEASE = "reviewer_lease"
|
||||
OWNERSHIP_CATEGORY_MERGER_LEASE = "merger_lease"
|
||||
OWNERSHIP_CATEGORY_CONTROLLER_LEASE = "controller_lease"
|
||||
OWNERSHIP_CATEGORY_RECONCILER_LEASE = "reconciler_lease"
|
||||
OWNERSHIP_CATEGORY_WORKTREE_BINDING = "worktree_binding"
|
||||
OWNERSHIP_CATEGORY_INVENTORY_ERROR = "ownership_inventory_error"
|
||||
|
||||
_ROLE_TO_OWNERSHIP_CATEGORY = {
|
||||
"author": OWNERSHIP_CATEGORY_AUTHOR_LEASE,
|
||||
"reviewer": OWNERSHIP_CATEGORY_REVIEWER_LEASE,
|
||||
"merger": OWNERSHIP_CATEGORY_MERGER_LEASE,
|
||||
"controller": OWNERSHIP_CATEGORY_CONTROLLER_LEASE,
|
||||
"reconciler": OWNERSHIP_CATEGORY_RECONCILER_LEASE,
|
||||
}
|
||||
|
||||
_ACTIVE_OWNERSHIP_STATUSES = frozenset(
|
||||
{"active", "live", "claimed", "in_progress", "working", "pushing", "pushed"}
|
||||
)
|
||||
_TERMINAL_OWNERSHIP_STATUSES = frozenset(
|
||||
{"released", "abandoned", "done", "blocked", "terminal", "closed"}
|
||||
)
|
||||
_EXPIRED_STATUSES = frozenset({"expired"})
|
||||
_STALE_STATUSES = frozenset({"stale", "stale_dead_process", "stale_missing_worktree"})
|
||||
|
||||
|
||||
def _norm_str(value: Any) -> str:
|
||||
return str(value or "").strip()
|
||||
|
||||
|
||||
def normalize_host(host: str | None) -> str:
|
||||
"""Normalize a host identity for ownership matching (no credentials)."""
|
||||
text = _norm_str(host).lower()
|
||||
for prefix in ("https://", "http://"):
|
||||
if text.startswith(prefix):
|
||||
text = text[len(prefix) :]
|
||||
# Drop path/query if a full URL slipped through.
|
||||
text = text.split("/", 1)[0]
|
||||
text = text.split("?", 1)[0]
|
||||
return text.rstrip(".")
|
||||
|
||||
|
||||
def ownership_category_for_role(role: str | None) -> str:
|
||||
"""Map a role kind to a non-secret ownership category label."""
|
||||
key = _norm_str(role).lower()
|
||||
return _ROLE_TO_OWNERSHIP_CATEGORY.get(key, f"{key or 'unknown'}_lease")
|
||||
|
||||
|
||||
def classify_branch_readback_http_status(
|
||||
status_code: int | None,
|
||||
*,
|
||||
not_found_scope: str | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Classify a GET-branch HTTP status into a secret-free readback result.
|
||||
|
||||
R1: A bare/generic/repository/wrong-host 404 never yields
|
||||
``verified_absent=True``. Only an authoritative *branch-scoped* not-found
|
||||
(``not_found_scope='branch'``) may verify deletion.
|
||||
"""
|
||||
if status_code == 404:
|
||||
scope = _norm_str(not_found_scope).lower() or NOT_FOUND_SCOPE_UNKNOWN
|
||||
if scope == NOT_FOUND_SCOPE_BRANCH:
|
||||
return {
|
||||
"status": READBACK_NOT_FOUND,
|
||||
"error_class": None,
|
||||
"verified_absent": True,
|
||||
"branch_present": False,
|
||||
"not_found_scope": NOT_FOUND_SCOPE_BRANCH,
|
||||
"reasons": [],
|
||||
}
|
||||
# repository / host / unknown / generic 404 — not verified absence
|
||||
reason = {
|
||||
NOT_FOUND_SCOPE_REPOSITORY: (
|
||||
"post-delete readback 404 is repository-scoped, not branch absence"
|
||||
),
|
||||
NOT_FOUND_SCOPE_HOST: (
|
||||
"post-delete readback 404 is host-scoped, not branch absence"
|
||||
),
|
||||
}.get(
|
||||
scope,
|
||||
"post-delete readback 404 is ambiguous (not branch-scoped); "
|
||||
"cannot verify absence",
|
||||
)
|
||||
return {
|
||||
"status": READBACK_AMBIGUOUS_404,
|
||||
"error_class": ERROR_CLASS_UNEXPECTED,
|
||||
"verified_absent": False,
|
||||
"branch_present": None,
|
||||
"not_found_scope": scope,
|
||||
"reasons": [reason],
|
||||
}
|
||||
if status_code in (401, 407):
|
||||
return {
|
||||
"status": READBACK_AUTHENTICATION,
|
||||
"error_class": ERROR_CLASS_AUTHENTICATION,
|
||||
"verified_absent": False,
|
||||
"branch_present": None,
|
||||
"reasons": ["post-delete branch readback authentication failed"],
|
||||
}
|
||||
if status_code == 403:
|
||||
return {
|
||||
"status": READBACK_AUTHORIZATION,
|
||||
"error_class": ERROR_CLASS_AUTHORIZATION,
|
||||
"verified_absent": False,
|
||||
"branch_present": None,
|
||||
"reasons": ["post-delete branch readback authorization failed"],
|
||||
}
|
||||
if status_code is not None and 200 <= int(status_code) < 300:
|
||||
return {
|
||||
"status": READBACK_EXISTS,
|
||||
"error_class": None,
|
||||
"verified_absent": False,
|
||||
"branch_present": True,
|
||||
"reasons": ["post-delete readback found branch still present"],
|
||||
}
|
||||
if status_code is not None and int(status_code) >= 500:
|
||||
return {
|
||||
"status": READBACK_TRANSPORT,
|
||||
"error_class": ERROR_CLASS_TRANSPORT,
|
||||
"verified_absent": False,
|
||||
"branch_present": None,
|
||||
"reasons": ["post-delete branch readback transport/upstream failure"],
|
||||
}
|
||||
return {
|
||||
"status": READBACK_UNEXPECTED,
|
||||
"error_class": ERROR_CLASS_UNEXPECTED,
|
||||
"verified_absent": False,
|
||||
"branch_present": None,
|
||||
"reasons": ["post-delete branch readback returned an unexpected response"],
|
||||
"http_status": status_code,
|
||||
}
|
||||
|
||||
|
||||
def _extract_http_status(exc: BaseException) -> int | None:
|
||||
"""Extract an HTTP status code from an exception chain (secret-free)."""
|
||||
seen: set[int] = set()
|
||||
current: BaseException | None = exc
|
||||
while current is not None and id(current) not in seen:
|
||||
seen.add(id(current))
|
||||
for attr in ("code", "status", "status_code"):
|
||||
value = getattr(current, attr, None)
|
||||
if isinstance(value, int) and 100 <= value <= 599:
|
||||
return value
|
||||
text = str(current) if current is not None else ""
|
||||
match = re.match(r"HTTP\s+(\d{3})\b", text)
|
||||
if match:
|
||||
return int(match.group(1))
|
||||
current = current.__cause__ or current.__context__
|
||||
return None
|
||||
|
||||
|
||||
def classify_branch_readback_exception(
|
||||
exc: BaseException,
|
||||
*,
|
||||
not_found_scope: str | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Classify a GET-branch exception without leaking credentials or bodies.
|
||||
|
||||
R1: substring ``404`` / ``not found`` alone never becomes verified_absent.
|
||||
Callers must pass ``not_found_scope='branch'`` only after authoritative
|
||||
proof that the repository/host is still reachable and the 404 is branch-level.
|
||||
"""
|
||||
status_code = _extract_http_status(exc)
|
||||
if status_code is None:
|
||||
lower = str(exc).lower() if exc is not None else ""
|
||||
if any(
|
||||
token in lower
|
||||
for token in (
|
||||
"timed out",
|
||||
"timeout",
|
||||
"connection",
|
||||
"network",
|
||||
"temporarily unavailable",
|
||||
"name or service not known",
|
||||
"nodename nor servname",
|
||||
)
|
||||
):
|
||||
return {
|
||||
"status": READBACK_TRANSPORT,
|
||||
"error_class": ERROR_CLASS_TRANSPORT,
|
||||
"verified_absent": False,
|
||||
"branch_present": None,
|
||||
"reasons": [
|
||||
"post-delete branch readback transport/upstream failure"
|
||||
],
|
||||
}
|
||||
if "unauthorized" in lower:
|
||||
status_code = 401
|
||||
elif "forbidden" in lower:
|
||||
status_code = 403
|
||||
elif "404" in lower or "not found" in lower:
|
||||
# Ambiguous: do NOT treat as branch absence without scope proof.
|
||||
status_code = 404
|
||||
if not_found_scope is None:
|
||||
not_found_scope = NOT_FOUND_SCOPE_UNKNOWN
|
||||
|
||||
if status_code is not None:
|
||||
return classify_branch_readback_http_status(
|
||||
status_code, not_found_scope=not_found_scope
|
||||
)
|
||||
|
||||
return {
|
||||
"status": READBACK_UNEXPECTED,
|
||||
"error_class": ERROR_CLASS_UNEXPECTED,
|
||||
"verified_absent": False,
|
||||
"branch_present": None,
|
||||
"reasons": ["post-delete branch readback returned an unexpected response"],
|
||||
}
|
||||
|
||||
|
||||
def assess_post_delete_readback(readback: dict[str, Any] | None) -> dict[str, Any]:
|
||||
"""Decide whether DELETE success may be reported after branch readback.
|
||||
|
||||
Success requires authoritative branch-scoped not-found
|
||||
(``verified_absent=True``). DELETE HTTP success alone is never enough.
|
||||
Generic/repository/host 404 cannot verify absence (R1).
|
||||
"""
|
||||
payload = dict(readback or {})
|
||||
status = _norm_str(payload.get("status")) or READBACK_UNEXPECTED
|
||||
# Only explicit verified_absent flag counts — never infer from status alone
|
||||
# when status is a bare not_found without branch scope proof.
|
||||
verified = bool(payload.get("verified_absent")) is True
|
||||
if verified and status == READBACK_NOT_FOUND:
|
||||
return {
|
||||
"ok": True,
|
||||
"success": True,
|
||||
"verified_absent": True,
|
||||
"readback": {
|
||||
"status": READBACK_NOT_FOUND,
|
||||
"verified_absent": True,
|
||||
"branch_present": False,
|
||||
"error_class": None,
|
||||
"not_found_scope": NOT_FOUND_SCOPE_BRANCH,
|
||||
},
|
||||
"reasons": [],
|
||||
}
|
||||
|
||||
reasons = list(payload.get("reasons") or [])
|
||||
if not reasons:
|
||||
if status == READBACK_EXISTS:
|
||||
reasons = ["post-delete readback found branch still present"]
|
||||
elif status == READBACK_AUTHENTICATION:
|
||||
reasons = ["post-delete branch readback authentication failed"]
|
||||
elif status == READBACK_AUTHORIZATION:
|
||||
reasons = ["post-delete branch readback authorization failed"]
|
||||
elif status == READBACK_TRANSPORT:
|
||||
reasons = ["post-delete branch readback transport/upstream failure"]
|
||||
elif status == READBACK_AMBIGUOUS_404:
|
||||
reasons = [
|
||||
"post-delete readback 404 is not branch-scoped; "
|
||||
"cannot verify absence"
|
||||
]
|
||||
else:
|
||||
reasons = ["post-delete branch readback could not verify deletion"]
|
||||
|
||||
return {
|
||||
"ok": False,
|
||||
"success": False,
|
||||
"verified_absent": False,
|
||||
"readback": {
|
||||
"status": status,
|
||||
"verified_absent": False,
|
||||
"branch_present": payload.get("branch_present"),
|
||||
"error_class": payload.get("error_class"),
|
||||
"not_found_scope": payload.get("not_found_scope"),
|
||||
},
|
||||
"reasons": reasons,
|
||||
"blocker_kind": "post_delete_readback_failed",
|
||||
}
|
||||
|
||||
|
||||
def cleanup_result_envelope(
|
||||
*,
|
||||
success: bool,
|
||||
performed: bool,
|
||||
delete_acknowledged: bool,
|
||||
verified_absent: bool,
|
||||
**extra: Any,
|
||||
) -> dict[str, Any]:
|
||||
"""R2: consistent top-level cleanup result fields on every return path."""
|
||||
out: dict[str, Any] = {
|
||||
"success": bool(success),
|
||||
"performed": bool(performed),
|
||||
"delete_acknowledged": bool(delete_acknowledged),
|
||||
"verified_absent": bool(verified_absent),
|
||||
}
|
||||
out.update(extra)
|
||||
return out
|
||||
|
||||
|
||||
def _repo_matches(
|
||||
record: dict[str, Any],
|
||||
*,
|
||||
remote: str,
|
||||
org: str,
|
||||
repo: str,
|
||||
host: str | None = None,
|
||||
) -> bool:
|
||||
"""Match ownership record to target remote/org/repo and normalized host."""
|
||||
if (
|
||||
_norm_str(record.get("remote")).lower() != _norm_str(remote).lower()
|
||||
or _norm_str(record.get("org")).lower() != _norm_str(org).lower()
|
||||
or _norm_str(record.get("repo")).lower() != _norm_str(repo).lower()
|
||||
):
|
||||
return False
|
||||
expected_host = normalize_host(host)
|
||||
record_host = normalize_host(record.get("host") or record.get("host_name"))
|
||||
# When both sides declare a host, they must agree after normalization.
|
||||
# A record host that disagrees with the expected host is out of scope.
|
||||
# Legacy records without host still match when remote/org/repo agree.
|
||||
if expected_host and record_host and expected_host != record_host:
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def _branch_matches(record: dict[str, Any], branch: str) -> bool:
|
||||
return _norm_str(record.get("branch")) == _norm_str(branch)
|
||||
|
||||
|
||||
def assess_ownership_record_activity(record: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Classify one ownership record as blocking or non-blocking.
|
||||
|
||||
Distinguishes active ownership from expired/released/terminal/stale records.
|
||||
Expired/stale records block unless reclaim_allowed is *explicitly* True
|
||||
(O2: never treat missing/unknown reclaim as auto-allowed).
|
||||
"""
|
||||
status = _norm_str(record.get("status")).lower()
|
||||
category = _norm_str(record.get("category")) or "unknown"
|
||||
reclaim_allowed = record.get("reclaim_allowed")
|
||||
|
||||
if category == OWNERSHIP_CATEGORY_INVENTORY_ERROR:
|
||||
return {
|
||||
"blocks": True,
|
||||
"status": status or "unknown",
|
||||
"category": category,
|
||||
"reason": "ownership inventory failed closed",
|
||||
}
|
||||
|
||||
if status in _TERMINAL_OWNERSHIP_STATUSES:
|
||||
return {
|
||||
"blocks": False,
|
||||
"status": status,
|
||||
"category": category,
|
||||
"reason": f"{category} ownership is terminal/released ({status})",
|
||||
}
|
||||
if status in _ACTIVE_OWNERSHIP_STATUSES:
|
||||
return {
|
||||
"blocks": True,
|
||||
"status": status,
|
||||
"category": category,
|
||||
"reason": f"active {category} ownership still uses the target branch",
|
||||
}
|
||||
if status in _EXPIRED_STATUSES or status in _STALE_STATUSES:
|
||||
# O2: only explicit reclaim_allowed=True skips the block.
|
||||
if reclaim_allowed is True:
|
||||
return {
|
||||
"blocks": False,
|
||||
"status": status,
|
||||
"category": category,
|
||||
"reason": (
|
||||
f"{category} ownership is {status} and reclaimable; "
|
||||
"does not block deletion"
|
||||
),
|
||||
}
|
||||
return {
|
||||
"blocks": True,
|
||||
"status": status,
|
||||
"category": category,
|
||||
"reason": (
|
||||
f"{status} {category} ownership still protects the target "
|
||||
"branch (reclaim not proven; fail closed)"
|
||||
),
|
||||
}
|
||||
# Unknown status → fail closed
|
||||
return {
|
||||
"blocks": True,
|
||||
"status": status or "unknown",
|
||||
"category": category,
|
||||
"reason": (
|
||||
f"unclassified {category} ownership status "
|
||||
f"'{status or 'unknown'}'; fail closed"
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
def assess_active_branch_ownership(
|
||||
*,
|
||||
remote: str,
|
||||
org: str,
|
||||
repo: str,
|
||||
branch: str,
|
||||
host: str | None = None,
|
||||
records: list[dict[str, Any]] | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Assess whether any active ownership still uses *branch* in *repo*.
|
||||
|
||||
Matching requires remote/org/repo/branch and, when provided, normalized
|
||||
host identity. Other repositories, hosts, or branches never false-block.
|
||||
"""
|
||||
target_branch = _norm_str(branch)
|
||||
expected_host = normalize_host(host)
|
||||
considered: list[dict[str, Any]] = []
|
||||
blocking: list[dict[str, Any]] = []
|
||||
ignored: list[dict[str, Any]] = []
|
||||
|
||||
for raw in records or []:
|
||||
if not isinstance(raw, dict):
|
||||
continue
|
||||
if not _repo_matches(
|
||||
raw, remote=remote, org=org, repo=repo, host=expected_host or None
|
||||
):
|
||||
ignored.append(
|
||||
{
|
||||
"category": _norm_str(raw.get("category")) or "unknown",
|
||||
"reason": "different repository or host scope",
|
||||
}
|
||||
)
|
||||
continue
|
||||
if not _branch_matches(raw, target_branch):
|
||||
ignored.append(
|
||||
{
|
||||
"category": _norm_str(raw.get("category")) or "unknown",
|
||||
"reason": "different branch",
|
||||
}
|
||||
)
|
||||
continue
|
||||
activity = assess_ownership_record_activity(raw)
|
||||
entry = {
|
||||
"category": activity["category"],
|
||||
"status": activity["status"],
|
||||
"blocks": activity["blocks"],
|
||||
"reason": activity["reason"],
|
||||
}
|
||||
considered.append(entry)
|
||||
if activity["blocks"]:
|
||||
blocking.append(entry)
|
||||
|
||||
block = bool(blocking)
|
||||
categories = sorted({b["category"] for b in blocking})
|
||||
reasons = [
|
||||
(
|
||||
"active ownership protects branch "
|
||||
f"'{target_branch}': " + "; ".join(b["reason"] for b in blocking)
|
||||
)
|
||||
] if block else []
|
||||
return {
|
||||
"block": block,
|
||||
"safe_to_delete": not block,
|
||||
"remote": remote,
|
||||
"org": org,
|
||||
"repo": repo,
|
||||
"host": expected_host or None,
|
||||
"branch": target_branch,
|
||||
"blocking_categories": categories,
|
||||
"blocking": blocking,
|
||||
"considered": considered,
|
||||
"ignored_out_of_scope": ignored,
|
||||
"reasons": reasons,
|
||||
"blocker_kind": "active_branch_ownership" if block else None,
|
||||
"recommended_action": "keep_remote_branch" if block else "delete_remote_branch",
|
||||
}
|
||||
|
||||
@@ -238,11 +238,43 @@ narrow operation set:
|
||||
- `gitea.issue.comment`
|
||||
- `gitea.issue.close`
|
||||
- `gitea.pr.close`
|
||||
- `gitea.branch.delete` (merged-branch cleanup only — see below)
|
||||
|
||||
Forbidden on reconciler profiles: `gitea.pr.approve`, `gitea.pr.merge`,
|
||||
`gitea.pr.review`, `gitea.pr.create`, `gitea.branch.push`, and
|
||||
`gitea.repo.commit`.
|
||||
|
||||
### Merged-branch cleanup ownership (`gitea.branch.delete`)
|
||||
|
||||
The reconciler is the repository-supported owner of merged-PR source-branch
|
||||
cleanup: `task_capability_map` maps `cleanup_merged_pr_branch` (and
|
||||
`reconciliation_cleanup`) to role `reconciler` with permission
|
||||
`gitea.branch.delete`. Post-merge branch lifecycle is reconciliation work —
|
||||
it happens after the author, reviewer, and merger roles have completed, and
|
||||
it must not be reachable from those roles.
|
||||
|
||||
Least-privilege constraints:
|
||||
|
||||
- `gitea.branch.delete` is granted **only** to reconciler profiles. Author,
|
||||
reviewer, and merger profiles must never hold it; `gitea_delete_branch`
|
||||
and `gitea_cleanup_merged_pr_branch` fail closed on any profile without
|
||||
the permission.
|
||||
- Even with the permission, reconciler deletion is only supported through the
|
||||
guarded `gitea_cleanup_merged_pr_branch` path (#514 / #687): the PR must be
|
||||
merged, the head an ancestor of the target, the branch not protected
|
||||
(`master`/`main`/`dev`), the branch not a preservation/evidence ref (e.g.
|
||||
`chore/issue-681-preserve-review-session-wip`), no open PR may still use the
|
||||
head, and an explicit `CLEANUP MERGED PR <n> BRANCH <branch>` confirmation is
|
||||
required. Raw `gitea_delete_branch` is **denied** to reconciler even when
|
||||
`gitea.branch.delete` is present.
|
||||
- Raw `git branch -d` / `git push --delete` cleanup remains blocked by
|
||||
`branch_cleanup_guard` and the final-report validator regardless of
|
||||
profile permissions.
|
||||
- `gitea.branch.delete` has no short alias in `GITEA_OPERATION_ALIASES`;
|
||||
write it fully qualified in `allowed_operations`. Migration must emit
|
||||
canonical names such as `gitea.pr.close` (never bare `pr.close` /
|
||||
`issue.close`, which the production normalizer rejects or drops).
|
||||
|
||||
Launch a static `gitea-reconciler` MCP namespace with
|
||||
`GITEA_MCP_PROFILE=prgs-reconciler`. Profile shape is validated by
|
||||
`reconciler_profile.assess_reconciler_profile` (#304). Use the
|
||||
@@ -251,6 +283,159 @@ Launch a static `gitea-reconciler` MCP namespace with
|
||||
fresh target-branch fetch, recorded target SHA, and ancestor proof. PRs whose
|
||||
heads are not already landed cannot be closed through this path.
|
||||
|
||||
### Operational runbook: grant reconciler `gitea.branch.delete` (#687)
|
||||
|
||||
Merging a code PR that updates `migrate_profiles.py` / `reconciler_profile.py`
|
||||
**does not** change the live operator profile on disk. Apply the profile
|
||||
change deliberately, then reconnect the client-managed namespace.
|
||||
|
||||
1. **Approved migration / profile-update command** (from the repo root, using
|
||||
the project venv if present):
|
||||
|
||||
```bash
|
||||
# Dry-run first (default): validates v2 output, writes nothing
|
||||
python3 migrate_profiles.py -i ~/.config/gitea-tools/profiles.json
|
||||
|
||||
# Apply: creates backup then writes migrated v2 config
|
||||
python3 migrate_profiles.py -i ~/.config/gitea-tools/profiles.json -w
|
||||
# Optional explicit paths:
|
||||
# python3 migrate_profiles.py -i ~/.config/gitea-tools/profiles.json \
|
||||
# -o ~/.config/gitea-tools/profiles.json \
|
||||
# --backup ~/.config/gitea-tools/profiles.json.bak -w
|
||||
```
|
||||
|
||||
If the live file is already v2, edit the reconciler identity’s
|
||||
`allowed_operations` / `forbidden_operations` under
|
||||
`environments.<env>.services.gitea.identities.reconciler` (or the
|
||||
`prgs-reconciler` alias target) so allowed includes the canonical set
|
||||
below — then re-validate with a load of the config (see step 3).
|
||||
|
||||
2. **Inspect the generated (or edited) profile** — confirm the reconciler
|
||||
identity, for example:
|
||||
|
||||
```bash
|
||||
python3 - <<'PY'
|
||||
import json
|
||||
from pathlib import Path
|
||||
cfg = json.loads(Path.home().joinpath(".config/gitea-tools/profiles.json").read_text())
|
||||
# v2 environments shape:
|
||||
ident = cfg["environments"]["prgs"]["services"]["gitea"]["identities"]["reconciler"]
|
||||
print("role:", ident.get("role"))
|
||||
print("allowed:", ident.get("allowed_operations"))
|
||||
print("forbidden:", ident.get("forbidden_operations"))
|
||||
PY
|
||||
```
|
||||
|
||||
3. **Validate canonical operation names and least privilege**
|
||||
|
||||
Expected canonical **allowed** (defaults after migration):
|
||||
|
||||
- `gitea.read`
|
||||
- `gitea.pr.close` (required)
|
||||
- `gitea.pr.comment`
|
||||
- `gitea.issue.comment`
|
||||
- `gitea.issue.close`
|
||||
- `gitea.branch.delete` (recommended; cleanup only)
|
||||
|
||||
Expected **forbidden** includes at least: `gitea.pr.approve`,
|
||||
`gitea.pr.merge`, `gitea.pr.review`, `gitea.pr.create`,
|
||||
`gitea.branch.push`, `gitea.repo.commit`.
|
||||
|
||||
No shorthand (`pr.close`, `issue.close`, `pr.comment`) may remain.
|
||||
Validate with the production loader:
|
||||
|
||||
```bash
|
||||
python3 - <<'PY'
|
||||
import gitea_config, reconciler_profile
|
||||
from pathlib import Path
|
||||
path = str(Path.home() / ".config/gitea-tools/profiles.json")
|
||||
gitea_config.load_config(path) # fails closed on invalid config
|
||||
# Or assess the reconciler lists directly after extracting them:
|
||||
# print(reconciler_profile.assess_reconciler_profile(allowed, forbidden))
|
||||
PY
|
||||
```
|
||||
|
||||
4. **Merging PR #688 (or any code PR) does not update the live profile.**
|
||||
Code changes only the migration helper, schema, docs, and tests. The
|
||||
operator must still run `migrate_profiles.py -w` or an equivalent
|
||||
authorized edit of `~/.config/gitea-tools/profiles.json`.
|
||||
|
||||
5. **Supported apply method:** `python3 migrate_profiles.py … -w` (backup
|
||||
created automatically) **or** operator-authorized edit of the live
|
||||
profiles file after backup. Unsupported: silent mtime tricks, manual
|
||||
process kill to “reload”, or undocumented env overrides.
|
||||
|
||||
6. **Backup and validation:** `-w` copies the input to
|
||||
`<input_path>.bak` (or `--backup PATH`) before writing. Re-run
|
||||
`load_config` / `assess_reconciler_profile` after write. Keep the
|
||||
`.bak` until live whoami/capability checks pass.
|
||||
|
||||
7. **Client-managed namespace reconnect/reload:** reconnect or reload the
|
||||
IDE MCP client so `gitea-reconciler` restarts from current `master` and
|
||||
the updated `GITEA_MCP_PROFILE=prgs-reconciler` config. Do not hand-launch
|
||||
`mcp_server.py` / `gitea_mcp_server.py` with ad hoc `GITEA_*` env
|
||||
(see #686 / #630).
|
||||
|
||||
8. **Live reverification** (through the client-managed `gitea-reconciler`
|
||||
namespace only):
|
||||
|
||||
- `gitea_whoami` → identity + profile `prgs-reconciler`
|
||||
- `gitea_assess_master_parity` → `stale=false`, `restart_required=false`
|
||||
- `gitea_resolve_task_capability(task="cleanup_merged_pr_branch")` →
|
||||
`allowed_in_current_session=true` only when permission and role match
|
||||
- `gitea_resolve_task_capability(task="delete_branch")` →
|
||||
**not** allowed for reconciler (role denial must be enforced)
|
||||
|
||||
9. **Guarded cleanup usage** (example for a merged PR whose source branch
|
||||
remains on the remote):
|
||||
|
||||
```text
|
||||
gitea_cleanup_merged_pr_branch(
|
||||
pr_number=<N>,
|
||||
branch=<exact PR head branch>,
|
||||
confirmation="CLEANUP MERGED PR <N> BRANCH <exact PR head branch>",
|
||||
remote="prgs",
|
||||
org="Scaled-Tech-Consulting",
|
||||
repo="Gitea-Tools",
|
||||
worktree_path="<path under branches/>",
|
||||
)
|
||||
```
|
||||
|
||||
The tool refuses unmerged PRs, protected branches, preservation/evidence
|
||||
branches, open-PR heads, mismatched branch names, and wrong confirmation.
|
||||
|
||||
10. **Prohibitions**
|
||||
|
||||
- No raw `git push --delete`, `git branch -d` / `-D`, or delete refspecs
|
||||
- No arbitrary `gitea_delete_branch` from reconciler
|
||||
- No unsupported profile switching mid-run without full re-preflight
|
||||
- No ad hoc hand-edits of live profiles **unless** operator-authorized,
|
||||
backed up, and revalidated as above
|
||||
|
||||
Canonical migrated reconciler example:
|
||||
|
||||
```json
|
||||
{
|
||||
"role": "reconciler",
|
||||
"allowed_operations": [
|
||||
"gitea.read",
|
||||
"gitea.pr.close",
|
||||
"gitea.pr.comment",
|
||||
"gitea.issue.comment",
|
||||
"gitea.issue.close",
|
||||
"gitea.branch.delete"
|
||||
],
|
||||
"forbidden_operations": [
|
||||
"gitea.pr.approve",
|
||||
"gitea.pr.merge",
|
||||
"gitea.pr.review",
|
||||
"gitea.pr.create",
|
||||
"gitea.branch.push",
|
||||
"gitea.repo.commit"
|
||||
]
|
||||
}
|
||||
```
|
||||
|
||||
## Identity and fail-closed rules
|
||||
|
||||
Before **any** mutating action, a workflow must know both:
|
||||
|
||||
+654
-62
@@ -5925,6 +5925,65 @@ def gitea_delete_branch(
|
||||
"permission_report": _permission_block_report("gitea.branch.delete"),
|
||||
}
|
||||
|
||||
# Possessing gitea.branch.delete alone is not enough for arbitrary deletion.
|
||||
# task_capability_map maps delete_branch → author; reconciler must use the
|
||||
# guarded cleanup_merged_pr_branch path only (#687 / #514).
|
||||
profile = get_profile()
|
||||
active_role = _profile_role_kind(profile)
|
||||
required_role = task_capability_map.required_role("delete_branch")
|
||||
if active_role == "reconciler":
|
||||
return {
|
||||
"success": False,
|
||||
"performed": False,
|
||||
"required_permission": "gitea.branch.delete",
|
||||
"required_role_kind": required_role,
|
||||
"active_role_kind": active_role,
|
||||
"reasons": [
|
||||
"reconciler profile cannot use raw gitea_delete_branch; "
|
||||
"use gitea_cleanup_merged_pr_branch for a fully merged PR "
|
||||
"source branch only (fail closed)"
|
||||
],
|
||||
"exact_next_action": (
|
||||
"Call gitea_cleanup_merged_pr_branch with pr_number, the "
|
||||
"exact PR head branch, and confirmation "
|
||||
"'CLEANUP MERGED PR <n> BRANCH <branch>' after capability "
|
||||
"resolve for cleanup_merged_pr_branch."
|
||||
),
|
||||
"permission_report": _permission_block_report("gitea.branch.delete"),
|
||||
}
|
||||
if active_role != required_role:
|
||||
return {
|
||||
"success": False,
|
||||
"performed": False,
|
||||
"required_permission": "gitea.branch.delete",
|
||||
"required_role_kind": required_role,
|
||||
"active_role_kind": active_role,
|
||||
"reasons": [
|
||||
f"Active profile role '{active_role}' cannot perform "
|
||||
f"{required_role} task 'delete_branch' even when "
|
||||
"gitea.branch.delete is present (fail closed)"
|
||||
],
|
||||
"permission_report": _permission_block_report("gitea.branch.delete"),
|
||||
}
|
||||
|
||||
if branch_cleanup_guard.is_preservation_or_evidence_branch(branch):
|
||||
return {
|
||||
"success": False,
|
||||
"performed": False,
|
||||
"required_permission": "gitea.branch.delete",
|
||||
"reasons": [
|
||||
f"branch '{branch}' is a preservation/evidence branch and "
|
||||
"cannot be deleted (fail closed)"
|
||||
],
|
||||
}
|
||||
if branch in branch_cleanup_guard.PROTECTED_BRANCHES:
|
||||
return {
|
||||
"success": False,
|
||||
"performed": False,
|
||||
"required_permission": "gitea.branch.delete",
|
||||
"reasons": [f"branch '{branch}' is protected (fail closed)"],
|
||||
}
|
||||
|
||||
audit_allowed, audit_reasons = (
|
||||
audit_reconciliation_mode.check_audit_mutation_allowed("delete_branch")
|
||||
)
|
||||
@@ -5967,41 +6026,49 @@ def gitea_cleanup_merged_pr_branch(
|
||||
"""Delete a merged PR source branch through the guarded MCP path (#514)."""
|
||||
gate_reasons = _profile_operation_gate("gitea.branch.delete")
|
||||
if gate_reasons:
|
||||
return {
|
||||
"success": False,
|
||||
"performed": False,
|
||||
"required_permission": "gitea.branch.delete",
|
||||
"reasons": gate_reasons,
|
||||
"permission_report": _permission_block_report("gitea.branch.delete"),
|
||||
}
|
||||
return branch_cleanup_guard.cleanup_result_envelope(
|
||||
success=False,
|
||||
performed=False,
|
||||
delete_acknowledged=False,
|
||||
verified_absent=False,
|
||||
required_permission="gitea.branch.delete",
|
||||
reasons=gate_reasons,
|
||||
permission_report=_permission_block_report("gitea.branch.delete"),
|
||||
)
|
||||
|
||||
profile = get_profile()
|
||||
active_role = _role_kind(
|
||||
profile.get("allowed_operations", []),
|
||||
profile.get("forbidden_operations", []),
|
||||
)
|
||||
if active_role == "reviewer":
|
||||
return {
|
||||
"success": False,
|
||||
"performed": False,
|
||||
"required_permission": "gitea.branch.delete",
|
||||
"reasons": [
|
||||
"reviewer profile is not authorized for merged branch cleanup "
|
||||
"(fail closed)"
|
||||
active_role = _profile_role_kind(profile)
|
||||
# cleanup_merged_pr_branch is reconciler-owned (task_capability_map).
|
||||
# Author/reviewer/merger must not reach this path even if they somehow
|
||||
# hold gitea.branch.delete.
|
||||
if active_role != "reconciler":
|
||||
return branch_cleanup_guard.cleanup_result_envelope(
|
||||
success=False,
|
||||
performed=False,
|
||||
delete_acknowledged=False,
|
||||
verified_absent=False,
|
||||
required_permission="gitea.branch.delete",
|
||||
required_role_kind="reconciler",
|
||||
active_role_kind=active_role,
|
||||
reasons=[
|
||||
f"profile role '{active_role}' is not authorized for merged "
|
||||
"branch cleanup; required role is reconciler (fail closed)"
|
||||
],
|
||||
"permission_report": _permission_block_report("gitea.branch.delete"),
|
||||
}
|
||||
permission_report=_permission_block_report("gitea.branch.delete"),
|
||||
)
|
||||
|
||||
if worktree_path is None or "/branches/" not in os.path.realpath(worktree_path):
|
||||
return {
|
||||
"success": False,
|
||||
"performed": False,
|
||||
"required_permission": "gitea.branch.delete",
|
||||
"reasons": [
|
||||
return branch_cleanup_guard.cleanup_result_envelope(
|
||||
success=False,
|
||||
performed=False,
|
||||
delete_acknowledged=False,
|
||||
verified_absent=False,
|
||||
required_permission="gitea.branch.delete",
|
||||
reasons=[
|
||||
"merged branch cleanup requires an explicit branches/ worktree "
|
||||
"path; root checkout branch ref mutation is blocked (fail closed)"
|
||||
],
|
||||
}
|
||||
)
|
||||
|
||||
verify_preflight_purity(
|
||||
remote,
|
||||
@@ -6019,16 +6086,18 @@ def gitea_cleanup_merged_pr_branch(
|
||||
target_branch = (pr.get("base") or {}).get("ref") or "master"
|
||||
|
||||
if branch and branch != pr_head.get("ref"):
|
||||
return {
|
||||
"success": False,
|
||||
"performed": False,
|
||||
"pr_number": pr_number,
|
||||
"branch": branch,
|
||||
"reasons": [
|
||||
return branch_cleanup_guard.cleanup_result_envelope(
|
||||
success=False,
|
||||
performed=False,
|
||||
delete_acknowledged=False,
|
||||
verified_absent=False,
|
||||
pr_number=pr_number,
|
||||
branch=branch,
|
||||
reasons=[
|
||||
f"requested branch '{branch}' does not match PR head "
|
||||
f"'{pr_head.get('ref')}'"
|
||||
],
|
||||
}
|
||||
)
|
||||
|
||||
open_prs = api_get_all(f"{base}/pulls?state=open", auth)
|
||||
open_heads = {
|
||||
@@ -6058,19 +6127,86 @@ def gitea_cleanup_merged_pr_branch(
|
||||
confirmation=confirmation,
|
||||
)
|
||||
if not assessment["safe_to_delete"]:
|
||||
return {
|
||||
"success": False,
|
||||
"performed": False,
|
||||
"pr_number": pr_number,
|
||||
"branch": head_branch,
|
||||
"assessment": assessment,
|
||||
"reasons": assessment["block_reasons"],
|
||||
}
|
||||
return branch_cleanup_guard.cleanup_result_envelope(
|
||||
success=False,
|
||||
performed=False,
|
||||
delete_acknowledged=False,
|
||||
verified_absent=False,
|
||||
pr_number=pr_number,
|
||||
branch=head_branch,
|
||||
assessment=assessment,
|
||||
reasons=assessment["block_reasons"],
|
||||
)
|
||||
|
||||
# #687: active ownership protection (sessions/leases/worktree bindings).
|
||||
# Do not rely only on the caller worktree living under branches/.
|
||||
ownership_bundle = _collect_branch_ownership_records(
|
||||
remote=remote,
|
||||
host=h,
|
||||
org=o,
|
||||
repo=r,
|
||||
branch=head_branch,
|
||||
pr_number=pr_number,
|
||||
project_root=PROJECT_ROOT,
|
||||
auth=auth,
|
||||
base_api=base,
|
||||
)
|
||||
ownership_records = ownership_bundle.get("records") or []
|
||||
if ownership_bundle.get("inventory_error"):
|
||||
# O1: control-plane / ownership inventory failure fails closed.
|
||||
ownership_records = list(ownership_records) + [
|
||||
{
|
||||
"category": branch_cleanup_guard.OWNERSHIP_CATEGORY_INVENTORY_ERROR,
|
||||
"status": "unknown",
|
||||
"remote": remote,
|
||||
"host": h,
|
||||
"org": o,
|
||||
"repo": r,
|
||||
"branch": head_branch,
|
||||
"reclaim_allowed": False,
|
||||
"role": "inventory",
|
||||
}
|
||||
]
|
||||
ownership = branch_cleanup_guard.assess_active_branch_ownership(
|
||||
remote=remote,
|
||||
org=o,
|
||||
repo=r,
|
||||
branch=head_branch,
|
||||
host=h,
|
||||
records=ownership_records,
|
||||
)
|
||||
if ownership.get("block"):
|
||||
return branch_cleanup_guard.cleanup_result_envelope(
|
||||
success=False,
|
||||
performed=False,
|
||||
delete_acknowledged=False,
|
||||
verified_absent=False,
|
||||
pr_number=pr_number,
|
||||
branch=head_branch,
|
||||
assessment=assessment,
|
||||
ownership={
|
||||
"block": True,
|
||||
"blocking_categories": ownership.get("blocking_categories") or [],
|
||||
"reasons": ownership.get("reasons") or [],
|
||||
"blocker_kind": ownership.get("blocker_kind"),
|
||||
"checked": True,
|
||||
},
|
||||
reasons=ownership.get("reasons")
|
||||
or ["active ownership protects the target branch"],
|
||||
blocker_kind="active_branch_ownership",
|
||||
)
|
||||
|
||||
import urllib.parse
|
||||
|
||||
encoded_branch = urllib.parse.quote(head_branch, safe="")
|
||||
url = f"{base}/branches/{encoded_branch}"
|
||||
request_metadata = {
|
||||
"branch": head_branch,
|
||||
"required_permission": "gitea.branch.delete",
|
||||
"cleanup_path": "gitea_cleanup_merged_pr_branch",
|
||||
"ownership_checked": True,
|
||||
"ownership_blocking_categories": [],
|
||||
}
|
||||
with _audited(
|
||||
"cleanup_merged_pr_branch",
|
||||
host=h,
|
||||
@@ -6079,36 +6215,416 @@ def gitea_cleanup_merged_pr_branch(
|
||||
repo=r,
|
||||
pr_number=pr_number,
|
||||
target_branch=head_branch,
|
||||
request_metadata={
|
||||
"branch": head_branch,
|
||||
"required_permission": "gitea.branch.delete",
|
||||
"cleanup_path": "gitea_cleanup_merged_pr_branch",
|
||||
},
|
||||
request_metadata=request_metadata,
|
||||
):
|
||||
api_request("DELETE", url, auth)
|
||||
return {
|
||||
"success": True,
|
||||
"performed": True,
|
||||
"pr_number": pr_number,
|
||||
"branch": head_branch,
|
||||
"message": f"Merged PR #{pr_number} source branch '{head_branch}' deleted.",
|
||||
"assessment": assessment,
|
||||
|
||||
# #687: authoritative post-delete readback. DELETE success alone is not
|
||||
# enough — only branch-scoped not-found proves deletion (R1).
|
||||
readback = _probe_remote_branch(h, o, r, auth, head_branch)
|
||||
readback_assessment = branch_cleanup_guard.assess_post_delete_readback(readback)
|
||||
verified_absent = bool(readback_assessment.get("verified_absent"))
|
||||
request_metadata["post_delete_readback"] = {
|
||||
"status": (readback_assessment.get("readback") or {}).get("status"),
|
||||
"verified_absent": verified_absent,
|
||||
"error_class": (readback_assessment.get("readback") or {}).get(
|
||||
"error_class"
|
||||
),
|
||||
"not_found_scope": (readback_assessment.get("readback") or {}).get(
|
||||
"not_found_scope"
|
||||
),
|
||||
}
|
||||
if gitea_audit.audit_enabled():
|
||||
_audit(
|
||||
"cleanup_merged_pr_branch_readback",
|
||||
host=h,
|
||||
remote=remote,
|
||||
org=o,
|
||||
repo=r,
|
||||
result=(
|
||||
gitea_audit.SUCCEEDED
|
||||
if readback_assessment.get("ok")
|
||||
else gitea_audit.FAILED
|
||||
),
|
||||
reason="; ".join(readback_assessment.get("reasons") or []) or None,
|
||||
request_metadata=request_metadata,
|
||||
pr_number=pr_number,
|
||||
target_branch=head_branch,
|
||||
)
|
||||
|
||||
if not readback_assessment.get("ok"):
|
||||
return branch_cleanup_guard.cleanup_result_envelope(
|
||||
success=False,
|
||||
performed=True,
|
||||
delete_acknowledged=True,
|
||||
verified_absent=False,
|
||||
pr_number=pr_number,
|
||||
branch=head_branch,
|
||||
assessment=assessment,
|
||||
ownership={
|
||||
"block": False,
|
||||
"blocking_categories": [],
|
||||
"checked": True,
|
||||
},
|
||||
readback=readback_assessment.get("readback"),
|
||||
reasons=readback_assessment.get("reasons")
|
||||
or ["post-delete branch readback could not verify deletion"],
|
||||
blocker_kind=readback_assessment.get("blocker_kind")
|
||||
or "post_delete_readback_failed",
|
||||
message=(
|
||||
f"DELETE accepted for '{head_branch}' but post-delete readback "
|
||||
"did not verify absence"
|
||||
),
|
||||
)
|
||||
|
||||
return branch_cleanup_guard.cleanup_result_envelope(
|
||||
success=True,
|
||||
performed=True,
|
||||
delete_acknowledged=True,
|
||||
verified_absent=True,
|
||||
pr_number=pr_number,
|
||||
branch=head_branch,
|
||||
message=(
|
||||
f"Merged PR #{pr_number} source branch '{head_branch}' deleted "
|
||||
"and verified absent via branch-scoped post-delete readback."
|
||||
),
|
||||
assessment=assessment,
|
||||
ownership={
|
||||
"block": False,
|
||||
"blocking_categories": [],
|
||||
"checked": True,
|
||||
},
|
||||
readback=readback_assessment.get("readback"),
|
||||
)
|
||||
|
||||
|
||||
def _remote_branch_exists(h: str, o: str, r: str, auth: str, branch: str) -> bool:
|
||||
def _probe_remote_branch(
|
||||
h: str, o: str, r: str, auth: str, branch: str
|
||||
) -> dict:
|
||||
"""GET a remote branch and return a secret-free structured readback.
|
||||
|
||||
R1: On HTTP 404, re-probe the repository endpoint. Only when the repo is
|
||||
still reachable is the 404 treated as branch-scoped absence. Generic,
|
||||
repository-scoped, or host-level 404 never sets verified_absent.
|
||||
"""
|
||||
import urllib.parse
|
||||
|
||||
encoded = urllib.parse.quote(branch, safe="")
|
||||
url = f"{repo_api_url(h, o, r)}/branches/{encoded}"
|
||||
try:
|
||||
api_request("GET", url, auth)
|
||||
return True
|
||||
return branch_cleanup_guard.classify_branch_readback_http_status(200)
|
||||
except Exception as exc:
|
||||
message = str(exc).lower()
|
||||
if "404" in message or "not found" in message:
|
||||
return False
|
||||
raise
|
||||
status = branch_cleanup_guard._extract_http_status(exc)
|
||||
if status != 404:
|
||||
return branch_cleanup_guard.classify_branch_readback_exception(exc)
|
||||
|
||||
# Distinguish branch vs repository/host 404 via repo reachability.
|
||||
repo_url = repo_api_url(h, o, r)
|
||||
try:
|
||||
api_request("GET", repo_url, auth)
|
||||
# Repo reachable → branch-scoped not-found.
|
||||
return branch_cleanup_guard.classify_branch_readback_http_status(
|
||||
404,
|
||||
not_found_scope=branch_cleanup_guard.NOT_FOUND_SCOPE_BRANCH,
|
||||
)
|
||||
except Exception as repo_exc:
|
||||
repo_status = branch_cleanup_guard._extract_http_status(repo_exc)
|
||||
if repo_status == 404:
|
||||
return branch_cleanup_guard.classify_branch_readback_http_status(
|
||||
404,
|
||||
not_found_scope=branch_cleanup_guard.NOT_FOUND_SCOPE_REPOSITORY,
|
||||
)
|
||||
if repo_status in (401, 407):
|
||||
return branch_cleanup_guard.classify_branch_readback_http_status(
|
||||
401
|
||||
)
|
||||
if repo_status == 403:
|
||||
return branch_cleanup_guard.classify_branch_readback_http_status(
|
||||
403
|
||||
)
|
||||
# Host/transport/unknown: not branch-verified.
|
||||
return branch_cleanup_guard.classify_branch_readback_http_status(
|
||||
404,
|
||||
not_found_scope=branch_cleanup_guard.NOT_FOUND_SCOPE_UNKNOWN,
|
||||
)
|
||||
|
||||
|
||||
def _remote_branch_exists(h: str, o: str, r: str, auth: str, branch: str) -> bool:
|
||||
"""True when the remote branch exists; False on authoritative branch 404.
|
||||
|
||||
Authentication, authorization, transport, and ambiguous 404 failures are
|
||||
raised rather than treated as absence.
|
||||
"""
|
||||
probe = _probe_remote_branch(h, o, r, auth, branch)
|
||||
status = probe.get("status")
|
||||
if (
|
||||
status == branch_cleanup_guard.READBACK_NOT_FOUND
|
||||
and probe.get("verified_absent")
|
||||
):
|
||||
return False
|
||||
if status == branch_cleanup_guard.READBACK_EXISTS:
|
||||
return True
|
||||
error_class = probe.get("error_class") or "unexpected"
|
||||
raise RuntimeError(
|
||||
f"remote branch probe failed ({error_class}/{status})"
|
||||
)
|
||||
|
||||
|
||||
def _collect_branch_ownership_records(
|
||||
*,
|
||||
remote: str,
|
||||
host: str,
|
||||
org: str,
|
||||
repo: str,
|
||||
branch: str,
|
||||
pr_number: int | None,
|
||||
project_root: str,
|
||||
auth: str | None = None,
|
||||
base_api: str | None = None,
|
||||
) -> dict:
|
||||
"""Gather canonical ownership records for *branch* (secret-free).
|
||||
|
||||
Sources:
|
||||
- author issue locks (task-session / lease files)
|
||||
- control-plane leases (author/reviewer/merger/controller/reconciler)
|
||||
- comment-backed active reviewer leases (O3)
|
||||
- local worktree bindings checked out to the branch
|
||||
|
||||
Returns ``{"records": [...], "inventory_error": bool}``. Control-plane
|
||||
inventory errors set inventory_error=True so callers fail closed (O1).
|
||||
"""
|
||||
records: list[dict] = []
|
||||
inventory_error = False
|
||||
target_branch = (branch or "").strip()
|
||||
if not target_branch:
|
||||
return {"records": records, "inventory_error": False}
|
||||
|
||||
host_n = branch_cleanup_guard.normalize_host(host)
|
||||
|
||||
def _base_rec(**kwargs):
|
||||
rec = {
|
||||
"remote": remote,
|
||||
"host": host_n or host,
|
||||
"org": org,
|
||||
"repo": repo,
|
||||
"branch": target_branch,
|
||||
}
|
||||
rec.update(kwargs)
|
||||
return rec
|
||||
|
||||
# --- Author issue locks (session + lease) ---
|
||||
try:
|
||||
for path in issue_lock_store.iter_lock_files():
|
||||
lock = issue_lock_store.read_lock_file(path)
|
||||
if not isinstance(lock, dict):
|
||||
continue
|
||||
if (
|
||||
str(lock.get("remote") or "") != str(remote)
|
||||
or str(lock.get("org") or "") != str(org)
|
||||
or str(lock.get("repo") or "") != str(repo)
|
||||
):
|
||||
continue
|
||||
# Host identity when present on the lock
|
||||
lock_host = branch_cleanup_guard.normalize_host(
|
||||
lock.get("host") or lock.get("host_name")
|
||||
)
|
||||
if host_n and lock_host and host_n != lock_host:
|
||||
continue
|
||||
if str(lock.get("branch_name") or "").strip() != target_branch:
|
||||
continue
|
||||
freshness = issue_lock_store.assess_lock_freshness(lock)
|
||||
reclaim = issue_lock_store.assess_expired_lock_reclaim(lock)
|
||||
status = str(freshness.get("status") or "unknown")
|
||||
if freshness.get("live"):
|
||||
status = "active"
|
||||
records.append(
|
||||
_base_rec(
|
||||
category=branch_cleanup_guard.OWNERSHIP_CATEGORY_AUTHOR_LEASE,
|
||||
status=status,
|
||||
reclaim_allowed=bool(reclaim.get("reclaim_allowed")),
|
||||
role="author",
|
||||
)
|
||||
)
|
||||
if freshness.get("live"):
|
||||
records.append(
|
||||
_base_rec(
|
||||
category=(
|
||||
branch_cleanup_guard.OWNERSHIP_CATEGORY_AUTHOR_SESSION
|
||||
),
|
||||
status="active",
|
||||
reclaim_allowed=False,
|
||||
role="author",
|
||||
)
|
||||
)
|
||||
except Exception:
|
||||
inventory_error = True
|
||||
records.append(
|
||||
_base_rec(
|
||||
category=branch_cleanup_guard.OWNERSHIP_CATEGORY_AUTHOR_LEASE,
|
||||
status="unknown",
|
||||
reclaim_allowed=False,
|
||||
role="author",
|
||||
)
|
||||
)
|
||||
|
||||
# --- Control-plane leases (role-tagged) ---
|
||||
try:
|
||||
db = None
|
||||
if hasattr(control_plane_db, "get_db"):
|
||||
try:
|
||||
db = control_plane_db.get_db()
|
||||
except Exception:
|
||||
db = None
|
||||
if db is None and hasattr(control_plane_db, "ControlPlaneDB"):
|
||||
try:
|
||||
db = control_plane_db.ControlPlaneDB()
|
||||
except Exception:
|
||||
db = None
|
||||
inventory_error = True
|
||||
if db is None:
|
||||
# O1: unavailable control-plane inventory fails closed.
|
||||
inventory_error = True
|
||||
else:
|
||||
listed = lease_lifecycle.list_active_leases(
|
||||
db,
|
||||
remote=remote,
|
||||
org=org,
|
||||
repo=repo,
|
||||
include_non_active=True,
|
||||
limit=200,
|
||||
)
|
||||
for lease in listed.get("leases") or []:
|
||||
work_kind = str(lease.get("work_kind") or "")
|
||||
work_number = lease.get("work_number")
|
||||
lease_branch = str(
|
||||
lease.get("branch")
|
||||
or lease.get("branch_name")
|
||||
or ""
|
||||
).strip()
|
||||
matches_branch = lease_branch == target_branch
|
||||
matches_pr = (
|
||||
work_kind == "pr"
|
||||
and pr_number is not None
|
||||
and int(work_number or 0) == int(pr_number)
|
||||
)
|
||||
if not (matches_branch or matches_pr):
|
||||
continue
|
||||
if matches_pr and not lease_branch:
|
||||
lease_branch = target_branch
|
||||
if lease_branch != target_branch:
|
||||
continue
|
||||
lease_host = branch_cleanup_guard.normalize_host(
|
||||
lease.get("host") or lease.get("host_name")
|
||||
)
|
||||
if host_n and lease_host and host_n != lease_host:
|
||||
continue
|
||||
fr = lease.get("freshness") or lease_lifecycle.classify_lease_freshness(
|
||||
lease
|
||||
)
|
||||
freshness_status = str(
|
||||
(fr.get("freshness") if isinstance(fr, dict) else None)
|
||||
or lease.get("status")
|
||||
or "unknown"
|
||||
)
|
||||
role = str(lease.get("role") or "unknown")
|
||||
category = branch_cleanup_guard.ownership_category_for_role(role)
|
||||
# O2: expired never auto-receives reclaim_allowed=True.
|
||||
if freshness_status == "active":
|
||||
status = "active"
|
||||
reclaim_allowed = False
|
||||
elif freshness_status in {"released", "abandoned"}:
|
||||
status = freshness_status
|
||||
reclaim_allowed = True
|
||||
elif freshness_status == "expired" or (
|
||||
isinstance(fr, dict) and fr.get("expired_by_time")
|
||||
):
|
||||
status = "expired"
|
||||
reclaim_allowed = False # O2 fail closed
|
||||
else:
|
||||
status = freshness_status
|
||||
reclaim_allowed = False
|
||||
records.append(
|
||||
_base_rec(
|
||||
category=category,
|
||||
status=status,
|
||||
reclaim_allowed=reclaim_allowed,
|
||||
role=role,
|
||||
host=lease_host or host_n or host,
|
||||
)
|
||||
)
|
||||
except Exception:
|
||||
# O1: fail closed on control-plane inventory errors.
|
||||
inventory_error = True
|
||||
records.append(
|
||||
_base_rec(
|
||||
category=branch_cleanup_guard.OWNERSHIP_CATEGORY_INVENTORY_ERROR,
|
||||
status="unknown",
|
||||
reclaim_allowed=False,
|
||||
role="control_plane",
|
||||
)
|
||||
)
|
||||
|
||||
# --- O3: comment-backed active reviewer leases ---
|
||||
if pr_number is not None and auth and base_api:
|
||||
try:
|
||||
comments = api_get_all(
|
||||
f"{base_api}/issues/{int(pr_number)}/comments", auth
|
||||
)
|
||||
active = reviewer_pr_lease.find_active_reviewer_lease(
|
||||
comments, pr_number=int(pr_number)
|
||||
)
|
||||
if active:
|
||||
records.append(
|
||||
_base_rec(
|
||||
category=(
|
||||
branch_cleanup_guard.OWNERSHIP_CATEGORY_REVIEWER_LEASE
|
||||
),
|
||||
status="active",
|
||||
reclaim_allowed=False,
|
||||
role="reviewer",
|
||||
)
|
||||
)
|
||||
except Exception:
|
||||
inventory_error = True
|
||||
records.append(
|
||||
_base_rec(
|
||||
category=branch_cleanup_guard.OWNERSHIP_CATEGORY_INVENTORY_ERROR,
|
||||
status="unknown",
|
||||
reclaim_allowed=False,
|
||||
role="reviewer_comment_lease",
|
||||
)
|
||||
)
|
||||
|
||||
# --- Worktree bindings checked out to the target branch ---
|
||||
try:
|
||||
for entry in worktree_cleanup_audit.list_worktrees(project_root):
|
||||
wt_branch = str(entry.get("branch") or "").strip()
|
||||
if wt_branch != target_branch:
|
||||
continue
|
||||
records.append(
|
||||
_base_rec(
|
||||
category=(
|
||||
branch_cleanup_guard.OWNERSHIP_CATEGORY_WORKTREE_BINDING
|
||||
),
|
||||
status="active",
|
||||
reclaim_allowed=False,
|
||||
role="worktree",
|
||||
)
|
||||
)
|
||||
except Exception:
|
||||
inventory_error = True
|
||||
records.append(
|
||||
_base_rec(
|
||||
category=branch_cleanup_guard.OWNERSHIP_CATEGORY_WORKTREE_BINDING,
|
||||
status="unknown",
|
||||
reclaim_allowed=False,
|
||||
role="worktree",
|
||||
)
|
||||
)
|
||||
|
||||
return {"records": records, "inventory_error": inventory_error}
|
||||
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
@@ -6295,6 +6811,66 @@ def gitea_reconcile_merged_cleanups(
|
||||
if remote_assessment.get("safe_to_delete_remote"):
|
||||
import urllib.parse
|
||||
|
||||
pr_num = entry.get("pr_number")
|
||||
try:
|
||||
pr_num_int = int(pr_num) if pr_num is not None else None
|
||||
except (TypeError, ValueError):
|
||||
pr_num_int = None
|
||||
ownership_bundle = _collect_branch_ownership_records(
|
||||
remote=remote,
|
||||
host=h,
|
||||
org=o,
|
||||
repo=r,
|
||||
branch=head_branch,
|
||||
pr_number=pr_num_int,
|
||||
project_root=PROJECT_ROOT,
|
||||
auth=auth,
|
||||
base_api=base,
|
||||
)
|
||||
ownership_records = list(ownership_bundle.get("records") or [])
|
||||
if ownership_bundle.get("inventory_error"):
|
||||
ownership_records.append(
|
||||
{
|
||||
"category": (
|
||||
branch_cleanup_guard.OWNERSHIP_CATEGORY_INVENTORY_ERROR
|
||||
),
|
||||
"status": "unknown",
|
||||
"remote": remote,
|
||||
"host": h,
|
||||
"org": o,
|
||||
"repo": r,
|
||||
"branch": head_branch,
|
||||
"reclaim_allowed": False,
|
||||
"role": "inventory",
|
||||
}
|
||||
)
|
||||
ownership = branch_cleanup_guard.assess_active_branch_ownership(
|
||||
remote=remote,
|
||||
org=o,
|
||||
repo=r,
|
||||
branch=head_branch,
|
||||
host=h,
|
||||
records=ownership_records,
|
||||
)
|
||||
if ownership.get("block"):
|
||||
actions.append(
|
||||
{
|
||||
"action": "delete_remote_branch",
|
||||
"branch": head_branch,
|
||||
"success": False,
|
||||
"performed": False,
|
||||
"delete_acknowledged": False,
|
||||
"verified_absent": False,
|
||||
"blocker_kind": "active_branch_ownership",
|
||||
"reasons": ownership.get("reasons") or [],
|
||||
"blocking_categories": ownership.get(
|
||||
"blocking_categories"
|
||||
)
|
||||
or [],
|
||||
}
|
||||
)
|
||||
continue
|
||||
|
||||
encoded = urllib.parse.quote(head_branch, safe="")
|
||||
url = f"{base}/branches/{encoded}"
|
||||
with _audited(
|
||||
@@ -6304,14 +6880,28 @@ def gitea_reconcile_merged_cleanups(
|
||||
org=o,
|
||||
repo=r,
|
||||
target_branch=head_branch,
|
||||
request_metadata={"branch": head_branch, "source": "reconcile_merged_cleanups"},
|
||||
request_metadata={
|
||||
"branch": head_branch,
|
||||
"source": "reconcile_merged_cleanups",
|
||||
"ownership_checked": True,
|
||||
},
|
||||
):
|
||||
api_request("DELETE", url, auth)
|
||||
readback = _probe_remote_branch(h, o, r, auth, head_branch)
|
||||
readback_assessment = branch_cleanup_guard.assess_post_delete_readback(
|
||||
readback
|
||||
)
|
||||
verified = bool(readback_assessment.get("verified_absent"))
|
||||
actions.append(
|
||||
{
|
||||
"action": "delete_remote_branch",
|
||||
"branch": head_branch,
|
||||
"success": True,
|
||||
"success": bool(readback_assessment.get("ok")),
|
||||
"performed": True,
|
||||
"delete_acknowledged": True,
|
||||
"verified_absent": verified,
|
||||
"readback": readback_assessment.get("readback"),
|
||||
"reasons": readback_assessment.get("reasons") or [],
|
||||
}
|
||||
)
|
||||
|
||||
@@ -11269,6 +11859,8 @@ def gitea_resolve_task_capability(
|
||||
"gitea_commit_files",
|
||||
"address_pr_change_requests",
|
||||
"delete_branch",
|
||||
"cleanup_merged_pr_branch",
|
||||
"reconciliation_cleanup",
|
||||
"work_issue",
|
||||
"work-issue",
|
||||
}
|
||||
|
||||
+127
-8
@@ -20,12 +20,114 @@ if PROJECT_ROOT not in sys.path:
|
||||
import gitea_config
|
||||
|
||||
|
||||
AUTHOR_DEFAULT_ALLOWED = ["read", "branch", "commit", "push", "open_pr", "comment"]
|
||||
AUTHOR_DEFAULT_FORBIDDEN = ["approve", "request_changes", "merge"]
|
||||
REVIEWER_DEFAULT_ALLOWED = [
|
||||
"read", "review", "comment", "approve", "request_changes", "merge"
|
||||
# Defaults emit *canonical* operation names only. Shorthand that is not in
|
||||
# gitea_config.GITEA_OPERATION_ALIASES (e.g. ``pr.close``, ``issue.close``)
|
||||
# is silently dropped by the production loader and must never appear here.
|
||||
AUTHOR_DEFAULT_ALLOWED = [
|
||||
"gitea.read",
|
||||
"gitea.branch.create",
|
||||
"gitea.repo.commit",
|
||||
"gitea.branch.push",
|
||||
"gitea.pr.create",
|
||||
"gitea.pr.comment",
|
||||
]
|
||||
REVIEWER_DEFAULT_FORBIDDEN = ["branch", "commit", "push", "open_pr"]
|
||||
AUTHOR_DEFAULT_FORBIDDEN = [
|
||||
"gitea.pr.approve",
|
||||
"gitea.pr.request_changes",
|
||||
"gitea.pr.merge",
|
||||
]
|
||||
REVIEWER_DEFAULT_ALLOWED = [
|
||||
"gitea.read",
|
||||
"gitea.pr.review",
|
||||
"gitea.pr.comment",
|
||||
"gitea.pr.approve",
|
||||
"gitea.pr.request_changes",
|
||||
"gitea.pr.merge",
|
||||
]
|
||||
REVIEWER_DEFAULT_FORBIDDEN = [
|
||||
"gitea.branch.create",
|
||||
"gitea.repo.commit",
|
||||
"gitea.branch.push",
|
||||
"gitea.pr.create",
|
||||
]
|
||||
# Required reconciler ops (read + pr.close) plus recommended comment/close and
|
||||
# branch.delete for guarded merged-PR cleanup. All names must normalize via
|
||||
# gitea_config.normalize_operation without being dropped.
|
||||
RECONCILER_DEFAULT_ALLOWED = [
|
||||
"gitea.read",
|
||||
"gitea.pr.close",
|
||||
"gitea.pr.comment",
|
||||
"gitea.issue.comment",
|
||||
"gitea.issue.close",
|
||||
"gitea.branch.delete",
|
||||
]
|
||||
RECONCILER_DEFAULT_FORBIDDEN = [
|
||||
"gitea.pr.approve",
|
||||
"gitea.pr.merge",
|
||||
"gitea.pr.review",
|
||||
"gitea.pr.create",
|
||||
"gitea.branch.push",
|
||||
"gitea.repo.commit",
|
||||
]
|
||||
|
||||
# Migration-only expansions for common shorthands that are *not* in
|
||||
# GITEA_OPERATION_ALIASES. Emitted output is always the canonical form so a
|
||||
# second canonicalize pass is a no-op (idempotent).
|
||||
_MIGRATION_ONLY_ALIASES = {
|
||||
"pr.close": "gitea.pr.close",
|
||||
"pr.comment": "gitea.pr.comment",
|
||||
"issue.close": "gitea.issue.close",
|
||||
"branch.delete": "gitea.branch.delete",
|
||||
}
|
||||
|
||||
# Reconciler required ops that must survive migration (from reconciler_profile).
|
||||
RECONCILER_REQUIRED_CANONICAL = ("gitea.read", "gitea.pr.close")
|
||||
|
||||
|
||||
def canonicalize_operation(op: str) -> str:
|
||||
"""Return a canonical operation name accepted by the production loader.
|
||||
|
||||
Fail closed on unknown/ambiguous spellings so required permissions cannot
|
||||
be silently dropped by ``check_operation`` later.
|
||||
"""
|
||||
if not isinstance(op, str) or not op.strip():
|
||||
raise ValueError("operation must be a non-empty string (fail closed)")
|
||||
op = op.strip()
|
||||
try:
|
||||
return gitea_config.normalize_operation(op)
|
||||
except gitea_config.ConfigError:
|
||||
pass
|
||||
if op in _MIGRATION_ONLY_ALIASES:
|
||||
return _MIGRATION_ONLY_ALIASES[op]
|
||||
raise ValueError(
|
||||
f"operation {op!r} cannot be canonicalized for migration "
|
||||
"(unknown/ambiguous; fail closed — production loader would drop it)"
|
||||
)
|
||||
|
||||
|
||||
def canonicalize_operations(ops, *, context: str = "operations") -> list[str]:
|
||||
"""Canonicalize a list of operations; preserve order, drop duplicates."""
|
||||
if not isinstance(ops, list):
|
||||
raise ValueError(f"{context} must be a list (fail closed)")
|
||||
out: list[str] = []
|
||||
seen: set[str] = set()
|
||||
for entry in ops:
|
||||
canon = canonicalize_operation(entry)
|
||||
if canon not in seen:
|
||||
seen.add(canon)
|
||||
out.append(canon)
|
||||
return out
|
||||
|
||||
|
||||
def _assert_reconciler_required_survive(allowed: list[str], profile_name: str) -> None:
|
||||
"""Fail visibly when migration would leave a reconciler without required ops."""
|
||||
missing = [op for op in RECONCILER_REQUIRED_CANONICAL if op not in set(allowed)]
|
||||
if missing:
|
||||
raise ValueError(
|
||||
f"Profile '{profile_name}' (reconciler) is missing required "
|
||||
f"operation(s) after migration: {missing}. Refusing to emit a "
|
||||
"profile that would silently fail pr.close / read (fail closed)."
|
||||
)
|
||||
|
||||
|
||||
def infer_role(name, execution_profile):
|
||||
@@ -90,9 +192,11 @@ def migrate_v1_to_v2(v1_data):
|
||||
ident_name = "reviewer"
|
||||
elif role == "author":
|
||||
ident_name = "author"
|
||||
elif role == "reconciler":
|
||||
ident_name = "reconciler"
|
||||
else:
|
||||
role = prof.get("role")
|
||||
if role not in (None, "author", "reviewer"):
|
||||
if role not in (None, "author", "reviewer", "reconciler"):
|
||||
raise ValueError(
|
||||
f"Profile '{name}' has unsupported role {role!r}"
|
||||
)
|
||||
@@ -124,20 +228,35 @@ def migrate_v1_to_v2(v1_data):
|
||||
raise ValueError(
|
||||
f"Profile '{name}' operation fields must be lists"
|
||||
)
|
||||
identity_data["allowed_operations"] = list(allowed)
|
||||
identity_data["forbidden_operations"] = list(forbidden)
|
||||
try:
|
||||
identity_data["allowed_operations"] = canonicalize_operations(
|
||||
allowed, context=f"profile '{name}' allowed_operations"
|
||||
)
|
||||
identity_data["forbidden_operations"] = canonicalize_operations(
|
||||
forbidden, context=f"profile '{name}' forbidden_operations"
|
||||
)
|
||||
except ValueError as exc:
|
||||
raise ValueError(f"Profile '{name}': {exc}") from exc
|
||||
elif role == "author":
|
||||
identity_data["allowed_operations"] = list(AUTHOR_DEFAULT_ALLOWED)
|
||||
identity_data["forbidden_operations"] = list(AUTHOR_DEFAULT_FORBIDDEN)
|
||||
elif role == "reviewer":
|
||||
identity_data["allowed_operations"] = list(REVIEWER_DEFAULT_ALLOWED)
|
||||
identity_data["forbidden_operations"] = list(REVIEWER_DEFAULT_FORBIDDEN)
|
||||
elif role == "reconciler":
|
||||
identity_data["allowed_operations"] = list(RECONCILER_DEFAULT_ALLOWED)
|
||||
identity_data["forbidden_operations"] = list(RECONCILER_DEFAULT_FORBIDDEN)
|
||||
else:
|
||||
raise ValueError(
|
||||
f"Profile '{name}' has no explicit operation lists and no "
|
||||
"unambiguous author/reviewer role marker (fail closed)"
|
||||
)
|
||||
|
||||
if role == "reconciler":
|
||||
_assert_reconciler_required_survive(
|
||||
identity_data["allowed_operations"], name
|
||||
)
|
||||
|
||||
# Nest inside environments/services structure
|
||||
env = environments.setdefault(env_name, {})
|
||||
services = env.setdefault("services", {})
|
||||
|
||||
@@ -18,6 +18,11 @@ RECONCILER_RECOMMENDED_OPERATIONS = (
|
||||
"gitea.pr.comment",
|
||||
"gitea.issue.comment",
|
||||
"gitea.issue.close",
|
||||
# Merged-branch cleanup is reconciler-owned (task_capability_map maps
|
||||
# cleanup_merged_pr_branch -> reconciler). The permission is only
|
||||
# exercisable through the guarded gitea_cleanup_merged_pr_branch path
|
||||
# (#514): merged proof, protected-branch refusal, explicit confirmation.
|
||||
"gitea.branch.delete",
|
||||
)
|
||||
|
||||
RECONCILER_FORBIDDEN_OPERATIONS = (
|
||||
|
||||
@@ -27,7 +27,13 @@ from task_capability_map import required_permission, required_role
|
||||
|
||||
DELETE_PROFILE = {
|
||||
"profile_name": "prgs-author-delete",
|
||||
"allowed_operations": ["gitea.read", "gitea.branch.delete"],
|
||||
"role": "author",
|
||||
"allowed_operations": [
|
||||
"gitea.read",
|
||||
"gitea.pr.create",
|
||||
"gitea.branch.push",
|
||||
"gitea.branch.delete",
|
||||
],
|
||||
"forbidden_operations": [],
|
||||
"audit_label": "prgs-author-delete",
|
||||
}
|
||||
|
||||
+1106
-27
File diff suppressed because it is too large
Load Diff
@@ -1425,10 +1425,16 @@ class TestReviewPR(unittest.TestCase):
|
||||
class TestDeleteBranch(unittest.TestCase):
|
||||
|
||||
DELETE_PROFILE = {
|
||||
"profile_name": "test-deleter",
|
||||
"allowed_operations": ["gitea.read", "gitea.branch.delete"],
|
||||
"profile_name": "test-author-deleter",
|
||||
"role": "author",
|
||||
"allowed_operations": [
|
||||
"gitea.read",
|
||||
"gitea.pr.create",
|
||||
"gitea.branch.push",
|
||||
"gitea.branch.delete",
|
||||
],
|
||||
"forbidden_operations": [],
|
||||
"audit_label": "test-deleter",
|
||||
"audit_label": "test-author-deleter",
|
||||
}
|
||||
|
||||
@patch("mcp_server.get_profile", return_value=DELETE_PROFILE)
|
||||
|
||||
@@ -92,14 +92,19 @@ class TestMigrateProfiles(unittest.TestCase):
|
||||
author = prgs_gitea["identities"]["author"]
|
||||
self.assertEqual(author["username"], "jcwalker3")
|
||||
self.assertEqual(author["auth"]["id"], "redacted-author-ref")
|
||||
self.assertEqual(author["allowed_operations"], ["read", "comment"])
|
||||
self.assertEqual(author["forbidden_operations"], ["approve", "merge"])
|
||||
self.assertEqual(
|
||||
author["allowed_operations"], ["gitea.read", "gitea.pr.comment"]
|
||||
)
|
||||
self.assertEqual(
|
||||
author["forbidden_operations"],
|
||||
["gitea.pr.approve", "gitea.pr.merge"],
|
||||
)
|
||||
|
||||
reviewer = prgs_gitea["identities"]["reviewer"]
|
||||
self.assertEqual(reviewer["role"], "reviewer")
|
||||
self.assertEqual(reviewer["username"], "sysadmin")
|
||||
self.assertEqual(reviewer["auth"]["id"], "redacted-reviewer-ref")
|
||||
self.assertIn("merge", reviewer["allowed_operations"])
|
||||
self.assertIn("gitea.pr.merge", reviewer["allowed_operations"])
|
||||
|
||||
def test_alias_generation(self):
|
||||
"""Test that aliases are correctly generated to support old profile names."""
|
||||
@@ -188,7 +193,7 @@ class TestMigrateProfiles(unittest.TestCase):
|
||||
self.assertNotIn("token", stdout_output.lower())
|
||||
|
||||
def test_explicit_operations_are_preserved(self):
|
||||
"""Explicit v1 permissions must not be replaced by role defaults."""
|
||||
"""Explicit v1 permissions are canonicalized, not replaced by role defaults."""
|
||||
v1_data = json.loads(json.dumps(self.v1_content))
|
||||
v1_data["profiles"]["prgs-reviewer"]["allowed_operations"] = ["read"]
|
||||
v1_data["profiles"]["prgs-reviewer"]["forbidden_operations"] = ["merge"]
|
||||
@@ -198,8 +203,8 @@ class TestMigrateProfiles(unittest.TestCase):
|
||||
v2_data["environments"]["prgs"]["services"]["gitea"]
|
||||
["identities"]["reviewer"]
|
||||
)
|
||||
self.assertEqual(reviewer["allowed_operations"], ["read"])
|
||||
self.assertEqual(reviewer["forbidden_operations"], ["merge"])
|
||||
self.assertEqual(reviewer["allowed_operations"], ["gitea.read"])
|
||||
self.assertEqual(reviewer["forbidden_operations"], ["gitea.pr.merge"])
|
||||
|
||||
def test_inferred_role_defaults_only_when_unambiguous(self):
|
||||
"""Role defaults are allowed only for clear author/reviewer profiles."""
|
||||
@@ -306,6 +311,171 @@ class TestMigrateProfiles(unittest.TestCase):
|
||||
migrate_profiles.main()
|
||||
self.assertEqual(cm.exception.code, 1)
|
||||
|
||||
def test_reconciler_profile_migration(self):
|
||||
"""Legacy reconciler shorthands migrate to valid canonical operations."""
|
||||
import gitea_config
|
||||
import reconciler_profile
|
||||
|
||||
v1_data = {
|
||||
"version": 1,
|
||||
"profiles": {
|
||||
"prgs-reconciler": {
|
||||
"base_url": "redacted-prgs-service",
|
||||
"username": "reconciler-agent",
|
||||
"auth": {"type": "keychain", "id": "reconciler-ref"},
|
||||
"execution_profile": "prgs-reconciler",
|
||||
"allowed_operations": [
|
||||
"read",
|
||||
"pr.close",
|
||||
"pr.comment",
|
||||
"issue.comment",
|
||||
"issue.close",
|
||||
"gitea.branch.delete",
|
||||
],
|
||||
"forbidden_operations": [
|
||||
"merge",
|
||||
"approve",
|
||||
"review",
|
||||
"pr.create",
|
||||
"branch.push",
|
||||
"commit",
|
||||
],
|
||||
}
|
||||
}
|
||||
}
|
||||
v2_data = migrate_profiles.migrate_v1_to_v2(v1_data)
|
||||
reconciler = (
|
||||
v2_data["environments"]["prgs"]["services"]["gitea"]
|
||||
["identities"]["reconciler"]
|
||||
)
|
||||
self.assertEqual(reconciler["role"], "reconciler")
|
||||
allowed = reconciler["allowed_operations"]
|
||||
forbidden = reconciler["forbidden_operations"]
|
||||
# No invalid shorthand remains
|
||||
for bad in ("pr.close", "pr.comment", "issue.close", "read", "merge"):
|
||||
self.assertNotIn(bad, allowed)
|
||||
self.assertNotIn(bad, forbidden)
|
||||
for required in (
|
||||
"gitea.read",
|
||||
"gitea.pr.close",
|
||||
"gitea.pr.comment",
|
||||
"gitea.issue.comment",
|
||||
"gitea.issue.close",
|
||||
"gitea.branch.delete",
|
||||
):
|
||||
self.assertIn(required, allowed)
|
||||
# Production loader accepts every allowed op
|
||||
self.assertEqual(
|
||||
gitea_config.normalize_operation(required), required
|
||||
)
|
||||
self.assertEqual(v2_data["aliases"]["prgs-reconciler"], "prgs.gitea.reconciler")
|
||||
assessment = reconciler_profile.assess_reconciler_profile(allowed, forbidden)
|
||||
self.assertTrue(assessment["valid"])
|
||||
self.assertTrue(migrate_profiles.validate_v2_data(v2_data))
|
||||
|
||||
def test_reconciler_profile_defaults(self):
|
||||
"""Reconciler defaults are fully canonical and loader-valid."""
|
||||
import gitea_config
|
||||
import reconciler_profile
|
||||
|
||||
v1_data = {
|
||||
"version": 1,
|
||||
"profiles": {
|
||||
"prgs-reconciler": {
|
||||
"base_url": "redacted-prgs-service",
|
||||
"username": "reconciler-agent",
|
||||
"auth": {"type": "keychain", "id": "reconciler-ref"},
|
||||
"execution_profile": "prgs-reconciler",
|
||||
}
|
||||
}
|
||||
}
|
||||
v2_data = migrate_profiles.migrate_v1_to_v2(v1_data)
|
||||
reconciler = (
|
||||
v2_data["environments"]["prgs"]["services"]["gitea"]
|
||||
["identities"]["reconciler"]
|
||||
)
|
||||
self.assertEqual(reconciler["role"], "reconciler")
|
||||
self.assertEqual(
|
||||
reconciler["allowed_operations"],
|
||||
migrate_profiles.RECONCILER_DEFAULT_ALLOWED,
|
||||
)
|
||||
self.assertEqual(
|
||||
reconciler["forbidden_operations"],
|
||||
migrate_profiles.RECONCILER_DEFAULT_FORBIDDEN,
|
||||
)
|
||||
for op in reconciler["allowed_operations"]:
|
||||
self.assertEqual(gitea_config.normalize_operation(op), op)
|
||||
self.assertTrue(op.startswith("gitea."))
|
||||
assessment = reconciler_profile.assess_reconciler_profile(
|
||||
reconciler["allowed_operations"],
|
||||
reconciler["forbidden_operations"],
|
||||
)
|
||||
self.assertTrue(assessment["valid"])
|
||||
self.assertNotIn(
|
||||
"gitea.branch.delete", assessment["missing_recommended_operations"]
|
||||
)
|
||||
|
||||
def test_reconciler_migration_idempotent_canonicalize(self):
|
||||
"""Second canonicalize of already-canonical ops is a no-op."""
|
||||
first = migrate_profiles.canonicalize_operations(
|
||||
list(migrate_profiles.RECONCILER_DEFAULT_ALLOWED)
|
||||
)
|
||||
second = migrate_profiles.canonicalize_operations(first)
|
||||
self.assertEqual(first, second)
|
||||
self.assertEqual(first, list(migrate_profiles.RECONCILER_DEFAULT_ALLOWED))
|
||||
|
||||
def test_reconciler_missing_required_fails_visibly(self):
|
||||
"""Missing gitea.pr.close after migration fails closed (not silent drop)."""
|
||||
v1_data = {
|
||||
"version": 1,
|
||||
"profiles": {
|
||||
"prgs-reconciler": {
|
||||
"base_url": "redacted-prgs-service",
|
||||
"username": "reconciler-agent",
|
||||
"auth": {"type": "keychain", "id": "reconciler-ref"},
|
||||
"execution_profile": "prgs-reconciler",
|
||||
"allowed_operations": ["read", "gitea.branch.delete"],
|
||||
"forbidden_operations": ["merge"],
|
||||
}
|
||||
},
|
||||
}
|
||||
with self.assertRaisesRegex(ValueError, "missing required"):
|
||||
migrate_profiles.migrate_v1_to_v2(v1_data)
|
||||
|
||||
def test_unknown_operation_fails_visibly(self):
|
||||
v1_data = {
|
||||
"version": 1,
|
||||
"profiles": {
|
||||
"prgs-author": {
|
||||
"base_url": "redacted-prgs-service",
|
||||
"username": "jcwalker3",
|
||||
"auth": {"type": "keychain", "id": "hidden-author-ref"},
|
||||
"execution_profile": "prgs-author",
|
||||
"allowed_operations": ["read", "not.a.real.op"],
|
||||
"forbidden_operations": ["merge"],
|
||||
}
|
||||
},
|
||||
}
|
||||
with self.assertRaisesRegex(ValueError, "cannot be canonicalized"):
|
||||
migrate_profiles.migrate_v1_to_v2(v1_data)
|
||||
|
||||
def test_role_inference_author_reviewer_merger_reconciler(self):
|
||||
self.assertEqual(
|
||||
migrate_profiles.infer_role("prgs-author", "prgs-author"), "author"
|
||||
)
|
||||
self.assertEqual(
|
||||
migrate_profiles.infer_role("prgs-reviewer", "prgs-reviewer"),
|
||||
"reviewer",
|
||||
)
|
||||
self.assertEqual(
|
||||
migrate_profiles.infer_role("prgs-reconciler", "prgs-reconciler"),
|
||||
"reconciler",
|
||||
)
|
||||
self.assertIsNone(
|
||||
migrate_profiles.infer_role("prgs-merger", "prgs-merger")
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
|
||||
@@ -82,6 +82,42 @@ class TestReconcilerProfileModel(unittest.TestCase):
|
||||
"reconciler",
|
||||
)
|
||||
|
||||
def test_branch_delete_is_recommended_for_reconciler(self):
|
||||
self.assertIn(
|
||||
"gitea.branch.delete",
|
||||
reconciler_profile.RECONCILER_RECOMMENDED_OPERATIONS,
|
||||
)
|
||||
self.assertNotIn(
|
||||
"gitea.branch.delete",
|
||||
reconciler_profile.RECONCILER_REQUIRED_OPERATIONS,
|
||||
)
|
||||
|
||||
def test_reconciler_with_branch_delete_stays_valid(self):
|
||||
allowed = PRGS_RECONCILER_ALLOWED + ["gitea.branch.delete"]
|
||||
result = reconciler_profile.assess_reconciler_profile(
|
||||
allowed,
|
||||
PRGS_RECONCILER_FORBIDDEN,
|
||||
)
|
||||
self.assertTrue(result["is_reconciler_profile"])
|
||||
self.assertTrue(result["valid"])
|
||||
self.assertNotIn(
|
||||
"gitea.branch.delete", result["missing_recommended_operations"]
|
||||
)
|
||||
self.assertEqual(
|
||||
mcp_server._role_kind(allowed, PRGS_RECONCILER_FORBIDDEN),
|
||||
"reconciler",
|
||||
)
|
||||
|
||||
def test_reconciler_without_branch_delete_reports_missing_recommended(self):
|
||||
result = reconciler_profile.assess_reconciler_profile(
|
||||
PRGS_RECONCILER_ALLOWED,
|
||||
PRGS_RECONCILER_FORBIDDEN,
|
||||
)
|
||||
self.assertTrue(result["valid"])
|
||||
self.assertIn(
|
||||
"gitea.branch.delete", result["missing_recommended_operations"]
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user