Local entry point for MCP Control Plane operational views.
"
""
+ "- Queue — live PR and issue dashboard (#429)
"
"- Projects — registry and onboarding (#427)
"
"- Prompts — canonical workflow prompt library (#428)
"
"- Runtime — MCP health and stale-runtime detection (#430)
"
@@ -52,6 +55,15 @@ async def health(_request: Request) -> JSONResponse:
})
+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))
@@ -151,6 +163,8 @@ def create_app() -> Starlette:
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"]),
diff --git a/webui/layout.py b/webui/layout.py
index d948204..a506c2c 100644
--- a/webui/layout.py
+++ b/webui/layout.py
@@ -4,6 +4,7 @@ from __future__ import annotations
NAV_ITEMS = (
("/", "Home"),
+ ("/queue", "Queue"),
("/projects", "Projects"),
("/prompts", "Prompts"),
("/runtime", "Runtime"),
@@ -145,6 +146,20 @@ def render_page(*, title: str, body_html: str, extra_head: str = "") -> str:
}}
.copy-btn:hover {{ filter: brightness(1.08); }}
.muted {{ color: var(--muted); }}
+ .badges {{ display: inline-flex; flex-wrap: wrap; gap: 0.35rem; margin-left: 0.5rem; }}
+ .badge {{
+ font-size: 0.72rem;
+ padding: 0.1rem 0.45rem;
+ border-radius: 999px;
+ border: 1px solid var(--border);
+ color: var(--muted);
+ text-transform: lowercase;
+ }}
+ .badge-claimed {{ color: #8fd19e; border-color: #3d6b4a; }}
+ .badge-blocked {{ color: #f0a8a8; border-color: #7a3b3b; }}
+ .badge-in-review {{ color: #9ec8f0; border-color: #3d5f7a; }}
+ .badge-duplicate {{ color: #e0c27a; border-color: #6b5730; }}
+ .badge-stale {{ color: #c9b8e8; border-color: #5a4a78; }}
{extra_head}
diff --git a/webui/queue_loader.py b/webui/queue_loader.py
new file mode 100644
index 0000000..57ebeeb
--- /dev/null
+++ b/webui/queue_loader.py
@@ -0,0 +1,385 @@
+"""Load live Gitea PR/issue queue state for the web UI dashboard (#429)."""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from typing import Any, Callable
+from urllib.parse import urlparse
+
+from gitea_auth import api_fetch_page, api_get_all, get_auth_header, repo_api_url
+
+from webui.project_registry import ProjectRecord, load_registry
+
+_CLOSES_RE = re.compile(r"(?:closes|fixes|resolves)\s+#(\d+)", re.I)
+_ISSUE_REF_RE = re.compile(r"#(\d+)")
+_STALE_DAYS = 14
+
+
+@dataclass(frozen=True)
+class PaginationMeta:
+ page: int
+ per_page: int
+ returned_count: int
+ has_more: bool
+ is_final_page: bool
+ inventory_complete: bool
+ pages_fetched: int
+
+
+@dataclass(frozen=True)
+class QueueItem:
+ number: int
+ title: str
+ badges: tuple[str, ...]
+ extra: dict[str, str]
+
+
+@dataclass(frozen=True)
+class QueueSnapshot:
+ project_id: str
+ repo_label: str
+ prs: tuple[QueueItem, ...]
+ issues: tuple[QueueItem, ...]
+ pr_pagination: PaginationMeta | None
+ issue_pagination: PaginationMeta | None
+ fetch_error: str | None = None
+
+
+def _host_from_url(remote_host: str) -> str:
+ parsed = urlparse(remote_host.strip())
+ return parsed.netloc or remote_host.strip().rstrip("/")
+
+
+def _extract_linked_issue(title: str | None, body: str | None = None) -> int | None:
+ for text in (title, body):
+ if not text:
+ continue
+ match = _CLOSES_RE.search(text)
+ if match:
+ return int(match.group(1))
+ if body:
+ refs = _ISSUE_REF_RE.findall(body)
+ if refs:
+ return int(refs[0])
+ return None
+
+
+def _is_stale(updated_at: str | None) -> bool:
+ if not updated_at:
+ return False
+ try:
+ normalized = updated_at.replace("Z", "+00:00")
+ parsed = datetime.fromisoformat(normalized)
+ if parsed.tzinfo is None:
+ parsed = parsed.replace(tzinfo=timezone.utc)
+ age = datetime.now(timezone.utc) - parsed.astimezone(timezone.utc)
+ return age.days >= _STALE_DAYS
+ except ValueError:
+ return False
+
+
+def _classify_pr(
+ pr: dict,
+ *,
+ issue_claimed: set[int],
+ issue_to_prs: dict[int, list[int]],
+) -> tuple[str, ...]:
+ badges: list[str] = []
+ mergeable = pr.get("mergeable")
+ if mergeable is False:
+ badges.append("blocked")
+ linked = _extract_linked_issue(pr.get("title"), pr.get("body"))
+ if linked is not None:
+ if linked in issue_claimed:
+ badges.append("claimed")
+ if len(issue_to_prs.get(linked, [])) > 1:
+ badges.append("duplicate")
+ if _is_stale(pr.get("updated_at")):
+ badges.append("stale")
+ if mergeable is True and "blocked" not in badges:
+ badges.append("in-review")
+ if not badges:
+ badges.append("open")
+ return tuple(dict.fromkeys(badges))
+
+
+def _classify_issue(
+ issue: dict,
+ *,
+ linked_prs: list[int],
+) -> tuple[str, ...]:
+ badges: list[str] = []
+ labels = [lb.get("name", "") for lb in issue.get("labels", [])]
+ if "status:in-progress" in labels:
+ badges.append("claimed")
+ if len(linked_prs) > 1:
+ badges.append("duplicate")
+ elif linked_prs and "claimed" not in badges:
+ badges.append("in-review")
+ if _is_stale(issue.get("updated_at")):
+ badges.append("stale")
+ if not badges:
+ badges.append("open")
+ return tuple(dict.fromkeys(badges))
+
+
+def _format_pr_item(pr: dict, badges: tuple[str, ...]) -> QueueItem:
+ head = pr.get("head") or {}
+ base = pr.get("base") or {}
+ mergeable = pr.get("mergeable")
+ merge_label = (
+ "mergeable" if mergeable is True else "conflicted" if mergeable is False else "unknown"
+ )
+ linked = _extract_linked_issue(pr.get("title"), pr.get("body"))
+ return QueueItem(
+ number=int(pr["number"]),
+ title=str(pr.get("title") or ""),
+ badges=badges,
+ extra={
+ "branch": f"{head.get('ref', '?')} → {base.get('ref', '?')}",
+ "head_sha": str(head.get("sha") or "")[:12],
+ "mergeable": merge_label,
+ "linked_issue": str(linked) if linked is not None else "",
+ },
+ )
+
+
+def _format_issue_item(issue: dict, badges: tuple[str, ...]) -> QueueItem:
+ labels = ", ".join(lb.get("name", "") for lb in issue.get("labels", []))
+ assignee = (issue.get("assignee") or {}).get("login", "")
+ return QueueItem(
+ number=int(issue["number"]),
+ title=str(issue.get("title") or ""),
+ badges=badges,
+ extra={
+ "labels": labels or "—",
+ "assignee": assignee or "unassigned",
+ "state": str(issue.get("state") or ""),
+ },
+ )
+
+
+def _pagination_from_pages(
+ *,
+ per_page: int,
+ pages_fetched: int,
+ returned_count: int,
+ is_final_page: bool,
+) -> PaginationMeta:
+ return PaginationMeta(
+ page=1,
+ per_page=per_page,
+ returned_count=returned_count,
+ has_more=not is_final_page,
+ is_final_page=is_final_page,
+ inventory_complete=is_final_page,
+ pages_fetched=pages_fetched,
+ )
+
+
+def _fetch_prs(
+ host: str,
+ org: str,
+ repo: str,
+ auth: str,
+ *,
+ per_page: int = 50,
+) -> tuple[list[dict], PaginationMeta]:
+ url = f"{repo_api_url(host, org, repo)}/pulls?state=open"
+ all_raw: list[dict] = []
+ pages_fetched = 0
+ is_final = False
+ page = 1
+ while pages_fetched < 20:
+ raw_page, meta = api_fetch_page(url, auth, page=page, limit=per_page)
+ pages_fetched += 1
+ all_raw.extend(raw_page)
+ is_final = bool(meta["is_final_page"])
+ if is_final:
+ break
+ page += 1
+ pagination = _pagination_from_pages(
+ per_page=per_page,
+ pages_fetched=pages_fetched,
+ returned_count=len(all_raw),
+ is_final_page=is_final,
+ )
+ return all_raw, pagination
+
+
+def _fetch_issues(
+ host: str,
+ org: str,
+ repo: str,
+ auth: str,
+ *,
+ per_page: int = 50,
+) -> tuple[list[dict], PaginationMeta]:
+ url = f"{repo_api_url(host, org, repo)}/issues?state=open&type=issues"
+ all_raw: list[dict] = []
+ page = 1
+ pages_fetched = 0
+ is_final = False
+ while pages_fetched < 20:
+ raw_page, meta = api_fetch_page(url, auth, page=page, limit=per_page)
+ pages_fetched += 1
+ all_raw.extend(raw_page)
+ is_final = bool(meta["is_final_page"])
+ if is_final:
+ break
+ page += 1
+ pagination = _pagination_from_pages(
+ per_page=per_page,
+ pages_fetched=pages_fetched,
+ returned_count=len(all_raw),
+ is_final_page=is_final,
+ )
+ return all_raw, pagination
+
+
+def load_queue_snapshot(
+ project_id: str | None = None,
+ *,
+ fetch_prs: Callable[..., tuple[list[dict], PaginationMeta]] | None = None,
+ fetch_issues: Callable[..., tuple[list[dict], PaginationMeta]] | None = None,
+) -> QueueSnapshot:
+ """Load open PR and issue queues for a registry project (default: first entry)."""
+ registry = load_registry()
+ project: ProjectRecord | None = None
+ if project_id:
+ for entry in registry.projects:
+ if entry.id == project_id:
+ project = entry
+ break
+ else:
+ project = registry.projects[0] if registry.projects else None
+
+ if project is None:
+ return QueueSnapshot(
+ project_id=project_id or "",
+ repo_label="",
+ prs=(),
+ issues=(),
+ pr_pagination=None,
+ issue_pagination=None,
+ fetch_error="project not found in registry",
+ )
+
+ 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
+ auth = get_auth_header(host) if using_live_fetch else "test-auth"
+ if using_live_fetch and not auth:
+ return QueueSnapshot(
+ project_id=project.id,
+ repo_label=f"{project.gitea_owner}/{project.repo_name}",
+ prs=(),
+ issues=(),
+ pr_pagination=None,
+ issue_pagination=None,
+ fetch_error=(
+ f"Gitea credentials unavailable for {host}; "
+ "queue cannot be loaded (fail closed — not showing empty queue)"
+ ),
+ )
+
+ try:
+ raw_prs, pr_pagination = pr_fetch(
+ host, project.gitea_owner, project.repo_name, auth
+ )
+ raw_issues, issue_pagination = issue_fetch(
+ host, project.gitea_owner, project.repo_name, auth
+ )
+ except Exception as exc: # noqa: BLE001 — surface operator-visible fetch errors
+ return QueueSnapshot(
+ project_id=project.id,
+ repo_label=f"{project.gitea_owner}/{project.repo_name}",
+ prs=(),
+ issues=(),
+ pr_pagination=None,
+ issue_pagination=None,
+ fetch_error=f"Gitea fetch failed: {exc}",
+ )
+
+ issue_to_prs: dict[int, list[int]] = {}
+ for pr in raw_prs:
+ linked = _extract_linked_issue(pr.get("title"), pr.get("body"))
+ if linked is not None:
+ issue_to_prs.setdefault(linked, []).append(int(pr["number"]))
+
+ issue_claimed = {
+ int(i["number"])
+ for i in raw_issues
+ if any(lb.get("name") == "status:in-progress" for lb in i.get("labels", []))
+ }
+
+ pr_items = tuple(
+ _format_pr_item(
+ pr,
+ _classify_pr(
+ pr,
+ issue_claimed=issue_claimed,
+ issue_to_prs=issue_to_prs,
+ ),
+ )
+ for pr in sorted(raw_prs, key=lambda p: int(p["number"]), reverse=True)
+ )
+ issue_items = tuple(
+ _format_issue_item(
+ issue,
+ _classify_issue(
+ issue,
+ linked_prs=issue_to_prs.get(int(issue["number"]), []),
+ ),
+ )
+ for issue in sorted(raw_issues, key=lambda i: int(i["number"]), reverse=True)
+ )
+
+ return QueueSnapshot(
+ project_id=project.id,
+ repo_label=f"{project.gitea_owner}/{project.repo_name}",
+ prs=pr_items,
+ issues=issue_items,
+ pr_pagination=pr_pagination,
+ issue_pagination=issue_pagination,
+ )
+
+
+def snapshot_to_dict(snapshot: QueueSnapshot) -> dict[str, Any]:
+ """JSON-serializable export for /api/queue."""
+
+ def _page(meta: PaginationMeta | None) -> dict[str, Any] | None:
+ if meta is None:
+ return None
+ return {
+ "page": meta.page,
+ "per_page": meta.per_page,
+ "returned_count": meta.returned_count,
+ "has_more": meta.has_more,
+ "is_final_page": meta.is_final_page,
+ "inventory_complete": meta.inventory_complete,
+ "pages_fetched": meta.pages_fetched,
+ }
+
+ def _item(item: QueueItem) -> dict[str, Any]:
+ return {
+ "number": item.number,
+ "title": item.title,
+ "badges": list(item.badges),
+ **item.extra,
+ }
+
+ return {
+ "project_id": snapshot.project_id,
+ "repo": snapshot.repo_label,
+ "fetch_error": snapshot.fetch_error,
+ "prs": [_item(p) for p in snapshot.prs],
+ "issues": [_item(i) for i in snapshot.issues],
+ "pagination": {
+ "prs": _page(snapshot.pr_pagination),
+ "issues": _page(snapshot.issue_pagination),
+ },
+ }
\ No newline at end of file
diff --git a/webui/queue_views.py b/webui/queue_views.py
new file mode 100644
index 0000000..8294b8e
--- /dev/null
+++ b/webui/queue_views.py
@@ -0,0 +1,110 @@
+"""HTML views for the live Gitea queue dashboard (#429)."""
+
+from __future__ import annotations
+
+import html
+
+from webui.queue_loader import PaginationMeta, QueueItem, QueueSnapshot
+
+
+def _badge_html(badges: tuple[str, ...]) -> str:
+ if not badges:
+ return ""
+ chips = "".join(
+ f'{html.escape(b)}'
+ for b in badges
+ )
+ return f'{chips}'
+
+
+def _pagination_html(label: str, meta: PaginationMeta | None) -> str:
+ if meta is None:
+ return (
+ f"{html.escape(label)} pagination: "
+ "unavailable
"
+ )
+ status = "complete" if meta.inventory_complete else "partial"
+ more = "yes" if meta.has_more else "no"
+ return (
+ f"{html.escape(label)} pagination ({status}): "
+ f"returned {meta.returned_count} · per_page {meta.per_page} · "
+ f"pages_fetched {meta.pages_fetched} · has_more {more} · "
+ f"final_page {'yes' if meta.is_final_page else 'no'}
"
+ )
+
+
+def _queue_table(
+ *,
+ title: str,
+ items: tuple[QueueItem, ...],
+ columns: tuple[tuple[str, str], ...],
+) -> str:
+ if not items:
+ return f"{html.escape(title)}
No open items.
"
+
+ headers = "".join(f"{html.escape(label)} | " for _, label in columns)
+ rows = []
+ for item in items:
+ cells = []
+ for key, _ in columns:
+ if key == "number":
+ cells.append(f"#{item.number} | ")
+ elif key == "title":
+ cells.append(
+ f"{html.escape(item.title)}{_badge_html(item.badges)} | "
+ )
+ else:
+ cells.append(f"{html.escape(item.extra.get(key, ''))} | ")
+ rows.append("" + "".join(cells) + "
")
+
+ body = (
+ f"{html.escape(title)}
"
+ f"{headers}
"
+ f"{''.join(rows)}
"
+ )
+ return body
+
+
+def render_queue_page(snapshot: QueueSnapshot) -> str:
+ error_block = ""
+ if snapshot.fetch_error:
+ error_block = (
+ f'Queue unavailable: '
+ f"{html.escape(snapshot.fetch_error)}
Live queue
"
+ f"Repository: {html.escape(snapshot.repo_label)} "
+ f"· project {html.escape(snapshot.project_id)}
"
+ f"{error_block}"
+ f"{_pagination_html('PR', snapshot.pr_pagination)}"
+ f"{pr_section}"
+ f"{_pagination_html('Issue', snapshot.issue_pagination)}"
+ f"{issue_section}"
+ "Read-only MVP — claims, reviews, and merges stay in Gitea/MCP tools.
"
+ )
\ No newline at end of file