mirror of
https://github.com/HKUDS/nanobot.git
synced 2026-08-05 17:08:33 +00:00
dd2cb4ca91
maintainer edit: CLI Apps settings now treats optional catalog caches as refresh candidates without blocking the initial payload, and pending polling stops when refresh is throttled instead of running indefinitely.
1318 lines
50 KiB
Python
1318 lines
50 KiB
Python
"""CLI-Anything catalog, install state, and safe CLI execution."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
import shlex
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from dataclasses import dataclass
|
|
from importlib import metadata as importlib_metadata
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from urllib.parse import urlparse
|
|
|
|
import httpx
|
|
|
|
from nanobot.apps.protocol import app_manifest, compact_dict
|
|
from nanobot.config.paths import get_runtime_subdir
|
|
from nanobot.security.workspace_policy import is_path_within
|
|
|
|
CLI_ANYTHING_REGISTRY_URL = "https://hkuds.github.io/CLI-Anything/registry.json"
|
|
CLI_ANYTHING_PUBLIC_REGISTRY_URL = "https://hkuds.github.io/CLI-Anything/public_registry.json"
|
|
CLI_ANYTHING_RAW_BASE = "https://raw.githubusercontent.com/HKUDS/CLI-Anything/main"
|
|
NANOBOT_EXTENSION_REGISTRY_URL = "https://raw.githubusercontent.com/Re-bin/nanobot-extension/main/registry.json"
|
|
NANOBOT_EXTENSION_RAW_BASE = "https://raw.githubusercontent.com/Re-bin/nanobot-extension/main"
|
|
_CATALOG_SOURCES = (
|
|
("harness", CLI_ANYTHING_REGISTRY_URL, CLI_ANYTHING_RAW_BASE, True),
|
|
("public", CLI_ANYTHING_PUBLIC_REGISTRY_URL, CLI_ANYTHING_RAW_BASE, True),
|
|
("extensions", NANOBOT_EXTENSION_REGISTRY_URL, NANOBOT_EXTENSION_RAW_BASE, False),
|
|
)
|
|
|
|
_MAX_TOOL_OUTPUT_CHARS = 12_000
|
|
_MAX_ARTIFACT_SCAN_PATHS = 4_000
|
|
_MAX_ARTIFACT_REPORT = 12
|
|
_SAFE_NAME_RE = re.compile(r"[^a-z0-9_-]+")
|
|
_SAFE_NPM_DIR_RE = re.compile(r"^[a-z0-9._-]+$", re.IGNORECASE)
|
|
_MENTION_RE = re.compile(r"(^|[\s([{])@([a-z0-9_-]+)\b", re.IGNORECASE)
|
|
_SHELL_META_CHARS = ("|", "&&", "||", ";", "$(", "`", ">", "<")
|
|
_ENDORSEMENT_WORD_RE = re.compile(r"\bofficial\s+", re.IGNORECASE)
|
|
_ARTIFACT_EXTENSIONS = frozenset({
|
|
".csv",
|
|
".drawio",
|
|
".gif",
|
|
".html",
|
|
".jpeg",
|
|
".jpg",
|
|
".json",
|
|
".md",
|
|
".pdf",
|
|
".png",
|
|
".svg",
|
|
".txt",
|
|
".vsdx",
|
|
".webp",
|
|
".xml",
|
|
})
|
|
_INLINE_ARTIFACT_EXTENSIONS = frozenset({".gif", ".jpeg", ".jpg", ".png", ".webp"})
|
|
_ARTIFACT_IGNORE_DIRS = frozenset({
|
|
".git",
|
|
".hg",
|
|
".mypy_cache",
|
|
".nanobot",
|
|
".pytest_cache",
|
|
".ruff_cache",
|
|
".venv",
|
|
"__pycache__",
|
|
"build",
|
|
"dist",
|
|
"node_modules",
|
|
"venv",
|
|
})
|
|
|
|
|
|
class CliAppError(ValueError):
|
|
"""User-facing CLI Apps failure."""
|
|
|
|
def __init__(self, message: str, *, status: int = 400) -> None:
|
|
super().__init__(message)
|
|
self.message = message
|
|
self.status = status
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class CliAppsRuntimeConfig:
|
|
"""Runtime knobs for CLI Apps."""
|
|
|
|
install_timeout: int = 300
|
|
run_timeout: int = 60
|
|
catalog_ttl_seconds: int = 3600
|
|
|
|
|
|
_BRANDS: dict[str, tuple[str, str]] = {
|
|
"1password-cli": ("1password", "#3B66BC"),
|
|
"arcgis": ("arcgis", "#2C7AC3"),
|
|
"arcgis-pro": ("arcgis", "#2C7AC3"),
|
|
"audacity": ("audacity", "#0000CC"),
|
|
"blender": ("blender", "#E87D0D"),
|
|
"browser": ("googlechrome", "#4285F4"),
|
|
"calibre": ("calibre", "#45B29D"),
|
|
"chromadb": ("chroma", "#FFDE2D"),
|
|
"comfyui": ("comfyui", "#111827"),
|
|
"contentful": ("contentful", "#2478CC"),
|
|
"dify": ("dify", "#155EEF"),
|
|
"drawio": ("diagramsdotnet", "#F08705"),
|
|
"elevenlabs": ("elevenlabs", "#000000"),
|
|
"eth2-quickstart": ("ethereum", "#627EEA"),
|
|
"firefly-iii": ("fireflyiii", "#CD5029"),
|
|
"freecad": ("freecad", "#418FDE"),
|
|
"generate-veo-video": ("googlegemini", "#8E75B2"),
|
|
"gimp": ("gimp", "#5C5543"),
|
|
"godot": ("godotengine", "#478CBF"),
|
|
"hacker-feeds-cli": ("rss", "#FFA500"),
|
|
"inkscape": ("inkscape", "#000000"),
|
|
"intelwatch": ("intel", "#0071C5"),
|
|
"iterm2": ("iterm2", "#000000"),
|
|
"jimeng": ("bytedance", "#3C8CFF"),
|
|
"joplin": ("joplin", "#1071D3"),
|
|
"kdenlive": ("kdenlive", "#527EB2"),
|
|
"krita": ("krita", "#3BABFF"),
|
|
"libreoffice": ("libreoffice", "#18A303"),
|
|
"mailchimp": ("mailchimp", "#FFE01B"),
|
|
"mermaid": ("mermaid", "#FF3670"),
|
|
"minimax": ("minimax", "#111827"),
|
|
"musescore": ("musescore", "#1A70B8"),
|
|
"n8n": ("n8n", "#EA4B71"),
|
|
"notebooklm": ("googlenotebooklm", "#4285F4"),
|
|
"obs-studio": ("obsstudio", "#302E31"),
|
|
"obsidian": ("obsidian", "#7C3AED"),
|
|
"ollama": ("ollama", "#000000"),
|
|
"pm2": ("pm2", "#2B037A"),
|
|
"qgis": ("qgis", "#589632"),
|
|
"safari": ("safari", "#006CFF"),
|
|
"sanity": ("sanity", "#F03E2F"),
|
|
"sentry": ("sentry", "#362D59"),
|
|
"sketch": ("sketch", "#F7B500"),
|
|
"shopify": ("shopify", "#7AB55C"),
|
|
"nsight-graphics": ("nvidia", "#76B900"),
|
|
"unrealinsights": ("unrealengine", "#0E1128"),
|
|
"ueatelier": ("unrealengine", "#0E1128"),
|
|
"ve-twini": ("x", "#000000"),
|
|
"wecom": ("wechat", "#07C160"),
|
|
"suno": ("suno", "#000000"),
|
|
"lldb": ("llvm", "#262D3A"),
|
|
"android-cli": ("android", "#3DDC84"),
|
|
"adguardhome": ("adguard", "#68BC71"),
|
|
"zotero": ("zotero", "#CC2936"),
|
|
"zoom": ("zoom", "#0B5CFF"),
|
|
}
|
|
|
|
_BRAND_DOMAINS: dict[str, tuple[str, str]] = {
|
|
"3mf": ("3mf.io", "#00A1DE"),
|
|
"anygen": ("anygen.io", "#111827"),
|
|
"clibrowser": ("github.com/allthingssecurity/clibrowser", "#24292F"),
|
|
"cloudanalyzer": ("github.com/rsasaki0109/CloudAnalyzer", "#2563EB"),
|
|
"cloudcompare": ("cloudcompare.org", "#4D83C3"),
|
|
"deployhq": ("deployhq.com", "#00A2D9"),
|
|
"exa": ("exa.ai", "#111827"),
|
|
"feishu": ("larksuite.com", "#00A5FF"),
|
|
"inkstitch": ("inkstitch.org", "#222222"),
|
|
"macrocli": ("github.com/HKUDS/CLI-Anything/tree/main/macrocli", "#24292F"),
|
|
"mubu": ("mubu.com", "#16A085"),
|
|
"nslogger": ("github.com/fpillet/NSLogger", "#24292F"),
|
|
"novita": ("novita.ai", "#7C3AED"),
|
|
"openscreen": ("openscreen.com", "#2563EB"),
|
|
"py4csr": ("github.com/yanmingyu92/py4csr", "#24292F"),
|
|
"quietshrink": ("github.com/achiya-automation/quietshrink", "#111827"),
|
|
"renderdoc": ("renderdoc.org", "#2C7DB8"),
|
|
"rms": ("rms.teltonika-networks.com", "#0054A6"),
|
|
"sbox": ("sbox.game", "#F59E0B"),
|
|
"seaclip": ("github.com/SeaClip-Lite/SeaClip", "#0284C7"),
|
|
"shotcut": ("shotcut.org", "#3B82F6"),
|
|
"slay-the-spire-ii": ("megacrit.com", "#B91C1C"),
|
|
"stata": ("stata.com", "#1F4E79"),
|
|
"unimol-tools": ("github.com/deepmodeling/Uni-Mol", "#4F46E5"),
|
|
"videocaptioner": ("github.com/WEIFENG2333/VideoCaptioner", "#2563EB"),
|
|
"wiremock": ("wiremock.org", "#FF6A00"),
|
|
}
|
|
|
|
_BRAND_ALIASES: dict[str, str] = {
|
|
"1password": "1password-cli",
|
|
"dify-workflow": "dify",
|
|
"feishu-lark": "feishu",
|
|
"lark-cli": "feishu",
|
|
"minimax-cli": "minimax",
|
|
"obsidian-cli": "obsidian",
|
|
"slay-the-spire-2": "slay-the-spire-ii",
|
|
"slay-the-spire-ii": "slay-the-spire-ii",
|
|
"unimol-tools": "unimol-tools",
|
|
"unimol": "unimol-tools",
|
|
"veo": "generate-veo-video",
|
|
}
|
|
|
|
_BRAND_TRAILING_WORDS = ("cli", "workflow", "workflows", "app", "apps", "tool", "tools")
|
|
|
|
|
|
def _now() -> float:
|
|
return time.time()
|
|
|
|
|
|
def _safe_skill_name(name: str) -> str:
|
|
clean = _SAFE_NAME_RE.sub("-", name.lower()).strip("-")
|
|
return f"cli-app-{clean or 'app'}"
|
|
|
|
|
|
def _has_shell_meta(command: str) -> bool:
|
|
return any(char in command for char in _SHELL_META_CHARS)
|
|
|
|
|
|
def _command_exists(command: str) -> bool:
|
|
try:
|
|
parts = shlex.split(command)
|
|
except ValueError:
|
|
return False
|
|
if not parts:
|
|
return False
|
|
return shutil.which(parts[0]) is not None
|
|
|
|
|
|
def _is_pip_install_command(command: str) -> bool:
|
|
try:
|
|
tokens = shlex.split(command)
|
|
except ValueError:
|
|
return False
|
|
return (
|
|
len(tokens) >= 3
|
|
and tokens[:2] == ["pip", "install"]
|
|
) or (
|
|
len(tokens) >= 5
|
|
and tokens[1:4] == ["-m", "pip", "install"]
|
|
and tokens[0] in {"python", "python3", sys.executable}
|
|
)
|
|
|
|
|
|
def _pip_uninstall_args_from_command(command: str) -> list[str] | None:
|
|
if not command or _has_shell_meta(command):
|
|
return None
|
|
try:
|
|
tokens = shlex.split(command)
|
|
except ValueError:
|
|
return None
|
|
if tokens[:2] == ["pip", "uninstall"]:
|
|
args = tokens[2:]
|
|
elif (
|
|
len(tokens) >= 5
|
|
and tokens[1:4] == ["-m", "pip", "uninstall"]
|
|
and tokens[0] in {"python", "python3", sys.executable}
|
|
):
|
|
args = tokens[4:]
|
|
else:
|
|
return None
|
|
packages = [arg for arg in args if arg not in {"-y", "--yes"}]
|
|
if not packages or any(arg.startswith("-") for arg in packages):
|
|
return None
|
|
return packages
|
|
|
|
|
|
def _console_script_distribution(entry_point: str) -> str | None:
|
|
if not entry_point:
|
|
return None
|
|
try:
|
|
distributions = importlib_metadata.distributions()
|
|
except Exception:
|
|
return None
|
|
for distribution in distributions:
|
|
try:
|
|
entry_points = distribution.entry_points
|
|
except Exception:
|
|
continue
|
|
for item in entry_points:
|
|
if item.group != "console_scripts" or item.name != entry_point:
|
|
continue
|
|
try:
|
|
name = distribution.metadata.get("Name")
|
|
except Exception:
|
|
name = None
|
|
return str(name or getattr(distribution, "name", "") or "").strip() or None
|
|
return None
|
|
|
|
|
|
def _brand_key(value: str) -> str:
|
|
return _SAFE_NAME_RE.sub("-", value.lower()).replace("_", "-").strip("-")
|
|
|
|
|
|
def _brand_candidates(app: dict[str, Any]) -> list[str]:
|
|
values = [
|
|
str(app.get("name") or ""),
|
|
str(app.get("display_name") or ""),
|
|
str(app.get("entry_point") or "").removeprefix("cli-anything-"),
|
|
]
|
|
seen: set[str] = set()
|
|
candidates: list[str] = []
|
|
for value in values:
|
|
key = _brand_key(value)
|
|
while key and key not in seen:
|
|
seen.add(key)
|
|
candidates.append(key)
|
|
parts = key.split("-")
|
|
if len(parts) <= 1 or parts[-1] not in _BRAND_TRAILING_WORDS:
|
|
break
|
|
key = "-".join(parts[:-1])
|
|
return candidates
|
|
|
|
|
|
def _brand_payload(app: dict[str, Any]) -> tuple[str | None, str | None]:
|
|
declared_logo = str(app.get("logo_url") or "").strip()
|
|
if declared_logo.startswith(("https://", "/")):
|
|
declared_color = str(app.get("brand_color") or "").strip()
|
|
return declared_logo, declared_color or None
|
|
|
|
brand = None
|
|
domain_brand = None
|
|
for candidate in _brand_candidates(app):
|
|
key = _BRAND_ALIASES.get(candidate, candidate)
|
|
brand = _BRANDS.get(key)
|
|
if brand:
|
|
break
|
|
domain_brand = _BRAND_DOMAINS.get(key)
|
|
if domain_brand:
|
|
break
|
|
if not brand:
|
|
if not domain_brand:
|
|
return None, None
|
|
domain, color = domain_brand
|
|
return f"https://www.google.com/s2/favicons?domain={domain}&sz=64", color
|
|
slug, color = brand
|
|
return f"https://cdn.simpleicons.org/{slug}/{color.lstrip('#')}", color
|
|
|
|
|
|
def _read_json(path: Path) -> dict[str, Any] | None:
|
|
try:
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError):
|
|
return None
|
|
return data if isinstance(data, dict) else None
|
|
|
|
|
|
def _write_json(path: Path, data: dict[str, Any]) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
payload = json.dumps(data, indent=2, ensure_ascii=False)
|
|
tmp_path = path.with_name(f".{path.name}.{os.getpid()}.{int(_now() * 1_000_000)}.tmp")
|
|
try:
|
|
tmp_path.write_text(payload, encoding="utf-8")
|
|
tmp_path.replace(path)
|
|
finally:
|
|
if tmp_path.exists():
|
|
tmp_path.unlink()
|
|
|
|
|
|
def _safe_skill_path(value: str) -> str | None:
|
|
if not value.startswith("skills/"):
|
|
return None
|
|
parts = value.split("/")
|
|
if any(part in {"", ".", ".."} for part in parts):
|
|
return None
|
|
return value if parts[-1] == "SKILL.md" else None
|
|
|
|
|
|
def _skill_content_url(skill_md: str, *, raw_base: str = CLI_ANYTHING_RAW_BASE) -> str | None:
|
|
safe_path = _safe_skill_path(skill_md)
|
|
if safe_path:
|
|
return f"{raw_base.rstrip('/')}/{safe_path}"
|
|
parsed = urlparse(skill_md)
|
|
if parsed.scheme != "https" or parsed.netloc != "raw.githubusercontent.com":
|
|
return None
|
|
raw_prefix = raw_base.rstrip("/") + "/"
|
|
if not skill_md.startswith(raw_prefix):
|
|
return None
|
|
suffix = skill_md.removeprefix(raw_prefix)
|
|
return skill_md if _safe_skill_path(suffix) else None
|
|
|
|
|
|
def _truncate(text: str, limit: int = _MAX_TOOL_OUTPUT_CHARS) -> str:
|
|
if len(text) <= limit:
|
|
return text
|
|
omitted = len(text) - limit
|
|
return text[:limit] + f"\n\n... truncated {omitted} characters ..."
|
|
|
|
|
|
def _catalog_description(app: dict[str, Any]) -> str:
|
|
"""Return catalog copy without implying vendor endorsement."""
|
|
description = str(app.get("description") or "")
|
|
return _ENDORSEMENT_WORD_RE.sub("", description).strip()
|
|
|
|
|
|
class CliAppManager:
|
|
"""Manage CLI-Anything registry entries and local install state."""
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
workspace: Path,
|
|
data_dir: Path | None = None,
|
|
runtime: CliAppsRuntimeConfig | None = None,
|
|
) -> None:
|
|
self.workspace = Path(workspace).expanduser()
|
|
self.data_dir = Path(data_dir) if data_dir is not None else get_runtime_subdir("cli-apps")
|
|
self.runtime = runtime or CliAppsRuntimeConfig()
|
|
|
|
@property
|
|
def installed_path(self) -> Path:
|
|
return self.data_dir / "installed.json"
|
|
|
|
def _cache_path(self, source: str) -> Path:
|
|
return self.data_dir / f"{source}_registry_cache.json"
|
|
|
|
def _cached_registry(self, cache_path: Path) -> tuple[dict[str, Any] | None, float]:
|
|
cached = _read_json(cache_path)
|
|
if not cached:
|
|
return None, 0.0
|
|
data = cached.get("data")
|
|
if not isinstance(data, dict):
|
|
return None, 0.0
|
|
try:
|
|
cached_at = float(cached.get("_cached_at", 0))
|
|
except (TypeError, ValueError):
|
|
cached_at = 0.0
|
|
return data, cached_at
|
|
|
|
def _load_installed(self) -> dict[str, Any]:
|
|
data = _read_json(self.installed_path) or {}
|
|
apps = data.get("apps") if isinstance(data.get("apps"), dict) else data
|
|
return apps if isinstance(apps, dict) else {}
|
|
|
|
def _save_installed(self, installed: dict[str, Any]) -> None:
|
|
_write_json(self.installed_path, {"schema_version": 1, "apps": installed})
|
|
|
|
def installed_names(self) -> list[str]:
|
|
"""Return registry names explicitly installed through CLI Apps."""
|
|
return sorted(str(name) for name in self._load_installed())
|
|
|
|
def _fetch_registry(
|
|
self,
|
|
url: str,
|
|
cache_path: Path,
|
|
*,
|
|
force_refresh: bool = False,
|
|
) -> dict[str, Any]:
|
|
data, cached_at = self._cached_registry(cache_path)
|
|
if (
|
|
not force_refresh
|
|
and data is not None
|
|
and _now() - cached_at < self.runtime.catalog_ttl_seconds
|
|
):
|
|
return data
|
|
|
|
try:
|
|
response = httpx.get(url, timeout=15.0, follow_redirects=True)
|
|
response.raise_for_status()
|
|
fetched = response.json()
|
|
if not isinstance(fetched, dict):
|
|
raise ValueError("registry response must be an object")
|
|
except Exception:
|
|
if data is not None:
|
|
return data
|
|
raise
|
|
|
|
_write_json(cache_path, {"_cached_at": _now(), "data": fetched})
|
|
return fetched
|
|
|
|
def catalog(
|
|
self,
|
|
*,
|
|
force_refresh: bool = False,
|
|
cache_only: bool = False,
|
|
) -> tuple[list[dict[str, Any]], str | None]:
|
|
registries: list[tuple[str, str, dict[str, Any]]] = []
|
|
for source, url, raw_base, required in _CATALOG_SOURCES:
|
|
try:
|
|
cache_path = self._cache_path(source)
|
|
if cache_only:
|
|
registry, _ = self._cached_registry(cache_path)
|
|
if registry is None:
|
|
continue
|
|
else:
|
|
registry = self._fetch_registry(
|
|
url,
|
|
cache_path,
|
|
force_refresh=force_refresh,
|
|
)
|
|
except Exception:
|
|
if required:
|
|
raise
|
|
continue
|
|
registries.append((source, raw_base, registry))
|
|
apps_by_name: dict[str, dict[str, Any]] = {}
|
|
updated_values: list[str] = []
|
|
for source, raw_base, registry in registries:
|
|
meta = registry.get("meta")
|
|
if isinstance(meta, dict) and isinstance(meta.get("updated"), str):
|
|
updated_values.append(meta["updated"])
|
|
for row in registry.get("clis", []):
|
|
if not isinstance(row, dict) or not row.get("name"):
|
|
continue
|
|
entry = dict(row)
|
|
entry["_source"] = source
|
|
entry["_raw_base"] = raw_base
|
|
key = str(entry["name"]).lower()
|
|
previous = apps_by_name.get(key)
|
|
if previous:
|
|
previous_source = str(previous.get("_source") or source)
|
|
merged_source = (
|
|
previous_source if previous_source == source else f"{previous_source}+{source}"
|
|
)
|
|
apps_by_name[key] = {**previous, **entry, "_source": merged_source}
|
|
else:
|
|
apps_by_name[key] = entry
|
|
return list(apps_by_name.values()), max(updated_values) if updated_values else None
|
|
|
|
def catalog_cache_fresh(self, *, include_optional: bool = False) -> bool:
|
|
for source, _url, _raw_base, required in _CATALOG_SOURCES:
|
|
if not required and not include_optional:
|
|
continue
|
|
data, cached_at = self._cached_registry(self._cache_path(source))
|
|
if data is None or _now() - cached_at >= self.runtime.catalog_ttl_seconds:
|
|
return False
|
|
return True
|
|
|
|
def _manifest_source(self, app: dict[str, Any]) -> str:
|
|
source = str(app.get("_source") or "harness")
|
|
if source == "extensions":
|
|
return "nanobot-extension"
|
|
return f"cli-anything:{source}"
|
|
|
|
def _trust_registry(self, app: dict[str, Any]) -> str:
|
|
return "nanobot-extension" if str(app.get("_source") or "") == "extensions" else "cli-anything"
|
|
|
|
def get_app(self, name: str, *, force_refresh: bool = False) -> dict[str, Any]:
|
|
wanted = name.lower()
|
|
for app in self.catalog(force_refresh=force_refresh)[0]:
|
|
if str(app.get("name", "")).lower() == wanted:
|
|
return app
|
|
raise CliAppError(f"CLI app '{name}' not found", status=404)
|
|
|
|
def mentioned_installed_apps(self, text: str) -> list[dict[str, str]]:
|
|
"""Return installed CLI Apps referenced as ``@name`` in user text."""
|
|
if "@" not in text:
|
|
return []
|
|
installed = self._load_installed()
|
|
if not installed:
|
|
return []
|
|
installed_by_name = {
|
|
str(name).lower(): (str(name), data if isinstance(data, dict) else {})
|
|
for name, data in installed.items()
|
|
}
|
|
seen: set[str] = set()
|
|
mentions: list[dict[str, str]] = []
|
|
for match in _MENTION_RE.finditer(text):
|
|
wanted = str(match.group(2)).lower()
|
|
if wanted in seen or wanted not in installed_by_name:
|
|
continue
|
|
installed_name, data = installed_by_name[wanted]
|
|
seen.add(wanted)
|
|
entry_point = str(data.get("entry_point") or "")
|
|
mentions.append(
|
|
{
|
|
"name": installed_name,
|
|
"entry_point": entry_point,
|
|
"source": str(data.get("source") or ""),
|
|
"skill": f"skills/{_safe_skill_name(installed_name)}/SKILL.md",
|
|
"tool": "run_cli_app",
|
|
}
|
|
)
|
|
return mentions
|
|
|
|
def _strategy(self, app: dict[str, Any]) -> str:
|
|
package_manager = str(app.get("package_manager") or "").lower()
|
|
install_strategy = str(app.get("install_strategy") or "").lower()
|
|
if package_manager == "bundled" or install_strategy == "bundled":
|
|
return "bundled"
|
|
if package_manager in {"npm", "brew", "uv", "pip"}:
|
|
return package_manager
|
|
if app.get("npm_package"):
|
|
return "npm"
|
|
install_cmd = str(app.get("install_cmd") or "")
|
|
if _is_pip_install_command(install_cmd):
|
|
return "pip"
|
|
return "unsupported"
|
|
|
|
def _install_supported(self, app: dict[str, Any]) -> bool:
|
|
if self._strategy(app) == "unsupported":
|
|
return False
|
|
install_cmd = str(app.get("install_cmd") or "")
|
|
return not _has_shell_meta(install_cmd)
|
|
|
|
def _skill_path(self, name: str) -> Path:
|
|
return self.workspace / "skills" / _safe_skill_name(name) / "SKILL.md"
|
|
|
|
def _app_payload(
|
|
self,
|
|
app: dict[str, Any],
|
|
installed: dict[str, Any],
|
|
) -> dict[str, Any]:
|
|
name = str(app["name"])
|
|
entry_point = str(app.get("entry_point") or "")
|
|
install_supported = self._install_supported(app)
|
|
is_installed = name in installed
|
|
available = bool(entry_point and shutil.which(entry_point))
|
|
if is_installed and available:
|
|
status = "installed"
|
|
elif is_installed:
|
|
status = "missing"
|
|
elif not install_supported:
|
|
status = "unsupported"
|
|
elif available:
|
|
status = "available"
|
|
else:
|
|
status = "not_installed"
|
|
logo_url, brand_color = _brand_payload(app)
|
|
return {
|
|
"name": name,
|
|
"display_name": app.get("display_name") or name,
|
|
"category": app.get("category") or "uncategorized",
|
|
"description": _catalog_description(app),
|
|
"requires": app.get("requires") or "",
|
|
"source": app.get("_source") or "harness",
|
|
"entry_point": entry_point,
|
|
"install_supported": install_supported,
|
|
"installed": is_installed,
|
|
"available": available,
|
|
"status": status,
|
|
"logo_url": logo_url,
|
|
"brand_color": brand_color,
|
|
"skill_installed": self._skill_path(name).is_file(),
|
|
"manifest": self._manifest_payload(app, logo_url=logo_url, brand_color=brand_color),
|
|
}
|
|
|
|
def _package_ref(self, app: dict[str, Any]) -> dict[str, Any] | None:
|
|
strategy = self._strategy(app)
|
|
name = ""
|
|
if strategy == "pip":
|
|
try:
|
|
uninstall = self._pip_uninstall_argv(app)
|
|
except CliAppError:
|
|
uninstall = None
|
|
name = uninstall[-1] if uninstall else ""
|
|
elif strategy == "npm":
|
|
name = str(app.get("npm_package") or "").strip()
|
|
elif strategy in {"brew", "uv"}:
|
|
try:
|
|
uninstall = self._argv_for_action(app, "uninstall")
|
|
except CliAppError:
|
|
uninstall = None
|
|
if uninstall:
|
|
name = uninstall[-1]
|
|
if not strategy or strategy in {"unsupported", "bundled"}:
|
|
return None
|
|
return compact_dict({"manager": strategy, "name": name})
|
|
|
|
def _manifest_payload(
|
|
self,
|
|
app: dict[str, Any],
|
|
*,
|
|
logo_url: str | None,
|
|
brand_color: str | None,
|
|
) -> dict[str, Any]:
|
|
name = str(app["name"])
|
|
entry_point = str(app.get("entry_point") or "")
|
|
strategy = self._strategy(app)
|
|
skill_path = f"skills/{_safe_skill_name(name)}/SKILL.md"
|
|
capabilities = [
|
|
compact_dict({
|
|
"type": "cli",
|
|
"entry_point": entry_point,
|
|
"package": self._package_ref(app),
|
|
}),
|
|
{"type": "skill", "path": skill_path},
|
|
]
|
|
install_supported = self._install_supported(app)
|
|
install = compact_dict({
|
|
"supported": install_supported,
|
|
"strategy": strategy,
|
|
"managed_paths": [skill_path],
|
|
"verification": ["entry_point_available"] if entry_point else [],
|
|
})
|
|
remove = compact_dict({
|
|
"supported": strategy != "unsupported",
|
|
"strategy": strategy,
|
|
"managed_paths": [skill_path],
|
|
"verification": (
|
|
["package_manager_ok", "entry_point_absent", "managed_paths_absent"]
|
|
if strategy not in {"bundled", "unsupported"}
|
|
else ["nanobot_state_absent", "managed_paths_absent"]
|
|
),
|
|
})
|
|
return app_manifest(
|
|
app_id=name,
|
|
display_name=str(app.get("display_name") or name),
|
|
version=str(app.get("version") or ""),
|
|
description=_catalog_description(app),
|
|
category=str(app.get("category") or "uncategorized"),
|
|
source=self._manifest_source(app),
|
|
logo_url=logo_url,
|
|
brand_color=brand_color,
|
|
capabilities=capabilities,
|
|
install=install,
|
|
remove=remove,
|
|
trust={
|
|
"registry": self._trust_registry(app),
|
|
"level": "catalog",
|
|
"review_status": "catalog_entry",
|
|
},
|
|
)
|
|
|
|
def payload(self, *, force_refresh: bool = False, cache_only: bool = False) -> dict[str, Any]:
|
|
apps, updated = self.catalog(force_refresh=force_refresh, cache_only=cache_only)
|
|
installed = self._load_installed()
|
|
rows = [self._app_payload(app, installed) for app in apps]
|
|
rows.sort(key=lambda item: (str(item["category"]), str(item["display_name"]).lower()))
|
|
return {
|
|
"apps": rows,
|
|
"installed_count": sum(1 for item in rows if item["installed"]),
|
|
"catalog_updated_at": updated,
|
|
}
|
|
|
|
def installed_payload(self) -> dict[str, Any]:
|
|
installed = self._load_installed()
|
|
rows = []
|
|
for name, raw_entry in sorted(installed.items()):
|
|
entry = raw_entry if isinstance(raw_entry, dict) else {}
|
|
strategy = str(entry.get("strategy") or "bundled")
|
|
app = {
|
|
"name": str(name),
|
|
"display_name": str(entry.get("display_name") or name),
|
|
"category": str(entry.get("category") or "installed"),
|
|
"description": str(entry.get("description") or ""),
|
|
"requires": str(entry.get("requires") or ""),
|
|
"_source": str(entry.get("source") or "local"),
|
|
"entry_point": str(entry.get("entry_point") or ""),
|
|
"package_manager": strategy,
|
|
}
|
|
rows.append(self._app_payload(app, installed))
|
|
return {
|
|
"apps": rows,
|
|
"installed_count": len(rows),
|
|
"catalog_updated_at": None,
|
|
}
|
|
|
|
def _pip_package_from_install(self, app: dict[str, Any]) -> str | None:
|
|
install_cmd = str(app.get("install_cmd") or "")
|
|
try:
|
|
tokens = shlex.split(install_cmd)
|
|
except ValueError:
|
|
return None
|
|
if tokens[:2] == ["pip", "install"]:
|
|
args = tokens[2:]
|
|
elif len(tokens) >= 5 and tokens[1:4] == ["-m", "pip", "install"]:
|
|
args = tokens[4:]
|
|
else:
|
|
return None
|
|
args = [arg for arg in args if not arg.startswith("-")]
|
|
if len(args) != 1 or args[0].startswith("git+"):
|
|
return None
|
|
return args[0]
|
|
|
|
@staticmethod
|
|
def _pip_available() -> bool:
|
|
"""Return True if pip is importable for the current interpreter."""
|
|
from importlib.util import find_spec
|
|
|
|
return find_spec("pip") is not None
|
|
|
|
def _pip_install_argv(self, app: dict[str, Any], *, update: bool = False) -> list[str]:
|
|
install_cmd = str(app.get("install_cmd") or "")
|
|
if not _is_pip_install_command(install_cmd) or _has_shell_meta(install_cmd):
|
|
raise CliAppError("unsupported pip install command")
|
|
tokens = shlex.split(install_cmd)
|
|
args = tokens[2:] if tokens[:2] == ["pip", "install"] else tokens[4:]
|
|
pip_available = self._pip_available()
|
|
if pip_available:
|
|
prefix = [sys.executable, "-m", "pip", "install"]
|
|
elif shutil.which("uv"):
|
|
prefix = ["uv", "pip", "install", "--python", sys.executable]
|
|
else:
|
|
raise CliAppError("pip is not available and uv is not installed")
|
|
if update:
|
|
if pip_available:
|
|
prefix.extend(["--upgrade", "--force-reinstall"])
|
|
else:
|
|
prefix.extend(["--upgrade", "--reinstall"])
|
|
return prefix + args
|
|
|
|
def _pip_uninstall_argv(
|
|
self,
|
|
app: dict[str, Any],
|
|
installed_entry: dict[str, Any] | None = None,
|
|
) -> list[str]:
|
|
if self._pip_available():
|
|
prefix = [sys.executable, "-m", "pip", "uninstall", "-y"]
|
|
elif shutil.which("uv"):
|
|
prefix = ["uv", "pip", "uninstall", "--python", sys.executable]
|
|
else:
|
|
raise CliAppError("pip is not available and uv is not installed")
|
|
distribution = str((installed_entry or {}).get("pip_distribution") or "").strip()
|
|
if distribution:
|
|
return [*prefix, distribution]
|
|
uninstall_cmd = str(app.get("uninstall_cmd") or "")
|
|
packages = _pip_uninstall_args_from_command(uninstall_cmd)
|
|
if packages:
|
|
return [*prefix, *packages]
|
|
package = str(app.get("pip_package") or "").strip() or self._pip_package_from_install(app)
|
|
if not package:
|
|
entry_point = str(app.get("entry_point") or "").strip()
|
|
package = entry_point if entry_point.startswith("cli-anything-") else f"cli-anything-{_brand_key(str(app['name']))}"
|
|
return [*prefix, package]
|
|
|
|
def _npm_argv(self, app: dict[str, Any], action: str) -> list[str]:
|
|
npm = shutil.which("npm")
|
|
if not npm:
|
|
raise CliAppError("npm is not installed")
|
|
package = str(app.get("npm_package") or "")
|
|
if not package:
|
|
raise CliAppError("registry entry has no npm_package")
|
|
if action == "install":
|
|
return [npm, "install", "-g", package]
|
|
if action == "update":
|
|
return [npm, "install", "-g", package + "@latest"]
|
|
return [npm, "uninstall", "-g", package]
|
|
|
|
def _cleanup_stale_npm_install(self, app: dict[str, Any]) -> bool:
|
|
npm = shutil.which("npm")
|
|
package = str(app.get("npm_package") or "").strip()
|
|
if not npm or not package or "/" in package or _SAFE_NPM_DIR_RE.match(package) is None:
|
|
return False
|
|
result = self._run_argv([npm, "root", "-g"], timeout=min(self.runtime.install_timeout, 30))
|
|
if result.returncode != 0:
|
|
return False
|
|
root = Path(result.stdout.strip()).expanduser()
|
|
try:
|
|
root = root.resolve(strict=True)
|
|
except OSError:
|
|
return False
|
|
targets = [root / package, *root.glob(f".{package}-*")]
|
|
removed = False
|
|
for target in targets:
|
|
try:
|
|
resolved = target.resolve(strict=False)
|
|
if not is_path_within(resolved, root) or not target.is_dir():
|
|
continue
|
|
shutil.rmtree(target)
|
|
removed = True
|
|
except OSError:
|
|
continue
|
|
return removed
|
|
|
|
def _retry_stale_npm_install(
|
|
self,
|
|
app: dict[str, Any],
|
|
argv: list[str],
|
|
result: subprocess.CompletedProcess[str],
|
|
) -> subprocess.CompletedProcess[str]:
|
|
output = f"{result.stderr}\n{result.stdout}"
|
|
if "ENOTEMPTY" not in output or "rename" not in output:
|
|
return result
|
|
if not self._cleanup_stale_npm_install(app):
|
|
return result
|
|
return self._run_argv(argv, timeout=self.runtime.install_timeout)
|
|
|
|
def _split_safe_command(self, app: dict[str, Any], key: str, expected: str) -> list[str]:
|
|
command = str(app.get(key) or "")
|
|
if not command:
|
|
raise CliAppError(f"no {key} is defined for {app['name']}")
|
|
if _has_shell_meta(command):
|
|
raise CliAppError("script-style install commands are disabled in this MVP")
|
|
try:
|
|
argv = shlex.split(command)
|
|
except ValueError as exc:
|
|
raise CliAppError(f"invalid command: {exc}") from exc
|
|
if not argv or argv[0] != expected:
|
|
raise CliAppError(f"unsupported {expected} command")
|
|
return argv
|
|
|
|
def _argv_for_action(
|
|
self,
|
|
app: dict[str, Any],
|
|
action: str,
|
|
installed_entry: dict[str, Any] | None = None,
|
|
) -> list[str] | None:
|
|
strategy = self._strategy(app)
|
|
if strategy == "pip":
|
|
if action == "install":
|
|
return self._pip_install_argv(app)
|
|
if action == "update":
|
|
return self._pip_install_argv(app, update=True)
|
|
return self._pip_uninstall_argv(app, installed_entry=installed_entry)
|
|
if strategy == "npm":
|
|
return self._npm_argv(app, action)
|
|
if strategy == "brew":
|
|
key = {"install": "install_cmd", "update": "update_cmd", "uninstall": "uninstall_cmd"}[action]
|
|
return self._split_safe_command(app, key, "brew")
|
|
if strategy == "uv":
|
|
key = {"install": "install_cmd", "update": "update_cmd", "uninstall": "uninstall_cmd"}[action]
|
|
return self._split_safe_command(app, key, "uv")
|
|
if strategy == "bundled":
|
|
return None
|
|
raise CliAppError("this CLI app uses an unsupported install strategy")
|
|
|
|
def _run_argv(self, argv: list[str], *, timeout: int) -> subprocess.CompletedProcess[str]:
|
|
return subprocess.run(
|
|
argv,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=timeout,
|
|
)
|
|
|
|
def _installed_entry(self, app: dict[str, Any]) -> dict[str, Any]:
|
|
entry_point = str(app.get("entry_point") or "")
|
|
strategy = self._strategy(app)
|
|
entry: dict[str, Any] = {
|
|
"version": app.get("version") or "unknown",
|
|
"entry_point": entry_point,
|
|
"source": app.get("_source") or "harness",
|
|
"strategy": strategy,
|
|
"installed_at": int(_now()),
|
|
}
|
|
resolved = shutil.which(entry_point) if entry_point else None
|
|
if resolved:
|
|
entry["entry_point_path"] = resolved
|
|
if strategy == "pip":
|
|
distribution = _console_script_distribution(entry_point)
|
|
if distribution:
|
|
entry["pip_distribution"] = distribution
|
|
return entry
|
|
|
|
def _fetch_skill_content(self, app: dict[str, Any]) -> str | None:
|
|
skill_md = str(app.get("skill_md") or "").strip()
|
|
if not skill_md:
|
|
return None
|
|
url = _skill_content_url(skill_md, raw_base=str(app.get("_raw_base") or CLI_ANYTHING_RAW_BASE))
|
|
if not url:
|
|
return None
|
|
try:
|
|
response = httpx.get(url, timeout=15.0, follow_redirects=True)
|
|
response.raise_for_status()
|
|
text = response.text
|
|
except Exception:
|
|
return None
|
|
if "SKILL.md" not in url and not text.lstrip().startswith("---"):
|
|
return None
|
|
return text if len(text) < 250_000 else None
|
|
|
|
def _fallback_skill(self, app: dict[str, Any]) -> str:
|
|
name = str(app.get("name") or "unknown")
|
|
display = str(app.get("display_name") or name)
|
|
entry = str(app.get("entry_point") or f"cli-anything-{name}")
|
|
description = _catalog_description(app) or f"Use {display} from nanobot."
|
|
return f"""---
|
|
name: {_safe_skill_name(name)}
|
|
description: >-
|
|
{description}
|
|
---
|
|
|
|
# {display}
|
|
|
|
Use this skill when the user asks nanobot to operate {display} through its installed CLI app.
|
|
|
|
If the user attached `@{name}` in chat, treat that as the selected app for the current turn.
|
|
|
|
## Commands
|
|
|
|
```bash
|
|
{entry} --help
|
|
{entry} --json --help
|
|
```
|
|
|
|
Prefer machine-readable output when the CLI supports `--json`.
|
|
"""
|
|
|
|
def _with_nanobot_skill_note(self, content: str, app: dict[str, Any]) -> str:
|
|
marker = "<!-- nanobot-cli-app-note -->"
|
|
if marker in content:
|
|
return content
|
|
name = str(app.get("name") or "unknown")
|
|
note = f"""{marker}
|
|
## Nanobot execution
|
|
|
|
Use the `run_cli_app` tool with `name="{name}"` for command execution. Do not invoke this CLI through shell unless the user explicitly asks. Prefer this skill when Runtime Context mentions `@{name}` as a CLI App Attachment.
|
|
"""
|
|
lines = content.splitlines(keepends=True)
|
|
if lines and lines[0].strip() == "---":
|
|
for index, line in enumerate(lines[1:], start=1):
|
|
if line.strip() == "---":
|
|
return "".join(lines[: index + 1]) + "\n" + note + "\n" + "".join(lines[index + 1 :])
|
|
return note + "\n" + content
|
|
|
|
def install_skill(self, app: dict[str, Any]) -> Path:
|
|
path = self._skill_path(str(app["name"]))
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
content = self._fetch_skill_content(app) or self._fallback_skill(app)
|
|
content = self._with_nanobot_skill_note(content, app)
|
|
path.write_text(content, encoding="utf-8")
|
|
return path
|
|
|
|
def remove_skill(self, name: str) -> None:
|
|
skill_dir = self._skill_path(name).parent
|
|
if skill_dir.is_dir():
|
|
shutil.rmtree(skill_dir)
|
|
|
|
def _record_installed(self, app: dict[str, Any]) -> dict[str, Any]:
|
|
installed = self._load_installed()
|
|
entry = self._installed_entry(app)
|
|
installed[str(app["name"])] = entry
|
|
self._save_installed(installed)
|
|
self.install_skill(app)
|
|
return entry
|
|
|
|
def install(self, name: str) -> dict[str, Any]:
|
|
app = self.get_app(name)
|
|
if not self._install_supported(app):
|
|
raise CliAppError("this CLI app uses an unsupported install strategy")
|
|
strategy = self._strategy(app)
|
|
entry_point = str(app.get("entry_point") or "")
|
|
if entry_point and shutil.which(entry_point):
|
|
self._record_installed(app)
|
|
return self.payload() | {
|
|
"last_action": {
|
|
"ok": True,
|
|
"message": f"CLI for {app['display_name']} is already available.",
|
|
"installed": True,
|
|
"verification": ["entry_point_available", "state_recorded", "managed_paths_present"],
|
|
}
|
|
}
|
|
if strategy == "bundled":
|
|
detect_cmd = str(app.get("detect_cmd") or app.get("entry_point") or "")
|
|
if detect_cmd and _command_exists(detect_cmd):
|
|
self._record_installed(app)
|
|
return self.payload() | {
|
|
"last_action": {
|
|
"ok": True,
|
|
"message": f"CLI for {app['display_name']} is available.",
|
|
"installed": True,
|
|
"verification": ["entry_point_available", "state_recorded"],
|
|
}
|
|
}
|
|
note = app.get("install_notes") or f"{app['display_name']} is bundled with its parent app."
|
|
raise CliAppError(str(note))
|
|
argv = self._argv_for_action(app, "install")
|
|
assert argv is not None
|
|
result = self._run_argv(argv, timeout=self.runtime.install_timeout)
|
|
if strategy == "npm" and result.returncode != 0:
|
|
result = self._retry_stale_npm_install(app, argv, result)
|
|
if result.returncode != 0:
|
|
raise CliAppError(_truncate(result.stderr or result.stdout or "install failed"), status=500)
|
|
self._record_installed(app)
|
|
return self.payload() | {
|
|
"last_action": {
|
|
"ok": True,
|
|
"message": f"Installed CLI for {app['display_name']}.",
|
|
"installed": True,
|
|
"verification": ["package_manager_ok", "state_recorded", "managed_paths_present"],
|
|
}
|
|
}
|
|
|
|
def update(self, name: str) -> dict[str, Any]:
|
|
app = self.get_app(name, force_refresh=True)
|
|
if str(app["name"]) not in self._load_installed():
|
|
raise CliAppError("CLI app is not installed")
|
|
if self._strategy(app) == "bundled":
|
|
self._record_installed(app)
|
|
return self.payload() | {
|
|
"last_action": {
|
|
"ok": True,
|
|
"message": f"Checked {app['display_name']}.",
|
|
"installed": True,
|
|
"verification": ["state_recorded"],
|
|
}
|
|
}
|
|
argv = self._argv_for_action(app, "update")
|
|
assert argv is not None
|
|
result = self._run_argv(argv, timeout=self.runtime.install_timeout)
|
|
if result.returncode != 0:
|
|
raise CliAppError(_truncate(result.stderr or result.stdout or "update failed"), status=500)
|
|
self._record_installed(app)
|
|
return self.payload() | {
|
|
"last_action": {
|
|
"ok": True,
|
|
"message": f"Updated CLI for {app['display_name']}.",
|
|
"installed": True,
|
|
"verification": ["package_manager_ok", "state_recorded", "managed_paths_present"],
|
|
}
|
|
}
|
|
|
|
def uninstall(self, name: str) -> dict[str, Any]:
|
|
app = self.get_app(name)
|
|
installed = self._load_installed()
|
|
if str(app["name"]) not in installed:
|
|
raise CliAppError("CLI app is not installed")
|
|
raw_installed_entry = installed.get(str(app["name"]))
|
|
installed_entry = raw_installed_entry if isinstance(raw_installed_entry, dict) else {}
|
|
strategy = self._strategy(app)
|
|
entry_point = str(app.get("entry_point") or "").strip()
|
|
managed_entry_path = str(installed_entry.get("entry_point_path") or "").strip()
|
|
if strategy != "bundled":
|
|
argv = self._argv_for_action(app, "uninstall", installed_entry=installed_entry)
|
|
assert argv is not None
|
|
result = self._run_argv(argv, timeout=self.runtime.install_timeout)
|
|
if result.returncode != 0:
|
|
raise CliAppError(_truncate(result.stderr or result.stdout or "uninstall failed"), status=500)
|
|
still_managed = bool(managed_entry_path and Path(managed_entry_path).exists())
|
|
still_available = bool(entry_point and shutil.which(entry_point))
|
|
if still_managed or (not managed_entry_path and still_available):
|
|
reason = (
|
|
f"the recorded entry point at {managed_entry_path} still exists"
|
|
if still_managed
|
|
else f"{entry_point} is still available on PATH"
|
|
)
|
|
message = (
|
|
f"Uninstall for {app['display_name']} completed, but {reason}, "
|
|
"so nanobot kept it installed."
|
|
)
|
|
return self.payload() | {
|
|
"last_action": {
|
|
"ok": False,
|
|
"message": message,
|
|
"removed": False,
|
|
"still_available": True,
|
|
"verification_failed": ["entry_point_absent"],
|
|
}
|
|
}
|
|
else:
|
|
still_available = bool(entry_point and shutil.which(entry_point))
|
|
installed.pop(str(app["name"]), None)
|
|
self._save_installed(installed)
|
|
self.remove_skill(str(app["name"]))
|
|
if strategy == "bundled" and still_available:
|
|
message = (
|
|
f"Removed {app['display_name']} from nanobot. {entry_point} "
|
|
"is still available because it is managed outside nanobot."
|
|
)
|
|
elif still_available:
|
|
message = (
|
|
f"Uninstalled CLI for {app['display_name']}, but another {entry_point} "
|
|
"is still available on PATH."
|
|
)
|
|
else:
|
|
message = f"Uninstalled CLI for {app['display_name']}."
|
|
return self.payload() | {
|
|
"last_action": {
|
|
"ok": True,
|
|
"message": message,
|
|
"removed": True,
|
|
"still_available": still_available,
|
|
"verification": ["state_absent", "managed_paths_absent"]
|
|
if still_available
|
|
else ["entry_point_absent", "state_absent", "managed_paths_absent"],
|
|
}
|
|
}
|
|
|
|
def test(self, name: str) -> dict[str, Any]:
|
|
app = self.get_app(name)
|
|
entry = str(app.get("entry_point") or "")
|
|
resolved = shutil.which(entry)
|
|
if not entry or not resolved:
|
|
raise CliAppError(f"{entry or name} is not available on PATH")
|
|
result = self._run_argv([resolved, "--help"], timeout=min(self.runtime.run_timeout, 30))
|
|
ok = result.returncode == 0
|
|
output = _truncate((result.stdout or result.stderr or "").strip(), 3000)
|
|
return self.payload() | {
|
|
"last_action": {
|
|
"ok": ok,
|
|
"message": f"{entry} --help exited {result.returncode}",
|
|
"output": output,
|
|
}
|
|
}
|
|
|
|
def _resolve_cwd(
|
|
self,
|
|
working_dir: str | None,
|
|
*,
|
|
restrict_to_workspace: bool,
|
|
) -> Path:
|
|
cwd = Path(working_dir).expanduser() if working_dir else self.workspace
|
|
cwd = cwd.resolve(strict=False)
|
|
workspace = self.workspace.resolve(strict=False)
|
|
if restrict_to_workspace and not is_path_within(cwd, workspace):
|
|
raise CliAppError("working_dir is outside the configured workspace")
|
|
return cwd
|
|
|
|
def _iter_artifact_candidates(self, cwd: Path) -> list[Path]:
|
|
if not cwd.is_dir():
|
|
return []
|
|
out: list[Path] = []
|
|
stack = [cwd]
|
|
scanned = 0
|
|
while stack and scanned < _MAX_ARTIFACT_SCAN_PATHS:
|
|
directory = stack.pop()
|
|
try:
|
|
entries = sorted(directory.iterdir(), key=lambda path: path.name.lower())
|
|
except OSError:
|
|
continue
|
|
for path in entries:
|
|
if scanned >= _MAX_ARTIFACT_SCAN_PATHS:
|
|
break
|
|
scanned += 1
|
|
try:
|
|
if path.is_dir() and not path.is_symlink():
|
|
if path.name not in _ARTIFACT_IGNORE_DIRS:
|
|
stack.append(path)
|
|
continue
|
|
if path.is_file() and path.suffix.lower() in _ARTIFACT_EXTENSIONS:
|
|
out.append(path.resolve(strict=False))
|
|
except OSError:
|
|
continue
|
|
return out
|
|
|
|
def _artifact_snapshot(self, cwd: Path) -> dict[Path, tuple[int, int]]:
|
|
snapshot: dict[Path, tuple[int, int]] = {}
|
|
for path in self._iter_artifact_candidates(cwd):
|
|
try:
|
|
stat = path.stat()
|
|
except OSError:
|
|
continue
|
|
snapshot[path] = (stat.st_mtime_ns, stat.st_size)
|
|
return snapshot
|
|
|
|
def _changed_artifacts(
|
|
self,
|
|
cwd: Path,
|
|
before: dict[Path, tuple[int, int]],
|
|
) -> list[Path]:
|
|
changed: list[tuple[int, Path]] = []
|
|
for path, stamp in self._artifact_snapshot(cwd).items():
|
|
if before.get(path) == stamp:
|
|
continue
|
|
changed.append((stamp[0], path))
|
|
changed.sort(key=lambda item: (item[0], item[1].name.lower()))
|
|
return [path for _, path in changed[-_MAX_ARTIFACT_REPORT:]]
|
|
|
|
def _format_artifact_path(self, cwd: Path, path: Path) -> str:
|
|
try:
|
|
return path.relative_to(cwd).as_posix()
|
|
except ValueError:
|
|
return path.name
|
|
|
|
@staticmethod
|
|
def _format_artifact_size(path: Path) -> str:
|
|
try:
|
|
size = path.stat().st_size
|
|
except OSError:
|
|
return "unknown size"
|
|
if size < 1024:
|
|
return f"{size} B"
|
|
if size < 1024 * 1024:
|
|
return f"{size / 1024:.1f} KB"
|
|
return f"{size / (1024 * 1024):.1f} MB"
|
|
|
|
def _format_artifact_lines(self, cwd: Path, paths: list[Path]) -> list[str]:
|
|
lines: list[str] = []
|
|
for path in paths:
|
|
rel = self._format_artifact_path(cwd, path)
|
|
ext = path.suffix.lower()
|
|
kind = (
|
|
"previewable image"
|
|
if ext in _INLINE_ARTIFACT_EXTENSIONS
|
|
else ext.lstrip(".") or "file"
|
|
)
|
|
lines.append(f"- {rel} ({kind}, {self._format_artifact_size(path)})")
|
|
return lines
|
|
|
|
def run(
|
|
self,
|
|
name: str,
|
|
args: list[str] | None = None,
|
|
*,
|
|
json_output: bool = False,
|
|
working_dir: str | None = None,
|
|
timeout: int | None = None,
|
|
restrict_to_workspace: bool = False,
|
|
) -> str:
|
|
app = self.get_app(name)
|
|
installed = self._load_installed()
|
|
if str(app["name"]) not in installed:
|
|
raise CliAppError(f"CLI app '{name}' is not installed")
|
|
cwd = self._resolve_cwd(working_dir, restrict_to_workspace=restrict_to_workspace)
|
|
entry = str(installed[str(app["name"])].get("entry_point") or app.get("entry_point") or "")
|
|
resolved = shutil.which(entry)
|
|
if not entry or not resolved:
|
|
raise CliAppError(f"{entry or name} is not available on PATH")
|
|
clean_args = [str(arg) for arg in (args or [])]
|
|
if json_output and "--json" not in clean_args:
|
|
clean_args = ["--json", *clean_args]
|
|
effective_timeout = max(1, min(timeout or self.runtime.run_timeout, 600))
|
|
artifact_snapshot = self._artifact_snapshot(cwd)
|
|
try:
|
|
result = subprocess.run(
|
|
[resolved, *clean_args],
|
|
cwd=str(cwd),
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=effective_timeout,
|
|
env=os.environ.copy(),
|
|
)
|
|
except subprocess.TimeoutExpired:
|
|
return f"CLI app '{name}' timed out after {effective_timeout}s"
|
|
output = [
|
|
f"CLI app '{name}' exited {result.returncode}.",
|
|
f"Command: {entry} {' '.join(shlex.quote(arg) for arg in clean_args)}".rstrip(),
|
|
]
|
|
if result.stdout:
|
|
output.append("\nSTDOUT:\n" + result.stdout.rstrip())
|
|
if result.stderr:
|
|
output.append("\nSTDERR:\n" + result.stderr.rstrip())
|
|
artifacts = self._changed_artifacts(cwd, artifact_snapshot)
|
|
if artifacts:
|
|
output.append(
|
|
"\nArtifacts created or updated:\n"
|
|
+ "\n".join(self._format_artifact_lines(cwd, artifacts))
|
|
)
|
|
if any(path.suffix.lower() in _INLINE_ARTIFACT_EXTENSIONS for path in artifacts):
|
|
output.append(
|
|
"\nTo show a preview in WebUI, reference a raster artifact with Markdown "
|
|
"using its workspace-relative path, for example ``."
|
|
)
|
|
return _truncate("\n".join(output))
|