diff --git a/src/cmdforge/cli/__init__.py b/src/cmdforge/cli/__init__.py index 335c1ee..a92f8a6 100644 --- a/src/cmdforge/cli/__init__.py +++ b/src/cmdforge/cli/__init__.py @@ -653,6 +653,18 @@ def cmd_inspect(args): print(f" {cat.name:24s} {cat.display:>8s}") print() + if report.reuse_opportunities: + print(f"Reuse opportunities ({len(report.reuse_opportunities)}):") + for opp in report.reuse_opportunities: + print(f" {opp['type'].upper()}: {opp['detail']}") + print() + + if report.audit_evidence: + ae = report.audit_evidence + print(f"Audit evidence (engine v{ae.get('engine_version', '?')}, {ae.get('timestamp', '')[:10]}):") + print(f" Checks: {', '.join(ae.get('checks_run', []))}") + print() + if not report.errors and not report.warnings and not report.suggestions: print("No issues found.") diff --git a/src/cmdforge/cli/picker.py b/src/cmdforge/cli/picker.py index b5ef402..7651b92 100644 --- a/src/cmdforge/cli/picker.py +++ b/src/cmdforge/cli/picker.py @@ -78,11 +78,43 @@ def get_tools() -> List[dict]: "flag": a.flag, "default": a.default or "", "desc": a.description or "" - } for a in tool.arguments] + } for a in tool.arguments], + "deprecated": tool.deprecated, + "deprecated_message": tool.deprecated_message, + "replacement": tool.replacement, }) return sorted(tools, key=lambda t: t["name"]) +def search_registry(query: str) -> List[dict]: + """Search the registry for tools matching the query. + + Returns a list of dicts with name, desc, and registry metadata. + Returns empty list if no token configured or search fails. + """ + if not query or len(query) < 2: + return [] + try: + from ..registry_client import get_client + client = get_client() + if not client or not client.token: + return [] + results = client.search_tools(query, per_page=5) + items = results.data if hasattr(results, "data") else results + return [ + { + "name": f"{item.get('owner', '')}/{item.get('name', '')}", + "desc": (item.get("description") or "")[:50], + "args": [], + "registry": True, + "downloads": item.get("downloads", 0), + } + for item in items + ] + except Exception: + return [] + + class TTYInput: """Read from /dev/tty for keyboard input, even when stdin is piped.""" @@ -157,6 +189,14 @@ def run_picker(tty_input: TTYInput) -> Optional[PickerResult]: matches.sort(key=lambda x: -x[1]) filtered = [m[0] for m in matches] + # If no local matches, search registry + registry_results = [] + if query and len(query) >= 2 and not filtered: + registry_results = search_registry(query) + for rt in registry_results: + rt["_registry"] = True + filtered = registry_results + if selected >= len(filtered): selected = max(0, len(filtered) - 1) @@ -181,13 +221,23 @@ def run_picker(tty_input: TTYInput) -> Optional[PickerResult]: # Items for i, t in enumerate(visible): actual_idx = scroll + i + is_registry = t.get("_registry", False) or t.get("registry", False) + is_deprecated = t.get("deprecated", False) + if actual_idx == selected: prefix = f"{CYAN}{BOLD}▸ {t['name']}{RESET}" else: prefix = f" {t['name']}" + # Add registry marker + if is_registry: + prefix += f" {YELLOW}[registry]{RESET}" + # Add deprecation marker + if is_deprecated: + prefix += f" {YELLOW}[deprecated]{RESET}" + # Add arg indicator - if t['args']: + if t.get('args'): prefix += f" {GREEN}⚙{RESET}" # Add description @@ -205,9 +255,24 @@ def run_picker(tty_input: TTYInput) -> Optional[PickerResult]: if ch in ('\r', '\n'): # Enter - run if filtered: + selected_tool = filtered[selected] clear_dropdown(last_drawn) _write(SHOW_CURSOR) - return PickerResult(filtered[selected]["name"], {}) + # If it's a registry tool, install it first + if selected_tool.get("_registry") or selected_tool.get("registry"): + _write(f"{YELLOW}Installing {selected_tool['name']}...{RESET}\n") + import subprocess + proc = subprocess.run( + ["cmdforge", "registry", "install", selected_tool["name"]], + capture_output=True, text=True + ) + if proc.returncode != 0: + _write(f"Install failed: {proc.stderr}\n") + return None + # Extract tool name from "owner/name" format + tool_name = selected_tool["name"].split("/")[-1] + return PickerResult(tool_name, {}) + return PickerResult(selected_tool["name"], {}) elif ch == '\t': # Tab - configure args or run if filtered: diff --git a/src/cmdforge/preflight.py b/src/cmdforge/preflight.py index ef6bfe7..a62575f 100644 --- a/src/cmdforge/preflight.py +++ b/src/cmdforge/preflight.py @@ -24,6 +24,11 @@ class PreflightReport: compatibility: List[Dict[str, Any]] = field(default_factory=list) contract_proposal: Optional[Dict[str, Any]] = None regression: Optional[Dict[str, Any]] = None + reuse_opportunities: List[Dict[str, Any]] = field(default_factory=list) + audit_evidence: Optional[Dict[str, Any]] = None + 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: @@ -39,6 +44,8 @@ class PreflightReport: compatibility=self.compatibility + other.compatibility, contract_proposal=self.contract_proposal or other.contract_proposal, regression=self.regression or other.regression, + reuse_opportunities=self.reuse_opportunities + other.reuse_opportunities, + audit_evidence=self.audit_evidence or other.audit_evidence, ) def to_dict(self) -> Dict[str, Any]: @@ -51,6 +58,8 @@ class PreflightReport: "compatibility": list(self.compatibility), "contract_proposal": copy.deepcopy(self.contract_proposal), "regression": copy.deepcopy(self.regression), + "reuse_opportunities": list(self.reuse_opportunities), + "audit_evidence": copy.deepcopy(self.audit_evidence) if self.audit_evidence else None, } @@ -78,6 +87,8 @@ def analyze_tool( if check_local_dependencies: _check_toolstep_compatibility(tool, report) _check_dependencies(tool, report) + _check_reuse_opportunities(tool, report) + _add_audit_evidence(tool, report) if registry_client: _check_similar_tools(tool, report, registry_client) return report @@ -360,6 +371,85 @@ def _is_semver(version: str) -> bool: return Version.parse(version) is not None +def _check_reuse_opportunities(tool: Tool, report: PreflightReport): + """Detect repeated step patterns that could be extracted into reusable tools.""" + from .tool import list_tools, load_tool + + if len(tool.steps) < 2: + return + + # Build a signature of each step for comparison + step_signatures = [] + for step in tool.steps: + sig = _step_signature(step) + if sig: + step_signatures.append(sig) + + if not step_signatures: + return + + # Check if this step sequence appears in other local tools + our_sequence = tuple(step_signatures) + for name in list_tools(): + if name == tool.name: + continue + other = load_tool(name) + if not other or len(other.steps) < 2: + continue + other_sigs = [_step_signature(s) for s in other.steps] + other_sigs = [s for s in other_sigs if s] + if our_sequence == tuple(other_sigs): + report.reuse_opportunities.append({ + "type": "duplicate_sequence", + "tool": name, + "detail": f"Step sequence identical to '{name}' — consider extracting shared logic", + }) + elif len(our_sequence) >= 2: + # Check for overlapping subsequences + for start in range(len(other_sigs) - 1): + for length in range(2, min(len(our_sequence), len(other_sigs) - start) + 1): + if our_sequence[:length] == tuple(other_sigs[start:start + length]): + report.reuse_opportunities.append({ + "type": "shared_subsequence", + "tool": name, + "length": length, + "detail": f"Shares {length} step(s) with '{name}' — possible extraction candidate", + }) + break + + +def _step_signature(step) -> str: + """Build a comparable signature for a step.""" + if hasattr(step, "prompt"): + return f"prompt:{step.provider}" + elif hasattr(step, "code"): + return "code" + elif hasattr(step, "tool"): + return f"tool:{step.tool}" + elif hasattr(step, "server"): + return f"mcp:{step.server}" + return "" + + +def _add_audit_evidence(tool: Tool, report: PreflightReport): + """Attach audit evidence metadata to the report.""" + from datetime import datetime, timezone + + report.audit_evidence = { + "tool_version": tool.version or "", + "engine_version": "1.0", + "timestamp": datetime.now(timezone.utc).isoformat(), + "checks_run": [ + "config_integrity", + "contracts", + "secrets", + "dependencies", + "toolstep_compatibility", + "reuse_opportunities", + ], + } + + def infer_contracts(tool: Tool) -> Dict[str, Any]: """Propose input/output schemas without mutating the tool.