Verify a plan file with comprehensive semantic checks.
(self, plan_path: Path)
| 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 | |
| 149 | result = VerificationResult(is_valid=True, plan_data=data) |
| 150 | result.warnings = warnings |
| 151 | return result |
| 152 | |
| 153 | # ── Per-step-type semantic checks ──────────────────────────────── |
| 154 |
no test coverage detected