"""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"""
| Item |
Title & Details |
Next Role |
{rows}
"""
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, a missing head pin, a merge conflict, or an active terminal review lock. Items labelled status:blocked route to section 4. 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)