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 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.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"})


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>"
        "</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 audit(_request: Request) -> HTMLResponse:
    return _stub_page(
        "Audit",
        "Report audit will accept pasted final reports and run validator previews.",
    )


async def worktrees(_request: Request) -> HTMLResponse:
    return _stub_page(
        "Worktrees",
        "Worktree hygiene will summarize branches/ session folders and cleanup risk.",
    )


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 method_not_allowed(request: Request, _exc: Exception) -> Response:
    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"]),
            Route("/worktrees", worktrees, methods=["GET"]),
            Route("/leases", leases, methods=["GET"]),
            Route("/api/leases", api_leases, methods=["GET"]),
        ],
        exception_handlers={405: method_not_allowed},
    )
    app.state.webui_bind_host = resolved_bind
    return app