fix(profiles): canonical reconciler ops and guard raw branch delete (#687)

Address REQUEST_CHANGES on PR #688:

- migrate_profiles: emit fully canonical reconciler/author/reviewer defaults;
  canonicalize explicit ops; fail if reconciler loses required pr.close/read
- gitea_delete_branch: deny reconciler (and non-author roles); refuse
  preservation/protected branches before any API call
- gitea_cleanup_merged_pr_branch: require reconciler role only; block
  preservation/evidence branches via branch_cleanup_guard
- docs: end-to-end operator runbook for profile migrate/apply/reconnect/cleanup
- tests: migration canonicalization, idempotency, raw-delete denial, guarded
  cleanup success and unmerged/preserve rejections

Closes #687.
This commit is contained in:
2026-07-12 22:35:14 -04:00
parent c7a444eb4b
commit 4a63578003
8 changed files with 702 additions and 55 deletions
+18
View File
@@ -7,6 +7,19 @@ from typing import Any
PROTECTED_BRANCHES = frozenset({"master", "main", "dev"}) 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 = ( _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+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), 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") reasons.append("PR head branch is missing")
if head_branch in protected: if head_branch in protected:
reasons.append(f"branch '{head_branch}' is 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: if head_branch in open_pr_heads:
reasons.append("an open PR still references this head branch") reasons.append("an open PR still references this head branch")
if head_on_target is False: if head_on_target is False:
+164 -7
View File
@@ -259,17 +259,21 @@ Least-privilege constraints:
reviewer, and merger profiles must never hold it; `gitea_delete_branch` reviewer, and merger profiles must never hold it; `gitea_delete_branch`
and `gitea_cleanup_merged_pr_branch` fail closed on any profile without and `gitea_cleanup_merged_pr_branch` fail closed on any profile without
the permission. the permission.
- Even with the permission, deletion is only supported through the guarded - Even with the permission, reconciler deletion is only supported through the
`gitea_cleanup_merged_pr_branch` path (#514): the PR must be merged, the guarded `gitea_cleanup_merged_pr_branch` path (#514 / #687): the PR must be
head an ancestor of the target, the branch not protected merged, the head an ancestor of the target, the branch not protected
(`master`/`main`/`dev`), no open PR may still use the head, and an (`master`/`main`/`dev`), the branch not a preservation/evidence ref (e.g.
explicit `CLEANUP MERGED PR <n> BRANCH <branch>` confirmation is `chore/issue-681-preserve-review-session-wip`), no open PR may still use the
required. 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 - Raw `git branch -d` / `git push --delete` cleanup remains blocked by
`branch_cleanup_guard` and the final-report validator regardless of `branch_cleanup_guard` and the final-report validator regardless of
profile permissions. profile permissions.
- `gitea.branch.delete` has no short alias in `GITEA_OPERATION_ALIASES`; - `gitea.branch.delete` has no short alias in `GITEA_OPERATION_ALIASES`;
write it fully qualified in `allowed_operations`. 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 Launch a static `gitea-reconciler` MCP namespace with
`GITEA_MCP_PROFILE=prgs-reconciler`. Profile shape is validated by `GITEA_MCP_PROFILE=prgs-reconciler`. Profile shape is validated by
@@ -279,6 +283,159 @@ Launch a static `gitea-reconciler` MCP namespace with
fresh target-branch fetch, recorded target SHA, and ancestor proof. PRs whose fresh target-branch fetch, recorded target SHA, and ancestor proof. PRs whose
heads are not already landed cannot be closed through this path. 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 identitys
`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 ## Identity and fail-closed rules
Before **any** mutating action, a workflow must know both: Before **any** mutating action, a workflow must know both:
+70 -7
View File
@@ -5925,6 +5925,65 @@ def gitea_delete_branch(
"permission_report": _permission_block_report("gitea.branch.delete"), "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_allowed, audit_reasons = (
audit_reconciliation_mode.check_audit_mutation_allowed("delete_branch") audit_reconciliation_mode.check_audit_mutation_allowed("delete_branch")
) )
@@ -5976,18 +6035,20 @@ def gitea_cleanup_merged_pr_branch(
} }
profile = get_profile() profile = get_profile()
active_role = _role_kind( active_role = _profile_role_kind(profile)
profile.get("allowed_operations", []), # cleanup_merged_pr_branch is reconciler-owned (task_capability_map).
profile.get("forbidden_operations", []), # Author/reviewer/merger must not reach this path even if they somehow
) # hold gitea.branch.delete.
if active_role == "reviewer": if active_role != "reconciler":
return { return {
"success": False, "success": False,
"performed": False, "performed": False,
"required_permission": "gitea.branch.delete", "required_permission": "gitea.branch.delete",
"required_role_kind": "reconciler",
"active_role_kind": active_role,
"reasons": [ "reasons": [
"reviewer profile is not authorized for merged branch cleanup " f"profile role '{active_role}' is not authorized for merged "
"(fail closed)" "branch cleanup; required role is reconciler (fail closed)"
], ],
"permission_report": _permission_block_report("gitea.branch.delete"), "permission_report": _permission_block_report("gitea.branch.delete"),
} }
@@ -11269,6 +11330,8 @@ def gitea_resolve_task_capability(
"gitea_commit_files", "gitea_commit_files",
"address_pr_change_requests", "address_pr_change_requests",
"delete_branch", "delete_branch",
"cleanup_merged_pr_branch",
"reconciliation_cleanup",
"work_issue", "work_issue",
"work-issue", "work-issue",
} }
+117 -9
View File
@@ -20,19 +20,115 @@ if PROJECT_ROOT not in sys.path:
import gitea_config import gitea_config
AUTHOR_DEFAULT_ALLOWED = ["read", "branch", "commit", "push", "open_pr", "comment"] # Defaults emit *canonical* operation names only. Shorthand that is not in
AUTHOR_DEFAULT_FORBIDDEN = ["approve", "request_changes", "merge"] # gitea_config.GITEA_OPERATION_ALIASES (e.g. ``pr.close``, ``issue.close``)
REVIEWER_DEFAULT_ALLOWED = [ # is silently dropped by the production loader and must never appear here.
"read", "review", "comment", "approve", "request_changes", "merge" 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 = [ RECONCILER_DEFAULT_ALLOWED = [
"read", "pr.close", "pr.comment", "issue.comment", "issue.close", "gitea.branch.delete" "gitea.read",
"gitea.pr.close",
"gitea.pr.comment",
"gitea.issue.comment",
"gitea.issue.close",
"gitea.branch.delete",
] ]
RECONCILER_DEFAULT_FORBIDDEN = [ RECONCILER_DEFAULT_FORBIDDEN = [
"approve", "merge", "review", "pr.create", "branch.push", "commit" "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): def infer_role(name, execution_profile):
"""Return the unambiguous role for a legacy profile name, or None.""" """Return the unambiguous role for a legacy profile name, or None."""
@@ -132,8 +228,15 @@ def migrate_v1_to_v2(v1_data):
raise ValueError( raise ValueError(
f"Profile '{name}' operation fields must be lists" f"Profile '{name}' operation fields must be lists"
) )
identity_data["allowed_operations"] = list(allowed) try:
identity_data["forbidden_operations"] = list(forbidden) 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": elif role == "author":
identity_data["allowed_operations"] = list(AUTHOR_DEFAULT_ALLOWED) identity_data["allowed_operations"] = list(AUTHOR_DEFAULT_ALLOWED)
identity_data["forbidden_operations"] = list(AUTHOR_DEFAULT_FORBIDDEN) identity_data["forbidden_operations"] = list(AUTHOR_DEFAULT_FORBIDDEN)
@@ -149,6 +252,11 @@ def migrate_v1_to_v2(v1_data):
"unambiguous author/reviewer role marker (fail closed)" "unambiguous author/reviewer role marker (fail closed)"
) )
if role == "reconciler":
_assert_reconciler_required_survive(
identity_data["allowed_operations"], name
)
# Nest inside environments/services structure # Nest inside environments/services structure
env = environments.setdefault(env_name, {}) env = environments.setdefault(env_name, {})
services = env.setdefault("services", {}) services = env.setdefault("services", {})
+7 -1
View File
@@ -27,7 +27,13 @@ from task_capability_map import required_permission, required_role
DELETE_PROFILE = { DELETE_PROFILE = {
"profile_name": "prgs-author-delete", "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": [], "forbidden_operations": [],
"audit_label": "prgs-author-delete", "audit_label": "prgs-author-delete",
} }
+189 -16
View File
@@ -8,10 +8,33 @@ import branch_cleanup_guard as guard # noqa: E402
import mcp_server # noqa: E402 import mcp_server # noqa: E402
import task_capability_map # noqa: E402 import task_capability_map # noqa: E402
from final_report_validator import assess_final_report_validator # noqa: E402 from final_report_validator import assess_final_report_validator # noqa: E402
from mcp_server import gitea_cleanup_merged_pr_branch # noqa: E402 from mcp_server import gitea_cleanup_merged_pr_branch, gitea_delete_branch # noqa: E402
FAKE_AUTH = "token fake" FAKE_AUTH = "token fake"
# Reconciler-shaped profile that holds branch.delete (recommended) plus
# required pr.close/read so _role_kind classifies as reconciler.
RECONCILER_WITH_DELETE = {
"profile_name": "prgs-reconciler",
"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",
],
}
class TestRawBranchDeleteGuard(unittest.TestCase): class TestRawBranchDeleteGuard(unittest.TestCase):
def test_detects_local_and_remote_raw_git_delete_commands(self): def test_detects_local_and_remote_raw_git_delete_commands(self):
@@ -134,11 +157,7 @@ class TestMergedPrBranchCleanupTool(unittest.TestCase):
def test_root_checkout_cleanup_fails_closed(self): def test_root_checkout_cleanup_fails_closed(self):
patch( patch(
"mcp_server.get_profile", "mcp_server.get_profile",
return_value={ return_value=dict(RECONCILER_WITH_DELETE),
"profile_name": "branch-cleanup",
"allowed_operations": ["gitea.read", "gitea.branch.delete"],
"forbidden_operations": [],
},
).start() ).start()
res = gitea_cleanup_merged_pr_branch( res = gitea_cleanup_merged_pr_branch(
pr_number=487, pr_number=487,
@@ -155,11 +174,7 @@ class TestMergedPrBranchCleanupTool(unittest.TestCase):
branch = "feat/issue-485-lease-comments-non-list-guard" branch = "feat/issue-485-lease-comments-non-list-guard"
patch( patch(
"mcp_server.get_profile", "mcp_server.get_profile",
return_value={ return_value=dict(RECONCILER_WITH_DELETE),
"profile_name": "branch-cleanup",
"allowed_operations": ["gitea.read", "gitea.branch.delete"],
"forbidden_operations": [],
},
).start() ).start()
self.mock_api.side_effect = [ self.mock_api.side_effect = [
{ {
@@ -190,11 +205,7 @@ class TestMergedPrBranchCleanupTool(unittest.TestCase):
branch = "feat/issue-485-lease-comments-non-list-guard" branch = "feat/issue-485-lease-comments-non-list-guard"
patch( patch(
"mcp_server.get_profile", "mcp_server.get_profile",
return_value={ return_value=dict(RECONCILER_WITH_DELETE),
"profile_name": "branch-cleanup",
"allowed_operations": ["gitea.read", "gitea.branch.delete"],
"forbidden_operations": [],
},
).start() ).start()
self.mock_api.side_effect = [ self.mock_api.side_effect = [
{ {
@@ -220,6 +231,168 @@ class TestMergedPrBranchCleanupTool(unittest.TestCase):
] ]
self.assertFalse(delete_calls) self.assertFalse(delete_calls)
def test_reconciler_with_branch_delete_cannot_raw_delete(self):
"""#687: reconciler + gitea.branch.delete still cannot call raw delete."""
patch(
"mcp_server.get_profile",
return_value=dict(RECONCILER_WITH_DELETE),
).start()
res = gitea_delete_branch(
branch="fix/issue-683-workflow-guard-hardening",
remote="prgs",
)
self.assertFalse(res.get("success", True))
self.assertFalse(res.get("performed", True))
reasons = " ".join(res.get("reasons") or [])
self.assertIn("raw gitea_delete_branch", reasons)
self.assertIn("cleanup_merged_pr_branch", reasons)
self.mock_api.assert_not_called()
def test_reconciler_raw_delete_denies_preservation_branch(self):
patch(
"mcp_server.get_profile",
return_value=dict(RECONCILER_WITH_DELETE),
).start()
res = gitea_delete_branch(
branch="chore/issue-681-preserve-review-session-wip",
remote="prgs",
)
self.assertFalse(res.get("performed", True))
self.mock_api.assert_not_called()
def test_author_with_branch_delete_role_ok_but_preserve_blocked(self):
"""Author role may use raw delete path when permitted; preserve fails closed."""
patch(
"mcp_server.get_profile",
return_value={
"profile_name": "prgs-author",
"role": "author",
"allowed_operations": [
"gitea.read",
"gitea.pr.create",
"gitea.branch.push",
"gitea.branch.delete",
],
"forbidden_operations": ["gitea.pr.approve", "gitea.pr.merge"],
},
).start()
res = gitea_delete_branch(
branch="chore/issue-681-preserve-review-session-wip",
remote="prgs",
)
self.assertFalse(res.get("performed", True))
self.assertIn("preservation", " ".join(res.get("reasons") or []))
self.mock_api.assert_not_called()
def test_unmerged_branch_cleanup_rejected(self):
branch = "feat/unmerged-work"
patch(
"mcp_server.get_profile",
return_value=dict(RECONCILER_WITH_DELETE),
).start()
self.mock_api.side_effect = [
{
"number": 999,
"merged": False,
"merged_at": None,
"head": {"ref": branch, "sha": "b" * 40},
"base": {"ref": "master"},
},
{},
]
res = gitea_cleanup_merged_pr_branch(
pr_number=999,
confirmation=f"CLEANUP MERGED PR 999 BRANCH {branch}",
branch=branch,
remote="prgs",
worktree_path="/tmp/repo/branches/cleanup",
)
self.assertFalse(res["performed"])
self.assertTrue(
any("not merged" in r for r in (res.get("reasons") or []))
)
delete_calls = [
call for call in self.mock_api.call_args_list if call.args[0] == "DELETE"
]
self.assertFalse(delete_calls)
def test_preservation_branch_cleanup_rejected(self):
branch = "chore/issue-681-preserve-review-session-wip"
patch(
"mcp_server.get_profile",
return_value=dict(RECONCILER_WITH_DELETE),
).start()
self.mock_api.side_effect = [
{
"number": 681,
"merged": True,
"merged_at": "2026-07-08T01:00:00Z",
"head": {"ref": branch, "sha": "c" * 40},
"base": {"ref": "master"},
},
{},
]
res = gitea_cleanup_merged_pr_branch(
pr_number=681,
confirmation=f"CLEANUP MERGED PR 681 BRANCH {branch}",
branch=branch,
remote="prgs",
worktree_path="/tmp/repo/branches/cleanup",
)
self.assertFalse(res["performed"])
self.assertTrue(
any("preservation" in r for r in (res.get("reasons") or []))
)
delete_calls = [
call for call in self.mock_api.call_args_list if call.args[0] == "DELETE"
]
self.assertFalse(delete_calls)
def test_non_reconciler_with_delete_denied_cleanup(self):
patch(
"mcp_server.get_profile",
return_value={
"profile_name": "prgs-author",
"role": "author",
"allowed_operations": [
"gitea.read",
"gitea.pr.create",
"gitea.branch.push",
"gitea.branch.delete",
],
"forbidden_operations": [],
},
).start()
res = gitea_cleanup_merged_pr_branch(
pr_number=487,
confirmation="CLEANUP MERGED PR 487 BRANCH feat/branch",
branch="feat/branch",
remote="prgs",
worktree_path="/tmp/repo/branches/cleanup",
)
self.assertFalse(res["performed"])
self.assertEqual(res.get("required_role_kind"), "reconciler")
self.mock_api.assert_not_called()
def test_assess_guard_rejects_preservation_branch(self):
assessment = guard.assess_merged_pr_branch_cleanup(
pr_number=681,
head_branch="chore/issue-681-preserve-review-session-wip",
merged=True,
remote_branch_exists=True,
open_pr_heads=set(),
head_on_target=True,
delete_capability_allowed=True,
confirmation=(
"CLEANUP MERGED PR 681 BRANCH "
"chore/issue-681-preserve-review-session-wip"
),
)
self.assertFalse(assessment["safe_to_delete"])
self.assertTrue(
any("preservation" in r for r in assessment["block_reasons"])
)
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()
+9 -3
View File
@@ -1425,10 +1425,16 @@ class TestReviewPR(unittest.TestCase):
class TestDeleteBranch(unittest.TestCase): class TestDeleteBranch(unittest.TestCase):
DELETE_PROFILE = { DELETE_PROFILE = {
"profile_name": "test-deleter", "profile_name": "test-author-deleter",
"allowed_operations": ["gitea.read", "gitea.branch.delete"], "role": "author",
"allowed_operations": [
"gitea.read",
"gitea.pr.create",
"gitea.branch.push",
"gitea.branch.delete",
],
"forbidden_operations": [], "forbidden_operations": [],
"audit_label": "test-deleter", "audit_label": "test-author-deleter",
} }
@patch("mcp_server.get_profile", return_value=DELETE_PROFILE) @patch("mcp_server.get_profile", return_value=DELETE_PROFILE)
+128 -12
View File
@@ -92,14 +92,19 @@ class TestMigrateProfiles(unittest.TestCase):
author = prgs_gitea["identities"]["author"] author = prgs_gitea["identities"]["author"]
self.assertEqual(author["username"], "jcwalker3") self.assertEqual(author["username"], "jcwalker3")
self.assertEqual(author["auth"]["id"], "redacted-author-ref") self.assertEqual(author["auth"]["id"], "redacted-author-ref")
self.assertEqual(author["allowed_operations"], ["read", "comment"]) self.assertEqual(
self.assertEqual(author["forbidden_operations"], ["approve", "merge"]) author["allowed_operations"], ["gitea.read", "gitea.pr.comment"]
)
self.assertEqual(
author["forbidden_operations"],
["gitea.pr.approve", "gitea.pr.merge"],
)
reviewer = prgs_gitea["identities"]["reviewer"] reviewer = prgs_gitea["identities"]["reviewer"]
self.assertEqual(reviewer["role"], "reviewer") self.assertEqual(reviewer["role"], "reviewer")
self.assertEqual(reviewer["username"], "sysadmin") self.assertEqual(reviewer["username"], "sysadmin")
self.assertEqual(reviewer["auth"]["id"], "redacted-reviewer-ref") 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): def test_alias_generation(self):
"""Test that aliases are correctly generated to support old profile names.""" """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()) self.assertNotIn("token", stdout_output.lower())
def test_explicit_operations_are_preserved(self): 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 = json.loads(json.dumps(self.v1_content))
v1_data["profiles"]["prgs-reviewer"]["allowed_operations"] = ["read"] v1_data["profiles"]["prgs-reviewer"]["allowed_operations"] = ["read"]
v1_data["profiles"]["prgs-reviewer"]["forbidden_operations"] = ["merge"] v1_data["profiles"]["prgs-reviewer"]["forbidden_operations"] = ["merge"]
@@ -198,8 +203,8 @@ class TestMigrateProfiles(unittest.TestCase):
v2_data["environments"]["prgs"]["services"]["gitea"] v2_data["environments"]["prgs"]["services"]["gitea"]
["identities"]["reviewer"] ["identities"]["reviewer"]
) )
self.assertEqual(reviewer["allowed_operations"], ["read"]) self.assertEqual(reviewer["allowed_operations"], ["gitea.read"])
self.assertEqual(reviewer["forbidden_operations"], ["merge"]) self.assertEqual(reviewer["forbidden_operations"], ["gitea.pr.merge"])
def test_inferred_role_defaults_only_when_unambiguous(self): def test_inferred_role_defaults_only_when_unambiguous(self):
"""Role defaults are allowed only for clear author/reviewer profiles.""" """Role defaults are allowed only for clear author/reviewer profiles."""
@@ -307,7 +312,10 @@ class TestMigrateProfiles(unittest.TestCase):
self.assertEqual(cm.exception.code, 1) self.assertEqual(cm.exception.code, 1)
def test_reconciler_profile_migration(self): def test_reconciler_profile_migration(self):
"""Verify that reconciler profiles with explicit operations migrate correctly.""" """Legacy reconciler shorthands migrate to valid canonical operations."""
import gitea_config
import reconciler_profile
v1_data = { v1_data = {
"version": 1, "version": 1,
"profiles": { "profiles": {
@@ -316,8 +324,22 @@ class TestMigrateProfiles(unittest.TestCase):
"username": "reconciler-agent", "username": "reconciler-agent",
"auth": {"type": "keychain", "id": "reconciler-ref"}, "auth": {"type": "keychain", "id": "reconciler-ref"},
"execution_profile": "prgs-reconciler", "execution_profile": "prgs-reconciler",
"allowed_operations": ["read", "pr.close", "gitea.branch.delete"], "allowed_operations": [
"forbidden_operations": ["merge", "approve"] "read",
"pr.close",
"pr.comment",
"issue.comment",
"issue.close",
"gitea.branch.delete",
],
"forbidden_operations": [
"merge",
"approve",
"review",
"pr.create",
"branch.push",
"commit",
],
} }
} }
} }
@@ -327,12 +349,35 @@ class TestMigrateProfiles(unittest.TestCase):
["identities"]["reconciler"] ["identities"]["reconciler"]
) )
self.assertEqual(reconciler["role"], "reconciler") self.assertEqual(reconciler["role"], "reconciler")
self.assertEqual(reconciler["allowed_operations"], ["read", "pr.close", "gitea.branch.delete"]) allowed = reconciler["allowed_operations"]
self.assertEqual(reconciler["forbidden_operations"], ["merge", "approve"]) 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") 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): def test_reconciler_profile_defaults(self):
"""Verify that reconciler profiles without explicit operations get defaults.""" """Reconciler defaults are fully canonical and loader-valid."""
import gitea_config
import reconciler_profile
v1_data = { v1_data = {
"version": 1, "version": 1,
"profiles": { "profiles": {
@@ -358,6 +403,77 @@ class TestMigrateProfiles(unittest.TestCase):
reconciler["forbidden_operations"], reconciler["forbidden_operations"],
migrate_profiles.RECONCILER_DEFAULT_FORBIDDEN, 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__": if __name__ == "__main__":