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 ./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 ### Run targeted tests
```bash ```bash
+65 -3
View File
@@ -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.
@@ -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. 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)
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) ## 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,
@@ -128,4 +190,4 @@ no tokens or MCP restart actions are exposed.
```bash ```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_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 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 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,
) )
+144
View File
@@ -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
View File
@@ -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
+3 -1
View File
@@ -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"]
+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. 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
+4
View File
@@ -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",
+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): class TestReviewerLeaseMutationGate(unittest.TestCase):
def setUp(self): def setUp(self):
leases.clear_session_lease() 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) self.assertEqual(data["pagination"]["issues"]["returned_count"], 3)
def test_queue_fail_closed_without_credentials(self): def test_queue_fail_closed_without_credentials(self):
with mock.patch("webui.queue_loader.get_auth_header", return_value=None): with mock.patch.dict("os.environ", {"WEBUI_TEST_OFFLINE": "0"}):
snapshot = load_queue_snapshot() with mock.patch("webui.queue_loader.get_auth_header", return_value=None):
snapshot = load_queue_snapshot()
self.assertIsNotNone(snapshot.fetch_error) self.assertIsNotNone(snapshot.fetch_error)
self.assertEqual(len(snapshot.prs), 0) self.assertEqual(len(snapshot.prs), 0)
@@ -285,4 +286,4 @@ class TestQueueFailClosedUx(unittest.TestCase):
if __name__ == "__main__": 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) 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
+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 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},
+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"), ("/audit", "Audit"),
("/worktrees", "Worktrees"), ("/worktrees", "Worktrees"),
("/leases", "Leases"), ("/leases", "Leases"),
("/actions", "Actions"),
) )
MVP_NOTICE = ( MVP_NOTICE = (
+24 -5
View File
@@ -228,10 +228,29 @@ def load_lease_snapshot(
) )
host = _host_from_url(project.remote_host) host = _host_from_url(project.remote_host)
pr_fetch = fetch_prs or _fetch_prs offline_test = (os.environ.get("WEBUI_TEST_OFFLINE") or "").strip() in {
issue_fetch = fetch_issues or _fetch_issues "1",
comment_fetch = fetch_comments or _fetch_comments "true",
using_live = fetch_prs is None or fetch_issues is None "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" auth = get_auth_header(host) if using_live else "test-auth"
if using_live and not 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), "collision_history": list(snapshot.collision_history),
"fetch_error": snapshot.fetch_error, "fetch_error": snapshot.fetch_error,
} }
+19 -4
View File
@@ -2,6 +2,7 @@
from __future__ import annotations from __future__ import annotations
import os
import re import re
from dataclasses import dataclass from dataclasses import dataclass
from datetime import datetime, timezone from datetime import datetime, timezone
@@ -268,9 +269,23 @@ def load_queue_snapshot(
) )
host = _host_from_url(project.remote_host) host = _host_from_url(project.remote_host)
pr_fetch = fetch_prs or _fetch_prs offline_test = (os.environ.get("WEBUI_TEST_OFFLINE") or "").strip() in {
issue_fetch = fetch_issues or _fetch_issues "1",
using_live_fetch = fetch_prs is None or fetch_issues is None "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" auth = get_auth_header(host) if using_live_fetch else "test-auth"
if using_live_fetch and not auth: if using_live_fetch and not auth:
return QueueSnapshot( return QueueSnapshot(
@@ -382,4 +397,4 @@ def snapshot_to_dict(snapshot: QueueSnapshot) -> dict[str, Any]:
"prs": _page(snapshot.pr_pagination), "prs": _page(snapshot.pr_pagination),
"issues": _page(snapshot.issue_pagination), "issues": _page(snapshot.issue_pagination),
}, },
} }
+18 -3
View File
@@ -217,6 +217,11 @@ def load_runtime_snapshot(
repo = _repo_root() repo = _repo_root()
host = _host_from_url(project.remote_host) host = _host_from_url(project.remote_host)
remote = _remote_name_for_host(host) remote = _remote_name_for_host(host)
offline_test = (os.environ.get("WEBUI_TEST_OFFLINE") or "").strip() in {
"1",
"true",
"yes",
}
profile = get_profile() profile = get_profile()
allowed = profile.get("allowed_operations") or [] allowed = profile.get("allowed_operations") or []
forbidden = profile.get("forbidden_operations") or [] forbidden = profile.get("forbidden_operations") or []
@@ -247,10 +252,20 @@ def load_runtime_snapshot(
fetch_error=str(exc), 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) 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" remote_ref = "prgs" if remote == "prgs" else "origin"
local_sha, remote_sha, behind = git_fn(repo, remote_ref, project.default_branch) 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], "schema_hashes": [_hash_row(item) for item in snapshot.schema_hashes],
"restart_guidance": snapshot.restart_guidance, "restart_guidance": snapshot.restart_guidance,
"fetch_error": snapshot.fetch_error, "fetch_error": snapshot.fetch_error,
} }