Compare commits

...
19 Commits
Author SHA1 Message Date
sysadmin fe29379ad2 test: keep webui CI hermetic offline 2026-07-09 11:53:08 -04:00
sysadmin 9a641ed4a5 feat(webui): test coverage and CI gate (#436)
Add consolidated MVP route/read-only tests, path-filter helpers, and
scripts/test-webui plus scripts/ci-webui-check for Jenkins path-filtered
runs on PRs touching web UI surfaces.

Closes #436
2026-07-09 11:52:58 -04:00
sysadmin 8d1098f916 Merge pull request 'feat: guard merged branch cleanup path' (#516) from feat/issue-514-branch-delete-guard into master 2026-07-09 10:45:05 -05:00
sysadmin 49aa3b2812 Merge pull request 'Web UI: Gated action framework (#434)' (#455) from feat/issue-434-gated-actions into master 2026-07-09 10:39:28 -05:00
sysadmin 94004f05ac Merge pull request 'fix: newest reviewer lease marker wins so release ends prior claims (Closes #577)' (#579) from fix/issue-577-reviewer-lease-newest-wins into master 2026-07-09 10:38:22 -05:00
sysadmin 9b96085d8d fix: newest lease marker wins so release ends prior claims
find_active_reviewer_lease walked past terminal markers and re-armed
older claimed leases after gitea_release_reviewer_pr_lease. Treat the
newest marker as authoritative (append-only ledger).

Closes #577
2026-07-09 10:09:38 -04:00
sysadmin ddb1dd941b Merge remote-tracking branch 'prgs/master' into feat/issue-434-gated-actions
# Conflicts:
#	docs/webui-local-dev.md
#	tests/test_webui_skeleton.py
#	webui/app.py
2026-07-09 09:38:20 -04:00
sysadmin d7c29afde5 Merge remote-tracking branch 'prgs/master' into feat/issue-514-branch-delete-guard
# Conflicts:
#	gitea_mcp_server.py
2026-07-09 09:35:55 -04:00
sysadmin 7acd55d2d2 Merge remote-tracking branch 'prgs/master' into feat/issue-514-branch-delete-guard
# Conflicts:
#	final_report_validator.py
2026-07-09 09:12:33 -04:00
sysadmin 46709537c1 Merge remote-tracking branch 'prgs/master' into feat/issue-434-gated-actions
# Conflicts:
#	docs/webui-local-dev.md
2026-07-09 08:41:08 -04:00
sysadmin 62ab4b4d95 Merge remote-tracking branch 'prgs/master' into feat/issue-514-branch-delete-guard
# Conflicts:
#	gitea_mcp_server.py
2026-07-09 08:28:50 -04:00
sysadmin d9cf26ddbf Merge branch 'master' into feat/issue-514-branch-delete-guard 2026-07-08 22:45:19 -05:00
sysadmin b71dfe9696 Merge branch 'master' into feat/issue-434-gated-actions 2026-07-08 22:40:02 -05:00
sysadmin 22d4735170 fix: resolve conflicts for PR #455
Merge prgs/master into PR branch.
2026-07-08 22:39:43 -04:00
sysadmin d57ca94ec5 fix: re-resolve conflicts for PR #516 after master advance 2026-07-08 22:33:21 -04:00
sysadminandClaude Opus 4.8 09d8db5b9e fix: resolve conflicts for PR #516
Merge prgs/master into PR branch to restore mergeability.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-08 22:28:13 -04:00
sysadmin 3550262613 fix: resolve conflicts for PR #516
Merge prgs/master into feat/issue-514-branch-delete-guard and preserve PR intent
alongside compatible master guardrails.
2026-07-08 11:10:21 -04:00
sysadmin 3cb05284b3 feat: guard merged branch cleanup path
Closes #514
2026-07-08 04:18:09 -04:00
sysadmin 84b841c727 feat(webui): gated action framework (#434)
Register future write actions with task_capability_map metadata,
mutation-ledger previews, and fail-closed execution in the read-only MVP.
Adds /actions routes, disabled UI buttons, and tests proving ungated
attempts cannot invoke mutations.

Closes #434
2026-07-07 15:34:39 -04:00
26 changed files with 1487 additions and 32 deletions
+98
View File
@@ -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",
}
+19
View File
@@ -54,6 +54,25 @@ Use `-q` for a compact summary and `-v` to see individual test names.
./venv/bin/python -m pytest tests/ -q
```
### Web UI suite (#436)
Hermetic unittest modules matching `test_webui_*.py` cover route rendering,
read-only guards, registry/prompt/queue loaders, and optional child-issue
modules when present on the branch.
```bash
./scripts/test-webui
./scripts/ci-webui-check # skip unless the diff touches web UI paths
WEBUI_CI_FORCE=1 ./scripts/ci-webui-check
```
`scripts/test-webui` defaults `WEBUI_TEST_OFFLINE=1` so route coverage never
needs Gitea credentials or MCP daemon credential access. Set
`WEBUI_TEST_OFFLINE=0` only for explicit operator live-fetch checks.
Wire `scripts/ci-webui-check` into Jenkins (or equivalent) for PRs that touch
`webui/`, `tests/test_webui_*`, or `docs/webui*`.
### Run targeted tests
```bash
+65 -3
View File
@@ -49,8 +49,9 @@ Optional environment variables:
| `/api/audit` | JSON validator preview (POST `report_text`, optional `task_kind`) |
| `/worktrees` | Worktree hygiene dashboard (#432) |
| `/api/worktrees` | JSON worktree scan with classifications and anomalies |
| `/leases` | Stub — lease visibility (#433) |
| `/worktrees` | Stub — hygiene dashboard (#432) |
| `/actions` | Gated write-action registry — all disabled in MVP (#434) |
| `/api/actions` | JSON action registry with capability metadata |
| `/api/actions/{id}/preview` | Mutation ledger preview (GET, read-only) |
| `/leases` | Lease and collision visibility (#433) |
| `/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
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)
`/worktrees` scans local `branches/` directories and registered git worktrees.
@@ -106,6 +116,58 @@ referenced by the issue lock file are flagged as anomalies (#404). The page
includes a copy/paste canonical cleanup prompt only — no deletion actions.
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)
Run the full hermetic web UI suite (all `test_webui_*.py` modules):
```bash
./scripts/test-webui
```
CI / Jenkins multibranch can call the path-filtered gate (runs only when the
diff touches `webui/`, `tests/test_webui_*`, or web UI docs/scripts):
```bash
./scripts/ci-webui-check
WEBUI_CI_FORCE=1 ./scripts/ci-webui-check # always run
```
`scripts/test-webui` sets `WEBUI_TEST_OFFLINE=1` by default. In that mode the
queue, lease, and runtime routes use empty offline snapshots instead of Gitea
credentials, so CI can run without MCP daemon credential access. Set
`WEBUI_TEST_OFFLINE=0` only when deliberately validating live fetch behavior.
Or invoke unittest directly:
```bash
python3 -m unittest discover -s tests -p 'test_webui_*.py' -q
```
## Lease visibility (#433)
`/leases` surfaces read-only lease and collision state: local issue lock file,
@@ -128,4 +190,4 @@ no tokens or MCP restart actions are exposed.
```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_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
```
```
+13
View File
@@ -11,6 +11,7 @@ import inspect
import re
from typing import Any, Callable
import branch_cleanup_guard
import issue_acceptance_gate
import issue_lock_provenance
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]]:
"""#403: require structured workflow-load helper result, not file-view narrative."""
if not report_text.strip():
@@ -1448,6 +1460,7 @@ _SHARED_ISSUE_LOCK_RULES = (
_rule_shared_issue_lock_external_state,
_rule_shared_manual_lock_pr_override,
_rule_shared_author_reviewer_same_run,
_rule_shared_raw_branch_delete_bypass,
_rule_shared_canonical_state_update,
)
+144
View File
@@ -832,6 +832,7 @@ import pr_work_lease # noqa: E402
import workflow_skill # noqa: E402
import conflict_fix_classification # noqa: E402
import native_mcp_preference # noqa: E402
import branch_cleanup_guard # noqa: E402
import thread_state_ledger_validator # 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."}
@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:
import urllib.parse
+21 -11
View File
@@ -190,18 +190,28 @@ def find_active_reviewer_lease(
pr_number: int,
now: datetime | None = 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)
for lease in reversed(_lease_entries(comments, pr_number=pr_number)):
phase = (lease.get("phase") or "").strip().lower()
if phase in _TERMINAL_PHASES:
continue
if _lease_expired(lease, now=now):
continue
if phase in _ACTIVE_PHASES or phase:
lease = dict(lease)
lease["freshness"] = classify_lease_freshness(lease, now=now)
return lease
entries = list(reversed(_lease_entries(comments, pr_number=pr_number)))
if not entries:
return None
newest = entries[0]
phase = (newest.get("phase") or "").strip().lower()
if phase in _TERMINAL_PHASES:
return None
if _lease_expired(newest, now=now):
return None
if phase in _ACTIVE_PHASES or phase:
lease = dict(newest)
lease["freshness"] = classify_lease_freshness(lease, now=now)
return lease
return None
+3 -1
View File
@@ -80,6 +80,7 @@ AUTHOR_TASKS = frozenset({
})
RECONCILER_TASKS = frozenset({
"cleanup_merged_pr_branch",
"reconcile_already_landed_pr",
"reconcile_already_landed",
"reconcile-landed-pr",
@@ -109,6 +110,7 @@ TASK_REQUIRED_ROLE = {
"reconcile_already_landed_pr": "reconciler",
"reconcile_already_landed": "reconciler",
"reconcile-landed-pr": "reconciler",
"cleanup_merged_pr_branch": "reconciler",
# #309: reconciler tasks close already-landed PRs/issues only.
"reconcile_close_landed_pr": "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:
"""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"]
+53
View File
@@ -0,0 +1,53 @@
#!/usr/bin/env bash
# Path-filtered CI gate for the internal web UI (#436).
# Runs the hermetic web UI unittest suite when a change touches UI surfaces.
set -euo pipefail
script_dir="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
repo_root="$(cd "$script_dir/.." && pwd)"
cd "$repo_root"
if [[ "${WEBUI_CI_FORCE:-}" == "1" ]]; then
exec "$script_dir/test-webui"
fi
base_ref="${WEBUI_CI_BASE_REF:-}"
if [[ -z "$base_ref" ]]; then
if git rev-parse --verify prgs/master >/dev/null 2>&1; then
base_ref="$(git merge-base HEAD prgs/master 2>/dev/null || true)"
fi
if [[ -z "$base_ref" ]]; then
base_ref="${GITHUB_BASE_REF:-${CHANGE_TARGET:-master}}"
if git rev-parse --verify "origin/$base_ref" >/dev/null 2>&1; then
base_ref="$(git merge-base HEAD "origin/$base_ref")"
elif git rev-parse --verify "$base_ref" >/dev/null 2>&1; then
base_ref="$(git merge-base HEAD "$base_ref")"
else
base_ref="HEAD~1"
fi
fi
fi
changed="$(git diff --name-only "$base_ref" HEAD 2>/dev/null || true)"
if [[ -z "$changed" ]]; then
echo "ci-webui-check: no changed files vs $base_ref; skipping web UI suite"
exit 0
fi
python_bin="${WEBUI_TEST_PYTHON:-python3}"
if [[ -x "$repo_root/venv/bin/python" ]]; then
python_bin="$repo_root/venv/bin/python"
fi
if printf '%s\n' "$changed" | "$python_bin" -c "
import sys
from webui.ci_paths import should_run_webui_ci
paths = [line.strip() for line in sys.stdin if line.strip()]
raise SystemExit(0 if should_run_webui_ci(paths) else 1)
"; then
echo "ci-webui-check: web UI surface changed; running suite"
exec "$script_dir/test-webui"
fi
echo "ci-webui-check: no web UI paths in diff; skipping suite"
exit 0
+14
View File
@@ -0,0 +1,14 @@
#!/usr/bin/env bash
set -euo pipefail
script_dir="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
repo_root="$(cd "$script_dir/.." && pwd)"
cd "$repo_root"
python_bin="${WEBUI_TEST_PYTHON:-python3}"
if [[ -x "$repo_root/venv/bin/python" ]]; then
python_bin="$repo_root/venv/bin/python"
fi
pattern="${WEBUI_TEST_PATTERN:-test_webui_*.py}"
export WEBUI_TEST_OFFLINE="${WEBUI_TEST_OFFLINE:-1}"
exec "$python_bin" -m unittest discover -s tests -p "$pattern" "$@"
@@ -936,6 +936,14 @@ Confirm:
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
transient and are removed automatically at successful completion once they are
clean, carry no open PR, and hold no active lease (#401). Use
+4
View File
@@ -100,6 +100,10 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
"permission": "gitea.branch.delete",
"role": "author",
},
"cleanup_merged_pr_branch": {
"permission": "gitea.branch.delete",
"role": "reconciler",
},
"commit_files": {
"permission": "gitea.repo.commit",
"role": "author",
+187
View File
@@ -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()
+39
View File
@@ -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):
def setUp(self):
leases.clear_session_lease()
+90
View File
@@ -0,0 +1,90 @@
"""Tests for web UI CI path-filter gate (#436)."""
import os
import subprocess
import sys
import unittest
from pathlib import Path
from unittest import mock
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from webui.ci_paths import should_run_webui_ci, touches_webui_surface
from webui.lease_loader import load_lease_snapshot
from webui.queue_loader import load_queue_snapshot
from webui.runtime_health import load_runtime_snapshot
_REPO_ROOT = Path(__file__).resolve().parent.parent
class TestCiPathMatching(unittest.TestCase):
def test_positive_paths(self):
for path in (
"webui/app.py",
"tests/test_webui_skeleton.py",
"docs/webui-local-dev.md",
"scripts/test-webui",
):
with self.subTest(path=path):
self.assertTrue(touches_webui_surface(path))
def test_negative_paths(self):
for path in ("mcp_server.py", "tests/test_mcp_server.py", "README.md"):
with self.subTest(path=path):
self.assertFalse(touches_webui_surface(path))
def test_should_run_aggregate(self):
self.assertTrue(should_run_webui_ci(["README.md", "webui/layout.py"]))
self.assertFalse(should_run_webui_ci(["mcp_server.py", "gitea_auth.py"]))
class TestCiRunnerScript(unittest.TestCase):
def test_test_webui_smoke(self):
script = _REPO_ROOT / "scripts" / "test-webui"
result = subprocess.run(
["bash", str(script), "-q"],
cwd=_REPO_ROOT,
capture_output=True,
text=True,
timeout=60,
env={
**os.environ,
"WEBUI_TEST_PATTERN": "test_webui_skeleton.py",
},
)
self.assertEqual(result.returncode, 0, result.stderr or result.stdout)
class TestOfflineWebuiCiMode(unittest.TestCase):
def test_offline_mode_avoids_live_queue_auth(self):
with mock.patch.dict(os.environ, {"WEBUI_TEST_OFFLINE": "1"}):
with mock.patch(
"webui.queue_loader.get_auth_header",
side_effect=AssertionError("live auth should not run"),
):
snapshot = load_queue_snapshot()
self.assertIsNone(snapshot.fetch_error)
self.assertEqual(snapshot.prs, ())
self.assertEqual(snapshot.issues, ())
def test_offline_mode_avoids_live_lease_auth(self):
with mock.patch.dict(os.environ, {"WEBUI_TEST_OFFLINE": "1"}):
with mock.patch(
"webui.lease_loader.get_auth_header",
side_effect=AssertionError("live auth should not run"),
):
snapshot = load_lease_snapshot()
self.assertIsNone(snapshot.fetch_error)
self.assertEqual(snapshot.reviewer_leases, ())
def test_offline_mode_avoids_live_runtime_identity(self):
with mock.patch.dict(os.environ, {"WEBUI_TEST_OFFLINE": "1"}):
with mock.patch(
"webui.runtime_health.get_auth_header",
side_effect=AssertionError("live auth should not run"),
):
snapshot = load_runtime_snapshot()
self.assertEqual(snapshot.identity_error, "offline web UI test mode")
if __name__ == "__main__":
unittest.main()
+128
View File
@@ -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()
+156
View File
@@ -0,0 +1,156 @@
"""Consolidated MVP coverage for the internal web UI (#436)."""
from __future__ import annotations
import importlib
import os
import sys
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from starlette.testclient import TestClient
from starlette.routing import Route
from webui.app import create_app
_REPO_ROOT = Path(__file__).resolve().parent.parent
# HTML pages and JSON exports registered on master MVP.
MVP_GET_ROUTES: tuple[tuple[str, str | tuple[str, ...]], ...] = (
("/", "Operator console"),
("/health", "mcp-control-plane-webui"),
("/queue", "Live queue"),
("/api/queue", ("prs", "issues")),
("/projects", "Gitea-Tools"),
("/api/projects", "projects"),
("/prompts", "Prompt library"),
("/api/prompts", "prompts"),
("/runtime", "Runtime"),
("/audit", "Audit"),
("/worktrees", "Worktrees"),
("/leases", "Leases"),
)
MUTATION_METHODS = ("POST", "PUT", "PATCH", "DELETE")
def _import_optional(module_name: str):
try:
return importlib.import_module(module_name)
except ImportError:
return None
class TestWebuiServerStartup(unittest.TestCase):
def test_create_app_imports_cleanly(self):
app = create_app()
self.assertIsNotNone(app)
def test_main_module_imports(self):
import webui.__main__ as main_mod
self.assertTrue(callable(main_mod.main))
class TestWebuiRouteRendering(unittest.TestCase):
def setUp(self):
self.client = TestClient(create_app())
def test_all_mvp_get_routes_render(self):
for path, needle in MVP_GET_ROUTES:
with self.subTest(path=path):
response = self.client.get(path)
self.assertEqual(response.status_code, 200, path)
if path == "/health":
assert isinstance(needle, str)
self.assertEqual(response.json()["service"], needle)
elif path.startswith("/api/"):
data = response.json()
if isinstance(needle, tuple):
for key in needle:
self.assertIn(key, data)
else:
self.assertIn(needle, data)
else:
assert isinstance(needle, str)
self.assertIn(needle, response.text)
def test_registered_routes_include_mvp_paths(self):
app = create_app()
get_paths = {
route.path
for route in app.routes
if isinstance(route, Route) and "GET" in route.methods
}
for path, _ in MVP_GET_ROUTES:
with self.subTest(path=path):
self.assertIn(path, get_paths)
class TestWebuiReadOnlyGuards(unittest.TestCase):
def setUp(self):
self.client = TestClient(create_app())
def test_mutation_methods_rejected_on_core_paths(self):
paths = ("/health", "/queue", "/projects", "/prompts", "/api/queue")
for path in paths:
for method in MUTATION_METHODS:
with self.subTest(path=path, method=method):
response = self.client.request(method, path)
self.assertEqual(response.status_code, 405)
self.assertEqual(response.json()["error"], "read-only-mvp")
class TestWebuiOptionalModules(unittest.TestCase):
"""Exercise child-issue modules when present on the branch under test."""
def test_audit_validator_basics_when_present(self):
mod = _import_optional("webui.audit_validator")
if mod is None:
self.skipTest("webui.audit_validator not merged on this branch")
report = "## Controller Handoff\n- Task: review PR #1\n"
self.assertEqual(mod.infer_task_kind(report), "review_pr")
result = mod.audit_report(report)
self.assertIn("grade", result)
def test_gated_actions_fail_closed_when_present(self):
mod = _import_optional("webui.gated_actions")
if mod is None:
self.skipTest("webui.gated_actions not merged on this branch")
registry = mod.build_action_registry()
for action in registry.actions:
with self.subTest(action=action.action_id):
self.assertFalse(action.enabled)
result = mod.attempt_action("merge_pr", pr_number=1)
self.assertFalse(result["success"])
def test_deployment_boundary_when_present(self):
mod = _import_optional("webui.deployment_boundary")
if mod is None:
self.skipTest("webui.deployment_boundary not merged on this branch")
assessment = mod.assess_bind_host("0.0.0.0")
self.assertEqual(assessment.disposition, "refuse")
class TestWebuiTestInventory(unittest.TestCase):
def test_webui_test_modules_exist(self):
modules = sorted(_REPO_ROOT.glob("tests/test_webui_*.py"))
names = [path.name for path in modules]
self.assertIn("test_webui_skeleton.py", names)
self.assertIn("test_webui_project_registry.py", names)
self.assertIn("test_webui_prompt_library.py", names)
self.assertIn("test_webui_queue_dashboard.py", names)
self.assertGreaterEqual(len(names), 5)
class TestWebuiCiScripts(unittest.TestCase):
def test_runner_scripts_exist(self):
for name in ("test-webui", "ci-webui-check"):
script = _REPO_ROOT / "scripts" / name
self.assertTrue(script.is_file(), name)
self.assertTrue(os.access(script, os.X_OK), name)
if __name__ == "__main__":
unittest.main()
+4 -3
View File
@@ -206,8 +206,9 @@ class TestQueueRoutes(unittest.TestCase):
self.assertEqual(data["pagination"]["issues"]["returned_count"], 3)
def test_queue_fail_closed_without_credentials(self):
with mock.patch("webui.queue_loader.get_auth_header", return_value=None):
snapshot = load_queue_snapshot()
with mock.patch.dict("os.environ", {"WEBUI_TEST_OFFLINE": "0"}):
with mock.patch("webui.queue_loader.get_auth_header", return_value=None):
snapshot = load_queue_snapshot()
self.assertIsNotNone(snapshot.fetch_error)
self.assertEqual(len(snapshot.prs), 0)
@@ -285,4 +286,4 @@ class TestQueueFailClosedUx(unittest.TestCase):
if __name__ == "__main__":
unittest.main()
unittest.main()
+7 -2
View File
@@ -61,6 +61,11 @@ class TestWebuiSkeleton(unittest.TestCase):
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):
response = self.client.post("/health")
self.assertEqual(response.status_code, 405)
@@ -72,8 +77,8 @@ class TestWebuiSkeleton(unittest.TestCase):
self.assertIn("Live queue", response.text)
def test_nav_links_on_all_pages(self):
paths = ("/", "/queue", "/projects", "/prompts", "/runtime", "/audit", "/worktrees", "/leases")
hrefs = ("/queue", "/projects", "/prompts", "/runtime", "/audit", "/worktrees", "/leases")
paths = ("/", "/queue", "/projects", "/prompts", "/runtime", "/audit", "/worktrees", "/leases", "/actions")
hrefs = ("/queue", "/projects", "/prompts", "/runtime", "/audit", "/worktrees", "/leases", "/actions")
for path in paths:
with self.subTest(path=path):
text = self.client.get(path).text
+50
View File
@@ -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 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_views import render_audit_page
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>Worktrees</strong> — branch hygiene dashboard (#432)</li>"
"<li><strong>Leases</strong> — collision and lease visibility (#433)</li>"
"<li><strong>Actions</strong> — gated write-action framework (#434)</li>"
"</ul>"
)
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()))
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:
path = request.url.path
if path in _AUDIT_MUTATION_PATHS and request.method == "POST":
@@ -234,6 +272,18 @@ def create_app() -> Starlette:
Route("/worktrees", worktrees, methods=["GET"]),
Route("/api/worktrees", api_worktrees, 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"]),
],
exception_handlers={405: method_not_allowed},
+20
View File
@@ -0,0 +1,20 @@
"""Path filters for web UI CI gating (#436)."""
from __future__ import annotations
import re
from collections.abc import Iterable
_WEBUI_TOUCH_RE = re.compile(
r"^(?:webui/|tests/test_webui_|docs/webui|scripts/(?:run-webui|test-webui|ci-webui-check)$)"
)
def touches_webui_surface(path: str) -> bool:
"""Return True when *path* is a web UI surface file."""
return bool(_WEBUI_TOUCH_RE.match(path.strip()))
def should_run_webui_ci(changed_paths: Iterable[str]) -> bool:
"""Return True when any changed path should trigger the web UI test suite."""
return any(touches_webui_surface(path) for path in changed_paths)
+101
View File
@@ -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)
+201
View File
@@ -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)
+1
View File
@@ -11,6 +11,7 @@ NAV_ITEMS = (
("/audit", "Audit"),
("/worktrees", "Worktrees"),
("/leases", "Leases"),
("/actions", "Actions"),
)
MVP_NOTICE = (
+24 -5
View File
@@ -228,10 +228,29 @@ def load_lease_snapshot(
)
host = _host_from_url(project.remote_host)
pr_fetch = fetch_prs or _fetch_prs
issue_fetch = fetch_issues or _fetch_issues
comment_fetch = fetch_comments or _fetch_comments
using_live = fetch_prs is None or fetch_issues is None
offline_test = (os.environ.get("WEBUI_TEST_OFFLINE") or "").strip() in {
"1",
"true",
"yes",
}
def _empty_pr_fetch(*_args, **_kwargs):
return [], None
def _empty_issue_fetch(*_args, **_kwargs):
return [], None
def _empty_comment_fetch(*_args, **_kwargs):
return []
pr_fetch = fetch_prs or (_empty_pr_fetch if offline_test else _fetch_prs)
issue_fetch = fetch_issues or (
_empty_issue_fetch if offline_test else _fetch_issues
)
comment_fetch = fetch_comments or (
_empty_comment_fetch if offline_test else _fetch_comments
)
using_live = not offline_test and (fetch_prs is None or fetch_issues is None)
auth = get_auth_header(host) if using_live else "test-auth"
if using_live and not auth:
@@ -328,4 +347,4 @@ def snapshot_to_dict(snapshot: LeaseSnapshot) -> dict[str, Any]:
],
"collision_history": list(snapshot.collision_history),
"fetch_error": snapshot.fetch_error,
}
}
+19 -4
View File
@@ -2,6 +2,7 @@
from __future__ import annotations
import os
import re
from dataclasses import dataclass
from datetime import datetime, timezone
@@ -268,9 +269,23 @@ def load_queue_snapshot(
)
host = _host_from_url(project.remote_host)
pr_fetch = fetch_prs or _fetch_prs
issue_fetch = fetch_issues or _fetch_issues
using_live_fetch = fetch_prs is None or fetch_issues is None
offline_test = (os.environ.get("WEBUI_TEST_OFFLINE") or "").strip() in {
"1",
"true",
"yes",
}
def _empty_fetch(*_args, **_kwargs):
return [], _pagination_from_pages(
per_page=50,
pages_fetched=1,
returned_count=0,
is_final_page=True,
)
pr_fetch = fetch_prs or (_empty_fetch if offline_test else _fetch_prs)
issue_fetch = fetch_issues or (_empty_fetch if offline_test else _fetch_issues)
using_live_fetch = not offline_test and (fetch_prs is None or fetch_issues is None)
auth = get_auth_header(host) if using_live_fetch else "test-auth"
if using_live_fetch and not auth:
return QueueSnapshot(
@@ -382,4 +397,4 @@ def snapshot_to_dict(snapshot: QueueSnapshot) -> dict[str, Any]:
"prs": _page(snapshot.pr_pagination),
"issues": _page(snapshot.issue_pagination),
},
}
}
+18 -3
View File
@@ -217,6 +217,11 @@ def load_runtime_snapshot(
repo = _repo_root()
host = _host_from_url(project.remote_host)
remote = _remote_name_for_host(host)
offline_test = (os.environ.get("WEBUI_TEST_OFFLINE") or "").strip() in {
"1",
"true",
"yes",
}
profile = get_profile()
allowed = profile.get("allowed_operations") or []
forbidden = profile.get("forbidden_operations") or []
@@ -247,10 +252,20 @@ def load_runtime_snapshot(
fetch_error=str(exc),
)
identity_fn = resolve_username or _resolve_username
identity_fn = resolve_username
if identity_fn is None:
if offline_test:
identity_fn = lambda _host: (None, "offline web UI test mode")
else:
identity_fn = _resolve_username
username, identity_error = identity_fn(host)
git_fn = git_sync or _git_sync_status
git_fn = git_sync
if git_fn is None:
if offline_test:
git_fn = lambda _repo, _remote, _branch: (None, None, None)
else:
git_fn = _git_sync_status
remote_ref = "prgs" if remote == "prgs" else "origin"
local_sha, remote_sha, behind = git_fn(repo, remote_ref, project.default_branch)
@@ -317,4 +332,4 @@ def snapshot_to_dict(snapshot: RuntimeSnapshot) -> dict[str, Any]:
"schema_hashes": [_hash_row(item) for item in snapshot.schema_hashes],
"restart_guidance": snapshot.restart_guidance,
"fetch_error": snapshot.fetch_error,
}
}