M8.PF: Add shared PreflightReport analyzer with cmdforge inspect command
This commit is contained in:
parent
0323a39afb
commit
f6bd640865
|
|
@ -90,6 +90,11 @@ def main():
|
|||
p_docs.add_argument("-e", "--edit", action="store_true", help="Edit/create README in $EDITOR")
|
||||
p_docs.set_defaults(func=cmd_docs)
|
||||
|
||||
# 'inspect' command
|
||||
p_inspect = subparsers.add_parser("inspect", help="Run preflight analysis on a tool")
|
||||
p_inspect.add_argument("name", help="Tool name")
|
||||
p_inspect.set_defaults(func=cmd_inspect)
|
||||
|
||||
# 'check' command
|
||||
p_check = subparsers.add_parser("check", help="Check dependencies for a tool (meta-tools)")
|
||||
p_check.add_argument("name", help="Tool name")
|
||||
|
|
@ -515,5 +520,47 @@ def main():
|
|||
return args.func(args)
|
||||
|
||||
|
||||
def cmd_inspect(args):
|
||||
"""Run preflight analysis on a tool."""
|
||||
from ..preflight import analyze_tool
|
||||
from ..tool import load_tool
|
||||
|
||||
tool = load_tool(args.name)
|
||||
if not tool:
|
||||
print(f"Error: Tool '{args.name}' not found.", file=open("/dev/stderr", "w"))
|
||||
return 1
|
||||
|
||||
report = analyze_tool(tool)
|
||||
|
||||
if report.errors:
|
||||
print(f"Errors ({len(report.errors)}):")
|
||||
for e in report.errors:
|
||||
print(f" ERROR: {e}")
|
||||
print()
|
||||
|
||||
if report.warnings:
|
||||
print(f"Warnings ({len(report.warnings)}):")
|
||||
for w in report.warnings:
|
||||
print(f" WARN: {w}")
|
||||
print()
|
||||
|
||||
if report.suggestions:
|
||||
print(f"Suggestions ({len(report.suggestions)}):")
|
||||
for s in report.suggestions:
|
||||
print(f" HINT: {s}")
|
||||
print()
|
||||
|
||||
if report.similar_tools:
|
||||
print(f"Similar registry tools ({len(report.similar_tools)}):")
|
||||
for t in report.similar_tools:
|
||||
print(f" - {t['name']}: {t.get('description', '')[:80]}")
|
||||
print()
|
||||
|
||||
if not report.errors and not report.warnings and not report.suggestions:
|
||||
print("No issues found.")
|
||||
|
||||
return 0 if not report.errors else 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
|
|
|
|||
|
|
@ -0,0 +1,116 @@
|
|||
"""Shared preflight analysis engine.
|
||||
|
||||
Produces a PreflightReport used by the GUI, CLI, and registry workflow.
|
||||
All checks are deterministic and evidence-based. Recommendations never
|
||||
automate away human judgment.
|
||||
"""
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
from .tool import Tool
|
||||
|
||||
|
||||
@dataclass
|
||||
class PreflightReport:
|
||||
"""Structured analysis of a tool before creation, save, or publish."""
|
||||
|
||||
errors: List[str] = field(default_factory=list)
|
||||
warnings: List[str] = field(default_factory=list)
|
||||
suggestions: List[str] = field(default_factory=list)
|
||||
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)
|
||||
|
||||
@property
|
||||
def ok(self) -> bool:
|
||||
return not self.errors
|
||||
|
||||
def merge(self, other: "PreflightReport") -> "PreflightReport":
|
||||
return PreflightReport(
|
||||
errors=self.errors + other.errors,
|
||||
warnings=self.warnings + other.warnings,
|
||||
suggestions=self.suggestions + other.suggestions,
|
||||
similar_tools=self.similar_tools + other.similar_tools,
|
||||
generated_tests=self.generated_tests + other.generated_tests,
|
||||
compatibility=self.compatibility + other.compatibility,
|
||||
)
|
||||
|
||||
|
||||
def analyze_tool(tool: Tool, registry_client=None) -> PreflightReport:
|
||||
"""Run all preflight checks on a tool.
|
||||
|
||||
Args:
|
||||
tool: The Tool to analyze.
|
||||
registry_client: Optional registry client for registry-side checks.
|
||||
|
||||
Returns:
|
||||
A PreflightReport with structured findings.
|
||||
"""
|
||||
report = PreflightReport()
|
||||
_check_config_integrity(tool, report)
|
||||
_check_contracts(tool, report)
|
||||
_check_secrets(tool, report)
|
||||
_check_dependencies(tool, report)
|
||||
if registry_client:
|
||||
_check_similar_tools(tool, report, registry_client)
|
||||
return report
|
||||
|
||||
|
||||
def _check_config_integrity(tool: Tool, report: PreflightReport):
|
||||
if not tool.name:
|
||||
report.errors.append("Tool name is required")
|
||||
if tool.version and not _is_semver(tool.version):
|
||||
report.errors.append(f"Version '{tool.version}' is not valid semver")
|
||||
if not tool.steps:
|
||||
report.warnings.append("Tool has no steps — output is a no-op passthrough")
|
||||
|
||||
|
||||
def _check_contracts(tool: Tool, report: PreflightReport):
|
||||
if tool.output_schema:
|
||||
schema = tool.output_schema
|
||||
if not isinstance(schema, dict):
|
||||
report.errors.append("output_schema must be a JSON Schema object")
|
||||
elif schema.get("type") != "object":
|
||||
report.warnings.append("output_schema root should have type: object")
|
||||
else:
|
||||
report.suggestions.append("Add an output_schema to enable automated verification")
|
||||
|
||||
|
||||
def _check_secrets(tool: Tool, report: PreflightReport):
|
||||
secret_patterns = ["api_key", "api_secret", "password", "token", "secret"]
|
||||
for step in tool.steps:
|
||||
if hasattr(step, "prompt"):
|
||||
prompt_lower = step.prompt.lower()
|
||||
for pat in secret_patterns:
|
||||
if pat in prompt_lower:
|
||||
report.warnings.append(
|
||||
f"Prompt step contains potential secret pattern '{pat}'"
|
||||
)
|
||||
break
|
||||
|
||||
|
||||
def _check_dependencies(tool: Tool, report: PreflightReport):
|
||||
from .tool import tool_exists
|
||||
for dep in tool.dependencies:
|
||||
ref = dep.get("name", dep) if isinstance(dep, dict) else dep
|
||||
if not tool_exists(ref):
|
||||
report.warnings.append(f"Dependency '{ref}' is not installed locally")
|
||||
|
||||
|
||||
def _check_similar_tools(tool: Tool, report: PreflightReport, client):
|
||||
try:
|
||||
results = client.search_tools(tool.name, limit=5)
|
||||
for item in results:
|
||||
report.similar_tools.append({
|
||||
"name": item.get("owner") + "/" + item.get("name"),
|
||||
"description": item.get("description", ""),
|
||||
"downloads": item.get("downloads", 0),
|
||||
})
|
||||
except Exception:
|
||||
report.warnings.append("Could not search registry for similar tools")
|
||||
|
||||
|
||||
def _is_semver(version: str) -> bool:
|
||||
import re
|
||||
return bool(re.match(r"^\d+\.\d+\.\d+", version))
|
||||
|
|
@ -0,0 +1,76 @@
|
|||
"""Tests for preflight analysis engine."""
|
||||
|
||||
from unittest.mock import patch
|
||||
|
||||
from cmdforge.preflight import PreflightReport, analyze_tool, _check_config_integrity
|
||||
|
||||
|
||||
class TestPreflightReport:
|
||||
def test_empty_report_is_ok(self):
|
||||
report = PreflightReport()
|
||||
assert report.ok
|
||||
|
||||
def test_errors_make_not_ok(self):
|
||||
report = PreflightReport(errors=["bad"])
|
||||
assert not report.ok
|
||||
|
||||
def test_merge(self):
|
||||
a = PreflightReport(errors=["e1"], warnings=["w1"])
|
||||
b = PreflightReport(errors=["e2"], suggestions=["s1"])
|
||||
c = a.merge(b)
|
||||
assert c.errors == ["e1", "e2"]
|
||||
assert c.warnings == ["w1"]
|
||||
assert c.suggestions == ["s1"]
|
||||
|
||||
|
||||
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"):
|
||||
tool = Tool(name="empty-tool")
|
||||
save_tool(tool)
|
||||
|
||||
report = analyze_tool(tool)
|
||||
assert report.ok
|
||||
assert any("no steps" in w.lower() for w in report.warnings)
|
||||
|
||||
def test_missing_name(self, tmp_path):
|
||||
from cmdforge.tool import Tool
|
||||
|
||||
report = PreflightReport()
|
||||
tool = Tool(name="")
|
||||
_check_config_integrity(tool, report)
|
||||
assert not report.ok
|
||||
|
||||
def test_invalid_version(self, tmp_path):
|
||||
from cmdforge.tool import Tool
|
||||
|
||||
report = PreflightReport()
|
||||
tool = Tool(name="test", version="not-semver")
|
||||
_check_config_integrity(tool, report)
|
||||
assert not report.ok
|
||||
|
||||
def test_ok_tool(self, tmp_path):
|
||||
from cmdforge.tool import Tool
|
||||
|
||||
report = PreflightReport()
|
||||
tool = Tool(name="ok-tool", version="1.0.0")
|
||||
_check_config_integrity(tool, report)
|
||||
assert report.ok
|
||||
|
||||
def test_secret_pattern_in_prompt(self, tmp_path):
|
||||
from cmdforge.tool import Tool, PromptStep
|
||||
|
||||
tool = Tool(
|
||||
name="leaky",
|
||||
steps=[PromptStep(
|
||||
prompt="Use api_key: abc123 to auth",
|
||||
provider="mock",
|
||||
output_var="out",
|
||||
)],
|
||||
)
|
||||
report = PreflightReport()
|
||||
from cmdforge.preflight import _check_secrets
|
||||
_check_secrets(tool, report)
|
||||
assert any("secret" in w.lower() for w in report.warnings)
|
||||
Loading…
Reference in New Issue