Coverage for little_loops / fsm / route_table.py: 0%
372 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-26 17:38 -0500
1"""Route table extraction, rendering, parsing, and application for FSM loops."""
3from __future__ import annotations
5import csv
6import io
7import os
8import re
9import subprocess
10import tempfile
11from dataclasses import dataclass, field
12from pathlib import Path
13from typing import TYPE_CHECKING, Any
15if TYPE_CHECKING:
16 from little_loops.fsm.policy_rules import Rule
17 from little_loops.fsm.schema import FSMLoop
19# Ordered standard verdict labels for column display
20_STANDARD_VERDICTS = ["yes", "no", "error", "partial", "blocked", "next", "default"]
22# Mapping from verdict label → YAML field name (shorthand mode)
23_VERDICT_TO_FIELD: dict[str, str] = {
24 "yes": "on_yes",
25 "no": "on_no",
26 "error": "on_error",
27 "partial": "on_partial",
28 "blocked": "on_blocked",
29 "next": "next",
30}
32EMPTY_CELL = "—"
35@dataclass
36class ParsedTable:
37 """Result of parsing an edited route table."""
39 matrix: dict[str, dict[str, str]]
40 new_stubs: list[str]
41 deleted_states: list[str]
44class RouteTableExtractor:
45 """Extract FSM routing into a state×verdict matrix."""
47 @staticmethod
48 def extract(fsm: FSMLoop) -> dict[str, dict[str, str]]:
49 """Return {state: {verdict: target}} for all states in fsm."""
50 matrix: dict[str, dict[str, str]] = {}
51 for name, state in fsm.states.items():
52 row: dict[str, str] = {}
53 # Shorthand fields
54 for verdict, target in [
55 ("yes", state.on_yes),
56 ("no", state.on_no),
57 ("error", state.on_error),
58 ("partial", state.on_partial),
59 ("blocked", state.on_blocked),
60 ("next", state.next),
61 ]:
62 if target:
63 row[verdict] = target
64 # route: block fields (may overlap shorthands; route takes precedence at runtime)
65 if state.route:
66 for verdict, target in state.route.routes.items():
67 row[verdict] = target
68 if state.route.default:
69 row["default"] = state.route.default
70 # extra_routes: on_<custom> → stored with prefix stripped
71 for verdict, target in state.extra_routes.items():
72 row[verdict] = target
73 matrix[name] = row
74 return matrix
77def _all_verdicts(matrix: dict[str, dict[str, str]]) -> list[str]:
78 """Return ordered verdict columns: standard first, then custom alpha-sorted."""
79 seen: set[str] = set()
80 for row in matrix.values():
81 seen.update(row.keys())
82 standard = [v for v in _STANDARD_VERDICTS if v in seen]
83 extra = sorted(seen - set(_STANDARD_VERDICTS))
84 return standard + extra
87class RouteTableRenderer:
88 """Render a state×verdict matrix as markdown or CSV."""
90 @staticmethod
91 def to_markdown(matrix: dict[str, dict[str, str]]) -> str:
92 verdicts = _all_verdicts(matrix)
93 headers = ["state"] + verdicts
94 rows = [
95 [state] + [row.get(v, EMPTY_CELL) for v in verdicts] for state, row in matrix.items()
96 ]
98 # Column widths
99 widths = [max(len(str(r[i])) for r in ([headers] + rows)) for i in range(len(headers))]
101 def fmt(cells: list[str]) -> str:
102 return "| " + " | ".join(str(c).ljust(widths[i]) for i, c in enumerate(cells)) + " |"
104 lines = [
105 fmt(headers),
106 "|" + "|".join("-" * (w + 2) for w in widths) + "|",
107 ]
108 lines.extend(fmt(row) for row in rows)
109 return "\n".join(lines) + "\n"
111 @staticmethod
112 def to_csv(matrix: dict[str, dict[str, str]]) -> str:
113 verdicts = _all_verdicts(matrix)
114 buf = io.StringIO()
115 writer = csv.DictWriter(buf, fieldnames=["state"] + verdicts)
116 writer.writeheader()
117 for state_name, row in matrix.items():
118 writer.writerow({"state": state_name, **{v: row.get(v, "") for v in verdicts}})
119 return buf.getvalue()
122class RouteTableParser:
123 """Parse an edited markdown or CSV table back to a state×verdict matrix."""
125 @staticmethod
126 def parse_markdown(text: str, known_states: set[str]) -> ParsedTable:
127 """Parse a markdown table into ParsedTable.
129 Unknown state names with all-empty verdict cells are classified as new_stubs.
130 Unknown state names with non-empty cells raise ValueError.
131 """
132 lines = [ln.strip() for ln in text.splitlines() if ln.strip().startswith("|")]
133 if len(lines) < 2:
134 raise ValueError("No valid markdown table found in input")
136 header_cells = [c.strip() for c in lines[0].strip("|").split("|")]
137 verdicts = header_cells[1:] # skip "state" column
139 result: dict[str, dict[str, str]] = {}
140 new_stubs: list[str] = []
141 for line in lines[2:]: # skip header + separator
142 cells = [c.strip() for c in line.strip("|").split("|")]
143 if not cells or not cells[0].strip():
144 continue
145 state_name = cells[0].strip()
146 row: dict[str, str] = {}
147 for i, verdict in enumerate(verdicts):
148 if i + 1 < len(cells):
149 val = cells[i + 1].strip()
150 if val and val not in (EMPTY_CELL, "-", ""):
151 row[verdict] = val
152 if state_name not in known_states:
153 if not row:
154 new_stubs.append(state_name)
155 else:
156 raise ValueError(
157 f"Unknown state in edited table: '{state_name}' "
158 f"(known: {sorted(known_states)})"
159 )
160 else:
161 result[state_name] = row
162 deleted_states = [s for s in known_states if s not in result]
163 return ParsedTable(matrix=result, new_stubs=new_stubs, deleted_states=deleted_states)
165 @staticmethod
166 def parse_csv(text: str, known_states: set[str]) -> ParsedTable:
167 """Parse a CSV table into ParsedTable.
169 Unknown state names with all-empty verdict cells are classified as new_stubs.
170 Unknown state names with non-empty cells raise ValueError.
171 """
172 reader = csv.DictReader(io.StringIO(text))
173 result: dict[str, dict[str, str]] = {}
174 new_stubs: list[str] = []
175 for csv_row in reader:
176 state_name = csv_row.pop("state", "").strip()
177 if not state_name:
178 continue
179 row = {k.strip(): v for k, v in csv_row.items() if v and v not in (EMPTY_CELL, "-")}
180 if state_name not in known_states:
181 if not row:
182 new_stubs.append(state_name)
183 else:
184 raise ValueError(
185 f"Unknown state in edited table: '{state_name}' "
186 f"(known: {sorted(known_states)})"
187 )
188 else:
189 result[state_name] = row
190 deleted_states = [s for s in known_states if s not in result]
191 return ParsedTable(matrix=result, new_stubs=new_stubs, deleted_states=deleted_states)
194class RouteTableApplier:
195 """Apply a matrix diff back to a loop YAML file, preserving non-route content."""
197 @staticmethod
198 def apply(
199 path: Path,
200 old_matrix: dict[str, dict[str, str]],
201 new_matrix: dict[str, dict[str, str]],
202 new_stubs: list[str] | None = None,
203 allow_delete: bool = False,
204 ) -> None:
205 """Write changed routes from new_matrix back to the YAML loop file.
207 Args:
208 path: Path to the loop YAML file.
209 old_matrix: Original state×verdict matrix before editing.
210 new_matrix: Edited state×verdict matrix (excludes deleted states and new stubs).
211 new_stubs: State names to insert as terminal: true stubs.
212 allow_delete: When True, removes states absent from new_matrix from the YAML.
213 When False, emits a warning but leaves absent states unchanged.
214 """
215 from io import StringIO
217 from ruamel.yaml import YAML
218 from ruamel.yaml.comments import CommentedMap
220 from little_loops.file_utils import atomic_write
222 yaml = YAML(typ="rt")
223 data = yaml.load(path)
224 states_data = data.get("states", {})
225 changed = False
227 # Detect deleted states (known states absent from the edited table)
228 deleted = set(old_matrix.keys()) - set(new_matrix.keys())
230 if deleted:
231 if allow_delete:
232 # Warn about dangling routes to deleted states
233 for remaining_state, remaining_routes in new_matrix.items():
234 for verdict, target in remaining_routes.items():
235 if target in deleted:
236 print(
237 f"⚠ Dangling route: '{remaining_state}' routes to "
238 f"deleted state '{target}' via '{verdict}'"
239 )
240 # Remove deleted states from YAML
241 for state_name in deleted:
242 states_data.pop(state_name, None)
243 changed = True
244 else:
245 for state_name in sorted(deleted):
246 print(
247 f"⚠ State '{state_name}' removed from table but "
248 f"--allow-delete not set; skipping deletion"
249 )
251 for state_name, new_row in new_matrix.items():
252 old_row = old_matrix.get(state_name, {})
253 if new_row == old_row:
254 continue
256 state_data = states_data.get(state_name)
257 if state_data is None:
258 continue
260 uses_route_block = "route" in state_data
262 # Apply added/changed verdicts
263 for verdict, new_target in new_row.items():
264 if old_row.get(verdict) == new_target:
265 continue
266 _write_route_field(state_data, verdict, new_target, uses_route_block)
267 changed = True
269 # Remove deleted verdicts
270 for verdict in old_row:
271 if verdict not in new_row:
272 _clear_route_field(state_data, verdict, uses_route_block)
273 changed = True
275 # Insert new terminal stubs
276 if new_stubs:
277 for stub_name in new_stubs:
278 states_data[stub_name] = CommentedMap({"terminal": True})
279 changed = True
281 if changed:
282 buf = StringIO()
283 yaml.dump(data, buf)
284 atomic_write(path, buf.getvalue())
287def _write_route_field(
288 state_data: Any,
289 verdict: str,
290 target: str,
291 uses_route_block: bool,
292) -> None:
293 """Write a single verdict→target mapping into the ruamel state CommentedMap."""
294 if verdict == "next":
295 state_data["next"] = target
296 return
297 if uses_route_block:
298 if "route" not in state_data:
299 state_data["route"] = {}
300 if verdict == "default":
301 state_data["route"]["_"] = target
302 else:
303 state_data["route"][verdict] = target
304 else:
305 field = _VERDICT_TO_FIELD.get(verdict)
306 if field:
307 state_data[field] = target
308 else:
309 # Extra route: stored as on_<verdict> in YAML
310 state_data[f"on_{verdict}"] = target
313def _clear_route_field(
314 state_data: Any,
315 verdict: str,
316 uses_route_block: bool,
317) -> None:
318 """Remove a verdict from the ruamel state CommentedMap."""
319 if verdict == "next":
320 state_data.pop("next", None)
321 return
322 if uses_route_block:
323 route = state_data.get("route", {})
324 if verdict == "default":
325 route.pop("_", None)
326 else:
327 route.pop(verdict, None)
328 else:
329 field = _VERDICT_TO_FIELD.get(verdict)
330 if field:
331 state_data.pop(field, None)
332 else:
333 state_data.pop(f"on_{verdict}", None)
336# ---------------------------------------------------------------------------
337# Compound decision-table types and helpers (ENH-2233)
338# ---------------------------------------------------------------------------
341@dataclass
342class ParsedDecisionTable:
343 """Result of parsing an edited compound policy-rule decision table."""
345 rules: list[Rule]
346 warnings: list[str] = field(default_factory=list)
349def _all_dims(rules: list[Rule]) -> list[str]:
350 """Return sorted union of dimension names from all non-catch-all rules."""
351 seen: set[str] = set()
352 for rule in rules:
353 for pred in rule.predicates:
354 seen.add(pred.dim)
355 return sorted(seen)
358def _detect_shadows(rules: list[Rule]) -> list[str]:
359 """Return warning strings for rules that are subsumed by an earlier rule."""
360 warnings: list[str] = []
361 for i, later in enumerate(rules):
362 if later.is_catchall:
363 continue
364 later_preds = {(p.dim, p.op, p.value) for p in later.predicates}
365 for j, earlier in enumerate(rules[:i]):
366 if earlier.is_catchall:
367 warnings.append(
368 f"Rule {i + 1} (→ {later.target}) is shadowed by catch-all rule {j + 1}"
369 )
370 break
371 earlier_preds = {(p.dim, p.op, p.value) for p in earlier.predicates}
372 if earlier_preds and earlier_preds.issubset(later_preds):
373 warnings.append(
374 f"Rule {i + 1} (→ {later.target}) may be shadowed by rule {j + 1} "
375 f"(→ {earlier.target}): earlier rule has fewer/equal constraints"
376 )
377 break
378 return warnings
381# Regex to parse a condition cell like ">=85", "<65", "==true"
382_COND_PATTERN = re.compile(r"^(>=|<=|==|!=|<|>)(.+)$")
385class PolicyRuleExtractor:
386 """Extract policy rules from a loop's context.policy_rules field."""
388 @staticmethod
389 def extract(fsm: FSMLoop) -> list[Rule]:
390 """Return parsed list of Rule from fsm.context['policy_rules']."""
391 from little_loops.fsm.policy_rules import parse_rules
393 text = str(fsm.context.get("policy_rules", ""))
394 return parse_rules(text)
397class CompoundGridRenderer:
398 """Render a policy rule list as a compound condition×action grid."""
400 @staticmethod
401 def to_markdown(rules: list[Rule]) -> str:
402 """Render rules as a markdown table with condition columns."""
403 dims = _all_dims(rules)
404 headers = ["#"] + dims + ["→ action"]
405 rows: list[list[str]] = []
406 for i, rule in enumerate(rules, 1):
407 if rule.is_catchall:
408 row = [str(i)] + ["*"] * len(dims) + [rule.target]
409 else:
410 pred_map = {p.dim: f"{p.op}{p.value}" for p in rule.predicates}
411 row = [str(i)] + [pred_map.get(d, EMPTY_CELL) for d in dims] + [rule.target]
412 rows.append(row)
414 widths = [max(len(str(r[i])) for r in ([headers] + rows)) for i in range(len(headers))]
416 def fmt(cells: list[str]) -> str:
417 return "| " + " | ".join(str(c).ljust(widths[i]) for i, c in enumerate(cells)) + " |"
419 lines = [
420 fmt(headers),
421 "|" + "|".join("-" * (w + 2) for w in widths) + "|",
422 ]
423 lines.extend(fmt(row) for row in rows)
424 return "\n".join(lines) + "\n"
426 @staticmethod
427 def to_csv(rules: list[Rule]) -> str:
428 """Render rules as CSV with condition columns."""
429 dims = _all_dims(rules)
430 buf = io.StringIO()
431 writer = csv.DictWriter(buf, fieldnames=["#"] + dims + ["→ action"])
432 writer.writeheader()
433 for i, rule in enumerate(rules, 1):
434 if rule.is_catchall:
435 row: dict[str, str] = {"#": str(i), "→ action": rule.target}
436 for d in dims:
437 row[d] = "*"
438 else:
439 pred_map = {p.dim: f"{p.op}{p.value}" for p in rule.predicates}
440 row = {"#": str(i), "→ action": rule.target}
441 for d in dims:
442 row[d] = pred_map.get(d, "")
443 writer.writerow(row)
444 return buf.getvalue()
447def _parse_cond_cell(dim: str, val: str) -> Any:
448 """Parse a condition cell string (e.g. '>=85') into a Predicate."""
449 from little_loops.fsm.policy_rules import Predicate
451 m = _COND_PATTERN.match(val)
452 if not m:
453 raise ValueError(
454 f"Cannot parse condition cell {val!r} for dimension {dim!r}; "
455 f"expected operator prefix (>=, <=, ==, !=, <, >)"
456 )
457 return Predicate(dim=dim, op=m.group(1), value=m.group(2).strip())
460def _parse_rule_cells(
461 dims: list[str], dim_cells: dict[str, str], action: str, known_states: set[str]
462) -> tuple[Any, list[str]]:
463 """Parse dimension cells into a Rule and return (rule, unknown_action_warnings)."""
464 from little_loops.fsm.policy_rules import Predicate
465 from little_loops.fsm.policy_rules import Rule as PolicyRule
467 warnings: list[str] = []
468 if known_states and action not in known_states:
469 warnings.append(f"Action state '{action}' is not a known state in this loop")
471 is_catchall = all(v in ("*", "", EMPTY_CELL, "-") for v in dim_cells.values()) and any(
472 v == "*" for v in dim_cells.values()
473 )
475 if is_catchall:
476 return PolicyRule(predicates=[], target=action, is_catchall=True), warnings
478 preds: list[Predicate] = []
479 for dim in dims:
480 val = dim_cells.get(dim, "").strip()
481 if not val or val in (EMPTY_CELL, "-", ""):
482 continue
483 if val == "*":
484 continue
485 preds.append(_parse_cond_cell(dim, val))
486 return PolicyRule(predicates=preds, target=action, is_catchall=False), warnings
489class CompoundGridParser:
490 """Parse an edited compound grid (markdown or CSV) back to a list of Rule."""
492 @staticmethod
493 def parse_markdown(text: str, known_states: set[str]) -> ParsedDecisionTable:
494 """Parse edited markdown table back to a ParsedDecisionTable.
496 Validates action states, warns on missing catch-all and shadowed rows.
497 """
498 lines = [ln.strip() for ln in text.splitlines() if ln.strip().startswith("|")]
499 if len(lines) < 2:
500 raise ValueError("No valid markdown table found in input")
502 header_cells = [c.strip() for c in lines[0].strip("|").split("|")]
503 if not header_cells or header_cells[-1] not in ("→ action",):
504 raise ValueError(
505 f"Decision table must end with a '→ action' column; got {header_cells[-1]!r}"
506 )
507 dims = header_cells[1:-1]
509 rules = []
510 all_warnings: list[str] = []
511 for line in lines[2:]:
512 cells = [c.strip() for c in line.strip("|").split("|")]
513 if not cells or not cells[0].strip():
514 continue
515 if len(cells) < len(dims) + 2:
516 continue
517 action = cells[-1].strip()
518 if not action or action in (EMPTY_CELL, "-"):
519 continue
520 dim_cells = {dims[k]: cells[k + 1].strip() for k in range(len(dims))}
521 rule, warns = _parse_rule_cells(dims, dim_cells, action, known_states)
522 rules.append(rule)
523 all_warnings.extend(warns)
525 if not rules or not rules[-1].is_catchall:
526 all_warnings.append(
527 "Decision table has no catch-all rule ('*'). "
528 "Unmatched inputs will produce no route."
529 )
530 all_warnings.extend(_detect_shadows(rules))
531 return ParsedDecisionTable(rules=rules, warnings=all_warnings)
533 @staticmethod
534 def parse_csv(text: str, known_states: set[str]) -> ParsedDecisionTable:
535 """Parse edited CSV table back to a ParsedDecisionTable."""
536 reader = csv.DictReader(io.StringIO(text))
537 fieldnames = list(reader.fieldnames or [])
538 if "→ action" not in fieldnames:
539 raise ValueError("CSV must have a '→ action' column")
540 dims = [f for f in fieldnames if f not in ("#", "→ action")]
542 rules = []
543 all_warnings: list[str] = []
544 for csv_row in reader:
545 action = csv_row.get("→ action", "").strip()
546 if not action:
547 continue
548 dim_cells = {d: csv_row.get(d, "").strip() for d in dims}
549 rule, warns = _parse_rule_cells(dims, dim_cells, action, known_states)
550 rules.append(rule)
551 all_warnings.extend(warns)
553 if not rules or not rules[-1].is_catchall:
554 all_warnings.append(
555 "Decision table has no catch-all rule ('*'). "
556 "Unmatched inputs will produce no route."
557 )
558 all_warnings.extend(_detect_shadows(rules))
559 return ParsedDecisionTable(rules=rules, warnings=all_warnings)
562class PolicyRuleApplier:
563 """Write edited policy rules back to a loop YAML file."""
565 @staticmethod
566 def apply(path: Path, rules: list[Rule]) -> None:
567 """Serialize rules and write back to context.policy_rules in the loop YAML."""
568 from io import StringIO
570 from ruamel.yaml import YAML
571 from ruamel.yaml.scalarstring import LiteralScalarString
573 from little_loops.file_utils import atomic_write
574 from little_loops.fsm.policy_rules import serialize_rules
576 yaml = YAML(typ="rt")
577 data = yaml.load(path)
578 serialized = serialize_rules(rules)
579 data["context"]["policy_rules"] = LiteralScalarString(serialized + "\n")
580 buf = StringIO()
581 yaml.dump(data, buf)
582 atomic_write(path, buf.getvalue())
585def detect_routing_gaps(fsm: FSMLoop) -> list[str]:
586 """Detect unreachable states, dead-end states, and missing verdict arms."""
587 from little_loops.fsm.validation import _find_reachable_states
589 warnings: list[str] = []
590 reachable = _find_reachable_states(fsm)
592 for state_name, state in fsm.states.items():
593 if state_name not in reachable:
594 warnings.append(f"Unreachable state: '{state_name}' has no route leading to it")
596 if state.terminal:
597 continue
599 has_any_route = (
600 state.on_yes
601 or state.on_no
602 or state.on_error
603 or state.on_partial
604 or state.on_blocked
605 or state.next
606 or state.route is not None
607 or state.extra_routes
608 )
609 if not has_any_route:
610 warnings.append(
611 f"Dead-end state: '{state_name}' has no outbound routes and is not terminal"
612 )
613 continue
615 # Missing arm: on_yes present but no on_no and no catch-all
616 has_default = (state.route and state.route.default is not None) or state.next is not None
617 if state.on_yes and not state.on_no and not has_default:
618 warnings.append(
619 f"Missing verdict arm: '{state_name}' has on_yes but no on_no or default"
620 )
622 return warnings
625def open_in_editor(table_text: str, fmt: str = "markdown") -> str | None:
626 """Write table to a temp file, open $EDITOR, return edited text (or None on cancel)."""
627 suffix = ".md" if fmt == "markdown" else ".csv"
628 with tempfile.NamedTemporaryFile(
629 mode="w", suffix=suffix, delete=False, encoding="utf-8"
630 ) as tmp:
631 tmp.write(table_text)
632 tmp_path = tmp.name
634 try:
635 editor = os.environ.get("EDITOR", "vi")
636 rc = subprocess.call([editor, tmp_path])
637 if rc != 0:
638 return None
639 return Path(tmp_path).read_text(encoding="utf-8")
640 finally:
641 try:
642 os.unlink(tmp_path)
643 except OSError:
644 pass