M8.C/A/D1/D2: Add reuse detection, audit evidence, deprecation, registry-aware picker
This commit is contained in:
@@ -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:
|
||||
|
||||
Reference in New Issue
Block a user