Backend features: - Tree sharing via secure tokens with expiration (Issue #16) - Draft tree status with conditional validation (Issue #25) - Save session as custom tree with fork tracking (Issue #17) - Tree validation system for publish requirements - Session-to-tree conversion preserving custom steps Database migrations: - 024: Tree sharing (tree_shares table, visibility field) - 025: Tree status field (draft/published) - 25b: Merge migration for indexes New endpoints: - POST /api/v1/trees/{id}/share - Generate share token - GET /api/v1/shared/{token} - Public tree access - POST /api/v1/trees/{id}/can-publish - Validate tree - POST /api/v1/sessions/{id}/save-as-tree - Convert session Test coverage: - 20 tests for draft trees functionality - 14 tests for session-to-tree conversion - 15 tests for tree sharing Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
This commit is contained in:
Michael Chihlas
206
backend/app/core/session_to_tree.py
Normal file
206
backend/app/core/session_to_tree.py
Normal file
@@ -0,0 +1,206 @@
|
||||
"""Helper module to convert sessions into tree structures."""
|
||||
import uuid
|
||||
from typing import Any
|
||||
|
||||
|
||||
def convert_session_to_tree(
|
||||
session_path: list[str],
|
||||
tree_snapshot: dict[str, Any],
|
||||
custom_steps: list[dict[str, Any]],
|
||||
decisions: list[dict[str, Any]]
|
||||
) -> dict[str, Any]:
|
||||
"""Convert a session's path and custom steps into a linear tree structure.
|
||||
|
||||
Creates a linear decision tree that represents the path taken through the
|
||||
original tree, including any custom steps inserted during the session.
|
||||
|
||||
Args:
|
||||
session_path: List of node IDs representing the path taken
|
||||
tree_snapshot: Original tree structure (for node details)
|
||||
custom_steps: Custom steps inserted during session
|
||||
decisions: Decision records with answers and notes
|
||||
|
||||
Returns:
|
||||
Tree structure dict representing the linear path
|
||||
"""
|
||||
if not session_path:
|
||||
# Return minimal valid tree if no path taken
|
||||
return {
|
||||
"id": str(uuid.uuid4()),
|
||||
"type": "solution",
|
||||
"solution": "Session had no recorded path",
|
||||
"children": []
|
||||
}
|
||||
|
||||
# Build a map of custom steps by their ID
|
||||
custom_steps_map = {}
|
||||
for step in custom_steps:
|
||||
if "id" in step:
|
||||
custom_steps_map[step["id"]] = step
|
||||
|
||||
# Build a map of decisions by node_id for quick lookup
|
||||
decisions_map = {}
|
||||
for decision in decisions:
|
||||
if "node_id" in decision:
|
||||
decisions_map[decision["node_id"]] = decision
|
||||
|
||||
# Build the linear tree structure
|
||||
root_node = None
|
||||
current_node = None
|
||||
|
||||
for i, node_id in enumerate(session_path):
|
||||
# Check if this is a custom step
|
||||
if node_id in custom_steps_map:
|
||||
step = custom_steps_map[node_id]
|
||||
new_node = _create_node_from_custom_step(step, node_id)
|
||||
else:
|
||||
# Find node in original tree
|
||||
original_node = _find_node_in_tree(tree_snapshot, node_id)
|
||||
if original_node:
|
||||
new_node = _create_node_from_original(original_node, decisions_map.get(node_id))
|
||||
else:
|
||||
# Node not found, create a placeholder
|
||||
new_node = {
|
||||
"id": node_id,
|
||||
"type": "action",
|
||||
"action": f"Step from original tree (node {node_id})",
|
||||
"children": []
|
||||
}
|
||||
|
||||
# Add notes from decision if available
|
||||
decision = decisions_map.get(node_id)
|
||||
if decision and decision.get("notes"):
|
||||
new_node["notes"] = decision["notes"]
|
||||
|
||||
# Build the chain
|
||||
if root_node is None:
|
||||
root_node = new_node
|
||||
current_node = root_node
|
||||
else:
|
||||
current_node["children"] = [new_node]
|
||||
current_node = new_node
|
||||
|
||||
return root_node
|
||||
|
||||
|
||||
def _find_node_in_tree(tree: dict[str, Any], node_id: str) -> dict[str, Any] | None:
|
||||
"""Recursively find a node in the tree structure by ID.
|
||||
|
||||
Args:
|
||||
tree: Tree structure dict
|
||||
node_id: Node ID to find
|
||||
|
||||
Returns:
|
||||
Node dict if found, None otherwise
|
||||
"""
|
||||
if tree.get("id") == node_id:
|
||||
return tree
|
||||
|
||||
for child in tree.get("children", []):
|
||||
result = _find_node_in_tree(child, node_id)
|
||||
if result:
|
||||
return result
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def _create_node_from_original(
|
||||
original_node: dict[str, Any],
|
||||
decision: dict[str, Any] | None = None
|
||||
) -> dict[str, Any]:
|
||||
"""Create a new node based on an original tree node.
|
||||
|
||||
Args:
|
||||
original_node: Original node from tree
|
||||
decision: Decision record for this node (optional)
|
||||
|
||||
Returns:
|
||||
New node dict for the linear tree
|
||||
"""
|
||||
node_type = original_node.get("type", "action")
|
||||
new_node = {
|
||||
"id": str(uuid.uuid4()), # Generate new ID for the saved tree
|
||||
"type": node_type,
|
||||
"children": []
|
||||
}
|
||||
|
||||
# Copy relevant content based on node type
|
||||
if node_type == "decision":
|
||||
new_node["question"] = original_node.get("question", "")
|
||||
if decision and decision.get("answer"):
|
||||
new_node["question"] += f"\n\nAnswer: {decision['answer']}"
|
||||
elif node_type == "action":
|
||||
new_node["action"] = original_node.get("action", "")
|
||||
if decision and decision.get("action_performed"):
|
||||
new_node["action"] = decision["action_performed"]
|
||||
elif node_type == "solution":
|
||||
new_node["solution"] = original_node.get("solution", "")
|
||||
|
||||
return new_node
|
||||
|
||||
|
||||
def _create_node_from_custom_step(
|
||||
custom_step: dict[str, Any],
|
||||
step_id: str
|
||||
) -> dict[str, Any]:
|
||||
"""Create a node from a custom step.
|
||||
|
||||
Args:
|
||||
custom_step: Custom step dict
|
||||
step_id: ID of the custom step
|
||||
|
||||
Returns:
|
||||
Node dict for the linear tree
|
||||
"""
|
||||
step_type = custom_step.get("type", "action")
|
||||
content = custom_step.get("content", "")
|
||||
|
||||
new_node = {
|
||||
"id": str(uuid.uuid4()),
|
||||
"type": step_type,
|
||||
"children": []
|
||||
}
|
||||
|
||||
# Map content to appropriate field based on type
|
||||
if step_type == "decision":
|
||||
new_node["question"] = content
|
||||
elif step_type == "action":
|
||||
new_node["action"] = content
|
||||
elif step_type == "solution":
|
||||
new_node["solution"] = content
|
||||
|
||||
# Add notes if present
|
||||
if custom_step.get("notes"):
|
||||
if step_type == "decision":
|
||||
new_node["question"] += f"\n\nNotes: {custom_step['notes']}"
|
||||
elif step_type == "action":
|
||||
new_node["action"] += f"\n\nNotes: {custom_step['notes']}"
|
||||
elif step_type == "solution":
|
||||
new_node["solution"] += f"\n\nNotes: {custom_step['notes']}"
|
||||
|
||||
return new_node
|
||||
|
||||
|
||||
def generate_tree_name_from_session(
|
||||
original_tree_name: str,
|
||||
ticket_number: str | None = None,
|
||||
client_name: str | None = None
|
||||
) -> str:
|
||||
"""Generate a descriptive name for the saved tree.
|
||||
|
||||
Args:
|
||||
original_tree_name: Name of the original tree
|
||||
ticket_number: Optional ticket number
|
||||
client_name: Optional client name
|
||||
|
||||
Returns:
|
||||
Generated tree name
|
||||
"""
|
||||
parts = [original_tree_name, "Session"]
|
||||
|
||||
if ticket_number:
|
||||
parts.append(f"(Ticket {ticket_number})")
|
||||
if client_name:
|
||||
parts.append(f"- {client_name}")
|
||||
|
||||
return " ".join(parts)
|
||||
151
backend/app/core/tree_validation.py
Normal file
151
backend/app/core/tree_validation.py
Normal file
@@ -0,0 +1,151 @@
|
||||
"""Tree validation helper module for draft/published workflow."""
|
||||
from typing import Any
|
||||
|
||||
|
||||
class TreeValidationError(Exception):
|
||||
"""Custom exception for tree validation errors."""
|
||||
def __init__(self, field: str, message: str):
|
||||
self.field = field
|
||||
self.message = message
|
||||
super().__init__(f"{field}: {message}")
|
||||
|
||||
|
||||
def validate_tree_structure(tree_structure: dict[str, Any]) -> tuple[bool, list[dict[str, str]]]:
|
||||
"""Validate tree structure for publishing.
|
||||
|
||||
A valid tree for publishing must have:
|
||||
- A root node with id, type, and appropriate content fields
|
||||
- All decision nodes must have a question field
|
||||
- All decision nodes with children must have at least 2 children
|
||||
- All action nodes must have an action field
|
||||
- All solution nodes must have a solution field
|
||||
- No orphaned nodes (all nodes reachable from root)
|
||||
|
||||
Args:
|
||||
tree_structure: The tree structure dict to validate
|
||||
|
||||
Returns:
|
||||
Tuple of (is_valid, list of errors)
|
||||
Each error is a dict with 'field' and 'message' keys
|
||||
"""
|
||||
errors = []
|
||||
|
||||
# Check root node exists
|
||||
if not tree_structure:
|
||||
errors.append({"field": "tree_structure", "message": "Tree structure cannot be empty"})
|
||||
return False, errors
|
||||
|
||||
if "id" not in tree_structure:
|
||||
errors.append({"field": "tree_structure.id", "message": "Root node must have an id"})
|
||||
|
||||
if "type" not in tree_structure:
|
||||
errors.append({"field": "tree_structure.type", "message": "Root node must have a type"})
|
||||
return False, errors
|
||||
|
||||
# Validate root node based on type
|
||||
_validate_node(tree_structure, "root", errors)
|
||||
|
||||
# Validate all child nodes recursively
|
||||
if "children" in tree_structure:
|
||||
_validate_children(tree_structure["children"], "root.children", errors)
|
||||
|
||||
return len(errors) == 0, errors
|
||||
|
||||
|
||||
def _validate_node(node: dict[str, Any], path: str, errors: list[dict[str, str]]) -> None:
|
||||
"""Validate a single node in the tree structure.
|
||||
|
||||
Args:
|
||||
node: The node dict to validate
|
||||
path: Current path in the tree (for error messages)
|
||||
errors: List to append errors to
|
||||
"""
|
||||
node_type = node.get("type")
|
||||
|
||||
if node_type == "decision":
|
||||
if "question" not in node or not node["question"]:
|
||||
errors.append({
|
||||
"field": f"{path}.question",
|
||||
"message": "Decision nodes must have a non-empty question"
|
||||
})
|
||||
|
||||
# If node has children, must have at least 2 (for decision branches)
|
||||
children = node.get("children", [])
|
||||
if children and len(children) < 2:
|
||||
errors.append({
|
||||
"field": f"{path}.children",
|
||||
"message": "Decision nodes with children must have at least 2 branches"
|
||||
})
|
||||
|
||||
elif node_type == "action":
|
||||
if "action" not in node or not node["action"]:
|
||||
errors.append({
|
||||
"field": f"{path}.action",
|
||||
"message": "Action nodes must have a non-empty action"
|
||||
})
|
||||
|
||||
elif node_type == "solution":
|
||||
if "solution" not in node or not node["solution"]:
|
||||
errors.append({
|
||||
"field": f"{path}.solution",
|
||||
"message": "Solution nodes must have a non-empty solution"
|
||||
})
|
||||
|
||||
else:
|
||||
errors.append({
|
||||
"field": f"{path}.type",
|
||||
"message": f"Unknown node type: {node_type}"
|
||||
})
|
||||
|
||||
|
||||
def _validate_children(children: list[dict[str, Any]], path: str, errors: list[dict[str, str]]) -> None:
|
||||
"""Recursively validate child nodes.
|
||||
|
||||
Args:
|
||||
children: List of child nodes
|
||||
path: Current path in the tree (for error messages)
|
||||
errors: List to append errors to
|
||||
"""
|
||||
for i, child in enumerate(children):
|
||||
child_path = f"{path}[{i}]"
|
||||
|
||||
if "id" not in child:
|
||||
errors.append({"field": f"{child_path}.id", "message": "Child node must have an id"})
|
||||
|
||||
if "type" not in child:
|
||||
errors.append({"field": f"{child_path}.type", "message": "Child node must have a type"})
|
||||
continue
|
||||
|
||||
_validate_node(child, child_path, errors)
|
||||
|
||||
# Recursively validate grandchildren
|
||||
if "children" in child:
|
||||
_validate_children(child["children"], f"{child_path}.children", errors)
|
||||
|
||||
|
||||
def can_publish_tree(tree_structure: dict[str, Any], name: str, description: str | None = None) -> tuple[bool, list[dict[str, str]]]:
|
||||
"""Check if a tree can be published.
|
||||
|
||||
Validates:
|
||||
- Tree has a name (non-empty)
|
||||
- Tree structure is valid
|
||||
|
||||
Args:
|
||||
tree_structure: The tree structure to validate
|
||||
name: The tree name
|
||||
description: Optional tree description
|
||||
|
||||
Returns:
|
||||
Tuple of (can_publish, list of errors)
|
||||
"""
|
||||
errors = []
|
||||
|
||||
# Validate name
|
||||
if not name or not name.strip():
|
||||
errors.append({"field": "name", "message": "Tree must have a name to be published"})
|
||||
|
||||
# Validate tree structure
|
||||
structure_valid, structure_errors = validate_tree_structure(tree_structure)
|
||||
errors.extend(structure_errors)
|
||||
|
||||
return len(errors) == 0, errors
|
||||
Reference in New Issue
Block a user