230 lines
8.6 KiB
Python
230 lines
8.6 KiB
Python
"""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.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.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
|
|
from webui.queue_views import render_queue_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>"
|
|
"</ul>"
|
|
)
|
|
return HTMLResponse(render_page(title="Home", body_html=body))
|
|
|
|
|
|
async def health(_request: Request) -> JSONResponse:
|
|
return JSONResponse({
|
|
"status": "ok",
|
|
"service": "mcp-control-plane-webui",
|
|
"mode": "read-only-mvp",
|
|
"timestamp": datetime.now(timezone.utc).isoformat(),
|
|
})
|
|
|
|
|
|
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(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:
|
|
return _stub_page(
|
|
"Runtime",
|
|
"Runtime health will report MCP profile, preflight, and stale-server signals.",
|
|
)
|
|
|
|
|
|
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:
|
|
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:
|
|
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() -> Starlette:
|
|
"""Build the read-only MVP Starlette app."""
|
|
return 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("/audit", audit, methods=["GET", "POST"]),
|
|
Route("/api/audit", api_audit, methods=["GET", "POST"]),
|
|
Route("/worktrees", worktrees, methods=["GET"]),
|
|
Route("/leases", leases, methods=["GET"]),
|
|
Route("/api/leases", api_leases, methods=["GET"]),
|
|
],
|
|
exception_handlers={405: method_not_allowed},
|
|
) |