Gitea-Tools/webui/app.py

"""Starlette application for the internal read-only web UI (#426)."""

from __future__ import annotations

from datetime import datetime, timezone

from starlette.applications import Starlette
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
from webui.prompt_library import find_prompt, library_to_dict
from webui.prompt_views import render_prompt_detail, render_prompts_page
from final_report_validator import FINAL_REPORT_TASK_KINDS

from webui.gated_actions import attempt_action, load_action_registry, preview_action
from webui.gated_action_views import render_actions_page
from webui.audit_validator import audit_report, audit_to_dict
from webui.audit_views import render_audit_page
from webui.lease_loader import load_lease_snapshot, snapshot_to_dict as lease_snapshot_to_dict
from webui.lease_views import render_leases_page
from webui.queue_loader import load_queue_snapshot, snapshot_to_dict as queue_snapshot_to_dict
from webui.queue_views import render_queue_page
from webui.worktree_scanner import load_hygiene_snapshot, snapshot_to_dict as worktree_snapshot_to_dict
from webui.worktree_views import render_worktrees_page
from webui.runtime_health import load_runtime_snapshot, snapshot_to_dict as runtime_snapshot_to_dict
from webui.runtime_views import render_runtime_page

_READ_ONLY_METHODS = frozenset({"GET", "HEAD", "OPTIONS"})
_AUDIT_MUTATION_PATHS = frozenset({"/audit", "/api/audit"})


def _stub_page(title: str, description: str) -> HTMLResponse:
    body = (
        f"<h2>{title}</h2>"
        f'<div class="stub"><p>{description}</p>'
        "<p>Implementation tracked in a child issue of #425.</p></div>"
    )
    return HTMLResponse(render_page(title=title, body_html=body))


async def home(_request: Request) -> HTMLResponse:
    body = (
        "<h2>Operator console</h2>"
        "<p>Local entry point for MCP Control Plane operational views.</p>"
        "<ul>"
        "<li><strong>Queue</strong> — live PR and issue dashboard (#429)</li>"
        "<li><strong>Projects</strong> — registry and onboarding (#427)</li>"
        "<li><strong>Prompts</strong> — canonical workflow prompt library (#428)</li>"
        "<li><strong>Runtime</strong> — MCP health and stale-runtime detection (#430)</li>"
        "<li><strong>Audit</strong> — final-report paste and validator preview (#431)</li>"
        "<li><strong>Worktrees</strong> — branch hygiene dashboard (#432)</li>"
        "<li><strong>Leases</strong> — collision and lease visibility (#433)</li>"
        "<li><strong>Actions</strong> — gated write-action framework (#434)</li>"
        "</ul>"
    )
    return HTMLResponse(render_page(title="Home", body_html=body))


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),
    })


async def queue(_request: Request) -> HTMLResponse:
    snapshot = load_queue_snapshot()
    return HTMLResponse(render_page(title="Queue", body_html=render_queue_page(snapshot)))


async def api_queue(_request: Request) -> JSONResponse:
    return JSONResponse(queue_snapshot_to_dict(load_queue_snapshot()))


async def projects(_request: Request) -> HTMLResponse:
    registry = load_registry()
    return HTMLResponse(render_projects_list(registry))


async def project_detail(request: Request) -> HTMLResponse:
    project_id = request.path_params["project_id"]
    registry = load_registry()
    project = find_project(registry, project_id)
    if project is None:
        return HTMLResponse(
            render_page(
                title="Project not found",
                body_html=(
                    "<h2>Project not found</h2>"
                    f"<p>No registry entry for <code>{project_id}</code>.</p>"
                    '<p><a href="/projects">← All projects</a></p>'
                ),
            ),
            status_code=404,
        )
    return HTMLResponse(render_project_detail(project))


async def api_projects(_request: Request) -> JSONResponse:
    registry = load_registry()
    return JSONResponse(registry_to_dict(registry))


async def prompts(_request: Request) -> HTMLResponse:
    return HTMLResponse(render_prompts_page())


async def prompt_detail(request: Request) -> HTMLResponse:
    prompt_id = request.path_params["prompt_id"]
    prompt = find_prompt(prompt_id)
    if prompt is None:
        return HTMLResponse(
            render_page(
                title="Prompt not found",
                body_html=(
                    "<h2>Prompt not found</h2>"
                    f"<p>No library entry for <code>{prompt_id}</code>.</p>"
                    '<p><a href="/prompts">← All prompts</a></p>'
                ),
            ),
            status_code=404,
        )
    return HTMLResponse(render_prompt_detail(prompt))


async def api_prompts(_request: Request) -> JSONResponse:
    return JSONResponse(library_to_dict())


async def runtime(_request: Request) -> HTMLResponse:
    snapshot = load_runtime_snapshot()
    return HTMLResponse(render_page(title="Runtime", body_html=render_runtime_page(snapshot)))


async def api_runtime(_request: Request) -> JSONResponse:
    return JSONResponse(runtime_snapshot_to_dict(load_runtime_snapshot()))


async def _parse_audit_form(request: Request) -> tuple[str, str | None]:
    if request.method == "GET":
        return "", None
    content_type = (request.headers.get("content-type") or "").lower()
    if "application/json" in content_type:
        payload = await request.json()
        if not isinstance(payload, dict):
            return "", None
        report_text = str(payload.get("report_text") or "")
        task_kind = payload.get("task_kind")
        return report_text, str(task_kind) if task_kind else None
    form = await request.form()
    report_text = str(form.get("report_text") or "")
    task_kind = form.get("task_kind")
    return report_text, str(task_kind) if task_kind else None


async def audit(request: Request) -> HTMLResponse:
    report_text, task_kind = await _parse_audit_form(request)
    result = None
    if request.method == "POST" and report_text.strip():
        result = audit_report(report_text, task_kind=task_kind)
    return HTMLResponse(
        render_audit_page(
            report_text=report_text,
            task_kind=task_kind,
            result=result,
        )
    )


async def api_audit(request: Request) -> JSONResponse:
    if request.method == "GET":
        return JSONResponse({
            "usage": "POST JSON with report_text and optional task_kind",
            "task_kinds": sorted(FINAL_REPORT_TASK_KINDS),
        })
    report_text, task_kind = await _parse_audit_form(request)
    if not report_text.strip():
        return JSONResponse({"error": "report_text required"}, status_code=400)
    return JSONResponse(audit_to_dict(audit_report(report_text, task_kind=task_kind)))


async def worktrees(_request: Request) -> HTMLResponse:
    snapshot = load_hygiene_snapshot()
    return HTMLResponse(render_worktrees_page(snapshot))


async def api_worktrees(_request: Request) -> JSONResponse:
    return JSONResponse(worktree_snapshot_to_dict(load_hygiene_snapshot()))


async def leases(_request: Request) -> HTMLResponse:
    snapshot = load_lease_snapshot()
    return HTMLResponse(render_leases_page(snapshot))


async def api_leases(_request: Request) -> JSONResponse:
    return JSONResponse(lease_snapshot_to_dict(load_lease_snapshot()))


async def actions(_request: Request) -> HTMLResponse:
    registry = load_action_registry()
    return HTMLResponse(render_actions_page(registry))


async def api_actions(_request: Request) -> JSONResponse:
    return JSONResponse(load_action_registry().to_dict())


async def api_action_preview(request: Request) -> JSONResponse:
    action_id = request.path_params["action_id"]
    params = dict(request.query_params)
    for key in ("issue_number", "pr_number"):
        if key in params:
            params[key] = int(params[key])
    result = preview_action(action_id, **params)
    if "error" in result:
        return JSONResponse(result, status_code=404)
    return JSONResponse(result)


async def api_action_attempt(request: Request) -> JSONResponse:
    """POST is rejected by read-only guard; kept for explicit fail-closed tests."""
    action_id = request.path_params["action_id"]
    try:
        body = await request.json()
    except Exception:
        body = {}
    if not isinstance(body, dict):
        body = {}
    result = attempt_action(action_id, **body)
    status = 403 if not result.get("success") else 200
    return JSONResponse(result, status_code=status)


async def method_not_allowed(request: Request, _exc: Exception) -> Response:
    path = request.url.path
    if path in _AUDIT_MUTATION_PATHS and request.method == "POST":
        return JSONResponse({"error": "not_found"}, status_code=404)
    if request.method not in _READ_ONLY_METHODS:
        return JSONResponse(
            {"error": "read-only-mvp", "detail": f"{request.method} not permitted"},
            status_code=405,
        )
    return JSONResponse({"error": "not_found"}, status_code=404)


def create_app(*, bind_host: str | None = None) -> Starlette:
    """Build the read-only MVP Starlette app."""
    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"]),
            Route("/health", health, methods=["GET"]),
            Route("/queue", queue, methods=["GET"]),
            Route("/api/queue", api_queue, methods=["GET"]),
            Route("/projects", projects, methods=["GET"]),
            Route("/projects/{project_id}", project_detail, methods=["GET"]),
            Route("/api/projects", api_projects, methods=["GET"]),
            Route("/prompts", prompts, methods=["GET"]),
            Route("/prompts/{prompt_id}", prompt_detail, methods=["GET"]),
            Route("/api/prompts", api_prompts, methods=["GET"]),
            Route("/runtime", runtime, methods=["GET"]),
            Route("/api/runtime", api_runtime, methods=["GET"]),
            Route("/audit", audit, methods=["GET", "POST"]),
            Route("/api/audit", api_audit, methods=["GET", "POST"]),
            Route("/worktrees", worktrees, methods=["GET"]),
            Route("/api/worktrees", api_worktrees, methods=["GET"]),
            Route("/leases", leases, methods=["GET"]),
            Route("/actions", actions, methods=["GET"]),
            Route("/api/actions", api_actions, methods=["GET"]),
            Route(
                "/api/actions/{action_id}/preview",
                api_action_preview,
                methods=["GET"],
            ),
            Route(
                "/api/actions/{action_id}/attempt",
                api_action_attempt,
                methods=["POST"],
            ),
            Route("/api/leases", api_leases, methods=["GET"]),
        ],
        exception_handlers={405: method_not_allowed},
    )
    app.state.webui_bind_host = resolved_bind
    return app