diff --git a/AGENTS.md b/AGENTS.md index cecd2d8..dc28206 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -23,6 +23,7 @@ cf # Interactive tool picker - `tool.py` - Tool/step dataclasses, including delegated `ToolStep` context and `McpStep` - `preflight.py` - Shared contract, dependency, secret-pattern, and registry-similarity analysis - `contract_testing.py` - Deterministic JSON Schema input generation and side-effect-safe conformance checks +- `schema_compat.py` - Conservative producer-to-consumer JSON Schema compatibility analysis - `runner.py` - Step execution, variable substitution, nested authorization and delegation - `providers.py` - AI providers, auto-discovery, fallback chains, and tool/MCP allowlists - `skills.py` - Per-provider Agent Skills loading and validation diff --git a/CLAUDE.md b/CLAUDE.md index 4b8feef..c20e804 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -87,6 +87,11 @@ Inspect also shows inferred contract proposals and runs deterministic contract conformance when both tool schemas are explicit. Prompt outputs are synthesized from their schemas; code, nested-tool, and MCP steps are reported as unsupported instead of being executed implicitly. +Use `cmdforge inspect --save-baseline` to explicitly promote passing +evidence to `conformance.json`. Later inspections and local publish dry-runs +rerun the stored synthetic inputs and report state, coverage, contract, and +normalized output changes. ToolStep preflight also compares the values the +runner supplies with the called tool's input contract. ### Step Types diff --git a/README.md b/README.md index e79c082..b75c775 100644 --- a/README.md +++ b/README.md @@ -137,6 +137,7 @@ cmdforge test mytool # Test with mock provider cmdforge check mytool # Check dependencies (meta-tools) cmdforge inspect mytool # Preflight, contract proposals, safe conformance tests cmdforge inspect mytool --registry # Also find similar registry tools +cmdforge inspect mytool --save-baseline # Approve passing evidence for regression checks cmdforge refresh # Update executable wrappers cmdforge docs mytool # View/create tool documentation diff --git a/src/cmdforge/cli/__init__.py b/src/cmdforge/cli/__init__.py index ffa44b2..4ea0f43 100644 --- a/src/cmdforge/cli/__init__.py +++ b/src/cmdforge/cli/__init__.py @@ -97,6 +97,10 @@ def main(): "--registry", action="store_true", help="Include similar tools from the configured registry", ) + p_inspect.add_argument( + "--save-baseline", action="store_true", + help="Save the current passing conformance evidence as the baseline", + ) p_inspect.set_defaults(func=cmd_inspect) # 'check' command @@ -593,6 +597,54 @@ def cmd_inspect(args): ) print() + if report.regression: + regression = report.regression + status = ( + "REGRESSION" if regression.get("has_regressions") + else "REVIEW" if regression.get("contract_changed") + else "STABLE" + ) + print( + f"Regression comparison: {status} — " + f"{regression.get('summary', 'no changes')}" + ) + print( + f" Baseline {regression.get('baseline_version') or '(unversioned)'} " + f"-> current {regression.get('current_version') or '(unversioned)'}" + ) + print() + + if report.compatibility: + print(f"ToolStep compatibility ({len(report.compatibility)}):") + for finding in report.compatibility: + print( + f" {str(finding.get('state', 'unknown')).upper()}: " + f"step {finding.get('step', '?')} -> " + f"{finding.get('tool', 'unknown')} — {finding.get('detail', '')}" + ) + print() + + if getattr(args, "save_baseline", False): + from ..contract_testing import ( + ConformanceReport, + ConformanceResult, + save_conformance_baseline, + ) + + conformance = ConformanceReport([ + ConformanceResult(**result) for result in report.generated_tests + ]) + if not tool.path: + print("Error: Tool has no local directory for a baseline.", file=sys.stderr) + return 1 + try: + path = save_conformance_baseline(tool, conformance, tool.path.parent) + except ValueError as exc: + print(f"Error: Baseline not saved: {exc}", file=sys.stderr) + return 1 + print(f"Saved conformance baseline: {path}") + print() + if not report.errors and not report.warnings and not report.suggestions: print("No issues found.") diff --git a/src/cmdforge/cli/registry_commands.py b/src/cmdforge/cli/registry_commands.py index 223d946..4c3d255 100644 --- a/src/cmdforge/cli/registry_commands.py +++ b/src/cmdforge/cli/registry_commands.py @@ -387,6 +387,28 @@ def _print_preflight_sections(report: dict, prefix: str = "") -> None: f"{result.get('step', 'unknown')} — {result.get('detail', '')}" ) + regression = report.get("regression") or {} + if regression: + marker = ( + "REGRESSION" if regression.get("has_regressions") + else "REVIEW" if regression.get("contract_changed") + else "STABLE" + ) + print( + f"{prefix}Regression comparison: {marker} — " + f"{regression.get('summary', 'no changes')}" + ) + + compatibility = report.get("compatibility") or [] + if compatibility: + print(f"{prefix}ToolStep compatibility ({len(compatibility)}):") + for finding in compatibility: + print( + f" {str(finding.get('state', 'unknown')).upper()}: " + f"step {finding.get('step', '?')} -> " + f"{finding.get('tool', 'unknown')} — {finding.get('detail', '')}" + ) + def _print_registry_suggestions(suggestions: dict) -> None: """Render registry-specific category, similarity, and scrutiny evidence.""" @@ -494,8 +516,10 @@ def _cmd_registry_publish(args): from ..preflight import analyze_tool from ..tool import Tool try: + local_tool = Tool.from_dict(data) + local_tool.path = config_path local_report = analyze_tool( - Tool.from_dict(data), include_contract_tests=True + local_tool, include_contract_tests=True ) except (KeyError, TypeError, ValueError) as exc: print(f"Local preflight error: {exc}", file=sys.stderr) diff --git a/src/cmdforge/contract_testing.py b/src/cmdforge/contract_testing.py index 0490630..53ab9a8 100644 --- a/src/cmdforge/contract_testing.py +++ b/src/cmdforge/contract_testing.py @@ -10,12 +10,16 @@ reported as unsupported instead of being run implicitly. from __future__ import annotations import copy +import hashlib import json import math +import os import re +import tempfile from dataclasses import asdict, dataclass, field from datetime import date, datetime, timezone -from typing import Any, Dict, List, Literal +from pathlib import Path +from typing import Any, Dict, List, Literal, Optional from uuid import UUID from .tool import CodeStep, McpStep, PromptStep, Tool, ToolStep @@ -34,6 +38,11 @@ class ConformanceResult: step: str state: ConformanceState detail: str = "" + case_id: str = "" + input_value: Any = None + input_fingerprint: str = "" + output_shape: Optional[Dict[str, Any]] = None + output_fingerprint: str = "" def __post_init__(self) -> None: if self.state not in _VALID_STATES: @@ -42,7 +51,7 @@ class ConformanceResult: def is_pass(self) -> bool: return self.state == "passed" - def to_dict(self) -> Dict[str, str]: + def to_dict(self) -> Dict[str, Any]: return asdict(self) @@ -84,7 +93,9 @@ class ConformanceReport: } -def run_contract_tests(tool: Tool) -> ConformanceReport: +def run_contract_tests( + tool: Tool, test_inputs: Optional[List[Any]] = None +) -> ConformanceReport: """Run safe structural conformance tests for a tool. This is deliberately not a semantic quality test. Original CodeStep, @@ -113,10 +124,20 @@ def run_contract_tests(tool: Tool) -> ConformanceReport: f"Safe conformance mode does not execute {unsafe}", ) + generation_issue = None try: - test_cases = _generate_inputs(tool.input_schema) + generated_cases = _generate_inputs(tool.input_schema) except UnsupportedContract as exc: - return _single_result("input-generation", "unsupported", str(exc)) + if not test_inputs: + return _single_result("input-generation", "unsupported", str(exc)) + generated_cases = [] + generation_issue = str(exc) + + test_cases = list(copy.deepcopy(test_inputs or [])) + generated_cases + deduplicated = {} + for test_input in test_cases: + deduplicated[_fingerprint(test_input)] = test_input + test_cases = list(deduplicated.values()) if not test_cases: return _single_result( @@ -124,31 +145,65 @@ def run_contract_tests(tool: Tool) -> ConformanceReport: ) report = ConformanceReport() - for index, test_input in enumerate(test_cases, start=1): - label = f"case-{index}" + if generation_issue: + report.results.append(ConformanceResult( + "input-generation", "unsupported", generation_issue + )) + for test_input in test_cases: + input_fingerprint = _fingerprint(test_input) + case_id = f"case-{input_fingerprint[:12]}" + evidence = { + "case_id": case_id, + "input_value": copy.deepcopy(test_input), + "input_fingerprint": input_fingerprint, + } + try: + _validate_instance(test_input, tool.input_schema, "input") + except UnsupportedContract as exc: + report.results.append(ConformanceResult( + case_id, "unsupported", str(exc), **evidence + )) + continue + except ValueError as exc: + report.results.append(ConformanceResult( + case_id, "failed", f"Stored {exc}", **evidence + )) + continue try: result = _run_with_stub(tool, test_input) except UnsupportedContract as exc: report.results.append( - ConformanceResult(label, "unsupported", str(exc)) + ConformanceResult(case_id, "unsupported", str(exc), **evidence) ) except Exception as exc: report.results.append( - ConformanceResult(label, "failed", f"Execution error: {exc}") + ConformanceResult( + case_id, "failed", f"Execution error: {exc}", **evidence + ) ) else: + output_evidence = { + "output_shape": _json_shape(result), + "output_fingerprint": _fingerprint(result), + } try: _validate_instance(result, tool.output_schema, "output") except UnsupportedContract as exc: report.results.append( - ConformanceResult(label, "unsupported", str(exc)) + ConformanceResult( + case_id, "unsupported", str(exc), + **evidence, **output_evidence, + ) ) except ValueError as exc: - report.results.append(ConformanceResult(label, "failed", str(exc))) + report.results.append(ConformanceResult( + case_id, "failed", str(exc), **evidence, **output_evidence + )) else: report.results.append( ConformanceResult( - label, "passed", "Output conforms to output_schema" + case_id, "passed", "Output conforms to output_schema", + **evidence, **output_evidence, ) ) return report @@ -160,6 +215,47 @@ def _single_result( return ConformanceReport([ConformanceResult(step, state, detail)]) +def _canonical_json(value: Any) -> str: + return json.dumps( + value, sort_keys=True, separators=(",", ":"), ensure_ascii=False, + default=str, + ) + + +def _fingerprint(value: Any) -> str: + return hashlib.sha256(_canonical_json(value).encode("utf-8")).hexdigest() + + +def _json_shape(value: Any) -> Dict[str, Any]: + if value is None: + return {"type": "null"} + if isinstance(value, bool): + return {"type": "boolean"} + if isinstance(value, int): + return {"type": "integer"} + if isinstance(value, float): + return {"type": "number"} + if isinstance(value, str): + return {"type": "string"} + if isinstance(value, dict): + return { + "type": "object", + "properties": { + str(key): _json_shape(item) + for key, item in sorted(value.items(), key=lambda pair: str(pair[0])) + }, + } + if isinstance(value, list): + shapes = { + _canonical_json(_json_shape(item)): _json_shape(item) for item in value + } + return { + "type": "array", + "items": [shapes[key] for key in sorted(shapes)], + } + return {"type": type(value).__name__} + + def _generate_inputs(schema: dict) -> List[Any]: """Generate deterministic values and validate every returned candidate.""" candidates: List[Any] = [] @@ -475,3 +571,336 @@ def _validate_output(output: Any, schema: dict) -> bool: return True except ValueError: return False + + +# --------------------------------------------------------------------------- +# M8.V2: Regression comparison +# --------------------------------------------------------------------------- + +CONFORMANCE_FILE = "conformance.json" +BASELINE_FORMAT_VERSION = 1 +CONFORMANCE_ENGINE_VERSION = 1 + + +class BaselineFormatError(ValueError): + """Raised when a stored conformance baseline is corrupt or incompatible.""" + + +@dataclass +class RegressionChange: + case_id: str + kind: str + baseline_state: str = "missing" + current_state: str = "missing" + detail: str = "" + + def to_dict(self) -> Dict[str, str]: + return asdict(self) + + +@dataclass +class RegressionDelta: + """Difference between baseline and current conformance results.""" + + tool: str + baseline_version: str + current_version: str + baseline_timestamp: str + current_timestamp: str + baseline_contract_hash: str + current_contract_hash: str + regressed: List[RegressionChange] = field(default_factory=list) + improved: List[RegressionChange] = field(default_factory=list) + new_failures: List[RegressionChange] = field(default_factory=list) + new_cases: List[RegressionChange] = field(default_factory=list) + removed_cases: List[RegressionChange] = field(default_factory=list) + coverage_losses: List[RegressionChange] = field(default_factory=list) + output_changes: List[RegressionChange] = field(default_factory=list) + unchanged: List[RegressionChange] = field(default_factory=list) + + @property + def has_regressions(self) -> bool: + return any(( + self.regressed, + self.new_failures, + self.removed_cases, + self.coverage_losses, + self.output_changes, + )) + + @property + def contract_changed(self) -> bool: + return self.baseline_contract_hash != self.current_contract_hash + + @property + def summary(self) -> str: + parts = [] + if self.contract_changed: + parts.append("contract changed") + if self.regressed: + parts.append(f"{len(self.regressed)} regression(s)") + if self.new_failures: + parts.append(f"{len(self.new_failures)} new failure(s)") + if self.improved: + parts.append(f"{len(self.improved)} improvement(s)") + if self.coverage_losses: + parts.append(f"{len(self.coverage_losses)} coverage loss(es)") + if self.removed_cases: + parts.append(f"{len(self.removed_cases)} removed case(s)") + if self.output_changes: + parts.append(f"{len(self.output_changes)} output change(s)") + if self.new_cases: + parts.append(f"{len(self.new_cases)} new case(s)") + if self.unchanged: + parts.append(f"{len(self.unchanged)} unchanged case(s)") + return ", ".join(parts) if parts else "no changes" + + def to_dict(self) -> Dict[str, Any]: + return { + "tool": self.tool, + "baseline_version": self.baseline_version, + "current_version": self.current_version, + "baseline_timestamp": self.baseline_timestamp, + "current_timestamp": self.current_timestamp, + "baseline_contract_hash": self.baseline_contract_hash, + "current_contract_hash": self.current_contract_hash, + "contract_changed": self.contract_changed, + "has_regressions": self.has_regressions, + "summary": self.summary, + **{ + name: [change.to_dict() for change in getattr(self, name)] + for name in ( + "regressed", "improved", "new_failures", "new_cases", + "removed_cases", "coverage_losses", "output_changes", + "unchanged", + ) + }, + } + + +def save_conformance_baseline(tool: Tool, report: ConformanceReport, tool_dir: Path) -> Path: + """Atomically persist a successful, reproducible conformance baseline.""" + if not report.results or any( + result.state != "passed" for result in report.results + ): + raise ValueError("Only a passing conformance report can become a baseline") + if any(not result.case_id or not result.input_fingerprint for result in report.results): + raise ValueError("Baseline results must include stable case evidence") + + tool_dir = Path(tool_dir) + tool_dir.mkdir(parents=True, exist_ok=True) + path = tool_dir / CONFORMANCE_FILE + data = { + "format_version": BASELINE_FORMAT_VERSION, + "engine_version": CONFORMANCE_ENGINE_VERSION, + "tool": {"name": tool.name, "version": tool.version}, + "timestamp": datetime.now(timezone.utc).isoformat(), + "contract_hash": _contract_hash(tool), + "contracts": { + "input_schema": copy.deepcopy(tool.input_schema), + "output_schema": copy.deepcopy(tool.output_schema), + }, + "results": [r.to_dict() for r in report.results], + } + _validate_baseline_data(data) + + descriptor, temporary_name = tempfile.mkstemp( + prefix=f".{CONFORMANCE_FILE}.", dir=tool_dir + ) + try: + with os.fdopen(descriptor, "w", encoding="utf-8") as handle: + json.dump(data, handle, indent=2, sort_keys=True) + handle.write("\n") + handle.flush() + os.fsync(handle.fileno()) + os.chmod(temporary_name, 0o600) + os.replace(temporary_name, path) + finally: + if os.path.exists(temporary_name): + os.unlink(temporary_name) + return path + + +def load_conformance_baseline(tool_dir: Path) -> Optional[dict]: + """Load and validate a previously saved conformance baseline.""" + path = Path(tool_dir) / CONFORMANCE_FILE + if not path.exists(): + return None + try: + data = json.loads(path.read_text(encoding="utf-8")) + except OSError as exc: + raise BaselineFormatError(f"Could not read baseline: {exc}") from exc + except json.JSONDecodeError as exc: + raise BaselineFormatError(f"Baseline is not valid JSON: {exc}") from exc + _validate_baseline_data(data) + return data + + +def compare_conformance( + baseline: dict, + current: ConformanceReport, + tool: Tool, +) -> RegressionDelta: + """Compare stable cases, states, coverage, and normalized output evidence.""" + _validate_baseline_data(baseline) + if baseline["tool"]["name"] != tool.name: + raise BaselineFormatError( + f"Baseline belongs to '{baseline['tool']['name']}', not '{tool.name}'" + ) + base_map = {result["case_id"]: result for result in baseline["results"]} + current_map = {result.case_id: result for result in current.results if result.case_id} + + delta = RegressionDelta( + tool=tool.name, + baseline_version=baseline["tool"]["version"], + current_version=tool.version, + baseline_timestamp=baseline["timestamp"], + current_timestamp=datetime.now(timezone.utc).isoformat(), + baseline_contract_hash=baseline["contract_hash"], + current_contract_hash=_contract_hash(tool), + ) + + for case_id, current_result in current_map.items(): + base_result = base_map.get(case_id) + if base_result is None: + if current_result.state == "failed": + delta.new_failures.append(_change( + case_id, "new_failure", None, current_result + )) + else: + delta.new_cases.append(_change( + case_id, "new_case", None, current_result + )) + continue + + baseline_state = base_result["state"] + if baseline_state == "passed" and current_result.state == "failed": + delta.regressed.append(_change( + case_id, "state_regression", base_result, current_result + )) + elif baseline_state == "passed" and current_result.state in ( + "unsupported", "not_run" + ): + delta.coverage_losses.append(_change( + case_id, "coverage_loss", base_result, current_result + )) + elif baseline_state != "passed" and current_result.state == "passed": + delta.improved.append(_change( + case_id, "improved", base_result, current_result + )) + elif ( + baseline_state == "passed" + and current_result.state == "passed" + and ( + base_result.get("output_shape") != current_result.output_shape + or base_result.get("output_fingerprint") + != current_result.output_fingerprint + ) + ): + delta.output_changes.append(_change( + case_id, "output_changed", base_result, current_result + )) + else: + delta.unchanged.append(_change( + case_id, "unchanged", base_result, current_result + )) + + for case_id, baseline_result in base_map.items(): + if case_id not in current_map: + delta.removed_cases.append(_change( + case_id, "case_removed", baseline_result, None + )) + return delta + + +def _change( + case_id: str, + kind: str, + baseline: Optional[dict], + current: Optional[ConformanceResult], +) -> RegressionChange: + baseline_state = baseline.get("state", "missing") if baseline else "missing" + current_state = current.state if current else "missing" + detail = current.detail if current else "Previously recorded case is absent" + return RegressionChange( + case_id=case_id, + kind=kind, + baseline_state=baseline_state, + current_state=current_state, + detail=detail, + ) + + +def _contract_hash(tool: Tool) -> str: + return _fingerprint({ + "input_schema": tool.input_schema, + "output_schema": tool.output_schema, + }) + + +def _validate_baseline_data(data: Any) -> None: + if not isinstance(data, dict): + raise BaselineFormatError("Baseline root must be an object") + if data.get("format_version") != BASELINE_FORMAT_VERSION: + raise BaselineFormatError("Unsupported baseline format version") + if data.get("engine_version") != CONFORMANCE_ENGINE_VERSION: + raise BaselineFormatError("Unsupported conformance engine version") + tool = data.get("tool") + if not isinstance(tool, dict) or not isinstance(tool.get("name"), str): + raise BaselineFormatError("Baseline tool metadata is invalid") + if not isinstance(tool.get("version"), str): + raise BaselineFormatError("Baseline tool version is invalid") + if not isinstance(data.get("timestamp"), str): + raise BaselineFormatError("Baseline timestamp is invalid") + if not isinstance(data.get("contract_hash"), str) or not re.fullmatch( + r"[0-9a-f]{64}", data["contract_hash"] + ): + raise BaselineFormatError("Baseline contract hash is invalid") + if not isinstance(data.get("contracts"), dict): + raise BaselineFormatError("Baseline contracts are invalid") + if _fingerprint(data["contracts"]) != data["contract_hash"]: + raise BaselineFormatError("Baseline contract hash does not match contracts") + results = data.get("results") + if not isinstance(results, list): + raise BaselineFormatError("Baseline results must be an array") + seen = set() + for result in results: + if not isinstance(result, dict): + raise BaselineFormatError("Baseline result must be an object") + case_id = result.get("case_id") + if not isinstance(case_id, str) or not case_id or case_id in seen: + raise BaselineFormatError("Baseline case IDs must be unique strings") + seen.add(case_id) + if result.get("state") not in _VALID_STATES: + raise BaselineFormatError(f"Invalid state for baseline case '{case_id}'") + if not isinstance(result.get("input_fingerprint"), str) or not re.fullmatch( + r"[0-9a-f]{64}", result["input_fingerprint"] + ): + raise BaselineFormatError( + f"Missing input fingerprint for baseline case '{case_id}'" + ) + if "input_value" not in result: + raise BaselineFormatError( + f"Missing input value for baseline case '{case_id}'" + ) + if _fingerprint(result["input_value"]) != result["input_fingerprint"]: + raise BaselineFormatError( + f"Input fingerprint mismatch for baseline case '{case_id}'" + ) + shape = result.get("output_shape") + if shape is not None and not isinstance(shape, dict): + raise BaselineFormatError( + f"Invalid output shape for baseline case '{case_id}'" + ) + if result["state"] == "passed": + if not isinstance(shape, dict): + raise BaselineFormatError( + f"Passing baseline case '{case_id}' lacks output shape" + ) + fingerprint = result.get("output_fingerprint") + if not isinstance(fingerprint, str) or not re.fullmatch( + r"[0-9a-f]{64}", fingerprint + ): + raise BaselineFormatError( + f"Passing baseline case '{case_id}' lacks output fingerprint" + ) diff --git a/src/cmdforge/gui/pages/tool_builder_page.py b/src/cmdforge/gui/pages/tool_builder_page.py index 40a3e74..b8dba07 100644 --- a/src/cmdforge/gui/pages/tool_builder_page.py +++ b/src/cmdforge/gui/pages/tool_builder_page.py @@ -1133,6 +1133,8 @@ class ToolBuilderPage(QWidget): input_schema=self._tool.input_schema if self._tool else None, output_schema=self._tool.output_schema if self._tool else None, ) + if self._tool and self._tool.path: + tool.path = self._tool.path from ...preflight import analyze_tool preflight = analyze_tool(tool, include_contract_tests=True) diff --git a/src/cmdforge/preflight.py b/src/cmdforge/preflight.py index 05ec2bb..ef6bfe7 100644 --- a/src/cmdforge/preflight.py +++ b/src/cmdforge/preflight.py @@ -23,6 +23,7 @@ class PreflightReport: generated_tests: List[Dict[str, Any]] = field(default_factory=list) compatibility: List[Dict[str, Any]] = field(default_factory=list) contract_proposal: Optional[Dict[str, Any]] = None + regression: Optional[Dict[str, Any]] = None @property def ok(self) -> bool: @@ -37,6 +38,7 @@ class PreflightReport: generated_tests=self.generated_tests + other.generated_tests, compatibility=self.compatibility + other.compatibility, contract_proposal=self.contract_proposal or other.contract_proposal, + regression=self.regression or other.regression, ) def to_dict(self) -> Dict[str, Any]: @@ -48,6 +50,7 @@ class PreflightReport: "generated_tests": list(self.generated_tests), "compatibility": list(self.compatibility), "contract_proposal": copy.deepcopy(self.contract_proposal), + "regression": copy.deepcopy(self.regression), } @@ -73,6 +76,7 @@ def analyze_tool( _add_contract_guidance(tool, report, include_contract_tests) _check_secrets(tool, report) if check_local_dependencies: + _check_toolstep_compatibility(tool, report) _check_dependencies(tool, report) if registry_client: _check_similar_tools(tool, report, registry_client) @@ -99,16 +103,32 @@ def _add_contract_guidance( if tool.input_schema is None or tool.output_schema is None: return - from .contract_testing import run_contract_tests + from .contract_testing import ( + BaselineFormatError, + compare_conformance, + load_conformance_baseline, + run_contract_tests, + ) - conformance = run_contract_tests(tool) + baseline = None + if tool.path: + try: + baseline = load_conformance_baseline(tool.path.parent) + except BaselineFormatError as exc: + report.warnings.append(f"Regression baseline is invalid: {exc}") + baseline_inputs = ( + [result["input_value"] for result in baseline["results"]] + if baseline is not None else None + ) + conformance = run_contract_tests(tool, test_inputs=baseline_inputs) report.generated_tests = [result.to_dict() for result in conformance.results] - if conformance.outcome == "failed": + states = {result.state for result in conformance.results} + if "failed" in states: details = "; ".join( result.detail for result in conformance.results if result.state == "failed" ) report.errors.append(f"Contract conformance failed: {details}") - elif conformance.outcome == "unsupported": + if "unsupported" in states: details = "; ".join( result.detail for result in conformance.results @@ -116,6 +136,20 @@ def _add_contract_guidance( ) report.warnings.append(f"Contract conformance unsupported: {details}") + if baseline is not None: + try: + delta = compare_conformance(baseline, conformance, tool) + except BaselineFormatError as exc: + report.warnings.append(f"Regression baseline is invalid: {exc}") + else: + report.regression = delta.to_dict() + if delta.has_regressions: + report.warnings.append(f"Regression comparison: {delta.summary}") + elif delta.contract_changed: + report.suggestions.append( + "Tool contracts changed since the saved conformance baseline" + ) + def _check_config_integrity(tool: Tool, report: PreflightReport): if not tool.name: @@ -159,6 +193,142 @@ def _check_secrets(tool: Tool, report: PreflightReport): break +def _check_toolstep_compatibility(tool: Tool, report: PreflightReport): + from .schema_compat import compare_json_schemas + from .tool import ToolStep, load_tool + + for index, step in enumerate(tool.steps): + if not isinstance(step, ToolStep): + continue + nested = load_tool(step.tool) + if not nested: + report.compatibility.append({ + "tool": step.tool, + "step": index + 1, + "state": "unresolved", + "detail": f"Tool '{step.tool}' not found locally", + "issues": ["Called tool is not installed locally"], + }) + report.suggestions.append( + f"Could not check ToolStep {index + 1} -> '{step.tool}': " + "tool is not installed locally" + ) + continue + if nested.input_schema is None: + finding = { + "tool": step.tool, + "step": index + 1, + "state": "unknown", + "detail": f"Tool '{step.tool}' has no input_schema", + "issues": ["Downstream input contract is missing"], + } + report.compatibility.append(finding) + report.suggestions.append(finding["detail"]) + continue + + supplied_schema, source_schema = _toolstep_supplied_schema( + tool, step, index, nested + ) + result = compare_json_schemas(supplied_schema, nested.input_schema) + finding = { + "tool": step.tool, + "step": index + 1, + "state": result.state, + "detail": result.detail, + "issues": result.issues, + "upstream_schema": source_schema, + "supplied_schema": supplied_schema, + "downstream_schema": copy.deepcopy(nested.input_schema), + } + report.compatibility.append(finding) + if result.state == "incompatible": + report.warnings.append( + f"ToolStep {index + 1} -> '{step.tool}' is incompatible: " + f"{result.detail}" + ) + elif result.state == "unknown": + report.suggestions.append( + f"Could not prove ToolStep {index + 1} -> '{step.tool}' " + f"compatibility: {result.detail}" + ) + + +def _toolstep_supplied_schema(tool: Tool, step, index: int, nested: Tool): + """Describe values the current runner actually supplies to a nested tool.""" + from .schema_compat import schema_for_value + + input_schema = ( + schema_for_value(step.input_template) + if "{" not in step.input_template + else {"type": "string"} + ) + properties: Dict[str, Any] = {"input": input_schema} + required = ["input"] + source_schema = _template_source_schema(tool, step.input_template, index) + + for name, value in step.args.items(): + # execute_tool_step substitutes every explicit argument through str(). + if isinstance(value, str) and "{" not in value: + properties[name] = schema_for_value(value) + else: + properties[name] = {"type": "string"} + required.append(name) + + for argument in nested.arguments: + if argument.variable in properties or argument.default is None: + continue + properties[argument.variable] = schema_for_value(argument.default) + required.append(argument.variable) + + supplied = { + "type": "object", + "properties": properties, + "required": required, + "additionalProperties": False, + } + if _schema_type_name(nested.input_schema) != "object": + # Scalar contracts describe stdin directly when no argument object is used. + supplied = properties["input"] + return supplied, source_schema + + +def _template_source_schema(tool: Tool, template: str, before_index: int): + match = re.fullmatch(r"\{([A-Za-z_][A-Za-z0-9_]*)\}", template.strip()) + if not match: + return {"type": "string"} + variable = match.group(1) + if tool.input_schema is not None: + if _schema_type_name(tool.input_schema) == "object": + schema = tool.input_schema.get("properties", {}).get(variable) + if schema is not None: + return copy.deepcopy(schema) + elif variable == "input": + return copy.deepcopy(tool.input_schema) + + from .tool import PromptStep, ToolStep, load_tool + for previous in reversed(tool.steps[:before_index]): + if getattr(previous, "output_var", None) != variable: + continue + if isinstance(previous, PromptStep): + return copy.deepcopy(previous.output_schema) + if isinstance(previous, ToolStep): + nested = load_tool(previous.tool) + return copy.deepcopy(nested.output_schema) if nested else None + return None + return None + + +def _schema_type_name(schema: Optional[dict]) -> Optional[str]: + if not schema: + return None + value = schema.get("type") + if isinstance(value, str): + return value + if "properties" in schema or "required" in schema: + return "object" + return None + + def _check_dependencies(tool: Tool, report: PreflightReport): from .tool import tool_exists for dep in tool.dependencies: diff --git a/src/cmdforge/schema_compat.py b/src/cmdforge/schema_compat.py new file mode 100644 index 0000000..1614582 --- /dev/null +++ b/src/cmdforge/schema_compat.py @@ -0,0 +1,348 @@ +"""Conservative JSON Schema compatibility checks for tool composition.""" + +from __future__ import annotations + +from dataclasses import asdict, dataclass, field +from typing import Any, Dict, List, Literal, Optional, Set + + +CompatibilityState = Literal["compatible", "incompatible", "unknown"] + + +@dataclass +class SchemaCompatibility: + state: CompatibilityState + detail: str + issues: List[str] = field(default_factory=list) + + def to_dict(self) -> Dict[str, Any]: + return asdict(self) + + +def compare_json_schemas( + producer: Optional[dict], consumer: Optional[dict], path: str = "$" +) -> SchemaCompatibility: + """Check whether every value allowed by producer is accepted by consumer. + + This intentionally implements a useful, conservative subset of JSON Schema + subsumption. Unsupported combinators and constraints return ``unknown`` + rather than claiming compatibility. + """ + if producer is None or consumer is None: + return _unknown(path, "schema evidence is missing") + if consumer == {}: + return _compatible("consumer accepts any JSON value") + if producer == {}: + return _unknown(path, "producer schema is unconstrained") + + concrete = _concrete_values(producer) + if concrete is not None: + failures = [value for value in concrete if not _value_matches(value, consumer)] + if failures: + return _incompatible( + path, f"producer can emit value not accepted by consumer: {failures[0]!r}" + ) + return _compatible("all producer const/enum values satisfy consumer") + if _concrete_values(consumer) is not None: + return _incompatible( + path, "producer is not restricted to the consumer's const/enum values" + ) + + if any(key in producer or key in consumer for key in ("oneOf", "anyOf", "allOf", "not", "$ref")): + return _unknown(path, "schema combinators or references require deeper analysis") + + advanced_constraints = ( + "contains", "minContains", "maxContains", "patternProperties", + "dependentSchemas", "propertyNames", "unevaluatedProperties", + "unevaluatedItems", "if", "then", "else", + ) + differing = [ + keyword for keyword in advanced_constraints + if keyword in consumer and producer.get(keyword) != consumer.get(keyword) + ] + if differing: + return _unknown( + path, "cannot prove constraint(s): " + ", ".join(differing) + ) + + producer_types = _types(producer) + consumer_types = _types(consumer) + if producer_types and consumer_types: + rejected = { + item for item in producer_types + if not any(_type_accepted(item, target) for target in consumer_types) + } + if rejected: + return _incompatible( + path, + "producer type(s) " + ", ".join(sorted(rejected)) + + " are not accepted by consumer type(s) " + + ", ".join(sorted(consumer_types)), + ) + elif not producer_types or not consumer_types: + return _unknown(path, "type constraints are incomplete") + + states: List[SchemaCompatibility] = [] + effective_types = producer_types or set() + if "object" in effective_types: + states.append(_compare_objects(producer, consumer, path)) + if "array" in effective_types: + states.append(_compare_arrays(producer, consumer, path)) + if "string" in effective_types: + states.append(_compare_strings(producer, consumer, path)) + if effective_types & {"integer", "number"}: + states.append(_compare_numbers(producer, consumer, path)) + + incompatible = [state for state in states if state.state == "incompatible"] + if incompatible: + issues = [issue for state in incompatible for issue in state.issues] + return SchemaCompatibility("incompatible", incompatible[0].detail, issues) + unknown = [state for state in states if state.state == "unknown"] + if unknown: + issues = [issue for state in unknown for issue in state.issues] + return SchemaCompatibility("unknown", unknown[0].detail, issues) + return _compatible("producer schema is accepted by consumer schema") + + +def schema_for_value(value: Any) -> dict: + """Return an exact schema for a literal value.""" + return {"const": value} + + +def _compare_objects(producer: dict, consumer: dict, path: str) -> SchemaCompatibility: + producer_properties = producer.get("properties", {}) + consumer_properties = consumer.get("properties", {}) + producer_required = set(producer.get("required", [])) + consumer_required = set(consumer.get("required", [])) + issues: List[str] = [] + unknown = False + incompatible = False + + if ( + consumer.get("additionalProperties") is False + and producer.get("additionalProperties") is not False + ): + issues.append(f"{path}: producer permits unspecified object fields") + incompatible = True + + producer_min = max(producer.get("minProperties", 0), len(producer_required)) + consumer_min = consumer.get("minProperties", 0) + if producer_min < consumer_min: + issues.append(f"{path}: producer permits too few object properties") + incompatible = True + producer_max = producer.get("maxProperties") + if producer.get("additionalProperties") is False: + closed_max = len(producer_properties) + producer_max = closed_max if producer_max is None else min(producer_max, closed_max) + consumer_max = consumer.get("maxProperties") + if consumer_max is not None and ( + producer_max is None or producer_max > consumer_max + ): + issues.append(f"{path}: producer permits too many object properties") + incompatible = True + + for name in sorted(consumer_required - producer_required): + issues.append(f"{path}.{name}: required consumer field is not guaranteed") + incompatible = True + + producer_additional = producer.get("additionalProperties", True) + for name, consumer_schema in consumer_properties.items(): + if name in producer_properties or producer_additional is False: + continue + if producer_additional is True: + issues.append( + f"{path}.{name}: producer may emit an unconstrained value" + ) + incompatible = True + continue + if isinstance(producer_additional, dict): + result = compare_json_schemas( + producer_additional, consumer_schema, f"{path}.{name}" + ) + issues.extend(result.issues) + incompatible = incompatible or result.state == "incompatible" + unknown = unknown or result.state == "unknown" + + if consumer.get("additionalProperties") is False: + for name in sorted(producer_required - set(consumer_properties)): + issues.append(f"{path}.{name}: consumer rejects this supplied field") + incompatible = True + + for name, producer_schema in producer_properties.items(): + consumer_schema = consumer_properties.get(name) + if consumer_schema is None: + additional = consumer.get("additionalProperties", True) + if additional is False and name in producer_required: + issues.append(f"{path}.{name}: consumer does not allow this field") + incompatible = True + elif isinstance(additional, dict): + result = compare_json_schemas( + producer_schema, additional, f"{path}.{name}" + ) + issues.extend(result.issues) + unknown = unknown or result.state == "unknown" + continue + result = compare_json_schemas( + producer_schema, consumer_schema, f"{path}.{name}" + ) + if result.state == "incompatible": + issues.extend(result.issues) + incompatible = True + elif result.state == "unknown": + unknown = True + issues.extend(result.issues) + + if incompatible: + return SchemaCompatibility("incompatible", issues[0], issues) + if unknown: + return SchemaCompatibility("unknown", issues[0] if issues else "object constraints are incomplete", issues) + return _compatible("object fields and required properties are compatible") + + +def _compare_arrays(producer: dict, consumer: dict, path: str) -> SchemaCompatibility: + if consumer.get("uniqueItems") is True and producer.get("uniqueItems") is not True: + return _unknown(path, "producer does not guarantee unique array items") + producer_min = producer.get("minItems", 0) + consumer_min = consumer.get("minItems", 0) + if producer_min < consumer_min: + return _incompatible(path, "producer permits fewer array items than consumer") + producer_max = producer.get("maxItems") + consumer_max = consumer.get("maxItems") + if consumer_max is not None and ( + producer_max is None or producer_max > consumer_max + ): + return _incompatible(path, "producer permits more array items than consumer") + if "items" in consumer: + if "items" not in producer: + return _unknown(path, "producer array item schema is missing") + return compare_json_schemas( + producer["items"], consumer["items"], f"{path}[]" + ) + return _compatible("array bounds are compatible") + + +def _compare_strings(producer: dict, consumer: dict, path: str) -> SchemaCompatibility: + if producer.get("minLength", 0) < consumer.get("minLength", 0): + return _incompatible(path, "producer permits strings shorter than consumer") + consumer_max = consumer.get("maxLength") + producer_max = producer.get("maxLength") + if consumer_max is not None and ( + producer_max is None or producer_max > consumer_max + ): + return _incompatible(path, "producer permits strings longer than consumer") + for keyword in ("pattern", "format"): + if keyword in consumer and producer.get(keyword) != consumer.get(keyword): + return _unknown(path, f"cannot prove differing {keyword} constraints") + return _compatible("string bounds are compatible") + + +def _compare_numbers(producer: dict, consumer: dict, path: str) -> SchemaCompatibility: + producer_min = _lower_bound(producer) + consumer_min = _lower_bound(consumer) + if consumer_min is not None and ( + producer_min is None or producer_min < consumer_min + ): + return _incompatible(path, "producer permits numbers below consumer minimum") + if "exclusiveMinimum" in consumer and producer_min == consumer_min: + if producer.get("exclusiveMinimum") != consumer.get("exclusiveMinimum"): + return _incompatible( + path, "producer includes the consumer's exclusive minimum" + ) + producer_max = _upper_bound(producer) + consumer_max = _upper_bound(consumer) + if consumer_max is not None and ( + producer_max is None or producer_max > consumer_max + ): + return _incompatible(path, "producer permits numbers above consumer maximum") + if "exclusiveMaximum" in consumer and producer_max == consumer_max: + if producer.get("exclusiveMaximum") != consumer.get("exclusiveMaximum"): + return _incompatible( + path, "producer includes the consumer's exclusive maximum" + ) + if "multipleOf" in consumer and producer.get("multipleOf") != consumer.get("multipleOf"): + return _unknown(path, "cannot prove differing multipleOf constraints") + return _compatible("numeric bounds are compatible") + + +def _types(schema: dict) -> Set[str]: + value = schema.get("type") + if isinstance(value, str): + return {value} + if isinstance(value, list): + return {item for item in value if isinstance(item, str)} + if "properties" in schema or "required" in schema: + return {"object"} + if "items" in schema: + return {"array"} + concrete = _concrete_values(schema) + if concrete: + return {_json_type(value) for value in concrete} + return set() + + +def _concrete_values(schema: dict) -> Optional[List[Any]]: + if "const" in schema: + return [schema["const"]] + if isinstance(schema.get("enum"), list) and schema["enum"]: + return list(schema["enum"]) + return None + + +def _value_matches(value: Any, schema: dict) -> bool: + try: + from jsonschema import validate + from jsonschema.exceptions import ValidationError + validate(instance=value, schema=schema) + return True + except ValidationError: + return False + + +def _json_type(value: Any) -> str: + if value is None: + return "null" + if isinstance(value, bool): + return "boolean" + if isinstance(value, int): + return "integer" + if isinstance(value, float): + return "number" + if isinstance(value, str): + return "string" + if isinstance(value, list): + return "array" + return "object" + + +def _type_accepted(producer: str, consumer: str) -> bool: + return producer == consumer or (producer == "integer" and consumer == "number") + + +def _lower_bound(schema: dict) -> Optional[float]: + if "exclusiveMinimum" in schema: + return float(schema["exclusiveMinimum"]) + if "minimum" in schema: + return float(schema["minimum"]) + return None + + +def _upper_bound(schema: dict) -> Optional[float]: + if "exclusiveMaximum" in schema: + return float(schema["exclusiveMaximum"]) + if "maximum" in schema: + return float(schema["maximum"]) + return None + + +def _compatible(detail: str) -> SchemaCompatibility: + return SchemaCompatibility("compatible", detail) + + +def _unknown(path: str, detail: str) -> SchemaCompatibility: + issue = f"{path}: {detail}" + return SchemaCompatibility("unknown", issue, [issue]) + + +def _incompatible(path: str, detail: str) -> SchemaCompatibility: + issue = f"{path}: {detail}" + return SchemaCompatibility("incompatible", issue, [issue]) diff --git a/tests/test_cli.py b/tests/test_cli.py index 65858c6..bdecfcc 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -589,6 +589,25 @@ def test_inspect_shows_contract_proposal_and_passing_conformance( assert "PASSED" in output +def test_inspect_can_save_passing_baseline(tmp_path, monkeypatch, capsys): + from cmdforge.cli import cmd_inspect + from cmdforge.contract_testing import CONFORMANCE_FILE + + tool = Tool( + name="baseline", version="1.0.0", input_schema={}, + output_schema={"type": "string"}, output="stable", + path=tmp_path / "config.yaml", + ) + monkeypatch.setattr("cmdforge.tool.load_tool", lambda name: tool) + + args = SimpleNamespace( + name="baseline", registry=False, save_baseline=True + ) + assert cmd_inspect(args) == 0 + assert (tmp_path / CONFORMANCE_FILE).exists() + assert "Saved conformance baseline" in capsys.readouterr().out + + def test_switch_to_existing_tool_closes_creation_page_first(): pytest.importorskip("PySide6") from cmdforge.gui.pages.tool_builder_page import _switch_to_existing_tool diff --git a/tests/test_contract_testing.py b/tests/test_contract_testing.py index 5b56fd5..2c7dc65 100644 --- a/tests/test_contract_testing.py +++ b/tests/test_contract_testing.py @@ -1,5 +1,7 @@ """Tests for deterministic contract conformance (M8.V1).""" +from pathlib import Path + import pytest from jsonschema import validate @@ -11,6 +13,7 @@ from cmdforge.contract_testing import ( _fallback_value, run_contract_tests, ) +from cmdforge.tool import Tool class TestInputGeneration: @@ -208,3 +211,158 @@ class TestRunContractTests: ) ) assert report.outcome == "failed" + + +class TestRegressionComparison: + @staticmethod + def contracted_tool(version="1.0.0", output_schema=None): + from cmdforge.tool import Tool + + return Tool( + name="test", + version=version, + input_schema={}, + output_schema=output_schema or {"type": "string"}, + output="stable", + ) + + def passing_report(self, tool=None): + return run_contract_tests(tool or self.contracted_tool()) + + def test_save_and_load_baseline(self, tmp_path): + from cmdforge.contract_testing import ( + save_conformance_baseline, load_conformance_baseline, + ) + import stat + + tool = self.contracted_tool() + report = self.passing_report(tool) + path = save_conformance_baseline(tool, report, tmp_path) + assert path.exists() + assert stat.S_IMODE(path.stat().st_mode) == 0o600 + + loaded = load_conformance_baseline(tmp_path) + assert loaded["format_version"] == 1 + assert loaded["tool"]["version"] == "1.0.0" + assert loaded["contract_hash"] + assert len(loaded["results"]) == 1 + assert loaded["results"][0]["case_id"].startswith("case-") + assert loaded["results"][0]["output_shape"] == {"type": "string"} + + def test_missing_baseline(self, tmp_path): + from cmdforge.contract_testing import load_conformance_baseline + assert load_conformance_baseline(tmp_path) is None + + def test_corrupt_baseline_is_rejected(self, tmp_path): + from cmdforge.contract_testing import ( + BaselineFormatError, CONFORMANCE_FILE, load_conformance_baseline, + ) + + (tmp_path / CONFORMANCE_FILE).write_text("[]") + with pytest.raises(BaselineFormatError, match="root"): + load_conformance_baseline(tmp_path) + + def test_tampered_baseline_evidence_is_rejected(self, tmp_path): + from cmdforge.contract_testing import ( + BaselineFormatError, CONFORMANCE_FILE, load_conformance_baseline, + save_conformance_baseline, + ) + import json + + tool = self.contracted_tool() + save_conformance_baseline(tool, self.passing_report(tool), tmp_path) + path = tmp_path / CONFORMANCE_FILE + data = json.loads(path.read_text()) + data["results"][0]["input_value"] = {"tampered": True} + path.write_text(json.dumps(data)) + + with pytest.raises(BaselineFormatError, match="fingerprint mismatch"): + load_conformance_baseline(tmp_path) + + def test_failed_report_cannot_replace_baseline(self, tmp_path): + from cmdforge.contract_testing import save_conformance_baseline + + report = run_contract_tests(Tool( + name="failed", input_schema={}, output_schema={"type": "integer"}, + output="text", + )) + with pytest.raises(ValueError, match="passing"): + save_conformance_baseline(self.contracted_tool(), report, tmp_path) + + def baseline_and_report(self, tmp_path): + from cmdforge.contract_testing import ( + load_conformance_baseline, save_conformance_baseline, + ) + + tool = self.contracted_tool() + report = self.passing_report(tool) + save_conformance_baseline(tool, report, tmp_path) + return load_conformance_baseline(tmp_path), report + + def test_detects_regression(self): + from dataclasses import replace + from cmdforge.contract_testing import ( + ConformanceReport, compare_conformance, + ) + import tempfile + + with tempfile.TemporaryDirectory() as directory: + baseline, report = self.baseline_and_report(Path(directory)) + current = ConformanceReport([ + replace(report.results[0], state="failed", detail="now broken") + ]) + delta = compare_conformance( + baseline, current, self.contracted_tool(version="2.0.0") + ) + assert len(delta.regressed) == 1 + assert delta.has_regressions + assert delta.current_version == "2.0.0" + + def test_lost_coverage_is_regression(self, tmp_path): + from dataclasses import replace + from cmdforge.contract_testing import ( + ConformanceReport, compare_conformance, + ) + baseline, report = self.baseline_and_report(tmp_path) + current = ConformanceReport([ + replace(report.results[0], state="unsupported", detail="lost") + ]) + delta = compare_conformance(baseline, current, self.contracted_tool()) + assert len(delta.coverage_losses) == 1 + assert delta.has_regressions + + def test_removed_case_is_regression(self, tmp_path): + from cmdforge.contract_testing import ( + ConformanceReport, compare_conformance, + ) + baseline, _ = self.baseline_and_report(tmp_path) + delta = compare_conformance( + baseline, ConformanceReport(), self.contracted_tool() + ) + assert len(delta.removed_cases) == 1 + assert delta.has_regressions + + def test_output_shape_or_value_change_is_flagged(self, tmp_path): + from dataclasses import replace + from cmdforge.contract_testing import ( + ConformanceReport, compare_conformance, + ) + baseline, report = self.baseline_and_report(tmp_path) + current = ConformanceReport([ + replace( + report.results[0], + output_shape={"type": "object", "properties": {}}, + output_fingerprint="changed", + ) + ]) + delta = compare_conformance(baseline, current, self.contracted_tool()) + assert len(delta.output_changes) == 1 + assert delta.has_regressions + + def test_contract_change_is_recorded(self, tmp_path): + from cmdforge.contract_testing import compare_conformance + + baseline, report = self.baseline_and_report(tmp_path) + changed = self.contracted_tool(output_schema={"type": ["string", "null"]}) + delta = compare_conformance(baseline, report, changed) + assert delta.contract_changed diff --git a/tests/test_preflight.py b/tests/test_preflight.py index 04dd5ef..509e878 100644 --- a/tests/test_preflight.py +++ b/tests/test_preflight.py @@ -342,3 +342,87 @@ class TestConformanceIntegration: assert report.ok assert report.generated_tests[0]["state"] == "unsupported" assert any("unsupported" in warning.lower() for warning in report.warnings) + + def test_saved_baseline_is_compared_during_local_preflight(self, tmp_path): + from cmdforge.contract_testing import ( + run_contract_tests, save_conformance_baseline, + ) + from cmdforge.tool import Tool + + baseline_tool = Tool( + name="regression", version="1.0.0", input_schema={}, + output_schema={"type": "string"}, output="before", + path=tmp_path / "config.yaml", + ) + save_conformance_baseline( + baseline_tool, run_contract_tests(baseline_tool), tmp_path + ) + current = Tool( + name="regression", version="2.0.0", input_schema={}, + output_schema={"type": "string"}, output="after", + path=tmp_path / "config.yaml", + ) + report = analyze_tool(current, include_contract_tests=True) + assert report.regression["has_regressions"] is True + assert report.regression["current_version"] == "2.0.0" + assert any("Regression comparison" in warning for warning in report.warnings) + + def test_baseline_inputs_are_rerun_after_input_schema_change(self, tmp_path): + from cmdforge.contract_testing import ( + run_contract_tests, save_conformance_baseline, + ) + from cmdforge.tool import Tool + + baseline_tool = Tool( + name="input-change", version="1.0.0", input_schema={}, + output_schema={"type": "string"}, output="stable", + path=tmp_path / "config.yaml", + ) + save_conformance_baseline( + baseline_tool, run_contract_tests(baseline_tool), tmp_path + ) + current = Tool( + name="input-change", version="2.0.0", + input_schema={ + "type": "object", + "properties": {"name": {"type": "string"}}, + "required": ["name"], + }, + output_schema={"type": "string"}, output="stable", + path=tmp_path / "config.yaml", + ) + report = analyze_tool(current, include_contract_tests=True) + assert report.regression["has_regressions"] is True + assert len(report.regression["regressed"]) == 1 + assert any( + result["input_value"] == {} and result["state"] == "failed" + for result in report.generated_tests + ) + + def test_invalid_baseline_is_an_advisory(self, tmp_path): + from cmdforge.contract_testing import CONFORMANCE_FILE + from cmdforge.tool import Tool + + (tmp_path / CONFORMANCE_FILE).write_text("not json") + tool = Tool( + name="invalid-baseline", input_schema={}, output_schema={}, + output="ok", path=tmp_path / "config.yaml", + ) + report = analyze_tool(tool, include_contract_tests=True) + assert report.ok + assert any("baseline is invalid" in warning.lower() for warning in report.warnings) + + def test_partial_unsupported_conformance_is_not_hidden_by_a_pass(self, monkeypatch): + from cmdforge.contract_testing import ConformanceReport, ConformanceResult + from cmdforge.tool import Tool + + monkeypatch.setattr( + "cmdforge.contract_testing.run_contract_tests", + lambda tool, test_inputs=None: ConformanceReport([ + ConformanceResult("case-a", "passed", "valid"), + ConformanceResult("input-generation", "unsupported", "unsupported keyword"), + ]), + ) + tool = Tool(name="mixed", input_schema={}, output_schema={}) + report = analyze_tool(tool, include_contract_tests=True) + assert any("unsupported keyword" in warning for warning in report.warnings) diff --git a/tests/test_schema_compat.py b/tests/test_schema_compat.py new file mode 100644 index 0000000..1e1a647 --- /dev/null +++ b/tests/test_schema_compat.py @@ -0,0 +1,186 @@ +"""Tests for conservative ToolStep JSON Schema compatibility.""" + +from unittest.mock import patch + +from cmdforge.preflight import ( + PreflightReport, _check_toolstep_compatibility, analyze_tool, +) +from cmdforge.schema_compat import compare_json_schemas +from cmdforge.tool import McpStep, PromptStep, Tool, ToolStep + + +class TestSchemaCompatibility: + def test_type_mismatch_is_incompatible(self): + result = compare_json_schemas( + {"type": "string"}, {"type": "integer"} + ) + assert result.state == "incompatible" + + def test_integer_is_accepted_by_number(self): + result = compare_json_schemas( + {"type": "integer"}, {"type": "number"} + ) + assert result.state == "compatible" + + def test_missing_required_property_is_incompatible(self): + result = compare_json_schemas( + {"type": "object", "properties": {}}, + { + "type": "object", + "properties": {"name": {"type": "string"}}, + "required": ["name"], + }, + ) + assert result.state == "incompatible" + assert "required" in result.detail + + def test_matching_object_property_is_compatible(self): + schema = { + "type": "object", + "properties": {"name": {"type": "string"}}, + "required": ["name"], + } + assert compare_json_schemas(schema, schema).state == "compatible" + + def test_literal_enum_value_is_checked(self): + assert compare_json_schemas( + {"const": "fast"}, {"type": "string", "enum": ["fast", "slow"]} + ).state == "compatible" + assert compare_json_schemas( + {"const": "invalid"}, {"type": "string", "enum": ["fast"]} + ).state == "incompatible" + + def test_unconstrained_producer_does_not_satisfy_consumer_enum(self): + result = compare_json_schemas( + {"type": "string"}, {"type": "string", "enum": ["fast"]} + ) + assert result.state == "incompatible" + + def test_closed_consumer_rejects_open_producer_object(self): + result = compare_json_schemas( + {"type": "object", "properties": {}}, + {"type": "object", "properties": {}, "additionalProperties": False}, + ) + assert result.state == "incompatible" + + def test_open_producer_cannot_guarantee_typed_optional_consumer_field(self): + result = compare_json_schemas( + {"type": "object", "properties": {}}, + { + "type": "object", + "properties": {"limit": {"type": "integer"}}, + }, + ) + assert result.state == "incompatible" + assert "limit" in result.detail + + def test_unproven_pattern_is_unknown(self): + result = compare_json_schemas( + {"type": "string"}, {"type": "string", "pattern": "^[a-z]+$"} + ) + assert result.state == "unknown" + + +class TestToolStepCompatibility: + @staticmethod + def child(input_schema): + return Tool(name="child", input_schema=input_schema) + + def analyze(self, parent, child): + report = PreflightReport() + with patch("cmdforge.tool.load_tool", return_value=child): + _check_toolstep_compatibility(parent, report) + return report + + def test_compatible_stdin_contract(self): + parent = Tool( + name="parent", + steps=[ToolStep(tool="child", output_var="result")], + ) + child = self.child({ + "type": "object", + "properties": {"input": {"type": "string"}}, + "required": ["input"], + }) + report = self.analyze(parent, child) + assert report.compatibility[0]["state"] == "compatible" + + def test_upstream_serialization_mismatch_is_reported(self): + producer = PromptStep( + prompt="produce", provider="mock", output_var="value", + output_schema={"type": "string"}, + ) + parent = Tool( + name="parent", + steps=[ + producer, + ToolStep( + tool="child", output_var="result", input_template="{value}" + ), + ], + ) + child = self.child({ + "type": "object", + "properties": {"input": {"type": "integer"}}, + "required": ["input"], + }) + report = self.analyze(parent, child) + finding = report.compatibility[0] + assert finding["state"] == "incompatible" + assert finding["upstream_schema"] == {"type": "string"} + assert any("incompatible" in warning for warning in report.warnings) + + def test_missing_required_argument_is_incompatible(self): + parent = Tool( + name="parent", + steps=[ToolStep(tool="child", output_var="result")], + ) + child = self.child({ + "type": "object", + "properties": { + "input": {"type": "string"}, + "limit": {"type": "integer"}, + }, + "required": ["input", "limit"], + }) + report = self.analyze(parent, child) + assert report.compatibility[0]["state"] == "incompatible" + assert "limit" in report.compatibility[0]["detail"] + + def test_explicit_argument_reflects_string_transport(self): + parent = Tool( + name="parent", + steps=[ToolStep( + tool="child", output_var="result", args={"limit": "{input}"} + )], + ) + child = self.child({ + "type": "object", + "properties": { + "input": {"type": "string"}, + "limit": {"type": "integer"}, + }, + "required": ["input", "limit"], + }) + assert self.analyze(parent, child).compatibility[0]["state"] == "incompatible" + + def test_mcp_steps_are_ignored(self): + parent = Tool( + name="parent", + steps=[McpStep(server="remote", tool="same-name", output_var="result")], + ) + report = PreflightReport() + with patch("cmdforge.tool.load_tool") as loader: + _check_toolstep_compatibility(parent, report) + loader.assert_not_called() + assert report.compatibility == [] + + def test_server_mode_does_not_resolve_local_tools(self): + parent = Tool( + name="published", + steps=[ToolStep(tool="child", output_var="result")], + ) + with patch("cmdforge.tool.load_tool") as loader: + report = analyze_tool(parent, check_local_dependencies=False) + loader.assert_not_called() + assert report.compatibility == []