feat: add dual-mode tree editor with Code Mode, variables, and markdown sync
Implements the full dual-mode tree editor (Plan Phases 1-5): Backend: - JSONB↔Markdown bidirectional serializer/parser with mistune - Markdown validator with line/column error reporting - 3 API endpoints: export-markdown, import-markdown, validate-markdown - Variable extraction/resolution service ([USER_INPUT], [VAR], [SAVE_AS]) - Session variables JSONB column (migration 028) - 39 tree markdown tests + variable service tests (403 total passing) Frontend: - Monaco-based Code Mode with custom Monarch tokenizer and dark theme - Autocomplete for @node_id refs, type values, variable names - Debounced validation (800ms) with inline Monaco error markers - Syntax help panel (absolute overlay, toggleable) - Starter template for new trees with valid cross-references - Bidirectional metadata sync (name/description/category/tags frontmatter) - Synchronous tree→markdown serializer (fixes async race condition) - Pre-save validation blocks save on broken refs or missing tree name - Mode-aware undo/redo: Monaco native in Code Mode, throttled zundo in Flow Mode - Variable prompt modal and frontend resolver for session navigation Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
chihlasm
491
backend/app/services/tree_markdown_parser.py
Normal file
491
backend/app/services/tree_markdown_parser.py
Normal file
@@ -0,0 +1,491 @@
|
||||
"""
|
||||
Markdown → JSONB parser for ResolutionFlow tree structures.
|
||||
|
||||
Parses ResolutionFlow Markdown format (frontmatter-delimited node blocks)
|
||||
back into the recursive tree_structure JSONB dict.
|
||||
"""
|
||||
import re
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any
|
||||
|
||||
|
||||
@dataclass
|
||||
class ParseError:
|
||||
"""A validation/parse error with location info."""
|
||||
line: int
|
||||
column: int
|
||||
message: str
|
||||
severity: str = "error" # 'error' or 'warning'
|
||||
|
||||
|
||||
@dataclass
|
||||
class ParseResult:
|
||||
"""Result of parsing markdown into a tree structure."""
|
||||
tree_structure: dict[str, Any] | None
|
||||
errors: list[ParseError] = field(default_factory=list)
|
||||
metadata: dict[str, Any] | None = None
|
||||
|
||||
|
||||
# Regex patterns
|
||||
FRONTMATTER_RE = re.compile(r"^---\s*$", re.MULTILINE)
|
||||
OPTION_RE = re.compile(
|
||||
r"^-\s*\[([A-Za-z0-9]+)\]\s*(.+?)(?:\s*→\s*@(\S+))?\s*$"
|
||||
)
|
||||
NEXT_NODE_RE = re.compile(r"^→\s*@(\S+)\s*$")
|
||||
EXPECTED_RE = re.compile(r"^\*\*Expected:\*\*\s*(.+)$")
|
||||
HEADING1_RE = re.compile(r"^#\s+(.+)$")
|
||||
HEADING2_RE = re.compile(r"^##\s+(.+)$")
|
||||
BLOCKQUOTE_RE = re.compile(r"^>\s*(.*)$")
|
||||
ORDERED_LIST_RE = re.compile(r"^\d+\.\s+(.+)$")
|
||||
COMMAND_BLOCK_START = re.compile(r"^```commands\s*$")
|
||||
COMMAND_BLOCK_END = re.compile(r"^```\s*$")
|
||||
|
||||
|
||||
def parse_markdown_to_tree(markdown: str) -> ParseResult:
|
||||
"""Parse ResolutionFlow markdown into a tree structure JSONB dict.
|
||||
|
||||
Args:
|
||||
markdown: The markdown string to parse.
|
||||
|
||||
Returns:
|
||||
ParseResult with tree_structure, errors, and optional metadata.
|
||||
"""
|
||||
errors: list[ParseError] = []
|
||||
raw_blocks = _split_into_blocks(markdown)
|
||||
|
||||
if not raw_blocks:
|
||||
errors.append(ParseError(line=1, column=1, message="No node blocks found"))
|
||||
return ParseResult(tree_structure=None, errors=errors)
|
||||
|
||||
# Check if the first block is a metadata block (has 'name' but no 'id'/'type')
|
||||
metadata = None
|
||||
node_blocks = raw_blocks
|
||||
first_block_text, _ = raw_blocks[0]
|
||||
meta = _try_parse_metadata_block(first_block_text)
|
||||
if meta is not None:
|
||||
metadata = meta
|
||||
node_blocks = raw_blocks[1:]
|
||||
|
||||
if not node_blocks:
|
||||
errors.append(ParseError(line=1, column=1, message="No node blocks found (only metadata)"))
|
||||
return ParseResult(tree_structure=None, errors=errors, metadata=metadata)
|
||||
|
||||
# Parse each block into a flat node dict
|
||||
flat_nodes: list[dict[str, Any]] = []
|
||||
for block_text, start_line in node_blocks:
|
||||
node, block_errors = _parse_block(block_text, start_line)
|
||||
errors.extend(block_errors)
|
||||
if node:
|
||||
flat_nodes.append(node)
|
||||
|
||||
if not flat_nodes:
|
||||
errors.append(ParseError(line=1, column=1, message="No valid nodes parsed"))
|
||||
return ParseResult(tree_structure=None, errors=errors)
|
||||
|
||||
# Check for duplicate IDs
|
||||
seen_ids: dict[str, int] = {}
|
||||
for node in flat_nodes:
|
||||
nid = node.get("id", "")
|
||||
if nid in seen_ids:
|
||||
errors.append(ParseError(
|
||||
line=node.get("_start_line", 1),
|
||||
column=1,
|
||||
message=f"Duplicate node ID: '{nid}'"
|
||||
))
|
||||
else:
|
||||
seen_ids[nid] = node.get("_start_line", 1)
|
||||
|
||||
# Reconstruct recursive tree from flat nodes
|
||||
tree, reconstruct_errors = _reconstruct_tree(flat_nodes)
|
||||
errors.extend(reconstruct_errors)
|
||||
|
||||
return ParseResult(tree_structure=tree, errors=errors, metadata=metadata)
|
||||
|
||||
|
||||
def _try_parse_metadata_block(block_text: str) -> dict[str, Any] | None:
|
||||
"""Try to parse a block as tree metadata (name, description, category, tags).
|
||||
|
||||
Returns metadata dict if the block contains 'name' but no 'id'/'type'.
|
||||
Returns None if it's a regular node block.
|
||||
"""
|
||||
lines = block_text.split("\n")
|
||||
fm_start = None
|
||||
fm_end = None
|
||||
for i, line in enumerate(lines):
|
||||
if line.strip() == "---":
|
||||
if fm_start is None:
|
||||
fm_start = i
|
||||
else:
|
||||
fm_end = i
|
||||
break
|
||||
|
||||
if fm_start is None or fm_end is None:
|
||||
return None
|
||||
|
||||
fm_data: dict[str, str] = {}
|
||||
for i in range(fm_start + 1, fm_end):
|
||||
line = lines[i].strip()
|
||||
if not line:
|
||||
continue
|
||||
if ":" in line:
|
||||
key, _, value = line.partition(":")
|
||||
fm_data[key.strip()] = value.strip()
|
||||
|
||||
# It's a metadata block if it has 'name' but no 'id' and no 'type'
|
||||
if "name" in fm_data and "id" not in fm_data and "type" not in fm_data:
|
||||
metadata: dict[str, Any] = {"name": fm_data["name"]}
|
||||
if "description" in fm_data:
|
||||
metadata["description"] = fm_data["description"]
|
||||
if "category" in fm_data:
|
||||
metadata["category"] = fm_data["category"]
|
||||
if "tags" in fm_data:
|
||||
tags_str = fm_data["tags"].strip("[]")
|
||||
metadata["tags"] = [t.strip() for t in tags_str.split(",") if t.strip()]
|
||||
return metadata
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def _split_into_blocks(markdown: str) -> list[tuple[str, int]]:
|
||||
"""Split markdown into blocks delimited by --- frontmatter markers.
|
||||
|
||||
Returns list of (block_text, start_line_number) tuples.
|
||||
"""
|
||||
lines = markdown.split("\n")
|
||||
blocks: list[tuple[str, int]] = []
|
||||
|
||||
# Find frontmatter boundaries (--- on its own line)
|
||||
fm_lines: list[int] = []
|
||||
for i, line in enumerate(lines):
|
||||
if line.strip() == "---":
|
||||
fm_lines.append(i)
|
||||
|
||||
# Pair up frontmatter markers: each block starts at a `---` and the
|
||||
# frontmatter ends at the next `---`. The body follows until the
|
||||
# next block's first `---` (or end of file).
|
||||
i = 0
|
||||
while i < len(fm_lines) - 1:
|
||||
start = fm_lines[i]
|
||||
end_fm = fm_lines[i + 1]
|
||||
|
||||
# Find the next block start (or EOF)
|
||||
next_block_start = len(lines)
|
||||
if i + 2 < len(fm_lines):
|
||||
next_block_start = fm_lines[i + 2]
|
||||
|
||||
block_lines = lines[start:next_block_start]
|
||||
block_text = "\n".join(block_lines)
|
||||
blocks.append((block_text, start + 1)) # 1-indexed line number
|
||||
|
||||
i += 2 # Jump to next frontmatter pair
|
||||
|
||||
return blocks
|
||||
|
||||
|
||||
def _parse_block(block_text: str, start_line: int) -> tuple[dict[str, Any] | None, list[ParseError]]:
|
||||
"""Parse a single frontmatter+body block into a node dict."""
|
||||
errors: list[ParseError] = []
|
||||
lines = block_text.split("\n")
|
||||
|
||||
# Extract frontmatter (between first and second ---)
|
||||
fm_start = None
|
||||
fm_end = None
|
||||
for i, line in enumerate(lines):
|
||||
if line.strip() == "---":
|
||||
if fm_start is None:
|
||||
fm_start = i
|
||||
else:
|
||||
fm_end = i
|
||||
break
|
||||
|
||||
if fm_start is None or fm_end is None:
|
||||
errors.append(ParseError(
|
||||
line=start_line, column=1,
|
||||
message="Block missing valid frontmatter delimiters"
|
||||
))
|
||||
return None, errors
|
||||
|
||||
# Parse YAML-like frontmatter (simple key: value)
|
||||
fm_data: dict[str, str] = {}
|
||||
for i in range(fm_start + 1, fm_end):
|
||||
line = lines[i].strip()
|
||||
if not line:
|
||||
continue
|
||||
if ":" in line:
|
||||
key, _, value = line.partition(":")
|
||||
fm_data[key.strip()] = value.strip()
|
||||
|
||||
node_id = fm_data.get("id", "")
|
||||
node_type = fm_data.get("type", "")
|
||||
parent_id = fm_data.get("parent")
|
||||
|
||||
if not node_id:
|
||||
errors.append(ParseError(
|
||||
line=start_line, column=1,
|
||||
message="Node block missing 'id' in frontmatter"
|
||||
))
|
||||
return None, errors
|
||||
|
||||
if node_type not in ("decision", "action", "solution"):
|
||||
errors.append(ParseError(
|
||||
line=start_line, column=1,
|
||||
message=f"Invalid node type: '{node_type}' (must be decision, action, or solution)"
|
||||
))
|
||||
return None, errors
|
||||
|
||||
# Parse body (everything after frontmatter)
|
||||
body_lines = lines[fm_end + 1:]
|
||||
body_text = "\n".join(body_lines)
|
||||
|
||||
node: dict[str, Any] = {
|
||||
"id": node_id,
|
||||
"type": node_type,
|
||||
"_parent_id": parent_id,
|
||||
"_start_line": start_line,
|
||||
}
|
||||
|
||||
if node_type == "decision":
|
||||
_parse_decision_body(body_lines, node, start_line + fm_end + 1, errors)
|
||||
elif node_type == "action":
|
||||
_parse_action_body(body_lines, node, start_line + fm_end + 1, errors)
|
||||
elif node_type == "solution":
|
||||
_parse_solution_body(body_lines, node, start_line + fm_end + 1, errors)
|
||||
|
||||
return node, errors
|
||||
|
||||
|
||||
def _parse_decision_body(
|
||||
lines: list[str],
|
||||
node: dict[str, Any],
|
||||
body_start_line: int,
|
||||
errors: list[ParseError],
|
||||
) -> None:
|
||||
"""Parse the body of a decision node."""
|
||||
question = ""
|
||||
help_text_lines: list[str] = []
|
||||
options: list[dict[str, Any]] = []
|
||||
|
||||
for i, line in enumerate(lines):
|
||||
stripped = line.strip()
|
||||
if not stripped:
|
||||
continue
|
||||
|
||||
# Check for heading (question)
|
||||
m = HEADING1_RE.match(stripped)
|
||||
if m:
|
||||
question = m.group(1).strip()
|
||||
continue
|
||||
|
||||
# Check for blockquote (help_text)
|
||||
m = BLOCKQUOTE_RE.match(stripped)
|
||||
if m:
|
||||
help_text_lines.append(m.group(1))
|
||||
continue
|
||||
|
||||
# Check for option
|
||||
m = OPTION_RE.match(stripped)
|
||||
if m:
|
||||
opt_label = m.group(2).strip()
|
||||
opt_next = m.group(3) or ""
|
||||
options.append({
|
||||
"id": f"opt_{node['id']}_{len(options)}",
|
||||
"label": opt_label,
|
||||
"next_node_id": opt_next,
|
||||
})
|
||||
continue
|
||||
|
||||
node["question"] = question
|
||||
node["help_text"] = "\n".join(help_text_lines) if help_text_lines else ""
|
||||
node["options"] = options
|
||||
node["children"] = []
|
||||
|
||||
|
||||
def _parse_action_body(
|
||||
lines: list[str],
|
||||
node: dict[str, Any],
|
||||
body_start_line: int,
|
||||
errors: list[ParseError],
|
||||
) -> None:
|
||||
"""Parse the body of an action node."""
|
||||
title = ""
|
||||
description_lines: list[str] = []
|
||||
commands: list[str] = []
|
||||
expected_outcome = ""
|
||||
next_node_id = ""
|
||||
in_command_block = False
|
||||
|
||||
for i, line in enumerate(lines):
|
||||
stripped = line.strip()
|
||||
|
||||
# Command block handling
|
||||
if in_command_block:
|
||||
if COMMAND_BLOCK_END.match(stripped):
|
||||
in_command_block = False
|
||||
else:
|
||||
commands.append(line.rstrip())
|
||||
continue
|
||||
|
||||
if COMMAND_BLOCK_START.match(stripped):
|
||||
in_command_block = True
|
||||
continue
|
||||
|
||||
if not stripped:
|
||||
# Blank lines are part of description
|
||||
if title and not expected_outcome and not next_node_id:
|
||||
description_lines.append("")
|
||||
continue
|
||||
|
||||
# Title
|
||||
m = HEADING2_RE.match(stripped)
|
||||
if m:
|
||||
title = m.group(1).strip()
|
||||
continue
|
||||
|
||||
# Expected outcome
|
||||
m = EXPECTED_RE.match(stripped)
|
||||
if m:
|
||||
expected_outcome = m.group(1).strip()
|
||||
continue
|
||||
|
||||
# Next node reference
|
||||
m = NEXT_NODE_RE.match(stripped)
|
||||
if m:
|
||||
next_node_id = m.group(1).strip()
|
||||
continue
|
||||
|
||||
# Everything else is description
|
||||
description_lines.append(stripped)
|
||||
|
||||
# Trim leading and trailing empty lines from description
|
||||
while description_lines and not description_lines[-1].strip():
|
||||
description_lines.pop()
|
||||
while description_lines and not description_lines[0].strip():
|
||||
description_lines.pop(0)
|
||||
|
||||
node["title"] = title
|
||||
node["description"] = "\n".join(description_lines)
|
||||
node["commands"] = commands if commands else []
|
||||
node["expected_outcome"] = expected_outcome
|
||||
node["next_node_id"] = next_node_id
|
||||
node["children"] = []
|
||||
|
||||
|
||||
def _parse_solution_body(
|
||||
lines: list[str],
|
||||
node: dict[str, Any],
|
||||
body_start_line: int,
|
||||
errors: list[ParseError],
|
||||
) -> None:
|
||||
"""Parse the body of a solution node."""
|
||||
title = ""
|
||||
description_lines: list[str] = []
|
||||
resolution_steps: list[str] = []
|
||||
|
||||
for i, line in enumerate(lines):
|
||||
stripped = line.strip()
|
||||
if not stripped:
|
||||
if title:
|
||||
description_lines.append("")
|
||||
continue
|
||||
|
||||
# Title
|
||||
m = HEADING2_RE.match(stripped)
|
||||
if m:
|
||||
title = m.group(1).strip()
|
||||
continue
|
||||
|
||||
# Ordered list item (resolution step)
|
||||
m = ORDERED_LIST_RE.match(stripped)
|
||||
if m:
|
||||
resolution_steps.append(m.group(1).strip())
|
||||
continue
|
||||
|
||||
# Everything else is description
|
||||
description_lines.append(stripped)
|
||||
|
||||
# Trim leading and trailing empty lines
|
||||
while description_lines and not description_lines[-1].strip():
|
||||
description_lines.pop()
|
||||
while description_lines and not description_lines[0].strip():
|
||||
description_lines.pop(0)
|
||||
|
||||
node["title"] = title
|
||||
node["description"] = "\n".join(description_lines)
|
||||
node["resolution_steps"] = resolution_steps
|
||||
node["solution"] = title # solution field required for publishing
|
||||
|
||||
|
||||
def _reconstruct_tree(flat_nodes: list[dict[str, Any]]) -> tuple[dict[str, Any] | None, list[ParseError]]:
|
||||
"""Reconstruct a recursive tree from flat nodes using parent references.
|
||||
|
||||
Returns (tree_structure, errors).
|
||||
"""
|
||||
errors: list[ParseError] = []
|
||||
|
||||
if not flat_nodes:
|
||||
return None, errors
|
||||
|
||||
# Build lookup
|
||||
node_map: dict[str, dict[str, Any]] = {}
|
||||
for node in flat_nodes:
|
||||
nid = node["id"]
|
||||
# Clean node (remove internal fields)
|
||||
clean = {k: v for k, v in node.items() if not k.startswith("_")}
|
||||
if "children" not in clean:
|
||||
clean["children"] = []
|
||||
node_map[nid] = clean
|
||||
|
||||
# Find root (node with no parent)
|
||||
root_id = None
|
||||
for node in flat_nodes:
|
||||
if node.get("_parent_id") is None:
|
||||
if root_id is not None:
|
||||
errors.append(ParseError(
|
||||
line=node.get("_start_line", 1),
|
||||
column=1,
|
||||
message=f"Multiple root nodes found: '{root_id}' and '{node['id']}'",
|
||||
))
|
||||
root_id = node["id"]
|
||||
|
||||
if root_id is None:
|
||||
# Fall back to first node
|
||||
root_id = flat_nodes[0]["id"]
|
||||
errors.append(ParseError(
|
||||
line=1, column=1,
|
||||
message="No root node found (no node without a parent). Using first node as root.",
|
||||
severity="warning"
|
||||
))
|
||||
|
||||
# Build children relationships
|
||||
for node in flat_nodes:
|
||||
parent_id = node.get("_parent_id")
|
||||
if parent_id and parent_id in node_map:
|
||||
child = node_map[node["id"]]
|
||||
node_map[parent_id]["children"].append(child)
|
||||
elif parent_id and parent_id not in node_map:
|
||||
errors.append(ParseError(
|
||||
line=node.get("_start_line", 1),
|
||||
column=1,
|
||||
message=f"Node '{node['id']}' references non-existent parent '{parent_id}'"
|
||||
))
|
||||
|
||||
# Validate option references
|
||||
for nid, node in node_map.items():
|
||||
if node.get("type") == "decision":
|
||||
for opt in node.get("options", []):
|
||||
ref = opt.get("next_node_id", "")
|
||||
if ref and ref not in node_map:
|
||||
errors.append(ParseError(
|
||||
line=1, column=1,
|
||||
message=f"Option '{opt.get('label', '')}' in node '{nid}' references non-existent node '@{ref}'"
|
||||
))
|
||||
elif node.get("type") == "action":
|
||||
ref = node.get("next_node_id", "")
|
||||
if ref and ref not in node_map:
|
||||
errors.append(ParseError(
|
||||
line=1, column=1,
|
||||
message=f"Action node '{nid}' references non-existent next node '@{ref}'"
|
||||
))
|
||||
|
||||
root = node_map.get(root_id)
|
||||
return root, errors
|
||||
157
backend/app/services/tree_markdown_service.py
Normal file
157
backend/app/services/tree_markdown_service.py
Normal file
@@ -0,0 +1,157 @@
|
||||
"""
|
||||
JSONB → Markdown serializer for ResolutionFlow tree structures.
|
||||
|
||||
Converts a tree_structure JSONB dict into the ResolutionFlow Markdown format
|
||||
where each node is a block delimited by YAML frontmatter.
|
||||
"""
|
||||
from typing import Any
|
||||
|
||||
|
||||
def serialize_tree_to_markdown(
|
||||
tree_structure: dict[str, Any],
|
||||
metadata: dict[str, Any] | None = None,
|
||||
) -> str:
|
||||
"""Convert a tree structure JSONB dict to ResolutionFlow Markdown format.
|
||||
|
||||
Args:
|
||||
tree_structure: The recursive tree_structure dict from the database.
|
||||
metadata: Optional tree metadata (name, description, category, tags)
|
||||
to include as a frontmatter block at the top.
|
||||
|
||||
Returns:
|
||||
Markdown string with YAML frontmatter blocks for each node.
|
||||
"""
|
||||
blocks: list[str] = []
|
||||
|
||||
if metadata:
|
||||
meta_lines = ["---"]
|
||||
if metadata.get("name"):
|
||||
meta_lines.append(f"name: {metadata['name']}")
|
||||
if metadata.get("description"):
|
||||
meta_lines.append(f"description: {metadata['description']}")
|
||||
if metadata.get("category"):
|
||||
meta_lines.append(f"category: {metadata['category']}")
|
||||
tags = metadata.get("tags")
|
||||
if tags:
|
||||
tags_str = ", ".join(tags) if isinstance(tags, list) else str(tags)
|
||||
meta_lines.append(f"tags: [{tags_str}]")
|
||||
meta_lines.append("---")
|
||||
blocks.append("\n".join(meta_lines))
|
||||
|
||||
_serialize_node(tree_structure, parent_id=None, blocks=blocks)
|
||||
return "\n\n".join(blocks) + "\n"
|
||||
|
||||
|
||||
def _serialize_node(
|
||||
node: dict[str, Any],
|
||||
parent_id: str | None,
|
||||
blocks: list[str],
|
||||
) -> None:
|
||||
"""Recursively serialize a single node and its children."""
|
||||
node_type = node.get("type", "decision")
|
||||
node_id = node.get("id", "unknown")
|
||||
|
||||
# Build frontmatter
|
||||
frontmatter_lines = [
|
||||
"---",
|
||||
f"id: {node_id}",
|
||||
f"type: {node_type}",
|
||||
]
|
||||
if parent_id is not None:
|
||||
frontmatter_lines.append(f"parent: {parent_id}")
|
||||
frontmatter_lines.append("---")
|
||||
|
||||
# Build body based on node type
|
||||
body_lines: list[str] = []
|
||||
|
||||
if node_type == "decision":
|
||||
_serialize_decision(node, body_lines)
|
||||
elif node_type == "action":
|
||||
_serialize_action(node, body_lines)
|
||||
elif node_type == "solution":
|
||||
_serialize_solution(node, body_lines)
|
||||
|
||||
block = "\n".join(frontmatter_lines) + "\n" + "\n".join(body_lines)
|
||||
blocks.append(block)
|
||||
|
||||
# Recurse into children
|
||||
children = node.get("children", [])
|
||||
if children:
|
||||
for child in children:
|
||||
_serialize_node(child, parent_id=node_id, blocks=blocks)
|
||||
|
||||
|
||||
def _serialize_decision(node: dict[str, Any], lines: list[str]) -> None:
|
||||
"""Serialize a decision node body."""
|
||||
question = node.get("question", "")
|
||||
if question:
|
||||
lines.append(f"# {question}")
|
||||
lines.append("")
|
||||
|
||||
help_text = node.get("help_text", "")
|
||||
if help_text:
|
||||
for ht_line in help_text.split("\n"):
|
||||
lines.append(f"> {ht_line}")
|
||||
lines.append("")
|
||||
|
||||
options = node.get("options", [])
|
||||
for i, opt in enumerate(options):
|
||||
label = opt.get("label", "")
|
||||
next_id = opt.get("next_node_id", "")
|
||||
letter = chr(ord("A") + i) if i < 26 else str(i + 1)
|
||||
if next_id:
|
||||
lines.append(f"- [{letter}] {label} → @{next_id}")
|
||||
else:
|
||||
lines.append(f"- [{letter}] {label}")
|
||||
|
||||
|
||||
def _serialize_action(node: dict[str, Any], lines: list[str]) -> None:
|
||||
"""Serialize an action node body."""
|
||||
title = node.get("title", "")
|
||||
if title:
|
||||
lines.append(f"## {title}")
|
||||
lines.append("")
|
||||
|
||||
description = node.get("description", "")
|
||||
if description:
|
||||
lines.append(description)
|
||||
lines.append("")
|
||||
|
||||
commands = node.get("commands", [])
|
||||
if commands:
|
||||
lines.append("```commands")
|
||||
for cmd in commands:
|
||||
lines.append(cmd)
|
||||
lines.append("```")
|
||||
lines.append("")
|
||||
|
||||
expected = node.get("expected_outcome", "")
|
||||
if expected:
|
||||
lines.append(f"**Expected:** {expected}")
|
||||
lines.append("")
|
||||
|
||||
next_id = node.get("next_node_id", "")
|
||||
if next_id:
|
||||
lines.append(f"→ @{next_id}")
|
||||
|
||||
|
||||
def _serialize_solution(node: dict[str, Any], lines: list[str]) -> None:
|
||||
"""Serialize a solution node body."""
|
||||
title = node.get("title", "")
|
||||
if title:
|
||||
lines.append(f"## {title}")
|
||||
lines.append("")
|
||||
|
||||
description = node.get("description", "")
|
||||
if description:
|
||||
lines.append(description)
|
||||
lines.append("")
|
||||
|
||||
steps = node.get("resolution_steps", [])
|
||||
if steps:
|
||||
for i, step in enumerate(steps, 1):
|
||||
lines.append(f"{i}. {step}")
|
||||
|
||||
solution = node.get("solution", "")
|
||||
if solution and not description and not steps:
|
||||
lines.append(solution)
|
||||
93
backend/app/services/tree_markdown_validator.py
Normal file
93
backend/app/services/tree_markdown_validator.py
Normal file
@@ -0,0 +1,93 @@
|
||||
"""
|
||||
Validation for ResolutionFlow tree markdown.
|
||||
|
||||
Validates markdown without saving, returning detailed errors with line numbers.
|
||||
"""
|
||||
from app.services.tree_markdown_parser import parse_markdown_to_tree, ParseError
|
||||
|
||||
|
||||
def validate_tree_markdown(markdown: str) -> list[ParseError]:
|
||||
"""Validate tree markdown and return all errors/warnings.
|
||||
|
||||
This wraps the parser and adds additional semantic checks.
|
||||
|
||||
Args:
|
||||
markdown: The markdown string to validate.
|
||||
|
||||
Returns:
|
||||
List of ParseError objects with line, column, message, severity.
|
||||
"""
|
||||
result = parse_markdown_to_tree(markdown)
|
||||
errors = list(result.errors)
|
||||
|
||||
# If parsing completely failed, return early
|
||||
if result.tree_structure is None:
|
||||
return errors
|
||||
|
||||
# Additional semantic validation on the parsed tree
|
||||
_validate_tree_semantics(result.tree_structure, errors)
|
||||
|
||||
return errors
|
||||
|
||||
|
||||
def _validate_tree_semantics(tree: dict, errors: list[ParseError]) -> None:
|
||||
"""Run semantic checks on a parsed tree structure."""
|
||||
all_ids: set[str] = set()
|
||||
has_solution = False
|
||||
|
||||
def _collect_ids(node: dict) -> None:
|
||||
nonlocal has_solution
|
||||
all_ids.add(node.get("id", ""))
|
||||
if node.get("type") == "solution":
|
||||
has_solution = True
|
||||
for child in node.get("children", []):
|
||||
_collect_ids(child)
|
||||
|
||||
_collect_ids(tree)
|
||||
|
||||
if not has_solution:
|
||||
errors.append(ParseError(
|
||||
line=1, column=1,
|
||||
message="Tree must have at least one solution node",
|
||||
severity="warning"
|
||||
))
|
||||
|
||||
# Check for empty required fields
|
||||
def _validate_node(node: dict) -> None:
|
||||
ntype = node.get("type", "")
|
||||
nid = node.get("id", "")
|
||||
|
||||
if ntype == "decision":
|
||||
if not node.get("question", "").strip():
|
||||
errors.append(ParseError(
|
||||
line=1, column=1,
|
||||
message=f"Decision node '{nid}' has an empty question",
|
||||
severity="warning"
|
||||
))
|
||||
if not node.get("options"):
|
||||
errors.append(ParseError(
|
||||
line=1, column=1,
|
||||
message=f"Decision node '{nid}' has no options",
|
||||
severity="warning"
|
||||
))
|
||||
|
||||
elif ntype == "action":
|
||||
if not node.get("title", "").strip():
|
||||
errors.append(ParseError(
|
||||
line=1, column=1,
|
||||
message=f"Action node '{nid}' has an empty title",
|
||||
severity="warning"
|
||||
))
|
||||
|
||||
elif ntype == "solution":
|
||||
if not node.get("title", "").strip():
|
||||
errors.append(ParseError(
|
||||
line=1, column=1,
|
||||
message=f"Solution node '{nid}' has an empty title",
|
||||
severity="warning"
|
||||
))
|
||||
|
||||
for child in node.get("children", []):
|
||||
_validate_node(child)
|
||||
|
||||
_validate_node(tree)
|
||||
134
backend/app/services/variable_service.py
Normal file
134
backend/app/services/variable_service.py
Normal file
@@ -0,0 +1,134 @@
|
||||
"""
|
||||
Variable extraction and resolution for ResolutionFlow tree structures.
|
||||
|
||||
Supports three variable tokens:
|
||||
- [USER_INPUT:prompt] — prompts user for input during session navigation
|
||||
- [VAR:name] — references a previously saved variable value
|
||||
- [SAVE_AS:name] — saves the current context as a named variable
|
||||
"""
|
||||
import re
|
||||
from dataclasses import dataclass
|
||||
from typing import Any
|
||||
|
||||
|
||||
@dataclass
|
||||
class VariableDefinition:
|
||||
"""A variable found in a tree structure."""
|
||||
name: str
|
||||
kind: str # 'user_input', 'reference', 'save_as'
|
||||
prompt: str # For user_input: the prompt text; for others: empty
|
||||
node_id: str # The node where this variable appears
|
||||
|
||||
|
||||
# Regex patterns for variable tokens
|
||||
USER_INPUT_RE = re.compile(r'\[USER_INPUT:([^\]]+)\]')
|
||||
VAR_REF_RE = re.compile(r'\[VAR:([^\]]+)\]')
|
||||
SAVE_AS_RE = re.compile(r'\[SAVE_AS:([^\]]+)\]')
|
||||
|
||||
|
||||
def extract_variables(tree_structure: dict[str, Any]) -> list[VariableDefinition]:
|
||||
"""Extract all variable definitions from a tree structure.
|
||||
|
||||
Traverses the tree recursively and finds all [USER_INPUT:...],
|
||||
[VAR:...], and [SAVE_AS:...] tokens.
|
||||
|
||||
Args:
|
||||
tree_structure: The recursive tree_structure dict.
|
||||
|
||||
Returns:
|
||||
List of VariableDefinition objects found in the tree.
|
||||
"""
|
||||
variables: list[VariableDefinition] = []
|
||||
_scan_node(tree_structure, variables)
|
||||
return variables
|
||||
|
||||
|
||||
def _scan_node(node: dict[str, Any], variables: list[VariableDefinition]) -> None:
|
||||
"""Scan a node and its children for variable tokens."""
|
||||
node_id = node.get("id", "unknown")
|
||||
|
||||
# Collect all text fields to scan
|
||||
text_fields = [
|
||||
node.get("question", ""),
|
||||
node.get("title", ""),
|
||||
node.get("description", ""),
|
||||
node.get("help_text", ""),
|
||||
node.get("expected_outcome", ""),
|
||||
node.get("solution", ""),
|
||||
]
|
||||
|
||||
# Include option labels
|
||||
for opt in node.get("options", []):
|
||||
text_fields.append(opt.get("label", ""))
|
||||
|
||||
# Include resolution steps
|
||||
for step in node.get("resolution_steps", []):
|
||||
text_fields.append(step)
|
||||
|
||||
# Include commands
|
||||
for cmd in node.get("commands", []):
|
||||
text_fields.append(cmd)
|
||||
|
||||
# Scan all text fields
|
||||
for text in text_fields:
|
||||
if not text:
|
||||
continue
|
||||
|
||||
for match in USER_INPUT_RE.finditer(text):
|
||||
variables.append(VariableDefinition(
|
||||
name=match.group(1).strip(),
|
||||
kind="user_input",
|
||||
prompt=match.group(1).strip(),
|
||||
node_id=node_id,
|
||||
))
|
||||
|
||||
for match in VAR_REF_RE.finditer(text):
|
||||
variables.append(VariableDefinition(
|
||||
name=match.group(1).strip(),
|
||||
kind="reference",
|
||||
prompt="",
|
||||
node_id=node_id,
|
||||
))
|
||||
|
||||
for match in SAVE_AS_RE.finditer(text):
|
||||
variables.append(VariableDefinition(
|
||||
name=match.group(1).strip(),
|
||||
kind="save_as",
|
||||
prompt="",
|
||||
node_id=node_id,
|
||||
))
|
||||
|
||||
# Recurse into children
|
||||
for child in node.get("children", []):
|
||||
_scan_node(child, variables)
|
||||
|
||||
|
||||
def resolve_variables(text: str, variables: dict[str, str]) -> str:
|
||||
"""Replace variable tokens in text with their resolved values.
|
||||
|
||||
- [VAR:name] → replaced with variables[name] if present
|
||||
- [USER_INPUT:prompt] → replaced with variables[prompt] if present
|
||||
- [SAVE_AS:name] → removed (save directives don't appear in output)
|
||||
|
||||
Args:
|
||||
text: The text containing variable tokens.
|
||||
variables: Dict mapping variable names to their values.
|
||||
|
||||
Returns:
|
||||
Text with variable tokens replaced.
|
||||
"""
|
||||
def _replace_var(match: re.Match) -> str:
|
||||
name = match.group(1).strip()
|
||||
return variables.get(name, f"[VAR:{name}]")
|
||||
|
||||
def _replace_input(match: re.Match) -> str:
|
||||
prompt = match.group(1).strip()
|
||||
return variables.get(prompt, f"[USER_INPUT:{prompt}]")
|
||||
|
||||
def _replace_save(match: re.Match) -> str:
|
||||
return "" # Save directives are removed from output
|
||||
|
||||
result = VAR_REF_RE.sub(_replace_var, text)
|
||||
result = USER_INPUT_RE.sub(_replace_input, result)
|
||||
result = SAVE_AS_RE.sub(_replace_save, result)
|
||||
return result
|
||||
Reference in New Issue
Block a user