Compare commits
7 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 1fabd54ae5 | |||
| 945e16c1e0 | |||
| 09e374165a | |||
| aa7674c79e | |||
| f8f072260a | |||
| 65017576b1 | |||
| 3b3a318a4c |
@@ -0,0 +1,689 @@
|
||||
"""Step definitions for structural component output validation.
|
||||
|
||||
Tests for features/structural_validation.feature - validates all four
|
||||
validators covering plan tree nodes, decision CLI dicts, structured
|
||||
output envelopes, and the unified dispatcher.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
from behave import given, then, when # type: ignore[import-untyped]
|
||||
from behave.runner import Context # type: ignore[import-untyped]
|
||||
|
||||
from cleveragents.core.validation import (
|
||||
ValidationError,
|
||||
validate_decision_dict,
|
||||
validate_plan_tree,
|
||||
validate_structured_component_output,
|
||||
validate_structured_output,
|
||||
)
|
||||
|
||||
# Helper: valid ULID strings for test data
|
||||
VALID_ULID_A = "01ARZ3NDEKTSV4XXFFJFRC889A"
|
||||
VALID_ULID_B = "01ARZ3NDEKTSV4XXFFJFRC889B"
|
||||
VALID_ULID_C = "01ARZ3NDEKTSV4XXFFJFRC889C"
|
||||
|
||||
|
||||
# ────────────────────────────────────────────────────────────
|
||||
# Plan tree scenarios
|
||||
# ────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@given("the plan tree contains 1 node(s)")
|
||||
def step_plan_tree_1_node(ctx: Context) -> None:
|
||||
"""Create a single-node plan tree."""
|
||||
ctx.tree_nodes = [
|
||||
{
|
||||
"decision_id": VALID_ULID_A,
|
||||
"type": "strategy_choice",
|
||||
"sequence": 0,
|
||||
"question": "Q",
|
||||
"children": [],
|
||||
}
|
||||
]
|
||||
|
||||
|
||||
@given("the plan tree contains 5 node(s)")
|
||||
def step_plan_tree_5_nodes(ctx: Context) -> None:
|
||||
"""Create a 5-node plan tree."""
|
||||
ctx.tree_nodes = [
|
||||
{
|
||||
"decision_id": f"01ARZ3NDEKTSV4XXFFJFRC{str(i).zfill(4)}",
|
||||
"type": "strategy_choice",
|
||||
"sequence": i,
|
||||
"question": f"Decision question #{i}",
|
||||
"children": [],
|
||||
**({"parent_decision_id": VALID_ULID_A} if i > 0 else {}),
|
||||
}
|
||||
for i in range(5)
|
||||
]
|
||||
|
||||
|
||||
@given("the plan tree contains 20 node(s)")
|
||||
def step_plan_tree_20_nodes(ctx: Context) -> None:
|
||||
"""Create a 20-node plan tree."""
|
||||
ctx.tree_nodes = [
|
||||
{
|
||||
"decision_id": f"01ARZ3NDEKTSV4XXFFJFRC{str(i).zfill(4)}",
|
||||
"type": "strategy_choice",
|
||||
"sequence": i,
|
||||
"question": f"Decision question #{i}",
|
||||
"children": [],
|
||||
**({"parent_decision_id": VALID_ULID_A} if i > 0 else {}),
|
||||
}
|
||||
for i in range(20)
|
||||
]
|
||||
|
||||
|
||||
def create_sequence_node(ctx: Context, seq_val: int) -> None:
|
||||
"""Create a single plan tree node with the given sequence value."""
|
||||
ctx.tree_nodes = [
|
||||
{
|
||||
"decision_id": VALID_ULID_A,
|
||||
"type": "strategy_choice",
|
||||
"sequence": seq_val,
|
||||
"question": "Q",
|
||||
"children": [],
|
||||
}
|
||||
]
|
||||
|
||||
|
||||
@given('the plan tree contains a node missing "decision_id"')
|
||||
def step_node_missing_decision_id(ctx: Context) -> None:
|
||||
"""Create a node missing the required decision_id field."""
|
||||
ctx.tree_nodes = [
|
||||
{
|
||||
"type": "strategy_choice",
|
||||
"sequence": 0,
|
||||
"question": "Missing ID",
|
||||
"children": [],
|
||||
}
|
||||
]
|
||||
|
||||
|
||||
@given('the plan tree contains a node with invalid ULID "not-a-ulid-12345"')
|
||||
def step_node_invalid_ulid(ctx: Context) -> None:
|
||||
"""Create a node with an invalid decision_id."""
|
||||
ctx.tree_nodes = [
|
||||
{
|
||||
"decision_id": "not-a-ulid-12345",
|
||||
"type": "strategy_choice",
|
||||
"sequence": 0,
|
||||
"question": "Invalid ULID test",
|
||||
"children": [],
|
||||
}
|
||||
]
|
||||
|
||||
|
||||
@given('the plan tree contains a node with empty "children" list')
|
||||
def step_node_empty_children(ctx: Context) -> None:
|
||||
"""Create a node with an empty children list - should be valid."""
|
||||
ctx.tree_nodes = [
|
||||
{
|
||||
"decision_id": VALID_ULID_A,
|
||||
"type": "strategy_choice",
|
||||
"sequence": 0,
|
||||
"question": "Question text",
|
||||
"children": [],
|
||||
}
|
||||
]
|
||||
|
||||
|
||||
@given("two sibling nodes share the same sequence number")
|
||||
def step_duplicate_sequences(ctx: Context) -> None:
|
||||
"""Create two sibling nodes with identical sequence."""
|
||||
ctx.tree_nodes = [
|
||||
{
|
||||
"decision_id": VALID_ULID_A,
|
||||
"type": "strategy_choice",
|
||||
"sequence": 5,
|
||||
"question": "First",
|
||||
"children": [],
|
||||
"parent_decision_id": VALID_ULID_B,
|
||||
},
|
||||
{
|
||||
"decision_id": VALID_ULID_C,
|
||||
"type": "implementation_choice",
|
||||
"sequence": 5,
|
||||
"question": "Second",
|
||||
"children": [],
|
||||
"parent_decision_id": VALID_ULID_B,
|
||||
},
|
||||
]
|
||||
|
||||
|
||||
@given('a plan tree node with sequence "0"')
|
||||
def step_node_sequence_zero(ctx: Context) -> None:
|
||||
create_sequence_node(ctx, 0)
|
||||
|
||||
|
||||
@given('a plan tree node with sequence "1"')
|
||||
def step_node_sequence_one(ctx: Context) -> None:
|
||||
create_sequence_node(ctx, 1)
|
||||
|
||||
|
||||
@given('a plan tree node with sequence "100"')
|
||||
def step_node_sequence_100(ctx: Context) -> None:
|
||||
create_sequence_node(ctx, 100)
|
||||
|
||||
|
||||
@given('a plan tree node with sequence "-1"')
|
||||
def step_node_sequence_negative(ctx: Context) -> None:
|
||||
create_sequence_node(ctx, -1)
|
||||
|
||||
|
||||
def create_sequence_node(ctx: Context, seq_val: int) -> None:
|
||||
"""Create a single plan tree node with the given sequence value."""
|
||||
ctx.tree_nodes = [
|
||||
{
|
||||
"decision_id": VALID_ULID_A,
|
||||
"type": "strategy_choice",
|
||||
"sequence": seq_val,
|
||||
"question": "Q",
|
||||
"children": [],
|
||||
}
|
||||
]
|
||||
|
||||
|
||||
# ────────────────────────────────────────────────────────────
|
||||
# Decision dict scenarios
|
||||
# ────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def _parse_value(raw: str) -> Any:
|
||||
"""Parse a behave value string into Python type."""
|
||||
stripped = raw.strip().strip('"').strip("'")
|
||||
if stripped in ("null", "none", "None"):
|
||||
return None
|
||||
if stripped == "true":
|
||||
return True
|
||||
if stripped == "false":
|
||||
return False
|
||||
try:
|
||||
return int(stripped)
|
||||
except ValueError:
|
||||
pass
|
||||
try:
|
||||
return float(stripped)
|
||||
except ValueError:
|
||||
pass
|
||||
return stripped
|
||||
|
||||
|
||||
def _ensure_decision_dict(ctx: Context) -> dict:
|
||||
"""Ensure ctx.decision_dict exists with defaults."""
|
||||
if not hasattr(ctx, "decision_dict"):
|
||||
ctx.decision_dict = {
|
||||
"decision_id": VALID_ULID_A,
|
||||
"plan_id": VALID_ULID_B,
|
||||
"type": "strategy_choice",
|
||||
"sequence": 1,
|
||||
"question": "Which approach?",
|
||||
"chosen": "Option A",
|
||||
"confidence": 0.75,
|
||||
"parent": "(root)",
|
||||
"is_correction": False,
|
||||
"superseded": False,
|
||||
}
|
||||
return ctx.decision_dict # type: ignore[return-value]
|
||||
|
||||
|
||||
@given('I have a decision dict with "decision_id" set to 01ARZ3NDEKTSV4XXFFJFRC889A')
|
||||
def step_decision_dict_field_decision_id(ctx: Context) -> None:
|
||||
d = _ensure_decision_dict(ctx)
|
||||
d["decision_id"] = "01ARZ3NDEKTSV4XXFFJFRC889A"
|
||||
|
||||
|
||||
@given('I have a decision dict with "plan_id" set to 01ARZ3NDEKTSV4XXFFJFRC889B')
|
||||
def step_decision_dict_field_plan_id(ctx: Context) -> None:
|
||||
d = _ensure_decision_dict(ctx)
|
||||
d["plan_id"] = "01ARZ3NDEKTSV4XXFFJFRC889B"
|
||||
|
||||
|
||||
@given('I have a decision dict with "type" set to strategy_choice')
|
||||
def step_decision_dict_field_type(ctx: Context) -> None:
|
||||
d = _ensure_decision_dict(ctx)
|
||||
d["type"] = "strategy_choice"
|
||||
|
||||
|
||||
@given('I have a decision dict with "sequence" set to 5')
|
||||
def step_decision_dict_field_sequence(ctx: Context) -> None:
|
||||
d = _ensure_decision_dict(ctx)
|
||||
d["sequence"] = 5
|
||||
|
||||
|
||||
@given('I have a decision dict with "question" set to Which approach?')
|
||||
def step_decision_dict_field_question(ctx: Context) -> None:
|
||||
d = _ensure_decision_dict(ctx)
|
||||
d["question"] = "Which approach?"
|
||||
|
||||
|
||||
@given('I have a decision dict with "chosen" set to A')
|
||||
def step_decision_dict_field_chosen(ctx: Context) -> None:
|
||||
d = _ensure_decision_dict(ctx)
|
||||
d["chosen"] = "A"
|
||||
|
||||
|
||||
@given('I have a decision dict with "confidence" set to 0.75')
|
||||
def step_decision_dict_field_confidence(ctx: Context) -> None:
|
||||
d = _ensure_decision_dict(ctx)
|
||||
d["confidence"] = 0.75
|
||||
|
||||
|
||||
@given('I have a decision dict with "parent" set to (root)')
|
||||
def step_decision_dict_field_parent(ctx: Context) -> None:
|
||||
d = _ensure_decision_dict(ctx)
|
||||
d["parent"] = "(root)"
|
||||
|
||||
|
||||
@given('I have a decision dict with "is_correction" set to false')
|
||||
def step_decision_dict_field_is_correction(ctx: Context) -> None:
|
||||
d = _ensure_decision_dict(ctx)
|
||||
d["is_correction"] = False
|
||||
|
||||
|
||||
@given('I have a decision dict with "superseded" set to false')
|
||||
def step_decision_dict_field_superseded(ctx: Context) -> None:
|
||||
d = _ensure_decision_dict(ctx)
|
||||
d["superseded"] = False
|
||||
|
||||
|
||||
@given('I have a decision dict with "confidence" set to 1.5')
|
||||
def step_decision_dict_confidence_15(ctx: Context) -> None:
|
||||
d = _ensure_decision_dict(ctx)
|
||||
d["confidence"] = 1.5
|
||||
|
||||
|
||||
@given('I have a decision dict with "confidence" set to null')
|
||||
def step_decision_dict_confidence_null(ctx: Context) -> None:
|
||||
d = _ensure_decision_dict(ctx)
|
||||
d["confidence"] = None
|
||||
|
||||
|
||||
@given('I have a decision dict with "question" set to 42')
|
||||
def step_decision_dict_question_42(ctx: Context) -> None:
|
||||
d = _ensure_decision_dict(ctx)
|
||||
d["question"] = 42
|
||||
|
||||
|
||||
@given('I have a decision dict with "parent" set to "random-string"')
|
||||
def step_decision_dict_parent_random(ctx: Context) -> None:
|
||||
d = _ensure_decision_dict(ctx)
|
||||
d["parent"] = "random-string"
|
||||
|
||||
|
||||
@given('I have a decision dict with "is_correction" set to "yes"')
|
||||
def step_decision_dict_is_correction_yes(ctx: Context) -> None:
|
||||
d = _ensure_decision_dict(ctx)
|
||||
d["is_correction"] = "yes"
|
||||
|
||||
|
||||
@given("I have a valid decision dict")
|
||||
@when("I validate the decision dict")
|
||||
def step_validate_decision(ctx: Context) -> None:
|
||||
"""Validate and store results."""
|
||||
d = _ensure_decision_dict(ctx)
|
||||
ctx.validation_result = validate_decision_dict(d)
|
||||
|
||||
|
||||
# ────────────────────────────────────────────────────────────
|
||||
# Structured output scenarios
|
||||
# ────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def _ensure_struct_output(ctx: Context) -> dict:
|
||||
"""Ensure ctx.struct_output exists with defaults."""
|
||||
if not hasattr(ctx, "struct_output"):
|
||||
ctx.struct_output = {
|
||||
"command": "agents plan list",
|
||||
"session_id": VALID_ULID_A,
|
||||
"status": "ok",
|
||||
"exit_code": 0,
|
||||
"elements": [],
|
||||
}
|
||||
return ctx.struct_output # type: ignore[return-value]
|
||||
|
||||
|
||||
@given("I have a valid structured output with all required fields")
|
||||
@when("I validate the structured output")
|
||||
def step_validate_structured(ctx: Context) -> None:
|
||||
"""Validate and store results."""
|
||||
o = _ensure_struct_output(ctx)
|
||||
ctx.validation_result = validate_structured_output(o)
|
||||
|
||||
|
||||
@given(
|
||||
'I have a structured output with "command" set to agents plan list (and all other required fields present)'
|
||||
)
|
||||
def step_structured_output_command(ctx: Context) -> None:
|
||||
o = _ensure_struct_output(ctx)
|
||||
o["command"] = "agents plan list"
|
||||
|
||||
|
||||
@given(
|
||||
'I have a structured output with "session_id" set to 01ARZ3NDEKTSV4XXFFJFRC889A (and all other required fields present)'
|
||||
)
|
||||
def step_structured_output_session_id(ctx: Context) -> None:
|
||||
o = _ensure_struct_output(ctx)
|
||||
o["session_id"] = "01ARZ3NDEKTSV4XXFFJFRC889A"
|
||||
|
||||
|
||||
@given(
|
||||
'I have a structured output with "status" set to ok (and all other required fields present)'
|
||||
)
|
||||
def step_structured_output_status_ok(ctx: Context) -> None:
|
||||
o = _ensure_struct_output(ctx)
|
||||
o["status"] = "ok"
|
||||
|
||||
|
||||
@given(
|
||||
'I have a structured output with "exit_code" set to 0 (and all other required fields present)'
|
||||
)
|
||||
def step_structured_output_exit_code_0(ctx: Context) -> None:
|
||||
o = _ensure_struct_output(ctx)
|
||||
o["exit_code"] = 0
|
||||
|
||||
|
||||
@given(
|
||||
'I have a structured output with "status" set to "running" (and all other required fields present)'
|
||||
)
|
||||
def step_structured_output_status_running(ctx: Context) -> None:
|
||||
o = _ensure_struct_output(ctx)
|
||||
o["status"] = "running"
|
||||
|
||||
|
||||
@given(
|
||||
'I have a structured output with "exit_code" set to "-1" (and all other required fields present)'
|
||||
)
|
||||
def step_structured_output_exit_code_neg1(ctx: Context) -> None:
|
||||
o = _ensure_struct_output(ctx)
|
||||
o["exit_code"] = -1
|
||||
|
||||
|
||||
@given(
|
||||
'I have a structured output with empty "elements" list (and all other fields valid)'
|
||||
)
|
||||
def step_empty_elements(ctx: Context) -> None:
|
||||
"""Empty elements is valid."""
|
||||
o = _ensure_struct_output(ctx)
|
||||
o["elements"] = []
|
||||
|
||||
|
||||
@given('I have a structured output with an element missing "kind" field')
|
||||
def step_invalid_element(ctx: Context) -> None:
|
||||
"""Create an invalid element."""
|
||||
o = _ensure_struct_output(ctx)
|
||||
o["elements"] = [{"not_kind": "panel"}]
|
||||
|
||||
|
||||
# ────────────────────────────────────────────────────────────
|
||||
# Dispatcher scenarios
|
||||
# ────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@given('I target validation for "plan_tree" with valid data')
|
||||
def step_dispatcher_target_plan_tree(ctx: Context) -> None:
|
||||
ctx.target_type = "plan_tree"
|
||||
ctx.validator_data = [
|
||||
{
|
||||
"decision_id": VALID_ULID_A,
|
||||
"type": "strategy_choice",
|
||||
"sequence": 0,
|
||||
"question": "Q",
|
||||
"children": [],
|
||||
}
|
||||
]
|
||||
|
||||
|
||||
@given('I target validation for "decision" with valid data')
|
||||
def step_dispatcher_target_decision(ctx: Context) -> None:
|
||||
ctx.target_type = "decision"
|
||||
ctx.validator_data = {
|
||||
"decision_id": VALID_ULID_A,
|
||||
"plan_id": VALID_ULID_B,
|
||||
"type": "strategy_choice",
|
||||
"sequence": 1,
|
||||
"question": "Q?",
|
||||
"chosen": "A",
|
||||
"confidence": 0.5,
|
||||
"parent": "(root)",
|
||||
"is_correction": False,
|
||||
"superseded": False,
|
||||
}
|
||||
|
||||
|
||||
@given('I target validation for "structured_output" with valid data')
|
||||
def step_dispatcher_target_struct(ctx: Context) -> None:
|
||||
ctx.target_type = "structured_output"
|
||||
ctx.validator_data = {
|
||||
"command": "test",
|
||||
"session_id": VALID_ULID_A,
|
||||
"status": "ok",
|
||||
"exit_code": 0,
|
||||
"elements": [],
|
||||
}
|
||||
|
||||
|
||||
@given('I target validation for "unknown_type" with valid data')
|
||||
def step_dispatcher_target_unknown(ctx: Context) -> None:
|
||||
"""Set up a dispatcher test for unknown target type."""
|
||||
ctx.target_type = "unknown_type"
|
||||
ctx.validator_data = {}
|
||||
|
||||
|
||||
@when("I call validate_structured_component_output")
|
||||
def step_call_dispatcher(ctx: Context) -> None:
|
||||
"""Call the dispatcher."""
|
||||
target = ctx.target_type if hasattr(ctx, "target_type") else "plan_tree" # type: ignore[attr-defined]
|
||||
data = ctx.validator_data if hasattr(ctx, "validator_data") else {} # type: ignore[attr-defined]
|
||||
try:
|
||||
ctx.dispatcher_result = validate_structured_component_output(target, data) # type: ignore[arg-type]
|
||||
ctx.dispatcher_error = None
|
||||
except ValidationError as exc:
|
||||
ctx.dispatcher_result = None
|
||||
ctx.dispatcher_error = exc
|
||||
|
||||
|
||||
# ────────────────────────────────────────────────────────────
|
||||
# Shared scenarios
|
||||
# ────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@given("the plan tree has {node_count:d} valid nodes")
|
||||
def step_valid_tree(ctx: Context, node_count: int = 3) -> None:
|
||||
"""Set up a default valid tree."""
|
||||
if not hasattr(ctx, "tree_nodes"):
|
||||
ctx.tree_nodes = [
|
||||
{
|
||||
"decision_id": f"01ARZ3NDEKTSV4XXFFJFRC{str(i).zfill(4)}",
|
||||
"type": "strategy_choice",
|
||||
"sequence": i,
|
||||
"question": f"Q{i}",
|
||||
"children": [],
|
||||
}
|
||||
for i in range(node_count)
|
||||
]
|
||||
|
||||
|
||||
@given("I have a plan tree with valid nodes")
|
||||
def step_valid_nodes_default(ctx: Context) -> None:
|
||||
"""Set default valid tree if no tree defined."""
|
||||
|
||||
|
||||
@when("I validate the plan tree")
|
||||
def step_validate_plan_tree(ctx: Context) -> None:
|
||||
"""Validate and store results."""
|
||||
nodes = (
|
||||
ctx.tree_nodes
|
||||
if hasattr(ctx, "tree_nodes")
|
||||
else [ # type: ignore[attr-defined]
|
||||
{
|
||||
"decision_id": VALID_ULID_A,
|
||||
"type": "strategy_choice",
|
||||
"sequence": i,
|
||||
"question": f"Q{i}",
|
||||
"children": [],
|
||||
}
|
||||
for i in range(3)
|
||||
]
|
||||
)
|
||||
ctx.validation_result = validate_plan_tree(nodes)
|
||||
|
||||
|
||||
@when("I validate the plan tree structure")
|
||||
def step_validate_structure(ctx: Context) -> None:
|
||||
"""Run plan_tree validation."""
|
||||
nodes = ctx.tree_nodes if hasattr(ctx, "tree_nodes") else [] # type: ignore[attr-defined]
|
||||
ctx.validation_result = validate_plan_tree(nodes)
|
||||
|
||||
|
||||
@given("I have a decision dict where whitespace may vary between fields")
|
||||
def step_whitespace_dict(ctx: Context) -> None:
|
||||
"""Whitespace in string values - structure still valid."""
|
||||
ctx.decision_dict = {
|
||||
"decision_id": VALID_ULID_A,
|
||||
"plan_id": VALID_ULID_B,
|
||||
"type": " strategy_choice ",
|
||||
"sequence": 1,
|
||||
"question": "Which approach? ",
|
||||
"chosen": " Option A ",
|
||||
"confidence": 0.75,
|
||||
"parent": "(root)",
|
||||
"is_correction": False,
|
||||
"superseded": False,
|
||||
}
|
||||
|
||||
|
||||
@given("a plan tree with parent and child node references connected by ULID")
|
||||
def step_parent_child_tree(ctx: Context) -> None:
|
||||
"""Parent-decoy pattern - root plus children."""
|
||||
ctx.tree_nodes = [
|
||||
{
|
||||
"decision_id": VALID_ULID_A,
|
||||
"type": "prompt_definition",
|
||||
"sequence": 0,
|
||||
"question": "Root question",
|
||||
"children": [{"decision_id": VALID_ULID_B}, {"decision_id": VALID_ULID_C}],
|
||||
"parent_decision_id": None,
|
||||
},
|
||||
{
|
||||
"decision_id": VALID_ULID_B,
|
||||
"type": "strategy_choice",
|
||||
"sequence": 1,
|
||||
"question": "Child A",
|
||||
"children": [],
|
||||
"parent_decision_id": VALID_ULID_A,
|
||||
},
|
||||
{
|
||||
"decision_id": VALID_ULID_C,
|
||||
"type": "strategy_choice",
|
||||
"sequence": 2,
|
||||
"question": "Child B",
|
||||
"children": [],
|
||||
"parent_decision_id": VALID_ULID_A,
|
||||
},
|
||||
]
|
||||
|
||||
|
||||
@given("I have invalid plan tree data")
|
||||
def step_invalid_tree_data(ctx: Context) -> None:
|
||||
"""Create invalid plan tree node."""
|
||||
ctx.tree_nodes = [
|
||||
{"not_a_key": "value"}, # missing required fields entirely
|
||||
]
|
||||
|
||||
|
||||
# ────────────────────────────────────────────────────────────
|
||||
# Assertions (shared / cross-scenario)
|
||||
# ────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@then("it should be structurally valid")
|
||||
def step_should_be_valid(ctx: Context) -> None:
|
||||
"""Assert validation passed."""
|
||||
result = getattr(ctx, "validation_result", {}) or {"valid": True}
|
||||
assert result["valid"], f"Expected valid but got errors: {result.get('errors', [])}"
|
||||
|
||||
|
||||
@then("it should be structurally invalid")
|
||||
def step_should_be_invalid(ctx: Context) -> None:
|
||||
"""Assert validation failed."""
|
||||
result = getattr(ctx, "validation_result", {}) or {"valid": True}
|
||||
assert not result["valid"], "Expected invalid but got valid"
|
||||
|
||||
|
||||
@then("it should report no errors")
|
||||
def step_no_errors(ctx: Context) -> None:
|
||||
"""Assert no error messages."""
|
||||
result = getattr(ctx, "validation_result", {}) or {}
|
||||
errs = result.get("errors", [])
|
||||
assert len(errs) == 0, f"Expected no errors but got: {errs}"
|
||||
|
||||
|
||||
@then('it should report errors including "{text}"')
|
||||
def step_errors_include(ctx: Context, text: str) -> None:
|
||||
"""Assert at least one error contains the given substring."""
|
||||
result = getattr(ctx, "validation_result", {}) or {}
|
||||
errs = result.get("errors", [])
|
||||
assert any(text in e for e in errs), f"'{text}' not found in errors: {errs}"
|
||||
|
||||
|
||||
@then('error message must include "{text}"')
|
||||
@then('error must include "{text}"')
|
||||
def step_error_contains(ctx: Context, text: str) -> None:
|
||||
"""Assert an error contains substring."""
|
||||
result = getattr(ctx, "validation_result", {}) or {}
|
||||
errs = result.get("errors", [])
|
||||
assert any(text in e for e in errs), f"'{text}' not found in errors: {errs}"
|
||||
|
||||
|
||||
@then('the first error message should identify "{prefix}" and the specific field issue')
|
||||
def step_first_error_identifies(ctx: Context, prefix: str) -> None:
|
||||
"""Assert first error contains prefix."""
|
||||
result = getattr(ctx, "validation_result", {}) or {}
|
||||
errs = result.get("errors", [])
|
||||
assert len(errs) > 0, "Expected at least one error"
|
||||
assert prefix in errs[0], f"First error '{errs[0]}' should contain '{prefix}'"
|
||||
|
||||
|
||||
@then('it should raise a ValidationError with message containing "{text}"')
|
||||
def step_raise_error_with_text(ctx: Context, text: str) -> None:
|
||||
"""Assert dispatcher raised specific error."""
|
||||
err = getattr(ctx, "dispatcher_error", None)
|
||||
assert err is not None, "Expected a ValidationError"
|
||||
assert isinstance(err, ValidationError), (
|
||||
f"Expected ValidationError, got {type(err)}"
|
||||
)
|
||||
assert text in str(err), f"'{text}' not in error message: {err}"
|
||||
|
||||
|
||||
@then("it should dispatch to the matching validator")
|
||||
@then("return a valid result")
|
||||
def step_dispatcher_valid(ctx: Context) -> None:
|
||||
"""Assert dispatcher succeeded and returned valid."""
|
||||
err = getattr(ctx, "dispatcher_error", None)
|
||||
res = getattr(ctx, "dispatcher_result", None)
|
||||
assert err is None, f"Dispatcher raised: {err}"
|
||||
assert res is not None, "Expected a result dict"
|
||||
assert res.get("valid"), f"Result not valid: {res}"
|
||||
|
||||
|
||||
@then("validation should pass (structure matched, not exact characters)")
|
||||
def step_structure_matched(ctx: Context) -> None:
|
||||
"""Whitespace-tolerant structural validation passes."""
|
||||
d = getattr(ctx, "decision_dict", {})
|
||||
ctx.validation_result = validate_decision_dict(d) # type: ignore[attr-defined]
|
||||
assert ctx.validation_result["valid"], (
|
||||
f"Structural validation failed: {ctx.validation_result.get('errors', [])}"
|
||||
) # type: ignore[attr-defined]
|
||||
|
||||
|
||||
@then("parent-child relationships should be validly connected")
|
||||
def step_relationships_valid(ctx: Context) -> None:
|
||||
"""Parent-child nodes validate without errors."""
|
||||
result = ctx.validation_result if hasattr(ctx, "validation_result") else {} # type: ignore[attr-defined]
|
||||
assert result.get("valid", False), (
|
||||
f"Relation validation failed: {result.get('errors', [])}"
|
||||
)
|
||||
@@ -0,0 +1,231 @@
|
||||
Feature: Structural component output validation
|
||||
|
||||
Output validation checks structural components (not exact characters).
|
||||
Required fields are validated for presence and correct type.
|
||||
Structural relationships are validated (parent-child, ordering).
|
||||
Minor formatting differences do not cause validation failures.
|
||||
Validation failures produce actionable error messages.
|
||||
|
||||
Based on Epic #8137 - structural output validation overhaul.
|
||||
|
||||
Scenario: validate_plan_tree accepts valid node count of 1
|
||||
Given the plan tree contains 1 node(s)
|
||||
When I validate the plan tree
|
||||
Then it should be structurally valid
|
||||
And it should report no errors
|
||||
|
||||
Scenario: validate_plan_tree accepts valid node count of 5
|
||||
Given the plan tree contains 5 node(s)
|
||||
When I validate the plan tree
|
||||
Then it should be structurally valid
|
||||
And it should report no errors
|
||||
|
||||
Scenario: validate_plan_tree accepts valid node count of 20
|
||||
Given the plan tree contains 20 node(s)
|
||||
When I validate the plan tree
|
||||
Then it should be structurally valid
|
||||
And it should report no errors
|
||||
|
||||
Scenario: validate_plan_tree rejects nodes missing required keys
|
||||
Given the plan tree contains a node missing "decision_id"
|
||||
When I validate the plan tree
|
||||
Then it should be structurally invalid
|
||||
And it should report errors including "missing required key(s)"
|
||||
|
||||
Scenario: validate_plan_tree rejects invalid ULID in decision_id
|
||||
Given the plan tree contains a node with invalid ULID "not-a-ulid-12345"
|
||||
When I validate the plan tree
|
||||
Then it should be structurally invalid
|
||||
And it should report errors including "must be a valid ULID"
|
||||
|
||||
Scenario: validate_plan_tree accepts empty children list
|
||||
Given the plan tree contains a node with empty "children" list
|
||||
When I validate the plan tree
|
||||
Then it should be structurally valid
|
||||
|
||||
Scenario: validate_plan_tree rejects duplicate sequences for siblings
|
||||
Given two sibling nodes share the same sequence number
|
||||
When I validate the plan tree
|
||||
Then it should be structurally invalid
|
||||
And it should report errors including "duplicate sequence"
|
||||
|
||||
Scenario: validate_plan_tree accepts non-negative sequence 0
|
||||
Given a plan tree node with sequence "0"
|
||||
When I validate the plan tree
|
||||
Then it should be structurally valid
|
||||
|
||||
Scenario: validate_plan_tree accepts non-negative sequence 1
|
||||
Given a plan tree node with sequence "1"
|
||||
When I validate the plan tree
|
||||
Then it should be structurally valid
|
||||
|
||||
Scenario: validate_plan_tree accepts non-negative sequence 100
|
||||
Given a plan tree node with sequence "100"
|
||||
When I validate the plan tree
|
||||
Then it should be structurally valid
|
||||
|
||||
Scenario: validate_plan_tree rejects negative sequence
|
||||
Given a plan tree node with sequence "-1"
|
||||
When I validate the plan tree
|
||||
Then it should be structurally invalid
|
||||
And error message must include "must be a non-negative integer"
|
||||
|
||||
Scenario: validate_decision_dict accepts valid CLI dict for decision_id field
|
||||
Given I have a decision dict with "decision_id" set to 01ARZ3NDEKTSV4XXFFJFRC889A
|
||||
When I validate the decision dict
|
||||
Then it should be structurally valid
|
||||
|
||||
Scenario: validate_decision_dict accepts valid CLI dict for plan_id field
|
||||
Given I have a decision dict with "plan_id" set to 01ARZ3NDEKTSV4XXFFJFRC889B
|
||||
When I validate the decision dict
|
||||
Then it should be structurally valid
|
||||
|
||||
Scenario: validate_decision_dict accepts valid CLI dict for type field
|
||||
Given I have a decision dict with "type" set to strategy_choice
|
||||
When I validate the decision dict
|
||||
Then it should be structurally valid
|
||||
|
||||
Scenario: validate_decision_dict accepts valid CLI dict for sequence field
|
||||
Given I have a decision dict with "sequence" set to 5
|
||||
When I validate the decision dict
|
||||
Then it should be structurally valid
|
||||
|
||||
Scenario: validate_decision_dict accepts valid CLI dict for question field
|
||||
Given I have a decision dict with "question" set to Which approach?
|
||||
When I validate the decision dict
|
||||
Then it should be structurally valid
|
||||
|
||||
Scenario: validate_decision_dict accepts valid CLI dict for chosen field
|
||||
Given I have a decision dict with "chosen" set to A
|
||||
When I validate the decision dict
|
||||
Then it should be structurally valid
|
||||
|
||||
Scenario: validate_decision_dict accepts valid CLI dict for confidence field
|
||||
Given I have a decision dict with "confidence" set to 0.75
|
||||
When I validate the decision dict
|
||||
Then it should be structurally valid
|
||||
|
||||
Scenario: validate_decision_dict accepts valid CLI dict for parent field
|
||||
Given I have a decision dict with "parent" set to (root)
|
||||
When I validate the decision dict
|
||||
Then it should be structurally valid
|
||||
|
||||
Scenario: validate_decision_dict accepts valid CLI dict for is_correction field
|
||||
Given I have a decision dict with "is_correction" set to false
|
||||
When I validate the decision dict
|
||||
Then it should be structurally valid
|
||||
|
||||
Scenario: validate_decision_dict accepts valid CLI dict for superseded field
|
||||
Given I have a decision dict with "superseded" set to false
|
||||
When I validate the decision dict
|
||||
Then it should be structurally valid
|
||||
|
||||
Scenario: validate_decision_dict rejects confidence outside [0,1]
|
||||
Given I have a decision dict with "confidence" set to 1.5
|
||||
When I validate the decision dict
|
||||
Then it should be structurally invalid
|
||||
And error message must include "must be in range [0.0, 1.0]"
|
||||
|
||||
Scenario: validate_decision_dict accepts None confidence
|
||||
Given I have a decision dict with "confidence" set to null
|
||||
When I validate the decision dict
|
||||
Then it should be structurally valid
|
||||
|
||||
Scenario: validate_decision_dict rejects non-string question
|
||||
Given I have a decision dict with "question" set to 42
|
||||
When I validate the decision dict
|
||||
Then it should be structurally invalid
|
||||
And error message must include "must be a non-empty string"
|
||||
|
||||
Scenario: validate_decision_dict rejects parent that is neither ULID nor "(root)"
|
||||
Given I have a decision dict with "parent" set to "random-string"
|
||||
When I validate the decision dict
|
||||
Then it should be structurally invalid
|
||||
And error message must include "must be a ULID or '(root)'"
|
||||
|
||||
Scenario: validate_decision_dict rejects is_correction that is not boolean
|
||||
Given I have a decision dict with "is_correction" set to "yes"
|
||||
When I validate the decision dict
|
||||
Then it should be structurally invalid
|
||||
And error message must include "must be a boolean"
|
||||
|
||||
Scenario: validate_structured_output accepts valid envelope for command field
|
||||
Given I have a structured output with "command" set to agents plan list (and all other required fields present)
|
||||
When I validate the structured output
|
||||
Then it should be structurally valid
|
||||
|
||||
Scenario: validate_structured_output accepts valid envelope for session_id field
|
||||
Given I have a structured output with "session_id" set to 01ARZ3NDEKTSV4XXFFJFRC889A (and all other required fields present)
|
||||
When I validate the structured output
|
||||
Then it should be structurally valid
|
||||
|
||||
Scenario: validate_structured_output accepts valid envelope for status field
|
||||
Given I have a structured output with "status" set to ok (and all other required fields present)
|
||||
When I validate the structured output
|
||||
Then it should be structurally valid
|
||||
|
||||
Scenario: validate_structured_output accepts valid envelope for exit_code field
|
||||
Given I have a structured output with "exit_code" set to 0 (and all other required fields present)
|
||||
When I validate the structured output
|
||||
Then it should be structurally valid
|
||||
|
||||
Scenario: validate_structured_output rejects invalid status value
|
||||
Given I have a structured output with "status" set to "running" (and all other required fields present)
|
||||
When I validate the structured output
|
||||
Then it should be structurally invalid
|
||||
And error must include "must be one of"
|
||||
|
||||
Scenario: validate_structured_output rejects negative exit_code
|
||||
Given I have a structured output with "exit_code" set to "-1" (and all other required fields present)
|
||||
When I validate the structured output
|
||||
Then it should be structurally invalid
|
||||
And error must include "must be >= 0"
|
||||
|
||||
Scenario: validate_structured_output accepts zero elements list
|
||||
Given I have a structured output with empty "elements" list (and all other fields valid)
|
||||
When I validate the structured output
|
||||
Then it should be structurally valid
|
||||
|
||||
Scenario: validate_structured_output rejects invalid element format
|
||||
Given I have a structured output with an element missing "kind" field
|
||||
When I validate the structured output
|
||||
Then it should be structurally invalid
|
||||
And error must include "missing 'kind' field"
|
||||
|
||||
Scenario: validate_structured_component_output routes plan_tree to correct validator
|
||||
Given I target validation for "plan_tree" with valid data
|
||||
When I call validate_structured_component_output
|
||||
Then it should dispatch to the matching validator
|
||||
And return a valid result
|
||||
|
||||
Scenario: validate_structured_component_output routes decision to correct validator
|
||||
Given I target validation for "decision" with valid data
|
||||
When I call validate_structured_component_output
|
||||
Then it should dispatch to the matching validator
|
||||
And return a valid result
|
||||
|
||||
Scenario: validate_structured_component_output routes structured_output to correct validator
|
||||
Given I target validation for "structured_output" with valid data
|
||||
When I call validate_structured_component_output
|
||||
Then it should dispatch to the matching validator
|
||||
And return a valid result
|
||||
|
||||
Scenario: validate_structured_component_output rejects unknown target type
|
||||
Given I target validation for "unknown_type" with valid data
|
||||
When I call validate_structured_component_output
|
||||
Then it should raise a ValidationError with message containing "Unknown target_type"
|
||||
|
||||
Scenario: Structural output is robust to minor formatting differences
|
||||
Given I have a decision dict where whitespace may vary between fields
|
||||
When I validate the decision dict
|
||||
Then validation should pass (structure matched, not exact characters)
|
||||
|
||||
Scenario: Plan tree nodes preserve parent-child structural relationships
|
||||
Given a plan tree with parent and child node references connected by ULID
|
||||
When I validate the plan tree structure
|
||||
Then parent-child relationships should be validly connected
|
||||
|
||||
Scenario: Validation error messages are actionable
|
||||
Given I have invalid plan tree data
|
||||
When I validate the plan tree
|
||||
Then the first error message should identify "node[0]" and the specific field issue
|
||||
@@ -21,14 +21,28 @@ from cleveragents.core.error_handling import (
|
||||
redact_value,
|
||||
wrap_unexpected,
|
||||
)
|
||||
from cleveragents.core.validation import (
|
||||
ValidationError,
|
||||
ValidationWarning,
|
||||
validate_decision_dict,
|
||||
validate_plan_tree,
|
||||
validate_structured_component_output,
|
||||
validate_structured_output,
|
||||
)
|
||||
|
||||
__all__: list[str] = [
|
||||
"ErrorCategory",
|
||||
"ErrorCode",
|
||||
"ErrorInfo",
|
||||
"ValidationError",
|
||||
"ValidationWarning",
|
||||
"classify_error",
|
||||
"format_error_for_cli",
|
||||
"redact_error_details",
|
||||
"redact_value",
|
||||
"validate_decision_dict",
|
||||
"validate_plan_tree",
|
||||
"validate_structured_component_output",
|
||||
"validate_structured_output",
|
||||
"wrap_unexpected",
|
||||
]
|
||||
|
||||
@@ -0,0 +1,482 @@
|
||||
"""Structural component output validators for plan trees, decisions, and sessions.
|
||||
|
||||
This module replaces legacy exact-character matching output validation
|
||||
with structural component checking. Output is validated against its
|
||||
schema shape - required fields, types, structural relationships, and
|
||||
ID formats - rather than character-by-character equality checks.
|
||||
|
||||
Validators
|
||||
----------
|
||||
|
||||
- :func:`validate_plan_tree` - validates plan tree node dicts for
|
||||
required keys (``decision_id``, ``type``, ``sequence``, ``question``,
|
||||
``children``), ULID format, correct types, and sibling ordering.
|
||||
|
||||
- :func:`validate_decision_dict` - validates a decision CLI output
|
||||
dictionary against the shape returned by :meth:`Decision.as_cli_dict`,
|
||||
checking field presence/types, ULID patterns, confidence range
|
||||
[0.0, 1.0], and boolean fields.
|
||||
|
||||
- :func:`validate_structured_output` - validates a
|
||||
:class:`StructuredOutput` envelope for command presence, session_id
|
||||
as ULID, status membership (``ok``, ``error``, ``warn``, ``info``),
|
||||
exit_code (non-negative int), and elements list integrity.
|
||||
|
||||
- :func:`validate_structured_component_output` - unified dispatcher
|
||||
that routes validation by ``target_type`` string to the correct
|
||||
validator implementation.
|
||||
|
||||
Based on:
|
||||
- docs/specification.md Output Rendering Framework
|
||||
- Epic #8137 - structural output validation overhaul
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from collections.abc import Callable
|
||||
from typing import Any
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Constants
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
ULID_PATTERN = r"^[0-9A-HJKMNP-TV-Z]{26}$"
|
||||
_ULID_RE = re.compile(ULID_PATTERN)
|
||||
|
||||
VALID_STATUSES = frozenset({"ok", "error", "warn", "info"})
|
||||
|
||||
PLAN_TREE_REQUIRED_KEYS = frozenset(
|
||||
{"decision_id", "type", "sequence", "question", "children"}
|
||||
)
|
||||
|
||||
DECISION_CLI_REQUIRED_FIELDS: dict[str, type | tuple[type, ...] | None] = {
|
||||
"decision_id": str, # ULID
|
||||
"plan_id": str, # ULID
|
||||
"type": str, # DecisionType string value
|
||||
"sequence": int, # sequence_number
|
||||
"question": str, # question text
|
||||
"chosen": str, # chosen_option text
|
||||
"confidence": (float, type(None)), # ge=0.0, le=1.0 or None
|
||||
"parent": str, # parent ULID or "(root)"
|
||||
"is_correction": bool, # True / False
|
||||
"superseded": bool, # True / False
|
||||
}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Error types
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class ValidationError(Exception):
|
||||
"""Raised when structural validation fails.
|
||||
|
||||
Attributes:
|
||||
message: Human-readable description of the failure.
|
||||
errors: List of individual field-level error strings.
|
||||
"""
|
||||
|
||||
def __init__(self, message: str, errors: list[str] | None = None) -> None:
|
||||
self.message = message
|
||||
self.errors = errors or []
|
||||
super().__init__(f"{message}: {'; '.join(self.errors)}")
|
||||
|
||||
|
||||
class ValidationWarning(Warning):
|
||||
"""Raised for non-fatal structural issues (e.g. optional field missing)."""
|
||||
|
||||
pass
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Plan tree validator
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def validate_plan_tree(nodes: list[dict[str, Any]]) -> dict[str, Any]:
|
||||
"""Validate a plan tree represented as a list of node dicts.
|
||||
|
||||
Each node must be a dictionary with the following required structural
|
||||
keys:
|
||||
|
||||
* ``decision_id`` - valid ULID (26-char base32 uppercase)
|
||||
* ``type`` - non-empty string describing the decision type
|
||||
* ``sequence`` - non-negative integer for sibling ordering
|
||||
* ``question`` - non-empty string describing the decision question
|
||||
* ``children`` - list of child node references or dicts
|
||||
|
||||
Args:
|
||||
nodes: Flat list of plan tree node dictionaries.
|
||||
|
||||
Returns:
|
||||
A summary dict with keys ``valid``, ``node_count``, and
|
||||
``errors`` / ``warnings`` lists.
|
||||
|
||||
Raises:
|
||||
ValidationError: If any node violates structural requirements.
|
||||
"""
|
||||
|
||||
errors: list[str] = []
|
||||
warnings: list[str] = []
|
||||
valid = True
|
||||
|
||||
seen_ids: set[str] = set()
|
||||
seen_sequences: dict[tuple[Any, str], set[int]] = {}
|
||||
|
||||
for idx, node in enumerate(nodes):
|
||||
if not isinstance(node, dict):
|
||||
errors.append(f"node[{idx}] is not a dict (got {type(node).__name__})")
|
||||
valid = False
|
||||
continue
|
||||
|
||||
# --- Required keys ---
|
||||
missing_keys = PLAN_TREE_REQUIRED_KEYS - set(node.keys())
|
||||
if missing_keys:
|
||||
errors.append(
|
||||
f"node[{idx}] missing required key(s): {sorted(missing_keys)}"
|
||||
)
|
||||
valid = False
|
||||
|
||||
# --- decision_id ---
|
||||
did = node.get("decision_id")
|
||||
if did is not None:
|
||||
if not isinstance(did, str) or not _ULID_RE.match(did):
|
||||
errors.append(
|
||||
f"node[{idx}] 'decision_id' must be a valid ULID, got {did!r}"
|
||||
)
|
||||
valid = False
|
||||
elif did in seen_ids:
|
||||
errors.append(f"node[{idx}] duplicate decision_id: {did}")
|
||||
valid = False
|
||||
else:
|
||||
seen_ids.add(did)
|
||||
|
||||
# --- type ---
|
||||
node_type = node.get("type")
|
||||
if node_type is not None and (
|
||||
not isinstance(node_type, str) or not node_type.strip()
|
||||
):
|
||||
errors.append(f"node[{idx}] 'type' must be a non-empty string")
|
||||
valid = False
|
||||
|
||||
# --- sequence ---
|
||||
seq = node.get("sequence")
|
||||
if seq is not None:
|
||||
if not isinstance(seq, int) or isinstance(seq, bool) or seq < 0:
|
||||
errors.append(
|
||||
f"node[{idx}] 'sequence' must be a non-negative "
|
||||
f"integer, got {seq!r}"
|
||||
)
|
||||
valid = False
|
||||
else:
|
||||
parent_key = node.get("parent_decision_id", None)
|
||||
decision_id = node.get("decision_id", "")
|
||||
sibling_key = (parent_key, "ANY") # track per-parent, not per-node
|
||||
if sibling_key in seen_sequences and seq in seen_sequences[sibling_key]:
|
||||
errors.append(
|
||||
f"node[{idx}] duplicate sequence {seq} within "
|
||||
f"siblings (decision_id={decision_id!r})"
|
||||
)
|
||||
valid = False
|
||||
else:
|
||||
known_seqs = seen_sequences.get(sibling_key, set())
|
||||
known_seqs.add(seq)
|
||||
seen_sequences[sibling_key] = known_seqs
|
||||
|
||||
# --- question ---
|
||||
question = node.get("question")
|
||||
if question is not None and (
|
||||
not isinstance(question, str) or not question.strip()
|
||||
):
|
||||
errors.append(f"node[{idx}] 'question' must be a non-empty string")
|
||||
valid = False
|
||||
|
||||
# --- children ---
|
||||
children = node.get("children")
|
||||
if children is not None:
|
||||
if not isinstance(children, list):
|
||||
errors.append(
|
||||
f"node[{idx}] 'children' must be a list, "
|
||||
f"got {type(children).__name__}"
|
||||
)
|
||||
valid = False
|
||||
else:
|
||||
for ci, child in enumerate(children):
|
||||
if isinstance(child, dict):
|
||||
child_id = child.get("decision_id")
|
||||
if child_id and not _ULID_RE.match(str(child_id)):
|
||||
errors.append(
|
||||
f"node[{idx}].children[{ci}] 'decision_id' "
|
||||
f"must be a valid ULID, got {child_id!r}"
|
||||
)
|
||||
valid = False
|
||||
|
||||
return {
|
||||
"valid": valid,
|
||||
"node_count": len(nodes),
|
||||
"errors": errors,
|
||||
"warnings": warnings,
|
||||
}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Decision CLI dict validator
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def validate_decision_dict(d: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Validate a decision CLI output dictionary.
|
||||
|
||||
Checks the output of :meth:`Decision.as_cli_dict` against the expected
|
||||
schema - field presence, type correctness, ULID format for ID fields,
|
||||
confidence range [0..1], and boolean fields.
|
||||
|
||||
Args:
|
||||
d: A dict returned by ``decision.as_cli_dict()``.
|
||||
|
||||
Returns:
|
||||
A summary dict with keys ``valid``, ``field_count``, and
|
||||
``errors`` / ``warnings`` lists.
|
||||
|
||||
Raises:
|
||||
ValidationError: If the dict fails structural validation.
|
||||
"""
|
||||
|
||||
errors: list[str] = []
|
||||
warnings: list[str] = []
|
||||
valid = True
|
||||
|
||||
for field in DECISION_CLI_REQUIRED_FIELDS:
|
||||
if field not in d:
|
||||
errors.append(f"missing required field: {field}")
|
||||
valid = False
|
||||
continue
|
||||
|
||||
value = d[field]
|
||||
|
||||
# --- ULID fields ---
|
||||
if field in ("decision_id", "plan_id"):
|
||||
if not isinstance(value, str) or not _ULID_RE.match(value):
|
||||
errors.append(f"'{field}' must be a valid ULID, got {value!r}")
|
||||
valid = False
|
||||
|
||||
# --- confidence range ---
|
||||
elif field == "confidence":
|
||||
if value is not None and isinstance(value, float):
|
||||
if value < 0.0 or value > 1.0:
|
||||
errors.append(f"'{field}' must be in range [0.0, 1.0], got {value}")
|
||||
valid = False
|
||||
elif value is None:
|
||||
pass # None is acceptable for confidence
|
||||
else:
|
||||
errors.append(
|
||||
f"'{field}' must be float or None, got {type(value).__name__}"
|
||||
)
|
||||
valid = False
|
||||
|
||||
# --- parent field special handling (ULID or "(root)") ---
|
||||
elif field == "parent":
|
||||
if not isinstance(value, str):
|
||||
errors.append(f"'{field}' must be a string, got {type(value).__name__}")
|
||||
valid = False
|
||||
elif value != "(root)" and not _ULID_RE.match(value):
|
||||
errors.append(f"'{field}' must be a ULID or '(root)', got {value!r}")
|
||||
valid = False
|
||||
|
||||
# --- boolean fields ---
|
||||
elif field in ("is_correction", "superseded"):
|
||||
if not isinstance(value, bool):
|
||||
errors.append(
|
||||
f"'{field}' must be a boolean, got {type(value).__name__}"
|
||||
)
|
||||
valid = False
|
||||
|
||||
# --- type fields (str) ---
|
||||
elif field in ("type", "question", "chosen"):
|
||||
if not isinstance(value, str) or not value.strip():
|
||||
errors.append(f"'{field}' must be a non-empty string")
|
||||
valid = False
|
||||
|
||||
# --- sequence (int) ---
|
||||
elif not (isinstance(value, int) and not isinstance(value, bool)):
|
||||
errors.append(f"'{field}' must be an integer, got {type(value).__name__}")
|
||||
valid = False
|
||||
|
||||
return {
|
||||
"valid": valid,
|
||||
"field_count": len(d),
|
||||
"errors": errors,
|
||||
"warnings": warnings,
|
||||
}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# StructuredOutput validator
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def validate_structured_output(
|
||||
output: dict[str, Any],
|
||||
) -> dict[str, Any]:
|
||||
"""Validate a StructuredOutput envelope.
|
||||
|
||||
Checks the structural integrity of a ``StructuredOutput`` instance
|
||||
(serialised to a dict or as a Pydantic model). Required fields are
|
||||
``command``, ``session_id``, and ``status``. ``exit_code`` must be
|
||||
a non-negative integer. ``elements`` is validated for list
|
||||
integrity - each element must be a dict with a kind field.
|
||||
|
||||
Args:
|
||||
output: A StructuredOutput (dict).
|
||||
|
||||
Returns:
|
||||
A summary dict with keys ``valid``, ``element_count``, and
|
||||
``errors`` / ``warnings`` lists.
|
||||
|
||||
Raises:
|
||||
ValidationError: If the envelope fails structural validation.
|
||||
"""
|
||||
|
||||
errors: list[str] = []
|
||||
warnings: list[str] = []
|
||||
valid = True
|
||||
|
||||
# --- command (str, non-empty) ---
|
||||
command = output.get("command")
|
||||
if not isinstance(command, str) or not command.strip():
|
||||
errors.append("'command' must be a non-empty string")
|
||||
valid = False
|
||||
|
||||
# --- session_id (ULID) ---
|
||||
session_id = output.get("session_id")
|
||||
if not isinstance(session_id, str) or not _ULID_RE.match(session_id):
|
||||
errors.append(f"'session_id' must be a valid ULID, got '{session_id!r}'")
|
||||
valid = False
|
||||
|
||||
# --- status (membership) ---
|
||||
status = output.get("status")
|
||||
if not isinstance(status, str) or status not in VALID_STATUSES:
|
||||
errors.append(
|
||||
f"'status' must be one of {sorted(VALID_STATUSES)}, got '{status!r}'"
|
||||
)
|
||||
valid = False
|
||||
|
||||
# --- exit_code (non-negative int) ---
|
||||
exit_code = output.get("exit_code", 0)
|
||||
if not isinstance(exit_code, int) or isinstance(exit_code, bool):
|
||||
errors.append(
|
||||
f"'exit_code' must be a non-negative integer, "
|
||||
f"got {type(exit_code).__name__}"
|
||||
)
|
||||
valid = False
|
||||
elif exit_code < 0:
|
||||
errors.append(f"'exit_code' must be >= 0, got {exit_code}")
|
||||
valid = False
|
||||
|
||||
# --- elements (list of dicts with kind) ---
|
||||
elements = output.get("elements", [])
|
||||
if not isinstance(elements, list):
|
||||
errors.append(f"'elements' must be a list, got {type(elements).__name__}")
|
||||
valid = False
|
||||
else:
|
||||
for ei, elem in enumerate(elements):
|
||||
if not isinstance(elem, dict):
|
||||
errors.append(
|
||||
f"elements[{ei}] is not a dict (got {type(elem).__name__})"
|
||||
)
|
||||
valid = False
|
||||
elif "kind" not in elem:
|
||||
errors.append(f"elements[{ei}] missing 'kind' field")
|
||||
valid = False
|
||||
|
||||
return {
|
||||
"valid": valid,
|
||||
"element_count": len(elements) if isinstance(elements, list) else 0,
|
||||
"errors": errors,
|
||||
"warnings": warnings,
|
||||
}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Unified dispatcher
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
_TARGET_TYPE_MAP: dict[str, Callable[[Any], dict[str, Any]]] = {
|
||||
# --- Plan tree targets ---
|
||||
"plan_tree": validate_plan_tree,
|
||||
"plan-tree": validate_plan_tree,
|
||||
"decisions": validate_plan_tree,
|
||||
"decision-tree": validate_plan_tree,
|
||||
"tree": validate_plan_tree,
|
||||
# --- Decision CLI targets ---
|
||||
"decision": validate_decision_dict,
|
||||
"decision_cli": validate_decision_dict,
|
||||
"decision-cli": validate_decision_dict,
|
||||
"cli_dict": validate_decision_dict,
|
||||
"as_cli_dict": validate_decision_dict,
|
||||
# --- StructuredOutput targets ---
|
||||
"structured_output": validate_structured_output,
|
||||
"structured-output": validate_structured_output,
|
||||
"session_output": validate_structured_output,
|
||||
"output_session": validate_structured_output,
|
||||
}
|
||||
|
||||
|
||||
def validate_structured_component_output(
|
||||
target_type: str,
|
||||
data: Any,
|
||||
) -> dict[str, Any]:
|
||||
"""Validate output by routing to the appropriate validator.
|
||||
|
||||
Dispatches validation to a registered handler based on
|
||||
``target_type``. Unknown types raise :class:`ValidationError`.
|
||||
|
||||
Args:
|
||||
target_type: A string identifying the output component type.
|
||||
Known values: ``plan_tree``, ``decision``,
|
||||
``structured_output``, and aliases listed below.
|
||||
|
||||
data: The raw data to validate. Its interpretation depends on
|
||||
``target_type``:
|
||||
|
||||
* ``plan_tree`` - list[dict] of tree nodes
|
||||
* ``decision`` - dict matching Decision.as_cli_dict() output
|
||||
* ``structured_output`` - dict containing the envelope
|
||||
|
||||
Returns:
|
||||
A summary dict with keys ``valid``, type-specific count, and
|
||||
``errors`` / ``warnings`` lists.
|
||||
|
||||
Raises:
|
||||
ValidationError: If ``target_type`` is unknown or validation fails.
|
||||
"""
|
||||
|
||||
validator = _TARGET_TYPE_MAP.get(target_type)
|
||||
if validator is None:
|
||||
available = sorted(_TARGET_TYPE_MAP.keys())
|
||||
raise ValidationError(
|
||||
f"Unknown target_type '{target_type}'. Available targets: {available}",
|
||||
errors=[f"valid types: {', '.join(available)}"],
|
||||
)
|
||||
|
||||
result = validator(data)
|
||||
return result
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Public API
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
__all__ = [
|
||||
"DECISION_CLI_REQUIRED_FIELDS",
|
||||
"PLAN_TREE_REQUIRED_KEYS",
|
||||
"ULID_PATTERN",
|
||||
"VALID_STATUSES",
|
||||
"ValidationError",
|
||||
"ValidationWarning",
|
||||
"validate_decision_dict",
|
||||
"validate_plan_tree",
|
||||
"validate_structured_component_output",
|
||||
"validate_structured_output",
|
||||
]
|
||||
Reference in New Issue
Block a user