Terminal / tools /registry_dev.py
Baida—-
deploy: full backend sync for Node A [Kernel Restore]
8b87770 verified
Raw
History Blame Contribute Delete
23.5 kB
"""registry_dev.py — Tool sviluppo: git, npm, pip, lint, scaffold tools leggeri.
Estratto da registry.py per ridurre il file principale.
Funzioni esportate:
_directory_tree, _file_search, _git_status, _git_clone, _git_diff,
_recall, _list_files, _diff_text, _validate_json, _lint_code_tool,
_git_push, _git_sync_vfs, _git_commit, _npm_install, _npm_run,
_pip_install, _type_check
"""
from __future__ import annotations
import httpx
import asyncio
import subprocess
import tempfile
import os
import sys
import logging
_logger = logging.getLogger("tools.registry")
# ─── S763: 10 tool mancanti (S760 dichiarati mai implementati) ──────────────
# Presenti in planner.py PLANNER_SYSTEM, agent.py _STEP_VISIBILITY/_NARR_QUICK,
# _TOOL_NEEDED_RE ma senza _fn in TOOL_REGISTRY -> KeyError silenzioso al runtime.
async def _directory_tree(path: str = ".", max_depth: int = 3, show_hidden: bool = False) -> dict:
"""S763: Albero filesystem con os.walk."""
import os as _os
try:
base = _os.path.abspath(path)
if not _os.path.isdir(base):
return {"error": f"Percorso non trovato: {path}"}
_IGNORE = {".git", "__pycache__", "node_modules", ".venv", "venv", "dist", "build", ".next", ".cache"}
lines: list = [f"{base}/"]
count = 0
for root, dirs, files in _os.walk(base):
depth = root.replace(base, "").count(_os.sep)
if depth >= max_depth:
dirs.clear()
continue
dirs[:] = sorted(d for d in dirs
if (show_hidden or not d.startswith(".")) and d not in _IGNORE)
ind = " " * (depth + 1)
for d in dirs:
lines.append(f"{ind}+-- {d}/")
for fname in sorted(files):
if not show_hidden and fname.startswith("."):
continue
if count >= 200:
lines.append(f"{ind}... (troncato)")
break
lines.append(f"{ind} {fname}")
count += 1
return {"ok": True, "path": base, "tree": "\n".join(lines), "count": count}
except Exception as e:
return {"error": str(e)[:300]}
async def _file_search(pattern: str, path: str = ".", file_glob: str = "*") -> dict:
"""S763: grep -rn con fallback Python os.walk+read."""
import os as _os
try:
proc = await asyncio.create_subprocess_exec(
"grep", "-rn", "--include", file_glob, "--color=never", "-m", "5",
pattern, _os.path.abspath(path),
stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
)
out, _ = await asyncio.wait_for(proc.communicate(), timeout=15)
lines = [l for l in out.decode("utf-8", errors="replace").splitlines() if l.strip()]
if proc.returncode == 1 and not lines:
return {"ok": True, "pattern": pattern, "matches": [], "count": 0, "note": "Nessun risultato"}
return {"ok": True, "pattern": pattern, "path": path, "matches": lines[:50], "count": len(lines)}
except (FileNotFoundError, asyncio.TimeoutError):
import re as _re2
matches: list = []
try:
_pat = _re2.compile(pattern, _re2.IGNORECASE)
for root, _, files in _os.walk(path):
for fname in files:
fpath = _os.path.join(root, fname)
try:
with open(fpath, "r", encoding="utf-8", errors="replace") as fh:
for i, line in enumerate(fh, 1):
if _pat.search(line):
matches.append(f"{fpath}:{i}: {line.rstrip()[:200]}")
if len(matches) >= 50:
break
except Exception:
continue
if len(matches) >= 50:
break
except Exception as ex:
return {"error": str(ex)[:300]}
return {"ok": True, "pattern": pattern, "matches": matches, "count": len(matches)}
except Exception as e:
return {"error": str(e)[:300]}
async def _git_status(cwd: str = ".") -> dict:
"""S763: branch + status --short + log -5. Read-only."""
try:
async def _rg(*args: str) -> str:
p = await asyncio.create_subprocess_exec(
"git", *args, cwd=cwd,
stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
)
out, _ = await asyncio.wait_for(p.communicate(), timeout=10)
return out.decode("utf-8", errors="replace").strip()
return {
"ok": True,
"branch": await _rg("rev-parse", "--abbrev-ref", "HEAD"),
"status": await _rg("status", "--short") or "(working tree clean)",
"log": await _rg("log", "--oneline", "-5"),
}
except Exception as e:
return {"error": str(e)[:300]}
async def _git_clone(url: str, directory: str = "", depth: int = 0) -> dict:
"""S763: git clone <url> [dir]. Timeout 120s. Risk medium."""
try:
cmd = ["git", "clone"]
if depth > 0:
cmd += ["--depth", str(depth)]
cmd.append(url)
if directory:
cmd.append(directory)
proc = await asyncio.create_subprocess_exec(
*cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
)
out, err = await asyncio.wait_for(proc.communicate(), timeout=120)
combined = (out + err).decode("utf-8", errors="replace")[:1000]
if proc.returncode == 0:
target = directory or url.rstrip("/").split("/")[-1].removesuffix(".git")
return {"ok": True, "directory": target, "output": combined}
return {"ok": False, "error": combined}
except asyncio.TimeoutError:
return {"ok": False, "error": "timeout 120s"}
except Exception as e:
return {"ok": False, "error": str(e)[:300]}
async def _git_diff(cwd: str = ".", staged: bool = False) -> dict:
"""S763: git diff [--cached]. Stat + diff max 3000 chars. Read-only."""
try:
extra = ["--cached"] if staged else []
async def _rg(*a: str) -> str:
p = await asyncio.create_subprocess_exec(
"git", *a, cwd=cwd,
stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
)
out, _ = await asyncio.wait_for(p.communicate(), timeout=15)
return out.decode("utf-8", errors="replace")
stat = await _rg("diff", *extra, "--stat", "--no-color")
diff = await _rg("diff", *extra, "--no-color")
return {"staged": staged, "stat": stat[:1000], "diff": diff[:3000] or "(nessuna modifica)"}
except Exception as e:
return {"error": str(e)[:300]}
async def _recall(query: str, limit: int = 5) -> dict:
"""S-GAP13: cerca in agentMemory (chiave/valore in-process). Fallback su entries recenti."""
try:
from api.state import _get_mem_manager_async as _gmm
mem = await _gmm()
if mem is None:
return {"results": [], "note": "MemoryManager non disponibile"}
results = []
try:
raw = await mem.search(query, limit=limit)
results = [{"key": r.get("key",""), "value": r.get("value",""), "score": r.get("score",0)} for r in (raw or [])]
except Exception:
try:
raw = await mem.list(limit=limit * 2)
q_lower = query.lower()
for r in (raw or []):
k = str(r.get("key","")).lower()
v = str(r.get("value","")).lower()
if q_lower in k or q_lower in v:
results.append({"key": r.get("key",""), "value": r.get("value","")})
if len(results) >= limit:
break
except Exception as _exc:
_logger.debug("[registry] silenced %s", type(_exc).__name__) # noqa: BLE001
return {"results": results, "count": len(results), "query": query}
except Exception as e:
return {"results": [], "error": str(e)[:200]}
async def _list_files(path: str = ".", recursive: bool = False, max_items: int = 100) -> dict:
"""S-GAP13: elenca file nella directory. os.listdir/os.walk. Max 100 items."""
import os as _os
try:
path = _os.path.abspath(path)
if not _os.path.exists(path):
return {"ok": False, "error": f"Path non trovato: {path}"}
if not _os.path.isdir(path):
return {"ok": False, "error": f"Non e una directory: {path}"}
items = []
if recursive:
for root, dirs, files in _os.walk(path):
dirs[:] = [d for d in dirs if not d.startswith('.') and d not in ('node_modules','__pycache__','.git')]
rel_root = _os.path.relpath(root, path)
for f in files:
rel = _os.path.join(rel_root, f) if rel_root != '.' else f
items.append(rel)
if len(items) >= max_items:
break
if len(items) >= max_items:
break
else:
for entry in _os.scandir(path):
items.append(entry.name + ('/' if entry.is_dir() else ''))
if len(items) >= max_items:
break
return {"ok": True, "path": path, "items": items, "count": len(items), "truncated": len(items) >= max_items}
except Exception as e:
return {"ok": False, "error": str(e)[:300]}
def _diff_text(text_a: str, text_b: str, context_lines: int = 3) -> dict:
"""S-GAP13: confronta due testi con difflib.unified_diff. Restituisce patch testo."""
import difflib
try:
lines_a = text_a.splitlines(keepends=True)
lines_b = text_b.splitlines(keepends=True)
diff = list(difflib.unified_diff(lines_a, lines_b, fromfile="a", tofile="b", n=context_lines))
patch = "".join(diff)
added = sum(1 for l in diff if l.startswith('+') and not l.startswith('+++'))
removed = sum(1 for l in diff if l.startswith('-') and not l.startswith('---'))
return {"patch": patch[:4000], "added": added, "removed": removed, "identical": len(diff) == 0}
except Exception as e:
return {"patch": "", "error": str(e)[:200]}
def _validate_json(json_str: str, schema: dict | None = None) -> dict:
"""S-GAP13: valida JSON (json.loads). Con schema dict usa jsonschema se disponibile."""
import json
try:
parsed = json.loads(json_str)
result: dict = {"valid": True, "type": type(parsed).__name__}
if schema:
try:
import jsonschema
jsonschema.validate(parsed, schema)
result["schema_valid"] = True
except ImportError:
result["schema_note"] = "jsonschema non installato — validazione struttura skippata"
except Exception as ve:
result["valid"] = False
result["schema_error"] = str(ve)[:400]
return result
except json.JSONDecodeError as e:
return {"valid": False, "error": f"JSON non valido: {e.msg} (riga {e.lineno}, col {e.colno})"}
except Exception as e:
return {"valid": False, "error": str(e)[:200]}
async def _lint_code_tool(content: str, language: str = "auto", path: str = "") -> dict:
"""S-GAP13: analisi statica codice. Wrapper di api.linter.lint_code. Auto-detect da estensione."""
try:
if language == "auto" and path:
ext = path.rsplit(".", 1)[-1].lower() if "." in path else ""
language = {"py": "python", "js": "javascript", "jsx": "javascript",
"ts": "typescript", "tsx": "typescript", "json": "json"}.get(ext, "python")
from api.linter import lint_code as _lc
return await _lc(content=content, language=language, path=path)
except Exception as e:
return {"ok": False, "errors": [], "warnings": [], "error": str(e)[:200]}
async def _git_push(remote: str = "origin", branch: str = "", cwd: str = ".") -> dict:
"""S-GAP12: git push <remote> [<branch>]. Timeout 70s. Risk high."""
import asyncio as _aio
try:
cmd = ["git", "push", remote]
if branch:
cmd.append(branch)
proc = await _aio.create_subprocess_exec(
*cmd, cwd=cwd,
stdout=_aio.subprocess.PIPE, stderr=_aio.subprocess.PIPE,
)
out, err = await _aio.wait_for(proc.communicate(), timeout=65)
combined = (out + err).decode("utf-8", errors="replace")[:800]
return {"ok": proc.returncode == 0, "output": combined, "code": proc.returncode}
except _aio.TimeoutError:
return {"ok": False, "error": "git push timeout (65s)"}
except Exception as e:
return {"ok": False, "error": str(e)[:300]}
async def _git_sync_vfs(
files: dict,
branch: str = "agent-state",
message: str = "chore(vfs): auto-sync session",
repo: str = "",
) -> dict:
"""
RF-1: git_sync_vfs — Commit atomico VFS→GitHub via Git Data API.
Flusso: GET HEAD → POST blob×N → POST tree → POST commit → PATCH/POST ref.
Branch inesistente: creato automaticamente da HEAD di main.
Repo: param repo oppure env GITHUB_REPO.
Fail-safe: ritorna error se GITHUB_TOKEN mancante.
"""
import base64
import httpx as _httpx
import os as _os
gh_token = _os.environ.get("GITHUB_TOKEN", "")
if not gh_token:
return {"success": False, "error": "GITHUB_TOKEN non configurato"}
if not files:
return {"success": False, "error": "Nessun file da sincronizzare"}
gh_repo = repo or _os.environ.get("GITHUB_REPO", "")
if not gh_repo:
return {"success": False, "error": "Specifica repo='owner/repo' oppure imposta GITHUB_REPO env"}
_hdrs = {
"Authorization": f"token {gh_token}",
"Accept": "application/vnd.github.v3+json",
"Content-Type": "application/json",
}
base_url = f"https://api.github.com/repos/{gh_repo}"
try:
async with _httpx.AsyncClient(timeout=30.0, headers=_hdrs) as _c:
# 1. Leggi HEAD branch target (o fallback a main)
base_sha: str | None = None
branch_exists = False
_ref_r = await _c.get(f"{base_url}/git/ref/heads/{branch}")
if _ref_r.status_code == 200:
base_sha = _ref_r.json()["object"]["sha"]
branch_exists = True
else:
_main_r = await _c.get(f"{base_url}/git/ref/heads/main")
if _main_r.status_code == 200:
base_sha = _main_r.json()["object"]["sha"]
else:
return {"success": False, "error": f"Impossibile leggere HEAD: {_main_r.status_code}"}
# 2. Crea blob per ogni file
tree_items: list[dict] = []
for fpath, content in files.items():
if not isinstance(content, str):
content = str(content)
encoded = base64.b64encode(content.encode("utf-8", errors="replace")).decode()
_blob_r = await _c.post(f"{base_url}/git/blobs", json={"content": encoded, "encoding": "base64"})
if _blob_r.status_code not in (200, 201):
return {"success": False, "error": f"Blob fail [{fpath}]: {_blob_r.status_code}"}
tree_items.append({"path": fpath, "mode": "100644", "type": "blob", "sha": _blob_r.json()["sha"]})
# 3. Crea tree
_tree_payload: dict = {"tree": tree_items}
if base_sha:
_tree_payload["base_tree"] = base_sha
_tree_r = await _c.post(f"{base_url}/git/trees", json=_tree_payload)
if _tree_r.status_code not in (200, 201):
return {"success": False, "error": f"Tree fail: {_tree_r.status_code}"}
tree_sha = _tree_r.json()["sha"]
# 4. Crea commit
_commit_payload: dict = {"message": message, "tree": tree_sha}
if base_sha:
_commit_payload["parents"] = [base_sha]
_commit_r = await _c.post(f"{base_url}/git/commits", json=_commit_payload)
if _commit_r.status_code not in (200, 201):
return {"success": False, "error": f"Commit fail: {_commit_r.status_code}"}
commit_sha = _commit_r.json()["sha"]
# 5. PATCH ref (o POST se branch nuovo)
if branch_exists:
_ref_upd = await _c.patch(f"{base_url}/git/refs/heads/{branch}", json={"sha": commit_sha})
else:
_ref_upd = await _c.post(f"{base_url}/git/refs", json={"ref": f"refs/heads/{branch}", "sha": commit_sha})
if _ref_upd.status_code not in (200, 201):
return {"success": False, "error": f"Ref update fail: {_ref_upd.status_code}{_ref_upd.text[:200]}"}
return {
"success": True,
"commit_sha": commit_sha,
"branch": branch,
"files_synced": len(tree_items),
"repo": gh_repo,
"url": f"https://github.com/{gh_repo}/tree/{branch}",
}
except Exception as _e:
return {"success": False, "error": f"git_sync_vfs errore: {str(_e)[:300]}"}
async def _git_commit(message: str, cwd: str = ".", push: bool = False, add_all: bool = True) -> dict:
"""S763: git add -A + commit -m. push=True per git push. Risk medium."""
try:
async def _run(cmd: list) -> tuple:
p = await asyncio.create_subprocess_exec(
*cmd, cwd=cwd,
stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
)
out, err = await asyncio.wait_for(p.communicate(), timeout=30)
return p.returncode, (out + err).decode("utf-8", errors="replace")[:500]
if add_all:
rc, out = await _run(["git", "add", "-A"])
if rc != 0:
return {"ok": False, "step": "git add", "error": out}
rc, out = await _run(["git", "commit", "-m", message])
if rc != 0:
return {"ok": False, "step": "git commit", "error": out}
result: dict = {"ok": True, "commit_output": out}
if push:
rc_p, out_p = await _run(["git", "push"])
result["push_ok"] = rc_p == 0
result["push_output"] = out_p
return result
except Exception as e:
return {"ok": False, "error": str(e)[:300]}
async def _npm_install(cwd: str = ".", manager: str = "auto", args: str = "") -> dict:
"""S763: npm/pnpm/yarn install. Auto-detecta da lockfile. Timeout 120s."""
import os as _os
try:
if manager == "auto":
manager = ("pnpm" if _os.path.exists(_os.path.join(cwd, "pnpm-lock.yaml"))
else "yarn" if _os.path.exists(_os.path.join(cwd, "yarn.lock"))
else "npm")
cmd = [manager, "install"] + ([args] if args else [])
proc = await asyncio.create_subprocess_exec(
*cmd, cwd=cwd,
stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
)
out, err = await asyncio.wait_for(proc.communicate(), timeout=120)
combined = (out + err).decode("utf-8", errors="replace")[:2000]
return {"ok": proc.returncode == 0, "manager": manager, "output": combined, "code": proc.returncode}
except asyncio.TimeoutError:
return {"ok": False, "error": f"{manager} install timeout (120s)"}
except Exception as e:
return {"ok": False, "error": str(e)[:300]}
async def _npm_run(script: str, cwd: str = ".", manager: str = "auto") -> dict:
"""S763: npm/pnpm/yarn run <script>. Auto-detecta manager. Timeout 120s."""
import os as _os
try:
if manager == "auto":
manager = ("pnpm" if _os.path.exists(_os.path.join(cwd, "pnpm-lock.yaml"))
else "yarn" if _os.path.exists(_os.path.join(cwd, "yarn.lock"))
else "npm")
proc = await asyncio.create_subprocess_exec(
manager, "run", script, cwd=cwd,
stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
)
out, err = await asyncio.wait_for(proc.communicate(), timeout=120)
combined = (out + err).decode("utf-8", errors="replace")[:3000]
return {"ok": proc.returncode == 0, "script": script, "manager": manager,
"output": combined, "code": proc.returncode}
except asyncio.TimeoutError:
return {"ok": False, "error": f"{manager} run {script} timeout (120s)"}
except Exception as e:
return {"ok": False, "error": str(e)[:300]}
async def _pip_install(packages: str, upgrade: bool = False, user: bool = False) -> dict:
"""S763: pip install. packages: stringa spazio/virgola separata. Timeout 120s."""
try:
cmd = [sys.executable, "-m", "pip", "install", "-q"]
if upgrade:
cmd.append("--upgrade")
if user:
cmd.append("--user")
pkgs = [p.strip() for p in packages.replace(",", " ").split() if p.strip()]
cmd.extend(pkgs)
proc = await asyncio.create_subprocess_exec(
*cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
)
out, err = await asyncio.wait_for(proc.communicate(), timeout=120)
combined = (out + err).decode("utf-8", errors="replace")[:2000]
return {"ok": proc.returncode == 0, "packages": pkgs, "output": combined, "code": proc.returncode}
except asyncio.TimeoutError:
return {"ok": False, "error": "pip install timeout (120s)"}
except Exception as e:
return {"ok": False, "error": str(e)[:300]}
async def _type_check(path: str = ".", checker: str = "auto", strict: bool = False) -> dict:
"""S763: tsc --noEmit o mypy. Auto-detecta da tsconfig.json/pyproject.toml. Timeout 60s."""
import os as _os
try:
if checker == "auto":
checker = ("tsc" if _os.path.exists(_os.path.join(path, "tsconfig.json")) else
"mypy" if (_os.path.exists(_os.path.join(path, "setup.py")) or
_os.path.exists(_os.path.join(path, "pyproject.toml"))) else "tsc")
if checker == "tsc":
cmd = ["npx", "tsc", "--noEmit"] + (["--strict"] if strict else [])
elif checker == "mypy":
cmd = ["mypy", path, "--ignore-missing-imports"] + (["--strict"] if strict else [])
else:
return {"error": f"Checker non supportato: {checker}"}
proc = await asyncio.create_subprocess_exec(
*cmd, cwd=path,
stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
)
out, err = await asyncio.wait_for(proc.communicate(), timeout=60)
combined = (out + err).decode("utf-8", errors="replace")[:3000]
ok = proc.returncode == 0
return {"ok": ok, "checker": checker, "output": combined[:1000],
"errors": combined if not ok else "", "code": proc.returncode}
except asyncio.TimeoutError:
return {"ok": False, "error": f"{checker} timeout (60s)"}
except Exception as e:
return {"ok": False, "error": str(e)[:300]}