775 lines
29 KiB
Python
775 lines
29 KiB
Python
"""Starlette application for the internal read-only web UI (#426)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
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.nav import NAV_GROUPS, STUB_PAGES
|
|
from webui.project_registry import (
|
|
ProjectRegistry,
|
|
RegistryError,
|
|
find_project,
|
|
known_project_ids,
|
|
load_registry,
|
|
project_detail_to_dict,
|
|
registry_to_dict,
|
|
)
|
|
from webui.project_views import (
|
|
render_project_detail,
|
|
render_projects_list,
|
|
render_registry_error,
|
|
)
|
|
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 import console_audit
|
|
from webui.console_authz import authorize, rbac_matrix, resolve_principal
|
|
from webui.console_redaction import redaction_policy
|
|
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
|
|
from webui.policy_inventory import load_policy_inventory, snapshot_to_dict as policy_snapshot_to_dict
|
|
from webui.policy_views import render_policy_page
|
|
from webui.timeline import load_timeline, snapshot_to_dict as timeline_snapshot_to_dict
|
|
from webui.analytics_loader import (
|
|
load_analytics,
|
|
record_usage,
|
|
snapshot_to_dict as analytics_snapshot_to_dict,
|
|
)
|
|
from webui.analytics_views import render_analytics_page
|
|
from webui.system_health import (
|
|
API_PATH as SYSTEM_HEALTH_API_PATH,
|
|
load_system_health,
|
|
process_uptime,
|
|
snapshot_to_dict as system_health_to_dict,
|
|
)
|
|
from webui.system_health_views import render_system_health_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))
|
|
|
|
|
|
_LEGACY_PAGES = (
|
|
("/queue", "Queue", "live PR and issue dashboard (#429)"),
|
|
("/projects", "Projects", "registry and onboarding (#427)"),
|
|
("/prompts", "Prompts", "canonical workflow prompt library (#428)"),
|
|
("/runtime", "Runtime", "MCP health and stale-runtime detection (#430)"),
|
|
("/audit", "Audit", "final-report paste and validator preview (#431)"),
|
|
("/worktrees", "Worktrees", "branch hygiene dashboard (#432)"),
|
|
("/leases", "Leases", "collision and lease visibility (#433)"),
|
|
("/actions", "Actions", "gated write-action framework (#434)"),
|
|
)
|
|
|
|
|
|
def _render_home_nav_groups() -> str:
|
|
groups = []
|
|
for group in NAV_GROUPS:
|
|
items = "".join(
|
|
f'<li><a href="{item.href}">{item.label}</a>'
|
|
+ ("" if item.status == "live" else " <span class=\"muted\">(stub)</span>")
|
|
+ "</li>"
|
|
for item in group.items
|
|
)
|
|
groups.append(f"<h3>{group.label}</h3><ul>{items}</ul>")
|
|
return "".join(groups)
|
|
|
|
|
|
async def home(_request: Request) -> HTMLResponse:
|
|
legacy = "".join(
|
|
f"<li><strong>{label}</strong> — {desc} "
|
|
f'(<a href="{href}">{href}</a>)</li>'
|
|
for href, label, desc in _LEGACY_PAGES
|
|
)
|
|
body = (
|
|
"<h2>Operator console</h2>"
|
|
"<p>Read-only home for the MCP Control Plane Phase 1 operator console. "
|
|
"Gitea, MCP capability gates, and canonical workflows remain the source "
|
|
"of truth; this console never mutates them.</p>"
|
|
"<h2>Phase 1 surfaces</h2>"
|
|
+ _render_home_nav_groups()
|
|
+ "<h2>MVP legacy pages</h2>"
|
|
"<ul>" + legacy + "</ul>"
|
|
)
|
|
return HTMLResponse(render_page(title="Home", body_html=body))
|
|
|
|
|
|
async def phase_stub(request: Request) -> HTMLResponse:
|
|
"""Graceful read-only placeholder for a not-yet-implemented Phase 1 surface."""
|
|
title, description = STUB_PAGES[request.url.path]
|
|
body = (
|
|
f"<h2>{title}</h2>"
|
|
f'<div class="stub"><p>{description}</p>'
|
|
"<p>Phase 1 shell placeholder — no write actions. Tracked under "
|
|
"epic #631.</p></div>"
|
|
)
|
|
return HTMLResponse(render_page(title=title, body_html=body))
|
|
|
|
|
|
async def health(_request: Request) -> JSONResponse:
|
|
"""Liveness only — deliberately cheap, runs no dependency probe (#634).
|
|
|
|
Every MVP key is retained so existing pollers keep working; the additions
|
|
are a pointer to the structured API and the in-memory process uptime.
|
|
Readiness lives at that API because answering it costs real probes.
|
|
"""
|
|
bind_host = _request.app.state.webui_bind_host
|
|
started_at, uptime_seconds = process_uptime()
|
|
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),
|
|
"started_at": started_at,
|
|
"uptime_seconds": uptime_seconds,
|
|
"system_health_api": SYSTEM_HEALTH_API_PATH,
|
|
})
|
|
|
|
|
|
def _truthy_flag(value: str | None) -> bool:
|
|
return (value or "").strip().lower() in {"1", "true", "yes", "on"}
|
|
|
|
|
|
async def api_system_health(request: Request) -> JSONResponse:
|
|
"""Structured read-only system health (#634).
|
|
|
|
`?deep=1` opts into the expensive network probe. The response status code
|
|
reflects readiness so automated checks can branch on it without parsing the
|
|
body: 200 when ready, 503 when a required dependency failed or never ran.
|
|
"""
|
|
deep = _truthy_flag(request.query_params.get("deep"))
|
|
snapshot = load_system_health(deep=deep)
|
|
payload = system_health_to_dict(snapshot)
|
|
return JSONResponse(payload, status_code=200 if snapshot.ready else 503)
|
|
|
|
|
|
async def system_health(request: Request) -> HTMLResponse:
|
|
"""Read-only system-health dashboard (#639).
|
|
|
|
Shares the #634 snapshot loader with the JSON API so the page can never
|
|
disagree with it. `?deep=1` opts into the network probe exactly as the API
|
|
does; the default page load stays cheap. The response is always 200: this
|
|
is an operator view that must render the degraded state, not withhold it.
|
|
"""
|
|
deep = _truthy_flag(request.query_params.get("deep"))
|
|
snapshot = load_system_health(deep=deep)
|
|
return HTMLResponse(
|
|
render_page(
|
|
title="System health",
|
|
body_html=render_system_health_page(snapshot),
|
|
)
|
|
)
|
|
|
|
|
|
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()))
|
|
|
|
|
|
def _load_project_registry() -> tuple[ProjectRegistry | None, RegistryError | None]:
|
|
"""Load the registry, converting validation failure into a fail-closed pair."""
|
|
try:
|
|
return load_registry(), None
|
|
except RegistryError as exc:
|
|
return None, exc
|
|
|
|
|
|
async def projects(_request: Request) -> HTMLResponse:
|
|
registry, error = _load_project_registry()
|
|
if error is not None:
|
|
return HTMLResponse(render_registry_error(error), status_code=500)
|
|
return HTMLResponse(render_projects_list(registry))
|
|
|
|
|
|
async def project_detail(request: Request) -> HTMLResponse:
|
|
project_id = request.path_params["project_id"]
|
|
registry, error = _load_project_registry()
|
|
if error is not None:
|
|
return HTMLResponse(render_registry_error(error), status_code=500)
|
|
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:
|
|
"""Unversioned MVP alias, retained through Phase 1 (#632 section 6)."""
|
|
registry, error = _load_project_registry()
|
|
if error is not None:
|
|
return JSONResponse(error.to_dict(), status_code=500)
|
|
return JSONResponse(registry_to_dict(registry))
|
|
|
|
|
|
async def api_v1_projects(_request: Request) -> JSONResponse:
|
|
registry, error = _load_project_registry()
|
|
if error is not None:
|
|
return JSONResponse(error.to_dict(), status_code=500)
|
|
return JSONResponse(registry_to_dict(registry))
|
|
|
|
|
|
async def api_v1_project_detail(request: Request) -> JSONResponse:
|
|
project_id = request.path_params["project_id"]
|
|
registry, error = _load_project_registry()
|
|
if error is not None:
|
|
return JSONResponse(error.to_dict(), status_code=500)
|
|
project = find_project(registry, project_id)
|
|
if project is None:
|
|
return JSONResponse(
|
|
{
|
|
"error": "project_not_found",
|
|
"project_id": project_id,
|
|
"known_project_ids": known_project_ids(registry),
|
|
"remediation": (
|
|
"Request one of the known project ids, or add the project to the "
|
|
"registry file named in 'source'."
|
|
),
|
|
"source": {
|
|
"kind": "file",
|
|
"path": str(registry.source_path),
|
|
"inventory_complete": True,
|
|
},
|
|
},
|
|
status_code=404,
|
|
)
|
|
return JSONResponse(project_detail_to_dict(registry, project))
|
|
|
|
|
|
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 policy(_request: Request) -> HTMLResponse:
|
|
snapshot = load_policy_inventory()
|
|
return HTMLResponse(
|
|
render_page(title="Policy", body_html=render_policy_page(snapshot))
|
|
)
|
|
|
|
|
|
async def api_v1_policy(_request: Request) -> JSONResponse:
|
|
return JSONResponse(policy_snapshot_to_dict(load_policy_inventory()))
|
|
|
|
|
|
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())
|
|
|
|
|
|
def _request_id() -> str:
|
|
return f"req-{uuid.uuid4().hex}"
|
|
|
|
|
|
def _audit_target(action_id: str, params: dict[str, object]) -> dict[str, object]:
|
|
"""Describe the action target for the audit record (never secrets)."""
|
|
if "pr_number" in params:
|
|
return {"kind": "pr", "ref": f"#{params['pr_number']}"}
|
|
if "issue_number" in params:
|
|
return {"kind": "issue", "ref": f"#{params['issue_number']}"}
|
|
if "branch_name" in params:
|
|
return {"kind": "branch", "ref": str(params["branch_name"])}
|
|
return {"kind": "unspecified", "ref": action_id}
|
|
|
|
|
|
def _authorize_request(
|
|
request: Request,
|
|
action_id: str,
|
|
params: dict[str, object],
|
|
*,
|
|
for_execution: bool,
|
|
result: str,
|
|
) -> dict[str, object]:
|
|
"""Resolve principal, decide, and audit. Returns the decision payload.
|
|
|
|
Phase 1 records the decision rather than enforcing it as the terminal
|
|
outcome: ``webui.gated_actions`` already fails closed for every action, so
|
|
this layer cannot loosen anything. Phase 2 enforces on this same decision.
|
|
"""
|
|
principal = resolve_principal(headers=dict(request.headers))
|
|
decision = authorize(action_id, principal, for_execution=for_execution)
|
|
console_audit.record_event(
|
|
action_id=action_id,
|
|
result=result,
|
|
decision=decision,
|
|
principal=principal,
|
|
target=_audit_target(action_id, params),
|
|
request_id=_request_id(),
|
|
detail=decision.detail,
|
|
)
|
|
return decision.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)
|
|
result["authorization"] = _authorize_request(
|
|
request,
|
|
action_id,
|
|
params,
|
|
for_execution=False,
|
|
result=console_audit.RESULT_PREVIEWED,
|
|
)
|
|
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)
|
|
authorization = _authorize_request(
|
|
request,
|
|
action_id,
|
|
body,
|
|
for_execution=True,
|
|
result=(
|
|
console_audit.RESULT_DENIED
|
|
if not result.get("success")
|
|
else console_audit.RESULT_ALLOWED
|
|
),
|
|
)
|
|
result["authorization"] = authorization
|
|
status = 403 if not result.get("success") else 200
|
|
return JSONResponse(result, status_code=status)
|
|
|
|
|
|
async def api_console_security_model(_request: Request) -> JSONResponse:
|
|
"""Read-only publication of the #633 authorization/redaction/audit model."""
|
|
return JSONResponse({
|
|
"rbac": rbac_matrix(),
|
|
"redaction": redaction_policy(),
|
|
"audit": console_audit.audit_policy(),
|
|
})
|
|
|
|
|
|
def _query_int(request: Request, key: str) -> int | None:
|
|
"""Parse an optional integer query parameter; None when absent/invalid."""
|
|
raw = request.query_params.get(key)
|
|
if raw is None or not str(raw).strip():
|
|
return None
|
|
try:
|
|
return int(str(raw).strip())
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
def _derive_remote(host: str) -> str:
|
|
"""Map a Gitea host to its known short remote name (control-plane scope key)."""
|
|
text = (host or "").lower()
|
|
if "prgs" in text:
|
|
return "prgs"
|
|
if "dadeschools" in text:
|
|
return "dadeschools"
|
|
return text.split(".")[0] if text else ""
|
|
|
|
|
|
def _timeline_comment_source(host: str, org: str, repo: str):
|
|
"""Build a fail-soft CTH-comment fetcher for one repo, or None when offline.
|
|
|
|
Returns a callable ``(kind, number) -> list[comment]``. Credentials or
|
|
network failures raise inside the callable so ``load_timeline`` degrades the
|
|
handoff source rather than the whole timeline. Offline test mode yields no
|
|
live source so the handoff section reports ``not run``.
|
|
"""
|
|
import os
|
|
|
|
from gitea_auth import api_fetch_page, get_auth_header, repo_api_url
|
|
|
|
offline = (os.environ.get("WEBUI_TEST_OFFLINE") or "").strip().lower() in {"1", "true", "yes"}
|
|
if offline:
|
|
return None
|
|
auth = get_auth_header(host)
|
|
if not auth:
|
|
return None
|
|
|
|
def _fetch(kind: str, number: int) -> list:
|
|
segment = "pulls" if kind == "pr" else "issues"
|
|
url = f"{repo_api_url(host, org, repo)}/{segment}/{int(number)}/comments"
|
|
comments: list = []
|
|
page = 1
|
|
while page <= 20:
|
|
raw, meta = api_fetch_page(url, auth, page=page, limit=50)
|
|
comments.extend(raw)
|
|
if bool(meta["is_final_page"]):
|
|
break
|
|
page += 1
|
|
return comments
|
|
|
|
return _fetch
|
|
|
|
|
|
async def api_v1_timeline(request: Request) -> JSONResponse:
|
|
"""Read-only workflow-event timeline (#637). Filter by issue/PR/session."""
|
|
from webui.queue_loader import _host_from_url # host normalisation helper
|
|
|
|
registry, error = _load_project_registry()
|
|
if error is not None:
|
|
return JSONResponse(error.to_dict(), status_code=500)
|
|
project = registry.projects[0] if registry.projects else None
|
|
|
|
org = request.query_params.get("org") or (project.gitea_owner if project else "")
|
|
repo = request.query_params.get("repo") or (project.repo_name if project else "")
|
|
host = _host_from_url(project.remote_host) if project else ""
|
|
remote = request.query_params.get("remote") or _derive_remote(host)
|
|
|
|
if not (remote and org and repo):
|
|
return JSONResponse(
|
|
{
|
|
"error": "timeline_scope_unresolved",
|
|
"detail": "no project in registry and no remote/org/repo query params provided",
|
|
},
|
|
status_code=400,
|
|
)
|
|
|
|
comment_source = _timeline_comment_source(host, org, repo) if (host and org and repo) else None
|
|
|
|
snapshot = load_timeline(
|
|
remote=remote,
|
|
org=org,
|
|
repo=repo,
|
|
issue=_query_int(request, "issue"),
|
|
pr=_query_int(request, "pr"),
|
|
session=(request.query_params.get("session") or None),
|
|
limit=_query_int(request, "limit"),
|
|
offset=_query_int(request, "offset"),
|
|
comment_source=comment_source,
|
|
)
|
|
# A filter no surviving source can carry is refused, not answered empty:
|
|
# a 200 with zero events would tell the operator no such activity exists.
|
|
status_code = 200 if snapshot.ok else 422
|
|
return JSONResponse(timeline_snapshot_to_dict(snapshot), status_code=status_code)
|
|
|
|
|
|
async def analytics(request: Request) -> HTMLResponse:
|
|
"""Read-only model usage, token cost, latency, and performance analytics HTML view (#651)."""
|
|
snapshot = load_analytics(
|
|
remote=request.query_params.get("remote"),
|
|
org=request.query_params.get("org"),
|
|
repo=request.query_params.get("repo"),
|
|
role=request.query_params.get("role"),
|
|
model=request.query_params.get("model"),
|
|
stage=request.query_params.get("stage"),
|
|
issue_number=_query_int(request, "issue"),
|
|
pr_number=_query_int(request, "pr"),
|
|
limit=_query_int(request, "limit") or 200,
|
|
)
|
|
return HTMLResponse(render_analytics_page(snapshot))
|
|
|
|
|
|
async def api_v1_analytics(request: Request) -> JSONResponse:
|
|
"""Read-only model usage, token cost, latency, and performance analytics API (#651)."""
|
|
snapshot = load_analytics(
|
|
remote=request.query_params.get("remote"),
|
|
org=request.query_params.get("org"),
|
|
repo=request.query_params.get("repo"),
|
|
role=request.query_params.get("role"),
|
|
model=request.query_params.get("model"),
|
|
stage=request.query_params.get("stage"),
|
|
issue_number=_query_int(request, "issue"),
|
|
pr_number=_query_int(request, "pr"),
|
|
limit=_query_int(request, "limit") or 500,
|
|
)
|
|
status_code = 200 if snapshot.ok else 500
|
|
return JSONResponse(analytics_snapshot_to_dict(snapshot), status_code=status_code)
|
|
|
|
|
|
async def api_v1_analytics_ingest(request: Request) -> JSONResponse:
|
|
"""Optional session instrumentation ingestion endpoint (#651).
|
|
|
|
Fail-closed write: every request is authorized through console_authz
|
|
(``record_analytics_usage``) before any control-plane DB mutation. Phase 1
|
|
keeps ``execution_enabled=False`` and denies unauthenticated callers, so
|
|
this route cannot be used as an unauthenticated write or XSS injection
|
|
vector (PR #876 F2).
|
|
"""
|
|
try:
|
|
body = await request.json()
|
|
except Exception:
|
|
body = {}
|
|
if not isinstance(body, dict):
|
|
body = {}
|
|
|
|
principal = resolve_principal(headers=dict(request.headers))
|
|
decision = authorize(
|
|
"record_analytics_usage", principal, for_execution=True
|
|
)
|
|
allowed = bool(decision.allowed and decision.execution_enabled)
|
|
console_audit.record_event(
|
|
action_id="record_analytics_usage",
|
|
result=(
|
|
console_audit.RESULT_ALLOWED
|
|
if allowed
|
|
else console_audit.RESULT_DENIED
|
|
),
|
|
decision=decision,
|
|
principal=principal,
|
|
target=_audit_target("record_analytics_usage", body),
|
|
request_id=_request_id(),
|
|
detail=decision.detail,
|
|
)
|
|
authorization = decision.to_dict()
|
|
if not allowed:
|
|
return JSONResponse(
|
|
{
|
|
"ok": False,
|
|
"error": "unauthorized",
|
|
"detail": (
|
|
"POST /api/v1/analytics/usage requires an authenticated "
|
|
"principal with record_analytics_usage execution enabled"
|
|
),
|
|
"authorization": authorization,
|
|
},
|
|
status_code=403,
|
|
)
|
|
|
|
usage_id = record_usage(
|
|
session_id=body.get("session_id"),
|
|
remote=body.get("remote", "dadeschools"),
|
|
org=body.get("org", ""),
|
|
repo=body.get("repo", ""),
|
|
project_id=body.get("project_id"),
|
|
role=body.get("role", "unknown"),
|
|
model=body.get("model", "unknown"),
|
|
issue_number=body.get("issue_number") or body.get("issue"),
|
|
pr_number=body.get("pr_number") or body.get("pr"),
|
|
stage=body.get("stage", "unknown"),
|
|
input_tokens=body.get("input_tokens"),
|
|
output_tokens=body.get("output_tokens"),
|
|
total_tokens=body.get("total_tokens"),
|
|
estimated_cost_usd=body.get("estimated_cost_usd"),
|
|
latency_ms=body.get("latency_ms"),
|
|
duration_ms=body.get("duration_ms"),
|
|
status=body.get("status", "success"),
|
|
metadata=body.get("metadata"),
|
|
)
|
|
return JSONResponse(
|
|
{"ok": True, "usage_id": usage_id, "authorization": authorization},
|
|
status_code=201,
|
|
)
|
|
|
|
|
|
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(SYSTEM_HEALTH_API_PATH, api_system_health, methods=["GET"]),
|
|
Route("/system-health", system_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("/api/v1/projects", api_v1_projects, methods=["GET"]),
|
|
Route(
|
|
"/api/v1/projects/{project_id}",
|
|
api_v1_project_detail,
|
|
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("/policy", policy, methods=["GET"]),
|
|
Route("/api/v1/policy", api_v1_policy, methods=["GET"]),
|
|
Route("/api/v1/timeline", api_v1_timeline, methods=["GET"]),
|
|
Route("/analytics", analytics, methods=["GET"]),
|
|
Route("/api/analytics", api_v1_analytics, methods=["GET"]),
|
|
Route("/api/v1/analytics", api_v1_analytics, methods=["GET"]),
|
|
Route("/api/v1/analytics/usage", api_v1_analytics_ingest, methods=["POST"]),
|
|
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"]),
|
|
Route(
|
|
"/api/console/security-model",
|
|
api_console_security_model,
|
|
methods=["GET"],
|
|
),
|
|
*[
|
|
Route(path, phase_stub, methods=["GET"])
|
|
for path in STUB_PAGES
|
|
],
|
|
],
|
|
exception_handlers={405: method_not_allowed},
|
|
)
|
|
app.state.webui_bind_host = resolved_bind
|
|
return app |