Semantic verifier for AgentSPEX YAML workflow plans. Performs: 1. YAML syntax check (via yaml.safe_load inside YAMLTaskParser) 2. Required field validation (name, goal, workflow) 3. Per-step-type semantic checks (required sub-fields, constraints) 4. Variable flow analysis (save_
| 89 | |
| 90 | |
| 91 | class PlanVerifier: |
| 92 | """Semantic verifier for AgentSPEX YAML workflow plans. |
| 93 | |
| 94 | Performs: |
| 95 | 1. YAML syntax check (via yaml.safe_load inside YAMLTaskParser) |
| 96 | 2. Required field validation (name, goal, workflow) |
| 97 | 3. Per-step-type semantic checks (required sub-fields, constraints) |
| 98 | 4. Variable flow analysis (save_as definitions vs {{variable}} references) |
| 99 | 5. Tool name validation against available MCP tools (warnings) |
| 100 | 6. Parallel/gather constraint checking |
| 101 | |
| 102 | Formal verification (Lean4) will be integrated when available. |
| 103 | """ |
| 104 | |
| 105 | def __init__(self, available_tool_names: Optional[set[str]] = None): |
| 106 | self._parser = YAMLTaskParser() |
| 107 | self._available_tools = available_tool_names or set() |
| 108 | |
| 109 | def verify(self, plan_path: Path) -> VerificationResult: |
| 110 | """Verify a plan file with comprehensive semantic checks.""" |
| 111 | try: |
| 112 | data = self._parser.load_task(str(plan_path)) |
| 113 | except FileNotFoundError: |
| 114 | return VerificationResult( |
| 115 | is_valid=False, error=f"File not found: {plan_path}" |
| 116 | ) |
| 117 | except ValueError as e: |
| 118 | return VerificationResult(is_valid=False, error=str(e)) |
| 119 | except Exception as e: |
| 120 | return VerificationResult(is_valid=False, error=f"Parse error: {e}") |
| 121 | |
| 122 | errors = [] |
| 123 | warnings = [] |
| 124 | |
| 125 | # Collect defined variables from parameters + save_as in workflow |
| 126 | defined_vars = set() |
| 127 | params = data.get("parameters", {}) |
| 128 | if isinstance(params, dict): |
| 129 | defined_vars.update(params.keys()) |
| 130 | # prev_output is always available |
| 131 | defined_vars.add("prev_output") |
| 132 | |
| 133 | # Run all checks |
| 134 | warnings.extend(self._check_tool_names(data)) |
| 135 | workflow = data.get("workflow", []) |
| 136 | self._check_steps(workflow, errors, warnings, defined_vars, path="workflow") |
| 137 | |
| 138 | if errors: |
| 139 | error_summary = "; ".join(errors[:5]) |
| 140 | if len(errors) > 5: |
| 141 | error_summary += f" ... and {len(errors) - 5} more error(s)" |
| 142 | return VerificationResult( |
| 143 | is_valid=False, |
| 144 | error=error_summary, |
| 145 | plan_data=data, |
| 146 | warnings=warnings, |
| 147 | ) |
| 148 |