From 84b841c72765ee1249cc6b9e4957ce1bf51ae606 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 15:34:39 -0400 Subject: [PATCH 01/12] 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 --- docs/webui-local-dev.md | 18 ++- tests/test_webui_gated_actions.py | 128 +++++++++++++++++++ tests/test_webui_skeleton.py | 11 +- webui/app.py | 50 ++++++++ webui/gated_action_views.py | 101 +++++++++++++++ webui/gated_actions.py | 201 ++++++++++++++++++++++++++++++ webui/layout.py | 1 + 7 files changed, 505 insertions(+), 5 deletions(-) create mode 100644 tests/test_webui_gated_actions.py create mode 100644 webui/gated_action_views.py create mode 100644 webui/gated_actions.py diff --git a/docs/webui-local-dev.md b/docs/webui-local-dev.md index 0a5b935..92e10b8 100644 --- a/docs/webui-local-dev.md +++ b/docs/webui-local-dev.md @@ -47,9 +47,12 @@ Optional environment variables: | `/audit` | Stub — report audit paste (#431) | | `/worktrees` | Stub — hygiene dashboard (#432) | | `/leases` | Stub — lease visibility (#433) | +| `/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) | -All routes are GET-only. POST/PUT/PATCH/DELETE return `405` with -`read-only-mvp`. +All routes are GET-only except registered POST handlers, which still return +`405` with `read-only-mvp` until write paths ship. ## Project registry (#427) @@ -82,8 +85,17 @@ 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. + ## 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 -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_gated_actions.py -q ``` \ No newline at end of file diff --git a/tests/test_webui_gated_actions.py b/tests/test_webui_gated_actions.py new file mode 100644 index 0000000..87d4086 --- /dev/null +++ b/tests/test_webui_gated_actions.py @@ -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() \ No newline at end of file diff --git a/tests/test_webui_skeleton.py b/tests/test_webui_skeleton.py index 5dd23a1..2960fc5 100644 --- a/tests/test_webui_skeleton.py +++ b/tests/test_webui_skeleton.py @@ -51,6 +51,11 @@ class TestWebuiSkeleton(unittest.TestCase): with self.subTest(path=path): self.assertEqual(self.client.get(path).status_code, 200) + 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) @@ -62,10 +67,12 @@ class TestWebuiSkeleton(unittest.TestCase): self.assertIn("Live queue", response.text) def test_nav_links_on_all_pages(self): - for path in ("/", "/queue", "/projects", "/prompts", "/runtime", "/audit"): + for path in ("/", "/queue", "/projects", "/prompts", "/runtime", "/audit", "/actions"): with self.subTest(path=path): text = self.client.get(path).text - for href in ("/queue", "/projects", "/prompts", "/runtime", "/audit"): + for href in ( + "/queue", "/projects", "/prompts", "/runtime", "/audit", "/actions", + ): self.assertIn(f'href="{href}"', text) diff --git a/webui/app.py b/webui/app.py index a97d1ad..4557f7b 100644 --- a/webui/app.py +++ b/webui/app.py @@ -14,6 +14,8 @@ from webui.project_registry import find_project, load_registry, registry_to_dict from webui.project_views import render_project_detail, render_projects_list from webui.prompt_library import find_prompt, library_to_dict from webui.prompt_views import render_prompt_detail, render_prompts_page +from webui.gated_actions import attempt_action, load_action_registry, preview_action +from webui.gated_action_views import render_actions_page from webui.queue_loader import load_queue_snapshot, snapshot_to_dict from webui.queue_views import render_queue_page @@ -41,6 +43,7 @@ async def home(_request: Request) -> HTMLResponse: "
  • Audit — final-report paste and validator preview (#431)
    • " "
  • Worktrees — branch hygiene dashboard (#432)
    • " "
  • Leases — collision and lease visibility (#433)
    • " + "
  • Actions — gated write-action framework (#434)
    • " "" ) return HTMLResponse(render_page(title="Home", body_html=body)) @@ -147,6 +150,41 @@ async def leases(_request: Request) -> HTMLResponse: ) +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: if request.method not in _READ_ONLY_METHODS: return JSONResponse( @@ -175,6 +213,18 @@ def create_app() -> Starlette: Route("/audit", audit, methods=["GET"]), Route("/worktrees", 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"], + ), ], exception_handlers={405: method_not_allowed}, ) \ No newline at end of file diff --git a/webui/gated_action_views.py b/webui/gated_action_views.py new file mode 100644 index 0000000..0d01d61 --- /dev/null +++ b/webui/gated_action_views.py @@ -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 ( + "
    " + "Preview mutation ledger" + f"
    {_escape(ledger_json)}
    " + f"

    Sample params: " + f"{_escape(json.dumps(sample_params))}

    " + f"

    JSON preview

    " + "
    " + ) + + +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 ( + "
    " + f"

    {_escape(action.label)}

    " + f"

    {_escape(action.description)}

    " + "

    " + f"Task {_escape(action.task_key)} · " + f"MCP {_escape(action.mcp_tool)}
    " + f"Requires {_escape(action.required_permission)} " + f"({_escape(action.required_role)} profile)

    " + f"" + f"{_ledger_block(action)}" + "
    " + ) + + +ACTION_PAGE_STYLES = """ + +""" + + +def render_actions_page(registry: ActionRegistry) -> str: + cards = "".join(_action_card(action) for action in registry.actions) + body = ( + "

    Gated actions

    " + "

    Future write actions are registered here but remain " + "disabled in the read-only MVP. Each action " + "declares the MCP capability and profile role from " + "task_capability_map; previews show the mutation " + "ledger before any execution path ships.

    " + f"

    {len(registry.actions)} registered actions · " + "execution: fail-closed

    " + f"{cards}" + "

    JSON registry

    " + f"{ACTION_PAGE_STYLES}" + ) + return render_page(title="Actions", body_html=body) \ No newline at end of file diff --git a/webui/gated_actions.py b/webui/gated_actions.py new file mode 100644 index 0000000..baefab8 --- /dev/null +++ b/webui/gated_actions.py @@ -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) \ No newline at end of file diff --git a/webui/layout.py b/webui/layout.py index a506c2c..47bedc1 100644 --- a/webui/layout.py +++ b/webui/layout.py @@ -11,6 +11,7 @@ NAV_ITEMS = ( ("/audit", "Audit"), ("/worktrees", "Worktrees"), ("/leases", "Leases"), + ("/actions", "Actions"), ) MVP_NOTICE = ( From 3cb05284b35bff23b8a5c68a7e67bd57f01ed5fa Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Wed, 8 Jul 2026 04:18:09 -0400 Subject: [PATCH 02/12] feat: guard merged branch cleanup path Closes #514 --- branch_cleanup_guard.py | 98 +++++++++ final_report_validator.py | 15 ++ gitea_mcp_server.py | 144 ++++++++++++++ role_session_router.py | 4 +- .../workflows/review-merge-pr.md | 9 + task_capability_map.py | 4 + tests/test_branch_cleanup_guard.py | 187 ++++++++++++++++++ 7 files changed, 460 insertions(+), 1 deletion(-) create mode 100644 branch_cleanup_guard.py create mode 100644 tests/test_branch_cleanup_guard.py diff --git a/branch_cleanup_guard.py b/branch_cleanup_guard.py new file mode 100644 index 0000000..66d2708 --- /dev/null +++ b/branch_cleanup_guard.py @@ -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", + } diff --git a/final_report_validator.py b/final_report_validator.py index 85d7b4e..62f95a7 100644 --- a/final_report_validator.py +++ b/final_report_validator.py @@ -11,6 +11,7 @@ import inspect import re from typing import Any, Callable +import branch_cleanup_guard import issue_lock_provenance from post_merge_cleanup_proof import assess_post_merge_cleanup_proof from review_proofs import ( @@ -1024,6 +1025,19 @@ 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_review_mutation( report_text: str, *, @@ -1061,6 +1075,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, ) _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = { diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 411a80b..d8db5f6 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -615,6 +615,7 @@ import reconciliation_workflow # noqa: E402 import review_merge_state_machine # noqa: E402 import pr_work_lease # noqa: E402 import native_mcp_preference # noqa: E402 +import branch_cleanup_guard # noqa: E402 # Keyed issue-lock storage (#443): per remote/org/repo/issue files under @@ -4116,6 +4117,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 diff --git a/role_session_router.py b/role_session_router.py index 8f30dc5..48dbdd2 100644 --- a/role_session_router.py +++ b/role_session_router.py @@ -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"] \ No newline at end of file + return assess_infra_stop(project_root)["infra_stop"] diff --git a/skills/llm-project-workflow/workflows/review-merge-pr.md b/skills/llm-project-workflow/workflows/review-merge-pr.md index 76a7000..e035fd4 100644 --- a/skills/llm-project-workflow/workflows/review-merge-pr.md +++ b/skills/llm-project-workflow/workflows/review-merge-pr.md @@ -871,6 +871,15 @@ 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 ` and `git push --delete ` 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. + Do not delete or mutate unrelated branches/worktrees. Do not touch the main checkout except to update the stable branch after merge if explicitly allowed by the workflow. diff --git a/task_capability_map.py b/task_capability_map.py index 1d40a78..c158cee 100644 --- a/task_capability_map.py +++ b/task_capability_map.py @@ -92,6 +92,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", diff --git a/tests/test_branch_cleanup_guard.py b/tests/test_branch_cleanup_guard.py new file mode 100644 index 0000000..dd4053f --- /dev/null +++ b/tests/test_branch_cleanup_guard.py @@ -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() From 9b96085d8d5fd6b7f54a3cac8573a261aad40972 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Thu, 9 Jul 2026 10:09:38 -0400 Subject: [PATCH 03/12] 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 --- reviewer_pr_lease.py | 32 +++++++++++++++++---------- tests/test_reviewer_pr_lease.py | 39 +++++++++++++++++++++++++++++++++ 2 files changed, 60 insertions(+), 11 deletions(-) diff --git a/reviewer_pr_lease.py b/reviewer_pr_lease.py index f55390b..2310843 100644 --- a/reviewer_pr_lease.py +++ b/reviewer_pr_lease.py @@ -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 diff --git a/tests/test_reviewer_pr_lease.py b/tests/test_reviewer_pr_lease.py index 97e5a2e..875d562 100644 --- a/tests/test_reviewer_pr_lease.py +++ b/tests/test_reviewer_pr_lease.py @@ -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() From 223cde1b941a6ead9d6435e11235010ee6222c05 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Thu, 9 Jul 2026 10:29:03 -0400 Subject: [PATCH 04/12] feat: terminal launcher diagnostics and mutation preflight (Closes #556) Add categorize-and-probe helpers for shell spawn failures, expose gitea_diagnose_terminal, and fail closed on mutation capability resolve when the terminal launcher is unhealthy (BLOCKED + DIAGNOSE). Document the operator path in the workflow runbooks. --- docs/llm-workflow-runbooks.md | 15 +++ gitea_mcp_server.py | 67 +++++++++++ native_mcp_preference.py | 155 +++++++++++++++++++++++++- tests/test_native_mcp_preference.py | 64 ++++++++++- tests/test_resolve_task_capability.py | 41 +++++++ 5 files changed, 340 insertions(+), 2 deletions(-) diff --git a/docs/llm-workflow-runbooks.md b/docs/llm-workflow-runbooks.md index ea2f64e..2a91259 100644 --- a/docs/llm-workflow-runbooks.md +++ b/docs/llm-workflow-runbooks.md @@ -661,6 +661,21 @@ session, clearing hung background terminals, switching to MCP-native commit, and the agent temp artifact cleanup checklist. Do **not** retry shell encoding in a loop and do **not** substitute WebFetch/Playwright/manual base64. +### Terminal launcher diagnostics (#556) + +When git/pytest finalization fails with opaque spawn errors (e.g. `os error 2`), +do **not** improvise shell wrappers or fall back to direct API / temp scripts. + +1. Call `gitea_diagnose_terminal` (optional `cwd`, optional `command`) — returns + categorized failure: missing cwd, cwd not a directory, missing executable, + missing runtime wrapper, missing shell, probe timeout, or session launcher + failure. +2. Call `gitea_get_shell_health` for the shell circuit-breaker state. +3. Mutation-capable `gitea_resolve_task_capability` probes the terminal launcher + before allowing git/pytest-oriented work; on failure it returns + `BLOCKED + DIAGNOSE` with `terminal_launcher_unhealthy` and diagnostics. +4. Emit the canonical `blocked-diagnose-report.md` template and stop. + ### Create an issue / child issues - **Profile:** issue-manager or author (any profile allowed to create issues). diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 099021f..0c05920 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -7906,6 +7906,36 @@ def gitea_get_shell_health() -> dict: return native_mcp_preference.shell_health_status() +@mcp.tool() +def gitea_diagnose_terminal( + cwd: str | None = None, + command: list[str] | None = None, +) -> dict: + """Read-only: Run active diagnostics on terminal launcher spawn ability (#556). + + Args: + cwd: Active worktree or current directory to test (defaults to GITEA_ACTIVE_WORKTREE). + command: Test command to attempt (defaults to ['git', '--version'] or ['echo', 'ok']). + """ + resolved_cwd = cwd or os.environ.get("GITEA_ACTIVE_WORKTREE") + if resolved_cwd: + resolved_cwd = os.path.realpath(resolved_cwd) + + probe_cmd = command or native_mcp_preference.default_terminal_probe_command() + + diag = native_mcp_preference.diagnose_terminal_failure(resolved_cwd, probe_cmd) + probe_res = native_mcp_preference.probe_terminal_spawn(resolved_cwd, probe_cmd) + + return { + "healthy": probe_res["healthy"], + "error_type": None if probe_res["healthy"] else probe_res["error_type"] or diag["error_type"], + "error_msg": "" if probe_res["healthy"] else probe_res["error_msg"] or diag["error_msg"], + "cwd": resolved_cwd, + "command": probe_cmd, + "diagnostics": diag, + } + + @mcp.tool() def gitea_validate_review_final_report( report_text: str, @@ -9412,6 +9442,43 @@ def gitea_resolve_task_capability( "exact_safe_next_action": next_safe_action, } + # ── Terminal Launcher Preflight Check (#556) ── + if task in native_mcp_preference.GITEA_MUTATION_TASKS: + in_test = _preflight_in_test_mode() + if not in_test or os.environ.get("GITEA_FORCE_TERMINAL_PROBE") == "1": + resolved_cwd = os.environ.get("GITEA_ACTIVE_WORKTREE") or PROJECT_ROOT + probe_res = native_mcp_preference.probe_terminal_spawn(resolved_cwd) + if not probe_res["healthy"]: + profile = get_profile() + h = host or (REMOTES.get(remote, {}).get("host") if remote in REMOTES else None) + username = _authenticated_username(h) if h else None + next_safe_action = ( + "BLOCKED + DIAGNOSE: terminal-launcher-failure " + f"({probe_res['error_type']}): {probe_res['error_msg']}. " + "Do not attempt git/pytest finalization or unsafe fallbacks. " + "Run gitea_diagnose_terminal, emit blocked-diagnose-report, and stop." + ) + return { + "requested_task": task, + "required_operation_permission": required_permission, + "required_role_kind": required_role, + "active_profile": profile["profile_name"], + "active_identity": username, + "active_profile_allowed_operations": profile.get("allowed_operations") or [], + "allowed_in_current_session": False, + "stop_required": True, + "infra_stop": True, + "blocked": True, + "blocked_reason": "BLOCKED + DIAGNOSE", + "terminal_launcher_unhealthy": True, + "terminal_launcher_diagnostics": probe_res, + "task_role_guidance": [next_safe_action], + "matching_configured_profile": [], + "runtime_switching_supported": gitea_config.is_runtime_switching_enabled(), + "different_mcp_namespace_required": False, + "exact_safe_next_action": next_safe_action, + } + record_preflight_check("capability", required_role, resolved_task=task) # Try automatic dispatch switching diff --git a/native_mcp_preference.py b/native_mcp_preference.py index 872b800..2e3d1ee 100644 --- a/native_mcp_preference.py +++ b/native_mcp_preference.py @@ -8,6 +8,9 @@ available unless explicit recovery-mode proof is supplied. from __future__ import annotations import re +import os +import shutil +import subprocess from typing import Any from reviewer_fallback import LOCAL_GITEA_SCRIPT_NAMES @@ -366,4 +369,154 @@ def assess_gitea_operation_path( else "proceed with native MCP" ) ), - } \ No newline at end of file + } + + +def default_terminal_probe_command() -> list[str]: + return ["git", "--version"] if shutil.which("git") else ["echo", "ok"] + + +def _resolve_executable(cwd: str | None, executable: str | None) -> str | None: + if not executable: + return None + if os.path.isabs(executable): + return executable if os.path.exists(executable) and os.access(executable, os.X_OK) else None + if "/" in executable or "\\" in executable: + target_cwd = cwd or os.getcwd() + full_path = os.path.abspath(os.path.join(target_cwd, executable)) + return full_path if os.path.exists(full_path) and os.access(full_path, os.X_OK) else None + return shutil.which(executable) + + +def diagnose_terminal_failure(cwd: str | None, command: list[str]) -> dict[str, Any]: + """Inspect and categorize command spawn failures (#556).""" + diagnostics = { + "cwd": cwd, + "command": command, + "cwd_exists": True, + "cwd_is_dir": True, + "shell_exists": True, + "executable_exists": True, + "resolved_executable": None, + "error_type": "session launcher failure", + "error_msg": ( + "Subprocess spawn failed despite valid CWD and executable. This may be due " + "to sandboxing, permission restrictions, or system resource limits." + ), + } + + if cwd: + if not os.path.exists(cwd): + diagnostics["cwd_exists"] = False + diagnostics["error_type"] = "missing cwd" + diagnostics["error_msg"] = f"Current working directory does not exist: {cwd}" + return diagnostics + if not os.path.isdir(cwd): + diagnostics["cwd_is_dir"] = False + diagnostics["error_type"] = "cwd is not a directory" + diagnostics["error_msg"] = f"Current working directory is not a directory: {cwd}" + return diagnostics + + exe = command[0] if command else None + if exe: + resolved_exe = _resolve_executable(cwd, exe) + diagnostics["resolved_executable"] = resolved_exe + if not resolved_exe: + diagnostics["executable_exists"] = False + if "venv" in exe or "pytest" in exe: + diagnostics["error_type"] = "missing runtime wrapper" + diagnostics["error_msg"] = ( + "Virtual environment runtime wrapper/executable not found or not " + f"executable: {exe}" + ) + else: + diagnostics["error_type"] = "missing executable" + diagnostics["error_msg"] = f"Executable not found in PATH or CWD: {exe}" + return diagnostics + + # Check common shell availability + has_any_shell = False + for shell_path in ("/bin/sh", "/bin/zsh", "/bin/bash"): + if os.path.exists(shell_path) and os.access(shell_path, os.X_OK): + has_any_shell = True + break + if not has_any_shell: + diagnostics["shell_exists"] = False + diagnostics["error_type"] = "missing shell" + diagnostics["error_msg"] = "No standard system shell (/bin/sh, /bin/zsh, /bin/bash) is available or executable." + return diagnostics + + return diagnostics + + +def probe_terminal_spawn( + cwd: str | None = None, + command: list[str] | None = None, +) -> dict[str, Any]: + """Proactively probe command spawning to detect launcher health (#556).""" + probe_cmd = command or default_terminal_probe_command() + diag = diagnose_terminal_failure(cwd, probe_cmd) + if diag["error_type"] != "session launcher failure": + return { + "healthy": False, + "error_type": diag["error_type"], + "error_msg": diag["error_msg"], + "cwd": cwd, + "command": probe_cmd, + "diagnostics": diag, + } + + try: + proc = subprocess.run( + probe_cmd, + capture_output=True, + text=True, + cwd=cwd, + timeout=5, + ) + if proc.returncode == 0: + return { + "healthy": True, + "error_type": None, + "error_msg": "", + "cwd": cwd, + "command": probe_cmd, + "diagnostics": diag, + } + # Non-zero status still proves the launcher spawned the process. + return { + "healthy": True, + "error_type": None, + "error_msg": "", + "cwd": cwd, + "command": probe_cmd, + "exit_code": proc.returncode, + "diagnostics": diag, + } + except FileNotFoundError: + return { + "healthy": False, + "error_type": diag["error_type"], + "error_msg": diag["error_msg"], + "cwd": cwd, + "command": probe_cmd, + "diagnostics": diag, + } + except subprocess.TimeoutExpired as exc: + return { + "healthy": False, + "error_type": "probe timeout", + "error_msg": f"Subprocess probe timed out after {exc.timeout} seconds.", + "cwd": cwd, + "command": probe_cmd, + "diagnostics": diag, + } + except Exception as exc: + return { + "healthy": False, + "error_type": diag["error_type"], + "error_msg": f"Subprocess spawn failed with exception: {exc}", + "cwd": cwd, + "command": probe_cmd, + "diagnostics": diag, + } diff --git a/tests/test_native_mcp_preference.py b/tests/test_native_mcp_preference.py index 7607814..82ec6f0 100644 --- a/tests/test_native_mcp_preference.py +++ b/tests/test_native_mcp_preference.py @@ -26,6 +26,68 @@ class TestShellHealthCircuitBreaker(unittest.TestCase): self.assertEqual(status["consecutive_spawn_failures"], 0) +class TestTerminalDiagnostics(unittest.TestCase): + def test_diagnose_missing_cwd(self): + res = nmp.diagnose_terminal_failure("/nonexistent_directory_for_test", ["git", "status"]) + self.assertFalse(res["cwd_exists"]) + self.assertEqual(res["error_type"], "missing cwd") + self.assertIn("does not exist", res["error_msg"]) + + def test_diagnose_cwd_is_not_dir(self): + import tempfile + import os + with tempfile.NamedTemporaryFile() as tmp: + res = nmp.diagnose_terminal_failure(tmp.name, ["git", "status"]) + self.assertFalse(res["cwd_is_dir"]) + self.assertEqual(res["error_type"], "cwd is not a directory") + + def test_diagnose_missing_executable(self): + res = nmp.diagnose_terminal_failure(None, ["nonexistent_exec_for_test"]) + self.assertFalse(res["executable_exists"]) + self.assertEqual(res["error_type"], "missing executable") + + def test_diagnose_missing_runtime_wrapper(self): + res = nmp.diagnose_terminal_failure("/tmp", ["venv/bin/pytest"]) + self.assertFalse(res["executable_exists"]) + self.assertEqual(res["error_type"], "missing runtime wrapper") + + def test_diagnose_relative_executable_from_cwd(self): + import os + import tempfile + + with tempfile.TemporaryDirectory() as tmp: + script = Path(tmp) / "tool" + script.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8") + os.chmod(script, 0o755) + + res = nmp.diagnose_terminal_failure(tmp, ["./tool"]) + self.assertTrue(res["executable_exists"]) + self.assertEqual(res["resolved_executable"], str(script)) + + def test_probe_terminal_spawn_success(self): + res = nmp.probe_terminal_spawn() + self.assertTrue(res["healthy"]) + self.assertIn("command", res) + self.assertIn("diagnostics", res) + + def test_probe_terminal_spawn_missing_cwd(self): + res = nmp.probe_terminal_spawn("/nonexistent_directory_for_test") + self.assertFalse(res["healthy"]) + self.assertEqual(res["error_type"], "missing cwd") + + def test_probe_terminal_spawn_honors_custom_command(self): + res = nmp.probe_terminal_spawn(command=[sys.executable, "-c", "raise SystemExit(3)"]) + self.assertTrue(res["healthy"]) + self.assertEqual(res["command"][0], sys.executable) + self.assertEqual(res["exit_code"], 3) + + def test_probe_terminal_spawn_blocks_missing_custom_command(self): + res = nmp.probe_terminal_spawn(command=["nonexistent_exec_for_test"]) + self.assertFalse(res["healthy"]) + self.assertEqual(res["error_type"], "missing executable") + self.assertEqual(res["command"], ["nonexistent_exec_for_test"]) + + class TestNativeMcpPreferenceGate(unittest.TestCase): def setUp(self): nmp.clear_shell_health_for_tests() @@ -118,4 +180,4 @@ class TestNativeMcpPreferenceGate(unittest.TestCase): if __name__ == "__main__": - unittest.main() \ No newline at end of file + unittest.main() diff --git a/tests/test_resolve_task_capability.py b/tests/test_resolve_task_capability.py index aa25ced..cdb2e9d 100644 --- a/tests/test_resolve_task_capability.py +++ b/tests/test_resolve_task_capability.py @@ -96,6 +96,14 @@ class TestResolveTaskCapability(unittest.TestCase): "GITEA_TOKEN_MERGER": "merger-pass", } + def test_diagnose_terminal_success_has_no_error(self): + res = mcp_server.gitea_diagnose_terminal( + command=[sys.executable, "-c", "raise SystemExit(0)"] + ) + self.assertTrue(res["healthy"]) + self.assertIsNone(res["error_type"]) + self.assertEqual(res["error_msg"], "") + @patch("mcp_server.api_request", return_value={"login": "author-user"}) @patch("mcp_server.get_auth_header", return_value="token author-pass") def test_resolve_author_capable_task(self, _auth, _api): @@ -110,6 +118,39 @@ class TestResolveTaskCapability(unittest.TestCase): self.assertFalse(res["different_mcp_namespace_required"]) self.assertIn("ready for operations", res["exact_safe_next_action"]) + @patch("mcp_server.native_mcp_preference.probe_terminal_spawn") + @patch("mcp_server.api_request", return_value={"login": "author-user"}) + @patch("mcp_server.get_auth_header", return_value="token author-pass") + def test_resolve_mutation_task_blocks_when_terminal_launcher_unhealthy( + self, + _auth, + _api, + mock_probe, + ): + mock_probe.return_value = { + "healthy": False, + "error_type": "missing cwd", + "error_msg": "Current working directory does not exist: /missing", + "cwd": "/missing", + "command": ["git", "--version"], + "diagnostics": {"cwd_exists": False}, + } + env = self._env("author-profile") + # Probe is skipped under pytest unless forced (mirrors production path). + env["GITEA_FORCE_TERMINAL_PROBE"] = "1" + with patch.dict(os.environ, env): + res = mcp_server.gitea_resolve_task_capability(task="create_issue", remote="prgs") + + self.assertFalse(res["allowed_in_current_session"]) + self.assertTrue(res["stop_required"]) + self.assertTrue(res["infra_stop"]) + self.assertTrue(res["terminal_launcher_unhealthy"]) + self.assertTrue(res["blocked"]) + self.assertEqual(res["blocked_reason"], "BLOCKED + DIAGNOSE") + self.assertEqual(res["terminal_launcher_diagnostics"]["error_type"], "missing cwd") + self.assertIn("terminal-launcher-failure", res["exact_safe_next_action"]) + self.assertIn("BLOCKED + DIAGNOSE", res["exact_safe_next_action"]) + @patch("mcp_server.api_request", return_value={"login": "author-user"}) @patch("mcp_server.get_auth_header", return_value="token author-pass") def test_resolve_reviewer_capable_task_blocked(self, _auth, _api): From f558b80cc8641fd60c8cb1f4ea9f9beee5f32f72 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Thu, 9 Jul 2026 10:43:42 -0400 Subject: [PATCH 05/12] test: cover explicit git -C terminal probe --- gitea_mcp_server.py | 12 ++++++++++-- native_mcp_preference.py | 13 ++++++++++--- tests/test_native_mcp_preference.py | 25 ++++++++++++++++++++++++- 3 files changed, 44 insertions(+), 6 deletions(-) diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 0c05920..c960185 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -7928,8 +7928,16 @@ def gitea_diagnose_terminal( return { "healthy": probe_res["healthy"], - "error_type": None if probe_res["healthy"] else probe_res["error_type"] or diag["error_type"], - "error_msg": "" if probe_res["healthy"] else probe_res["error_msg"] or diag["error_msg"], + "error_type": ( + None + if probe_res["healthy"] + else probe_res["error_type"] or diag["error_type"] + ), + "error_msg": ( + "" + if probe_res["healthy"] + else probe_res["error_msg"] or diag["error_msg"] + ), "cwd": resolved_cwd, "command": probe_cmd, "diagnostics": diag, diff --git a/native_mcp_preference.py b/native_mcp_preference.py index 2e3d1ee..60cb420 100644 --- a/native_mcp_preference.py +++ b/native_mcp_preference.py @@ -380,11 +380,15 @@ def _resolve_executable(cwd: str | None, executable: str | None) -> str | None: if not executable: return None if os.path.isabs(executable): - return executable if os.path.exists(executable) and os.access(executable, os.X_OK) else None + if os.path.exists(executable) and os.access(executable, os.X_OK): + return executable + return None if "/" in executable or "\\" in executable: target_cwd = cwd or os.getcwd() full_path = os.path.abspath(os.path.join(target_cwd, executable)) - return full_path if os.path.exists(full_path) and os.access(full_path, os.X_OK) else None + if os.path.exists(full_path) and os.access(full_path, os.X_OK): + return full_path + return None return shutil.which(executable) @@ -443,7 +447,10 @@ def diagnose_terminal_failure(cwd: str | None, command: list[str]) -> dict[str, if not has_any_shell: diagnostics["shell_exists"] = False diagnostics["error_type"] = "missing shell" - diagnostics["error_msg"] = "No standard system shell (/bin/sh, /bin/zsh, /bin/bash) is available or executable." + diagnostics["error_msg"] = ( + "No standard system shell (/bin/sh, /bin/zsh, /bin/bash) is " + "available or executable." + ) return diagnostics return diagnostics diff --git a/tests/test_native_mcp_preference.py b/tests/test_native_mcp_preference.py index 82ec6f0..9566759 100644 --- a/tests/test_native_mcp_preference.py +++ b/tests/test_native_mcp_preference.py @@ -1,6 +1,7 @@ """Tests for native MCP preference gate and shell circuit breaker (#270).""" import sys import unittest +import shutil from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) @@ -76,11 +77,33 @@ class TestTerminalDiagnostics(unittest.TestCase): self.assertEqual(res["error_type"], "missing cwd") def test_probe_terminal_spawn_honors_custom_command(self): - res = nmp.probe_terminal_spawn(command=[sys.executable, "-c", "raise SystemExit(3)"]) + res = nmp.probe_terminal_spawn( + command=[sys.executable, "-c", "raise SystemExit(3)"] + ) self.assertTrue(res["healthy"]) self.assertEqual(res["command"][0], sys.executable) self.assertEqual(res["exit_code"], 3) + @unittest.skipUnless(shutil.which("git"), "git is required for git -C probe") + def test_probe_terminal_spawn_supports_explicit_git_c_worktree(self): + import subprocess + import tempfile + + with tempfile.TemporaryDirectory() as tmp: + subprocess.run( + ["git", "init", tmp], + check=True, + capture_output=True, + text=True, + ) + res = nmp.probe_terminal_spawn( + cwd="/", + command=["git", "-C", tmp, "status", "--short"], + ) + + self.assertTrue(res["healthy"]) + self.assertEqual(res["command"][1], "-C") + def test_probe_terminal_spawn_blocks_missing_custom_command(self): res = nmp.probe_terminal_spawn(command=["nonexistent_exec_for_test"]) self.assertFalse(res["healthy"]) From 8d6d3cb014ed26bacc63ad336fb8b97c3d3bd32b Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Thu, 9 Jul 2026 11:31:48 -0400 Subject: [PATCH 06/12] fix: refine terminal preflight check to avoid false-positive PATH blocks on server --- gitea_mcp_server.py | 63 +++++++++++++++++++++++++-------------------- 1 file changed, 35 insertions(+), 28 deletions(-) diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index c960185..241b278 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -9457,35 +9457,42 @@ def gitea_resolve_task_capability( resolved_cwd = os.environ.get("GITEA_ACTIVE_WORKTREE") or PROJECT_ROOT probe_res = native_mcp_preference.probe_terminal_spawn(resolved_cwd) if not probe_res["healthy"]: - profile = get_profile() - h = host or (REMOTES.get(remote, {}).get("host") if remote in REMOTES else None) - username = _authenticated_username(h) if h else None - next_safe_action = ( - "BLOCKED + DIAGNOSE: terminal-launcher-failure " - f"({probe_res['error_type']}): {probe_res['error_msg']}. " - "Do not attempt git/pytest finalization or unsafe fallbacks. " - "Run gitea_diagnose_terminal, emit blocked-diagnose-report, and stop." + # Only block if the failure is due to a missing CWD, missing shell, or the circuit breaker is tripped. + # Other failures (like missing executable in the MCP process's limited PATH) should not block the agent's shell capability. + is_real_blocker = ( + probe_res.get("error_type") in ("missing cwd", "missing shell") + or native_mcp_preference.shell_health_status().get("hard_stopped") ) - return { - "requested_task": task, - "required_operation_permission": required_permission, - "required_role_kind": required_role, - "active_profile": profile["profile_name"], - "active_identity": username, - "active_profile_allowed_operations": profile.get("allowed_operations") or [], - "allowed_in_current_session": False, - "stop_required": True, - "infra_stop": True, - "blocked": True, - "blocked_reason": "BLOCKED + DIAGNOSE", - "terminal_launcher_unhealthy": True, - "terminal_launcher_diagnostics": probe_res, - "task_role_guidance": [next_safe_action], - "matching_configured_profile": [], - "runtime_switching_supported": gitea_config.is_runtime_switching_enabled(), - "different_mcp_namespace_required": False, - "exact_safe_next_action": next_safe_action, - } + if is_real_blocker: + profile = get_profile() + h = host or (REMOTES.get(remote, {}).get("host") if remote in REMOTES else None) + username = _authenticated_username(h) if h else None + next_safe_action = ( + "BLOCKED + DIAGNOSE: terminal-launcher-failure " + f"({probe_res['error_type']}): {probe_res['error_msg']}. " + "Do not attempt git/pytest finalization or unsafe fallbacks. " + "Run gitea_diagnose_terminal, emit blocked-diagnose-report, and stop." + ) + return { + "requested_task": task, + "required_operation_permission": required_permission, + "required_role_kind": required_role, + "active_profile": profile["profile_name"], + "active_identity": username, + "active_profile_allowed_operations": profile.get("allowed_operations") or [], + "allowed_in_current_session": False, + "stop_required": True, + "infra_stop": True, + "blocked": True, + "blocked_reason": "BLOCKED + DIAGNOSE", + "terminal_launcher_unhealthy": True, + "terminal_launcher_diagnostics": probe_res, + "task_role_guidance": [next_safe_action], + "matching_configured_profile": [], + "runtime_switching_supported": gitea_config.is_runtime_switching_enabled(), + "different_mcp_namespace_required": False, + "exact_safe_next_action": next_safe_action, + } record_preflight_check("capability", required_role, resolved_task=task) From 975674d7f5e2d40b3a3e09ddaf9f17483f4cbf2f Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 15:37:32 -0400 Subject: [PATCH 07/12] feat(webui): auth and deployment boundary (#435) Add bind-host assessment that refuses 0.0.0.0/:: without override, warns on non-loopback binds, and documents internal-only MVP serving. Health endpoint exposes deployment metadata; docs cover Access/VPN/WARP and runtime env assumptions without embedding secrets in the client. Closes #435 --- docs/webui-deployment.md | 59 +++++++++ docs/webui-local-dev.md | 15 +++ tests/test_webui_deployment_boundary.py | 106 ++++++++++++++++ tests/test_webui_skeleton.py | 2 + webui/__main__.py | 14 +++ webui/app.py | 14 ++- webui/deployment_boundary.py | 155 ++++++++++++++++++++++++ 7 files changed, 362 insertions(+), 3 deletions(-) create mode 100644 docs/webui-deployment.md create mode 100644 tests/test_webui_deployment_boundary.py create mode 100644 webui/deployment_boundary.py diff --git a/docs/webui-deployment.md b/docs/webui-deployment.md new file mode 100644 index 0000000..22b3906 --- /dev/null +++ b/docs/webui-deployment.md @@ -0,0 +1,59 @@ +# Web UI deployment boundary (#435) + +The MCP Control Plane web UI is an **internal operator console**, not a +customer-facing application. The MVP assumes local or trusted-network access +only. + +## MVP deployment model + +- **Default bind:** `127.0.0.1:8765` (`WEBUI_HOST` / `WEBUI_PORT`) +- **Authentication:** none in MVP — protection comes from network placement +- **Mutations:** read-only routes; gated write actions remain disabled (#434) +- **Secrets:** resolved server-side via `gitea_auth` / `GITEA_MCP_CONFIG`; never + embedded in HTML, JavaScript, or browser storage + +Do **not** expose the UI on the public internet without an access layer. + +## Beyond localhost + +If the UI must be reachable outside the operator laptop: + +1. Prefer **Cloudflare Access**, **Cloudflare WARP**, or an org **VPN** so only + authenticated staff reach the service. +2. Bind to a specific interface only when necessary — never `0.0.0.0` / `::` + without understanding the exposure. +3. Set explicit override env vars only after access controls are in place: + - `WEBUI_ALLOW_PUBLIC_BIND=1` — acknowledges all-interface bind (`0.0.0.0`, `::`) + - `WEBUI_ALLOW_REMOTE_BIND=1` — acknowledges a non-loopback host + +Startup **refuses** all-interface binds unless `WEBUI_ALLOW_PUBLIC_BIND=1`. +Non-loopback binds log a warning unless `WEBUI_ALLOW_REMOTE_BIND=1`. + +## Environment variables + +| Variable | Default | Purpose | +|----------|---------|---------| +| `WEBUI_HOST` | `127.0.0.1` | Bind address | +| `WEBUI_PORT` | `8765` | Listen port | +| `WEBUI_REPO_ROOT` | repository root | Workflow/schema hash root for prompt library | +| `WEBUI_PROJECT_REGISTRY` | packaged JSON | Project registry file path | +| `GITEA_MCP_CONFIG` | unset | Server-side MCP profile config path (optional) | +| `GITEA_MCP_PROFILE` | unset | Active MCP profile name (optional) | +| `WEBUI_ALLOW_PUBLIC_BIND` | unset | Acknowledge `0.0.0.0` / `::` bind | +| `WEBUI_ALLOW_REMOTE_BIND` | unset | Acknowledge non-loopback bind | + +Gitea credentials (`GITEA_TOKEN_*`, `.env.*`, keychain refs) are read only on +the server when a page needs live Gitea data (e.g. `/queue`). They are not +shipped to the browser. + +## Health / deployment metadata + +`GET /health` includes a `deployment` object with bind disposition, runtime +assumption paths, and the client-secret policy. Use it to verify an instance is +configured for internal-only operation. + +## Non-goals (MVP) + +- Full SSO or session login in the UI +- Hosting on the public internet without Access/VPN/WARP +- Embedding Gitea tokens in the frontend bundle \ No newline at end of file diff --git a/docs/webui-local-dev.md b/docs/webui-local-dev.md index 990914e..73c4f47 100644 --- a/docs/webui-local-dev.md +++ b/docs/webui-local-dev.md @@ -29,6 +29,13 @@ Optional environment variables: |----------|---------|---------| | `WEBUI_HOST` | `127.0.0.1` | Bind address (keep local for MVP) | | `WEBUI_PORT` | `8765` | Listen port | +| `WEBUI_REPO_ROOT` | repository root | Prompt library workflow hash root | +| `WEBUI_PROJECT_REGISTRY` | packaged JSON | Project registry path | +| `GITEA_MCP_CONFIG` | unset | Server-side MCP profile config (never sent to browser) | +| `GITEA_MCP_PROFILE` | unset | Active MCP profile name (server-side only) | + +See [webui-deployment.md](webui-deployment.md) for internal-only serving, +Cloudflare Access/WARP/VPN guidance, and unsafe bind overrides (#435). ## Routes (MVP) @@ -134,10 +141,18 @@ 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. +## Deployment boundary (#435) + +MVP serves on loopback by default. Binding `0.0.0.0` or `::` is **refused** +unless `WEBUI_ALLOW_PUBLIC_BIND=1`. Non-loopback hosts log a warning unless +`WEBUI_ALLOW_REMOTE_BIND=1`. `GET /health` exposes `deployment` metadata. + ## 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 + +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_deployment_boundary.py -q ``` \ No newline at end of file diff --git a/tests/test_webui_deployment_boundary.py b/tests/test_webui_deployment_boundary.py new file mode 100644 index 0000000..9f5b625 --- /dev/null +++ b/tests/test_webui_deployment_boundary.py @@ -0,0 +1,106 @@ +"""Tests for web UI deployment and auth boundary (#435).""" +import os +import subprocess +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from starlette.testclient import TestClient + +from webui.app import create_app +from webui.deployment_boundary import ( + assess_bind_host, + runtime_assumptions, + scan_text_for_client_secrets, +) + + +class TestBindAssessment(unittest.TestCase): + def test_loopback_is_safe(self): + for host in ("127.0.0.1", "localhost", "::1"): + with self.subTest(host=host): + result = assess_bind_host(host) + self.assertEqual(result.disposition, "safe") + + def test_all_interfaces_refused_by_default(self): + for host in ("0.0.0.0", "::", "*"): + with self.subTest(host=host): + result = assess_bind_host(host) + self.assertEqual(result.disposition, "refuse") + + def test_all_interfaces_allowed_with_override(self): + prior = os.environ.pop("WEBUI_ALLOW_PUBLIC_BIND", None) + try: + os.environ["WEBUI_ALLOW_PUBLIC_BIND"] = "1" + result = assess_bind_host("0.0.0.0") + self.assertEqual(result.disposition, "warn") + finally: + os.environ.pop("WEBUI_ALLOW_PUBLIC_BIND", None) + if prior is not None: + os.environ["WEBUI_ALLOW_PUBLIC_BIND"] = prior + + def test_non_loopback_warns_without_override(self): + prior = os.environ.pop("WEBUI_ALLOW_REMOTE_BIND", None) + try: + os.environ.pop("WEBUI_ALLOW_REMOTE_BIND", None) + result = assess_bind_host("192.168.1.10") + self.assertEqual(result.disposition, "warn") + finally: + if prior is not None: + os.environ["WEBUI_ALLOW_REMOTE_BIND"] = prior + + def test_runtime_assumptions_exclude_secrets(self): + assumptions = runtime_assumptions() + blob = " ".join(str(v) for v in assumptions.values()) + self.assertNotIn("GITEA_TOKEN", blob) + self.assertIn("internal-operator-console", assumptions["deployment_mode"]) + + +class TestDeploymentRoutes(unittest.TestCase): + def setUp(self): + self.client = TestClient(create_app(bind_host="127.0.0.1")) + + def test_health_includes_deployment_metadata(self): + response = self.client.get("/health") + self.assertEqual(response.status_code, 200) + data = response.json() + self.assertIn("deployment", data) + deployment = data["deployment"] + self.assertEqual(deployment["mode"], "internal-operator-console") + self.assertEqual(deployment["mvp_auth"], "none") + self.assertEqual(deployment["bind"]["disposition"], "safe") + self.assertIn("runtime_assumptions", deployment) + + def test_rendered_pages_contain_no_embedded_secrets(self): + for path in ("/", "/projects", "/prompts", "/queue"): + with self.subTest(path=path): + text = self.client.get(path).text + self.assertEqual(scan_text_for_client_secrets(text), []) + + +class TestStartupRefusal(unittest.TestCase): + def test_refuses_all_interface_bind(self): + env = {**os.environ, "WEBUI_HOST": "0.0.0.0"} + env.pop("WEBUI_ALLOW_PUBLIC_BIND", None) + repo_root = Path(__file__).resolve().parent.parent + result = subprocess.run( + [sys.executable, "-m", "webui"], + cwd=repo_root, + env=env, + capture_output=True, + text=True, + timeout=3, + ) + self.assertEqual(result.returncode, 1) + + +class TestClientSecretScanner(unittest.TestCase): + def test_detects_token_like_content(self): + findings = scan_text_for_client_secrets("var x = GITEA_TOKEN_PRGS") + self.assertTrue(findings) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file diff --git a/tests/test_webui_skeleton.py b/tests/test_webui_skeleton.py index d9de18e..07593d8 100644 --- a/tests/test_webui_skeleton.py +++ b/tests/test_webui_skeleton.py @@ -22,6 +22,8 @@ class TestWebuiSkeleton(unittest.TestCase): self.assertEqual(data["service"], "mcp-control-plane-webui") self.assertEqual(data["mode"], "read-only-mvp") self.assertIn("timestamp", data) + self.assertIn("deployment", data) + self.assertEqual(data["deployment"]["mode"], "internal-operator-console") def test_home_renders(self): response = self.client.get("/") diff --git a/webui/__main__.py b/webui/__main__.py index c55b05e..5c427da 100644 --- a/webui/__main__.py +++ b/webui/__main__.py @@ -2,16 +2,30 @@ from __future__ import annotations +import logging import os import uvicorn from webui.app import create_app +from webui.deployment_boundary import assess_bind_host + +logger = logging.getLogger("webui") + + +def _validate_bind_host(host: str) -> None: + assessment = assess_bind_host(host) + if assessment.disposition == "refuse": + logger.error("webui bind refused: %s", assessment.message) + raise SystemExit(1) + if assessment.disposition == "warn": + logger.warning("webui bind warning: %s", assessment.message) def main() -> None: host = os.environ.get("WEBUI_HOST", "127.0.0.1") port = int(os.environ.get("WEBUI_PORT", "8765")) + _validate_bind_host(host) uvicorn.run(create_app(), host=host, port=port, log_level="info") diff --git a/webui/app.py b/webui/app.py index 4ab38ed..d0f832b 100644 --- a/webui/app.py +++ b/webui/app.py @@ -9,6 +9,7 @@ from starlette.requests import Request from starlette.responses import HTMLResponse, JSONResponse, Response from starlette.routing import Route +from webui.deployment_boundary import deployment_snapshot from webui.layout import render_page from webui.project_registry import find_project, load_registry, registry_to_dict from webui.project_views import render_project_detail, render_projects_list @@ -61,11 +62,13 @@ async def home(_request: Request) -> HTMLResponse: async def health(_request: Request) -> JSONResponse: + bind_host = _request.app.state.webui_bind_host return JSONResponse({ "status": "ok", "service": "mcp-control-plane-webui", "mode": "read-only-mvp", "timestamp": datetime.now(timezone.utc).isoformat(), + "deployment": deployment_snapshot(bind_host=bind_host), }) @@ -250,9 +253,12 @@ async def method_not_allowed(request: Request, _exc: Exception) -> Response: return JSONResponse({"error": "not_found"}, status_code=404) -def create_app() -> Starlette: +def create_app(*, bind_host: str | None = None) -> Starlette: """Build the read-only MVP Starlette app.""" - return Starlette( + import os + + resolved_bind = bind_host or os.environ.get("WEBUI_HOST", "127.0.0.1") + app = Starlette( debug=False, routes=[ Route("/", home, methods=["GET"]), @@ -287,4 +293,6 @@ def create_app() -> Starlette: Route("/api/leases", api_leases, methods=["GET"]), ], exception_handlers={405: method_not_allowed}, - ) \ No newline at end of file + ) + app.state.webui_bind_host = resolved_bind + return app \ No newline at end of file diff --git a/webui/deployment_boundary.py b/webui/deployment_boundary.py new file mode 100644 index 0000000..e2df05c --- /dev/null +++ b/webui/deployment_boundary.py @@ -0,0 +1,155 @@ +"""Internal-only deployment boundary for the operator web UI (#435).""" + +from __future__ import annotations + +import ipaddress +import os +import re +from dataclasses import asdict, dataclass +from typing import Literal + +Disposition = Literal["safe", "warn", "refuse"] + +_LOOPBACK_HOSTS = frozenset({"127.0.0.1", "localhost", "::1"}) +_ALL_INTERFACE_HOSTS = frozenset({"0.0.0.0", "::", "*"}) + +_ALLOW_PUBLIC_BIND_ENV = "WEBUI_ALLOW_PUBLIC_BIND" +_ALLOW_REMOTE_BIND_ENV = "WEBUI_ALLOW_REMOTE_BIND" + +_FORBIDDEN_CLIENT_PATTERNS = ( + re.compile(r"GITEA_(?:TOKEN|PASS|PASSWORD)", re.I), + re.compile(r"Bearer\s+[A-Za-z0-9._\-]{20,}"), + re.compile(r"password\s*[:=]\s*['\"][^'\"]+['\"]", re.I), +) + + +@dataclass(frozen=True) +class BindAssessment: + host: str + disposition: Disposition + message: str + override_env: str | None = None + + +def _truthy_env(name: str) -> bool: + return os.environ.get(name, "").strip().lower() in {"1", "true", "yes"} + + +def assess_bind_host(host: str) -> BindAssessment: + """Classify a bind host as safe, warn, or refuse (fail closed on 0.0.0.0/::).""" + normalized = (host or "").strip().lower() + if not normalized: + return BindAssessment( + host=host, + disposition="refuse", + message="WEBUI_HOST must not be empty.", + ) + + if normalized in _LOOPBACK_HOSTS: + return BindAssessment( + host=host, + disposition="safe", + message="Loopback bind — suitable for local operator console.", + ) + + if normalized in _ALL_INTERFACE_HOSTS: + if _truthy_env(_ALLOW_PUBLIC_BIND_ENV): + return BindAssessment( + host=host, + disposition="warn", + message=( + "Binding all interfaces. Protect with Cloudflare Access, " + "WARP, VPN, or equivalent before exposing beyond localhost." + ), + override_env=_ALLOW_PUBLIC_BIND_ENV, + ) + return BindAssessment( + host=host, + disposition="refuse", + message=( + f"Refusing all-interface bind ({host!r}). Set " + f"{_ALLOW_PUBLIC_BIND_ENV}=1 only when fronted by trusted " + "network access controls." + ), + override_env=_ALLOW_PUBLIC_BIND_ENV, + ) + + try: + addr = ipaddress.ip_address(normalized) + if addr.is_loopback: + return BindAssessment( + host=host, + disposition="safe", + message="Loopback bind — suitable for local operator console.", + ) + except ValueError: + pass + + if _truthy_env(_ALLOW_REMOTE_BIND_ENV): + return BindAssessment( + host=host, + disposition="warn", + message=( + "Non-loopback bind acknowledged. Restrict to a trusted network " + "and add Cloudflare Access, WARP, or VPN if reachable beyond it." + ), + override_env=_ALLOW_REMOTE_BIND_ENV, + ) + + return BindAssessment( + host=host, + disposition="warn", + message=( + f"Non-loopback bind ({host!r}). MVP expects 127.0.0.1; set " + f"{_ALLOW_REMOTE_BIND_ENV}=1 to acknowledge a trusted-network bind." + ), + override_env=_ALLOW_REMOTE_BIND_ENV, + ) + + +def runtime_assumptions() -> dict[str, str]: + """Documented runtime paths and hosts — never includes secrets.""" + repo_root = (os.environ.get("WEBUI_REPO_ROOT") or "").strip() + registry = (os.environ.get("WEBUI_PROJECT_REGISTRY") or "").strip() + profile_config = (os.environ.get("GITEA_MCP_CONFIG") or "").strip() + profile_name = (os.environ.get("GITEA_MCP_PROFILE") or "").strip() + return { + "deployment_mode": "internal-operator-console", + "webui_host_env": "WEBUI_HOST", + "webui_port_env": "WEBUI_PORT", + "webui_repo_root_env": "WEBUI_REPO_ROOT", + "webui_repo_root": repo_root or "(defaults to repository root)", + "webui_project_registry_env": "WEBUI_PROJECT_REGISTRY", + "webui_project_registry": registry or "(packaged webui/data/projects.registry.json)", + "gitea_mcp_config_env": "GITEA_MCP_CONFIG", + "gitea_mcp_config": profile_config or "(optional; server-side only)", + "gitea_mcp_profile_env": "GITEA_MCP_PROFILE", + "gitea_mcp_profile": profile_name or "(optional; server-side only)", + "gitea_credentials": "Resolved server-side via gitea_auth; never embedded in HTML/JS", + "public_bind_override_env": _ALLOW_PUBLIC_BIND_ENV, + "remote_bind_override_env": _ALLOW_REMOTE_BIND_ENV, + } + + +def scan_text_for_client_secrets(text: str) -> list[str]: + """Return human-readable findings if *text* looks like it embeds secrets.""" + findings: list[str] = [] + for pattern in _FORBIDDEN_CLIENT_PATTERNS: + if pattern.search(text): + findings.append(f"matched forbidden client pattern: {pattern.pattern}") + return findings + + +def deployment_snapshot(*, bind_host: str | None = None) -> dict[str, object]: + host = bind_host if bind_host is not None else os.environ.get("WEBUI_HOST", "127.0.0.1") + bind = assess_bind_host(host) + return { + "mode": "internal-operator-console", + "mvp_auth": "none", + "bind": asdict(bind), + "runtime_assumptions": runtime_assumptions(), + "client_secret_policy": ( + "No Gitea tokens, passwords, or profile secrets in HTML, JS, " + "or browser storage." + ), + } \ No newline at end of file From 1334b0b3207c19fe8c46b7771fa894357f3a2f55 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Thu, 9 Jul 2026 08:10:36 -0400 Subject: [PATCH 08/12] fix: restore ISSUE_LOCK_FILE import and fail-closed public bind Conflict resolution left webui/lease_loader.py importing ISSUE_LOCK_FILE from merged_cleanup_reconcile (not exported); import from issue_lock_provenance instead. Validate WEBUI_HOST before loading the full app stack so 0.0.0.0 refuse exits immediately. Addresses reviewer findings on PR #456 / issue #435. --- webui/__main__.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/webui/__main__.py b/webui/__main__.py index 5c427da..9265ea8 100644 --- a/webui/__main__.py +++ b/webui/__main__.py @@ -7,7 +7,6 @@ import os import uvicorn -from webui.app import create_app from webui.deployment_boundary import assess_bind_host logger = logging.getLogger("webui") @@ -25,9 +24,13 @@ def _validate_bind_host(host: str) -> None: def main() -> None: host = os.environ.get("WEBUI_HOST", "127.0.0.1") port = int(os.environ.get("WEBUI_PORT", "8765")) + # Validate bind host before importing the full app stack so refuse paths + # fail closed quickly (tests and operators must not hang on create_app). _validate_bind_host(host) + from webui.app import create_app + uvicorn.run(create_app(), host=host, port=port, log_level="info") if __name__ == "__main__": - main() \ No newline at end of file + main() From 23535e30bc4aa44d321a82c2bdcca209fafb6da0 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 15:46:05 -0400 Subject: [PATCH 09/12] 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 --- docs/developer-testing-guidelines.md | 15 +++ docs/webui-local-dev.md | 21 ++++ scripts/ci-webui-check | 53 +++++++++ scripts/test-webui | 13 +++ tests/test_webui_ci_gate.py | 54 ++++++++++ tests/test_webui_mvp_suite.py | 156 +++++++++++++++++++++++++++ webui/ci_paths.py | 20 ++++ 7 files changed, 332 insertions(+) create mode 100755 scripts/ci-webui-check create mode 100755 scripts/test-webui create mode 100644 tests/test_webui_ci_gate.py create mode 100644 tests/test_webui_mvp_suite.py create mode 100644 webui/ci_paths.py diff --git a/docs/developer-testing-guidelines.md b/docs/developer-testing-guidelines.md index d85c670..61abd75 100644 --- a/docs/developer-testing-guidelines.md +++ b/docs/developer-testing-guidelines.md @@ -54,6 +54,21 @@ 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 +``` + +Wire `scripts/ci-webui-check` into Jenkins (or equivalent) for PRs that touch +`webui/`, `tests/test_webui_*`, or `docs/webui*`. + ### Run targeted tests ```bash diff --git a/docs/webui-local-dev.md b/docs/webui-local-dev.md index 73c4f47..1ee929b 100644 --- a/docs/webui-local-dev.md +++ b/docs/webui-local-dev.md @@ -155,4 +155,25 @@ pytest tests/test_webui_skeleton.py tests/test_webui_project_registry.py tests/t 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_deployment_boundary.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 +``` + +Or invoke unittest directly: + +```bash +python3 -m unittest discover -s tests -p 'test_webui_*.py' -q ``` \ No newline at end of file diff --git a/scripts/ci-webui-check b/scripts/ci-webui-check new file mode 100755 index 0000000..a679993 --- /dev/null +++ b/scripts/ci-webui-check @@ -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 \ No newline at end of file diff --git a/scripts/test-webui b/scripts/test-webui new file mode 100755 index 0000000..d7d4e31 --- /dev/null +++ b/scripts/test-webui @@ -0,0 +1,13 @@ +#!/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}" +exec "$python_bin" -m unittest discover -s tests -p "$pattern" "$@" \ No newline at end of file diff --git a/tests/test_webui_ci_gate.py b/tests/test_webui_ci_gate.py new file mode 100644 index 0000000..2f4a577 --- /dev/null +++ b/tests/test_webui_ci_gate.py @@ -0,0 +1,54 @@ +"""Tests for web UI CI path-filter gate (#436).""" +import os +import subprocess +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from webui.ci_paths import should_run_webui_ci, touches_webui_surface + +_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) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file diff --git a/tests/test_webui_mvp_suite.py b/tests/test_webui_mvp_suite.py new file mode 100644 index 0000000..2d5bd54 --- /dev/null +++ b/tests/test_webui_mvp_suite.py @@ -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() \ No newline at end of file diff --git a/webui/ci_paths.py b/webui/ci_paths.py new file mode 100644 index 0000000..779acf4 --- /dev/null +++ b/webui/ci_paths.py @@ -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) \ No newline at end of file From 64b83cd1fbca546c7e665734a52be35c19825472 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Thu, 9 Jul 2026 11:34:33 -0400 Subject: [PATCH 10/12] test: keep webui CI hermetic offline --- docs/developer-testing-guidelines.md | 4 +++ docs/webui-local-dev.md | 31 ++++++++++++++++++++++- scripts/test-webui | 3 ++- tests/test_webui_ci_gate.py | 38 +++++++++++++++++++++++++++- tests/test_webui_queue_dashboard.py | 7 ++--- webui/lease_loader.py | 29 +++++++++++++++++---- webui/queue_loader.py | 23 ++++++++++++++--- webui/runtime_health.py | 21 ++++++++++++--- 8 files changed, 138 insertions(+), 18 deletions(-) diff --git a/docs/developer-testing-guidelines.md b/docs/developer-testing-guidelines.md index 61abd75..3e180fc 100644 --- a/docs/developer-testing-guidelines.md +++ b/docs/developer-testing-guidelines.md @@ -66,6 +66,10 @@ modules when present on the branch. 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*`. diff --git a/docs/webui-local-dev.md b/docs/webui-local-dev.md index 1ee929b..f1e9145 100644 --- a/docs/webui-local-dev.md +++ b/docs/webui-local-dev.md @@ -172,8 +172,37 @@ diff touches `webui/`, `tests/test_webui_*`, or web UI docs/scripts): 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 -``` \ No newline at end of file +``` + +## 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 +(``, #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_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 +``` diff --git a/scripts/test-webui b/scripts/test-webui index d7d4e31..a5c49b7 100755 --- a/scripts/test-webui +++ b/scripts/test-webui @@ -10,4 +10,5 @@ if [[ -x "$repo_root/venv/bin/python" ]]; then fi pattern="${WEBUI_TEST_PATTERN:-test_webui_*.py}" -exec "$python_bin" -m unittest discover -s tests -p "$pattern" "$@" \ No newline at end of file +export WEBUI_TEST_OFFLINE="${WEBUI_TEST_OFFLINE:-1}" +exec "$python_bin" -m unittest discover -s tests -p "$pattern" "$@" diff --git a/tests/test_webui_ci_gate.py b/tests/test_webui_ci_gate.py index 2f4a577..3d2c602 100644 --- a/tests/test_webui_ci_gate.py +++ b/tests/test_webui_ci_gate.py @@ -4,10 +4,14 @@ 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 @@ -50,5 +54,37 @@ class TestCiRunnerScript(unittest.TestCase): 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() \ No newline at end of file + unittest.main() diff --git a/tests/test_webui_queue_dashboard.py b/tests/test_webui_queue_dashboard.py index cbdcc16..90ff8e7 100644 --- a/tests/test_webui_queue_dashboard.py +++ b/tests/test_webui_queue_dashboard.py @@ -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() \ No newline at end of file + unittest.main() diff --git a/webui/lease_loader.py b/webui/lease_loader.py index 318a46f..cd05d61 100644 --- a/webui/lease_loader.py +++ b/webui/lease_loader.py @@ -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, - } \ No newline at end of file + } diff --git a/webui/queue_loader.py b/webui/queue_loader.py index a70e58d..a6d35c7 100644 --- a/webui/queue_loader.py +++ b/webui/queue_loader.py @@ -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), }, - } \ No newline at end of file + } diff --git a/webui/runtime_health.py b/webui/runtime_health.py index 8f3b5d6..ee44187 100644 --- a/webui/runtime_health.py +++ b/webui/runtime_health.py @@ -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, - } \ No newline at end of file + } From 7f2a5a3d3c567053dade714aeba28c18fa82bb34 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Thu, 9 Jul 2026 10:12:17 -0400 Subject: [PATCH 11/12] fix: resolve reviewer lease identity via host not remote alias MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit _authenticated_username expects a host for auth/API calls. Lease acquire/adopt/gate/release paths passed the remote alias (prgs), so identity resolved empty and same-identity release fail-closed across sessions — blocking author conflict-fix pushes after REQUEST_CHANGES. Pass resolved host h from _resolve into _authenticated_username. --- gitea_mcp_server.py | 14 +++++--- tests/test_mcp_server.py | 78 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 88 insertions(+), 4 deletions(-) diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 49460df..e41b202 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -6153,7 +6153,9 @@ def _reviewer_pr_lease_gate( """Return block reasons when the session lacks an owned PR reviewer lease.""" session = reviewer_pr_lease.get_session_lease() session_id = (session or {}).get("session_id") - identity = _authenticated_username(remote) or "" + # Resolve host first: _authenticated_username expects a host, not remote name. + h, _, _ = _resolve(remote, host, org, repo) + identity = _authenticated_username(h) or "" try: comments = _fetch_pr_comments( pr_number, remote=remote, host=host, org=org, repo=repo) @@ -6207,7 +6209,8 @@ def gitea_acquire_reviewer_pr_lease( h, o, r = _resolve(remote, host, org, repo) auth = _auth(h) profile = get_profile() - identity = _authenticated_username(remote) or profile.get("username") or "" + # Host (not remote name) — remote aliases like "prgs" do not resolve identity. + identity = _authenticated_username(h) or profile.get("username") or "" sid = (session_id or "").strip() or reviewer_pr_lease.new_session_id() repo_label = f"{o}/{r}" @@ -6350,7 +6353,8 @@ def gitea_adopt_merger_pr_lease( auth = _auth(h) profile = get_profile() profile_name = profile.get("profile_name") or "unknown" - identity = _authenticated_username(remote) or profile.get("username") or "" + # Host (not remote name) — remote aliases like "prgs" do not resolve identity. + identity = _authenticated_username(h) or profile.get("username") or "" sid = (session_id or "").strip() or reviewer_pr_lease.new_session_id() repo_label = f"{o}/{r}" @@ -6738,7 +6742,9 @@ def gitea_release_reviewer_pr_lease( "reasons": ["no active reviewer lease found on PR"], } - identity = _authenticated_username(remote) or "" + # Host (not remote name). Passing "prgs"/"dadeschools" yields empty identity + # and incorrectly blocks same-identity release across sessions. + identity = _authenticated_username(h) or "" session = reviewer_pr_lease.get_session_lease() or {} session_id = session.get("session_id") diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index c3b4643..c054d7b 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -4288,3 +4288,81 @@ class TestIssue546Deadlock(unittest.TestCase): self.assertIsNone(reviewer_pr_lease.get_session_lease()) # verify comment phase is released self.assertIn("phase: released", posted_comments[0]["body"]) + + def test_release_reviewer_pr_lease_same_identity_cross_session(self): + """Same reviewer identity may release a lease claimed by another session. + + Regression: release used to call ``_authenticated_username(remote)`` + with the remote alias (e.g. ``prgs``) instead of the resolved host, + so identity resolved empty and same-identity release fail-closed even + when the authenticated user owned the lease. + """ + import mcp_server + import reviewer_pr_lease + + mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs") + mcp_server.gitea_load_review_workflow() + reviewer_pr_lease.clear_session_lease() + + foreign_session = "46820-dd78cdc4aacc" + comments = [ + _reviewer_lease_comment( + 457, + session_id=foreign_session, + head_sha="ef287eaf5841d9e542a4e888ce0ae7d679dbd685", + reviewer="sysadmin", + ) + ] + posted = [] + resolved_hosts = [] + + def mock_api(method, url, auth, data=None): + if "/api/v1/user" in url: + return {"login": "sysadmin"} + if "/comments" in url: + if method == "POST": + new_c = { + "id": 8071, + "body": data["body"], + "user": {"login": "sysadmin"}, + } + comments.append(new_c) + posted.append(new_c) + return {"id": 8071} + return comments + if "/pulls/" in url: + return { + "user": {"login": "jcwalker3"}, + "state": "open", + "head": {"sha": "ef287eaf5841d9e542a4e888ce0ae7d679dbd685"}, + "mergeable": False, + } + return {} + + def tracking_username(host): + resolved_hosts.append(host) + # Empty identity when passed a remote alias (the pre-fix bug path). + if host in {"prgs", "dadeschools"}: + return None + return "sysadmin" + + with patch("mcp_server.api_request", side_effect=mock_api), patch( + "mcp_server._authenticated_username", side_effect=tracking_username + ), patch( + "mcp_server._resolve", + return_value=("gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools"), + ), patch("mcp_server._auth", return_value={"Authorization": "token x"}), patch( + "mcp_server._verify_role_mutation_workspace", return_value=None + ): + res = mcp_server.gitea_release_reviewer_pr_lease( + pr_number=457, remote="prgs" + ) + + self.assertTrue(res.get("success"), res) + self.assertTrue(res.get("released"), res) + self.assertEqual(res.get("comment_id"), 8071) + self.assertTrue(posted) + self.assertIn("phase: released", posted[0]["body"]) + # Must resolve identity with host, never the remote alias alone. + self.assertIn("gitea.prgs.cc", resolved_hosts) + self.assertNotIn("prgs", resolved_hosts) From fe9faec741960136b9a49140ed2f1a5486cf4035 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Thu, 9 Jul 2026 10:49:31 -0400 Subject: [PATCH 12/12] test: cover reviewer lease identity host resolution --- tests/test_mcp_server.py | 155 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 155 insertions(+) diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index c054d7b..d52263b 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -4289,6 +4289,161 @@ class TestIssue546Deadlock(unittest.TestCase): # verify comment phase is released self.assertIn("phase: released", posted_comments[0]["body"]) + def test_reviewer_pr_lease_gate_resolves_identity_with_host(self): + """Lease mutation gate must resolve identity using host, not remote alias.""" + import mcp_server + + resolved_hosts = [] + + def tracking_username(host): + resolved_hosts.append(host) + return None if host in {"prgs", "dadeschools"} else "reviewer-bot" + + with _install_owned_reviewer_lease( + 550, + session_id=_DEFAULT_LEASE_SESSION, + head_sha="abc123", + ), patch( + "mcp_server._resolve", + return_value=("gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools"), + ), patch( + "mcp_server._authenticated_username", + side_effect=tracking_username, + ): + reasons = mcp_server._reviewer_pr_lease_gate( + pr_number=550, + remote="prgs", + host=None, + org=None, + repo=None, + mutation="approve", + live_head_sha="abc123", + pinned_head_sha="abc123", + ) + + self.assertEqual(reasons, []) + self.assertIn("gitea.prgs.cc", resolved_hosts) + self.assertNotIn("prgs", resolved_hosts) + + def test_acquire_reviewer_pr_lease_resolves_identity_with_host(self): + """Acquire path must not pass a remote alias into identity lookup.""" + import mcp_server + + resolved_hosts = [] + posted = [] + + def tracking_username(host): + resolved_hosts.append(host) + return None if host in {"prgs", "dadeschools"} else "reviewer-bot" + + def mock_api(method, url, auth, data=None): + if "/pulls/" in url: + return { + "user": {"login": "author-user"}, + "state": "open", + "head": {"sha": "abc123"}, + "mergeable": True, + } + if "/comments" in url: + if method == "POST": + posted.append(data["body"]) + return {"id": 1003} + return [] + return {} + + with patch("mcp_server.api_request", side_effect=mock_api), patch( + "mcp_server._authenticated_username", + side_effect=tracking_username, + ), patch( + "mcp_server._resolve", + return_value=("gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools"), + ): + res = mcp_server.gitea_acquire_reviewer_pr_lease( + pr_number=550, + worktree="/workspace", + candidate_head="abc123", + remote="prgs", + ) + + self.assertTrue(res.get("success"), res) + self.assertTrue(res.get("acquired"), res) + self.assertEqual(res.get("comment_id"), 1003) + self.assertTrue(posted) + self.assertIn("gitea.prgs.cc", resolved_hosts) + self.assertNotIn("prgs", resolved_hosts) + + def test_adopt_merger_pr_lease_resolves_identity_with_host(self): + """Adopt path must not pass a remote alias into identity lookup.""" + import mcp_server + import reviewer_pr_lease + + reviewer_pr_lease.clear_session_lease() + head = "a" * 40 + resolved_hosts = [] + posted = [] + reviewer_comment = _reviewer_lease_comment( + 550, + session_id="reviewer-session", + head_sha=head, + reviewer="reviewer-bot", + ) + + def tracking_username(host): + resolved_hosts.append(host) + return None if host in {"prgs", "dadeschools"} else "merger-bot" + + def mock_api(method, url, auth, data=None): + if "/pulls/" in url and "/comments" not in url: + return { + "number": 550, + "user": {"login": "author-user"}, + "state": "open", + "head": {"sha": head}, + "mergeable": True, + "merged": False, + } + if "/comments" in url: + if method == "POST": + posted.append(data["body"]) + return {"id": 1004} + return [reviewer_comment] + return {} + + merger_profile = { + "profile_name": "prgs-merger", + "allowed_operations": ["gitea.read", "gitea.pr.comment", "gitea.pr.merge"], + "forbidden_operations": [], + } + with patch("mcp_server.get_profile", return_value=merger_profile), patch( + "mcp_server.api_request", + side_effect=mock_api, + ), patch( + "mcp_server._authenticated_username", + side_effect=tracking_username, + ), patch( + "mcp_server._resolve", + return_value=("gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools"), + ), patch( + "mcp_server.gitea_get_pr_review_feedback", + return_value={ + "approval_at_current_head": True, + "latest_approved_head_sha": head, + }, + ): + res = mcp_server.gitea_adopt_merger_pr_lease( + pr_number=550, + worktree="/workspace", + expected_head_sha=head, + remote="prgs", + ) + + self.assertTrue(res.get("success"), res) + self.assertTrue(res.get("adopted"), res) + self.assertEqual(res.get("adoption_comment_id"), 1004) + self.assertTrue(posted) + self.assertIn("gitea.prgs.cc", resolved_hosts) + self.assertNotIn("prgs", resolved_hosts) + def test_release_reviewer_pr_lease_same_identity_cross_session(self): """Same reviewer identity may release a lease claimed by another session.