Coverage for little_loops / cli / loop / layout.py: 55%
986 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-04-11 23:20 -0500
1"""FSM diagram layout engine.
3Implements adaptive layout for FSM diagrams using a Sugiyama-inspired
4layered graph drawing approach. Supports topology detection, vertical
5linear chains, side-by-side branching, and margin back-edge rendering.
7Extracted from info.py and extended with adaptive layout capabilities.
8"""
10from __future__ import annotations
12import re
13from collections import deque
15from wcwidth import wcswidth as _wcswidth
16from wcwidth import wcwidth as _wcwidth
18from little_loops.cli.output import colorize, terminal_width
19from little_loops.fsm.schema import FSMLoop, StateConfig
21_ANSI_ESCAPE_RE = re.compile(r"\033\[[0-9;]*m")
23# ---------------------------------------------------------------------------
24# Edge label colorization
25# ---------------------------------------------------------------------------
27_EDGE_LABEL_COLORS: dict[str, str] = {
28 "yes": "32",
29 "no": "38;5;208",
30 "error": "31",
31 "partial": "33",
32 "next": "2",
33 "_": "2",
34 "blocked": "31",
35 "retry_exhausted": "38;5;208",
36}
39def _colorize_label(label: str) -> str:
40 """Colorize a (possibly compound) edge label like 'no/error'."""
41 parts = label.split("/")
42 code = ""
43 for part in parts:
44 if part in ("no", "error"):
45 code = _EDGE_LABEL_COLORS["no"]
46 break
47 if part == "partial" and not code:
48 code = _EDGE_LABEL_COLORS["partial"]
49 elif part == "yes" and not code:
50 code = _EDGE_LABEL_COLORS["yes"]
51 elif part in ("next", "_") and not code:
52 code = _EDGE_LABEL_COLORS["next"]
53 return colorize(label, code) if code else label
56def _edge_line_color(label: str) -> str:
57 """Return the ANSI SGR code to use for connector characters of an edge.
59 Applies the same priority cascade as ``_colorize_label`` so that line
60 characters (│, ─, ▼, ▶, corners) match the semantic color of their label.
61 Returns an empty string when no color applies (callers treat this as "no-op").
62 """
63 parts = label.split("/")
64 code = ""
65 for part in parts:
66 if part in ("no", "error", "blocked", "retry_exhausted"):
67 return _EDGE_LABEL_COLORS.get(part, "31")
68 if part == "partial" and not code:
69 code = _EDGE_LABEL_COLORS["partial"]
70 elif part == "yes" and not code:
71 code = _EDGE_LABEL_COLORS["yes"]
72 elif part in ("next", "_") and not code:
73 code = _EDGE_LABEL_COLORS["next"]
74 return code
77def _colorize_diagram_labels(diagram: str, colors: dict[str, str] | None = None) -> str:
78 """Apply ANSI color to known edge labels in a rendered diagram string.
80 Labels are colorized only when bounded by box-drawing or whitespace chars
81 to avoid coloring partial matches inside state names.
83 Args:
84 colors: Optional label→SGR-code mapping; falls back to _EDGE_LABEL_COLORS if None.
85 """
86 label_colors = colors if colors is not None else _EDGE_LABEL_COLORS
87 for label, code in label_colors.items():
88 colored = colorize(label, code)
89 diagram = re.sub(
90 r"(?<=[─ │▶\n])" + re.escape(label) + r"(?=[─ │▶\n])",
91 colored,
92 diagram,
93 )
94 return diagram
97# ---------------------------------------------------------------------------
98# State box badge definitions
99# ---------------------------------------------------------------------------
101_ACTION_TYPE_BADGES: dict[str, str] = {
102 "prompt": "\u2726", # ✦
103 "slash_command": "/\u2501\u25ba", # /━►
104 "shell": "\u276f_", # ❯_
105 "mcp_tool": "\u26a1", # ⚡
106}
108_SUB_LOOP_BADGE = "\u21b3\u27f3" # ↳⟳
109_ROUTE_BADGE = "\u2443" # ⑃
112def _badge_display_width(badge: str) -> int:
113 """Compute terminal display width of a badge string using wcwidth."""
114 w = _wcswidth(badge)
115 return w if w >= 0 else len(badge)
118def _get_state_badge(state: StateConfig | None, badges: dict[str, str] | None = None) -> str:
119 """Return the unicode badge string for a state, or '' if none."""
120 if state is None:
121 return ""
122 effective = {**_ACTION_TYPE_BADGES, **(badges or {})}
123 sub_loop_badge = (badges or {}).get("sub_loop", _SUB_LOOP_BADGE)
124 route_badge = (badges or {}).get("route", _ROUTE_BADGE)
125 if state.loop is not None:
126 return sub_loop_badge
127 if state.action_type:
128 return effective.get(state.action_type, f"[{state.action_type}]")
129 if state.action:
130 return effective["shell"]
131 if state.route is not None:
132 return route_badge
133 return ""
136# ---------------------------------------------------------------------------
137# Box content helpers for multi-row diagram boxes
138# ---------------------------------------------------------------------------
141def _box_inner_lines(
142 state: StateConfig | None,
143 display_label: str,
144 verbose: bool,
145 inner_width: int,
146) -> list[str]:
147 """Return interior lines for a state box (between top and bottom borders).
149 The first line is always ``display_label`` + type badge (if any).
150 Subsequent lines are action content lines. All lines fit within
151 ``inner_width`` characters (content is truncated or wrapped accordingly).
152 """
153 # Badge is now rendered in the top-right corner by _draw_box; name row is label only
154 name_line = display_label[:inner_width]
156 lines: list[str] = [name_line]
158 # Action lines
159 if state and state.action:
160 action_src = [ln.rstrip() for ln in state.action.strip().splitlines()]
161 if verbose:
162 for src in action_src:
163 if not src:
164 continue
165 # Wrap long lines to inner_width
166 while len(src) > inner_width:
167 lines.append(src[:inner_width])
168 src = src[inner_width:]
169 if src:
170 lines.append(src)
171 else:
172 # Show first non-empty line, truncated
173 first = next((ln for ln in action_src if ln), "")
174 if len(first) > inner_width:
175 first = first[: inner_width - 1] + "\u2026"
176 if first:
177 lines.append(first)
179 return lines
182# ---------------------------------------------------------------------------
183# Topology detection
184# ---------------------------------------------------------------------------
187def _collect_edges(fsm: FSMLoop) -> list[tuple[str, str, str]]:
188 """Collect all (source, target, label) edges from an FSM."""
189 edges: list[tuple[str, str, str]] = []
190 for name, state in fsm.states.items():
191 if state.on_yes:
192 edges.append((name, state.on_yes, "yes"))
193 if state.on_no:
194 edges.append((name, state.on_no, "no"))
195 if state.on_error:
196 edges.append((name, state.on_error, "error"))
197 if state.on_partial:
198 edges.append((name, state.on_partial, "partial"))
199 if state.on_blocked:
200 edges.append((name, state.on_blocked, "blocked"))
201 if state.on_retry_exhausted:
202 edges.append((name, state.on_retry_exhausted, "retry_exhausted"))
203 if state.next:
204 edges.append((name, state.next, "next"))
205 if state.route:
206 for verdict, target in state.route.routes.items():
207 edges.append((name, target, verdict))
208 if state.route.default:
209 edges.append((name, state.route.default, "_"))
210 for verdict, target in state.extra_routes.items():
211 edges.append((name, target, verdict))
212 return edges
215def _bfs_order(initial: str, edges: list[tuple[str, str, str]]) -> tuple[list[str], dict[str, int]]:
216 """BFS from initial state. Returns (order, depth_map)."""
217 order: list[str] = []
218 depth: dict[str, int] = {initial: 0}
219 queue: deque[str] = deque([initial])
220 while queue:
221 node = queue.popleft()
222 order.append(node)
223 for src, dst, _ in edges:
224 if src == node and dst not in depth:
225 depth[dst] = depth[node] + 1
226 queue.append(dst)
227 return order, depth
230def _trace_main_path(
231 fsm: FSMLoop, edges: list[tuple[str, str, str]]
232) -> tuple[list[str], set[tuple[str, str]]]:
233 """Trace the main (happy) path through the FSM."""
234 visited: set[str] = set()
235 main_path: list[str] = []
236 main_edge_set: set[tuple[str, str]] = set()
237 current = fsm.initial
238 while current and current not in visited:
239 visited.add(current)
240 main_path.append(current)
241 st = fsm.states.get(current)
242 if not st or st.terminal:
243 break
244 nxt: str = st.on_yes or st.next or ""
245 if not nxt and st.route:
246 nxt = next(iter(st.route.routes.values()), "") or st.route.default or ""
247 if nxt:
248 main_edge_set.add((current, nxt))
249 current = nxt
250 else:
251 break
252 return main_path, main_edge_set
255def _classify_edges(
256 edges: list[tuple[str, str, str]],
257 main_edge_set: set[tuple[str, str]],
258 bfs_order: list[str],
259) -> tuple[list[tuple[str, str, str]], list[tuple[str, str, str]]]:
260 """Split non-main edges into branches and back_edges."""
261 main_consumed: set[int] = set()
262 for src, dst in main_edge_set:
263 for i, (s, d, _) in enumerate(edges):
264 if s == src and d == dst and i not in main_consumed:
265 main_consumed.add(i)
266 break
268 branches: list[tuple[str, str, str]] = []
269 back_edges: list[tuple[str, str, str]] = []
270 for i, (src, dst, label) in enumerate(edges):
271 if i in main_consumed:
272 continue
273 src_pos = bfs_order.index(src) if src in bfs_order else len(bfs_order)
274 dst_pos = bfs_order.index(dst) if dst in bfs_order else len(bfs_order)
275 if dst == src or dst_pos < src_pos:
276 back_edges.append((src, dst, label))
277 else:
278 branches.append((src, dst, label))
279 return branches, back_edges
282class TopologyDetector:
283 """Classify FSM graph topology for layout strategy selection."""
285 def __init__(
286 self,
287 edges: list[tuple[str, str, str]],
288 main_path: list[str],
289 branches: list[tuple[str, str, str]],
290 back_edges: list[tuple[str, str, str]],
291 ) -> None:
292 self._edges = edges
293 self._main_path = main_path
294 self._branches = branches
295 self._back_edges = back_edges
297 def classify(self) -> str:
298 """Return 'linear', 'tree', or 'general'.
300 Linear: main path only, no non-self branches, only self-loop back-edges.
301 Tree: branches exist but no fan-in (every non-initial state has ≤1 incoming).
302 General: everything else (full Sugiyama needed).
303 """
304 non_self_branches = [b for b in self._branches if b[0] != b[1]]
305 non_self_back = [b for b in self._back_edges if b[0] != b[1]]
307 if not non_self_branches and not non_self_back:
308 return "linear"
310 # Check for fan-in: any state with >1 incoming forward edge
311 in_count: dict[str, int] = {}
312 for _, dst, _ in self._edges:
313 # Only count forward edges (not back-edges)
314 in_count[dst] = in_count.get(dst, 0) + 1
316 if not non_self_back and all(v <= 1 for v in in_count.values()):
317 return "tree"
319 return "general"
322# ---------------------------------------------------------------------------
323# Sugiyama layout pipeline
324# ---------------------------------------------------------------------------
327class LayerAssigner:
328 """Assign nodes to layers using longest-path + width constraint."""
330 def __init__(
331 self,
332 all_states: list[str],
333 edges: list[tuple[str, str, str]],
334 back_edge_set: set[tuple[str, str]],
335 initial: str,
336 max_width: int = 4,
337 ) -> None:
338 self._all_states = all_states
339 self._edges = edges
340 self._back_edge_set = back_edge_set
341 self._initial = initial
342 self._max_width = max_width
344 def assign(self) -> list[list[str]]:
345 """Return list of layers, each a list of state names (top to bottom)."""
346 # Build adjacency (forward edges only)
347 forward: dict[str, list[str]] = {s: [] for s in self._all_states}
348 reverse: dict[str, list[str]] = {s: [] for s in self._all_states}
349 seen_edges: set[tuple[str, str]] = set()
350 for src, dst, _ in self._edges:
351 if (src, dst) in self._back_edge_set or src == dst:
352 continue
353 if src in forward and dst in forward and (src, dst) not in seen_edges:
354 forward[src].append(dst)
355 reverse[dst].append(src)
356 seen_edges.add((src, dst))
358 # Longest-path layer assignment (topological order)
359 layer_of: dict[str, int] = {}
361 # Kahn's algorithm for topological order
362 in_degree = {s: len(reverse[s]) for s in self._all_states}
363 queue: deque[str] = deque()
364 for s in self._all_states:
365 if in_degree[s] == 0:
366 queue.append(s)
368 topo_order: list[str] = []
369 while queue:
370 node = queue.popleft()
371 topo_order.append(node)
372 for dst in forward[node]:
373 in_degree[dst] -= 1
374 if in_degree[dst] == 0:
375 queue.append(dst)
377 # Handle nodes not reached by topo sort (cycles in forward graph)
378 for s in self._all_states:
379 if s not in set(topo_order):
380 topo_order.append(s)
382 # Assign layers: longest path from root
383 for node in topo_order:
384 if not reverse[node]:
385 layer_of[node] = 0
386 else:
387 layer_of[node] = max(
388 (layer_of.get(p, 0) + 1 for p in reverse[node]),
389 default=0,
390 )
392 # Ensure initial state is at layer 0
393 if self._initial in layer_of and layer_of[self._initial] != 0:
394 offset = layer_of[self._initial]
395 for s in layer_of:
396 layer_of[s] -= offset
398 # Build layers list
399 max_layer = max(layer_of.values(), default=0)
400 layers: list[list[str]] = [[] for _ in range(max_layer + 1)]
401 for s in self._all_states:
402 layer = layer_of.get(s, 0)
403 layers[layer].append(s)
405 # Width constraint: if any layer exceeds max_width, split
406 if self._max_width > 0:
407 new_layers: list[list[str]] = []
408 for grp in layers:
409 remaining = list(grp)
410 while len(remaining) > self._max_width:
411 new_layers.append(remaining[: self._max_width])
412 remaining = remaining[self._max_width :]
413 if remaining:
414 new_layers.append(remaining)
415 layers = new_layers
417 return layers
420class CrossingMinimizer:
421 """Minimize edge crossings using barycenter heuristic."""
423 def __init__(
424 self,
425 layers: list[list[str]],
426 edges: list[tuple[str, str, str]],
427 back_edge_set: set[tuple[str, str]],
428 ) -> None:
429 self._layers = layers
430 self._edges = edges
431 self._back_edge_set = back_edge_set
433 def minimize(self) -> list[list[str]]:
434 """Return reordered layers with reduced crossings."""
435 # Build position lookup
436 layer_of: dict[str, int] = {}
437 for i, layer in enumerate(self._layers):
438 for s in layer:
439 layer_of[s] = i
441 # Forward adjacency (non-back, non-self)
442 adj_down: dict[str, list[str]] = {}
443 adj_up: dict[str, list[str]] = {}
444 for src, dst, _ in self._edges:
445 if (src, dst) in self._back_edge_set or src == dst:
446 continue
447 if src in layer_of and dst in layer_of:
448 adj_down.setdefault(src, []).append(dst)
449 adj_up.setdefault(dst, []).append(src)
451 layers = [list(layer) for layer in self._layers]
453 # 3 sweeps: down, up, down
454 for sweep in range(3):
455 if sweep % 2 == 0:
456 # Top-down sweep
457 for i in range(1, len(layers)):
458 pos_above = {s: j for j, s in enumerate(layers[i - 1])}
459 bary: dict[str, float] = {}
460 for s in layers[i]:
461 parents = [p for p in adj_up.get(s, []) if p in pos_above]
462 if parents:
463 bary[s] = sum(pos_above[p] for p in parents) / len(parents)
464 else:
465 bary[s] = float(layers[i].index(s))
466 layers[i].sort(key=lambda s: bary.get(s, 0))
467 else:
468 # Bottom-up sweep
469 for i in range(len(layers) - 2, -1, -1):
470 pos_below = {s: j for j, s in enumerate(layers[i + 1])}
471 bary_up: dict[str, float] = {}
472 for s in layers[i]:
473 children = [c for c in adj_down.get(s, []) if c in pos_below]
474 if children:
475 bary_up[s] = sum(pos_below[c] for c in children) / len(children)
476 else:
477 bary_up[s] = float(layers[i].index(s))
478 layers[i].sort(key=lambda s: bary_up.get(s, 0))
480 return layers
483# ---------------------------------------------------------------------------
484# Rendering helpers
485# ---------------------------------------------------------------------------
488def _compute_display_labels(
489 all_states: list[str],
490 initial: str,
491 terminal_states: set[str],
492) -> dict[str, str]:
493 """Compute display labels with → prefix and ◉ suffix."""
494 display_label: dict[str, str] = {}
495 for s in all_states:
496 label = s
497 if s == initial:
498 label = "\u2192 " + label
499 if s in terminal_states:
500 label = label + " \u25c9"
501 display_label[s] = label
502 return display_label
505def _compute_box_sizes(
506 all_states: list[str],
507 display_label: dict[str, str],
508 fsm_states: dict[str, StateConfig] | None,
509 verbose: bool,
510 max_box_inner: int,
511 badges: dict[str, str] | None = None,
512) -> tuple[dict[str, list[str]], dict[str, int], dict[str, int], dict[str, str]]:
513 """Compute box content, widths, and heights for all states.
515 Returns (box_inner, box_width, box_height, box_badge).
516 """
517 box_inner: dict[str, list[str]] = {}
518 box_width: dict[str, int] = {}
519 box_badge: dict[str, str] = {}
521 for s in all_states:
522 state_obj = fsm_states.get(s) if fsm_states else None
524 badge = _get_state_badge(state_obj, badges)
525 badge_w = _badge_display_width(badge) if badge else 0
526 box_badge[s] = badge
528 # Width must fit: name label on content row, badge on top border (with one space
529 # of padding on each side: " badge ")
530 base_w = max(len(display_label[s]), badge_w + 2 if badge_w else 0)
532 inner_w = base_w
533 if state_obj and state_obj.action and max_box_inner > 0:
534 action_lines = state_obj.action.strip().splitlines()
535 if verbose:
536 max_action_w = max(
537 (len(ln.rstrip()) for ln in action_lines if ln.rstrip()), default=0
538 )
539 inner_w = max(base_w, min(max_action_w, max_box_inner))
540 else:
541 first_action = next((ln.rstrip() for ln in action_lines if ln.rstrip()), "")
542 inner_w = max(base_w, min(len(first_action), max_box_inner))
544 content = _box_inner_lines(state_obj, display_label[s], verbose, inner_w)
545 actual_w = max(len(ln) for ln in content)
546 inner_w = max(inner_w, actual_w)
547 box_inner[s] = content
548 box_width[s] = inner_w + 4 # "│ " + content + " │"
550 box_height: dict[str, int] = {s: len(box_inner[s]) + 2 for s in all_states}
551 return box_inner, box_width, box_height, box_badge
554def _draw_box(
555 grid: list[list[str]],
556 row: int,
557 col: int,
558 width: int,
559 height: int,
560 content: list[str],
561 is_highlighted: bool,
562 highlight_color: str,
563 badge: str = "",
564) -> None:
565 """Draw a state box onto a character grid at (row, col).
567 If *badge* is provided it is placed right-aligned in the top border row with
568 one space of padding on each side (``─ badge ┐``), colorized via ``_bc()``.
569 """
570 total_width = len(grid[0]) if grid else 0
572 def _bc(ch: str) -> str:
573 return colorize(ch, highlight_color) if is_highlighted else ch
575 # Top border: ┌ ─ ─ … ─ ┐
576 if col < total_width:
577 grid[row][col] = _bc("\u250c")
578 for j in range(1, width - 1):
579 if col + j < total_width:
580 grid[row][col + j] = _bc("\u2500")
581 if col + width - 1 < total_width:
582 grid[row][col + width - 1] = _bc("\u2510")
584 # Overlay badge in top-right corner with one space of padding on each side: " badge ┐"
585 if badge:
586 badge_w = _badge_display_width(badge)
587 # Trailing space between badge end and ┐
588 trail_pos = col + width - 2
589 if col + 1 <= trail_pos < col + width - 1 and trail_pos < total_width:
590 grid[row][trail_pos] = _bc(" ")
591 # Leading space before badge
592 lead_pos = col + width - badge_w - 3
593 if col + 1 <= lead_pos < col + width - 1 and lead_pos < total_width:
594 grid[row][lead_pos] = _bc(" ")
595 # Badge characters (shifted left by 1 compared to no-padding placement)
596 pos = col + width - 1 - badge_w - 1
597 for ch in badge:
598 ch_w = _wcwidth(ch)
599 if ch_w < 1:
600 ch_w = 1
601 if col + 1 <= pos < col + width - 1 and pos < total_width:
602 grid[row][pos] = _bc(ch)
603 if ch_w == 2 and pos + 1 < col + width - 1 and pos + 1 < total_width:
604 grid[row][pos + 1] = ""
605 pos += ch_w
607 # Content rows
608 for i, line in enumerate(content):
609 r = row + i + 1
610 if r >= len(grid):
611 break
612 if col < total_width:
613 grid[r][col] = _bc("\u2502")
614 if col + width - 1 < total_width:
615 grid[r][col + width - 1] = _bc("\u2502")
616 if is_highlighted and i == 0:
617 colored_line = colorize(line, f"{highlight_color};1")
618 if col + 2 < total_width:
619 grid[r][col + 2] = colored_line
620 for j in range(1, len(line)):
621 if col + 2 + j < col + width - 1:
622 grid[r][col + 2 + j] = ""
623 elif i == 0:
624 bold_line = colorize(line, "1")
625 if col + 2 < total_width:
626 grid[r][col + 2] = bold_line
627 for j in range(1, len(line)):
628 if col + 2 + j < col + width - 1:
629 grid[r][col + 2 + j] = ""
630 else:
631 for j, ch in enumerate(line):
632 if col + 2 + j < col + width - 1:
633 grid[r][col + 2 + j] = ch
635 # Padding rows between content and bottom border
636 for i in range(len(content) + 1, height - 1):
637 r = row + i
638 if r >= len(grid):
639 break
640 if col < total_width:
641 grid[r][col] = _bc("\u2502")
642 if col + width - 1 < total_width:
643 grid[r][col + width - 1] = _bc("\u2502")
645 # Bottom border
646 brow = row + height - 1
647 if brow < len(grid):
648 if col < total_width:
649 grid[brow][col] = _bc("\u2514")
650 for j in range(1, width - 1):
651 if col + j < total_width:
652 grid[brow][col + j] = _bc("\u2500")
653 if col + width - 1 < total_width:
654 grid[brow][col + width - 1] = _bc("\u2518")
657# ---------------------------------------------------------------------------
658# Layered (vertical) renderer
659# ---------------------------------------------------------------------------
662def _render_layered_diagram(
663 layers: list[list[str]],
664 edges: list[tuple[str, str, str]],
665 main_edge_set: set[tuple[str, str]],
666 branches: list[tuple[str, str, str]],
667 back_edges: list[tuple[str, str, str]],
668 initial: str,
669 terminal_states: set[str] | None,
670 fsm_states: dict[str, StateConfig] | None,
671 verbose: bool,
672 highlight_state: str | None,
673 highlight_color: str,
674 edge_label_colors: dict[str, str] | None = None,
675 badges: dict[str, str] | None = None,
676) -> str:
677 """Render FSM using layered (Sugiyama-style) vertical layout."""
678 terminal_states = terminal_states or set()
679 fsm_states = fsm_states or {}
680 tw = terminal_width()
682 # Flatten layers to get all states
683 all_states = [s for layer in layers for s in layer]
684 if not all_states:
685 return ""
687 display_label = _compute_display_labels(all_states, initial, terminal_states)
689 # Compute max_box_inner based on widest layer
690 max_layer_size = max(len(layer) for layer in layers)
691 if verbose:
692 max_box_inner = max(20, min(60, (tw - 4) // max(1, max_layer_size) - 6))
693 else:
694 max_box_inner = max(20, min(40, (tw - 4) // max(1, max_layer_size) - 6))
696 box_inner, box_width, box_height, box_badge = _compute_box_sizes(
697 all_states, display_label, fsm_states, verbose, max_box_inner, badges
698 )
700 # Post-hoc layer merge: re-merge adjacent single-state layers that were
701 # over-split by the conservative max_width_per_layer estimate. Only merge
702 # when both layers receive an edge from the same source state (indicating
703 # they were originally one layer split by width constraint).
704 available_w = tw - 20 # conservative content-area estimate
705 gap_between = 6
706 # Build edge target sets: for each state, which earlier states point to it
707 _incoming: dict[str, set[str]] = {s: set() for layer in layers for s in layer}
708 for src, dst, _ in edges:
709 if src != dst and dst in _incoming:
710 _incoming[dst].add(src)
711 merged = True
712 while merged:
713 merged = False
714 i = 0
715 while i < len(layers) - 1:
716 la, lb = layers[i], layers[i + 1]
717 # Only merge single-state layers that share an incoming source
718 if len(la) == 1 and len(lb) == 1:
719 sources_a = _incoming.get(la[0], set())
720 sources_b = _incoming.get(lb[0], set())
721 shared_source = sources_a & sources_b
722 else:
723 shared_source = set()
724 combined_w = (
725 sum(box_width[s] for s in la)
726 + gap_between * (len(la) - 1)
727 + gap_between
728 + sum(box_width[s] for s in lb)
729 + gap_between * (len(lb) - 1)
730 )
731 if shared_source and combined_w <= available_w and len(la) + len(lb) <= 4:
732 layers[i] = la + lb
733 layers.pop(i + 1)
734 merged = True
735 else:
736 i += 1
738 # Collect ALL non-self-loop forward edge labels (main + branches + same-depth back-edges)
739 # Multiple edges between the same pair are combined as "label1/label2"
740 forward_edge_labels: dict[tuple[str, str], str] = {}
741 for src, dst, lbl in edges:
742 if src == dst:
743 continue
744 if (src, dst) in main_edge_set or (src, dst, lbl) in branches:
745 if (src, dst) in forward_edge_labels:
746 forward_edge_labels[(src, dst)] += "/" + lbl
747 else:
748 forward_edge_labels[(src, dst)] = lbl
750 # True back-edges: only those going to an earlier layer (computed after layer assignment)
751 # Will be finalized below after col positions are computed
752 # Combine same-pair back-edges into single entries with merged labels (e.g. "error/fail")
753 back_edge_labels_initial: dict[tuple[str, str], str] = {}
754 for s, d, lbl in back_edges:
755 if s != d:
756 if (s, d) in back_edge_labels_initial:
757 back_edge_labels_initial[(s, d)] += "/" + lbl
758 else:
759 back_edge_labels_initial[(s, d)] = lbl
761 # Pre-compute layer positions to detect main-path cycle edges early.
762 # This ensures back_edge_margin accounts for ALL backward-pointing edges
763 # (including main-path cycles like commit → initial_state) before column
764 # positions are computed.
765 prelim_layer_of: dict[str, int] = {}
766 for li, layer in enumerate(layers):
767 for s in layer:
768 prelim_layer_of[s] = li
770 # Include main-path/branch edges that point backward in margin estimate
771 all_back_labels: dict[tuple[str, str], str] = dict(back_edge_labels_initial)
772 for (src, dst), lbl in forward_edge_labels.items():
773 src_layer = prelim_layer_of.get(src, -1)
774 dst_layer = prelim_layer_of.get(dst, -1)
775 if dst_layer < src_layer:
776 if (src, dst) in all_back_labels:
777 all_back_labels[(src, dst)] += "/" + lbl
778 else:
779 all_back_labels[(src, dst)] = lbl
781 non_self_back_initial = [(s, d, lbl) for (s, d), lbl in all_back_labels.items()]
782 back_edge_margin = 0
783 if non_self_back_initial:
784 max_label_len = max(len(lbl) for _, _, lbl in non_self_back_initial)
785 n_back_initial = len(non_self_back_initial)
786 back_edge_margin = max_label_len + max(6, 2 * n_back_initial + 2)
788 content_left = 2 + back_edge_margin
790 # Self-loops per state
791 self_loops: dict[str, list[str]] = {}
792 for src, dst, lbl in back_edges:
793 if src == dst:
794 self_loops.setdefault(src, []).append(lbl)
796 # --- Compute a common center column for alignment ---
797 # For layers with single boxes, we want vertical alignment through a
798 # shared center column. Use the widest single-state layer's center.
799 max_single_w = max((box_width[layer[0]] for layer in layers if len(layer) == 1), default=0)
800 # The common center is at content_left + max_single_w // 2
801 common_center = content_left + max_single_w // 2
803 # Compute column positions per layer
804 col_start: dict[str, int] = {}
805 col_center: dict[str, int] = {}
806 layer_of: dict[str, int] = {}
808 for li, layer in enumerate(layers):
809 if len(layer) == 1:
810 # Single-state layer: center-align to common center
811 sname = layer[0]
812 col_start[sname] = common_center - box_width[sname] // 2
813 col_center[sname] = common_center
814 layer_of[sname] = li
815 else:
816 # Multi-state layer: place side-by-side, centered around common_center
817 gap_between = 6
818 total_layer_w = sum(box_width[s] for s in layer) + gap_between * (len(layer) - 1)
819 x = common_center - total_layer_w // 2
820 x = max(content_left, x)
821 for i, sname in enumerate(layer):
822 col_start[sname] = x
823 col_center[sname] = x + box_width[sname] // 2
824 layer_of[sname] = li
825 if i < len(layer) - 1:
826 next_s = layer[i + 1]
827 # Check for edge labels in both directions between adjacent states
828 label_fwd = forward_edge_labels.get((sname, next_s), "")
829 label_rev = forward_edge_labels.get((next_s, sname), "")
830 max_label = max(len(label_fwd), len(label_rev))
831 gap = max(gap_between, max_label + 6) if max_label > 0 else gap_between
832 x += box_width[sname] + gap
833 else:
834 x += box_width[sname]
836 # Reclassify back-edges based on actual layer positions
837 # Only edges going to an earlier layer are true margin back-edges
838 # Combine same-pair edges into merged labels (e.g. "error/fail")
839 back_edge_labels_reclass: dict[tuple[str, str], str] = {}
840 same_layer_edges: list[tuple[str, str, str]] = []
841 for src, dst, lbl in back_edges:
842 if src == dst:
843 continue
844 src_layer = layer_of.get(src, -1)
845 dst_layer = layer_of.get(dst, -1)
846 if dst_layer < src_layer:
847 if (src, dst) in back_edge_labels_reclass:
848 back_edge_labels_reclass[(src, dst)] += "/" + lbl
849 else:
850 back_edge_labels_reclass[(src, dst)] = lbl
851 elif dst_layer == src_layer:
852 same_layer_edges.append((src, dst, lbl))
853 else: # dst_layer > src_layer: actually forward edge
854 if (src, dst) in forward_edge_labels:
855 forward_edge_labels[(src, dst)] += "/" + lbl
856 else:
857 forward_edge_labels[(src, dst)] = lbl
859 # Also reclassify main/branch edges in forward_edge_labels that point backward
860 # after layer assignment (e.g. main-path cycle edges like commit → initial_state)
861 backward_in_fwd: list[tuple[str, str]] = []
862 for (src, dst), lbl in forward_edge_labels.items():
863 src_layer = layer_of.get(src, -1)
864 dst_layer = layer_of.get(dst, -1)
865 if dst_layer < src_layer:
866 backward_in_fwd.append((src, dst))
867 if (src, dst) in back_edge_labels_reclass:
868 back_edge_labels_reclass[(src, dst)] += "/" + lbl
869 else:
870 back_edge_labels_reclass[(src, dst)] = lbl
871 elif dst_layer == src_layer and src != dst:
872 backward_in_fwd.append((src, dst))
873 same_layer_edges.append((src, dst, lbl))
874 for key in backward_in_fwd:
875 del forward_edge_labels[key]
877 # Add same-layer back-edges to forward_edge_labels so gap calculation accounts for them
878 for src, dst, lbl in same_layer_edges:
879 if (src, dst) in forward_edge_labels:
880 forward_edge_labels[(src, dst)] += "/" + lbl
881 else:
882 forward_edge_labels[(src, dst)] = lbl
884 # Recalculate inter-box gaps for layers with newly discovered same-layer edges
885 affected_layers: set[int] = set()
886 for src, dst, _lbl in same_layer_edges:
887 sl = layer_of.get(src, -1)
888 dl = layer_of.get(dst, -1)
889 if sl >= 0:
890 affected_layers.add(sl)
891 if dl >= 0:
892 affected_layers.add(dl)
893 for li in affected_layers:
894 layer = layers[li]
895 if len(layer) < 2:
896 continue
897 gap_between = 6
898 total_layer_w = sum(box_width[s] for s in layer)
899 # For non-adjacent same-layer edges the label lands in the gap immediately
900 # adjacent to the source box (left of src for right-to-left, right of src
901 # for left-to-right). Collect those requirements so the gap is wide enough.
902 extra_gap_req: dict[tuple[str, str], int] = {}
903 for src, dst, lbl in same_layer_edges:
904 if layer_of.get(src) != li or layer_of.get(dst) != li:
905 continue
906 try:
907 si, di = layer.index(src), layer.index(dst)
908 except ValueError:
909 continue
910 if abs(si - di) <= 1:
911 continue # adjacent — already handled by forward_edge_labels
912 if si > di:
913 key = (layer[si - 1], src) # gap to the left of src
914 else:
915 key = (src, layer[si + 1]) # gap to the right of src
916 extra_gap_req[key] = max(extra_gap_req.get(key, 0), len(lbl))
917 # Recalculate gaps with label-aware spacing
918 gaps: list[int] = []
919 for i in range(len(layer) - 1):
920 sname, next_s = layer[i], layer[i + 1]
921 label_fwd = forward_edge_labels.get((sname, next_s), "")
922 label_rev = forward_edge_labels.get((next_s, sname), "")
923 max_label = max(len(label_fwd), len(label_rev), extra_gap_req.get((sname, next_s), 0))
924 gap = max(gap_between, max_label + 6) if max_label > 0 else gap_between
925 gaps.append(gap)
926 total_layer_w += sum(gaps)
927 x = common_center - total_layer_w // 2
928 x = max(content_left, x)
929 for i, sname in enumerate(layer):
930 col_start[sname] = x
931 col_center[sname] = x + box_width[sname] // 2
932 if i < len(layer) - 1:
933 x += box_width[sname] + gaps[i]
934 else:
935 x += box_width[sname]
937 non_self_back = [(s, d, lbl) for (s, d), lbl in back_edge_labels_reclass.items()]
939 # Recalculate back-edge margin if it changed
940 if non_self_back:
941 max_label_len = max(len(lbl) for _, _, lbl in non_self_back)
942 n_back = len(non_self_back)
943 actual_margin = max_label_len + max(6, 2 * n_back + 2)
944 if actual_margin != back_edge_margin:
945 # Need to recalculate positions (rare case - usually matches)
946 back_edge_margin = actual_margin
947 content_left = 2 + back_edge_margin
949 # Identify forward skip-layer edges (span > 1 layer, not handled by consecutive renderer)
950 skip_forward_edges: list[tuple[str, str, str]] = []
951 for (src, dst), lbl in forward_edge_labels.items():
952 src_layer = layer_of.get(src, -1)
953 dst_layer = layer_of.get(dst, -1)
954 if dst_layer > src_layer + 1:
955 skip_forward_edges.append((src, dst, lbl))
957 # Pre-compute right margin width for forward skip-layer edges
958 right_edge_margin = 0
959 if skip_forward_edges:
960 max_fwd_label_len = max(len(lbl) for _, _, lbl in skip_forward_edges)
961 right_edge_margin = max_fwd_label_len + 6
963 # Compute total width needed
964 total_content_w = 0
965 for s in all_states:
966 right = col_start[s] + box_width[s]
967 total_content_w = max(total_content_w, right)
968 total_width = max(total_content_w + right_edge_margin + 4, tw)
970 # Compute vertical positions with space for self-loops and inter-layer arrows
971 row_start: dict[str, int] = {}
972 y = 0
973 for li, layer in enumerate(layers):
974 layer_h = max(box_height[s] for s in layer)
975 for sname in layer:
976 row_start[sname] = y
977 y += layer_h
979 # Add self-loop row if any state in this layer has self-loops
980 has_self_loop = any(s in self_loops for s in layer)
981 if has_self_loop:
982 y += 1 # self-loop marker row
984 if li < len(layers) - 1:
985 y += 3 # arrow gap: │, label, ▼
987 total_height = y
989 # Build character grid
990 grid: list[list[str]] = [[" "] * total_width for _ in range(total_height)]
992 # Draw boxes
993 for sname in all_states:
994 is_highlighted = highlight_state is not None and sname == highlight_state
995 _draw_box(
996 grid,
997 row_start[sname],
998 col_start[sname],
999 box_width[sname],
1000 box_height[sname],
1001 box_inner[sname],
1002 is_highlighted,
1003 highlight_color,
1004 badge=box_badge[sname],
1005 )
1007 # Precompute box-occupied (row, col) pairs so connector lines can avoid overwriting box cells
1008 _box_occ: dict[int, set[int]] = {}
1009 for _s in all_states:
1010 for _r in range(row_start[_s], row_start[_s] + box_height[_s]):
1011 _row_set = _box_occ.setdefault(_r, set())
1012 for _c in range(col_start[_s], col_start[_s] + box_width[_s]):
1013 _row_set.add(_c)
1015 # Draw self-loop markers immediately below their boxes
1016 for sname, labels in self_loops.items():
1017 marker = "\u21ba " + ", ".join(labels)
1018 r = row_start[sname] + box_height[sname]
1019 if r < total_height:
1020 cx = col_center[sname]
1021 pos = max(0, cx - len(marker) // 2)
1022 for j, ch in enumerate(marker):
1023 if pos + j < total_width:
1024 grid[r][pos + j] = ch
1026 # Draw forward edges between layers (vertical arrows with labels)
1027 for li in range(len(layers) - 1):
1028 layer_h = max(box_height[s] for s in layers[li])
1029 has_self_loop = any(s in self_loops for s in layers[li])
1030 self_loop_offset = 1 if has_self_loop else 0
1032 # Arrow area starts after box bottom + self-loop row
1033 arrow_start_row = row_start[layers[li][0]] + layer_h + self_loop_offset
1034 arrow_end_row = row_start[layers[li + 1][0]] - 1
1036 # Collect all inter-layer edges from this layer to the next
1037 inter_edges: list[tuple[str, str, str]] = []
1038 for src in layers[li]:
1039 for dst in layers[li + 1]:
1040 label = forward_edge_labels.get((src, dst))
1041 if label is not None:
1042 inter_edges.append((src, dst, label))
1044 # Draw each edge with its own vertical pipe to the target's center
1045 for src, dst, label in inter_edges:
1046 dst_cc = col_center[dst]
1047 src_left = col_start[src]
1048 src_right = src_left + box_width[src]
1049 ec = _edge_line_color(label) # ANSI code for this edge's connector chars
1051 def _lc(ch: str, _ec: str = ec) -> str: # noqa: E306
1052 return colorize(ch, _ec) if _ec else ch
1054 # Horizontal connector when pipe is outside source box range
1055 if dst_cc >= src_right or dst_cc < src_left:
1056 conn_row = arrow_start_row
1057 if 0 <= conn_row < total_height:
1058 if dst_cc >= src_right:
1059 # Pipe right of source: └───┐
1060 src_cc = col_center[src]
1061 if 0 <= src_cc < total_width and grid[conn_row][src_cc] == " ":
1062 grid[conn_row][src_cc] = _lc("\u2514") # └
1063 start_c = src_cc + 1
1064 else:
1065 start_c = src_right
1066 for c in range(start_c, dst_cc):
1067 if 0 <= c < total_width:
1068 grid[conn_row][c] = _lc("\u2500")
1069 if 0 <= dst_cc < total_width:
1070 grid[conn_row][dst_cc] = _lc("\u2510") # ┐
1071 else:
1072 # Pipe left of source: ┌───┘
1073 src_cc = col_center[src]
1074 if 0 <= src_cc < total_width and grid[conn_row][src_cc] == " ":
1075 end_c = src_cc
1076 grid[conn_row][src_cc] = _lc("\u2518") # ┘
1077 else:
1078 end_c = src_left
1079 for c in range(dst_cc + 1, end_c):
1080 if 0 <= c < total_width:
1081 grid[conn_row][c] = _lc("\u2500")
1082 if 0 <= dst_cc < total_width:
1083 grid[conn_row][dst_cc] = _lc("\u250c") # ┌
1084 pipe_start = arrow_start_row + 1
1085 else:
1086 pipe_start = arrow_start_row
1088 # Draw vertical pipe at destination's center column
1089 for r in range(pipe_start, arrow_end_row):
1090 if 0 <= dst_cc < total_width and r < total_height:
1091 grid[r][dst_cc] = _lc("\u2502")
1093 # Arrow tip at destination center
1094 if arrow_end_row < total_height and 0 <= dst_cc < total_width:
1095 grid[arrow_end_row][dst_cc] = _lc("\u25bc")
1097 # Label to the right of the pipe (or left if it would overlap)
1098 label_row = arrow_start_row
1099 if label_row < total_height:
1100 label_start = dst_cc + 2
1101 for j, ch in enumerate(label):
1102 if label_start + j < total_width:
1103 grid[label_row][label_start + j] = ch
1105 # Post-pass: connect horizontal gaps for multi-branch sources
1106 if len(inter_edges) >= 2 and 0 <= arrow_start_row < total_height:
1107 src_targets: dict[str, list[int]] = {}
1108 for src, dst, _ in inter_edges:
1109 if dst in col_center:
1110 src_targets.setdefault(src, []).append(col_center[dst])
1111 for _src, centers in src_targets.items():
1112 if len(centers) < 2:
1113 continue
1114 left = min(centers)
1115 right = max(centers)
1116 for c in range(left + 1, right):
1117 if 0 <= c < total_width:
1118 cell = grid[arrow_start_row][c]
1119 if cell == " ":
1120 grid[arrow_start_row][c] = "\u2500" # ─
1121 elif cell == "\u2502": # │ → ┼
1122 grid[arrow_start_row][c] = "\u253c"
1123 elif cell == "\u2518": # ┘ → ┴
1124 grid[arrow_start_row][c] = "\u2534"
1125 elif cell == "\u2514": # └ → ┴
1126 grid[arrow_start_row][c] = "\u2534"
1127 elif cell == "\u2510": # ┐ → ┬
1128 grid[arrow_start_row][c] = "\u252c"
1129 elif cell == "\u250c": # ┌ → ┬
1130 grid[arrow_start_row][c] = "\u252c"
1131 # Update boundary junction chars where the horizontal bar meets a pipe
1132 if 0 <= left < total_width and grid[arrow_start_row][left] == "\u2502": # │ → ├
1133 grid[arrow_start_row][left] = "\u251c"
1134 if 0 <= right < total_width and grid[arrow_start_row][right] == "\u2502": # │ → ┤
1135 grid[arrow_start_row][right] = "\u2524"
1137 # Draw same-layer edges (horizontal arrows between states on same layer)
1138 # Includes both branches and reclassified back-edges within same layer
1139 all_same_layer: list[tuple[str, str, str]] = list(same_layer_edges)
1140 for _li, layer in enumerate(layers):
1141 for i, src in enumerate(layer):
1142 for j, dst in enumerate(layer):
1143 if i == j:
1144 continue
1145 label = forward_edge_labels.get((src, dst))
1146 if label is not None and (src, dst, label) not in all_same_layer:
1147 all_same_layer.append((src, dst, label))
1149 for src, dst, label in all_same_layer:
1150 if src not in col_start or dst not in col_start:
1151 continue
1152 name_row = row_start[src] + 1
1153 src_right = col_start[src] + box_width[src]
1154 dst_right = col_start[dst] + box_width[dst]
1155 dst_left = col_start[dst]
1156 src_left = col_start[src]
1157 _row_boxes = _box_occ.get(name_row, set())
1158 ec = _edge_line_color(label)
1160 def _lc(ch: str, _ec: str = ec) -> str: # noqa: E306
1161 return colorize(ch, _ec) if _ec else ch
1163 if dst_left >= src_right:
1164 # Left to right horizontal arrow: src ──label──▶ dst
1165 start = src_right
1166 end = dst_left
1167 edge_text = "\u2500" + label + "\u2500\u2500\u25b6"
1168 available = end - start
1169 if available < len(edge_text):
1170 edge_text = edge_text[: max(1, available)]
1171 left_dashes = max(0, available - len(edge_text))
1172 for k in range(left_dashes):
1173 pos = start + k
1174 if pos < total_width and name_row < total_height and pos not in _row_boxes:
1175 grid[name_row][pos] = _lc("\u2500")
1176 for k, ch in enumerate(edge_text):
1177 pos = start + left_dashes + k
1178 if (
1179 0 <= pos < end
1180 and pos < total_width
1181 and name_row < total_height
1182 and pos not in _row_boxes
1183 ):
1184 grid[name_row][pos] = _lc(ch)
1185 elif dst_right <= src_left:
1186 # Right to left: dst is left of src: src → dst drawn as dst ◀──label── src
1187 start = dst_right
1188 end = src_left
1189 edge_text = "\u2500" + label + "\u2500\u2500\u25b6"
1190 available = end - start
1191 if available < len(edge_text):
1192 edge_text = edge_text[: max(1, available)]
1193 left_dashes = max(0, available - len(edge_text))
1194 for k in range(left_dashes):
1195 pos = start + k
1196 if pos < total_width and name_row < total_height and pos not in _row_boxes:
1197 grid[name_row][pos] = _lc("\u2500")
1198 for k, ch in enumerate(edge_text):
1199 pos = start + left_dashes + k
1200 if (
1201 0 <= pos < end
1202 and pos < total_width
1203 and name_row < total_height
1204 and pos not in _row_boxes
1205 ):
1206 grid[name_row][pos] = _lc(ch)
1208 # Back-edges: left-margin vertical arrows with labels
1209 if non_self_back:
1210 sorted_back = sorted(
1211 non_self_back,
1212 key=lambda e: abs(row_start.get(e[0], 0) - row_start.get(e[1], 0)),
1213 reverse=True,
1214 )
1215 used_cols: list[int] = []
1216 # Compute rightmost pipe column so labels go right of ALL pipes
1217 rightmost_pipe_col = 1 + (len(sorted_back) - 1) * 2
1219 for src, dst, label in sorted_back:
1220 # Source: name row of source box; target: name row of target box
1221 src_row = row_start.get(src, 0) + 1 # name row, not bottom border
1222 dst_row = row_start.get(dst, 0) + 1 # name row
1224 # Find a free column in the margin
1225 col = 1
1226 for uc in sorted(used_cols):
1227 if uc == col:
1228 col += 2
1229 used_cols.append(col)
1231 if col >= content_left - 1:
1232 continue
1234 top_row = min(src_row, dst_row)
1235 bot_row = max(src_row, dst_row)
1236 ec = _edge_line_color(label)
1238 def _lc(ch: str, _ec: str = ec) -> str: # noqa: E306
1239 return colorize(ch, _ec) if _ec else ch
1241 # Draw vertical line in margin (exclude corner rows handled below)
1242 for r in range(top_row + 1, bot_row):
1243 if 0 <= r < total_height and col < total_width:
1244 cell = grid[r][col]
1245 if cell == "\u2500": # ─ → ┼ (junction, leave uncolored)
1246 grid[r][col] = "\u253c"
1247 elif cell == " ":
1248 grid[r][col] = _lc("\u2502")
1250 # Horizontal connector from source box to margin
1251 # Draw right-to-left, crossing existing pipes with junction chars
1252 if 0 <= src_row < total_height:
1253 src_left = col_start.get(src, col + 1)
1254 _src_row_boxes = _box_occ.get(src_row, set())
1255 for c in range(col + 1, src_left):
1256 if c < total_width and c not in _src_row_boxes:
1257 cell = grid[src_row][c]
1258 if cell == " ":
1259 grid[src_row][c] = _lc("\u2500") # ─
1260 elif cell == "\u2502": # │ → ┼ (junction)
1261 grid[src_row][c] = "\u253c"
1262 elif cell == "\u2514": # └ → ┴ (junction)
1263 grid[src_row][c] = "\u2534"
1264 elif cell == "\u250c": # ┌ → ┬ (junction)
1265 grid[src_row][c] = "\u252c"
1266 elif cell == "\u251c": # ├ → ┼ (junction)
1267 grid[src_row][c] = "\u253c"
1268 # Leave ─, ▶, box chars unchanged
1270 # Horizontal connector from margin to target box
1271 # Draw right-to-left, crossing existing pipes with junction chars
1272 dst_left = col_start.get(dst, col + 1)
1273 if 0 <= dst_row < total_height:
1274 _dst_row_boxes = _box_occ.get(dst_row, set())
1275 for c in range(col + 1, dst_left):
1276 if c < total_width and c not in _dst_row_boxes:
1277 cell = grid[dst_row][c]
1278 if cell == " ":
1279 grid[dst_row][c] = _lc("\u2500") # ─
1280 elif cell == "\u2502": # │ → ┼ (junction)
1281 grid[dst_row][c] = "\u253c"
1282 elif cell == "\u2514": # └ → ┴ (junction)
1283 grid[dst_row][c] = "\u2534"
1284 elif cell == "\u250c": # ┌ → ┬ (junction)
1285 grid[dst_row][c] = "\u252c"
1286 elif cell == "\u251c": # ├ → ┼ (junction)
1287 grid[dst_row][c] = "\u253c"
1289 # Corner characters at pipe-to-horizontal turn points
1290 for row in (src_row, dst_row):
1291 if 0 <= row < total_height and col < total_width:
1292 existing = grid[row][col]
1293 if row == bot_row:
1294 # Pipe ends, turns right: └; if horizontal already crosses here: ┴
1295 grid[row][col] = "\u2534" if existing == "\u2500" else _lc("\u2514")
1296 else: # row == top_row
1297 # Pipe starts going down, turns right: ┌; if horizontal already crosses here: ┬
1298 grid[row][col] = "\u252c" if existing == "\u2500" else _lc("\u250c")
1300 # Arrow tip at target: place ▶ at end of horizontal connector (entering box from left)
1301 if 0 <= dst_row < total_height and dst_left - 1 > col and dst_left - 1 < total_width:
1302 grid[dst_row][dst_left - 1] = _lc("\u25b6")
1304 # Label on the margin line (right of ALL pipes, not just this one)
1305 label_row_pos = (top_row + bot_row) // 2
1306 if 0 <= label_row_pos < total_height:
1307 label_start = rightmost_pipe_col + 2
1308 for j, ch in enumerate(label):
1309 if label_start + j < content_left - 1 and label_start + j < total_width:
1310 grid[label_row_pos][label_start + j] = _lc(ch)
1312 # Forward skip-layer edges: right-margin vertical arrows with labels
1313 # Symmetric to the left-margin back-edge renderer above
1314 if skip_forward_edges:
1315 sorted_fwd_skip = sorted(
1316 skip_forward_edges,
1317 key=lambda e: abs(row_start.get(e[0], 0) - row_start.get(e[1], 0)),
1318 reverse=True,
1319 )
1320 used_fwd_cols: list[int] = []
1321 # Rightmost pipe column (furthest from content) for label placement
1322 rightmost_fwd_pipe_col = total_content_w + 2 + (len(sorted_fwd_skip) - 1) * 2
1324 for src, dst, label in sorted_fwd_skip:
1325 src_row = row_start.get(src, 0) + 1 # name row
1326 dst_row = row_start.get(dst, 0) + 1 # name row
1328 # Allocate column in right margin (starting from content edge, going right)
1329 col = total_content_w + 2
1330 for uc in sorted(used_fwd_cols):
1331 if uc == col:
1332 col += 2
1333 used_fwd_cols.append(col)
1335 if col >= total_width:
1336 continue
1338 top_row = min(src_row, dst_row)
1339 bot_row = max(src_row, dst_row)
1340 ec = _edge_line_color(label)
1342 def _lc(ch: str, _ec: str = ec) -> str: # noqa: E306
1343 return colorize(ch, _ec) if _ec else ch
1345 # Draw vertical line in right margin (exclude corner rows handled below)
1346 for r in range(top_row + 1, bot_row):
1347 if 0 <= r < total_height and col < total_width:
1348 cell = grid[r][col]
1349 if cell == "\u2500": # ─ → ┼ (junction)
1350 grid[r][col] = "\u253c"
1351 elif cell == " ":
1352 grid[r][col] = _lc("\u2502")
1354 # Horizontal connector from source box right side to margin
1355 # Draw left-to-right, crossing existing pipes with junction chars
1356 src_right = col_start.get(src, 0) + box_width.get(src, 0)
1357 _src_row_boxes = _box_occ.get(src_row, set())
1358 if 0 <= src_row < total_height:
1359 for c in range(src_right, col):
1360 if 0 <= c < total_width and c not in _src_row_boxes:
1361 cell = grid[src_row][c]
1362 if cell == " ":
1363 grid[src_row][c] = _lc("\u2500") # ─
1364 elif cell == "\u2502": # │ → ┼ (junction)
1365 grid[src_row][c] = "\u253c"
1366 elif cell == "\u2518": # ┘ → ┴ (junction)
1367 grid[src_row][c] = "\u2534"
1368 elif cell == "\u2510": # ┐ → ┬ (junction)
1369 grid[src_row][c] = "\u252c"
1370 elif cell == "\u2524": # ┤ → ┼ (junction)
1371 grid[src_row][c] = "\u253c"
1372 # Leave ─, ◀, box chars unchanged
1374 # Horizontal connector from margin to destination box right side
1375 dst_right = col_start.get(dst, 0) + box_width.get(dst, 0)
1376 _dst_row_boxes = _box_occ.get(dst_row, set())
1377 if 0 <= dst_row < total_height:
1378 for c in range(dst_right, col):
1379 if 0 <= c < total_width and c not in _dst_row_boxes:
1380 cell = grid[dst_row][c]
1381 if cell == " ":
1382 grid[dst_row][c] = _lc("\u2500") # ─
1383 elif cell == "\u2502": # │ → ┼ (junction)
1384 grid[dst_row][c] = "\u253c"
1385 elif cell == "\u2518": # ┘ → ┴ (junction)
1386 grid[dst_row][c] = "\u2534"
1387 elif cell == "\u2510": # ┐ → ┬ (junction)
1388 grid[dst_row][c] = "\u252c"
1389 elif cell == "\u2524": # ┤ → ┼ (junction)
1390 grid[dst_row][c] = "\u253c"
1392 # Corner characters at pipe-to-horizontal turn points
1393 for row in (src_row, dst_row):
1394 if 0 <= row < total_height and col < total_width:
1395 existing = grid[row][col]
1396 if row == bot_row:
1397 # Pipe ends, turns left: ┘; if horizontal crosses: ┤
1398 grid[row][col] = "\u2524" if existing == "\u2500" else _lc("\u2518")
1399 else: # row == top_row
1400 # Pipe starts going down, turns left: ┐; if horizontal crosses: ┤
1401 grid[row][col] = "\u2524" if existing == "\u2500" else _lc("\u2510")
1403 # Arrow tip at target: ◀ entering box from right side
1404 if 0 <= dst_row < total_height and dst_right < col and dst_right < total_width:
1405 grid[dst_row][dst_right] = _lc("\u25c0")
1407 # Label on the margin line (right of ALL pipes, mirroring left-margin approach)
1408 label_row_pos = (top_row + bot_row) // 2
1409 if 0 <= label_row_pos < total_height:
1410 label_start = rightmost_fwd_pipe_col + 2
1411 for j, ch in enumerate(label):
1412 if label_start + j < total_width:
1413 grid[label_row_pos][label_start + j] = _lc(ch)
1415 # Convert grid to string
1416 lines = ["".join(row).rstrip() for row in grid]
1418 # Remove trailing empty lines
1419 while lines and not lines[-1].strip():
1420 lines.pop()
1422 # Center diagram
1423 max_line_len = max((len(_ANSI_ESCAPE_RE.sub("", ln)) for ln in lines), default=0)
1424 diagram_indent = max(0, (tw - max_line_len) // 2)
1425 if diagram_indent > 0:
1426 lines = [" " * diagram_indent + ln if ln.strip() else ln for ln in lines]
1428 return _colorize_diagram_labels("\n".join(lines), edge_label_colors)
1431# ---------------------------------------------------------------------------
1432# FSM diagram renderer (main entry point)
1433# ---------------------------------------------------------------------------
1436def _render_fsm_diagram(
1437 fsm: FSMLoop,
1438 verbose: bool = False,
1439 highlight_state: str | None = None,
1440 highlight_color: str = "32",
1441 edge_label_colors: dict[str, str] | None = None,
1442 badges: dict[str, str] | None = None,
1443) -> str:
1444 """Render an adaptive text diagram of the FSM graph.
1446 Detects FSM topology and selects appropriate layout:
1447 - Linear chains: vertical top-to-bottom
1448 - Branching/cyclic: layered Sugiyama-style
1450 Args:
1451 fsm: The FSM loop to render.
1452 verbose: If True, show expanded action content in boxes.
1453 highlight_state: If provided, render this state's box with the highlight color.
1454 highlight_color: ANSI SGR code for the highlighted state (default: green).
1455 edge_label_colors: Optional label→SGR-code mapping for transition labels.
1456 Falls back to hardcoded defaults when None.
1457 badges: Optional glyph-key→string mapping for state type badges.
1458 Falls back to hardcoded defaults when None.
1459 """
1460 edges = _collect_edges(fsm)
1461 bfs_order_list, bfs_depth = _bfs_order(fsm.initial, edges)
1462 main_path, main_edge_set = _trace_main_path(fsm, edges)
1463 branches, back_edges = _classify_edges(edges, main_edge_set, bfs_order_list)
1465 terminal_states = {name for name, state in fsm.states.items() if state.terminal}
1467 # Collect all states
1468 all_states = list(main_path)
1469 for src, dst, _ in branches:
1470 for s in (src, dst):
1471 if s not in all_states:
1472 all_states.append(s)
1474 # Topology detection
1475 detector = TopologyDetector(edges, main_path, branches, back_edges)
1476 topology = detector.classify()
1478 # Build back-edge set for layout pipeline
1479 back_edge_set: set[tuple[str, str]] = set()
1480 for src, dst, _ in back_edges:
1481 if src != dst:
1482 back_edge_set.add((src, dst))
1484 tw = terminal_width()
1486 if topology == "linear" and len(all_states) <= 1:
1487 # Single state or empty — use simple horizontal
1488 return _render_horizontal_simple(
1489 main_path,
1490 edges,
1491 main_edge_set,
1492 branches,
1493 back_edges,
1494 bfs_order_list,
1495 fsm.initial,
1496 terminal_states,
1497 fsm.states,
1498 verbose,
1499 highlight_state,
1500 highlight_color,
1501 edge_label_colors,
1502 badges,
1503 )
1505 # Compute max node width to determine width constraint
1506 # Quick estimate: widest state name or badge + padding
1507 max_node_w = 30 # reasonable default
1508 for s in all_states:
1509 st = fsm.states.get(s)
1510 badge = _get_state_badge(st, badges)
1511 badge_w = _badge_display_width(badge) if badge else 0
1512 label = s
1513 if s == fsm.initial:
1514 label = "\u2192 " + label
1515 if s in terminal_states:
1516 label = label + " \u25c9"
1517 w = max(len(label), badge_w)
1518 max_node_w = max(max_node_w, w + 4 + 4) # inner + borders + padding
1520 max_width_per_layer = max(1, (tw - 10) // (max_node_w + 4))
1522 # Layer assignment
1523 assigner = LayerAssigner(all_states, edges, back_edge_set, fsm.initial, max_width_per_layer)
1524 layers = assigner.assign()
1526 # Crossing minimization
1527 minimizer = CrossingMinimizer(layers, edges, back_edge_set)
1528 layers = minimizer.minimize()
1530 return _render_layered_diagram(
1531 layers,
1532 edges,
1533 main_edge_set,
1534 branches,
1535 back_edges,
1536 fsm.initial,
1537 terminal_states,
1538 fsm.states,
1539 verbose,
1540 highlight_state,
1541 highlight_color,
1542 edge_label_colors,
1543 badges,
1544 )
1547def _render_horizontal_simple(
1548 main_path: list[str],
1549 edges: list[tuple[str, str, str]],
1550 main_edge_set: set[tuple[str, str]],
1551 branches: list[tuple[str, str, str]],
1552 back_edges: list[tuple[str, str, str]],
1553 bfs_order: list[str],
1554 initial: str,
1555 terminal_states: set[str],
1556 fsm_states: dict[str, StateConfig],
1557 verbose: bool,
1558 highlight_state: str | None,
1559 highlight_color: str,
1560 edge_label_colors: dict[str, str] | None = None,
1561 badges: dict[str, str] | None = None,
1562) -> str:
1563 """Simple horizontal rendering for single-state or very simple FSMs."""
1564 if not main_path:
1565 return ""
1567 all_states = list(main_path)
1568 display_label = _compute_display_labels(all_states, initial, terminal_states)
1570 tw = terminal_width()
1571 num_main = max(1, len(main_path))
1572 if verbose and fsm_states and main_path:
1573 max_box_inner = max(20, min(60, (tw - 4) // num_main - 6))
1574 else:
1575 max_box_inner = max(20, min(40, (tw - 4) // num_main - 6))
1577 box_inner, box_width, box_height, box_badge = _compute_box_sizes(
1578 all_states, display_label, fsm_states, verbose, max_box_inner, badges
1579 )
1581 main_height = max((box_height[s] for s in main_path), default=3)
1582 total_width = tw
1584 # Column positions
1585 col_start: dict[str, int] = {}
1586 col_center: dict[str, int] = {}
1587 x = 2
1588 for i, sname in enumerate(main_path):
1589 col_start[sname] = x
1590 col_center[sname] = x + box_width[sname] // 2
1591 x += box_width[sname]
1592 if i < len(main_path) - 1:
1593 x += 4
1595 rows: list[list[str]] = [[" "] * total_width for _ in range(main_height)]
1597 for sname in main_path:
1598 is_highlighted = highlight_state is not None and sname == highlight_state
1599 _draw_box(
1600 rows,
1601 0,
1602 col_start[sname],
1603 box_width[sname],
1604 main_height,
1605 box_inner[sname],
1606 is_highlighted,
1607 highlight_color,
1608 badge=box_badge[sname],
1609 )
1611 # Self-loops
1612 self_loops_list = [(s, d, lbl) for s, d, lbl in back_edges if s == d]
1613 lines = ["".join(row).rstrip() for row in rows]
1614 if self_loops_list:
1615 self_labels: dict[str, list[str]] = {}
1616 for src, _, label in self_loops_list:
1617 self_labels.setdefault(src, []).append(label)
1618 for sname, labels in self_labels.items():
1619 marker = "\u21ba " + ", ".join(labels)
1620 self_row = [" "] * total_width
1621 cx = col_center.get(sname, 0)
1622 pos = max(0, cx - len(marker) // 2)
1623 for j, ch in enumerate(marker):
1624 if pos + j < total_width:
1625 self_row[pos + j] = ch
1626 lines.append("".join(self_row).rstrip())
1628 diagram_indent = max(0, (tw - (x + 4)) // 2)
1629 if diagram_indent > 0:
1630 lines = [" " * diagram_indent + ln if ln.strip() else ln for ln in lines]
1632 return _colorize_diagram_labels("\n".join(lines), edge_label_colors)