Coverage for little_loops / fsm / route_table.py: 0%
374 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-29 00:54 -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
15from little_loops.fsm.policy_rules import _ALL_OPS
17if TYPE_CHECKING:
18 from little_loops.fsm.policy_rules import Rule
19 from little_loops.fsm.schema import FSMLoop
21# Ordered standard verdict labels for column display
22_STANDARD_VERDICTS = ["yes", "no", "error", "partial", "blocked", "next", "default"]
24# Mapping from verdict label → YAML field name (shorthand mode)
25_VERDICT_TO_FIELD: dict[str, str] = {
26 "yes": "on_yes",
27 "no": "on_no",
28 "error": "on_error",
29 "partial": "on_partial",
30 "blocked": "on_blocked",
31 "next": "next",
32}
34EMPTY_CELL = "—"
37@dataclass
38class ParsedTable:
39 """Result of parsing an edited route table."""
41 matrix: dict[str, dict[str, str]]
42 new_stubs: list[str]
43 deleted_states: list[str]
46class RouteTableExtractor:
47 """Extract FSM routing into a state×verdict matrix."""
49 @staticmethod
50 def extract(fsm: FSMLoop) -> dict[str, dict[str, str]]:
51 """Return {state: {verdict: target}} for all states in fsm."""
52 matrix: dict[str, dict[str, str]] = {}
53 for name, state in fsm.states.items():
54 row: dict[str, str] = {}
55 # Shorthand fields
56 for verdict, target in [
57 ("yes", state.on_yes),
58 ("no", state.on_no),
59 ("error", state.on_error),
60 ("partial", state.on_partial),
61 ("blocked", state.on_blocked),
62 ("next", state.next),
63 ]:
64 if target:
65 row[verdict] = target
66 # route: block fields (may overlap shorthands; route takes precedence at runtime)
67 if state.route:
68 for verdict, target in state.route.routes.items():
69 row[verdict] = target
70 if state.route.default:
71 row["default"] = state.route.default
72 # extra_routes: on_<custom> → stored with prefix stripped
73 for verdict, target in state.extra_routes.items():
74 row[verdict] = target
75 matrix[name] = row
76 return matrix
79def _all_verdicts(matrix: dict[str, dict[str, str]]) -> list[str]:
80 """Return ordered verdict columns: standard first, then custom alpha-sorted."""
81 seen: set[str] = set()
82 for row in matrix.values():
83 seen.update(row.keys())
84 standard = [v for v in _STANDARD_VERDICTS if v in seen]
85 extra = sorted(seen - set(_STANDARD_VERDICTS))
86 return standard + extra
89class RouteTableRenderer:
90 """Render a state×verdict matrix as markdown or CSV."""
92 @staticmethod
93 def to_markdown(matrix: dict[str, dict[str, str]]) -> str:
94 verdicts = _all_verdicts(matrix)
95 headers = ["state"] + verdicts
96 rows = [
97 [state] + [row.get(v, EMPTY_CELL) for v in verdicts] for state, row in matrix.items()
98 ]
100 # Column widths
101 widths = [max(len(str(r[i])) for r in ([headers] + rows)) for i in range(len(headers))]
103 def fmt(cells: list[str]) -> str:
104 return "| " + " | ".join(str(c).ljust(widths[i]) for i, c in enumerate(cells)) + " |"
106 lines = [
107 fmt(headers),
108 "|" + "|".join("-" * (w + 2) for w in widths) + "|",
109 ]
110 lines.extend(fmt(row) for row in rows)
111 return "\n".join(lines) + "\n"
113 @staticmethod
114 def to_csv(matrix: dict[str, dict[str, str]]) -> str:
115 verdicts = _all_verdicts(matrix)
116 buf = io.StringIO()
117 writer = csv.DictWriter(buf, fieldnames=["state"] + verdicts)
118 writer.writeheader()
119 for state_name, row in matrix.items():
120 writer.writerow({"state": state_name, **{v: row.get(v, "") for v in verdicts}})
121 return buf.getvalue()
124class RouteTableParser:
125 """Parse an edited markdown or CSV table back to a state×verdict matrix."""
127 @staticmethod
128 def parse_markdown(text: str, known_states: set[str]) -> ParsedTable:
129 """Parse a markdown table into ParsedTable.
131 Unknown state names with all-empty verdict cells are classified as new_stubs.
132 Unknown state names with non-empty cells raise ValueError.
133 """
134 lines = [ln.strip() for ln in text.splitlines() if ln.strip().startswith("|")]
135 if len(lines) < 2:
136 raise ValueError("No valid markdown table found in input")
138 header_cells = [c.strip() for c in lines[0].strip("|").split("|")]
139 verdicts = header_cells[1:] # skip "state" column
141 result: dict[str, dict[str, str]] = {}
142 new_stubs: list[str] = []
143 for line in lines[2:]: # skip header + separator
144 cells = [c.strip() for c in line.strip("|").split("|")]
145 if not cells or not cells[0].strip():
146 continue
147 state_name = cells[0].strip()
148 row: dict[str, str] = {}
149 for i, verdict in enumerate(verdicts):
150 if i + 1 < len(cells):
151 val = cells[i + 1].strip()
152 if val and val not in (EMPTY_CELL, "-", ""):
153 row[verdict] = val
154 if state_name not in known_states:
155 if not row:
156 new_stubs.append(state_name)
157 else:
158 raise ValueError(
159 f"Unknown state in edited table: '{state_name}' "
160 f"(known: {sorted(known_states)})"
161 )
162 else:
163 result[state_name] = row
164 deleted_states = [s for s in known_states if s not in result]
165 return ParsedTable(matrix=result, new_stubs=new_stubs, deleted_states=deleted_states)
167 @staticmethod
168 def parse_csv(text: str, known_states: set[str]) -> ParsedTable:
169 """Parse a CSV table into ParsedTable.
171 Unknown state names with all-empty verdict cells are classified as new_stubs.
172 Unknown state names with non-empty cells raise ValueError.
173 """
174 reader = csv.DictReader(io.StringIO(text))
175 result: dict[str, dict[str, str]] = {}
176 new_stubs: list[str] = []
177 for csv_row in reader:
178 state_name = csv_row.pop("state", "").strip()
179 if not state_name:
180 continue
181 row = {k.strip(): v for k, v in csv_row.items() if v and v not in (EMPTY_CELL, "-")}
182 if state_name not in known_states:
183 if not row:
184 new_stubs.append(state_name)
185 else:
186 raise ValueError(
187 f"Unknown state in edited table: '{state_name}' "
188 f"(known: {sorted(known_states)})"
189 )
190 else:
191 result[state_name] = row
192 deleted_states = [s for s in known_states if s not in result]
193 return ParsedTable(matrix=result, new_stubs=new_stubs, deleted_states=deleted_states)
196class RouteTableApplier:
197 """Apply a matrix diff back to a loop YAML file, preserving non-route content."""
199 @staticmethod
200 def apply(
201 path: Path,
202 old_matrix: dict[str, dict[str, str]],
203 new_matrix: dict[str, dict[str, str]],
204 new_stubs: list[str] | None = None,
205 allow_delete: bool = False,
206 ) -> None:
207 """Write changed routes from new_matrix back to the YAML loop file.
209 Args:
210 path: Path to the loop YAML file.
211 old_matrix: Original state×verdict matrix before editing.
212 new_matrix: Edited state×verdict matrix (excludes deleted states and new stubs).
213 new_stubs: State names to insert as terminal: true stubs.
214 allow_delete: When True, removes states absent from new_matrix from the YAML.
215 When False, emits a warning but leaves absent states unchanged.
216 """
217 from io import StringIO
219 from ruamel.yaml import YAML
220 from ruamel.yaml.comments import CommentedMap
222 from little_loops.file_utils import atomic_write
224 yaml = YAML(typ="rt")
225 data = yaml.load(path)
226 states_data = data.get("states", {})
227 changed = False
229 # Detect deleted states (known states absent from the edited table)
230 deleted = set(old_matrix.keys()) - set(new_matrix.keys())
232 if deleted:
233 if allow_delete:
234 # Warn about dangling routes to deleted states
235 for remaining_state, remaining_routes in new_matrix.items():
236 for verdict, target in remaining_routes.items():
237 if target in deleted:
238 print(
239 f"⚠ Dangling route: '{remaining_state}' routes to "
240 f"deleted state '{target}' via '{verdict}'"
241 )
242 # Remove deleted states from YAML
243 for state_name in deleted:
244 states_data.pop(state_name, None)
245 changed = True
246 else:
247 for state_name in sorted(deleted):
248 print(
249 f"⚠ State '{state_name}' removed from table but "
250 f"--allow-delete not set; skipping deletion"
251 )
253 for state_name, new_row in new_matrix.items():
254 old_row = old_matrix.get(state_name, {})
255 if new_row == old_row:
256 continue
258 state_data = states_data.get(state_name)
259 if state_data is None:
260 continue
262 uses_route_block = "route" in state_data
264 # Apply added/changed verdicts
265 for verdict, new_target in new_row.items():
266 if old_row.get(verdict) == new_target:
267 continue
268 _write_route_field(state_data, verdict, new_target, uses_route_block)
269 changed = True
271 # Remove deleted verdicts
272 for verdict in old_row:
273 if verdict not in new_row:
274 _clear_route_field(state_data, verdict, uses_route_block)
275 changed = True
277 # Insert new terminal stubs
278 if new_stubs:
279 for stub_name in new_stubs:
280 states_data[stub_name] = CommentedMap({"terminal": True})
281 changed = True
283 if changed:
284 buf = StringIO()
285 yaml.dump(data, buf)
286 atomic_write(path, buf.getvalue())
289def _write_route_field(
290 state_data: Any,
291 verdict: str,
292 target: str,
293 uses_route_block: bool,
294) -> None:
295 """Write a single verdict→target mapping into the ruamel state CommentedMap."""
296 if verdict == "next":
297 state_data["next"] = target
298 return
299 if uses_route_block:
300 if "route" not in state_data:
301 state_data["route"] = {}
302 if verdict == "default":
303 state_data["route"]["_"] = target
304 else:
305 state_data["route"][verdict] = target
306 else:
307 field = _VERDICT_TO_FIELD.get(verdict)
308 if field:
309 state_data[field] = target
310 else:
311 # Extra route: stored as on_<verdict> in YAML
312 state_data[f"on_{verdict}"] = target
315def _clear_route_field(
316 state_data: Any,
317 verdict: str,
318 uses_route_block: bool,
319) -> None:
320 """Remove a verdict from the ruamel state CommentedMap."""
321 if verdict == "next":
322 state_data.pop("next", None)
323 return
324 if uses_route_block:
325 route = state_data.get("route", {})
326 if verdict == "default":
327 route.pop("_", None)
328 else:
329 route.pop(verdict, None)
330 else:
331 field = _VERDICT_TO_FIELD.get(verdict)
332 if field:
333 state_data.pop(field, None)
334 else:
335 state_data.pop(f"on_{verdict}", None)
338# ---------------------------------------------------------------------------
339# Compound decision-table types and helpers (ENH-2233)
340# ---------------------------------------------------------------------------
343@dataclass
344class ParsedDecisionTable:
345 """Result of parsing an edited compound policy-rule decision table."""
347 rules: list[Rule]
348 warnings: list[str] = field(default_factory=list)
351def _all_dims(rules: list[Rule]) -> list[str]:
352 """Return sorted union of dimension names from all non-catch-all rules."""
353 seen: set[str] = set()
354 for rule in rules:
355 for pred in rule.predicates:
356 seen.add(pred.dim)
357 return sorted(seen)
360def _detect_shadows(rules: list[Rule]) -> list[str]:
361 """Return warning strings for rules that are subsumed by an earlier rule."""
362 warnings: list[str] = []
363 for i, later in enumerate(rules):
364 if later.is_catchall:
365 continue
366 later_preds = {(p.dim, p.op, p.value) for p in later.predicates}
367 for j, earlier in enumerate(rules[:i]):
368 if earlier.is_catchall:
369 warnings.append(
370 f"Rule {i + 1} (→ {later.target}) is shadowed by catch-all rule {j + 1}"
371 )
372 break
373 earlier_preds = {(p.dim, p.op, p.value) for p in earlier.predicates}
374 if earlier_preds and earlier_preds.issubset(later_preds):
375 warnings.append(
376 f"Rule {i + 1} (→ {later.target}) may be shadowed by rule {j + 1} "
377 f"(→ {earlier.target}): earlier rule has fewer/equal constraints"
378 )
379 break
380 return warnings
383# Multi-char operators must precede their single-char prefixes for longest-match
384# (e.g. ">=" before ">"). Sort by length descending; alphabetical order is NOT safe.
385_OP_ALT = "|".join(sorted(_ALL_OPS, key=len, reverse=True))
386# Regex to parse a condition cell like ">=85", "<65", "==true"
387_COND_PATTERN = re.compile(rf"^({_OP_ALT})(.+)$")
390class PolicyRuleExtractor:
391 """Extract policy rules from a loop's context.policy_rules field."""
393 @staticmethod
394 def extract(fsm: FSMLoop) -> list[Rule]:
395 """Return parsed list of Rule from fsm.context['policy_rules']."""
396 from little_loops.fsm.policy_rules import parse_rules
398 text = str(fsm.context.get("policy_rules", ""))
399 return parse_rules(text)
402class CompoundGridRenderer:
403 """Render a policy rule list as a compound condition×action grid."""
405 @staticmethod
406 def to_markdown(rules: list[Rule]) -> str:
407 """Render rules as a markdown table with condition columns."""
408 dims = _all_dims(rules)
409 headers = ["#"] + dims + ["→ action"]
410 rows: list[list[str]] = []
411 for i, rule in enumerate(rules, 1):
412 if rule.is_catchall:
413 row = [str(i)] + ["*"] * len(dims) + [rule.target]
414 else:
415 pred_map = {p.dim: f"{p.op}{p.value}" for p in rule.predicates}
416 row = [str(i)] + [pred_map.get(d, EMPTY_CELL) for d in dims] + [rule.target]
417 rows.append(row)
419 widths = [max(len(str(r[i])) for r in ([headers] + rows)) for i in range(len(headers))]
421 def fmt(cells: list[str]) -> str:
422 return "| " + " | ".join(str(c).ljust(widths[i]) for i, c in enumerate(cells)) + " |"
424 lines = [
425 fmt(headers),
426 "|" + "|".join("-" * (w + 2) for w in widths) + "|",
427 ]
428 lines.extend(fmt(row) for row in rows)
429 return "\n".join(lines) + "\n"
431 @staticmethod
432 def to_csv(rules: list[Rule]) -> str:
433 """Render rules as CSV with condition columns."""
434 dims = _all_dims(rules)
435 buf = io.StringIO()
436 writer = csv.DictWriter(buf, fieldnames=["#"] + dims + ["→ action"])
437 writer.writeheader()
438 for i, rule in enumerate(rules, 1):
439 if rule.is_catchall:
440 row: dict[str, str] = {"#": str(i), "→ action": rule.target}
441 for d in dims:
442 row[d] = "*"
443 else:
444 pred_map = {p.dim: f"{p.op}{p.value}" for p in rule.predicates}
445 row = {"#": str(i), "→ action": rule.target}
446 for d in dims:
447 row[d] = pred_map.get(d, "")
448 writer.writerow(row)
449 return buf.getvalue()
452def _parse_cond_cell(dim: str, val: str) -> Any:
453 """Parse a condition cell string (e.g. '>=85') into a Predicate."""
454 from little_loops.fsm.policy_rules import Predicate
456 m = _COND_PATTERN.match(val)
457 if not m:
458 raise ValueError(
459 f"Cannot parse condition cell {val!r} for dimension {dim!r}; "
460 f"expected operator prefix ({_OP_ALT.replace('|', ', ')})"
461 )
462 return Predicate(dim=dim, op=m.group(1), value=m.group(2).strip())
465def _parse_rule_cells(
466 dims: list[str], dim_cells: dict[str, str], action: str, known_states: set[str]
467) -> tuple[Any, list[str]]:
468 """Parse dimension cells into a Rule and return (rule, unknown_action_warnings)."""
469 from little_loops.fsm.policy_rules import Predicate
470 from little_loops.fsm.policy_rules import Rule as PolicyRule
472 warnings: list[str] = []
473 if known_states and action not in known_states:
474 warnings.append(f"Action state '{action}' is not a known state in this loop")
476 is_catchall = all(v in ("*", "", EMPTY_CELL, "-") for v in dim_cells.values()) and any(
477 v == "*" for v in dim_cells.values()
478 )
480 if is_catchall:
481 return PolicyRule(predicates=[], target=action, is_catchall=True), warnings
483 preds: list[Predicate] = []
484 for dim in dims:
485 val = dim_cells.get(dim, "").strip()
486 if not val or val in (EMPTY_CELL, "-", ""):
487 continue
488 if val == "*":
489 continue
490 preds.append(_parse_cond_cell(dim, val))
491 return PolicyRule(predicates=preds, target=action, is_catchall=False), warnings
494class CompoundGridParser:
495 """Parse an edited compound grid (markdown or CSV) back to a list of Rule."""
497 @staticmethod
498 def parse_markdown(text: str, known_states: set[str]) -> ParsedDecisionTable:
499 """Parse edited markdown table back to a ParsedDecisionTable.
501 Validates action states, warns on missing catch-all and shadowed rows.
502 """
503 lines = [ln.strip() for ln in text.splitlines() if ln.strip().startswith("|")]
504 if len(lines) < 2:
505 raise ValueError("No valid markdown table found in input")
507 header_cells = [c.strip() for c in lines[0].strip("|").split("|")]
508 if not header_cells or header_cells[-1] not in ("→ action",):
509 raise ValueError(
510 f"Decision table must end with a '→ action' column; got {header_cells[-1]!r}"
511 )
512 dims = header_cells[1:-1]
514 rules = []
515 all_warnings: list[str] = []
516 for line in lines[2:]:
517 cells = [c.strip() for c in line.strip("|").split("|")]
518 if not cells or not cells[0].strip():
519 continue
520 if len(cells) < len(dims) + 2:
521 continue
522 action = cells[-1].strip()
523 if not action or action in (EMPTY_CELL, "-"):
524 continue
525 dim_cells = {dims[k]: cells[k + 1].strip() for k in range(len(dims))}
526 rule, warns = _parse_rule_cells(dims, dim_cells, action, known_states)
527 rules.append(rule)
528 all_warnings.extend(warns)
530 if not rules or not rules[-1].is_catchall:
531 all_warnings.append(
532 "Decision table has no catch-all rule ('*'). "
533 "Unmatched inputs will produce no route."
534 )
535 all_warnings.extend(_detect_shadows(rules))
536 return ParsedDecisionTable(rules=rules, warnings=all_warnings)
538 @staticmethod
539 def parse_csv(text: str, known_states: set[str]) -> ParsedDecisionTable:
540 """Parse edited CSV table back to a ParsedDecisionTable."""
541 reader = csv.DictReader(io.StringIO(text))
542 fieldnames = list(reader.fieldnames or [])
543 if "→ action" not in fieldnames:
544 raise ValueError("CSV must have a '→ action' column")
545 dims = [f for f in fieldnames if f not in ("#", "→ action")]
547 rules = []
548 all_warnings: list[str] = []
549 for csv_row in reader:
550 action = csv_row.get("→ action", "").strip()
551 if not action:
552 continue
553 dim_cells = {d: csv_row.get(d, "").strip() for d in dims}
554 rule, warns = _parse_rule_cells(dims, dim_cells, action, known_states)
555 rules.append(rule)
556 all_warnings.extend(warns)
558 if not rules or not rules[-1].is_catchall:
559 all_warnings.append(
560 "Decision table has no catch-all rule ('*'). "
561 "Unmatched inputs will produce no route."
562 )
563 all_warnings.extend(_detect_shadows(rules))
564 return ParsedDecisionTable(rules=rules, warnings=all_warnings)
567class PolicyRuleApplier:
568 """Write edited policy rules back to a loop YAML file."""
570 @staticmethod
571 def apply(path: Path, rules: list[Rule]) -> None:
572 """Serialize rules and write back to context.policy_rules in the loop YAML."""
573 from io import StringIO
575 from ruamel.yaml import YAML
576 from ruamel.yaml.scalarstring import LiteralScalarString
578 from little_loops.file_utils import atomic_write
579 from little_loops.fsm.policy_rules import serialize_rules
581 yaml = YAML(typ="rt")
582 data = yaml.load(path)
583 serialized = serialize_rules(rules)
584 data["context"]["policy_rules"] = LiteralScalarString(serialized + "\n")
585 buf = StringIO()
586 yaml.dump(data, buf)
587 atomic_write(path, buf.getvalue())
590def detect_routing_gaps(fsm: FSMLoop) -> list[str]:
591 """Detect unreachable states, dead-end states, and missing verdict arms."""
592 from little_loops.fsm.validation import _find_reachable_states
594 warnings: list[str] = []
595 reachable = _find_reachable_states(fsm)
597 for state_name, state in fsm.states.items():
598 if state_name not in reachable:
599 warnings.append(f"Unreachable state: '{state_name}' has no route leading to it")
601 if state.terminal:
602 continue
604 has_any_route = (
605 state.on_yes
606 or state.on_no
607 or state.on_error
608 or state.on_partial
609 or state.on_blocked
610 or state.next
611 or state.route is not None
612 or state.extra_routes
613 )
614 if not has_any_route:
615 warnings.append(
616 f"Dead-end state: '{state_name}' has no outbound routes and is not terminal"
617 )
618 continue
620 # Missing arm: on_yes present but no on_no and no catch-all
621 has_default = (state.route and state.route.default is not None) or state.next is not None
622 if state.on_yes and not state.on_no and not has_default:
623 warnings.append(
624 f"Missing verdict arm: '{state_name}' has on_yes but no on_no or default"
625 )
627 return warnings
630def open_in_editor(table_text: str, fmt: str = "markdown") -> str | None:
631 """Write table to a temp file, open $EDITOR, return edited text (or None on cancel)."""
632 suffix = ".md" if fmt == "markdown" else ".csv"
633 with tempfile.NamedTemporaryFile(
634 mode="w", suffix=suffix, delete=False, encoding="utf-8"
635 ) as tmp:
636 tmp.write(table_text)
637 tmp_path = tmp.name
639 try:
640 editor = os.environ.get("EDITOR", "vi")
641 rc = subprocess.call([editor, tmp_path])
642 if rc != 0:
643 return None
644 return Path(tmp_path).read_text(encoding="utf-8")
645 finally:
646 try:
647 os.unlink(tmp_path)
648 except OSError:
649 pass