From d81286e32256da0110ee9e481de67c35407e4d06 Mon Sep 17 00:00:00 2001 From: rob Date: Mon, 20 Jul 2026 12:22:26 -0300 Subject: [PATCH] M8.0/M8.V1: Add contract inference engine and deterministic conformance testing --- AGENTS.md | 1 + CLAUDE.md | 4 + README.md | 2 +- src/cmdforge/cli/__init__.py | 31 +- src/cmdforge/cli/registry_commands.py | 26 +- src/cmdforge/contract_testing.py | 477 ++++++++++++++++++++ src/cmdforge/gui/pages/tool_builder_page.py | 31 +- src/cmdforge/preflight.py | 174 +++++++ tests/test_cli.py | 29 ++ tests/test_collection_api.py | 1 + tests/test_contract_testing.py | 210 +++++++++ tests/test_preflight.py | 210 ++++++++- 12 files changed, 1191 insertions(+), 5 deletions(-) create mode 100644 src/cmdforge/contract_testing.py create mode 100644 tests/test_contract_testing.py diff --git a/AGENTS.md b/AGENTS.md index 9ee351e..cecd2d8 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -22,6 +22,7 @@ cf # Interactive tool picker - `cli/` - Routes all subcommands (list, create, run, test, providers, registry, collections, deps, install, etc.) - `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 - `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 6bdb3fb..4b8feef 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -83,6 +83,10 @@ Tools are YAML configs with: Run `cmdforge inspect [--registry]` for the shared local preflight report. `cmdforge registry publish --dry-run` runs local checks first and, when authenticated, the registry's publish-time checks without publishing. +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. ### Step Types diff --git a/README.md b/README.md index e1b289e..e79c082 100644 --- a/README.md +++ b/README.md @@ -135,7 +135,7 @@ cmdforge delete mytool # Delete tool cmdforge run mytool # Run a tool cmdforge test mytool # Test with mock provider cmdforge check mytool # Check dependencies (meta-tools) -cmdforge inspect mytool # Validate contracts and run local preflight checks +cmdforge inspect mytool # Preflight, contract proposals, safe conformance tests cmdforge inspect mytool --registry # Also find similar registry tools 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 d6e2d45..ffa44b2 100644 --- a/src/cmdforge/cli/__init__.py +++ b/src/cmdforge/cli/__init__.py @@ -538,7 +538,11 @@ def cmd_inspect(args): if getattr(args, "registry", False): from ..registry_client import get_client registry_client = get_client() - report = analyze_tool(tool, registry_client=registry_client) + report = analyze_tool( + tool, + registry_client=registry_client, + include_contract_tests=True, + ) if report.errors: print(f"Errors ({len(report.errors)}):") @@ -564,6 +568,31 @@ def cmd_inspect(args): print(f" - {t['name']}: {t.get('description', '')[:80]}") print() + proposal = report.contract_proposal or {} + proposed_schemas = { + key: proposal.get(key) + for key in ("input_schema", "output_schema") + if proposal.get(key) is not None + } + if proposed_schemas: + import yaml + + print(f"Contract proposal ({proposal.get('confidence', 'low')} confidence):") + print(yaml.safe_dump(proposed_schemas, sort_keys=False).rstrip()) + for evidence in proposal.get("evidence", []): + print(f" EVIDENCE: {evidence}") + print(" Contracts are suggestions only and were not saved.") + print() + + if report.generated_tests: + print(f"Contract conformance ({len(report.generated_tests)} case(s)):") + for result in report.generated_tests: + print( + f" {result['state'].upper()}: {result['step']} — " + f"{result.get('detail', '')}" + ) + 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 9265c52..223d946 100644 --- a/src/cmdforge/cli/registry_commands.py +++ b/src/cmdforge/cli/registry_commands.py @@ -365,6 +365,28 @@ def _print_preflight_sections(report: dict, prefix: str = "") -> None: for value in values: print(f" {marker}: {value}") + proposal = report.get("contract_proposal") or {} + proposed = { + key: proposal.get(key) + for key in ("input_schema", "output_schema") + if proposal.get(key) is not None + } + if proposed: + print( + f"{prefix}Contract proposal " + f"({proposal.get('confidence', 'low')} confidence; not saved):" + ) + print(yaml.safe_dump(proposed, sort_keys=False).rstrip()) + + generated = report.get("generated_tests") or [] + if generated: + print(f"{prefix}Contract conformance ({len(generated)} case(s)):") + for result in generated: + print( + f" {str(result.get('state', 'not_run')).upper()}: " + f"{result.get('step', 'unknown')} — {result.get('detail', '')}" + ) + def _print_registry_suggestions(suggestions: dict) -> None: """Render registry-specific category, similarity, and scrutiny evidence.""" @@ -472,7 +494,9 @@ def _cmd_registry_publish(args): from ..preflight import analyze_tool from ..tool import Tool try: - local_report = analyze_tool(Tool.from_dict(data)) + local_report = analyze_tool( + Tool.from_dict(data), include_contract_tests=True + ) except (KeyError, TypeError, ValueError) as exc: print(f"Local preflight error: {exc}", file=sys.stderr) return 1 diff --git a/src/cmdforge/contract_testing.py b/src/cmdforge/contract_testing.py new file mode 100644 index 0000000..0490630 --- /dev/null +++ b/src/cmdforge/contract_testing.py @@ -0,0 +1,477 @@ +"""Deterministic, side-effect-safe contract conformance verification. + +The verifier generates inputs that are validated against ``input_schema``, +replaces prompt steps with schema-derived deterministic values, executes the +remaining safe pipeline, and validates its final result. It never calls an AI +provider. Steps that could execute user code or contact another tool/server are +reported as unsupported instead of being run implicitly. +""" + +from __future__ import annotations + +import copy +import json +import math +import re +from dataclasses import asdict, dataclass, field +from datetime import date, datetime, timezone +from typing import Any, Dict, List, Literal +from uuid import UUID + +from .tool import CodeStep, McpStep, PromptStep, Tool, ToolStep + + +ConformanceState = Literal["passed", "failed", "unsupported", "not_run"] +_VALID_STATES = ("passed", "failed", "unsupported", "not_run") + + +class UnsupportedContract(ValueError): + """Raised when a valid deterministic value cannot be produced safely.""" + + +@dataclass +class ConformanceResult: + step: str + state: ConformanceState + detail: str = "" + + def __post_init__(self) -> None: + if self.state not in _VALID_STATES: + raise ValueError(f"Unknown conformance state: {self.state}") + + def is_pass(self) -> bool: + return self.state == "passed" + + def to_dict(self) -> Dict[str, str]: + return asdict(self) + + +@dataclass +class ConformanceReport: + results: List[ConformanceResult] = field(default_factory=list) + + @property + def outcome(self) -> ConformanceState: + states = {result.state for result in self.results} + if "failed" in states: + return "failed" + if "passed" in states: + return "passed" + if "unsupported" in states: + return "unsupported" + return "not_run" + + @property + def passed(self) -> bool: + return self.outcome == "passed" + + @property + def failed(self) -> bool: + return self.outcome == "failed" + + @property + def summary(self) -> Dict[str, int]: + counts = {state: 0 for state in _VALID_STATES} + for result in self.results: + counts[result.state] += 1 + return counts + + def to_dict(self) -> Dict[str, Any]: + return { + "outcome": self.outcome, + "summary": self.summary, + "results": [result.to_dict() for result in self.results], + } + + +def run_contract_tests(tool: Tool) -> ConformanceReport: + """Run safe structural conformance tests for a tool. + + This is deliberately not a semantic quality test. Original CodeStep, + ToolStep, and McpStep instances are not executed because they may have side + effects, make network calls, or invoke real providers. + """ + if tool.input_schema is None or tool.output_schema is None: + return _single_result( + "preflight", + "not_run", + "Both input_schema and output_schema are required", + ) + + unsafe = next( + ( + type(step).__name__ + for step in tool.steps + if isinstance(step, (CodeStep, ToolStep, McpStep)) + ), + None, + ) + if unsafe: + return _single_result( + "preflight", + "unsupported", + f"Safe conformance mode does not execute {unsafe}", + ) + + try: + test_cases = _generate_inputs(tool.input_schema) + except UnsupportedContract as exc: + return _single_result("input-generation", "unsupported", str(exc)) + + if not test_cases: + return _single_result( + "input-generation", "not_run", "No deterministic inputs generated" + ) + + report = ConformanceReport() + for index, test_input in enumerate(test_cases, start=1): + label = f"case-{index}" + try: + result = _run_with_stub(tool, test_input) + except UnsupportedContract as exc: + report.results.append( + ConformanceResult(label, "unsupported", str(exc)) + ) + except Exception as exc: + report.results.append( + ConformanceResult(label, "failed", f"Execution error: {exc}") + ) + else: + try: + _validate_instance(result, tool.output_schema, "output") + except UnsupportedContract as exc: + report.results.append( + ConformanceResult(label, "unsupported", str(exc)) + ) + except ValueError as exc: + report.results.append(ConformanceResult(label, "failed", str(exc))) + else: + report.results.append( + ConformanceResult( + label, "passed", "Output conforms to output_schema" + ) + ) + return report + + +def _single_result( + step: str, state: ConformanceState, detail: str +) -> ConformanceReport: + return ConformanceReport([ConformanceResult(step, state, detail)]) + + +def _generate_inputs(schema: dict) -> List[Any]: + """Generate deterministic values and validate every returned candidate.""" + candidates: List[Any] = [] + schema_type = _schema_type(schema) + + if schema_type == "object": + enum_properties = { + name: spec["enum"] + for name, spec in schema.get("properties", {}).items() + if isinstance(spec, dict) and spec.get("enum") + } + variant_count = min(max((len(v) for v in enum_properties.values()), default=1), 3) + for index in range(variant_count): + overrides = { + name: values[index % len(values)] + for name, values in enum_properties.items() + } + candidates.append(_object_value(schema, overrides)) + else: + candidates.append(_value_for_schema(schema, "input")) + + unique: List[Any] = [] + seen = set() + for candidate in candidates: + try: + _validate_instance(candidate, schema, "generated input") + except ValueError as exc: + raise UnsupportedContract( + f"Could not generate a valid input: {exc}" + ) from exc + marker = json.dumps(candidate, sort_keys=True, default=str) + if marker not in seen: + seen.add(marker) + unique.append(candidate) + return unique + + +def _schema_type(schema: dict) -> str | None: + schema_type = schema.get("type") + if isinstance(schema_type, list): + return next((value for value in schema_type if value != "null"), "null") + if schema_type is None and ("properties" in schema or "required" in schema): + return "object" + return schema_type + + +def _value_for_schema(schema: dict, name: str) -> Any: + for keyword in ("const", "default"): + if keyword in schema: + return copy.deepcopy(schema[keyword]) + if schema.get("examples"): + return copy.deepcopy(schema["examples"][0]) + if schema.get("enum"): + return copy.deepcopy(schema["enum"][0]) + + alternatives = schema.get("oneOf") or schema.get("anyOf") + if alternatives: + for alternative in alternatives: + try: + candidate = _value_for_schema(alternative, name) + _validate_instance(candidate, schema, name) + return candidate + except (UnsupportedContract, ValueError): + continue + raise UnsupportedContract(f"Cannot synthesize {name} from alternatives") + + if schema.get("allOf"): + merged = _merge_all_of(schema) + candidate = _value_for_schema(merged, name) + _validate_instance(candidate, schema, name) + return candidate + + schema_type = _schema_type(schema) + if schema_type == "object": + return _object_value(schema, {}) + if schema_type == "array": + item_schema = schema.get("items", {}) + count = max(0, schema.get("minItems", 0)) + if count == 0 and schema.get("contains"): + return [_value_for_schema(schema["contains"], f"{name}-item")] + return [ + _vary_value(_value_for_schema(item_schema, f"{name}-item"), index) + for index in range(count) + ] + if schema_type == "string": + return _string_value(schema, name) + if schema_type in ("integer", "number"): + return _number_value(schema, integer=schema_type == "integer") + if schema_type == "boolean": + return True + if schema_type == "null": + return None + if schema_type is None and not schema: + return {} + if schema_type is None: + candidate = {} + try: + _validate_instance(candidate, schema, name) + return candidate + except ValueError as exc: + raise UnsupportedContract(str(exc)) from exc + raise UnsupportedContract(f"Unsupported JSON Schema type for {name}: {schema_type}") + + +def _object_value(schema: dict, overrides: Dict[str, Any]) -> Dict[str, Any]: + properties = schema.get("properties", {}) + required = list(schema.get("required", [])) + result: Dict[str, Any] = {} + + names = list(required) + for name, spec in properties.items(): + if name in overrides or any( + keyword in spec for keyword in ("const", "default", "examples", "enum") + ): + if name not in names: + names.append(name) + + minimum = schema.get("minProperties", 0) + for name in properties: + if len(names) >= minimum: + break + if name not in names: + names.append(name) + + for name in names: + if name in overrides: + result[name] = copy.deepcopy(overrides[name]) + else: + result[name] = _value_for_schema(properties.get(name, {}), name) + + for trigger, dependents in schema.get("dependentRequired", {}).items(): + if trigger in result: + for dependent in dependents: + if dependent not in result: + result[dependent] = _value_for_schema( + properties.get(dependent, {}), dependent + ) + return result + + +def _string_value(schema: dict, name: str) -> str: + formats = { + "date": date(2020, 1, 2).isoformat(), + "date-time": datetime(2020, 1, 2, 3, 4, 5, tzinfo=timezone.utc).isoformat(), + "email": "test@example.com", + "hostname": "example.com", + "ipv4": "192.0.2.1", + "uri": "https://example.com/", + "uuid": str(UUID(int=0)), + } + minimum = max(0, schema.get("minLength", 0)) + maximum = schema.get("maxLength") + seeds = [formats.get(schema.get("format")), f"test-{name}", "a", "0", ""] + pattern = schema.get("pattern") + for seed in seeds: + if seed is None: + continue + candidate = seed + if len(candidate) < minimum: + candidate += "a" * (minimum - len(candidate)) + if maximum is not None: + candidate = candidate[:maximum] + if pattern and re.search(pattern, candidate) is None: + continue + return candidate + raise UnsupportedContract(f"Cannot synthesize string '{name}' for pattern/bounds") + + +def _number_value(schema: dict, *, integer: bool) -> int | float: + lower = schema.get("minimum") + if "exclusiveMinimum" in schema: + exclusive_lower = schema["exclusiveMinimum"] + (1 if integer else 0.5) + lower = exclusive_lower if lower is None else max(lower, exclusive_lower) + upper = schema.get("maximum") + if "exclusiveMaximum" in schema: + exclusive_upper = schema["exclusiveMaximum"] - (1 if integer else 0.5) + upper = exclusive_upper if upper is None else min(upper, exclusive_upper) + value = 0 + if lower is not None: + value = max(value, lower) + if upper is not None: + value = min(value, upper) + multiple = schema.get("multipleOf") + if multiple: + if lower is not None: + value = math.ceil(value / multiple) * multiple + else: + value = math.floor(value / multiple) * multiple + if (lower is not None and value < lower) or (upper is not None and value > upper): + raise UnsupportedContract("Numeric bounds do not admit a deterministic value") + return int(value) if integer else float(value) + + +def _vary_value(value: Any, index: int) -> Any: + if index == 0: + return value + if isinstance(value, bool) or value is None: + return value + if isinstance(value, int): + return value + index + if isinstance(value, float): + return value + float(index) + if isinstance(value, str): + return f"{value}{index}" + return value + + +def _merge_all_of(schema: dict) -> dict: + merged = {key: copy.deepcopy(value) for key, value in schema.items() if key != "allOf"} + properties = copy.deepcopy(merged.get("properties", {})) + required = list(merged.get("required", [])) + for part in schema["allOf"]: + properties.update(copy.deepcopy(part.get("properties", {}))) + for name in part.get("required", []): + if name not in required: + required.append(name) + for key, value in part.items(): + if key not in ("properties", "required"): + merged.setdefault(key, copy.deepcopy(value)) + if properties: + merged["properties"] = properties + if required: + merged["required"] = required + return merged + + +def _validate_instance(instance: Any, schema: dict, label: str) -> None: + try: + from jsonschema import FormatChecker + from jsonschema.exceptions import ValidationError + from jsonschema.validators import validator_for + except ImportError as exc: + raise UnsupportedContract("jsonschema is required for conformance testing") from exc + + validator = validator_for(schema)(schema, format_checker=FormatChecker()) + try: + validator.validate(instance) + except ValidationError as exc: + path = ".".join(str(part) for part in exc.absolute_path) or "root" + raise ValueError(f"{label} does not conform at {path}: {exc.message}") from exc + + +def _run_with_stub(tool: Tool, inputs: Any) -> Any: + """Execute only a prompt/no-step pipeline with generated prompt outputs.""" + from .runner import DEFAULT_OUTPUT_SCHEMA, run_tool + + tool_copy = copy.deepcopy(tool) + transformed = [] + final_variable = _single_output_variable(tool.output) + for step in tool_copy.steps: + if not isinstance(step, PromptStep): + raise UnsupportedContract( + f"Safe conformance mode does not execute {type(step).__name__}" + ) + schema = step.output_schema + if schema is None and step.output_var == final_variable: + schema = tool.output_schema + if schema is None: + schema = DEFAULT_OUTPUT_SCHEMA + value = _value_for_schema(schema, step.output_var) + try: + _validate_instance(value, schema, f"stub output '{step.output_var}'") + except ValueError as exc: + raise UnsupportedContract( + f"Could not synthesize '{step.output_var}': {exc}" + ) from exc + transformed.append( + CodeStep( + code=f"globals()[{step.output_var!r}] = {value!r}", + output_var=step.output_var, + name=f"Contract stub for {step.name or step.output_var}", + ) + ) + tool_copy.steps = transformed + + if isinstance(inputs, dict): + custom_args = {key: value for key, value in inputs.items() if key != "input"} + input_text = inputs.get("input", "") + else: + custom_args = {} + input_text = inputs + + output, exit_code = run_tool( + tool_copy, + input_text="" if input_text is None else str(input_text), + custom_args=custom_args, + provider_override=None, + dry_run=False, + verbose=False, + ) + if exit_code != 0: + raise RuntimeError(f"Tool execution failed with exit code {exit_code}") + try: + return json.loads(output) + except (json.JSONDecodeError, TypeError): + return output + + +def _single_output_variable(output: str) -> str | None: + match = re.fullmatch(r"\{([A-Za-z_][A-Za-z0-9_]*)\}", output.strip()) + return match.group(1) if match else None + + +def _fallback_value(prop: dict, name: str, required: list | None = None) -> Any: + """Backward-compatible helper used by callers and focused tests.""" + return _value_for_schema(prop, name) + + +def _validate_output(output: Any, schema: dict) -> bool: + """Backward-compatible boolean output validator.""" + try: + _validate_instance(output, schema, "output") + return True + except ValueError: + return False diff --git a/src/cmdforge/gui/pages/tool_builder_page.py b/src/cmdforge/gui/pages/tool_builder_page.py index e95c544..40a3e74 100644 --- a/src/cmdforge/gui/pages/tool_builder_page.py +++ b/src/cmdforge/gui/pages/tool_builder_page.py @@ -1135,7 +1135,7 @@ class ToolBuilderPage(QWidget): ) from ...preflight import analyze_tool - preflight = analyze_tool(tool) + preflight = analyze_tool(tool, include_contract_tests=True) if preflight.errors: QMessageBox.warning( self, @@ -1143,6 +1143,35 @@ class ToolBuilderPage(QWidget): "\n".join(preflight.errors), ) return + + proposal = preflight.contract_proposal or {} + proposed_schemas = { + key: proposal.get(key) + for key in ("input_schema", "output_schema") + if proposal.get(key) is not None + } + if proposed_schemas: + msg = QMessageBox(self) + msg.setIcon(QMessageBox.Information) + msg.setWindowTitle("Contract Proposal") + msg.setText( + "CmdForge inferred a possible tool contract. " + "It will not be applied automatically." + ) + msg.setInformativeText( + "Review the proposal below. Contract editing and approval " + "will be added by the guided-creation milestone." + ) + msg.setDetailedText( + yaml.safe_dump(proposed_schemas, sort_keys=False).rstrip() + ) + save_button = msg.addButton( + "Save Without Applying", QMessageBox.AcceptRole + ) + msg.addButton(QMessageBox.Cancel) + msg.exec() + if msg.clickedButton() is not save_button: + return advisory_count = len(preflight.warnings) + len(preflight.suggestions) # Preserve source if editing diff --git a/src/cmdforge/preflight.py b/src/cmdforge/preflight.py index 730a77d..05ec2bb 100644 --- a/src/cmdforge/preflight.py +++ b/src/cmdforge/preflight.py @@ -4,6 +4,8 @@ Produces a PreflightReport used by the GUI, CLI, and registry workflow. All checks are deterministic and evidence-based. """ +import copy +import re from dataclasses import dataclass, field from typing import Any, Dict, List, Optional @@ -20,6 +22,7 @@ class PreflightReport: similar_tools: List[Dict[str, Any]] = field(default_factory=list) 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 @property def ok(self) -> bool: @@ -33,6 +36,7 @@ class PreflightReport: similar_tools=self.similar_tools + other.similar_tools, generated_tests=self.generated_tests + other.generated_tests, compatibility=self.compatibility + other.compatibility, + contract_proposal=self.contract_proposal or other.contract_proposal, ) def to_dict(self) -> Dict[str, Any]: @@ -43,6 +47,7 @@ class PreflightReport: "similar_tools": list(self.similar_tools), "generated_tests": list(self.generated_tests), "compatibility": list(self.compatibility), + "contract_proposal": copy.deepcopy(self.contract_proposal), } @@ -51,6 +56,7 @@ def analyze_tool( registry_client=None, *, check_local_dependencies: bool = True, + include_contract_tests: bool = False, ) -> PreflightReport: """Run all preflight checks on a tool. @@ -64,6 +70,7 @@ def analyze_tool( report = PreflightReport() _check_config_integrity(tool, report) _check_contracts(tool, report) + _add_contract_guidance(tool, report, include_contract_tests) _check_secrets(tool, report) if check_local_dependencies: _check_dependencies(tool, report) @@ -72,6 +79,44 @@ def analyze_tool( return report +def _add_contract_guidance( + tool: Tool, report: PreflightReport, include_contract_tests: bool +) -> None: + proposal = infer_contracts(tool) + report.contract_proposal = proposal + inferred = [ + label + for label in ("input", "output") + if proposal[f"{label}_source"] == "inferred" + ] + if inferred: + report.suggestions.append( + "Review inferred " + " and ".join(inferred) + " contract proposal" + ) + + if not include_contract_tests: + return + if tool.input_schema is None or tool.output_schema is None: + return + + from .contract_testing import run_contract_tests + + conformance = run_contract_tests(tool) + report.generated_tests = [result.to_dict() for result in conformance.results] + if conformance.outcome == "failed": + 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": + details = "; ".join( + result.detail + for result in conformance.results + if result.state == "unsupported" + ) + report.warnings.append(f"Contract conformance unsupported: {details}") + + def _check_config_integrity(tool: Tool, report: PreflightReport): if not tool.name: report.errors.append("Tool name is required") @@ -143,3 +188,132 @@ def _check_similar_tools(tool: Tool, report: PreflightReport, client): def _is_semver(version: str) -> bool: from .semver import Version return Version.parse(version) is not None + + +def infer_contracts(tool: Tool) -> Dict[str, Any]: + """Propose input/output schemas without mutating the tool. + + Returns a dict with: + - input_schema: proposed JSON Schema (or None) + - output_schema: proposed JSON Schema (or None) + - evidence: list of human-readable provenance strings + - confidence: "high" / "medium" / "low" + + Never overwrites an explicit schema. ``{}`` counts as explicit. + """ + evidence = [] + proposed_input = None + proposed_output = None + + if tool.input_schema is None: + proposed_input, input_evidence = _infer_input_schema(tool) + evidence.extend(input_evidence) + + if tool.output_schema is None: + proposed_output, output_evidence = _infer_output_schema(tool) + evidence.extend(output_evidence) + + effective_input = tool.input_schema if tool.input_schema is not None else proposed_input + effective_output = tool.output_schema if tool.output_schema is not None else proposed_output + confidence = _confidence(effective_input, effective_output) + + return { + "input_schema": proposed_input, + "output_schema": proposed_output, + "input_source": "explicit" if tool.input_schema is not None else ( + "inferred" if proposed_input is not None else "unavailable" + ), + "output_source": "explicit" if tool.output_schema is not None else ( + "inferred" if proposed_output is not None else "unavailable" + ), + "evidence": evidence, + "confidence": confidence, + } + + +def _infer_input_schema(tool: Tool) -> tuple: + props = {} + required = [] + evidence = [] + has_stdin = _contains_input_reference(tool.output) or any( + _contains_input_reference(vars(step)) for step in tool.steps + ) + + if has_stdin: + props["input"] = {"type": "string", "description": "Standard input text"} + evidence.append("stdin detected from {input} usage") + + for arg in tool.arguments: + prop = {"type": arg.type or "string"} + if arg.description: + prop["description"] = arg.description + if arg.default is not None: + prop["default"] = arg.default + if arg.enum: + prop["enum"] = arg.enum + props[arg.variable] = prop + if arg.required: + required.append(arg.variable) + evidence.append(f"argument {arg.flag} ({arg.type or 'string'})") + + if not props: + return None, [] + + schema = {"type": "object", "properties": props} + if required: + schema["required"] = required + return schema, evidence + + +def _infer_output_schema(tool: Tool) -> tuple: + evidence = [] + + output = tool.output.strip() + if not output or output == "{input}": + return None, evidence + + var_match = None + m = re.match(r"^\{([a-zA-Z_][a-zA-Z0-9_]*)\}$", output) + if m: + var_match = m.group(1) + + if not var_match: + evidence.append("output template is a composite expression — schema must be defined manually") + return None, evidence + + # Look for the step that produces this variable + for step in reversed(tool.steps): + if hasattr(step, "output_var") and step.output_var == var_match: + schema = getattr(step, "output_schema", None) + if schema is not None: + evidence.append( + f"output variable '{var_match}' has an explicit step schema — using as tool output schema" + ) + return copy.deepcopy(schema), evidence + elif hasattr(step, "provider"): + evidence.append( + f"output variable '{var_match}' comes from a provider step with no schema — define output_schema on the step" + ) + return None, evidence + break + + evidence.append(f"output variable '{var_match}' source not found") + return None, evidence + + +def _contains_input_reference(value: Any) -> bool: + if isinstance(value, str): + return "{input}" in value or "{input." in value + if isinstance(value, dict): + return any(_contains_input_reference(item) for item in value.values()) + if isinstance(value, (list, tuple)): + return any(_contains_input_reference(item) for item in value) + return False + + +def _confidence(input_schema, output_schema) -> str: + if input_schema is not None and output_schema is not None: + return "high" + if input_schema is not None or output_schema is not None: + return "medium" + return "low" diff --git a/tests/test_cli.py b/tests/test_cli.py index 6a6c5c2..65858c6 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -560,6 +560,35 @@ def test_inspect_with_registry_uses_similarity_results(monkeypatch, capsys): assert "official/summarize" in capsys.readouterr().out +def test_inspect_shows_contract_proposal_and_passing_conformance( + monkeypatch, capsys +): + from cmdforge.cli import cmd_inspect + + step_schema = { + "type": "object", + "properties": {"answer": {"type": "string"}}, + "required": ["answer"], + } + tool = Tool( + name="contracted", + arguments=[], + input_schema={}, + output_schema=step_schema, + steps=[PromptStep( + prompt="Answer", provider="paid-provider", output_var="answer", + output_schema=step_schema, + )], + output="{answer}", + ) + monkeypatch.setattr("cmdforge.tool.load_tool", lambda name: tool) + + assert cmd_inspect(SimpleNamespace(name="contracted", registry=False)) == 0 + output = capsys.readouterr().out + assert "Contract conformance" in output + assert "PASSED" in output + + 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_collection_api.py b/tests/test_collection_api.py index 2247136..809a762 100644 --- a/tests/test_collection_api.py +++ b/tests/test_collection_api.py @@ -180,6 +180,7 @@ class TestPublishPreflightEndpoint: assert response.status_code == 200 report = response.get_json()["data"]["preflight"] assert report["errors"] == [] + assert report["generated_tests"] == [] def test_invalid_contract_is_rejected(self, client, auth_headers): response = client.post( diff --git a/tests/test_contract_testing.py b/tests/test_contract_testing.py new file mode 100644 index 0000000..5b56fd5 --- /dev/null +++ b/tests/test_contract_testing.py @@ -0,0 +1,210 @@ +"""Tests for deterministic contract conformance (M8.V1).""" + +import pytest +from jsonschema import validate + +from cmdforge.contract_testing import ( + ConformanceResult, + ConformanceReport, + UnsupportedContract, + _generate_inputs, + _fallback_value, + run_contract_tests, +) + + +class TestInputGeneration: + def test_enum_generates_variants(self): + schema = { + "type": "object", + "properties": { + "mode": {"type": "string", "enum": ["fast", "accurate", "balanced"]}, + "limit": {"type": "integer", "default": 10}, + }, + "required": ["mode"], + } + cases = _generate_inputs(schema) + assert len(cases) >= 1 + assert cases[0]["mode"] in ("fast", "accurate", "balanced") + + def test_fallback_string(self): + cases = _generate_inputs({ + "type": "object", + "properties": {"name": {"type": "string"}}, + "required": ["name"], + }) + assert "test-name" in cases[0]["name"] + + def test_fallback_integer(self): + cases = _generate_inputs({ + "type": "object", + "properties": {"count": {"type": "integer"}}, + "required": ["count"], + }) + assert cases[0]["count"] == 0 + + def test_default_used(self): + cases = _generate_inputs({ + "type": "object", + "properties": {"name": {"type": "string", "default": "World"}}, + "required": ["name"], + }) + assert cases[0]["name"] == "World" + + def test_scalar_schema_generates_scalar(self): + cases = _generate_inputs({"type": "string"}) + assert cases == ["test-input"] + + def test_generated_value_respects_numeric_bounds(self): + schema = { + "type": "object", + "properties": { + "count": {"type": "integer", "minimum": 1, "maximum": 3} + }, + "required": ["count"], + } + case = _generate_inputs(schema)[0] + validate(case, schema) + assert 1 <= case["count"] <= 3 + + def test_array_and_nested_object_are_supported(self): + schema = { + "type": "array", + "minItems": 1, + "items": { + "type": "object", + "properties": {"name": {"type": "string"}}, + "required": ["name"], + }, + } + value = _generate_inputs(schema)[0] + validate(value, schema) + + def test_unsynthesizable_pattern_is_unsupported(self): + with pytest.raises(UnsupportedContract): + _generate_inputs({"type": "string", "pattern": "^z{20}$"}) + + def test_invalid_default_is_unsupported_not_returned(self): + with pytest.raises(UnsupportedContract, match="valid input"): + _generate_inputs({"type": "integer", "default": "wrong"}) + + +class TestFallbackValue: + def test_explicit_default(self): + assert _fallback_value({"default": "hi"}, "x", []) == "hi" + + def test_examples(self): + assert _fallback_value({"examples": ["a", "b"]}, "x", []) == "a" + + def test_enum(self): + assert _fallback_value({"enum": ["yes", "no"]}, "x", []) == "yes" + + def test_type_fallback(self): + assert _fallback_value({"type": "boolean"}, "flag", []) is True + + +class TestConformanceReport: + def test_empty_is_not_run_not_pass(self): + report = ConformanceReport() + assert report.outcome == "not_run" + assert not report.passed + + def test_invalid_state_is_rejected(self): + with pytest.raises(ValueError, match="Unknown conformance state"): + ConformanceResult("case", "unknown") + + def test_summary_counts(self): + report = ConformanceReport(results=[ + ConformanceResult("a", "passed"), + ConformanceResult("b", "passed"), + ConformanceResult("c", "failed", "error"), + ]) + assert report.summary == {"passed": 2, "failed": 1, "unsupported": 0, "not_run": 0} + assert not report.passed + + +class TestRunContractTests: + def test_missing_schemas_returns_not_run(self, tmp_path): + from unittest.mock import patch + from cmdforge.tool import Tool + + with patch("cmdforge.tool.TOOLS_DIR", tmp_path / ".cmdforge"): + tool = Tool(name="bare") + report = run_contract_tests(tool) + assert report.results[0].state == "not_run" + + def test_simple_tool_passes(self, tmp_path): + from unittest.mock import patch + from cmdforge.tool import Tool, PromptStep + + with patch("cmdforge.tool.TOOLS_DIR", tmp_path / ".cmdforge"): + tool = Tool( + name="greet", + input_schema={ + "type": "object", + "properties": { + "name": {"type": "string", "default": "World"}, + }, + }, + output_schema={ + "type": "object", + "properties": {"result": {"type": "string"}}, + "required": ["result"], + }, + steps=[ + PromptStep( + prompt="Greet {name}", + provider="real-provider-that-must-not-run", + output_var="response", + output_schema={ + "type": "object", + "properties": {"result": {"type": "string"}}, + "required": ["result"], + }, + ), + ], + output="{response}", + ) + report = run_contract_tests(tool) + assert report.outcome == "passed" + assert report.results[0].state == "passed" + + def test_empty_schemas_are_explicit_and_runnable(self): + from cmdforge.tool import Tool + + report = run_contract_tests( + Tool(name="anything", input_schema={}, output_schema={}, output="ok") + ) + assert report.outcome == "passed" + + @pytest.mark.parametrize("step_type", ["code", "tool", "mcp"]) + def test_side_effecting_steps_are_unsupported(self, step_type): + from cmdforge.tool import CodeStep, McpStep, Tool, ToolStep + + steps = { + "code": CodeStep(code="raise AssertionError('must not run')", output_var="x"), + "tool": ToolStep(tool="external", output_var="x"), + "mcp": McpStep(server="remote", tool="external", output_var="x"), + } + tool = Tool( + name="unsafe", + input_schema={}, + output_schema={}, + steps=[steps[step_type]], + output="{x}", + ) + report = run_contract_tests(tool) + assert report.outcome == "unsupported" + + def test_schema_mismatch_is_a_failure(self): + from cmdforge.tool import Tool + + report = run_contract_tests( + Tool( + name="mismatch", + input_schema={}, + output_schema={"type": "integer"}, + output="not-an-integer", + ) + ) + assert report.outcome == "failed" diff --git a/tests/test_preflight.py b/tests/test_preflight.py index a97ad08..04dd5ef 100644 --- a/tests/test_preflight.py +++ b/tests/test_preflight.py @@ -38,12 +38,22 @@ class TestPreflightReport: serialized["errors"].append("another") assert report.errors == ["bad"] + def test_to_dict_copies_contract_proposal(self): + report = PreflightReport( + contract_proposal={"input_schema": {"type": "string"}} + ) + serialized = report.to_dict() + serialized["contract_proposal"]["input_schema"]["type"] = "integer" + assert report.contract_proposal["input_schema"]["type"] == "string" + class TestAnalyzeTool: def test_no_steps_warns(self, tmp_path): from cmdforge.tool import Tool, save_tool - with patch("cmdforge.tool.TOOLS_DIR", tmp_path / ".cmdforge"): + with patch("cmdforge.tool.TOOLS_DIR", tmp_path / ".cmdforge"), patch( + "cmdforge.tool.BIN_DIR", tmp_path / "bin" + ): tool = Tool(name="empty-tool") save_tool(tool) @@ -134,3 +144,201 @@ class TestAnalyzeTool: _check_similar_tools( SimpleNamespace(name="summarize"), PreflightReport(), Client() ) + + +class TestContractInference: + def test_no_arguments_no_input(self): + from cmdforge.tool import Tool + from cmdforge.preflight import infer_contracts + + tool = Tool(name="empty", output="{response}") + result = infer_contracts(tool) + assert result["input_schema"] is None + assert result["output_schema"] is None + assert result["confidence"] == "low" + + def test_argument_becomes_property(self): + from cmdforge.tool import Tool, ToolArgument + from cmdforge.preflight import infer_contracts + + tool = Tool( + name="greet", + arguments=[ToolArgument(flag="--name", variable="name", description="Who to greet")], + output="Hello", + ) + result = infer_contracts(tool) + assert result["input_schema"] is not None + assert "name" in result["input_schema"]["properties"] + assert result["input_schema"]["properties"]["name"]["description"] == "Who to greet" + + def test_stdin_detected_when_input_used(self): + from cmdforge.tool import Tool, PromptStep + from cmdforge.preflight import infer_contracts + + tool = Tool( + name="echo", + steps=[PromptStep(prompt="Repeat: {input}", provider="mock", output_var="response")], + output="{response}", + ) + result = infer_contracts(tool) + assert result["input_schema"] is not None + assert "input" in result["input_schema"]["properties"] + assert any("stdin" in e.lower() for e in result["evidence"]) + + def test_enum_propagates(self): + from cmdforge.tool import Tool, ToolArgument + from cmdforge.preflight import infer_contracts + + arg = ToolArgument(flag="--mode", variable="mode") + arg.enum = ["fast", "accurate"] + tool = Tool(name="tool", arguments=[arg], output="done") + result = infer_contracts(tool) + assert result["input_schema"]["properties"]["mode"]["enum"] == ["fast", "accurate"] + + def test_explicit_empty_schema_not_overwritten(self): + from cmdforge.tool import Tool + from cmdforge.preflight import infer_contracts + + tool = Tool(name="tool", input_schema={}, output_schema={}) + result = infer_contracts(tool) + assert result["input_schema"] is None + assert result["output_schema"] is None + + def test_output_propagated_from_step_schema(self): + from cmdforge.tool import Tool, PromptStep + from cmdforge.preflight import infer_contracts + + step = PromptStep( + prompt="Analyze: {input}", + provider="mock", + output_var="analysis", + ) + step.output_schema = { + "type": "object", + "properties": {"summary": {"type": "string"}}, + } + tool = Tool(name="analyze", steps=[step], output="{analysis}") + result = infer_contracts(tool) + assert result["output_schema"] is not None + assert result["output_schema"]["type"] == "object" + assert "summary" in result["output_schema"]["properties"] + assert result["confidence"] == "high" + + @pytest.mark.parametrize("schema_type", ["string", "number", "array", "boolean"]) + def test_non_object_output_schema_is_propagated(self, schema_type): + from cmdforge.tool import PromptStep, Tool + from cmdforge.preflight import infer_contracts + + schema = {"type": schema_type} + step = PromptStep( + prompt="Generate", provider="mock", output_var="result", + output_schema=schema, + ) + proposal = infer_contracts( + Tool(name="typed", steps=[step], output="{result}") + ) + assert proposal["output_schema"] == schema + + def test_proposed_schema_does_not_alias_step_schema(self): + from cmdforge.tool import PromptStep, Tool + from cmdforge.preflight import infer_contracts + + step = PromptStep( + prompt="Generate", provider="mock", output_var="result", + output_schema={ + "type": "object", + "properties": {"value": {"type": "string"}}, + }, + ) + proposal = infer_contracts( + Tool(name="copy", steps=[step], output="{result}") + ) + proposal["output_schema"]["properties"]["value"]["type"] = "integer" + assert step.output_schema["properties"]["value"]["type"] == "string" + + def test_stdin_detected_in_nested_tool_input(self): + from cmdforge.tool import Tool, ToolStep + from cmdforge.preflight import infer_contracts + + proposal = infer_contracts(Tool( + name="nested", + steps=[ToolStep(tool="child", output_var="result")], + output="{result}", + )) + assert "input" in proposal["input_schema"]["properties"] + + def test_evidence_uses_real_argument_flag(self): + from cmdforge.tool import Tool, ToolArgument + from cmdforge.preflight import infer_contracts + + proposal = infer_contracts(Tool( + name="flags", + arguments=[ToolArgument(flag="-n", variable="name")], + output="constant", + )) + assert any("argument -n" in item for item in proposal["evidence"]) + + def test_explicit_contracts_produce_high_confidence(self): + from cmdforge.tool import Tool + from cmdforge.preflight import infer_contracts + + proposal = infer_contracts( + Tool(name="explicit", input_schema={}, output_schema={}) + ) + assert proposal["confidence"] == "high" + + def test_bare_variable_output_stays_bare(self): + from cmdforge.tool import Tool + from cmdforge.preflight import infer_contracts + + tool = Tool(name="passthrough", steps=[], output="{response}") + result = infer_contracts(tool) + assert result["output_schema"] is None + assert any("source not found" in e.lower() for e in result["evidence"]) + + +class TestConformanceIntegration: + def test_contract_tests_are_opt_in_for_server_side_preflight(self): + from cmdforge.tool import Tool + + tool = Tool(name="server-safe", input_schema={}, output_schema={}) + with patch("cmdforge.contract_testing.run_contract_tests") as runner: + analyze_tool(tool) + runner.assert_not_called() + + def test_analyze_tool_records_passing_contract_case(self): + from cmdforge.tool import PromptStep, Tool + + schema = { + "type": "object", + "properties": {"answer": {"type": "string"}}, + "required": ["answer"], + } + tool = Tool( + name="contracted", + input_schema={}, + output_schema=schema, + steps=[PromptStep( + prompt="Answer", provider="paid-provider", output_var="answer", + output_schema=schema, + )], + output="{answer}", + ) + report = analyze_tool(tool, include_contract_tests=True) + assert report.ok + assert report.generated_tests[0]["state"] == "passed" + + def test_unsupported_execution_is_advisory_not_failure(self): + from cmdforge.tool import CodeStep, Tool + + tool = Tool( + name="code-tool", + input_schema={}, + output_schema={}, + steps=[CodeStep(code="result = 1", output_var="result")], + output="{result}", + ) + report = analyze_tool(tool, include_contract_tests=True) + assert report.ok + assert report.generated_tests[0]["state"] == "unsupported" + assert any("unsupported" in warning.lower() for warning in report.warnings)