Compare commits
19
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
fe29379ad2 | ||
|
|
9a641ed4a5 | ||
|
|
8d1098f916 | ||
|
|
49aa3b2812 | ||
|
|
94004f05ac | ||
|
|
9b96085d8d | ||
|
|
ddb1dd941b | ||
|
|
d7c29afde5 | ||
|
|
7acd55d2d2 | ||
|
|
46709537c1 | ||
|
|
62ab4b4d95 | ||
|
|
d9cf26ddbf | ||
|
|
b71dfe9696 | ||
|
|
22d4735170 | ||
|
|
d57ca94ec5 | ||
|
|
09d8db5b9e | ||
|
|
3550262613 | ||
|
|
3cb05284b3 | ||
|
|
84b841c727 |
@@ -0,0 +1,98 @@
|
|||||||
|
"""Guards for merged-PR branch cleanup and raw git delete bypasses (#514)."""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import re
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
|
PROTECTED_BRANCHES = frozenset({"master", "main", "dev"})
|
||||||
|
|
||||||
|
_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),
|
||||||
|
re.compile(r"\bgit(?:\s+-C\s+\S+)?\s+push\b[^\n\r]*\s:[^\s`]+", re.I),
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def raw_branch_delete_commands(text: str | None) -> list[str]:
|
||||||
|
"""Return raw git branch-delete commands cited in *text*."""
|
||||||
|
if not text:
|
||||||
|
return []
|
||||||
|
commands: list[str] = []
|
||||||
|
for pattern in _RAW_BRANCH_DELETE_PATTERNS:
|
||||||
|
commands.extend(match.group(0).strip("` ") for match in pattern.finditer(text))
|
||||||
|
return list(dict.fromkeys(commands))
|
||||||
|
|
||||||
|
|
||||||
|
def assess_raw_branch_delete_report(text: str | None) -> dict[str, Any]:
|
||||||
|
"""Fail closed when a report uses raw git branch deletion as cleanup proof."""
|
||||||
|
commands = raw_branch_delete_commands(text)
|
||||||
|
reasons = [
|
||||||
|
(
|
||||||
|
"raw git branch deletion bypasses MCP branch.delete cleanup gates: "
|
||||||
|
f"{command}"
|
||||||
|
)
|
||||||
|
for command in commands
|
||||||
|
]
|
||||||
|
return {
|
||||||
|
"proven": not reasons,
|
||||||
|
"block": bool(reasons),
|
||||||
|
"commands": commands,
|
||||||
|
"reasons": reasons,
|
||||||
|
"safe_next_action": (
|
||||||
|
"use gitea_cleanup_merged_pr_branch or another approved cleanup "
|
||||||
|
"helper with explicit branch.delete capability"
|
||||||
|
if reasons
|
||||||
|
else "proceed"
|
||||||
|
),
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def assess_merged_pr_branch_cleanup(
|
||||||
|
*,
|
||||||
|
pr_number: int,
|
||||||
|
head_branch: str,
|
||||||
|
merged: bool,
|
||||||
|
remote_branch_exists: bool,
|
||||||
|
open_pr_heads: set[str],
|
||||||
|
head_on_target: bool | None,
|
||||||
|
delete_capability_allowed: bool,
|
||||||
|
confirmation: str | None,
|
||||||
|
protected_branches: frozenset[str] | None = None,
|
||||||
|
) -> dict[str, Any]:
|
||||||
|
"""Assess whether a merged PR source branch can be deleted via MCP."""
|
||||||
|
protected = protected_branches or PROTECTED_BRANCHES
|
||||||
|
expected_confirmation = f"CLEANUP MERGED PR {pr_number} BRANCH {head_branch}"
|
||||||
|
reasons: list[str] = []
|
||||||
|
if not merged:
|
||||||
|
reasons.append("PR is not merged")
|
||||||
|
if not remote_branch_exists:
|
||||||
|
reasons.append("remote branch already absent")
|
||||||
|
if not head_branch:
|
||||||
|
reasons.append("PR head branch is missing")
|
||||||
|
if head_branch in protected:
|
||||||
|
reasons.append(f"branch '{head_branch}' is protected")
|
||||||
|
if head_branch in open_pr_heads:
|
||||||
|
reasons.append("an open PR still references this head branch")
|
||||||
|
if head_on_target is False:
|
||||||
|
reasons.append("PR head is not an ancestor of the target branch")
|
||||||
|
if head_on_target is None:
|
||||||
|
reasons.append("PR head ancestry could not be proven")
|
||||||
|
if not delete_capability_allowed:
|
||||||
|
reasons.append("gitea.branch.delete capability is not allowed")
|
||||||
|
if confirmation != expected_confirmation:
|
||||||
|
reasons.append(
|
||||||
|
"confirmation must equal "
|
||||||
|
f"'{expected_confirmation}' for branch cleanup"
|
||||||
|
)
|
||||||
|
|
||||||
|
safe = not reasons
|
||||||
|
return {
|
||||||
|
"pr_number": pr_number,
|
||||||
|
"head_branch": head_branch,
|
||||||
|
"expected_confirmation": expected_confirmation,
|
||||||
|
"remote_branch_exists": remote_branch_exists,
|
||||||
|
"safe_to_delete": safe,
|
||||||
|
"block_reasons": reasons,
|
||||||
|
"recommended_action": "delete_remote_branch" if safe else "keep_remote_branch",
|
||||||
|
}
|
||||||
+38
-2
@@ -49,8 +49,9 @@ Optional environment variables:
|
|||||||
| `/api/audit` | JSON validator preview (POST `report_text`, optional `task_kind`) |
|
| `/api/audit` | JSON validator preview (POST `report_text`, optional `task_kind`) |
|
||||||
| `/worktrees` | Worktree hygiene dashboard (#432) |
|
| `/worktrees` | Worktree hygiene dashboard (#432) |
|
||||||
| `/api/worktrees` | JSON worktree scan with classifications and anomalies |
|
| `/api/worktrees` | JSON worktree scan with classifications and anomalies |
|
||||||
| `/leases` | Stub — lease visibility (#433) |
|
| `/actions` | Gated write-action registry — all disabled in MVP (#434) |
|
||||||
| `/worktrees` | Stub — hygiene dashboard (#432) |
|
| `/api/actions` | JSON action registry with capability metadata |
|
||||||
|
| `/api/actions/{id}/preview` | Mutation ledger preview (GET, read-only) |
|
||||||
| `/leases` | Lease and collision visibility (#433) |
|
| `/leases` | Lease and collision visibility (#433) |
|
||||||
| `/api/leases` | JSON lease/collision export |
|
| `/api/leases` | JSON lease/collision export |
|
||||||
|
|
||||||
@@ -97,6 +98,15 @@ credentials. The UI surfaces pagination proof (returned count, pages fetched,
|
|||||||
If credentials are missing or the fetch fails, the page shows an explicit error
|
If credentials are missing or the fetch fails, the page shows an explicit error
|
||||||
instead of an empty queue (fail closed).
|
instead of an empty queue (fail closed).
|
||||||
|
|
||||||
|
## Gated actions (#434)
|
||||||
|
|
||||||
|
`/actions` registers future write actions (claim, comment, review, merge,
|
||||||
|
delete branch, create PR/issue). Each entry declares the MCP tool, required
|
||||||
|
permission, and profile role from `task_capability_map.py` — aligned with
|
||||||
|
`gitea_resolve_task_capability`. Buttons are disabled; previews always render
|
||||||
|
a mutation ledger. Direct `attempt_action` calls fail closed without invoking
|
||||||
|
MCP tools.
|
||||||
|
|
||||||
## Worktree hygiene (#432)
|
## Worktree hygiene (#432)
|
||||||
|
|
||||||
`/worktrees` scans local `branches/` directories and registered git worktrees.
|
`/worktrees` scans local `branches/` directories and registered git worktrees.
|
||||||
@@ -107,6 +117,30 @@ includes a copy/paste canonical cleanup prompt only — no deletion actions.
|
|||||||
|
|
||||||
Override scan root with `WEBUI_REPO_ROOT` (defaults to repository root).
|
Override scan root with `WEBUI_REPO_ROOT` (defaults to repository root).
|
||||||
|
|
||||||
|
## Lease visibility (#433)
|
||||||
|
|
||||||
|
`/leases` surfaces read-only lease and collision state: local issue lock file,
|
||||||
|
in-progress claim inventory (#268), reviewer PR lease comments when present
|
||||||
|
(`<!-- mcp-review-lease:v1 -->`, #407), duplicate open PRs per issue (#400),
|
||||||
|
and duplicate local branches per issue. Links to collision-history backend
|
||||||
|
issues (#267, #268, #400, #407) are included. No lease acquire/release from UI.
|
||||||
|
|
||||||
|
## Runtime health (#430)
|
||||||
|
|
||||||
|
`/runtime` surfaces read-only MCP/runtime diagnostics for the default registry
|
||||||
|
project: active profile and role kind, authenticated identity (when credentials
|
||||||
|
are available), config model/mode, local vs remote `master` SHA sync, shell
|
||||||
|
health, workflow/schema SHA-256 hashes, and stale-runtime warnings when the
|
||||||
|
checkout is behind merged safety-gate changes. Restart guidance links to #420;
|
||||||
|
no tokens or MCP restart actions are exposed.
|
||||||
|
|
||||||
|
## Tests
|
||||||
|
|
||||||
|
```bash
|
||||||
|
pytest tests/test_webui_skeleton.py tests/test_webui_project_registry.py tests/test_webui_prompt_library.py tests/test_webui_queue_dashboard.py tests/test_webui_gated_actions.py -q
|
||||||
|
pytest tests/test_webui_skeleton.py tests/test_webui_project_registry.py tests/test_webui_prompt_library.py tests/test_webui_queue_dashboard.py tests/test_webui_audit.py tests/test_webui_worktree_hygiene.py -q
|
||||||
|
pytest tests/test_webui_skeleton.py tests/test_webui_project_registry.py tests/test_webui_prompt_library.py tests/test_webui_queue_dashboard.py tests/test_webui_lease_visibility.py tests/test_webui_runtime_health.py -q
|
||||||
|
|
||||||
## Tests (#436)
|
## Tests (#436)
|
||||||
|
|
||||||
Run the full hermetic web UI suite (all `test_webui_*.py` modules):
|
Run the full hermetic web UI suite (all `test_webui_*.py` modules):
|
||||||
@@ -132,6 +166,8 @@ Or invoke unittest directly:
|
|||||||
|
|
||||||
```bash
|
```bash
|
||||||
python3 -m unittest discover -s tests -p 'test_webui_*.py' -q
|
python3 -m unittest discover -s tests -p 'test_webui_*.py' -q
|
||||||
|
```
|
||||||
|
|
||||||
## Lease visibility (#433)
|
## Lease visibility (#433)
|
||||||
|
|
||||||
`/leases` surfaces read-only lease and collision state: local issue lock file,
|
`/leases` surfaces read-only lease and collision state: local issue lock file,
|
||||||
|
|||||||
@@ -11,6 +11,7 @@ import inspect
|
|||||||
import re
|
import re
|
||||||
from typing import Any, Callable
|
from typing import Any, Callable
|
||||||
|
|
||||||
|
import branch_cleanup_guard
|
||||||
import issue_acceptance_gate
|
import issue_acceptance_gate
|
||||||
import issue_lock_provenance
|
import issue_lock_provenance
|
||||||
import reviewer_handoff_consistency
|
import reviewer_handoff_consistency
|
||||||
@@ -1294,6 +1295,17 @@ def _rule_shared_author_reviewer_same_run(report_text: str) -> list[dict[str, st
|
|||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def _rule_shared_raw_branch_delete_bypass(report_text: str) -> list[dict[str, str]]:
|
||||||
|
result = branch_cleanup_guard.assess_raw_branch_delete_report(report_text)
|
||||||
|
if not result["block"]:
|
||||||
|
return []
|
||||||
|
return _findings_from_reasons(
|
||||||
|
"shared.raw_branch_delete_bypass",
|
||||||
|
result["reasons"],
|
||||||
|
field="Cleanup mutations",
|
||||||
|
severity="block",
|
||||||
|
safe_next_action=result["safe_next_action"],
|
||||||
|
)
|
||||||
def _rule_reviewer_workflow_load_boundary(report_text: str) -> list[dict[str, str]]:
|
def _rule_reviewer_workflow_load_boundary(report_text: str) -> list[dict[str, str]]:
|
||||||
"""#403: require structured workflow-load helper result, not file-view narrative."""
|
"""#403: require structured workflow-load helper result, not file-view narrative."""
|
||||||
if not report_text.strip():
|
if not report_text.strip():
|
||||||
@@ -1448,6 +1460,7 @@ _SHARED_ISSUE_LOCK_RULES = (
|
|||||||
_rule_shared_issue_lock_external_state,
|
_rule_shared_issue_lock_external_state,
|
||||||
_rule_shared_manual_lock_pr_override,
|
_rule_shared_manual_lock_pr_override,
|
||||||
_rule_shared_author_reviewer_same_run,
|
_rule_shared_author_reviewer_same_run,
|
||||||
|
_rule_shared_raw_branch_delete_bypass,
|
||||||
_rule_shared_canonical_state_update,
|
_rule_shared_canonical_state_update,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|||||||
@@ -832,6 +832,7 @@ import pr_work_lease # noqa: E402
|
|||||||
import workflow_skill # noqa: E402
|
import workflow_skill # noqa: E402
|
||||||
import conflict_fix_classification # noqa: E402
|
import conflict_fix_classification # noqa: E402
|
||||||
import native_mcp_preference # noqa: E402
|
import native_mcp_preference # noqa: E402
|
||||||
|
import branch_cleanup_guard # noqa: E402
|
||||||
import thread_state_ledger_validator # noqa: E402
|
import thread_state_ledger_validator # noqa: E402
|
||||||
import master_parity_gate # noqa: E402
|
import master_parity_gate # noqa: E402
|
||||||
|
|
||||||
@@ -4852,6 +4853,149 @@ def gitea_delete_branch(
|
|||||||
return {"success": True, "message": f"Remote branch '{branch}' deleted."}
|
return {"success": True, "message": f"Remote branch '{branch}' deleted."}
|
||||||
|
|
||||||
|
|
||||||
|
@mcp.tool()
|
||||||
|
def gitea_cleanup_merged_pr_branch(
|
||||||
|
pr_number: int,
|
||||||
|
confirmation: str,
|
||||||
|
branch: str | None = None,
|
||||||
|
remote: str = "dadeschools",
|
||||||
|
host: str | None = None,
|
||||||
|
org: str | None = None,
|
||||||
|
repo: str | None = None,
|
||||||
|
worktree_path: str | None = None,
|
||||||
|
) -> dict:
|
||||||
|
"""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"),
|
||||||
|
}
|
||||||
|
|
||||||
|
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)"
|
||||||
|
],
|
||||||
|
"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": [
|
||||||
|
"merged branch cleanup requires an explicit branches/ worktree "
|
||||||
|
"path; root checkout branch ref mutation is blocked (fail closed)"
|
||||||
|
],
|
||||||
|
}
|
||||||
|
|
||||||
|
verify_preflight_purity(
|
||||||
|
remote,
|
||||||
|
worktree_path=worktree_path,
|
||||||
|
task="cleanup_merged_pr_branch",
|
||||||
|
)
|
||||||
|
|
||||||
|
h, o, r = _resolve(remote, host, org, repo)
|
||||||
|
auth = _auth(h)
|
||||||
|
base = repo_api_url(h, o, r)
|
||||||
|
pr = api_request("GET", f"{base}/pulls/{pr_number}", auth)
|
||||||
|
pr_head = pr.get("head") or {}
|
||||||
|
head_branch = branch or pr_head.get("ref") or ""
|
||||||
|
head_sha = pr_head.get("sha")
|
||||||
|
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": [
|
||||||
|
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 = {
|
||||||
|
str((open_pr.get("head") or {}).get("ref"))
|
||||||
|
for open_pr in open_prs
|
||||||
|
if (open_pr.get("head") or {}).get("ref")
|
||||||
|
}
|
||||||
|
remote_exists = _remote_branch_exists(h, o, r, auth, head_branch)
|
||||||
|
target_ref = (
|
||||||
|
f"{remote}/{target_branch}"
|
||||||
|
if remote in REMOTES
|
||||||
|
else f"origin/{target_branch}"
|
||||||
|
)
|
||||||
|
head_on_target = merged_cleanup_reconcile.is_head_ancestor_of_ref(
|
||||||
|
PROJECT_ROOT,
|
||||||
|
head_sha,
|
||||||
|
target_ref,
|
||||||
|
)
|
||||||
|
assessment = branch_cleanup_guard.assess_merged_pr_branch_cleanup(
|
||||||
|
pr_number=pr_number,
|
||||||
|
head_branch=head_branch,
|
||||||
|
merged=bool(pr.get("merged") or pr.get("merged_at")),
|
||||||
|
remote_branch_exists=remote_exists,
|
||||||
|
open_pr_heads=open_heads,
|
||||||
|
head_on_target=head_on_target,
|
||||||
|
delete_capability_allowed=True,
|
||||||
|
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"],
|
||||||
|
}
|
||||||
|
|
||||||
|
import urllib.parse
|
||||||
|
|
||||||
|
encoded_branch = urllib.parse.quote(head_branch, safe="")
|
||||||
|
url = f"{base}/branches/{encoded_branch}"
|
||||||
|
with _audited(
|
||||||
|
"cleanup_merged_pr_branch",
|
||||||
|
host=h,
|
||||||
|
remote=remote,
|
||||||
|
org=o,
|
||||||
|
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",
|
||||||
|
},
|
||||||
|
):
|
||||||
|
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,
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
def _remote_branch_exists(h: str, o: str, r: str, auth: str, branch: str) -> bool:
|
def _remote_branch_exists(h: str, o: str, r: str, auth: str, branch: str) -> bool:
|
||||||
import urllib.parse
|
import urllib.parse
|
||||||
|
|
||||||
|
|||||||
+21
-11
@@ -190,18 +190,28 @@ def find_active_reviewer_lease(
|
|||||||
pr_number: int,
|
pr_number: int,
|
||||||
now: datetime | None = None,
|
now: datetime | None = None,
|
||||||
) -> dict[str, Any] | None:
|
) -> dict[str, Any] | None:
|
||||||
"""Newest non-terminal, unexpired lease for *pr_number*."""
|
"""Return the newest unexpired non-terminal lease for *pr_number*.
|
||||||
|
|
||||||
|
Append-only lease markers form a ledger: the **newest** marker is
|
||||||
|
authoritative. A later terminal phase (``released`` / ``done`` /
|
||||||
|
``blocked``) ends the lease even when older ``claimed`` markers remain
|
||||||
|
on the thread (#577). Skipping only terminal markers and walking older
|
||||||
|
claims incorrectly re-arms a lease after a successful release.
|
||||||
|
"""
|
||||||
now = now or datetime.now(timezone.utc)
|
now = now or datetime.now(timezone.utc)
|
||||||
for lease in reversed(_lease_entries(comments, pr_number=pr_number)):
|
entries = list(reversed(_lease_entries(comments, pr_number=pr_number)))
|
||||||
phase = (lease.get("phase") or "").strip().lower()
|
if not entries:
|
||||||
if phase in _TERMINAL_PHASES:
|
return None
|
||||||
continue
|
newest = entries[0]
|
||||||
if _lease_expired(lease, now=now):
|
phase = (newest.get("phase") or "").strip().lower()
|
||||||
continue
|
if phase in _TERMINAL_PHASES:
|
||||||
if phase in _ACTIVE_PHASES or phase:
|
return None
|
||||||
lease = dict(lease)
|
if _lease_expired(newest, now=now):
|
||||||
lease["freshness"] = classify_lease_freshness(lease, now=now)
|
return None
|
||||||
return lease
|
if phase in _ACTIVE_PHASES or phase:
|
||||||
|
lease = dict(newest)
|
||||||
|
lease["freshness"] = classify_lease_freshness(lease, now=now)
|
||||||
|
return lease
|
||||||
return None
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -80,6 +80,7 @@ AUTHOR_TASKS = frozenset({
|
|||||||
})
|
})
|
||||||
|
|
||||||
RECONCILER_TASKS = frozenset({
|
RECONCILER_TASKS = frozenset({
|
||||||
|
"cleanup_merged_pr_branch",
|
||||||
"reconcile_already_landed_pr",
|
"reconcile_already_landed_pr",
|
||||||
"reconcile_already_landed",
|
"reconcile_already_landed",
|
||||||
"reconcile-landed-pr",
|
"reconcile-landed-pr",
|
||||||
@@ -109,6 +110,7 @@ TASK_REQUIRED_ROLE = {
|
|||||||
"reconcile_already_landed_pr": "reconciler",
|
"reconcile_already_landed_pr": "reconciler",
|
||||||
"reconcile_already_landed": "reconciler",
|
"reconcile_already_landed": "reconciler",
|
||||||
"reconcile-landed-pr": "reconciler",
|
"reconcile-landed-pr": "reconciler",
|
||||||
|
"cleanup_merged_pr_branch": "reconciler",
|
||||||
# #309: reconciler tasks close already-landed PRs/issues only.
|
# #309: reconciler tasks close already-landed PRs/issues only.
|
||||||
"reconcile_close_landed_pr": "reconciler",
|
"reconcile_close_landed_pr": "reconciler",
|
||||||
"reconcile_close_landed_issue": "reconciler",
|
"reconcile_close_landed_issue": "reconciler",
|
||||||
@@ -406,4 +408,4 @@ def assess_infra_stop(project_root: str | None = None) -> dict:
|
|||||||
|
|
||||||
def check_mid_merge(project_root: str | None = None) -> bool:
|
def check_mid_merge(project_root: str | None = None) -> bool:
|
||||||
"""Return True if the repository is mid-merge, mid-rebase, or has conflict markers."""
|
"""Return True if the repository is mid-merge, mid-rebase, or has conflict markers."""
|
||||||
return assess_infra_stop(project_root)["infra_stop"]
|
return assess_infra_stop(project_root)["infra_stop"]
|
||||||
|
|||||||
@@ -936,6 +936,14 @@ Confirm:
|
|||||||
|
|
||||||
Clean only the session-owned `branches/` review worktree if the project workflow explicitly allows cleanup.
|
Clean only the session-owned `branches/` review worktree if the project workflow explicitly allows cleanup.
|
||||||
|
|
||||||
|
Do not delete source branches from reviewer mode with raw git commands. Commands
|
||||||
|
such as `git branch -d <branch>` and `git push <remote> --delete <branch>` are
|
||||||
|
not proof of authorized cleanup and bypass MCP role/capability gates. Merged PR
|
||||||
|
source branch cleanup must use an explicit MCP cleanup tool such as
|
||||||
|
`gitea_cleanup_merged_pr_branch` or another approved cleanup helper with exact
|
||||||
|
`gitea.branch.delete` authority, merged-PR proof, no open PR using the branch,
|
||||||
|
target-branch ancestry proof, a `branches/` worktree path, and cleanup mutations
|
||||||
|
reported separately from review mutations.
|
||||||
Review, baseline, and merge-simulation worktrees created during this run are
|
Review, baseline, and merge-simulation worktrees created during this run are
|
||||||
transient and are removed automatically at successful completion once they are
|
transient and are removed automatically at successful completion once they are
|
||||||
clean, carry no open PR, and hold no active lease (#401). Use
|
clean, carry no open PR, and hold no active lease (#401). Use
|
||||||
|
|||||||
@@ -100,6 +100,10 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
|
|||||||
"permission": "gitea.branch.delete",
|
"permission": "gitea.branch.delete",
|
||||||
"role": "author",
|
"role": "author",
|
||||||
},
|
},
|
||||||
|
"cleanup_merged_pr_branch": {
|
||||||
|
"permission": "gitea.branch.delete",
|
||||||
|
"role": "reconciler",
|
||||||
|
},
|
||||||
"commit_files": {
|
"commit_files": {
|
||||||
"permission": "gitea.repo.commit",
|
"permission": "gitea.repo.commit",
|
||||||
"role": "author",
|
"role": "author",
|
||||||
|
|||||||
@@ -0,0 +1,187 @@
|
|||||||
|
import sys
|
||||||
|
import unittest
|
||||||
|
from unittest.mock import patch
|
||||||
|
|
||||||
|
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
|
||||||
|
|
||||||
|
import branch_cleanup_guard as guard # noqa: E402
|
||||||
|
import mcp_server # noqa: E402
|
||||||
|
import task_capability_map # noqa: E402
|
||||||
|
from final_report_validator import assess_final_report_validator # noqa: E402
|
||||||
|
from mcp_server import gitea_cleanup_merged_pr_branch # noqa: E402
|
||||||
|
|
||||||
|
FAKE_AUTH = "token fake"
|
||||||
|
|
||||||
|
|
||||||
|
class TestRawBranchDeleteGuard(unittest.TestCase):
|
||||||
|
def test_detects_local_and_remote_raw_git_delete_commands(self):
|
||||||
|
text = """
|
||||||
|
Cleanup mutations:
|
||||||
|
- git branch -d feat/issue-485-lease-comments-non-list-guard
|
||||||
|
- git push prgs --delete feat/issue-485-lease-comments-non-list-guard
|
||||||
|
"""
|
||||||
|
commands = guard.raw_branch_delete_commands(text)
|
||||||
|
self.assertEqual(len(commands), 2)
|
||||||
|
self.assertIn("git branch -d", commands[0])
|
||||||
|
self.assertIn("git push prgs --delete", commands[1])
|
||||||
|
|
||||||
|
def test_final_report_blocks_raw_delete_as_cleanup_proof(self):
|
||||||
|
report = """
|
||||||
|
## Controller Handoff
|
||||||
|
- Task: review_pr
|
||||||
|
- Cleanup mutations: git push prgs --delete feat/branch
|
||||||
|
"""
|
||||||
|
result = assess_final_report_validator(report, "review_pr")
|
||||||
|
self.assertTrue(result["blocked"])
|
||||||
|
self.assertTrue(
|
||||||
|
any("shared.raw_branch_delete_bypass" in r for r in result["reasons"])
|
||||||
|
)
|
||||||
|
|
||||||
|
def test_cleanup_task_requires_branch_delete_capability(self):
|
||||||
|
self.assertEqual(
|
||||||
|
task_capability_map.required_permission("cleanup_merged_pr_branch"),
|
||||||
|
"gitea.branch.delete",
|
||||||
|
)
|
||||||
|
self.assertEqual(
|
||||||
|
task_capability_map.required_role("cleanup_merged_pr_branch"),
|
||||||
|
"reconciler",
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
class TestMergedPrBranchCleanupTool(unittest.TestCase):
|
||||||
|
def setUp(self):
|
||||||
|
self._remotes = patch.dict(
|
||||||
|
mcp_server.REMOTES,
|
||||||
|
{
|
||||||
|
"prgs": {
|
||||||
|
"host": "gitea.example.com",
|
||||||
|
"org": "Example-Org",
|
||||||
|
"repo": "Example-Repo",
|
||||||
|
}
|
||||||
|
},
|
||||||
|
)
|
||||||
|
self._remotes.start()
|
||||||
|
patch("gitea_audit.audit_enabled", return_value=False).start()
|
||||||
|
self.mock_api = patch("mcp_server.api_request").start()
|
||||||
|
self.mock_all = patch("mcp_server.api_get_all", return_value=[]).start()
|
||||||
|
patch("mcp_server.get_auth_header", return_value=FAKE_AUTH).start()
|
||||||
|
patch(
|
||||||
|
"mcp_server.merged_cleanup_reconcile.is_head_ancestor_of_ref",
|
||||||
|
return_value=True,
|
||||||
|
).start()
|
||||||
|
|
||||||
|
def tearDown(self):
|
||||||
|
patch.stopall()
|
||||||
|
|
||||||
|
def test_reviewer_without_delete_authority_fails_closed(self):
|
||||||
|
patch(
|
||||||
|
"mcp_server.get_profile",
|
||||||
|
return_value={
|
||||||
|
"profile_name": "reviewer",
|
||||||
|
"allowed_operations": ["gitea.read", "gitea.pr.review"],
|
||||||
|
"forbidden_operations": ["gitea.branch.delete"],
|
||||||
|
},
|
||||||
|
).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["required_permission"], "gitea.branch.delete")
|
||||||
|
self.mock_api.assert_not_called()
|
||||||
|
|
||||||
|
def test_root_checkout_cleanup_fails_closed(self):
|
||||||
|
patch(
|
||||||
|
"mcp_server.get_profile",
|
||||||
|
return_value={
|
||||||
|
"profile_name": "branch-cleanup",
|
||||||
|
"allowed_operations": ["gitea.read", "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/Gitea-Tools",
|
||||||
|
)
|
||||||
|
self.assertFalse(res["performed"])
|
||||||
|
self.assertIn("root checkout", " ".join(res["reasons"]))
|
||||||
|
self.mock_api.assert_not_called()
|
||||||
|
|
||||||
|
def test_authorized_cleanup_deletes_through_api_path(self):
|
||||||
|
branch = "feat/issue-485-lease-comments-non-list-guard"
|
||||||
|
patch(
|
||||||
|
"mcp_server.get_profile",
|
||||||
|
return_value={
|
||||||
|
"profile_name": "branch-cleanup",
|
||||||
|
"allowed_operations": ["gitea.read", "gitea.branch.delete"],
|
||||||
|
"forbidden_operations": [],
|
||||||
|
},
|
||||||
|
).start()
|
||||||
|
self.mock_api.side_effect = [
|
||||||
|
{
|
||||||
|
"number": 487,
|
||||||
|
"merged": True,
|
||||||
|
"merged_at": "2026-07-08T01:00:00Z",
|
||||||
|
"head": {"ref": branch, "sha": "a" * 40},
|
||||||
|
"base": {"ref": "master"},
|
||||||
|
},
|
||||||
|
{},
|
||||||
|
{},
|
||||||
|
]
|
||||||
|
res = gitea_cleanup_merged_pr_branch(
|
||||||
|
pr_number=487,
|
||||||
|
confirmation=f"CLEANUP MERGED PR 487 BRANCH {branch}",
|
||||||
|
branch=branch,
|
||||||
|
remote="prgs",
|
||||||
|
worktree_path="/tmp/repo/branches/cleanup",
|
||||||
|
)
|
||||||
|
self.assertTrue(res["performed"])
|
||||||
|
delete_calls = [
|
||||||
|
call for call in self.mock_api.call_args_list if call.args[0] == "DELETE"
|
||||||
|
]
|
||||||
|
self.assertEqual(len(delete_calls), 1)
|
||||||
|
self.assertIn("branches/feat%2Fissue-485", delete_calls[0].args[1])
|
||||||
|
|
||||||
|
def test_confirmation_required_before_delete(self):
|
||||||
|
branch = "feat/issue-485-lease-comments-non-list-guard"
|
||||||
|
patch(
|
||||||
|
"mcp_server.get_profile",
|
||||||
|
return_value={
|
||||||
|
"profile_name": "branch-cleanup",
|
||||||
|
"allowed_operations": ["gitea.read", "gitea.branch.delete"],
|
||||||
|
"forbidden_operations": [],
|
||||||
|
},
|
||||||
|
).start()
|
||||||
|
self.mock_api.side_effect = [
|
||||||
|
{
|
||||||
|
"number": 487,
|
||||||
|
"merged": True,
|
||||||
|
"merged_at": "2026-07-08T01:00:00Z",
|
||||||
|
"head": {"ref": branch, "sha": "a" * 40},
|
||||||
|
"base": {"ref": "master"},
|
||||||
|
},
|
||||||
|
{},
|
||||||
|
]
|
||||||
|
res = gitea_cleanup_merged_pr_branch(
|
||||||
|
pr_number=487,
|
||||||
|
confirmation="wrong",
|
||||||
|
branch=branch,
|
||||||
|
remote="prgs",
|
||||||
|
worktree_path="/tmp/repo/branches/cleanup",
|
||||||
|
)
|
||||||
|
self.assertFalse(res["performed"])
|
||||||
|
self.assertIn("confirmation", " ".join(res["reasons"]))
|
||||||
|
delete_calls = [
|
||||||
|
call for call in self.mock_api.call_args_list if call.args[0] == "DELETE"
|
||||||
|
]
|
||||||
|
self.assertFalse(delete_calls)
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
unittest.main()
|
||||||
@@ -98,6 +98,45 @@ class TestReviewerLeaseFreshness(unittest.TestCase):
|
|||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
class TestReviewerLeaseNewestWins(unittest.TestCase):
|
||||||
|
"""#577: terminal release must supersede older claimed markers."""
|
||||||
|
|
||||||
|
def test_released_after_claimed_has_no_active_lease(self):
|
||||||
|
claim = _lease_comment(516, "session-a", phase="claimed")
|
||||||
|
claim["id"] = 1
|
||||||
|
release = _lease_comment(516, "session-a", phase="released")
|
||||||
|
release["id"] = 2
|
||||||
|
active = leases.find_active_reviewer_lease(
|
||||||
|
[claim, release], pr_number=516
|
||||||
|
)
|
||||||
|
self.assertIsNone(active)
|
||||||
|
|
||||||
|
def test_new_claim_after_release_is_active(self):
|
||||||
|
claim = _lease_comment(516, "session-a", phase="claimed")
|
||||||
|
claim["id"] = 1
|
||||||
|
release = _lease_comment(516, "session-a", phase="released")
|
||||||
|
release["id"] = 2
|
||||||
|
reclaim = _lease_comment(516, "session-b", phase="claimed")
|
||||||
|
reclaim["id"] = 3
|
||||||
|
active = leases.find_active_reviewer_lease(
|
||||||
|
[claim, release, reclaim], pr_number=516
|
||||||
|
)
|
||||||
|
self.assertIsNotNone(active)
|
||||||
|
self.assertEqual(active.get("session_id"), "session-b")
|
||||||
|
self.assertEqual(active.get("phase"), "claimed")
|
||||||
|
self.assertEqual(active.get("comment_id"), 3)
|
||||||
|
|
||||||
|
def test_done_after_claimed_has_no_active_lease(self):
|
||||||
|
claim = _lease_comment(516, "session-a", phase="claimed")
|
||||||
|
claim["id"] = 10
|
||||||
|
done = _lease_comment(516, "session-a", phase="done")
|
||||||
|
done["id"] = 11
|
||||||
|
active = leases.find_active_reviewer_lease(
|
||||||
|
[claim, done], pr_number=516
|
||||||
|
)
|
||||||
|
self.assertIsNone(active)
|
||||||
|
|
||||||
|
|
||||||
class TestReviewerLeaseMutationGate(unittest.TestCase):
|
class TestReviewerLeaseMutationGate(unittest.TestCase):
|
||||||
def setUp(self):
|
def setUp(self):
|
||||||
leases.clear_session_lease()
|
leases.clear_session_lease()
|
||||||
|
|||||||
@@ -0,0 +1,128 @@
|
|||||||
|
"""Tests for web UI gated action framework (#434)."""
|
||||||
|
import sys
|
||||||
|
import unittest
|
||||||
|
from pathlib import Path
|
||||||
|
from unittest.mock import patch
|
||||||
|
|
||||||
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||||
|
|
||||||
|
from starlette.testclient import TestClient
|
||||||
|
|
||||||
|
from task_capability_map import TASK_CAPABILITY_MAP
|
||||||
|
from webui.app import create_app
|
||||||
|
from webui.gated_actions import (
|
||||||
|
attempt_action,
|
||||||
|
build_action_registry,
|
||||||
|
load_action_registry,
|
||||||
|
preview_action,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
class TestGatedActionRegistry(unittest.TestCase):
|
||||||
|
def test_all_actions_disabled_by_default(self):
|
||||||
|
registry = build_action_registry()
|
||||||
|
self.assertGreater(len(registry.actions), 0)
|
||||||
|
for action in registry.actions:
|
||||||
|
with self.subTest(action=action.action_id):
|
||||||
|
self.assertFalse(action.enabled)
|
||||||
|
|
||||||
|
def test_actions_align_with_task_capability_map(self):
|
||||||
|
registry = build_action_registry()
|
||||||
|
for action in registry.actions:
|
||||||
|
with self.subTest(task=action.task_key):
|
||||||
|
self.assertIn(action.task_key, TASK_CAPABILITY_MAP)
|
||||||
|
self.assertEqual(
|
||||||
|
action.required_permission,
|
||||||
|
TASK_CAPABILITY_MAP[action.task_key]["permission"],
|
||||||
|
)
|
||||||
|
self.assertEqual(
|
||||||
|
action.required_role,
|
||||||
|
TASK_CAPABILITY_MAP[action.task_key]["role"],
|
||||||
|
)
|
||||||
|
|
||||||
|
def test_preview_includes_mutation_ledger(self):
|
||||||
|
preview = preview_action("merge_pr", pr_number=42)
|
||||||
|
self.assertNotIn("error", preview)
|
||||||
|
ledger = preview["mutation_ledger"]
|
||||||
|
self.assertEqual(len(ledger), 1)
|
||||||
|
self.assertEqual(ledger[0]["mcp_tool"], "gitea_merge_pr")
|
||||||
|
self.assertEqual(ledger[0]["permission"], "gitea.pr.merge")
|
||||||
|
self.assertIn("42", ledger[0]["target"])
|
||||||
|
|
||||||
|
def test_attempt_fails_closed_without_mcp_calls(self):
|
||||||
|
registry = build_action_registry()
|
||||||
|
with patch("webui.gated_actions._MVP_ACTIONS_ENABLED", False):
|
||||||
|
for action in registry.actions:
|
||||||
|
with self.subTest(action=action.action_id):
|
||||||
|
result = attempt_action(action.action_id, issue_number=1)
|
||||||
|
self.assertFalse(result["success"])
|
||||||
|
self.assertEqual(result["error"], "action_disabled")
|
||||||
|
self.assertIn("preview", result)
|
||||||
|
|
||||||
|
def test_attempt_module_has_no_mcp_imports(self):
|
||||||
|
"""Ungated execution path must not import MCP server tools."""
|
||||||
|
import webui.gated_actions as ga
|
||||||
|
|
||||||
|
source_path = Path(ga.__file__).resolve()
|
||||||
|
source = source_path.read_text(encoding="utf-8")
|
||||||
|
self.assertNotIn("mcp_server", source)
|
||||||
|
result = attempt_action("merge_pr", pr_number=99)
|
||||||
|
self.assertFalse(result["success"])
|
||||||
|
|
||||||
|
def test_unknown_action_fails_closed(self):
|
||||||
|
result = attempt_action("nonexistent_action")
|
||||||
|
self.assertFalse(result["success"])
|
||||||
|
self.assertEqual(result["error"], "unknown_action")
|
||||||
|
|
||||||
|
|
||||||
|
class TestGatedActionRoutes(unittest.TestCase):
|
||||||
|
def setUp(self):
|
||||||
|
self.client = TestClient(create_app())
|
||||||
|
|
||||||
|
def test_actions_page_renders_disabled_buttons(self):
|
||||||
|
response = self.client.get("/actions")
|
||||||
|
self.assertEqual(response.status_code, 200)
|
||||||
|
self.assertIn("Gated actions", response.text)
|
||||||
|
self.assertIn("disabled", response.text.lower())
|
||||||
|
self.assertIn("mutation ledger", response.text.lower())
|
||||||
|
self.assertIn("aria-disabled", response.text)
|
||||||
|
self.assertIn(" disabled", response.text)
|
||||||
|
|
||||||
|
def test_api_actions_registry(self):
|
||||||
|
response = self.client.get("/api/actions")
|
||||||
|
self.assertEqual(response.status_code, 200)
|
||||||
|
data = response.json()
|
||||||
|
self.assertFalse(data["mvp_actions_enabled"])
|
||||||
|
action_ids = {item["action_id"] for item in data["actions"]}
|
||||||
|
self.assertIn("merge_pr", action_ids)
|
||||||
|
self.assertIn("claim_issue", action_ids)
|
||||||
|
for item in data["actions"]:
|
||||||
|
self.assertFalse(item["enabled"])
|
||||||
|
|
||||||
|
def test_api_action_preview_json(self):
|
||||||
|
response = self.client.get(
|
||||||
|
"/api/actions/review_pr/preview?pr_number=12"
|
||||||
|
)
|
||||||
|
self.assertEqual(response.status_code, 200)
|
||||||
|
data = response.json()
|
||||||
|
self.assertEqual(data["task_key"], "review_pr")
|
||||||
|
self.assertEqual(data["required_role"], "reviewer")
|
||||||
|
self.assertEqual(len(data["mutation_ledger"]), 1)
|
||||||
|
|
||||||
|
def test_post_attempt_fails_closed(self):
|
||||||
|
response = self.client.post(
|
||||||
|
"/api/actions/merge_pr/attempt",
|
||||||
|
json={"pr_number": 1},
|
||||||
|
)
|
||||||
|
self.assertEqual(response.status_code, 403)
|
||||||
|
data = response.json()
|
||||||
|
self.assertFalse(data["success"])
|
||||||
|
self.assertEqual(data["error"], "action_disabled")
|
||||||
|
|
||||||
|
def test_nav_includes_actions_link(self):
|
||||||
|
text = self.client.get("/").text
|
||||||
|
self.assertIn('href="/actions"', text)
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
unittest.main()
|
||||||
@@ -61,6 +61,11 @@ class TestWebuiSkeleton(unittest.TestCase):
|
|||||||
self.assertIn("Collision warnings", response.text)
|
self.assertIn("Collision warnings", response.text)
|
||||||
|
|
||||||
|
|
||||||
|
def test_actions_is_implemented(self):
|
||||||
|
response = self.client.get("/actions")
|
||||||
|
self.assertEqual(response.status_code, 200)
|
||||||
|
self.assertIn("Gated actions", response.text)
|
||||||
|
|
||||||
def test_post_is_rejected(self):
|
def test_post_is_rejected(self):
|
||||||
response = self.client.post("/health")
|
response = self.client.post("/health")
|
||||||
self.assertEqual(response.status_code, 405)
|
self.assertEqual(response.status_code, 405)
|
||||||
@@ -72,8 +77,8 @@ class TestWebuiSkeleton(unittest.TestCase):
|
|||||||
self.assertIn("Live queue", response.text)
|
self.assertIn("Live queue", response.text)
|
||||||
|
|
||||||
def test_nav_links_on_all_pages(self):
|
def test_nav_links_on_all_pages(self):
|
||||||
paths = ("/", "/queue", "/projects", "/prompts", "/runtime", "/audit", "/worktrees", "/leases")
|
paths = ("/", "/queue", "/projects", "/prompts", "/runtime", "/audit", "/worktrees", "/leases", "/actions")
|
||||||
hrefs = ("/queue", "/projects", "/prompts", "/runtime", "/audit", "/worktrees", "/leases")
|
hrefs = ("/queue", "/projects", "/prompts", "/runtime", "/audit", "/worktrees", "/leases", "/actions")
|
||||||
for path in paths:
|
for path in paths:
|
||||||
with self.subTest(path=path):
|
with self.subTest(path=path):
|
||||||
text = self.client.get(path).text
|
text = self.client.get(path).text
|
||||||
|
|||||||
@@ -16,6 +16,8 @@ from webui.prompt_library import find_prompt, library_to_dict
|
|||||||
from webui.prompt_views import render_prompt_detail, render_prompts_page
|
from webui.prompt_views import render_prompt_detail, render_prompts_page
|
||||||
from final_report_validator import FINAL_REPORT_TASK_KINDS
|
from final_report_validator import FINAL_REPORT_TASK_KINDS
|
||||||
|
|
||||||
|
from webui.gated_actions import attempt_action, load_action_registry, preview_action
|
||||||
|
from webui.gated_action_views import render_actions_page
|
||||||
from webui.audit_validator import audit_report, audit_to_dict
|
from webui.audit_validator import audit_report, audit_to_dict
|
||||||
from webui.audit_views import render_audit_page
|
from webui.audit_views import render_audit_page
|
||||||
from webui.lease_loader import load_lease_snapshot, snapshot_to_dict as lease_snapshot_to_dict
|
from webui.lease_loader import load_lease_snapshot, snapshot_to_dict as lease_snapshot_to_dict
|
||||||
@@ -52,6 +54,7 @@ async def home(_request: Request) -> HTMLResponse:
|
|||||||
"<li><strong>Audit</strong> — final-report paste and validator preview (#431)</li>"
|
"<li><strong>Audit</strong> — final-report paste and validator preview (#431)</li>"
|
||||||
"<li><strong>Worktrees</strong> — branch hygiene dashboard (#432)</li>"
|
"<li><strong>Worktrees</strong> — branch hygiene dashboard (#432)</li>"
|
||||||
"<li><strong>Leases</strong> — collision and lease visibility (#433)</li>"
|
"<li><strong>Leases</strong> — collision and lease visibility (#433)</li>"
|
||||||
|
"<li><strong>Actions</strong> — gated write-action framework (#434)</li>"
|
||||||
"</ul>"
|
"</ul>"
|
||||||
)
|
)
|
||||||
return HTMLResponse(render_page(title="Home", body_html=body))
|
return HTMLResponse(render_page(title="Home", body_html=body))
|
||||||
@@ -200,6 +203,41 @@ async def api_leases(_request: Request) -> JSONResponse:
|
|||||||
return JSONResponse(lease_snapshot_to_dict(load_lease_snapshot()))
|
return JSONResponse(lease_snapshot_to_dict(load_lease_snapshot()))
|
||||||
|
|
||||||
|
|
||||||
|
async def actions(_request: Request) -> HTMLResponse:
|
||||||
|
registry = load_action_registry()
|
||||||
|
return HTMLResponse(render_actions_page(registry))
|
||||||
|
|
||||||
|
|
||||||
|
async def api_actions(_request: Request) -> JSONResponse:
|
||||||
|
return JSONResponse(load_action_registry().to_dict())
|
||||||
|
|
||||||
|
|
||||||
|
async def api_action_preview(request: Request) -> JSONResponse:
|
||||||
|
action_id = request.path_params["action_id"]
|
||||||
|
params = dict(request.query_params)
|
||||||
|
for key in ("issue_number", "pr_number"):
|
||||||
|
if key in params:
|
||||||
|
params[key] = int(params[key])
|
||||||
|
result = preview_action(action_id, **params)
|
||||||
|
if "error" in result:
|
||||||
|
return JSONResponse(result, status_code=404)
|
||||||
|
return JSONResponse(result)
|
||||||
|
|
||||||
|
|
||||||
|
async def api_action_attempt(request: Request) -> JSONResponse:
|
||||||
|
"""POST is rejected by read-only guard; kept for explicit fail-closed tests."""
|
||||||
|
action_id = request.path_params["action_id"]
|
||||||
|
try:
|
||||||
|
body = await request.json()
|
||||||
|
except Exception:
|
||||||
|
body = {}
|
||||||
|
if not isinstance(body, dict):
|
||||||
|
body = {}
|
||||||
|
result = attempt_action(action_id, **body)
|
||||||
|
status = 403 if not result.get("success") else 200
|
||||||
|
return JSONResponse(result, status_code=status)
|
||||||
|
|
||||||
|
|
||||||
async def method_not_allowed(request: Request, _exc: Exception) -> Response:
|
async def method_not_allowed(request: Request, _exc: Exception) -> Response:
|
||||||
path = request.url.path
|
path = request.url.path
|
||||||
if path in _AUDIT_MUTATION_PATHS and request.method == "POST":
|
if path in _AUDIT_MUTATION_PATHS and request.method == "POST":
|
||||||
@@ -234,6 +272,18 @@ def create_app() -> Starlette:
|
|||||||
Route("/worktrees", worktrees, methods=["GET"]),
|
Route("/worktrees", worktrees, methods=["GET"]),
|
||||||
Route("/api/worktrees", api_worktrees, methods=["GET"]),
|
Route("/api/worktrees", api_worktrees, methods=["GET"]),
|
||||||
Route("/leases", leases, methods=["GET"]),
|
Route("/leases", leases, methods=["GET"]),
|
||||||
|
Route("/actions", actions, methods=["GET"]),
|
||||||
|
Route("/api/actions", api_actions, methods=["GET"]),
|
||||||
|
Route(
|
||||||
|
"/api/actions/{action_id}/preview",
|
||||||
|
api_action_preview,
|
||||||
|
methods=["GET"],
|
||||||
|
),
|
||||||
|
Route(
|
||||||
|
"/api/actions/{action_id}/attempt",
|
||||||
|
api_action_attempt,
|
||||||
|
methods=["POST"],
|
||||||
|
),
|
||||||
Route("/api/leases", api_leases, methods=["GET"]),
|
Route("/api/leases", api_leases, methods=["GET"]),
|
||||||
],
|
],
|
||||||
exception_handlers={405: method_not_allowed},
|
exception_handlers={405: method_not_allowed},
|
||||||
|
|||||||
@@ -0,0 +1,101 @@
|
|||||||
|
"""HTML views for gated action previews (#434)."""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import html
|
||||||
|
import json
|
||||||
|
|
||||||
|
from webui.gated_actions import ActionRegistry, GatedAction, preview_action
|
||||||
|
from webui.layout import render_page
|
||||||
|
|
||||||
|
|
||||||
|
def _escape(text: str) -> str:
|
||||||
|
return html.escape(text, quote=True)
|
||||||
|
|
||||||
|
|
||||||
|
def _ledger_block(action: GatedAction) -> str:
|
||||||
|
sample_params: dict[str, object]
|
||||||
|
if action.action_id == "create_issue":
|
||||||
|
sample_params = {"title": "Example issue"}
|
||||||
|
elif action.action_id == "create_pr":
|
||||||
|
sample_params = {"head": "feat/issue-99-example", "base": "master"}
|
||||||
|
elif action.action_id == "delete_branch":
|
||||||
|
sample_params = {"branch_name": "feat/issue-99-example"}
|
||||||
|
elif action.action_id in {"review_pr", "merge_pr", "comment_pr", "close_pr"}:
|
||||||
|
sample_params = {"pr_number": 99}
|
||||||
|
else:
|
||||||
|
sample_params = {"issue_number": 99}
|
||||||
|
|
||||||
|
preview = preview_action(action.action_id, **sample_params)
|
||||||
|
ledger_json = json.dumps(preview.get("mutation_ledger", []), indent=2)
|
||||||
|
return (
|
||||||
|
"<details class='action-preview'>"
|
||||||
|
"<summary>Preview mutation ledger</summary>"
|
||||||
|
f"<pre class='prompt-text'>{_escape(ledger_json)}</pre>"
|
||||||
|
f"<p class='muted meta'>Sample params: "
|
||||||
|
f"<code>{_escape(json.dumps(sample_params))}</code></p>"
|
||||||
|
f"<p><a href=\"/api/actions/{_escape(action.action_id)}/preview?"
|
||||||
|
f"{_preview_query(sample_params)}\">JSON preview</a></p>"
|
||||||
|
"</details>"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def _preview_query(params: dict[str, object]) -> str:
|
||||||
|
parts = []
|
||||||
|
for key, value in params.items():
|
||||||
|
parts.append(f"{key}={_escape(str(value))}")
|
||||||
|
return "&".join(parts)
|
||||||
|
|
||||||
|
|
||||||
|
def _action_card(action: GatedAction) -> str:
|
||||||
|
disabled_attr = " disabled" if not action.enabled else ""
|
||||||
|
return (
|
||||||
|
"<article class='prompt-card action-card'>"
|
||||||
|
f"<h3>{_escape(action.label)}</h3>"
|
||||||
|
f"<p class='muted'>{_escape(action.description)}</p>"
|
||||||
|
"<p class='meta'>"
|
||||||
|
f"Task <code>{_escape(action.task_key)}</code> · "
|
||||||
|
f"MCP <code>{_escape(action.mcp_tool)}</code><br>"
|
||||||
|
f"Requires <code>{_escape(action.required_permission)}</code> "
|
||||||
|
f"({_escape(action.required_role)} profile)</p>"
|
||||||
|
f"<button type='button' class='copy-btn action-btn'{disabled_attr} "
|
||||||
|
f"aria-disabled='true' title='Disabled in read-only MVP'>"
|
||||||
|
f"{_escape(action.label)}</button>"
|
||||||
|
f"{_ledger_block(action)}"
|
||||||
|
"</article>"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
ACTION_PAGE_STYLES = """
|
||||||
|
<style>
|
||||||
|
.action-card .action-btn[disabled] {
|
||||||
|
opacity: 0.45;
|
||||||
|
cursor: not-allowed;
|
||||||
|
filter: none;
|
||||||
|
}
|
||||||
|
.action-preview { margin-top: 0.75rem; }
|
||||||
|
.action-preview summary {
|
||||||
|
cursor: pointer;
|
||||||
|
color: var(--accent);
|
||||||
|
font-size: 0.9rem;
|
||||||
|
}
|
||||||
|
</style>
|
||||||
|
"""
|
||||||
|
|
||||||
|
|
||||||
|
def render_actions_page(registry: ActionRegistry) -> str:
|
||||||
|
cards = "".join(_action_card(action) for action in registry.actions)
|
||||||
|
body = (
|
||||||
|
"<h2>Gated actions</h2>"
|
||||||
|
"<p>Future write actions are registered here but remain "
|
||||||
|
"<strong>disabled</strong> in the read-only MVP. Each action "
|
||||||
|
"declares the MCP capability and profile role from "
|
||||||
|
"<code>task_capability_map</code>; previews show the mutation "
|
||||||
|
"ledger before any execution path ships.</p>"
|
||||||
|
f"<p class='meta'>{len(registry.actions)} registered actions · "
|
||||||
|
"execution: fail-closed</p>"
|
||||||
|
f"{cards}"
|
||||||
|
"<p><a href=\"/api/actions\">JSON registry</a></p>"
|
||||||
|
f"{ACTION_PAGE_STYLES}"
|
||||||
|
)
|
||||||
|
return render_page(title="Actions", body_html=body)
|
||||||
@@ -0,0 +1,201 @@
|
|||||||
|
"""Disabled-by-default gated action registry for future UI writes (#434).
|
||||||
|
|
||||||
|
Every action declares the MCP capability and profile role from
|
||||||
|
``task_capability_map``. Previews always render a mutation ledger; execution
|
||||||
|
is fail-closed in the read-only MVP (no MCP or shell fallbacks).
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from dataclasses import asdict, dataclass, field
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
|
from task_capability_map import required_permission, required_role
|
||||||
|
|
||||||
|
# MVP: no UI action may execute mutations until capability gates ship.
|
||||||
|
_MVP_ACTIONS_ENABLED = False
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True)
|
||||||
|
class MutationLedgerEntry:
|
||||||
|
"""One planned mutation step shown before any write executes."""
|
||||||
|
|
||||||
|
sequence: int
|
||||||
|
mcp_tool: str
|
||||||
|
permission: str
|
||||||
|
target: str
|
||||||
|
summary: str
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True)
|
||||||
|
class GatedAction:
|
||||||
|
"""Registry entry tying a UI action to resolver task metadata."""
|
||||||
|
|
||||||
|
action_id: str
|
||||||
|
label: str
|
||||||
|
task_key: str
|
||||||
|
mcp_tool: str
|
||||||
|
description: str
|
||||||
|
enabled: bool = False
|
||||||
|
|
||||||
|
@property
|
||||||
|
def required_permission(self) -> str:
|
||||||
|
return required_permission(self.task_key)
|
||||||
|
|
||||||
|
@property
|
||||||
|
def required_role(self) -> str:
|
||||||
|
return required_role(self.task_key)
|
||||||
|
|
||||||
|
def mutation_ledger(self, **params: Any) -> tuple[MutationLedgerEntry, ...]:
|
||||||
|
"""Build the mutation ledger preview for *params* (read-only)."""
|
||||||
|
target = _format_target(self.action_id, params)
|
||||||
|
return (
|
||||||
|
MutationLedgerEntry(
|
||||||
|
sequence=1,
|
||||||
|
mcp_tool=self.mcp_tool,
|
||||||
|
permission=self.required_permission,
|
||||||
|
target=target,
|
||||||
|
summary=self.description,
|
||||||
|
),
|
||||||
|
)
|
||||||
|
|
||||||
|
def preview(self, **params: Any) -> dict[str, Any]:
|
||||||
|
ledger = self.mutation_ledger(**params)
|
||||||
|
return {
|
||||||
|
"action_id": self.action_id,
|
||||||
|
"label": self.label,
|
||||||
|
"task_key": self.task_key,
|
||||||
|
"mcp_tool": self.mcp_tool,
|
||||||
|
"required_permission": self.required_permission,
|
||||||
|
"required_role": self.required_role,
|
||||||
|
"enabled": self.enabled and _MVP_ACTIONS_ENABLED,
|
||||||
|
"mvp_mode": "read-only",
|
||||||
|
"mutation_ledger": [asdict(entry) for entry in ledger],
|
||||||
|
"params": dict(params),
|
||||||
|
}
|
||||||
|
|
||||||
|
def attempt(self, **params: Any) -> dict[str, Any]:
|
||||||
|
"""Fail closed — never invokes MCP tools in MVP."""
|
||||||
|
preview = self.preview(**params)
|
||||||
|
if not preview["enabled"]:
|
||||||
|
return {
|
||||||
|
"success": False,
|
||||||
|
"error": "action_disabled",
|
||||||
|
"detail": (
|
||||||
|
"UI actions are disabled in the read-only MVP. "
|
||||||
|
"Use MCP tools with capability gates."
|
||||||
|
),
|
||||||
|
"preview": preview,
|
||||||
|
}
|
||||||
|
return {
|
||||||
|
"success": False,
|
||||||
|
"error": "action_not_implemented",
|
||||||
|
"detail": "Gated execution path is not wired in MVP.",
|
||||||
|
"preview": preview,
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def _format_target(action_id: str, params: dict[str, Any]) -> str:
|
||||||
|
if action_id == "claim_issue":
|
||||||
|
return f"issue #{params.get('issue_number', '?')}"
|
||||||
|
if action_id in {"comment_issue", "close_issue"}:
|
||||||
|
return f"issue #{params.get('issue_number', '?')}"
|
||||||
|
if action_id in {"review_pr", "merge_pr", "comment_pr", "close_pr"}:
|
||||||
|
return f"PR #{params.get('pr_number', '?')}"
|
||||||
|
if action_id == "delete_branch":
|
||||||
|
return f"branch {params.get('branch_name', '?')!r}"
|
||||||
|
if action_id == "create_pr":
|
||||||
|
return (
|
||||||
|
f"PR {params.get('head', '?')} → {params.get('base', 'master')}"
|
||||||
|
)
|
||||||
|
if action_id == "create_issue":
|
||||||
|
return f"issue {params.get('title', '?')!r}"
|
||||||
|
return "unspecified"
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class ActionRegistry:
|
||||||
|
"""Ordered registry of gated UI actions."""
|
||||||
|
|
||||||
|
actions: tuple[GatedAction, ...] = field(default_factory=tuple)
|
||||||
|
|
||||||
|
def get(self, action_id: str) -> GatedAction | None:
|
||||||
|
for action in self.actions:
|
||||||
|
if action.action_id == action_id:
|
||||||
|
return action
|
||||||
|
return None
|
||||||
|
|
||||||
|
def to_dict(self) -> dict[str, Any]:
|
||||||
|
return {
|
||||||
|
"mvp_actions_enabled": _MVP_ACTIONS_ENABLED,
|
||||||
|
"actions": [
|
||||||
|
{
|
||||||
|
"action_id": action.action_id,
|
||||||
|
"label": action.label,
|
||||||
|
"task_key": action.task_key,
|
||||||
|
"mcp_tool": action.mcp_tool,
|
||||||
|
"required_permission": action.required_permission,
|
||||||
|
"required_role": action.required_role,
|
||||||
|
"enabled": action.enabled,
|
||||||
|
"description": action.description,
|
||||||
|
}
|
||||||
|
for action in self.actions
|
||||||
|
],
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def build_action_registry() -> ActionRegistry:
|
||||||
|
"""Return the canonical MVP action set (all disabled)."""
|
||||||
|
specs = (
|
||||||
|
("claim_issue", "Claim issue", "claim_issue", "gitea_mark_issue",
|
||||||
|
"Apply status:in-progress via issue comment gate."),
|
||||||
|
("comment_issue", "Comment on issue", "comment_issue",
|
||||||
|
"gitea_create_issue_comment", "Post an issue comment."),
|
||||||
|
("create_issue", "Create issue", "create_issue", "gitea_create_issue",
|
||||||
|
"Open a new tracking issue."),
|
||||||
|
("create_pr", "Create pull request", "create_pr", "gitea_create_pr",
|
||||||
|
"Open a PR from a locked feature branch."),
|
||||||
|
("review_pr", "Review PR", "review_pr", "gitea_review_pr",
|
||||||
|
"Submit approve/request-changes/comment review."),
|
||||||
|
("merge_pr", "Merge PR", "merge_pr", "gitea_merge_pr",
|
||||||
|
"Merge an approved pull request."),
|
||||||
|
("delete_branch", "Delete branch", "delete_branch",
|
||||||
|
"gitea_delete_branch", "Remove a remote feature branch."),
|
||||||
|
("comment_pr", "Comment on PR", "comment_pr",
|
||||||
|
"gitea_create_issue_comment", "Post a PR review thread comment."),
|
||||||
|
("close_pr", "Close PR", "close_pr", "gitea_edit_pr",
|
||||||
|
"Close a pull request without merge."),
|
||||||
|
)
|
||||||
|
actions = tuple(
|
||||||
|
GatedAction(
|
||||||
|
action_id=action_id,
|
||||||
|
label=label,
|
||||||
|
task_key=task_key,
|
||||||
|
mcp_tool=mcp_tool,
|
||||||
|
description=description,
|
||||||
|
enabled=False,
|
||||||
|
)
|
||||||
|
for action_id, label, task_key, mcp_tool, description in specs
|
||||||
|
)
|
||||||
|
return ActionRegistry(actions=actions)
|
||||||
|
|
||||||
|
|
||||||
|
_REGISTRY = build_action_registry()
|
||||||
|
|
||||||
|
|
||||||
|
def load_action_registry() -> ActionRegistry:
|
||||||
|
return _REGISTRY
|
||||||
|
|
||||||
|
|
||||||
|
def preview_action(action_id: str, **params: Any) -> dict[str, Any]:
|
||||||
|
action = _REGISTRY.get(action_id)
|
||||||
|
if action is None:
|
||||||
|
return {"error": "unknown_action", "action_id": action_id}
|
||||||
|
return action.preview(**params)
|
||||||
|
|
||||||
|
|
||||||
|
def attempt_action(action_id: str, **params: Any) -> dict[str, Any]:
|
||||||
|
action = _REGISTRY.get(action_id)
|
||||||
|
if action is None:
|
||||||
|
return {"success": False, "error": "unknown_action", "action_id": action_id}
|
||||||
|
return action.attempt(**params)
|
||||||
@@ -11,6 +11,7 @@ NAV_ITEMS = (
|
|||||||
("/audit", "Audit"),
|
("/audit", "Audit"),
|
||||||
("/worktrees", "Worktrees"),
|
("/worktrees", "Worktrees"),
|
||||||
("/leases", "Leases"),
|
("/leases", "Leases"),
|
||||||
|
("/actions", "Actions"),
|
||||||
)
|
)
|
||||||
|
|
||||||
MVP_NOTICE = (
|
MVP_NOTICE = (
|
||||||
|
|||||||
Reference in New Issue
Block a user