Coverage for little_loops / cli / loop / layout.py: 0%
1156 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-05-28 13:07 -0500
« prev ^ index » next coverage.py v7.12.0, created at 2026-05-28 13:07 -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 "rate_limit_exhausted": "38;5;214",
37 "throttle_hard": "38;5;196",
38}
41def _colorize_label(label: str) -> str:
42 """Colorize a (possibly compound) edge label like 'no/error'."""
43 parts = label.split("/")
44 code = ""
45 for part in parts:
46 if part in ("no", "error"):
47 code = _EDGE_LABEL_COLORS["no"]
48 break
49 if part == "partial" and not code:
50 code = _EDGE_LABEL_COLORS["partial"]
51 elif part == "yes" and not code:
52 code = _EDGE_LABEL_COLORS["yes"]
53 elif part in ("next", "_") and not code:
54 code = _EDGE_LABEL_COLORS["next"]
55 return colorize(label, code) if code else label
58def _edge_line_color(label: str) -> str:
59 """Return the ANSI SGR code to use for connector characters of an edge.
61 Applies the same priority cascade as ``_colorize_label`` so that line
62 characters (│, ─, ▼, ▶, corners) match the semantic color of their label.
63 Returns an empty string when no color applies (callers treat this as "no-op").
64 """
65 parts = label.split("/")
66 code = ""
67 for part in parts:
68 if part in (
69 "no",
70 "error",
71 "blocked",
72 "retry_exhausted",
73 "rate_limit_exhausted",
74 "throttle_hard",
75 ):
76 return _EDGE_LABEL_COLORS.get(part, "31")
77 if part == "partial" and not code:
78 code = _EDGE_LABEL_COLORS["partial"]
79 elif part == "yes" and not code:
80 code = _EDGE_LABEL_COLORS["yes"]
81 elif part in ("next", "_") and not code:
82 code = _EDGE_LABEL_COLORS["next"]
83 return code
86def _colorize_diagram_labels(diagram: str, colors: dict[str, str] | None = None) -> str:
87 """Apply ANSI color to known edge labels in a rendered diagram string.
89 Labels are colorized only when bounded by box-drawing or whitespace chars
90 to avoid coloring partial matches inside state names.
92 Args:
93 colors: Optional label→SGR-code mapping; falls back to _EDGE_LABEL_COLORS if None.
94 """
95 label_colors = colors if colors is not None else _EDGE_LABEL_COLORS
96 for label, code in label_colors.items():
97 colored = colorize(label, code)
98 diagram = re.sub(
99 r"(?<=[─ │▶\n])" + re.escape(label) + r"(?=[─ │▶\n])",
100 colored,
101 diagram,
102 )
103 return diagram
106# ---------------------------------------------------------------------------
107# State box badge definitions
108# ---------------------------------------------------------------------------
110_ACTION_TYPE_BADGES: dict[str, str] = {
111 "prompt": "\u2726", # ✦
112 "slash_command": "/\u2501\u25ba", # /━►
113 "shell": "\u276f_", # ❯_
114 "mcp_tool": "\u26a1", # ⚡
115}
117_SUB_LOOP_BADGE = "\u21b3\u27f3" # ↳⟳
118_ROUTE_BADGE = "\u2443" # ⑃
121def _badge_display_width(badge: str) -> int:
122 """Compute terminal display width of a badge string using wcwidth."""
123 w = _wcswidth(badge)
124 return w if w >= 0 else len(badge)
127def _get_state_badge(state: StateConfig | None, badges: dict[str, str] | None = None) -> str:
128 """Return the unicode badge string for a state, or '' if none."""
129 if state is None:
130 return ""
131 effective = {**_ACTION_TYPE_BADGES, **(badges or {})}
132 sub_loop_badge = (badges or {}).get("sub_loop", _SUB_LOOP_BADGE)
133 route_badge = (badges or {}).get("route", _ROUTE_BADGE)
134 if state.loop is not None:
135 return sub_loop_badge
136 if state.action_type:
137 return effective.get(state.action_type, f"[{state.action_type}]")
138 if state.action:
139 return effective["shell"]
140 if state.route is not None:
141 return route_badge
142 return ""
145# ---------------------------------------------------------------------------
146# Box content helpers for multi-row diagram boxes
147# ---------------------------------------------------------------------------
150def _box_inner_lines(
151 state: StateConfig | None,
152 display_label: str,
153 verbose: bool,
154 inner_width: int,
155 title_only: bool = False,
156) -> list[str]:
157 """Return interior lines for a state box (between top and bottom borders).
159 The first line is always ``display_label`` + type badge (if any).
160 Subsequent lines are action content lines. All lines fit within
161 ``inner_width`` characters (content is truncated or wrapped accordingly).
163 When ``title_only`` is True, only the name row is returned (used by
164 ``--show-diagrams=mini`` for skeleton rendering).
165 """
166 # Badge is now rendered in the top-right corner by _draw_box; name row is label only
167 name_line = display_label[:inner_width]
169 lines: list[str] = [name_line]
171 if title_only:
172 return lines
174 # Action lines
175 if state and state.action:
176 action_src = [ln.rstrip() for ln in state.action.strip().splitlines()]
177 if verbose:
178 for src in action_src:
179 if not src:
180 continue
181 # Wrap long lines to inner_width
182 while len(src) > inner_width:
183 lines.append(src[:inner_width])
184 src = src[inner_width:]
185 if src:
186 lines.append(src)
187 else:
188 # Show first non-empty line, truncated
189 first = next((ln for ln in action_src if ln), "")
190 if len(first) > inner_width:
191 first = first[: inner_width - 1] + "\u2026"
192 if first:
193 lines.append(first)
195 return lines
198# ---------------------------------------------------------------------------
199# Topology detection
200# ---------------------------------------------------------------------------
203def _collect_edges(fsm: FSMLoop) -> list[tuple[str, str, str]]:
204 """Collect all (source, target, label) edges from an FSM."""
205 edges: list[tuple[str, str, str]] = []
206 for name, state in fsm.states.items():
207 if state.on_yes:
208 edges.append((name, state.on_yes, "yes"))
209 if state.on_no:
210 edges.append((name, state.on_no, "no"))
211 if state.on_error:
212 edges.append((name, state.on_error, "error"))
213 if state.on_partial:
214 edges.append((name, state.on_partial, "partial"))
215 if state.on_blocked:
216 edges.append((name, state.on_blocked, "blocked"))
217 if state.on_retry_exhausted:
218 edges.append((name, state.on_retry_exhausted, "retry_exhausted"))
219 if state.on_rate_limit_exhausted:
220 edges.append((name, state.on_rate_limit_exhausted, "rate_limit_exhausted"))
221 if state.on_throttle_hard:
222 edges.append((name, state.on_throttle_hard, "throttle_hard"))
223 if state.next:
224 edges.append((name, state.next, "next"))
225 if state.route:
226 for verdict, target in state.route.routes.items():
227 edges.append((name, target, verdict))
228 if state.route.default:
229 edges.append((name, state.route.default, "_"))
230 for verdict, target in state.extra_routes.items():
231 edges.append((name, target, verdict))
232 return edges
235def _bfs_order(initial: str, edges: list[tuple[str, str, str]]) -> tuple[list[str], dict[str, int]]:
236 """BFS from initial state. Returns (order, depth_map)."""
237 order: list[str] = []
238 depth: dict[str, int] = {initial: 0}
239 queue: deque[str] = deque([initial])
240 while queue:
241 node = queue.popleft()
242 order.append(node)
243 for src, dst, _ in edges:
244 if src == node and dst not in depth:
245 depth[dst] = depth[node] + 1
246 queue.append(dst)
247 return order, depth
250def _trace_main_path(
251 fsm: FSMLoop, edges: list[tuple[str, str, str]]
252) -> tuple[list[str], set[tuple[str, str]]]:
253 """Trace the main (happy) path through the FSM."""
254 visited: set[str] = set()
255 main_path: list[str] = []
256 main_edge_set: set[tuple[str, str]] = set()
257 current = fsm.initial
258 while current and current not in visited:
259 visited.add(current)
260 main_path.append(current)
261 st = fsm.states.get(current)
262 if not st or st.terminal:
263 break
264 nxt: str = st.on_yes or st.next or ""
265 if not nxt and st.route:
266 nxt = next(iter(st.route.routes.values()), "") or st.route.default or ""
267 if nxt:
268 main_edge_set.add((current, nxt))
269 current = nxt
270 else:
271 break
272 return main_path, main_edge_set
275def _classify_edges(
276 edges: list[tuple[str, str, str]],
277 main_edge_set: set[tuple[str, str]],
278 bfs_order: list[str],
279) -> tuple[list[tuple[str, str, str]], list[tuple[str, str, str]]]:
280 """Split non-main edges into branches and back_edges."""
281 main_consumed: set[int] = set()
282 for src, dst in main_edge_set:
283 for i, (s, d, _) in enumerate(edges):
284 if s == src and d == dst and i not in main_consumed:
285 main_consumed.add(i)
286 break
288 branches: list[tuple[str, str, str]] = []
289 back_edges: list[tuple[str, str, str]] = []
290 for i, (src, dst, label) in enumerate(edges):
291 if i in main_consumed:
292 continue
293 src_pos = bfs_order.index(src) if src in bfs_order else len(bfs_order)
294 dst_pos = bfs_order.index(dst) if dst in bfs_order else len(bfs_order)
295 if dst == src or dst_pos < src_pos:
296 back_edges.append((src, dst, label))
297 else:
298 branches.append((src, dst, label))
299 return branches, back_edges
302class TopologyDetector:
303 """Classify FSM graph topology for layout strategy selection."""
305 def __init__(
306 self,
307 edges: list[tuple[str, str, str]],
308 main_path: list[str],
309 branches: list[tuple[str, str, str]],
310 back_edges: list[tuple[str, str, str]],
311 ) -> None:
312 self._edges = edges
313 self._main_path = main_path
314 self._branches = branches
315 self._back_edges = back_edges
317 def classify(self) -> str:
318 """Return 'linear', 'tree', or 'general'.
320 Linear: main path only, no non-self branches, only self-loop back-edges.
321 Tree: branches exist but no fan-in (every non-initial state has ≤1 incoming).
322 General: everything else (full Sugiyama needed).
323 """
324 non_self_branches = [b for b in self._branches if b[0] != b[1]]
325 non_self_back = [b for b in self._back_edges if b[0] != b[1]]
327 if not non_self_branches and not non_self_back:
328 return "linear"
330 # Check for fan-in: any state with >1 incoming forward edge
331 in_count: dict[str, int] = {}
332 for _, dst, _ in self._edges:
333 # Only count forward edges (not back-edges)
334 in_count[dst] = in_count.get(dst, 0) + 1
336 if not non_self_back and all(v <= 1 for v in in_count.values()):
337 return "tree"
339 return "general"
342# ---------------------------------------------------------------------------
343# Sugiyama layout pipeline
344# ---------------------------------------------------------------------------
347class LayerAssigner:
348 """Assign nodes to layers using longest-path + width constraint."""
350 def __init__(
351 self,
352 all_states: list[str],
353 edges: list[tuple[str, str, str]],
354 back_edge_set: set[tuple[str, str]],
355 initial: str,
356 max_width: int = 4,
357 ) -> None:
358 self._all_states = all_states
359 self._edges = edges
360 self._back_edge_set = back_edge_set
361 self._initial = initial
362 self._max_width = max_width
364 def assign(self) -> list[list[str]]:
365 """Return list of layers, each a list of state names (top to bottom)."""
366 # Build adjacency (forward edges only)
367 forward: dict[str, list[str]] = {s: [] for s in self._all_states}
368 reverse: dict[str, list[str]] = {s: [] for s in self._all_states}
369 seen_edges: set[tuple[str, str]] = set()
370 for src, dst, _ in self._edges:
371 if (src, dst) in self._back_edge_set or src == dst:
372 continue
373 if src in forward and dst in forward and (src, dst) not in seen_edges:
374 forward[src].append(dst)
375 reverse[dst].append(src)
376 seen_edges.add((src, dst))
378 # Longest-path layer assignment (topological order)
379 layer_of: dict[str, int] = {}
381 # Kahn's algorithm for topological order
382 in_degree = {s: len(reverse[s]) for s in self._all_states}
383 queue: deque[str] = deque()
384 for s in self._all_states:
385 if in_degree[s] == 0:
386 queue.append(s)
388 topo_order: list[str] = []
389 while queue:
390 node = queue.popleft()
391 topo_order.append(node)
392 for dst in forward[node]:
393 in_degree[dst] -= 1
394 if in_degree[dst] == 0:
395 queue.append(dst)
397 # Handle nodes not reached by topo sort (cycles in forward graph)
398 for s in self._all_states:
399 if s not in set(topo_order):
400 topo_order.append(s)
402 # Assign layers: longest path from root
403 for node in topo_order:
404 if not reverse[node]:
405 layer_of[node] = 0
406 else:
407 layer_of[node] = max(
408 (layer_of.get(p, 0) + 1 for p in reverse[node]),
409 default=0,
410 )
412 # Ensure initial state is at layer 0
413 if self._initial in layer_of and layer_of[self._initial] != 0:
414 offset = layer_of[self._initial]
415 for s in layer_of:
416 layer_of[s] -= offset
418 # Build layers list
419 max_layer = max(layer_of.values(), default=0)
420 layers: list[list[str]] = [[] for _ in range(max_layer + 1)]
421 for s in self._all_states:
422 layer = layer_of.get(s, 0)
423 layers[layer].append(s)
425 # Width constraint: if any layer exceeds max_width, split
426 if self._max_width > 0:
427 new_layers: list[list[str]] = []
428 for grp in layers:
429 remaining = list(grp)
430 while len(remaining) > self._max_width:
431 new_layers.append(remaining[: self._max_width])
432 remaining = remaining[self._max_width :]
433 if remaining:
434 new_layers.append(remaining)
435 layers = new_layers
437 return layers
440class CrossingMinimizer:
441 """Minimize edge crossings using barycenter heuristic."""
443 def __init__(
444 self,
445 layers: list[list[str]],
446 edges: list[tuple[str, str, str]],
447 back_edge_set: set[tuple[str, str]],
448 ) -> None:
449 self._layers = layers
450 self._edges = edges
451 self._back_edge_set = back_edge_set
453 def minimize(self) -> list[list[str]]:
454 """Return reordered layers with reduced crossings."""
455 # Build position lookup
456 layer_of: dict[str, int] = {}
457 for i, layer in enumerate(self._layers):
458 for s in layer:
459 layer_of[s] = i
461 # Forward adjacency (non-back, non-self)
462 adj_down: dict[str, list[str]] = {}
463 adj_up: dict[str, list[str]] = {}
464 for src, dst, _ in self._edges:
465 if (src, dst) in self._back_edge_set or src == dst:
466 continue
467 if src in layer_of and dst in layer_of:
468 adj_down.setdefault(src, []).append(dst)
469 adj_up.setdefault(dst, []).append(src)
471 layers = [list(layer) for layer in self._layers]
473 # 3 sweeps: down, up, down
474 for sweep in range(3):
475 if sweep % 2 == 0:
476 # Top-down sweep
477 for i in range(1, len(layers)):
478 pos_above = {s: j for j, s in enumerate(layers[i - 1])}
479 bary: dict[str, float] = {}
480 for s in layers[i]:
481 parents = [p for p in adj_up.get(s, []) if p in pos_above]
482 if parents:
483 bary[s] = sum(pos_above[p] for p in parents) / len(parents)
484 else:
485 bary[s] = float(layers[i].index(s))
486 layers[i].sort(key=lambda s: bary.get(s, 0))
487 else:
488 # Bottom-up sweep
489 for i in range(len(layers) - 2, -1, -1):
490 pos_below = {s: j for j, s in enumerate(layers[i + 1])}
491 bary_up: dict[str, float] = {}
492 for s in layers[i]:
493 children = [c for c in adj_down.get(s, []) if c in pos_below]
494 if children:
495 bary_up[s] = sum(pos_below[c] for c in children) / len(children)
496 else:
497 bary_up[s] = float(layers[i].index(s))
498 layers[i].sort(key=lambda s: bary_up.get(s, 0))
500 return layers
503# ---------------------------------------------------------------------------
504# Rendering helpers
505# ---------------------------------------------------------------------------
508def _compute_display_labels(
509 all_states: list[str],
510 initial: str,
511 terminal_states: set[str],
512) -> dict[str, str]:
513 """Compute display labels with → prefix and ◉ suffix."""
514 display_label: dict[str, str] = {}
515 for s in all_states:
516 label = s
517 if s == initial:
518 label = "\u2192 " + label
519 if s in terminal_states:
520 label = label + " \u25c9"
521 display_label[s] = label
522 return display_label
525def _compute_box_sizes(
526 all_states: list[str],
527 display_label: dict[str, str],
528 fsm_states: dict[str, StateConfig] | None,
529 verbose: bool,
530 max_box_inner: int,
531 badges: dict[str, str] | None = None,
532 title_only: bool = False,
533) -> tuple[dict[str, list[str]], dict[str, int], dict[str, int], dict[str, str]]:
534 """Compute box content, widths, and heights for all states.
536 Returns (box_inner, box_width, box_height, box_badge).
538 When ``title_only`` is True, action body lines are suppressed (used by
539 ``--show-diagrams=mini`` for skeleton rendering); box widths are computed
540 from the name label / badge only.
541 """
542 box_inner: dict[str, list[str]] = {}
543 box_width: dict[str, int] = {}
544 box_badge: dict[str, str] = {}
546 for s in all_states:
547 state_obj = fsm_states.get(s) if fsm_states else None
549 badge = _get_state_badge(state_obj, badges)
550 badge_w = _badge_display_width(badge) if badge else 0
551 box_badge[s] = badge
553 # Width must fit: name label on content row, badge on top border (with one space
554 # of padding on each side: " badge ")
555 base_w = max(len(display_label[s]), badge_w + 2 if badge_w else 0)
557 inner_w = base_w
558 if not title_only and state_obj and state_obj.action and max_box_inner > 0:
559 action_lines = state_obj.action.strip().splitlines()
560 if verbose:
561 max_action_w = max(
562 (len(ln.rstrip()) for ln in action_lines if ln.rstrip()), default=0
563 )
564 inner_w = max(base_w, min(max_action_w, max_box_inner))
565 else:
566 first_action = next((ln.rstrip() for ln in action_lines if ln.rstrip()), "")
567 inner_w = max(base_w, min(len(first_action), max_box_inner))
569 content = _box_inner_lines(
570 state_obj, display_label[s], verbose, inner_w, title_only=title_only
571 )
572 actual_w = max(len(ln) for ln in content)
573 inner_w = max(inner_w, actual_w)
574 box_inner[s] = content
575 box_width[s] = inner_w + 4 # "│ " + content + " │"
577 box_height: dict[str, int] = {s: len(box_inner[s]) + 2 for s in all_states}
578 return box_inner, box_width, box_height, box_badge
581def _draw_box(
582 grid: list[list[str]],
583 row: int,
584 col: int,
585 width: int,
586 height: int,
587 content: list[str],
588 is_highlighted: bool,
589 highlight_color: str,
590 badge: str = "",
591) -> None:
592 """Draw a state box onto a character grid at (row, col).
594 If *badge* is provided it is placed right-aligned in the top border row with
595 one space of padding on each side (``─ badge ┐``), colorized via ``_bc()``.
596 """
597 total_width = len(grid[0]) if grid else 0
598 try:
599 bg_code: str | None = str(int(highlight_color) + 10)
600 except (ValueError, TypeError):
601 bg_code = None
603 def _bc(ch: str) -> str:
604 if not is_highlighted:
605 return ch
606 if bg_code:
607 return colorize(ch, f"{highlight_color};{bg_code}")
608 return colorize(ch, highlight_color)
610 # Top border: ┌ ─ ─ … ─ ┐
611 if col < total_width:
612 grid[row][col] = _bc("\u250c")
613 for j in range(1, width - 1):
614 if col + j < total_width:
615 grid[row][col + j] = _bc("\u2500")
616 if col + width - 1 < total_width:
617 grid[row][col + width - 1] = _bc("\u2510")
619 # Overlay badge in top-right corner with one space of padding on each side: " badge ┐"
620 if badge:
621 badge_w = _badge_display_width(badge)
622 # Trailing space between badge end and ┐
623 trail_pos = col + width - 2
624 if col + 1 <= trail_pos < col + width - 1 and trail_pos < total_width:
625 grid[row][trail_pos] = _bc(" ")
626 # Leading space before badge
627 lead_pos = col + width - badge_w - 3
628 if col + 1 <= lead_pos < col + width - 1 and lead_pos < total_width:
629 grid[row][lead_pos] = _bc(" ")
630 # Badge characters (shifted left by 1 compared to no-padding placement)
631 pos = col + width - 1 - badge_w - 1
632 for ch in badge:
633 ch_w = _wcwidth(ch)
634 if ch_w < 1:
635 ch_w = 1
636 if col + 1 <= pos < col + width - 1 and pos < total_width:
637 grid[row][pos] = _bc(ch)
638 if ch_w == 2 and pos + 1 < col + width - 1 and pos + 1 < total_width:
639 grid[row][pos + 1] = ""
640 pos += ch_w
642 # Pre-fill all interior cells with bg color so padding rows and gaps are filled
643 if is_highlighted and bg_code:
644 for ri in range(row + 1, row + height - 1):
645 if ri >= len(grid):
646 break
647 for ci in range(col + 1, col + width - 1):
648 if ci < total_width:
649 grid[ri][ci] = colorize(" ", bg_code)
651 # Content rows
652 for i, line in enumerate(content):
653 r = row + i + 1
654 if r >= len(grid):
655 break
656 if col < total_width:
657 grid[r][col] = _bc("\u2502")
658 if col + width - 1 < total_width:
659 grid[r][col + width - 1] = _bc("\u2502")
660 if is_highlighted and i == 0:
661 name_code = f"97;{bg_code};1" if bg_code else f"{highlight_color};1"
662 colored_line = colorize(line, name_code)
663 if col + 2 < total_width:
664 grid[r][col + 2] = colored_line
665 for j in range(1, len(line)):
666 if col + 2 + j < col + width - 1:
667 grid[r][col + 2 + j] = ""
668 elif i == 0:
669 bold_line = colorize(line, "1")
670 if col + 2 < total_width:
671 grid[r][col + 2] = bold_line
672 for j in range(1, len(line)):
673 if col + 2 + j < col + width - 1:
674 grid[r][col + 2 + j] = ""
675 else:
676 for j, ch in enumerate(line):
677 if col + 2 + j < col + width - 1:
678 if is_highlighted and bg_code:
679 grid[r][col + 2 + j] = colorize(ch, f"97;{bg_code}")
680 else:
681 grid[r][col + 2 + j] = ch
683 # Padding rows between content and bottom border
684 for i in range(len(content) + 1, height - 1):
685 r = row + i
686 if r >= len(grid):
687 break
688 if col < total_width:
689 grid[r][col] = _bc("\u2502")
690 if col + width - 1 < total_width:
691 grid[r][col + width - 1] = _bc("\u2502")
693 # Bottom border
694 brow = row + height - 1
695 if brow < len(grid):
696 if col < total_width:
697 grid[brow][col] = _bc("\u2514")
698 for j in range(1, width - 1):
699 if col + j < total_width:
700 grid[brow][col + j] = _bc("\u2500")
701 if col + width - 1 < total_width:
702 grid[brow][col + width - 1] = _bc("\u2518")
705# ---------------------------------------------------------------------------
706# Layered (vertical) renderer
707# ---------------------------------------------------------------------------
710def _render_layered_diagram(
711 layers: list[list[str]],
712 edges: list[tuple[str, str, str]],
713 main_edge_set: set[tuple[str, str]],
714 branches: list[tuple[str, str, str]],
715 back_edges: list[tuple[str, str, str]],
716 initial: str,
717 terminal_states: set[str] | None,
718 fsm_states: dict[str, StateConfig] | None,
719 verbose: bool,
720 highlight_state: str | None,
721 highlight_color: str,
722 edge_label_colors: dict[str, str] | None = None,
723 badges: dict[str, str] | None = None,
724 title_only: bool = False,
725 suppress_labels: bool = False,
726) -> str:
727 """Render FSM using layered (Sugiyama-style) vertical layout.
729 When ``title_only`` is True, per-state body lines are suppressed.
730 When ``suppress_labels`` is True, inter-state edges render without labels.
731 """
732 terminal_states = terminal_states or set()
733 fsm_states = fsm_states or {}
734 tw = terminal_width()
736 # Flatten layers to get all states
737 all_states = [s for layer in layers for s in layer]
738 if not all_states:
739 return ""
741 display_label = _compute_display_labels(all_states, initial, terminal_states)
743 # Compute max_box_inner based on widest layer
744 max_layer_size = max(len(layer) for layer in layers)
745 if verbose:
746 max_box_inner = max(20, min(60, (tw - 4) // max(1, max_layer_size) - 6))
747 else:
748 max_box_inner = max(20, min(40, (tw - 4) // max(1, max_layer_size) - 6))
750 box_inner, box_width, box_height, box_badge = _compute_box_sizes(
751 all_states,
752 display_label,
753 fsm_states,
754 verbose,
755 max_box_inner,
756 badges,
757 title_only=title_only,
758 )
760 # Post-hoc layer merge: re-merge adjacent single-state layers that were
761 # over-split by the conservative max_width_per_layer estimate. Only merge
762 # when both layers receive an edge from the same source state (indicating
763 # they were originally one layer split by width constraint).
764 available_w = tw - 20 # conservative content-area estimate
765 gap_between = 6
766 # Build edge target sets: for each state, which earlier states point to it
767 _incoming: dict[str, set[str]] = {s: set() for layer in layers for s in layer}
768 for src, dst, _ in edges:
769 if src != dst and dst in _incoming:
770 _incoming[dst].add(src)
771 merged = True
772 while merged:
773 merged = False
774 i = 0
775 while i < len(layers) - 1:
776 la, lb = layers[i], layers[i + 1]
777 # Only merge single-state layers that share an incoming source
778 if len(la) == 1 and len(lb) == 1:
779 sources_a = _incoming.get(la[0], set())
780 sources_b = _incoming.get(lb[0], set())
781 shared_source = sources_a & sources_b
782 else:
783 shared_source = set()
784 combined_w = (
785 sum(box_width[s] for s in la)
786 + gap_between * (len(la) - 1)
787 + gap_between
788 + sum(box_width[s] for s in lb)
789 + gap_between * (len(lb) - 1)
790 )
791 if shared_source and combined_w <= available_w and len(la) + len(lb) <= 4:
792 layers[i] = la + lb
793 layers.pop(i + 1)
794 merged = True
795 else:
796 i += 1
798 # Collect ALL non-self-loop forward edge labels (main + branches + same-depth back-edges)
799 # Multiple edges between the same pair are combined as "label1/label2"
800 forward_edge_labels: dict[tuple[str, str], str] = {}
801 for src, dst, lbl in edges:
802 if src == dst:
803 continue
804 if (src, dst) in main_edge_set or (src, dst, lbl) in branches:
805 if (src, dst) in forward_edge_labels:
806 forward_edge_labels[(src, dst)] += "/" + lbl
807 else:
808 forward_edge_labels[(src, dst)] = lbl
810 # True back-edges: only those going to an earlier layer (computed after layer assignment)
811 # Will be finalized below after col positions are computed
812 # Combine same-pair back-edges into single entries with merged labels (e.g. "error/fail")
813 back_edge_labels_initial: dict[tuple[str, str], str] = {}
814 for s, d, lbl in back_edges:
815 if s != d:
816 if (s, d) in back_edge_labels_initial:
817 back_edge_labels_initial[(s, d)] += "/" + lbl
818 else:
819 back_edge_labels_initial[(s, d)] = lbl
821 # Pre-compute layer positions to detect main-path cycle edges early.
822 # This ensures back_edge_margin accounts for ALL backward-pointing edges
823 # (including main-path cycles like commit → initial_state) before column
824 # positions are computed.
825 prelim_layer_of: dict[str, int] = {}
826 for li, layer in enumerate(layers):
827 for s in layer:
828 prelim_layer_of[s] = li
830 # Include main-path/branch edges that point backward in margin estimate
831 all_back_labels: dict[tuple[str, str], str] = dict(back_edge_labels_initial)
832 for (src, dst), lbl in forward_edge_labels.items():
833 src_layer = prelim_layer_of.get(src, -1)
834 dst_layer = prelim_layer_of.get(dst, -1)
835 if dst_layer < src_layer:
836 if (src, dst) in all_back_labels:
837 all_back_labels[(src, dst)] += "/" + lbl
838 else:
839 all_back_labels[(src, dst)] = lbl
841 non_self_back_initial = [(s, d, lbl) for (s, d), lbl in all_back_labels.items()]
842 back_edge_margin = 0
843 if non_self_back_initial:
844 max_label_len = max(len(lbl) for _, _, lbl in non_self_back_initial)
845 n_back_initial = len(non_self_back_initial)
846 back_edge_margin = max_label_len + max(6, 2 * n_back_initial + 2)
848 content_left = 2 + back_edge_margin
850 # Self-loops per state
851 self_loops: dict[str, list[str]] = {}
852 for src, dst, lbl in back_edges:
853 if src == dst:
854 self_loops.setdefault(src, []).append(lbl)
856 # --- Compute a common center column for alignment ---
857 # For layers with single boxes, we want vertical alignment through a
858 # shared center column. Use the widest single-state layer's center.
859 max_single_w = max((box_width[layer[0]] for layer in layers if len(layer) == 1), default=0)
860 # The common center is at content_left + max_single_w // 2
861 common_center = content_left + max_single_w // 2
863 # Compute column positions per layer
864 col_start: dict[str, int] = {}
865 col_center: dict[str, int] = {}
866 layer_of: dict[str, int] = {}
868 for li, layer in enumerate(layers):
869 if len(layer) == 1:
870 # Single-state layer: center-align to common center
871 sname = layer[0]
872 col_start[sname] = common_center - box_width[sname] // 2
873 col_center[sname] = common_center
874 layer_of[sname] = li
875 else:
876 # Multi-state layer: place side-by-side, centered around common_center
877 gap_between = 6
878 total_layer_w = sum(box_width[s] for s in layer) + gap_between * (len(layer) - 1)
879 x = common_center - total_layer_w // 2
880 x = max(content_left, x)
881 for i, sname in enumerate(layer):
882 col_start[sname] = x
883 col_center[sname] = x + box_width[sname] // 2
884 layer_of[sname] = li
885 if i < len(layer) - 1:
886 next_s = layer[i + 1]
887 # Check for edge labels in both directions between adjacent states
888 label_fwd = forward_edge_labels.get((sname, next_s), "")
889 label_rev = forward_edge_labels.get((next_s, sname), "")
890 max_label = max(len(label_fwd), len(label_rev))
891 gap = max(gap_between, max_label + 6) if max_label > 0 else gap_between
892 x += box_width[sname] + gap
893 else:
894 x += box_width[sname]
896 # Reclassify back-edges based on actual layer positions
897 # Only edges going to an earlier layer are true margin back-edges
898 # Combine same-pair edges into merged labels (e.g. "error/fail")
899 back_edge_labels_reclass: dict[tuple[str, str], str] = {}
900 same_layer_edges: list[tuple[str, str, str]] = []
901 for src, dst, lbl in back_edges:
902 if src == dst:
903 continue
904 src_layer = layer_of.get(src, -1)
905 dst_layer = layer_of.get(dst, -1)
906 if dst_layer < src_layer:
907 if (src, dst) in back_edge_labels_reclass:
908 back_edge_labels_reclass[(src, dst)] += "/" + lbl
909 else:
910 back_edge_labels_reclass[(src, dst)] = lbl
911 elif dst_layer == src_layer:
912 same_layer_edges.append((src, dst, lbl))
913 else: # dst_layer > src_layer: actually forward edge
914 if (src, dst) in forward_edge_labels:
915 forward_edge_labels[(src, dst)] += "/" + lbl
916 else:
917 forward_edge_labels[(src, dst)] = lbl
919 # Also reclassify main/branch edges in forward_edge_labels that point backward
920 # after layer assignment (e.g. main-path cycle edges like commit → initial_state)
921 backward_in_fwd: list[tuple[str, str]] = []
922 for (src, dst), lbl in forward_edge_labels.items():
923 src_layer = layer_of.get(src, -1)
924 dst_layer = layer_of.get(dst, -1)
925 if dst_layer < src_layer:
926 backward_in_fwd.append((src, dst))
927 if (src, dst) in back_edge_labels_reclass:
928 back_edge_labels_reclass[(src, dst)] += "/" + lbl
929 else:
930 back_edge_labels_reclass[(src, dst)] = lbl
931 elif dst_layer == src_layer and src != dst:
932 backward_in_fwd.append((src, dst))
933 same_layer_edges.append((src, dst, lbl))
934 for key in backward_in_fwd:
935 del forward_edge_labels[key]
937 # Add same-layer back-edges to forward_edge_labels so gap calculation accounts for them
938 for src, dst, lbl in same_layer_edges:
939 if (src, dst) in forward_edge_labels:
940 forward_edge_labels[(src, dst)] += "/" + lbl
941 else:
942 forward_edge_labels[(src, dst)] = lbl
944 # Recalculate inter-box gaps for layers with newly discovered same-layer edges
945 affected_layers: set[int] = set()
946 for src, dst, _lbl in same_layer_edges:
947 sl = layer_of.get(src, -1)
948 dl = layer_of.get(dst, -1)
949 if sl >= 0:
950 affected_layers.add(sl)
951 if dl >= 0:
952 affected_layers.add(dl)
953 for li in affected_layers:
954 layer = layers[li]
955 if len(layer) < 2:
956 continue
957 gap_between = 6
958 total_layer_w = sum(box_width[s] for s in layer)
959 # For non-adjacent same-layer edges the label lands in the gap immediately
960 # adjacent to the source box (left of src for right-to-left, right of src
961 # for left-to-right). Collect those requirements so the gap is wide enough.
962 extra_gap_req: dict[tuple[str, str], int] = {}
963 for src, dst, lbl in same_layer_edges:
964 if layer_of.get(src) != li or layer_of.get(dst) != li:
965 continue
966 try:
967 si, di = layer.index(src), layer.index(dst)
968 except ValueError:
969 continue
970 if abs(si - di) <= 1:
971 continue # adjacent — already handled by forward_edge_labels
972 if si > di:
973 key = (layer[si - 1], src) # gap to the left of src
974 else:
975 key = (src, layer[si + 1]) # gap to the right of src
976 extra_gap_req[key] = max(extra_gap_req.get(key, 0), len(lbl))
977 # Recalculate gaps with label-aware spacing
978 gaps: list[int] = []
979 for i in range(len(layer) - 1):
980 sname, next_s = layer[i], layer[i + 1]
981 label_fwd = forward_edge_labels.get((sname, next_s), "")
982 label_rev = forward_edge_labels.get((next_s, sname), "")
983 max_label = max(len(label_fwd), len(label_rev), extra_gap_req.get((sname, next_s), 0))
984 gap = max(gap_between, max_label + 6) if max_label > 0 else gap_between
985 gaps.append(gap)
986 total_layer_w += sum(gaps)
987 x = common_center - total_layer_w // 2
988 x = max(content_left, x)
989 for i, sname in enumerate(layer):
990 col_start[sname] = x
991 col_center[sname] = x + box_width[sname] // 2
992 if i < len(layer) - 1:
993 x += box_width[sname] + gaps[i]
994 else:
995 x += box_width[sname]
997 non_self_back = [(s, d, lbl) for (s, d), lbl in back_edge_labels_reclass.items()]
999 # Recalculate back-edge margin if it changed
1000 if non_self_back:
1001 max_label_len = max(len(lbl) for _, _, lbl in non_self_back)
1002 n_back = len(non_self_back)
1003 actual_margin = max_label_len + max(6, 2 * n_back + 2)
1004 if actual_margin != back_edge_margin:
1005 # Need to recalculate positions (rare case - usually matches)
1006 back_edge_margin = actual_margin
1007 content_left = 2 + back_edge_margin
1009 # Identify forward skip-layer edges (span > 1 layer, not handled by consecutive renderer)
1010 skip_forward_edges: list[tuple[str, str, str]] = []
1011 for (src, dst), lbl in forward_edge_labels.items():
1012 src_layer = layer_of.get(src, -1)
1013 dst_layer = layer_of.get(dst, -1)
1014 if dst_layer > src_layer + 1:
1015 skip_forward_edges.append((src, dst, lbl))
1017 # Pre-compute right margin width for forward skip-layer edges
1018 right_edge_margin = 0
1019 if skip_forward_edges:
1020 max_fwd_label_len = max(len(lbl) for _, _, lbl in skip_forward_edges)
1021 right_edge_margin = max_fwd_label_len + 6
1023 # Compute total width needed
1024 total_content_w = 0
1025 for s in all_states:
1026 right = col_start[s] + box_width[s]
1027 total_content_w = max(total_content_w, right)
1028 total_width = max(total_content_w + right_edge_margin + 4, tw)
1030 # Compute vertical positions with space for self-loops and inter-layer arrows
1031 row_start: dict[str, int] = {}
1032 y = 0
1033 for li, layer in enumerate(layers):
1034 layer_h = max(box_height[s] for s in layer)
1035 for sname in layer:
1036 row_start[sname] = y
1037 y += layer_h
1039 # Add self-loop row if any state in this layer has self-loops
1040 has_self_loop = any(s in self_loops for s in layer)
1041 if has_self_loop:
1042 y += 1 # self-loop marker row
1044 if li < len(layers) - 1:
1045 y += 3 # arrow gap: │, label, ▼
1047 total_height = y
1049 # Build character grid
1050 grid: list[list[str]] = [[" "] * total_width for _ in range(total_height)]
1052 # Draw boxes
1053 for sname in all_states:
1054 is_highlighted = highlight_state is not None and sname == highlight_state
1055 _draw_box(
1056 grid,
1057 row_start[sname],
1058 col_start[sname],
1059 box_width[sname],
1060 box_height[sname],
1061 box_inner[sname],
1062 is_highlighted,
1063 highlight_color,
1064 badge=box_badge[sname],
1065 )
1067 # Precompute box-occupied (row, col) pairs so connector lines can avoid overwriting box cells
1068 _box_occ: dict[int, set[int]] = {}
1069 for _s in all_states:
1070 for _r in range(row_start[_s], row_start[_s] + box_height[_s]):
1071 _row_set = _box_occ.setdefault(_r, set())
1072 for _c in range(col_start[_s], col_start[_s] + box_width[_s]):
1073 _row_set.add(_c)
1075 # Draw self-loop markers immediately below their boxes
1076 for sname, labels in self_loops.items():
1077 marker = "\u21ba" if suppress_labels else "\u21ba " + ", ".join(labels)
1078 r = row_start[sname] + box_height[sname]
1079 if r < total_height:
1080 cx = col_center[sname]
1081 pos = max(0, cx - len(marker) // 2)
1082 for j, ch in enumerate(marker):
1083 if pos + j < total_width:
1084 grid[r][pos + j] = ch
1086 # Shared row tracker: prevents two labels (back-edge, skip-forward, or adjacent)
1087 # landing on the same grid row, which would clobber the first label written there.
1088 used_label_rows: set[int] = set()
1090 # Draw forward edges between layers (vertical arrows with labels)
1091 for li in range(len(layers) - 1):
1092 layer_h = max(box_height[s] for s in layers[li])
1093 has_self_loop = any(s in self_loops for s in layers[li])
1094 self_loop_offset = 1 if has_self_loop else 0
1096 # Arrow area starts after box bottom + self-loop row
1097 arrow_start_row = row_start[layers[li][0]] + layer_h + self_loop_offset
1098 arrow_end_row = row_start[layers[li + 1][0]] - 1
1100 # Collect all inter-layer edges from this layer to the next
1101 inter_edges: list[tuple[str, str, str]] = []
1102 for src in layers[li]:
1103 for dst in layers[li + 1]:
1104 label = forward_edge_labels.get((src, dst))
1105 if label is not None:
1106 inter_edges.append((src, dst, label))
1108 # Draw each edge with its own vertical pipe to the target's center
1109 for src, dst, label in inter_edges:
1110 dst_cc = col_center[dst]
1111 src_left = col_start[src]
1112 src_right = src_left + box_width[src]
1113 ec = _edge_line_color(label) # ANSI code for this edge's connector chars
1115 def _lc(ch: str, _ec: str = ec) -> str: # noqa: E306
1116 return colorize(ch, _ec) if _ec else ch
1118 # Horizontal connector when pipe is outside source box range
1119 if dst_cc >= src_right or dst_cc < src_left:
1120 conn_row = arrow_start_row
1121 if 0 <= conn_row < total_height:
1122 if dst_cc >= src_right:
1123 # Pipe right of source: └───┐
1124 src_cc = col_center[src]
1125 if 0 <= src_cc < total_width and grid[conn_row][src_cc] == " ":
1126 grid[conn_row][src_cc] = _lc("\u2514") # └
1127 start_c = src_cc + 1
1128 else:
1129 start_c = src_right
1130 for c in range(start_c, dst_cc):
1131 if 0 <= c < total_width:
1132 grid[conn_row][c] = _lc("\u2500")
1133 if 0 <= dst_cc < total_width:
1134 grid[conn_row][dst_cc] = _lc("\u2510") # ┐
1135 else:
1136 # Pipe left of source: ┌───┘
1137 src_cc = col_center[src]
1138 if 0 <= src_cc < total_width and grid[conn_row][src_cc] == " ":
1139 end_c = src_cc
1140 grid[conn_row][src_cc] = _lc("\u2518") # ┘
1141 else:
1142 end_c = src_left
1143 for c in range(dst_cc + 1, end_c):
1144 if 0 <= c < total_width:
1145 grid[conn_row][c] = _lc("\u2500")
1146 if 0 <= dst_cc < total_width:
1147 grid[conn_row][dst_cc] = _lc("\u250c") # ┌
1148 pipe_start = arrow_start_row + 1
1149 else:
1150 pipe_start = arrow_start_row
1152 # Draw vertical pipe at destination's center column
1153 for r in range(pipe_start, arrow_end_row):
1154 if 0 <= dst_cc < total_width and r < total_height:
1155 grid[r][dst_cc] = _lc("\u2502")
1157 # Arrow tip at destination center
1158 if arrow_end_row < total_height and 0 <= dst_cc < total_width:
1159 grid[arrow_end_row][dst_cc] = _lc("\u25bc")
1161 # Label to the right of the pipe. When skip-layer forward edges
1162 # exist their vertical pipes occupy columns starting at
1163 # total_content_w+2, so clamp to total_content_w to avoid
1164 # overwriting them (BUG-1500). Without skip-layer edges the right
1165 # margin is empty and the full total_width is available.
1166 label_row = arrow_start_row
1167 # Nudge if this row already has a label from another inter-layer edge
1168 # (e.g., two edges from the same source go to states in the same layer).
1169 # Try pipe rows (pipe_start..arrow_end_row) before giving up (BUG-1501).
1170 if label_row in used_label_rows:
1171 for _cand in range(pipe_start, arrow_end_row + 1):
1172 if _cand not in used_label_rows:
1173 label_row = _cand
1174 break
1175 if label_row < total_height and not suppress_labels:
1176 used_label_rows.add(label_row)
1177 label_start = dst_cc + 2
1178 max_col = total_content_w if skip_forward_edges else total_width
1179 max_label = max_col - label_start
1180 if 0 < max_label < len(label):
1181 label = label[: max_label - 1] + "…"
1182 for j, ch in enumerate(label):
1183 if label_start + j < max_col:
1184 grid[label_row][label_start + j] = ch
1186 # Post-pass: connect horizontal gaps for multi-branch sources
1187 if len(inter_edges) >= 2 and 0 <= arrow_start_row < total_height:
1188 src_targets: dict[str, list[int]] = {}
1189 for src, dst, _ in inter_edges:
1190 if dst in col_center:
1191 src_targets.setdefault(src, []).append(col_center[dst])
1192 for _src, centers in src_targets.items():
1193 if len(centers) < 2:
1194 continue
1195 left = min(centers)
1196 right = max(centers)
1197 for c in range(left + 1, right):
1198 if 0 <= c < total_width:
1199 cell = grid[arrow_start_row][c]
1200 if cell == " ":
1201 grid[arrow_start_row][c] = "\u2500" # ─
1202 elif cell == "\u2502": # │ → ┼
1203 grid[arrow_start_row][c] = "\u253c"
1204 elif cell == "\u2518": # ┘ → ┴
1205 grid[arrow_start_row][c] = "\u2534"
1206 elif cell == "\u2514": # └ → ┴
1207 grid[arrow_start_row][c] = "\u2534"
1208 elif cell == "\u2510": # ┐ → ┬
1209 grid[arrow_start_row][c] = "\u252c"
1210 elif cell == "\u250c": # ┌ → ┬
1211 grid[arrow_start_row][c] = "\u252c"
1212 # Update boundary junction chars where the horizontal bar meets a pipe
1213 if 0 <= left < total_width and grid[arrow_start_row][left] == "\u2502": # │ → ├
1214 grid[arrow_start_row][left] = "\u251c"
1215 if 0 <= right < total_width and grid[arrow_start_row][right] == "\u2502": # │ → ┤
1216 grid[arrow_start_row][right] = "\u2524"
1218 # Draw same-layer edges (horizontal arrows between states on same layer)
1219 # Includes both branches and reclassified back-edges within same layer
1220 all_same_layer: list[tuple[str, str, str]] = list(same_layer_edges)
1221 for _li, layer in enumerate(layers):
1222 for i, src in enumerate(layer):
1223 for j, dst in enumerate(layer):
1224 if i == j:
1225 continue
1226 label = forward_edge_labels.get((src, dst))
1227 if label is not None and (src, dst, label) not in all_same_layer:
1228 all_same_layer.append((src, dst, label))
1230 for src, dst, label in all_same_layer:
1231 if src not in col_start or dst not in col_start:
1232 continue
1233 if suppress_labels:
1234 label = ""
1235 name_row = row_start[src] + 1
1236 src_right = col_start[src] + box_width[src]
1237 dst_right = col_start[dst] + box_width[dst]
1238 dst_left = col_start[dst]
1239 src_left = col_start[src]
1240 _row_boxes = _box_occ.get(name_row, set())
1241 ec = _edge_line_color(label)
1243 def _lc(ch: str, _ec: str = ec) -> str: # noqa: E306
1244 return colorize(ch, _ec) if _ec else ch
1246 if dst_left >= src_right:
1247 # Left to right horizontal arrow: src ──label──▶ dst
1248 start = src_right
1249 end = dst_left
1250 edge_text = "\u2500" + label + "\u2500\u2500\u25b6"
1251 available = end - start
1252 if available < len(edge_text):
1253 edge_text = edge_text[: max(1, available)]
1254 left_dashes = max(0, available - len(edge_text))
1255 for k in range(left_dashes):
1256 pos = start + k
1257 if pos < total_width and name_row < total_height and pos not in _row_boxes:
1258 grid[name_row][pos] = _lc("\u2500")
1259 for k, ch in enumerate(edge_text):
1260 pos = start + left_dashes + k
1261 if (
1262 0 <= pos < end
1263 and pos < total_width
1264 and name_row < total_height
1265 and pos not in _row_boxes
1266 ):
1267 grid[name_row][pos] = _lc(ch)
1268 elif dst_right <= src_left:
1269 # Right to left: dst is left of src: src → dst drawn as dst ◀──label── src
1270 start = dst_right
1271 end = src_left
1272 edge_text = "\u25c4\u2500\u2500" + label + "\u2500"
1273 available = end - start
1274 if available < len(edge_text):
1275 edge_text = edge_text[: max(1, available)]
1276 for k, ch in enumerate(edge_text):
1277 pos = start + k
1278 if (
1279 0 <= pos < end
1280 and pos < total_width
1281 and name_row < total_height
1282 and pos not in _row_boxes
1283 ):
1284 grid[name_row][pos] = _lc(ch)
1285 for k in range(start + len(edge_text), end):
1286 if k < total_width and name_row < total_height and k not in _row_boxes:
1287 grid[name_row][k] = _lc("\u2500")
1289 # Back-edges: left-margin vertical arrows with labels
1290 if non_self_back:
1291 sorted_back = sorted(
1292 non_self_back,
1293 key=lambda e: abs(row_start.get(e[0], 0) - row_start.get(e[1], 0)),
1294 reverse=True,
1295 )
1296 used_cols: list[int] = []
1297 # Compute rightmost pipe column so labels go right of ALL pipes
1298 rightmost_pipe_col = 1 + (len(sorted_back) - 1) * 2
1300 for src, dst, label in sorted_back:
1301 # Source: name row of source box; target: name row of target box
1302 src_row = row_start.get(src, 0) + 1 # name row, not bottom border
1303 dst_row = row_start.get(dst, 0) + 1 # name row
1305 # Find a free column in the margin
1306 col = 1
1307 for uc in sorted(used_cols):
1308 if uc == col:
1309 col += 2
1310 used_cols.append(col)
1312 if col >= content_left - 1:
1313 continue
1315 top_row = min(src_row, dst_row)
1316 bot_row = max(src_row, dst_row)
1317 ec = _edge_line_color(label)
1319 def _lc(ch: str, _ec: str = ec) -> str: # noqa: E306
1320 return colorize(ch, _ec) if _ec else ch
1322 # Draw vertical line in margin (exclude corner rows handled below)
1323 for r in range(top_row + 1, bot_row):
1324 if 0 <= r < total_height and col < total_width:
1325 cell = grid[r][col]
1326 if cell == "\u2500": # ─ → ┼ (junction, leave uncolored)
1327 grid[r][col] = "\u253c"
1328 elif cell == " ":
1329 grid[r][col] = _lc("\u2502")
1331 # Horizontal connector from source box to margin
1332 # Draw right-to-left, crossing existing pipes with junction chars
1333 if 0 <= src_row < total_height:
1334 src_left = col_start.get(src, col + 1)
1335 _src_row_boxes = _box_occ.get(src_row, set())
1336 for c in range(col + 1, src_left):
1337 if c < total_width and c not in _src_row_boxes:
1338 cell = grid[src_row][c]
1339 if cell == " ":
1340 grid[src_row][c] = _lc("\u2500") # ─
1341 elif cell == "\u2502": # │ → ┼ (junction)
1342 grid[src_row][c] = "\u253c"
1343 elif cell == "\u2514": # └ → ┴ (junction)
1344 grid[src_row][c] = "\u2534"
1345 elif cell == "\u250c": # ┌ → ┬ (junction)
1346 grid[src_row][c] = "\u252c"
1347 elif cell == "\u251c": # ├ → ┼ (junction)
1348 grid[src_row][c] = "\u253c"
1349 # Leave ─, ▶, box chars unchanged
1351 # Horizontal connector from margin to target box
1352 # Draw right-to-left, crossing existing pipes with junction chars
1353 dst_left = col_start.get(dst, col + 1)
1354 if 0 <= dst_row < total_height:
1355 _dst_row_boxes = _box_occ.get(dst_row, set())
1356 for c in range(col + 1, dst_left):
1357 if c < total_width and c not in _dst_row_boxes:
1358 cell = grid[dst_row][c]
1359 if cell == " ":
1360 grid[dst_row][c] = _lc("\u2500") # ─
1361 elif cell == "\u2502": # │ → ┼ (junction)
1362 grid[dst_row][c] = "\u253c"
1363 elif cell == "\u2514": # └ → ┴ (junction)
1364 grid[dst_row][c] = "\u2534"
1365 elif cell == "\u250c": # ┌ → ┬ (junction)
1366 grid[dst_row][c] = "\u252c"
1367 elif cell == "\u251c": # ├ → ┼ (junction)
1368 grid[dst_row][c] = "\u253c"
1370 # Corner characters at pipe-to-horizontal turn points
1371 for row in (src_row, dst_row):
1372 if 0 <= row < total_height and col < total_width:
1373 existing = grid[row][col]
1374 if row == bot_row:
1375 # Pipe ends, turns right: └; if horizontal already crosses here: ┴
1376 grid[row][col] = "\u2534" if existing == "\u2500" else _lc("\u2514")
1377 else: # row == top_row
1378 # Pipe starts going down, turns right: ┌; if horizontal already crosses here: ┬
1379 grid[row][col] = "\u252c" if existing == "\u2500" else _lc("\u250c")
1381 # Arrow tip at target: place ▶ at end of horizontal connector (entering box from left)
1382 if 0 <= dst_row < total_height and dst_left - 1 > col and dst_left - 1 < total_width:
1383 grid[dst_row][dst_left - 1] = _lc("\u25b6")
1385 # Label on the margin line (right of ALL pipes, not just this one)
1386 label_row_pos = (top_row + bot_row) // 2
1387 # Nudge away from already-used rows to prevent clobbering earlier labels
1388 if label_row_pos in used_label_rows and top_row + 1 < bot_row:
1389 midpoint = label_row_pos
1390 found = False
1391 for _off in range(1, bot_row - top_row):
1392 for _cand in (midpoint - _off, midpoint + _off):
1393 if top_row < _cand < bot_row and _cand not in used_label_rows:
1394 label_row_pos = _cand
1395 found = True
1396 break
1397 if found:
1398 break
1399 if not found:
1400 label_row_pos = top_row + 1
1401 used_label_rows.add(label_row_pos)
1402 if 0 <= label_row_pos < total_height and not title_only:
1403 label_start = rightmost_pipe_col + 2
1404 for j, ch in enumerate(label):
1405 if label_start + j < content_left - 1 and label_start + j < total_width:
1406 grid[label_row_pos][label_start + j] = _lc(ch)
1408 # Forward skip-layer edges: right-margin vertical arrows with labels
1409 # Symmetric to the left-margin back-edge renderer above
1410 if skip_forward_edges:
1411 sorted_fwd_skip = sorted(
1412 skip_forward_edges,
1413 key=lambda e: abs(row_start.get(e[0], 0) - row_start.get(e[1], 0)),
1414 reverse=True,
1415 )
1416 used_fwd_cols: list[int] = []
1417 # Rightmost pipe column (furthest from content) for label placement
1418 rightmost_fwd_pipe_col = total_content_w + 2 + (len(sorted_fwd_skip) - 1) * 2
1420 for src, dst, label in sorted_fwd_skip:
1421 src_row = row_start.get(src, 0) + 1 # name row
1422 dst_row = row_start.get(dst, 0) + 1 # name row
1424 # Allocate column in right margin (starting from content edge, going right)
1425 col = total_content_w + 2
1426 for uc in sorted(used_fwd_cols):
1427 if uc == col:
1428 col += 2
1429 used_fwd_cols.append(col)
1431 if col >= total_width:
1432 continue
1434 top_row = min(src_row, dst_row)
1435 bot_row = max(src_row, dst_row)
1436 ec = _edge_line_color(label)
1438 def _lc(ch: str, _ec: str = ec) -> str: # noqa: E306
1439 return colorize(ch, _ec) if _ec else ch
1441 # Draw vertical line in right margin (exclude corner rows handled below)
1442 for r in range(top_row + 1, bot_row):
1443 if 0 <= r < total_height and col < total_width:
1444 cell = grid[r][col]
1445 if cell == "\u2500": # ─ → ┼ (junction)
1446 grid[r][col] = "\u253c"
1447 elif cell == " ":
1448 grid[r][col] = _lc("\u2502")
1450 # Horizontal connector from source box right side to margin
1451 # Draw left-to-right, crossing existing pipes with junction chars
1452 src_right = col_start.get(src, 0) + box_width.get(src, 0)
1453 _src_row_boxes = _box_occ.get(src_row, set())
1454 if 0 <= src_row < total_height:
1455 for c in range(src_right, col):
1456 if 0 <= c < total_width and c not in _src_row_boxes:
1457 cell = grid[src_row][c]
1458 if cell == " ":
1459 grid[src_row][c] = _lc("\u2500") # ─
1460 elif cell == "\u2502": # │ → ┼ (junction)
1461 grid[src_row][c] = "\u253c"
1462 elif cell == "\u2518": # ┘ → ┴ (junction)
1463 grid[src_row][c] = "\u2534"
1464 elif cell == "\u2510": # ┐ → ┬ (junction)
1465 grid[src_row][c] = "\u252c"
1466 elif cell == "\u2524": # ┤ → ┼ (junction)
1467 grid[src_row][c] = "\u253c"
1468 # Leave ─, ◀, box chars unchanged
1470 # Horizontal connector from margin to destination box right side
1471 dst_right = col_start.get(dst, 0) + box_width.get(dst, 0)
1472 _dst_row_boxes = _box_occ.get(dst_row, set())
1473 if 0 <= dst_row < total_height:
1474 for c in range(dst_right, col):
1475 if 0 <= c < total_width and c not in _dst_row_boxes:
1476 cell = grid[dst_row][c]
1477 if cell == " ":
1478 grid[dst_row][c] = _lc("\u2500") # ─
1479 elif cell == "\u2502": # │ → ┼ (junction)
1480 grid[dst_row][c] = "\u253c"
1481 elif cell == "\u2518": # ┘ → ┴ (junction)
1482 grid[dst_row][c] = "\u2534"
1483 elif cell == "\u2510": # ┐ → ┬ (junction)
1484 grid[dst_row][c] = "\u252c"
1485 elif cell == "\u2524": # ┤ → ┼ (junction)
1486 grid[dst_row][c] = "\u253c"
1488 # Corner characters at pipe-to-horizontal turn points
1489 for row in (src_row, dst_row):
1490 if 0 <= row < total_height and col < total_width:
1491 existing = grid[row][col]
1492 if row == bot_row:
1493 # Pipe ends, turns left: ┘; if horizontal crosses: ┤
1494 grid[row][col] = "\u2524" if existing == "\u2500" else _lc("\u2518")
1495 else: # row == top_row
1496 # Pipe starts going down, turns left: ┐; if horizontal crosses: ┤
1497 grid[row][col] = "\u2524" if existing == "\u2500" else _lc("\u2510")
1499 # Arrow tip at target: ◀ entering box from right side
1500 if 0 <= dst_row < total_height and dst_right < col and dst_right < total_width:
1501 grid[dst_row][dst_right] = _lc("\u25c0")
1503 # Label on the margin line (right of ALL pipes, mirroring left-margin
1504 # approach). Truncate with … when label would exceed total_width
1505 # to prevent extending diagram lines far past the content area (BUG-1500).
1506 label_row_pos = (top_row + bot_row) // 2
1507 # Nudge to avoid row collision with a previously-placed label (back- or fwd-edge)
1508 if label_row_pos in used_label_rows and top_row + 1 < bot_row:
1509 midpoint = label_row_pos
1510 found = False
1511 for _off in range(1, bot_row - top_row):
1512 for _cand in (midpoint - _off, midpoint + _off):
1513 if top_row < _cand < bot_row and _cand not in used_label_rows:
1514 label_row_pos = _cand
1515 found = True
1516 break
1517 if found:
1518 break
1519 if not found:
1520 label_row_pos = top_row + 1
1521 used_label_rows.add(label_row_pos)
1522 if 0 <= label_row_pos < total_height and not title_only:
1523 label_start = rightmost_fwd_pipe_col + 2
1524 max_label = total_width - label_start
1525 if 0 < max_label < len(label):
1526 label = label[: max_label - 1] + "…"
1527 for j, ch in enumerate(label):
1528 if label_start + j < total_width:
1529 grid[label_row_pos][label_start + j] = _lc(ch)
1531 # Convert grid to string
1532 lines = ["".join(row).rstrip() for row in grid]
1534 # Remove trailing empty lines
1535 while lines and not lines[-1].strip():
1536 lines.pop()
1538 # Center diagram
1539 max_line_len = max((len(_ANSI_ESCAPE_RE.sub("", ln)) for ln in lines), default=0)
1540 diagram_indent = max(0, (tw - max_line_len) // 2)
1541 if diagram_indent > 0:
1542 lines = [" " * diagram_indent + ln if ln.strip() else ln for ln in lines]
1544 return _colorize_diagram_labels("\n".join(lines), edge_label_colors)
1547# ---------------------------------------------------------------------------
1548# FSM diagram renderer (main entry point)
1549# ---------------------------------------------------------------------------
1552_MAIN_PATH_EDGE_LABELS: frozenset[str] = frozenset({"yes", "no", "next", "_"})
1555def _filter_main_path_graph(
1556 fsm: FSMLoop, edges: list[tuple[str, str, str]]
1557) -> tuple[list[tuple[str, str, str]], set[str]]:
1558 """Filter edges to the main happy-path subgraph and the states reachable through it.
1560 Drops off-happy-path labels (anything not in ``_MAIN_PATH_EDGE_LABELS`` and not
1561 a ``route``-verdict label) plus any state unreachable from ``fsm.initial`` once
1562 those edges are gone. Route-verdict labels (everything emitted by ``state.route``
1563 other than the default ``_``) are kept since routes encode normal branching.
1564 """
1565 route_verdicts: set[str] = set()
1566 for _name, state in fsm.states.items():
1567 if state.route is not None:
1568 route_verdicts.update(state.route.routes.keys())
1569 route_verdicts.update(state.extra_routes.keys())
1571 def _is_main_label(label: str) -> bool:
1572 return label in _MAIN_PATH_EDGE_LABELS or label in route_verdicts
1574 filtered_edges = [(s, t, lbl) for (s, t, lbl) in edges if _is_main_label(lbl)]
1575 _bfs_visited_order, depth_map = _bfs_order(fsm.initial, filtered_edges)
1576 reachable: set[str] = set(depth_map.keys())
1577 filtered_edges = [
1578 (s, t, lbl) for (s, t, lbl) in filtered_edges if s in reachable and t in reachable
1579 ]
1580 return filtered_edges, reachable
1583def _render_fsm_diagram(
1584 fsm: FSMLoop,
1585 verbose: bool = False,
1586 highlight_state: str | None = None,
1587 highlight_color: str = "32",
1588 edge_label_colors: dict[str, str] | None = None,
1589 badges: dict[str, str] | None = None,
1590 mode: str = "full",
1591 *,
1592 suppress_labels: bool = False,
1593 title_only: bool = False,
1594) -> str:
1595 """Render an adaptive text diagram of the FSM graph.
1597 Detects FSM topology and selects appropriate layout:
1598 - Linear chains: vertical top-to-bottom
1599 - Branching/cyclic: layered Sugiyama-style
1601 Args:
1602 fsm: The FSM loop to render.
1603 verbose: If True, show expanded action content in boxes.
1604 highlight_state: If provided, render this state's box with the highlight color.
1605 highlight_color: ANSI SGR code for the highlighted state (default: green).
1606 edge_label_colors: Optional label→SGR-code mapping for transition labels.
1607 Falls back to hardcoded defaults when None.
1608 badges: Optional glyph-key→string mapping for state type badges.
1609 Falls back to hardcoded defaults when None.
1610 mode: Controls edge scope: "main" (default when filtering active) hides
1611 off-happy-path edges. "full" renders every edge and state. "mini" is
1612 an alias for main-scope; use ``suppress_labels=True, title_only=True``
1613 instead. Callers that need full-detail dumps (e.g. ``ll-loop info``)
1614 keep the default "full".
1615 suppress_labels: If True, edge labels are omitted from all rendered edges.
1616 title_only: If True, state boxes show only the state name (no action body).
1617 """
1618 edges = _collect_edges(fsm)
1619 if mode in ("main", "mini"):
1620 edges, _reachable = _filter_main_path_graph(fsm, edges)
1621 bfs_order_list, bfs_depth = _bfs_order(fsm.initial, edges)
1622 main_path, main_edge_set = _trace_main_path(fsm, edges)
1623 branches, back_edges = _classify_edges(edges, main_edge_set, bfs_order_list)
1625 terminal_states = {name for name, state in fsm.states.items() if state.terminal}
1627 # Collect all states
1628 all_states = list(main_path)
1629 for src, dst, _ in branches:
1630 for s in (src, dst):
1631 if s not in all_states:
1632 all_states.append(s)
1634 # Topology detection
1635 detector = TopologyDetector(edges, main_path, branches, back_edges)
1636 topology = detector.classify()
1638 # Build back-edge set for layout pipeline
1639 back_edge_set: set[tuple[str, str]] = set()
1640 for src, dst, _ in back_edges:
1641 if src != dst:
1642 back_edge_set.add((src, dst))
1644 tw = terminal_width()
1646 if topology == "linear" and len(all_states) <= 1:
1647 # Single state or empty — use simple horizontal
1648 return _render_horizontal_simple(
1649 main_path,
1650 edges,
1651 main_edge_set,
1652 branches,
1653 back_edges,
1654 bfs_order_list,
1655 fsm.initial,
1656 terminal_states,
1657 fsm.states,
1658 verbose,
1659 highlight_state,
1660 highlight_color,
1661 edge_label_colors,
1662 badges,
1663 title_only=title_only or (mode == "mini"),
1664 suppress_labels=suppress_labels or (mode == "mini"),
1665 )
1667 # Compute max node width to determine width constraint
1668 # Quick estimate: widest state name or badge + padding
1669 max_node_w = 30 # reasonable default
1670 for s in all_states:
1671 st = fsm.states.get(s)
1672 badge = _get_state_badge(st, badges)
1673 badge_w = _badge_display_width(badge) if badge else 0
1674 label = s
1675 if s == fsm.initial:
1676 label = "\u2192 " + label
1677 if s in terminal_states:
1678 label = label + " \u25c9"
1679 w = max(len(label), badge_w)
1680 max_node_w = max(max_node_w, w + 4 + 4) # inner + borders + padding
1682 max_width_per_layer = max(1, (tw - 10) // (max_node_w + 4))
1684 # Layer assignment
1685 assigner = LayerAssigner(all_states, edges, back_edge_set, fsm.initial, max_width_per_layer)
1686 layers = assigner.assign()
1688 # Crossing minimization
1689 minimizer = CrossingMinimizer(layers, edges, back_edge_set)
1690 layers = minimizer.minimize()
1692 return _render_layered_diagram(
1693 layers,
1694 edges,
1695 main_edge_set,
1696 branches,
1697 back_edges,
1698 fsm.initial,
1699 terminal_states,
1700 fsm.states,
1701 verbose,
1702 highlight_state,
1703 highlight_color,
1704 edge_label_colors,
1705 badges,
1706 title_only=title_only or (mode == "mini"),
1707 suppress_labels=suppress_labels or (mode == "mini"),
1708 )
1711_PREV_STATE_COLOR = "33" # ANSI orange/yellow border for the just-prior FSM state.
1714def _render_neighborhood_diagram(
1715 fsm: FSMLoop,
1716 active_state: str,
1717 *,
1718 edge_label_colors: dict[str, str] | None = None,
1719 badges: dict[str, str] | None = None,
1720 highlight_color: str = "32",
1721 mode: str = "full",
1722 prev_state: str | None = None,
1723) -> str:
1724 """Render a compact 1-hop neighborhood: predecessors → [active] → successors.
1726 Suitable as a fallback when the full FSM diagram does not fit the viewport.
1727 Bounded: ``max(len(preds), len(succs), 1) * 3`` rows (each state box is 3
1728 lines tall). Returns the empty string when ``active_state`` is not in
1729 ``fsm.states``.
1731 Self-loops are collapsed: a state that only points to itself contributes
1732 neither predecessors nor successors here.
1734 Args:
1735 mode: ``"full"`` (default) includes every edge. ``"main"`` filters edges
1736 through ``_filter_main_path_graph`` so off-happy-path predecessors
1737 (e.g. those connected only via ``on_error``) are hidden. Falls back
1738 to ``"full"`` if the active state would be filtered out.
1739 prev_state: Name of the predecessor the FSM most recently transitioned
1740 from. When that name appears in the rendered pred stack, its box is
1741 drawn with the orange ``_PREV_STATE_COLOR`` border. Silently
1742 skipped if the name is missing from the pred stack.
1743 """
1744 if active_state not in fsm.states:
1745 return ""
1747 edges = _collect_edges(fsm)
1748 if mode == "main":
1749 filtered_edges, reachable = _filter_main_path_graph(fsm, edges)
1750 if active_state in reachable:
1751 edges = filtered_edges
1752 preds = sorted({s for (s, t, _lbl) in edges if t == active_state and s != active_state})
1753 succs = sorted({t for (s, t, _lbl) in edges if s == active_state and t != active_state})
1755 terminal_states = {n for n, st in fsm.states.items() if st.terminal}
1757 def _label(name: str) -> str:
1758 label = name
1759 if name == fsm.initial:
1760 label = "→ " + label
1761 if name in terminal_states:
1762 label = label + " ◉"
1763 return label
1765 pred_labels = [_label(p) for p in preds]
1766 active_label = _label(active_state)
1767 succ_labels = [_label(s) for s in succs]
1769 inner_pred = max((len(lbl) for lbl in pred_labels), default=0)
1770 inner_active = len(active_label)
1771 inner_succ = max((len(lbl) for lbl in succ_labels), default=0)
1773 box_w_pred = inner_pred + 4 if pred_labels else 0
1774 box_w_active = inner_active + 4
1775 box_w_succ = inner_succ + 4 if succ_labels else 0
1777 n_rows = max(len(pred_labels), len(succ_labels), 1)
1778 try:
1779 nd_bg_code: str | None = str(int(highlight_color) + 10)
1780 except (ValueError, TypeError):
1781 nd_bg_code = None
1783 def _make_box(
1784 label: str,
1785 inner_w: int,
1786 highlighted: bool,
1787 *,
1788 border_color: str | None = None,
1789 ) -> list[str]:
1790 top = "┌" + "─" * (inner_w + 2) + "┐"
1791 bot = "└" + "─" * (inner_w + 2) + "┘"
1792 padded = label.ljust(inner_w)
1793 if highlighted:
1794 border_code = f"{highlight_color};{nd_bg_code}" if nd_bg_code else highlight_color
1795 top = colorize(top, border_code)
1796 bot = colorize(bot, border_code)
1797 if nd_bg_code:
1798 mid = (
1799 colorize("│", border_code)
1800 + colorize(" ", nd_bg_code)
1801 + colorize(padded, f"97;{nd_bg_code};1")
1802 + colorize(" ", nd_bg_code)
1803 + colorize("│", border_code)
1804 )
1805 else:
1806 mid = (
1807 colorize("│", border_code)
1808 + " "
1809 + colorize(padded, f"{highlight_color};1")
1810 + " "
1811 + colorize("│", border_code)
1812 )
1813 elif border_color is not None:
1814 top = colorize(top, border_color)
1815 bot = colorize(bot, border_color)
1816 mid = (
1817 colorize("│", border_color)
1818 + " "
1819 + colorize(padded, "1")
1820 + " "
1821 + colorize("│", border_color)
1822 )
1823 else:
1824 mid = "│ " + colorize(padded, "1") + " │"
1825 return [top, mid, bot]
1827 center_idx = (n_rows - 1) // 2
1829 def _build_stack(
1830 labels: list[str],
1831 box_w: int,
1832 *,
1833 color_for: dict[str, str] | None = None,
1834 ) -> list[str]:
1835 rows: list[str] = []
1836 # Align the stack to the arrow row (the active state's slot). Without
1837 # this, a shorter stack always sits at row 0 and the single ``──▶``
1838 # arrow drawn at ``active_line_offset`` points into empty space.
1839 # ``min(center_idx, n_rows - len(labels))`` caps the start so longer
1840 # but still-smaller stacks (e.g. 4 succs vs. 5 preds) don't overflow.
1841 start = max(0, min(center_idx, n_rows - len(labels)))
1842 color_map = color_for or {}
1843 for i in range(n_rows):
1844 j = i - start
1845 if 0 <= j < len(labels):
1846 border = color_map.get(labels[j])
1847 rows.extend(_make_box(labels[j], box_w - 4, False, border_color=border))
1848 else:
1849 rows.extend([" " * box_w] * 3)
1850 return rows
1852 pred_color_for: dict[str, str] = {}
1853 if prev_state is not None and prev_state in preds:
1854 pred_color_for[_label(prev_state)] = _PREV_STATE_COLOR
1856 pred_col = (
1857 _build_stack(pred_labels, box_w_pred, color_for=pred_color_for) if pred_labels else None
1858 )
1859 succ_col = _build_stack(succ_labels, box_w_succ) if succ_labels else None
1860 active_rows: list[str] = []
1861 for i in range(n_rows):
1862 if i == center_idx:
1863 active_rows.extend(_make_box(active_label, inner_active, True))
1864 else:
1865 active_rows.extend([" " * box_w_active] * 3)
1867 arrow = " ──▶ "
1868 arrow_blank = " " * len(arrow)
1869 active_line_offset = center_idx * 3 + 1
1871 total_lines = n_rows * 3
1872 out_lines: list[str] = []
1873 for i in range(total_lines):
1874 parts: list[str] = []
1875 if pred_col is not None:
1876 parts.append(pred_col[i])
1877 parts.append(arrow if i == active_line_offset else arrow_blank)
1878 parts.append(active_rows[i])
1879 if succ_col is not None:
1880 parts.append(arrow if i == active_line_offset else arrow_blank)
1881 parts.append(succ_col[i])
1882 out_lines.append("".join(parts).rstrip())
1884 return "\n".join(out_lines)
1887def _render_horizontal_simple(
1888 main_path: list[str],
1889 edges: list[tuple[str, str, str]],
1890 main_edge_set: set[tuple[str, str]],
1891 branches: list[tuple[str, str, str]],
1892 back_edges: list[tuple[str, str, str]],
1893 bfs_order: list[str],
1894 initial: str,
1895 terminal_states: set[str],
1896 fsm_states: dict[str, StateConfig],
1897 verbose: bool,
1898 highlight_state: str | None,
1899 highlight_color: str,
1900 edge_label_colors: dict[str, str] | None = None,
1901 badges: dict[str, str] | None = None,
1902 title_only: bool = False,
1903 suppress_labels: bool = False,
1904) -> str:
1905 """Simple horizontal rendering for single-state or very simple FSMs.
1907 When ``title_only`` is True, per-state body lines and self-loop labels are suppressed.
1908 When ``suppress_labels`` is True, self-loop markers omit label text.
1909 """
1910 if not main_path:
1911 return ""
1912 all_states = list(main_path)
1913 display_label = _compute_display_labels(all_states, initial, terminal_states)
1915 tw = terminal_width()
1916 num_main = max(1, len(main_path))
1917 if verbose and fsm_states and main_path:
1918 max_box_inner = max(20, min(60, (tw - 4) // num_main - 6))
1919 else:
1920 max_box_inner = max(20, min(40, (tw - 4) // num_main - 6))
1922 box_inner, box_width, box_height, box_badge = _compute_box_sizes(
1923 all_states,
1924 display_label,
1925 fsm_states,
1926 verbose,
1927 max_box_inner,
1928 badges,
1929 title_only=title_only,
1930 )
1932 main_height = max((box_height[s] for s in main_path), default=3)
1933 total_width = tw
1935 # Column positions
1936 col_start: dict[str, int] = {}
1937 col_center: dict[str, int] = {}
1938 x = 2
1939 for i, sname in enumerate(main_path):
1940 col_start[sname] = x
1941 col_center[sname] = x + box_width[sname] // 2
1942 x += box_width[sname]
1943 if i < len(main_path) - 1:
1944 x += 4
1946 rows: list[list[str]] = [[" "] * total_width for _ in range(main_height)]
1948 for sname in main_path:
1949 is_highlighted = highlight_state is not None and sname == highlight_state
1950 _draw_box(
1951 rows,
1952 0,
1953 col_start[sname],
1954 box_width[sname],
1955 main_height,
1956 box_inner[sname],
1957 is_highlighted,
1958 highlight_color,
1959 badge=box_badge[sname],
1960 )
1962 # Self-loops
1963 self_loops_list = [(s, d, lbl) for s, d, lbl in back_edges if s == d]
1964 lines = ["".join(row).rstrip() for row in rows]
1965 if self_loops_list:
1966 self_labels: dict[str, list[str]] = {}
1967 for src, _, label in self_loops_list:
1968 self_labels.setdefault(src, []).append(label)
1969 for sname, labels in self_labels.items():
1970 marker = "\u21ba" if suppress_labels else "\u21ba " + ", ".join(labels)
1971 self_row = [" "] * total_width
1972 cx = col_center.get(sname, 0)
1973 pos = max(0, cx - len(marker) // 2)
1974 for j, ch in enumerate(marker):
1975 if pos + j < total_width:
1976 self_row[pos + j] = ch
1977 lines.append("".join(self_row).rstrip())
1979 diagram_indent = max(0, (tw - (x + 4)) // 2)
1980 if diagram_indent > 0:
1981 lines = [" " * diagram_indent + ln if ln.strip() else ln for ln in lines]
1983 return _colorize_diagram_labels("\n".join(lines), edge_label_colors)