feat: add backend validation for fallback steps (Task 16)

Validate fallback_steps in procedural flow validation: required fields,
no nested fallback_steps, no duplicate IDs. Add FallbackStepRecord schema
and fallback_decisions field to SessionResponse.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
chihlasm
2026-03-16 01:13:34 -04:00
parent 5e4d323ef1
commit a0ba253428
3 changed files with 64 additions and 0 deletions

View File

@@ -301,6 +301,31 @@ def validate_generated_procedural_steps(tree: dict[str, Any]) -> list[str]:
f"Must be one of: {', '.join(sorted(VALID_CONTENT_TYPES))}"
)
# Validate fallback_steps if present (one level deep only)
fallback_steps = step.get("fallback_steps")
if fallback_steps is not None:
if not isinstance(fallback_steps, list):
errors.append(f"Step '{step_id or f'index {i}'}' fallback_steps must be an array")
else:
fallback_ids: set[str] = set()
for j, fb_step in enumerate(fallback_steps):
if not isinstance(fb_step, dict):
errors.append(f"Fallback step at {step_id}[{j}] is not an object")
continue
fb_id = fb_step.get("id")
if not fb_id or not isinstance(fb_id, str):
errors.append(f"Fallback step at {step_id}[{j}] missing or invalid 'id'")
elif fb_id in all_ids or fb_id in fallback_ids:
errors.append(f"Duplicate fallback step ID: '{fb_id}' (collides with primary or other fallback steps)")
else:
fallback_ids.add(fb_id)
all_ids.add(fb_id)
fb_title = fb_step.get("title")
if not fb_title or not isinstance(fb_title, str):
errors.append(f"Fallback step '{fb_id or f'{step_id}[{j}]'}' missing or invalid 'title'")
if fb_step.get("fallback_steps"):
errors.append(f"Fallback step '{fb_id or f'{step_id}[{j}]'}' cannot have its own fallback_steps (one level deep only)")
# Must have exactly one procedure_end as the last step
if procedure_end_count == 0:
errors.append("Procedural flow must have exactly one 'procedure_end' step")

View File

@@ -208,6 +208,34 @@ def validate_procedural_structure(tree_structure: dict[str, Any]) -> tuple[bool,
if content_type and content_type not in VALID_CONTENT_TYPES:
errors.append({"field": f"{path}.content_type", "message": f"Invalid content_type: {content_type}. Must be one of: {', '.join(VALID_CONTENT_TYPES)}"})
# Validate fallback_steps if present (one level deep only)
fallback_steps = step.get("fallback_steps")
if fallback_steps is not None:
if not isinstance(fallback_steps, list):
errors.append({"field": f"{path}.fallback_steps", "message": "fallback_steps must be an array"})
else:
fallback_ids: set[str] = set()
for j, fb_step in enumerate(fallback_steps):
fb_path = f"{path}.fallback_steps[{j}]"
if not isinstance(fb_step, dict):
errors.append({"field": fb_path, "message": "Fallback step must be an object"})
continue
fb_id = fb_step.get("id")
if not fb_id:
errors.append({"field": f"{fb_path}.id", "message": "Fallback step must have an id"})
elif fb_id in seen_ids or fb_id in fallback_ids:
errors.append({"field": f"{fb_path}.id", "message": f"Duplicate fallback step id: {fb_id}"})
else:
fallback_ids.add(fb_id)
seen_ids.add(fb_id)
if not fb_step.get("title"):
errors.append({"field": f"{fb_path}.title", "message": "Fallback step must have a non-empty title"})
fb_type = fb_step.get("type")
if fb_type and fb_type not in VALID_STEP_TYPES:
errors.append({"field": f"{fb_path}.type", "message": f"Invalid fallback step type: {fb_type}"})
if fb_step.get("fallback_steps"):
errors.append({"field": f"{fb_path}.fallback_steps", "message": "Fallback steps cannot have their own fallback_steps (one level deep only)"})
# Must have exactly one end step
if end_count == 0:
errors.append({"field": "tree_structure.steps", "message": "Procedural tree must have a procedure_end step as the last step"})

View File

@@ -98,6 +98,9 @@ class SessionResponse(BaseModel):
psa_ticket_id: Optional[str] = None
psa_connection_id: Optional[UUID] = None
# Fallback step decisions
fallback_decisions: list[dict[str, Any]] = Field(default_factory=list)
class Config:
from_attributes = True
@@ -123,6 +126,14 @@ class SessionComplete(BaseModel):
next_steps: Optional[str] = None
class FallbackStepRecord(BaseModel):
parent_step_id: str
fallback_step_id: str
completed_at: str | None = None
notes: str | None = None
outcome: Literal['resolved', 'not_resolved', 'skipped']
class SessionVariablesUpdate(BaseModel):
"""Partial update to session variables (dict merge)."""
variables: dict[str, str] = Field(..., description="Key-value pairs to merge into session_variables")