LZH-YS1998
170 lines
6.8 KiB
Python
170 lines
6.8 KiB
Python
"""Read-only work-item projection visualisation for company-mode runs."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from rich.console import Group, RenderableType
|
|
from rich.panel import Panel
|
|
from rich.text import Text
|
|
from textual.widgets import Static
|
|
|
|
from ..state.models import PipelineSnapshot, PipelineWorkItemView
|
|
from ..state.store import BoardStateStore
|
|
from .render_utils import status_style, truncate_text
|
|
|
|
_STATUS_SYMBOL = {
|
|
"done": ("\u2713", "bold #22c55e"), # ✓ green
|
|
"running": ("\u25cf", "bold #38bdf8"), # ● blue
|
|
"idle": ("\u25cf", "bold #38bdf8"), # ● blue
|
|
"pending": ("\u25cb", "dim"), # ○ gray
|
|
"failed": ("\u2717", "bold #ef4444"), # ✗ red
|
|
"cancelled": ("\u2717", "bold #9333ea"), # ✗ purple
|
|
"blocked": ("\u25a0", "bold #f59e0b"), # ■ yellow
|
|
"awaiting_peer": ("\u25a0", "bold #f59e0b"),
|
|
"awaiting_review": ("\u25a0", "bold #f59e0b"),
|
|
}
|
|
|
|
|
|
def _fmt_elapsed(sec: float) -> str:
|
|
if sec <= 0:
|
|
return "--"
|
|
if sec < 60:
|
|
return f"{int(sec)}s"
|
|
if sec < 3600:
|
|
return f"{int(sec // 60)}m"
|
|
return f"{int(sec // 3600)}h{int((sec % 3600) // 60)}m"
|
|
|
|
|
|
class PipelineViewWidget(Static):
|
|
"""Render the work-item projection as an ASCII dependency view."""
|
|
|
|
def __init__(self, state: BoardStateStore) -> None:
|
|
super().__init__(id="pipeline-view")
|
|
self.state = state
|
|
self.pipeline: PipelineSnapshot | None = None
|
|
|
|
def set_pipeline(self, pipeline: PipelineSnapshot | None) -> None:
|
|
self.pipeline = pipeline
|
|
self.refresh()
|
|
|
|
def render(self) -> RenderableType:
|
|
focused = self.state.pane_focus == "main" and self.state.view_mode == "pipeline"
|
|
if self.pipeline is None or not self.pipeline.work_items:
|
|
return Panel(
|
|
Text("No work-item projection data. Select a company-mode task and press 4.", style="dim"),
|
|
title="Projection [Focused]" if focused else "Projection",
|
|
border_style="cyan" if focused else "white",
|
|
)
|
|
|
|
pipe = self.pipeline
|
|
parts: list[RenderableType] = []
|
|
|
|
# Header
|
|
header = Text()
|
|
header.append(truncate_text(pipe.parent_title, 50), style="bold white")
|
|
if pipe.profile:
|
|
header.append(f" {pipe.profile}", style="dim italic")
|
|
parts.append(header)
|
|
|
|
# Work-item rows — linear chain rendering
|
|
parts.append(self._render_work_item_chain(pipe.work_items))
|
|
|
|
# Progress footer
|
|
footer = self._render_footer(pipe)
|
|
parts.append(footer)
|
|
|
|
title = "Projection [Focused]" if focused else "Projection"
|
|
return Panel(Group(*parts), title=title, border_style="cyan" if focused else "white")
|
|
|
|
def _render_work_item_chain(self, work_items: list[PipelineWorkItemView]) -> Text:
|
|
"""Render work items as a vertical list with dependency arrows.
|
|
|
|
Using vertical layout for reliable terminal rendering — horizontal
|
|
box-drawing DAGs break on narrow terminals.
|
|
"""
|
|
text = Text()
|
|
# Build dependency lookup for rendering connectors
|
|
projection_ids = {item.projection_id for item in work_items}
|
|
|
|
for i, item in enumerate(work_items):
|
|
sym, sym_style = _STATUS_SYMBOL.get(item.status, ("\u25cb", "dim"))
|
|
is_selected = (
|
|
self.state.selected_task_id is not None
|
|
and item.task_id == self.state.selected_task_id
|
|
)
|
|
|
|
# Connector line from previous work item
|
|
if i > 0:
|
|
if item.dependencies:
|
|
# Show which projections this depends on
|
|
dep_labels = [d for d in item.dependencies if d in projection_ids]
|
|
if dep_labels:
|
|
text.append(f"\n \u2502 after: {', '.join(dep_labels)}", style="dim")
|
|
text.append("\n \u2502\n \u25bc\n", style="dim")
|
|
|
|
# Work-item box — single-line compact rendering
|
|
border_style = "bold cyan" if is_selected else "dim"
|
|
text.append(" \u250c\u2500 ", style=border_style)
|
|
text.append(f"{item.projection_id}", style="bold #38bdf8" if is_selected else "bold white")
|
|
|
|
# Parallel group indicator
|
|
if item.parallel_group:
|
|
text.append(f" \u2261{item.parallel_group}", style="dim") # ≡
|
|
|
|
# Gate indicator
|
|
if item.has_gate:
|
|
gate_label = item.gate_type or "gate"
|
|
text.append(f" \u229e {gate_label}", style="bold #f59e0b") # ⊞
|
|
|
|
text.append(" \u2500\u2510\n", style=border_style)
|
|
|
|
# Status line
|
|
text.append(" \u2502 ", style=border_style)
|
|
text.append(f"{sym} {item.status}", style=sym_style)
|
|
text.append(f" {truncate_text(item.title, 30)}", style="white")
|
|
text.append("\n", style="")
|
|
|
|
# Detail line: assignee, elapsed, tool
|
|
detail_parts: list[str] = []
|
|
if item.assigned_to:
|
|
detail_parts.append(item.assigned_to)
|
|
if item.elapsed_sec > 0 and item.status != "pending":
|
|
detail_parts.append(_fmt_elapsed(item.elapsed_sec))
|
|
if item.current_tool:
|
|
detail_parts.append(f"\u2699{item.current_tool}") # ⚙
|
|
if item.tool_elapsed_ms > 0:
|
|
detail_parts.append(f"{item.tool_elapsed_ms}ms")
|
|
if item.context_remaining_pct > 0:
|
|
detail_parts.append(f"ctx {item.context_remaining_pct}%")
|
|
if item.turn_cost_usd > 0:
|
|
detail_parts.append(f"${item.turn_cost_usd:.4f}")
|
|
|
|
if detail_parts:
|
|
text.append(" \u2502 ", style=border_style)
|
|
text.append(" ".join(detail_parts), style="dim")
|
|
text.append("\n", style="")
|
|
|
|
if item.last_tool_summary:
|
|
text.append(" \u2502 ", style=border_style)
|
|
text.append(truncate_text(item.last_tool_summary, 44), style="dim")
|
|
text.append("\n", style="")
|
|
|
|
text.append(" \u2514", style=border_style)
|
|
text.append("\u2500" * 40, style=border_style)
|
|
text.append("\u2518\n", style=border_style)
|
|
|
|
return text
|
|
|
|
@staticmethod
|
|
def _render_footer(pipe: PipelineSnapshot) -> Text:
|
|
text = Text("\n")
|
|
# Progress bar
|
|
ratio = min(1.0, pipe.done_count / pipe.total_count) if pipe.total_count > 0 else 0.0
|
|
bar_width = 20
|
|
filled = int(ratio * bar_width)
|
|
text.append("Progress: ", style="dim")
|
|
text.append("\u2588" * filled, style="bold #22c55e")
|
|
text.append("\u2591" * (bar_width - filled), style="dim")
|
|
text.append(f" {pipe.done_count}/{pipe.total_count} projected steps", style="white")
|
|
text.append(f" Elapsed: {_fmt_elapsed(pipe.elapsed_sec)}", style="dim")
|
|
return text
|