CHANGELOG.md

@@ -5,6 +5,14 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 Changed `wf10_batch.robot` to be less likely to create files, and
 `plan_generation_graph.robot` to give more test answers.
 
+- Implemented structural component output validation (issue #8164): replaces
+  exact character matching with structural checking for plan tree nodes, decision
+  CLI dicts, and structured session snapshots.  Validators enforce required field
+  presence, correct types (ULID, int, string, bool), structural relationships
+  (parent-child chains, sequence ordering), and produce actionable error messages
+  identifying the exact component path that failed validation.  Full Behave BDD
+  test coverage added via `features/structural_validation.feature`.
+
 - Hardened the TDD bug-fix quality gate for issue #629: PR parsing now
   requires whole-word closing keywords (avoids false positives like
   "prefixes #12"), TDD bug tag discovery now uses exact token matching

src/cleveragents/cli/output/__init__.py

@@ -204,6 +204,18 @@ from cleveragents.cli.output.selection import (
     select_materializer,
 )
 from cleveragents.cli.output.session import OutputSession, StructuredOutput
+from cleveragents.cli.output.validators import (
+    DECISION_CLI_KEYS,
+    PLAN_TREE_NODE_KEYS,
+    STRUCTURED_OUTPUT_KEYS,
+    VALID_STATUS_VALUES,
+    OutputValidationError,
+    OutputValidationResult,
+    validate_decision_dict,
+    validate_plan_tree,
+    validate_structured_component_output,
+    validate_structured_output,
+)
 
 __all__ = [
     "MAX_ELEMENTS_PER_SESSION",
@@ -257,4 +269,8 @@ __all__ = [
     "detect_terminal_capabilities",
     "select_materializer",
     "strip_terminal_escapes",
+    "validate_decision_dict",
+    "validate_plan_tree",
+    "validate_structured_component_output",
+    "validate_structured_output",
 ]

src/cleveragents/cli/output/validators.py

@@ -0,0 +1,419 @@
+"""Structural component output validation for CleverAgents CLI output.
+
+Replaces exact character matching with structural component checking.
+Validates that required fields are present with correct types, that
+structural relationships (parent-child chains, ordering) hold, and
+that minor formatting differences do not cause validation failures.
+
+Based on the v3.2.0 milestone specification:
+    "Output validation is flexible - checks structural components,
+     not exact character matching."
+    (docs/specification.md Testing Strategy)
+
+Validation targets
+------------------
+- **Plan tree output** (from ``build_decision_tree(): list[dict]``):
+  Each node must contain required keys with valid values.  Children
+  form a proper hierarchy.  Optional fields are present when indicated.
+
+- **Decision CLI dict** (from ``Decision.as_cli_dict() -> dict``):
+  Must match the stable schema produced by the Decision domain model;
+  missing keys, wrong types, and inconsistent correction metadata are
+  all caught with actionable error messages.
+
+- **Structured session snapshots** (from ``OutputSession.snapshot()/close()``):
+  The ``StructuredOutput`` envelope is validated for structural integrity:
+  required fields present, elements list well-formed, status consistent
+  with exit code.
+
+Error messages
+--------------
+Every validation failure identifies the exact path to the problematic
+component and describes which required field or constraint was violated,
+making it actionable for developers fixing test cases or production bugs.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Any
+
+
+class OutputValidationError(Exception):
+    """Raised when structural component validation fails."""
+
+    def __init__(
+        self,
+        message: str,
+        *,
+        errors: list[str] | None = None,
+        schema_name: str = "unknown",
+        path: str = "<root>",
+    ) -> None:
+        super().__init__(message)
+        self.errors = errors or []
+        self.schema_name = schema_name
+        self.path = path
+
+
+class OutputValidationResult:
+    """Result of a structural validation pass."""
+
+    def __init__(
+        self,
+        *,
+        valid: bool = True,
+        message: str = "",
+        errors: list[str] | None = None,
+        schema_name: str = "unknown",
+        path: str = "<root>",
+    ) -> None:
+        self.valid = valid
+        self.message = message
+        self.errors = errors or []
+        self.schema_name = schema_name
+        self.path = path
+
+    def __bool__(self) -> bool:
+        return self.valid
+
+
+# Schema definitions
+PLAN_TREE_NODE_KEYS: set[str] = {
+    "decision_id",
+    "type",
+    "sequence",
+    "question",
+    "children",
+}
+DECISION_CLI_KEYS: set[str] = {
+    "decision_id",
+    "plan_id",
+    "type",
+    "sequence",
+    "question",
+    "chosen",
+    "confidence",
+    "parent",
+    "is_correction",
+    "superseded",
+}
+STRUCTURED_OUTPUT_KEYS: set[str] = {
+    "command",
+    "session_id",
+    "status",
+    "exit_code",
+}
+VALID_STATUS_VALUES: frozenset[str] = frozenset({"ok", "error", "timeout"})
+
+_VALID_DECISION_TYPES: frozenset[str] = frozenset(
+    {
+        "prompt_definition",
+        "invariant_enforced",
+        "strategy_choice",
+        "implementation_choice",
+        "resource_selection",
+        "subplan_spawn",
+        "subplan_parallel_spawn",
+        "tool_invocation",
+        "error_recovery",
+        "validation_response",
+        "user_intervention",
+    }
+)
+
+
+def _is_valid_ulid(value: Any) -> bool:
+    """Return True if *value* is a valid ULID string (26 chars)."""
+    if not isinstance(value, str) or len(value) != 26:
+        return False
+    ulid_chars = set("0123456789ABCDEFGHJKMNPQRSTVWXYZ")
+    return all(c in ulid_chars for c in value)
+
+
+def _is_valid_sequence(value: Any) -> bool:
+    """Return True if *value* is a valid non-negative int."""
+    return isinstance(value, int) and value >= 0
+
+
+def _is_valid_confidence(value: Any) -> bool:
+    """Return True if *value* is None or float in [0.0, 1.0]."""
+    if value is None:
+        return True
+    try:
+        f = float(value)
+        return 0.0 <= f <= 1.0
+    except (TypeError, ValueError):
+        return False
+
+
+def _path(base: str, comp: str) -> str:
+    """Join path strings."""
+    if base == "<root>":
+        return comp
+    return f"{base}.{comp}"
+
+
+# --- Plan tree validator ---
+
+
+def validate_plan_tree(
+    data: Any, *, check_ordering: bool = True,
+) -> OutputValidationResult:
+    """Validate a plan tree output (list of root node dicts)."""
+    if not isinstance(data, list):
+        return OutputValidationResult(
+            valid=False, message="Plan tree root must be a list",
+            errors=[f"Expected list, got {type(data).__name__}"],
+            schema_name="plan_tree_root",
+        )
+    if len(data) == 0:
+        return OutputValidationResult(valid=True, message="Empty plan tree is valid")
+
+    errors = []
+    for i, node in enumerate(data):
+        errors.extend(_check_node(node, _path("<root>", f"[{i}]")))
+
+    if check_ordering and len(data) >= 2:
+        seq_0: int | None = None
+        seq_1: int | None = None
+        for i in range(1, len(data)):
+            prev_seq = data[i - 1].get("sequence")
+            curr_seq = data[i].get("sequence")
+            if isinstance(prev_seq, int) and isinstance(curr_seq, int):
+                seq_0 = prev_seq
+                seq_1 = curr_seq
+                if seq_0 > seq_1:
+                    errors.append(f"[{i}] sequence {prev_seq} > {curr_seq}")
+
+    _validate_children(data, "<root>", check_ordering, errors)
+    if errors:
+        msg = (
+            f"Plan tree failed ({len(errors)} error{'s' if len(errors)>1 else ''})"
+        )
+        return OutputValidationResult(
+            valid=False, message=msg, errors=errors, schema_name="plan_tree_root",
+        )
+    n = len(data)
+    msg = f"Plan tree valid ({n} root node{'s' if n > 1 else ''})"
+    return OutputValidationResult(
+        valid=True, message=msg, schema_name="plan_tree_root",
+    )
+
+
+def _check_node(node: Any, prefix: str) -> list[str]:
+    """Validate a single plan tree node. Returns error strings."""
+    errors = []
+    if not isinstance(node, dict):
+        return [f"{prefix}: expected dict, got {type(node).__name__}"]
+    missing = sorted(PLAN_TREE_NODE_KEYS - set(node.keys()))
+    if missing:
+        errors.append(f"{prefix}: missing keys: {', '.join(missing)}")
+    did = node.get("decision_id")
+    if "decision_id" not in missing and did is not None and not _is_valid_ulid(did):
+        errors.append(f"{prefix}.decision_id: invalid ULID")
+    dtype = node.get("type")
+    type_missing = "type" not in set(node.keys())
+    if not type_missing and dtype is not None and dtype not in _VALID_DECISION_TYPES:
+        errors.append(f"{prefix}.type: unexpected {dtype!r}")
+    seq = node.get("sequence")
+    if (
+        not ("sequence" in set(node.keys()) or seq is None)
+    ) and not _is_valid_sequence(seq):
+        pass  # Missing already reported above
+    elif not type_missing or "sequence" not in missing:
+        seq_val = node.get("sequence")
+        s_missing = "sequence" in missing if isinstance(seq_val, int) else False
+        if (not s_missing and seq_val is not None and not _is_valid_sequence(seq_val)):
+            errors.append(f"{prefix}.sequence: expected non-negative int")
+    question = node.get("question")
+    q_miss = "question" in missing or question is None
+    if not q_miss and not isinstance(question, str):
+        errors.append(f"{prefix}.question: expected string")
+    children = node.get("children")
+    c_miss = "children" in missing or children is None
+    if not c_miss and not isinstance(children, list):
+        children_type = type(children).__name__
+        errors.append(f"{prefix}.children: expected list, got {children_type}")
+    return errors
+
+
+def _validate_children(
+    nodes: list[dict[str, Any]],
+    parent_path: str,
+    check_ordering: bool,
+    errors: list[str],
+) -> None:
+    """Recursively validate children of plan tree nodes."""
+    for idx, node in enumerate(nodes):
+        if not isinstance(node, dict):
+            continue
+        ch = node.get("children")
+        if not isinstance(ch, list):
+            continue
+        base = _path(parent_path, f"[{idx}].children")
+        ci_errors = []
+        for ci in range(len(ch)):
+            ci_errors.extend(_check_node(ch[ci], _path(base, f"[{ci}]")))
+        errors.extend(ci_errors)
+        if check_ordering and len(ch) >= 2:
+            for i in range(1, len(ch)):
+                ps = ch[i - 1].get("sequence", -1)
+                cs = ch[i].get("sequence", -1)
+                if isinstance(ps, int) and isinstance(cs, int) and ps > cs:
+                    errors.append(f"{base}[{i}] ordering violation seq {ps} > seq {cs}")
+    for idx, node in enumerate(nodes):
+        ch = node.get("children")
+        if isinstance(ch, list) and ch:
+            _validate_children(
+                ch, _path(parent_path, f"[{idx}].children"), check_ordering, errors,
+            )
+
+
+# --- Decision dict validator ---
+
+
+def validate_decision_dict(data: Any) -> OutputValidationResult:
+    """Validate a decision CLI dict."""
+    if not isinstance(data, dict):
+        return OutputValidationResult(
+        valid=False,
+        message="Decision output must be a dict",
+        errors=[f"Expected dict, got {type(data).__name__}"],
+        schema_name="decision_cli_dict",
+    )
+
+    missing = sorted(DECISION_CLI_KEYS - set(data.keys()))
+    errors: list[str] = []
+    if missing:
+        errors.append(f"missing keys: {', '.join(missing)}")
+
+    for key, check in [
+        ("decision_id", _is_valid_ulid),
+        ("plan_id", _is_valid_ulid),
+    ]:
+        val = data.get(key)
+        if not check(val):
+            errors.append(f"{key}: invalid format")
+
+    dtype = data.get("type")
+    if not isinstance(dtype, str) or not dtype:
+        errors.append("type: expected string")
+    elif dtype not in _VALID_DECISION_TYPES:
+        errors.append(f"type: unexpected {dtype!r}")
+
+    seq = data.get("sequence")
+    if not _is_valid_sequence(seq):
+        errors.append("sequence: expected non-negative int")
+
+    _STR = str
+    for key, _check_type in [("question", _STR), ("chosen", _STR)]:
+        val = data.get(key)
+        if not isinstance(val, (str, type(None))) and val is not None:
+            errors.append(f"{key}: expected string or null")
+        elif val is None or not isinstance(val, str):
+            pass  # May be in missing list
+        elif not val.strip():
+            errors.append(f"{key}: expected non-empty")
+
+    confidence = data.get("confidence")
+    if confidence is not None and not _is_valid_confidence(confidence):
+        errors.append("confidence: expected 0.0-1.0")
+
+    parent = data.get("parent")
+    if isinstance(parent, str) and parent != "(root)" and not _is_valid_ulid(parent):
+        errors.append("parent: expected ULID or '(root)'")
+    elif not isinstance(parent, str) and parent is not None:
+        errors.append("parent: expected string")
+
+    ic = data.get("is_correction")
+    if not isinstance(ic, bool):
+        errors.append(f"is_correction: expected bool, got {type(ic).__name__}")
+    s = data.get("superseded")
+    if not isinstance(s, bool):
+        errors.append(f"superseded: expected bool, got {type(s).__name__}")
+
+    return OutputValidationResult(
+        valid=len(errors) == 0, message="decision_cli_dict",
+        errors=errors, schema_name="decision_cli_dict",
+    )
+
+
+# --- Structured output validator ---
+
+
+def validate_structured_output(data: Any) -> OutputValidationResult:
+    """Validate a structured session output envelope."""
+    if not isinstance(data, dict):
+        return OutputValidationResult(
+        valid=False,
+        message="Structured output must be a dict",
+        errors=[f"Expected dict, got {type(data).__name__}"],
+        schema_name="structured_output_root",
+    )
+
+    missing = sorted(STRUCTURED_OUTPUT_KEYS - set(data.keys()))
+    errors: list[str] = []
+    if missing:
+        errors.append(f"missing keys: {', '.join(missing)}")
+
+    cmd = data.get("command")
+    if "command" not in missing and cmd is not None and not isinstance(cmd, str):
+        errors.append("command: expected string")
+
+    sid = data.get("session_id")
+    if "session_id" not in missing and sid is not None and not _is_valid_ulid(sid):
+        errors.append("session_id: invalid ULID")
+
+    status = data.get("status")
+    smissing = "status" not in set(data.keys()) or status is None
+    if not smissing and isinstance(status, str) and status not in VALID_STATUS_VALUES:
+        errors.append(f"status: expected {sorted(VALID_STATUS_VALUES)}, got {status!r}")
+
+    ec = data.get("exit_code")
+    if "exit_code" not in missing and ec is not None and not isinstance(ec, int):
+        errors.append("exit_code: expected int")
+
+    elements = data.get("elements")
+    if "elements" in set(data.keys()) or elements is not None:
+        if isinstance(elements, list):
+            for ei, elem in enumerate(elements):
+                if not isinstance(elem, dict):
+                    errors.append(f"elements[{ei}]: expected dict")
+                elif "type" not in elem:
+                    errors.append(f"elements[{ei}]: missing 'type'")
+        elif elements is not None:
+            errors.append(f"elements: expected list, got {type(elements).__name__}")
+
+    timing = data.get("timing")
+    if "timing" in set(data.keys()) or timing is not None:
+        if isinstance(timing, dict):
+            for tk, tv in timing.items():
+                if isinstance(tv, str):
+                    try:
+                        datetime.fromisoformat(tv)
+                    except (ValueError, TypeError):
+                        errors.append(f"timing.{tk}: expected ISO-8601")
+        elif timing is not None:
+            errors.append("timing: expected dict when present")
+
+    return OutputValidationResult(
+        valid=len(errors) == 0, message="structured_output",
+        errors=errors, schema_name="structured_output_root",
+    )
+
+
+# --- Unified validator ---
+
+
+def validate_structured_component_output(
+    data: Any, target_type: str,
+) -> OutputValidationResult:
+    """Validate structured output for a given target type."""
+    if target_type == "plan_tree":
+        return validate_plan_tree(data)
+    if target_type == "decision_dict":
+        return validate_decision_dict(data)
+    if target_type == "structured_output":
+        return validate_structured_output(data)
+    raise ValueError(f"Unknown target_type '{target_type}'")