From 069a9af7e6aa2c2994e07199d1b0814819457017 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Fri, 24 Jul 2026 17:34:50 -0400 Subject: [PATCH] feat(webui): implement workflow traffic-control view (Closes #640) --- docs/webui-local-dev.md | 2 + tests/test_webui_traffic_control.py | 192 +++++++++++++++++ webui/app.py | 13 ++ webui/nav.py | 1 + webui/traffic_loader.py | 309 ++++++++++++++++++++++++++++ webui/traffic_views.py | 170 +++++++++++++++ 6 files changed, 687 insertions(+) create mode 100644 tests/test_webui_traffic_control.py create mode 100644 webui/traffic_loader.py create mode 100644 webui/traffic_views.py diff --git a/docs/webui-local-dev.md b/docs/webui-local-dev.md index 15c00e4..cb2bb8b 100644 --- a/docs/webui-local-dev.md +++ b/docs/webui-local-dev.md @@ -57,6 +57,8 @@ status, onboarding checklist state, and the fail-closed error payloads (#635). | `/system-health` | System-health dashboard — readiness, version/uptime, dependencies, MCP namespaces, stale-runtime parity (#639) | | `/queue` | Live PR and issue queue dashboard (#429) | | `/api/queue` | JSON queue export with pagination metadata | +| `/traffic` | Workflow traffic-control view — runnable, leased, blocked, needs-controller, terminal-complete (#640) | +| `/api/traffic` | JSON traffic-control export with state classifications and next safe role actions | | `/projects` | Project registry list with status and onboarding progress (#427, #635) | | `/projects/{id}` | Project detail + onboarding checklist | | `/api/v1/projects` | Versioned JSON registry export (#635) | diff --git a/tests/test_webui_traffic_control.py b/tests/test_webui_traffic_control.py new file mode 100644 index 0000000..6d85661 --- /dev/null +++ b/tests/test_webui_traffic_control.py @@ -0,0 +1,192 @@ +"""Tests for web UI workflow traffic-control view (#640).""" + +import sys +import unittest +from pathlib import Path +from unittest import mock + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from starlette.testclient import TestClient + +from webui.app import create_app +from webui.traffic_loader import ( + TrafficItem, + TrafficSnapshot, + load_traffic_snapshot, + snapshot_to_dict, +) +from webui.traffic_views import render_traffic_page +from allocator_service import WorkCandidate + + +class TestTrafficClassification(unittest.TestCase): + def test_runnable_candidate_classification(self): + cand = WorkCandidate( + kind="issue", + number=640, + state="open", + labels=("status:ready",), + title="Web Console: Workflow traffic-control view (Phase 1)", + priority=20, + ) + snap = load_traffic_snapshot(candidates=[cand]) + self.assertEqual(len(snap.runnable), 1) + self.assertEqual(snap.runnable[0].number, 640) + self.assertTrue(snap.runnable[0].is_safe) + self.assertEqual(snap.runnable[0].traffic_state, "runnable") + + def test_blocked_dependency_candidate_classification(self): + cand = WorkCandidate( + kind="issue", + number=643, + state="open", + labels=("status:ready",), + title="Web Console: Requests & intent preview (Phase 2)", + priority=20, + dependency_unmet=True, + dependency_reason="issue#643 depends on unresolved issue(s) #640; they are not closed", + ) + snap = load_traffic_snapshot(candidates=[cand]) + self.assertEqual(len(snap.blocked), 1) + self.assertEqual(snap.blocked[0].number, 643) + self.assertFalse(snap.blocked[0].is_safe) + self.assertEqual(snap.blocked[0].traffic_state, "blocked") + self.assertIn("depends on unresolved issue(s) #640", snap.blocked[0].block_reason) + + def test_leased_candidate_classification(self): + cand = WorkCandidate( + kind="issue", + number=640, + state="open", + labels=("status:in-progress",), + title="Web Console: Workflow traffic-control view (Phase 1)", + priority=20, + ) + lease = { + "kind": "issue", + "number": 640, + "session_id": "prgs-author-12345", + "role": "author", + "status": "active", + } + snap = load_traffic_snapshot(candidates=[cand], leases=[lease]) + self.assertEqual(len(snap.leased), 1) + self.assertEqual(snap.leased[0].number, 640) + self.assertEqual(snap.leased[0].traffic_state, "leased") + self.assertIsNotNone(snap.leased[0].lease_info) + + def test_needs_controller_candidate_classification(self): + cand = WorkCandidate( + kind="issue", + number=700, + state="open", + labels=("status:blocked",), + title="Controller intervention needed", + priority=10, + blocked=True, + ) + snap = load_traffic_snapshot(candidates=[cand]) + self.assertEqual(len(snap.needs_controller), 1) + self.assertEqual(snap.needs_controller[0].number, 700) + + +class TestTrafficLoader(unittest.TestCase): + def test_snapshot_to_dict_export(self): + cand = WorkCandidate( + kind="issue", + number=640, + state="open", + labels=("status:ready",), + title="Traffic control test", + priority=20, + ) + snap = load_traffic_snapshot(candidates=[cand]) + data = snapshot_to_dict(snap) + self.assertEqual(data["project_id"], "gitea-tools") + self.assertEqual(len(data["runnable"]), 1) + self.assertTrue(data["inventory_complete"]) + + def test_fail_closed_error_handling(self): + with mock.patch("webui.traffic_loader.load_queue_snapshot", side_effect=RuntimeError("Gitea connection failed")): + snap = load_traffic_snapshot() + self.assertIsNotNone(snap.fetch_error) + self.assertIn("Failed to load traffic state", snap.fetch_error) + self.assertEqual(len(snap.runnable), 0) + self.assertFalse(snap.inventory_complete) + + +class TestTrafficRoutesAndRendering(unittest.TestCase): + def setUp(self): + self.client = TestClient(create_app()) + + def test_traffic_html_page_rendering(self): + cand1 = WorkCandidate( + kind="issue", + number=640, + state="open", + labels=("status:ready",), + title="Traffic View Implementation", + priority=20, + ) + cand2 = WorkCandidate( + kind="issue", + number=643, + state="open", + labels=("status:ready",), + title="Dependent Feature", + priority=20, + dependency_unmet=True, + dependency_reason="issue#643 depends on unresolved issue(s) #640; they are not closed", + ) + snap = load_traffic_snapshot(candidates=[cand1, cand2]) + with mock.patch("webui.app.load_traffic_snapshot", return_value=snap): + response = self.client.get("/traffic") + + self.assertEqual(response.status_code, 200) + self.assertIn("Workflow Traffic Control", response.text) + self.assertIn("1. Runnable Lanes", response.text) + self.assertIn("3. Blocked Items", response.text) + self.assertIn("Traffic View Implementation", response.text) + self.assertIn("depends on unresolved issue(s) #640", response.text) + + def test_api_traffic_json_route(self): + cand = WorkCandidate( + kind="issue", + number=640, + state="open", + labels=("status:ready",), + title="Traffic View API Test", + priority=20, + ) + snap = load_traffic_snapshot(candidates=[cand]) + with mock.patch("webui.app.load_traffic_snapshot", return_value=snap): + response = self.client.get("/api/traffic") + + self.assertEqual(response.status_code, 200) + data = response.json() + self.assertEqual(data["project_id"], "gitea-tools") + self.assertEqual(len(data["runnable"]), 1) + self.assertEqual(data["runnable"][0]["number"], 640) + + def test_render_traffic_fail_closed_page(self): + snap = TrafficSnapshot( + project_id="gitea-tools", + repo_label="Scaled-Tech-Consulting/Gitea-Tools", + runnable=(), + leased=(), + blocked=(), + needs_controller=(), + terminal_complete=(), + next_roles=(), + fetch_error="Gitea credentials unavailable for gitea.prgs.cc", + inventory_complete=False, + ) + html = render_traffic_page(snap) + self.assertIn("Traffic data unavailable", html) + self.assertIn("Fail closed", html) + self.assertNotIn("1. Runnable Lanes", html) + + +if __name__ == "__main__": + unittest.main() diff --git a/webui/app.py b/webui/app.py index 2c7b146..e593860 100644 --- a/webui/app.py +++ b/webui/app.py @@ -42,6 +42,8 @@ from webui.lease_loader import load_lease_snapshot, snapshot_to_dict as lease_sn 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.traffic_loader import load_traffic_snapshot, snapshot_to_dict as traffic_snapshot_to_dict +from webui.traffic_views import render_traffic_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 @@ -200,6 +202,15 @@ async def api_queue(_request: Request) -> JSONResponse: return JSONResponse(queue_snapshot_to_dict(load_queue_snapshot())) +async def traffic(_request: Request) -> HTMLResponse: + snapshot = load_traffic_snapshot() + return HTMLResponse(render_traffic_page(snapshot)) + + +async def api_traffic(_request: Request) -> JSONResponse: + return JSONResponse(traffic_snapshot_to_dict(load_traffic_snapshot())) + + def _load_project_registry() -> tuple[ProjectRegistry | None, RegistryError | None]: """Load the registry, converting validation failure into a fail-closed pair.""" try: @@ -736,6 +747,8 @@ def create_app(*, bind_host: str | None = None) -> Starlette: Route("/system-health", system_health, methods=["GET"]), Route("/queue", queue, methods=["GET"]), Route("/api/queue", api_queue, methods=["GET"]), + Route("/traffic", traffic, methods=["GET"]), + Route("/api/traffic", api_traffic, 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/nav.py b/webui/nav.py index 45d1212..129b2a4 100644 --- a/webui/nav.py +++ b/webui/nav.py @@ -41,6 +41,7 @@ NAV_GROUPS: tuple[NavGroup, ...] = ( NavItem("/system-health", "System health"), )), NavGroup("Traffic", ( + NavItem("/traffic", "Traffic control"), NavItem("/queue", "Queue"), NavItem("/leases", "Leases"), NavItem("/actions", "Actions"), diff --git a/webui/traffic_loader.py b/webui/traffic_loader.py new file mode 100644 index 0000000..9753d15 --- /dev/null +++ b/webui/traffic_loader.py @@ -0,0 +1,309 @@ +"""Traffic-control view loader for Phase 1 operator web console (#640). + +Combines queue snapshots, inventory leases, dependency graph classifications, +and workflow dashboard rules to deliver full traffic-control visibility: +runnable, leased (in-progress), blocked (dependency/lock), needs-controller, +and terminal-complete candidates. +""" + +from __future__ import annotations + +import os +from dataclasses import dataclass +from typing import Any, Callable, Sequence + +from webui.project_registry import find_project, load_registry +from webui.queue_loader import load_queue_snapshot, QueueSnapshot +from webui.lease_loader import load_lease_snapshot, LeaseSnapshot +from workflow_dashboard import ( + DashboardSnapshot, + QueueEntry, + RoleNextAction, + build_workflow_dashboard, + DASHBOARD_ROLES, +) +from allocator_service import WorkCandidate + + +@dataclass(frozen=True) +class TrafficItem: + kind: str # "issue" or "pr" + number: int + title: str + traffic_state: str # "runnable", "leased", "blocked", "needs_controller", "terminal_complete" + expected_role: str + safe_for_roles: tuple[str, ...] + badges: tuple[str, ...] + block_reason: str | None = None + lease_info: dict[str, Any] | None = None + head_sha: str | None = None + + @property + def is_safe(self) -> bool: + return self.block_reason is None and bool(self.safe_for_roles) + + def as_dict(self) -> dict[str, Any]: + return { + "kind": self.kind, + "number": self.number, + "title": self.title, + "traffic_state": self.traffic_state, + "expected_role": self.expected_role, + "safe_for_roles": list(self.safe_for_roles), + "badges": list(self.badges), + "block_reason": self.block_reason, + "lease_info": self.lease_info, + "head_sha": self.head_sha, + "is_safe": self.is_safe, + } + + +@dataclass(frozen=True) +class TrafficSnapshot: + project_id: str + repo_label: str + runnable: tuple[TrafficItem, ...] + leased: tuple[TrafficItem, ...] + blocked: tuple[TrafficItem, ...] + needs_controller: tuple[TrafficItem, ...] + terminal_complete: tuple[TrafficItem, ...] + next_roles: tuple[dict[str, Any], ...] + fetch_error: str | None = None + inventory_complete: bool = True + + def as_dict(self) -> dict[str, Any]: + return { + "project_id": self.project_id, + "repo_label": self.repo_label, + "runnable": [i.as_dict() for i in self.runnable], + "leased": [i.as_dict() for i in self.leased], + "blocked": [i.as_dict() for i in self.blocked], + "needs_controller": [i.as_dict() for i in self.needs_controller], + "terminal_complete": [i.as_dict() for i in self.terminal_complete], + "next_roles": list(self.next_roles), + "fetch_error": self.fetch_error, + "inventory_complete": self.inventory_complete, + } + + +def _classify_traffic_item( + entry: QueueEntry, + *, + lease_info: dict[str, Any] | None = None, +) -> TrafficItem: + """Classify a QueueEntry into a TrafficItem with explicit traffic state.""" + badges = list(entry.badges) + block_reason = entry.block_reason + expected_role = entry.expected_role + + entry_is_safe = entry.block_reason is None and bool(entry.safe_for_roles) + if lease_info is not None or "in-progress" in badges or "claimed" in badges: + state = "leased" + elif expected_role == "reconciler" or "terminal-lock" in badges: + state = "terminal_complete" + elif expected_role == "controller" or "contaminated" in badges or "needs-controller" in badges: + state = "needs_controller" + elif block_reason is not None or "blocked" in badges or "dependency-unmet" in badges or "blocked-by-terminal" in badges: + state = "blocked" + elif entry_is_safe: + state = "runnable" + else: + state = "needs_controller" + + return TrafficItem( + kind=entry.kind, + number=entry.number, + title=entry.title, + traffic_state=state, + expected_role=expected_role, + safe_for_roles=entry.safe_for_roles, + badges=tuple(badges), + block_reason=block_reason, + lease_info=lease_info, + head_sha=entry.head_sha, + ) + + +def load_traffic_snapshot( + *, + candidates: Sequence[WorkCandidate] | None = None, + leases: Sequence[dict[str, Any]] | None = None, + terminal_pr: int | None = None, + fetch_queue_snapshot: Callable[[], QueueSnapshot] | None = None, + fetch_lease_snapshot: Callable[[], LeaseSnapshot] | None = None, + project_id: str = "gitea-tools", +) -> TrafficSnapshot: + """Load and compute the traffic-control snapshot.""" + try: + reg = load_registry() + proj = find_project(reg, project_id) + repo_label = proj.remote_repo if proj else "Scaled-Tech-Consulting/Gitea-Tools" + except Exception: + repo_label = "Scaled-Tech-Consulting/Gitea-Tools" + + # Injected candidates path (pure unit testing) + if candidates is not None: + dashboard = build_workflow_dashboard( + candidates=candidates, + leases=leases, + terminal_pr=terminal_pr, + inventory_complete=True, + ) + return _build_traffic_snapshot_from_dashboard( + project_id=project_id, + repo_label=repo_label, + dashboard=dashboard, + leases=leases or (), + ) + + # Live snapshot loading + q_loader = fetch_queue_snapshot or load_queue_snapshot + l_loader = fetch_lease_snapshot or load_lease_snapshot + + try: + q_snap = q_loader() + l_snap = l_loader() + except Exception as exc: # noqa: BLE001 + return TrafficSnapshot( + project_id=project_id, + repo_label=repo_label, + runnable=(), + leased=(), + blocked=(), + needs_controller=(), + terminal_complete=(), + next_roles=(), + fetch_error=f"Failed to load traffic state: {exc}", + inventory_complete=False, + ) + + if q_snap.fetch_error or l_snap.fetch_error: + err = q_snap.fetch_error or l_snap.fetch_error + return TrafficSnapshot( + project_id=project_id, + repo_label=repo_label, + runnable=(), + leased=(), + blocked=(), + needs_controller=(), + terminal_complete=(), + next_roles=(), + fetch_error=err, + inventory_complete=False, + ) + + # Build WorkCandidates from live queue snapshot + candidate_list: list[WorkCandidate] = [] + for pr in q_snap.prs: + linked = int(pr.extra["linked_issue"]) if pr.extra.get("linked_issue") and pr.extra["linked_issue"].isdigit() else None + candidate_list.append( + WorkCandidate( + kind="pr", + number=pr.number, + state="open", + labels=(), + title=pr.title, + priority=10 if "request-changes" in pr.badges else 5, + request_changes_current_head="request-changes" in pr.badges, + approval_on_current_head="approved" in pr.badges or "merge-ready" in pr.badges, + mergeable="blocked" not in pr.badges, + blocked="blocked" in pr.badges, + linked_issue_number=linked, + ) + ) + + for issue in q_snap.issues: + labels = [b for b in issue.badges if b.startswith("status:") or b in ("discussion", "blocked", "ready", "in-progress")] + candidate_list.append( + WorkCandidate( + kind="issue", + number=issue.number, + state="open", + labels=tuple(labels), + title=issue.title, + priority=20 if "status:ready" in labels else 10, + blocked="blocked" in labels or "status:blocked" in labels, + ) + ) + + raw_leases: list[dict[str, Any]] = [] + if l_snap.claim_inventory and "active_claims" in l_snap.claim_inventory: + raw_leases.extend(l_snap.claim_inventory["active_claims"]) + for r_lease in l_snap.reviewer_leases: + raw_leases.append(r_lease) + + dashboard = build_workflow_dashboard( + candidates=candidate_list, + leases=raw_leases, + inventory_complete=q_snap.pr_pagination.inventory_complete if q_snap.pr_pagination else True, + ) + + return _build_traffic_snapshot_from_dashboard( + project_id=project_id, + repo_label=repo_label, + dashboard=dashboard, + leases=raw_leases, + ) + + +def _build_traffic_snapshot_from_dashboard( + *, + project_id: str, + repo_label: str, + dashboard: DashboardSnapshot, + leases: Sequence[dict[str, Any]], +) -> TrafficSnapshot: + """Classify dashboard entries into the 5 traffic state buckets.""" + all_entries = dashboard.open_prs + dashboard.open_issues + + # Map leased work numbers + lease_map: dict[tuple[str, int], dict[str, Any]] = {} + for lease in leases: + if isinstance(lease, dict): + kind = str(lease.get("kind") or lease.get("work_kind") or "issue").lower() + num = lease.get("number") or lease.get("work_number") or lease.get("issue_number") or lease.get("pr_number") + if num is not None: + try: + lease_map[(kind, int(num))] = lease + except (ValueError, TypeError): + pass + + runnable: list[TrafficItem] = [] + leased: list[TrafficItem] = [] + blocked: list[TrafficItem] = [] + needs_controller: list[TrafficItem] = [] + terminal_complete: list[TrafficItem] = [] + + for entry in all_entries: + l_info = lease_map.get((entry.kind, entry.number)) + item = _classify_traffic_item(entry, lease_info=l_info) + + if item.traffic_state == "leased": + leased.append(item) + elif item.traffic_state == "terminal_complete": + terminal_complete.append(item) + elif item.traffic_state == "blocked": + blocked.append(item) + elif item.traffic_state == "needs_controller": + needs_controller.append(item) + else: + runnable.append(item) + + next_roles = [dashboard.next_safe_by_role[r].as_dict() for r in DASHBOARD_ROLES if r in dashboard.next_safe_by_role] + + return TrafficSnapshot( + project_id=project_id, + repo_label=repo_label, + runnable=tuple(runnable), + leased=tuple(leased), + blocked=tuple(blocked), + needs_controller=tuple(needs_controller), + terminal_complete=tuple(terminal_complete), + next_roles=tuple(next_roles), + fetch_error=None, + inventory_complete=dashboard.inventory_complete, + ) + + +def snapshot_to_dict(snapshot: TrafficSnapshot) -> dict[str, Any]: + return snapshot.as_dict() diff --git a/webui/traffic_views.py b/webui/traffic_views.py new file mode 100644 index 0000000..e247915 --- /dev/null +++ b/webui/traffic_views.py @@ -0,0 +1,170 @@ +"""HTML rendering for Phase 1 Traffic-Control View (#640).""" + +from __future__ import annotations + +from html import escape +from typing import Sequence + +from webui.layout import render_page +from webui.traffic_loader import TrafficItem, TrafficSnapshot + + +def _render_badges(badges: Sequence[str]) -> str: + if not badges: + return "" + out = [] + for b in badges: + cls = "badge" + b_lower = b.lower() + if "blocked" in b_lower or "unmet" in b_lower: + cls += " badge-blocked" + elif "claimed" in b_lower or "in-progress" in b_lower or "leased" in b_lower: + cls += " badge-claimed" + elif "review" in b_lower or "ready" in b_lower: + cls += " badge-in-review" + elif "duplicate" in b_lower: + cls += " badge-duplicate" + elif "stale" in b_lower: + cls += " badge-stale" + out.append(f'{escape(b)}') + return f'
{"".join(out)}
' + + +def _render_traffic_item_row(item: TrafficItem) -> str: + kind_label = escape(item.kind.upper()) + num_str = f"#{item.number}" + title_str = escape(item.title) + role_str = escape(item.expected_role) + badges_html = _render_badges(item.badges) + + reason_html = "" + if item.block_reason: + reason_html = f'
Blocker: {escape(item.block_reason)}
' + + lease_html = "" + if item.lease_info: + owner = escape(str(item.lease_info.get("session_id") or item.lease_info.get("reviewer_identity") or "active worker")) + lease_html = f'
Lease: {owner}
' + + return f""" + {kind_label} {num_str} + +
{title_str} {badges_html}
+ {reason_html} + {lease_html} + + {role_str} +""" + + +def _render_traffic_table(items: Sequence[TrafficItem], empty_message: str) -> str: + if not items: + return f'

{escape(empty_message)}

' + + rows = "".join(_render_traffic_item_row(item) for item in items) + return f""" + + + + + + + + + {rows} + +
ItemTitle & DetailsNext Role
""" + + +def _render_next_roles(next_roles: Sequence[dict]) -> str: + if not next_roles: + return "" + + cards = [] + for r in next_roles: + role = escape(r.get("role", "unknown")) + status = r.get("status", "idle") + prompt = escape(r.get("prompt", "")) + + status_cls = "badge-health-ok" if status == "safe" else ("badge-blocked" if "blocked" in status else "badge-health-skipped") + cards.append(f"""
+
+

Role: {role}

+ status: {escape(status)} +
+

{prompt}

+
""") + + return f"""
+

Next Safe Role Actions

+ {"".join(cards)} +
""" + + +def render_traffic_page(snapshot: TrafficSnapshot) -> str: + """Render the full HTML view for workflow traffic control.""" + if snapshot.fetch_error: + body = f"""

Workflow Traffic Control

+

Repository: {escape(snapshot.repo_label)}

+
+

Traffic data unavailable

+

{escape(snapshot.fetch_error)}

+

Fail closed: traffic state cannot be established cleanly. Check credentials or remote connectivity.

+
""" + return render_page(title="Traffic Control", body_html=body) + + runnable_count = len(snapshot.runnable) + leased_count = len(snapshot.leased) + blocked_count = len(snapshot.blocked) + controller_count = len(snapshot.needs_controller) + terminal_count = len(snapshot.terminal_complete) + + summary_bar = f"""
+
Runnable: {runnable_count}
+
Leased: {leased_count}
+
Blocked: {blocked_count}
+
Needs Controller: {controller_count}
+
Terminal Complete: {terminal_count}
+
""" + + next_roles_html = _render_next_roles(snapshot.next_roles) + + sections_html = f""" +
+

1. Runnable Lanes (Ready for Allocation)

+

Safe work items with no unmet dependencies or active leases. Safe for allocation.

+ {_render_traffic_table(snapshot.runnable, "No runnable items ready for allocation.")} +
+ +
+

2. In-Progress Work (Active Leases)

+

Work items currently leased and actively being worked by an assigned role session.

+ {_render_traffic_table(snapshot.leased, "No active leases in flight.")} +
+ +
+

3. Blocked Items (Dependencies / Locks)

+

Items blocked by unmet dependency issues, status:blocked, or active terminal review locks. Never presented as safe.

+ {_render_traffic_table(snapshot.blocked, "No blocked items.")} +
+ +
+

4. Needs Controller Intervention

+

Items requiring controller routing, diagnosis, or cross-role assignment.

+ {_render_traffic_table(snapshot.needs_controller, "No items requiring controller intervention.")} +
+ +
+

5. Terminal / Complete Candidates

+

Items ready for terminal reconciliation or post-merge worktree cleanup.

+ {_render_traffic_table(snapshot.terminal_complete, "No terminal complete candidates.")} +
+""" + + body = f"""

Workflow Traffic Control

+

Repository: {escape(snapshot.repo_label)}

+{summary_bar} +{next_roles_html} +{sections_html}""" + + return render_page(title="Traffic Control", body_html=body)