fix: procedural KB import — add title field, correct step types, validation gate

- _build_procedural_tree now maps AI node types (step, action, warning) to
  valid procedural types (procedure_step, section_header, procedure_end)
- Generates 'title' field from content text — fixes "Unnamed step" in editor
- Auto-appends procedure_end step if AI didn't generate one
- Adds procedural validation gate at commit endpoint (same as troubleshooting)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Michael Chihlas
2026-03-11 09:12:43 -04:00
parent c920e825c6
commit 00e1e701ca

View File

@@ -28,6 +28,7 @@ from app.core.rate_limit import limiter
from app.core.subscriptions import get_plan_limits
from app.core.ai_quota_service import get_user_plan
from app.core.ai_tree_validator import validate_generated_tree
from app.core.tree_validation import validate_procedural_structure
from app.core.kb_extraction_service import extract_text
from app.core.kb_conversion_service import convert_document
from app.models.kb_import import KBImport, KBImportNode
@@ -561,6 +562,22 @@ async def commit_import(
"validation_errors": validation_errors,
},
)
else:
# Procedural/maintenance validation
is_valid, proc_errors = validate_procedural_structure(tree_structure)
if not is_valid:
error_messages = [e.get("message", str(e)) for e in proc_errors]
logger.warning(
"KB commit blocked: procedural flow failed validation with %d errors: %s",
len(proc_errors), "; ".join(error_messages[:5]),
)
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail={
"message": "The converted flow has structural issues that need to be fixed before committing.",
"validation_errors": error_messages,
},
)
# Build intake_form for procedural flows
intake_form = None
@@ -881,17 +898,59 @@ def _demote_decision_to_action(node: dict, siblings: list[dict], index: int) ->
# Delete the broken _repair_tree and replace with the working version
# by removing the first broken attempt
def _build_procedural_tree(nodes: list[KBImportNode]) -> dict:
"""Build a procedural tree_structure from import nodes."""
"""Build a procedural tree_structure from import nodes.
Maps AI node types to valid procedural step types:
- step/action/warning → procedure_step
- section_header → section_header
Adds a procedure_end step at the end if missing.
Each step requires 'title' (from content text) and 'content' fields.
"""
# Type mapping from AI output to valid step types
TYPE_MAP = {
"step": "procedure_step",
"action": "procedure_step",
"warning": "procedure_step",
"question": "procedure_step",
"resolution": "procedure_step",
"section_header": "section_header",
"procedure_step": "procedure_step",
"procedure_end": "procedure_end",
}
steps = []
for node in sorted(nodes, key=lambda n: n.node_order):
content = node.content
step = {
raw_type = node.node_type
step_type = TYPE_MAP.get(raw_type, "procedure_step")
step_content = content.get("content", "")
step_title = content.get("title") or content.get("question") or step_content[:80] or "Step"
step: dict = {
"id": content.get("original_id", str(node.id)),
"type": node.node_type,
"content": content.get("content", ""),
"type": step_type,
"title": step_title,
"content": step_content,
}
# Preserve content_type if present
content_type = content.get("content_type")
if content_type:
step["content_type"] = content_type
steps.append(step)
# Ensure a procedure_end exists at the end
has_end = any(s["type"] == "procedure_end" for s in steps)
if not has_end and steps:
steps.append({
"id": "procedure-end",
"type": "procedure_end",
"title": "Procedure Complete",
"content": "All steps have been completed.",
})
return {
"id": "root",
"type": "procedural",