"""
registry.py — Tool Registry
Ogni tool ha: goal, required_inputs, risk_level, fallbacks, _fn.
"""
import httpx
import asyncio
import subprocess
import tempfile
import os
import sys
import ipaddress
import socket
from contextvars import ContextVar
import logging
_logger = logging.getLogger("tools.registry")
# ─── SEC-FS-JAIL: confinamento path per read_file/write_file/apply_patch ──
# GAP: pathlib.Path(path) usato as-is permetteva accesso a qualunque file
# leggibile/scrivibile dal processo (env mount, secrets, codice di altri
# servizi). Confina tutte le operazioni FS del tool registry sotto una root
# esplicita, configurabile via env (default: cwd del processo backend).
_FS_JAIL_ROOT = _pathlib_module = None
def _fs_jail_root():
import pathlib as _pl
global _FS_JAIL_ROOT
if _FS_JAIL_ROOT is None:
_root = os.getenv("FS_TOOL_ROOT", os.getcwd())
_FS_JAIL_ROOT = _pl.Path(_root).resolve()
return _FS_JAIL_ROOT
def _safe_fs_path(path: str):
"""Risolve 'path' e verifica che sia contenuto in _fs_jail_root().
Ritorna (resolved_path, None) se ok, oppure (None, error_msg) se rifiutato.
Blocca path assoluti fuori jail e traversal (../) che escono dalla root."""
import pathlib as _pl
try:
_root = _fs_jail_root()
_candidate = (_root / path) if not _pl.Path(path).is_absolute() else _pl.Path(path)
_resolved = _candidate.resolve()
_resolved.relative_to(_root)
return _resolved, None
except ValueError:
return None, f"Accesso negato: path fuori dalla sandbox consentita ({path})"
except Exception as exc:
return None, f"Path non valido: {exc}"
# ─── SEC-SSRF: validazione IP per call_api ─────────────────────────────
# GAP: _call_api chiamava qualsiasi URL senza risolvere l'hostname o
# bloccare IP privati/loopback/link-local, permettendo di colpire endpoint
# interni del backend stesso (es. http://localhost:PORT/api/vault/...).
_SSRF_ALLOWED_SCHEMES = frozenset({"http", "https"})
def _ssrf_check(url: str):
"""Ritorna None se l'URL è sicuro da chiamare, stringa di errore se bloccato."""
from urllib.parse import urlparse
try:
_parsed = urlparse(url)
except Exception as exc:
return f"URL non valido: {exc}"
if _parsed.scheme.lower() not in _SSRF_ALLOWED_SCHEMES:
return f"Scheme non permesso: '{_parsed.scheme}' (solo http/https)"
_host = _parsed.hostname or ""
if not _host:
return "URL senza hostname valido"
if _host.lower() in ("localhost", "0.0.0.0", "::1"):
return "SSRF bloccato: hostname locale non permesso"
try:
_ip = ipaddress.ip_address(socket.gethostbyname(_host))
if _ip.is_private or _ip.is_loopback or _ip.is_link_local or _ip.is_reserved or _ip.is_multicast:
return "SSRF bloccato: IP privato/loopback/riservato non permesso"
except socket.gaierror:
return f"Hostname non risolvibile: {_host}"
except Exception:
pass # errori di parsing IP imprevisti: lascia proseguire, httpx gestirà l'errore di rete
return None
# ─── S749-GAP1/D: backend-exec microservice client + ContextVar session ──
_EXEC_ENGINE_URL: str = os.getenv("EXEC_ENGINE_URL", "").rstrip("/")
_EXEC_TOKEN_HDR: str = os.getenv("EXEC_TOKEN", "")
# S749-D: ContextVar per session_id isolata per task — thread-safe, asyncio-safe.
# unified_loop.run() imposta questo valore al boot del task e lo resetta nel finally.
# Default "agent_default" se chiamato fuori dal loop (es. test unitari).
_agent_session_id_var: ContextVar[str] = ContextVar("agent_session_id", default="agent_default")
# Shell validation -> tools._shell_safety
from tools._shell_safety import (
validate_shell_command as _validate_shell_cmd,
run_shell_safe_sync as _run_shell_sync,
safe_shell_env as _safe_env,
)
import re as _re_registry
_REGISTRY_SAFE_CMD_RE = _re_registry.compile(
r'^(ls|cat|echo|pwd|whoami|date|uname|python3?|node|npm|pip[3]?|pnpm|'
r'grep|find|head|tail|wc|sort|uniq|diff|mkdir|touch|cp|mv|chmod|'
r'git\s+(status|log|diff|show)|curl\s+https?://|wget\s+https?://)(\s|$)',
_re_registry.IGNORECASE,
)
_REGISTRY_SHELL_METACHAR_RE = _re_registry.compile(r'[;&|`<>\n\r]|\$[\(\{]')
# P17-B3: persistent HTTP client per backend-exec — evita nuovo TCP/TLS handshake ad ogni call.
# Risparmio stimato: ~100-300ms per chiamata run_python/execute_shell su Railway.
# Reset automatico su connessione persa (is_closed check in _get_exec_client).
_exec_http_client: "httpx.AsyncClient | None" = None
def _get_exec_client() -> "httpx.AsyncClient":
"""Lazy singleton httpx.AsyncClient con keepalive per backend-exec."""
global _exec_http_client
if _exec_http_client is None or _exec_http_client.is_closed:
_exec_http_client = httpx.AsyncClient(
timeout=httpx.Timeout(20.0, connect=5.0),
limits=httpx.Limits(
max_connections=5,
max_keepalive_connections=3,
keepalive_expiry=60.0,
),
)
return _exec_http_client
async def _call_exec_engine(payload: dict, endpoint: str = "/api/exec") -> "dict | None":
"""Chiama backend-exec microservice (Railway) con fallback silente.
Ritorna il JSON della risposta se status 200, altrimenti None.
Se EXEC_ENGINE_URL non configurato ritorna None immediatamente.
Timeout conservativo 20s — mai blocca il loop principale.
S750-GAP-A: 1 retry con back-off 2s — gestisce cold start Railway (~1-2s attesa).
"""
if not _EXEC_ENGINE_URL:
return None
headers: dict = {}
if _EXEC_TOKEN_HDR:
headers["X-Exec-Token"] = _EXEC_TOKEN_HDR
for _attempt in range(2): # S750-GAP-A: tentativi 0 e 1
try:
_c = _get_exec_client() # P17-B3: persistent client — reusa connessione TCP
_r = await _c.post(
f"{_EXEC_ENGINE_URL}{endpoint}",
json=payload,
headers=headers,
)
if _r.status_code == 200:
return _r.json()
except (httpx.TransportError, httpx.ConnectError, ConnectionError, OSError) as _exc:
_logger.debug("[registry] P17-B3 transport error, resetting client: %s", type(_exc).__name__) # noqa: BLE001
global _exec_http_client
_exec_http_client = None # P17-B3: reset SOLO su errori trasporto/rete
except Exception as _exc:
_logger.debug("[registry] P17-B3 silenced (no reset): %s", type(_exc).__name__) # noqa: BLE001
if _attempt == 0:
await asyncio.sleep(2.0) # back-off 2s prima del retry
return None
# ─── Tool functions ────────────────────────────────────────
async def _web_search(query: str, max_results: int = 5) -> dict:
"""
S357: Parallelismo completo — tutti e 4 i provider lanciati con asyncio.gather.
Worst case: 10s (timeout singolo provider) invece di 40s+ (4 × 10s sequenziali).
Merge con deduplicazione per URL, priorità Brave > Tavily > Wikipedia > DDG.
"""
import re as _re, html as _html, urllib.parse as _urlparse, urllib.request as _urlreq
_headers = {"User-Agent": "Mozilla/5.0 (compatible; AgentBot/1.0)"}
brave_key = os.environ.get("BRAVE_SEARCH_API_KEY", "")
tavily_key = os.environ.get("TAVILY_API_KEY", "")
async def _brave() -> list:
if not brave_key:
return []
try:
async with httpx.AsyncClient(timeout=10) as c:
r = await c.get(
"https://api.search.brave.com/res/v1/web/search",
params={"q": query, "count": max_results, "text_decorations": "0"},
headers={**_headers, "Accept": "application/json", "X-Subscription-Token": brave_key},
)
if r.status_code == 200:
return [
# S602: snippet 300→500 — Brave description[:300] parity con Tavily
{"title": it["title"], "snippet": (it.get("description") or "")[:500],
"url": it["url"], "source": "Brave"}
for it in r.json().get("web", {}).get("results", [])[:max_results]
if it.get("title") and it.get("url")
]
except Exception as _exc:
_logger.debug("[registry] silenced %s", type(_exc).__name__) # noqa: BLE001
return []
async def _tavily() -> list:
if not tavily_key:
return []
try:
async with httpx.AsyncClient(timeout=10) as c:
r = await c.post(
"https://api.tavily.com/search",
json={"api_key": tavily_key, "query": query,
"max_results": max_results, "include_answer": False},
headers={**_headers, "Content-Type": "application/json"},
)
if r.status_code == 200:
return [
# S600: snippet 300→500 — Tavily restituisce snippet più ricchi
{"title": it["title"], "snippet": (it.get("content") or "")[:500],
"url": it["url"], "source": "Tavily"}
for it in r.json().get("results", [])[:max_results]
if it.get("title") and it.get("url")
]
except Exception as _exc:
_logger.debug("[registry] silenced %s", type(_exc).__name__) # noqa: BLE001
return []
async def _wikipedia() -> list:
try:
wiki_qs = _urlparse.urlencode({
"action": "query", "list": "search", "srsearch": query,
"format": "json", "utf8": "1", "srlimit": min(max_results, 4), "srnamespace": "0",
})
wiki_req = _urlreq.Request(
f"https://en.wikipedia.org/w/api.php?{wiki_qs}",
headers={"User-Agent": "agente-ai/3.2"},
)
wiki_data = await asyncio.to_thread(
lambda: __import__("json").loads(_urlreq.urlopen(wiki_req, timeout=7).read())
)
return [
{
"title": item.get("title", ""),
"url": "https://en.wikipedia.org/wiki/" + item.get("title", "").replace(" ", "_"),
# S600: snippet 300→500 — Wikipedia snippet può essere più lungo
"snippet": _html.unescape(_re.sub(r"<[^>]+>", "", item.get("snippet", "")))[:500],
"source": "Wikipedia",
}
for item in wiki_data.get("query", {}).get("search", [])[:max_results]
]
except Exception:
return []
async def _ddg() -> list:
try:
ddg_qs = _urlparse.urlencode({"q": query, "kl": "it-it"})
ddg_req = _urlreq.Request(
f"https://html.duckduckgo.com/html/?{ddg_qs}",
headers={
"User-Agent": "Mozilla/5.0 (iPhone; CPU iPhone OS 17_0 like Mac OS X) AppleWebKit/605.1.15 Safari/604.1",
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"Accept-Language": "it-IT,it;q=0.9,en;q=0.8",
},
)
raw = await asyncio.to_thread(
lambda: _urlreq.urlopen(ddg_req, timeout=10).read().decode("utf-8", errors="replace")
)
links = _re.findall(r'class="result__a"[^>]+href="([^"]+)"[^>]*>(.*?)', raw, _re.S)
snippets = _re.findall(r'class="result__snippet"[^>]*>(.*?)', raw, _re.S)
out = []
for i, (href, title_html) in enumerate(links[:max_results]):
title = _html.unescape(_re.sub(r"<[^>]+>", "", title_html)).strip()
snippet = _html.unescape(_re.sub(r"<[^>]+>", "", snippets[i] if i < len(snippets) else "")).strip()
if title and href.startswith("http"):
# S600: snippet 300→500 — DDG snippet spesso viene troncato a 300
out.append({"title": title, "url": href, "snippet": snippet[:500], "source": "DDG"})
return out
except Exception:
return []
async def _jina() -> list:
"""S378: Jina Search — gratuito, no API key, fallback affidabile."""
try:
import urllib.parse as _up
jina_url = f"https://s.jina.ai/?q={_up.quote(query)}"
async with httpx.AsyncClient(timeout=10) as c:
r = await c.get(
jina_url,
headers={**_headers, "Accept": "application/json", "X-No-Cache": "true"},
)
if r.status_code == 200:
data = r.json().get("data", [])
return [
{
"title": item.get("title", ""),
"url": item.get("url", ""),
# S602: snippet 300→500 — Jina description/content parity con altri provider
"snippet": (item.get("description") or item.get("content") or "")[:500],
"source": "Jina",
}
for item in data[:max_results]
if item.get("title") and item.get("url")
]
except Exception as _exc:
_logger.debug("[registry] silenced %s", type(_exc).__name__) # noqa: BLE001
return []
async def _hackernews() -> list:
"""S763: HackerNews via Algolia — gratuito, no API key, qualità alta su query tech."""
try:
async with httpx.AsyncClient(timeout=8) as c:
r = await c.get(
"https://hn.algolia.com/api/v1/search",
params={"query": query, "hitsPerPage": min(max_results, 4), "tags": "story"},
)
if r.status_code == 200:
return [
{"title": h.get("title", "")[:300], # S607: 200→300
"snippet": (h.get("story_text") or "")[:300], # S583: 250→300
"url": h.get("url") or f"https://news.ycombinator.com/item?id={h.get('objectID','')}",
"source": "HackerNews"}
for h in r.json().get("hits", [])
if h.get("title")
]
except Exception as _exc:
_logger.debug("[registry] silenced %s", type(_exc).__name__) # noqa: BLE001
return []
# S357: lancio parallelo — worst case = max timeout singolo (10s), non 4×10s=40s
# S378: Jina aggiunto come 5° provider (gratuito, no API key)
# GAP3-fix: HackerNews aggiunto come 6° provider (tech quality, da web_search.py ora integrato)
_gather_results = await asyncio.gather(
_brave(), _tavily(), _wikipedia(), _ddg(), _jina(), _hackernews(),
return_exceptions=True,
)
all_lists = [r for r in _gather_results if not isinstance(r, BaseException)]
# Merge con deduplicazione per URL (priorità: Brave > Tavily > Wikipedia > DDG > Jina)
seen_urls: set = set()
results: list = []
for res_list in all_lists:
for item in (res_list if isinstance(res_list, list) else []):
url = item.get("url", "")
if url and url not in seen_urls:
seen_urls.add(url)
results.append(item)
if len(results) >= max_results:
break
if len(results) >= max_results:
break
return {"query": query, "results": results[:max_results]}
async def _read_page(url: str, query: str = "") -> dict:
"""S763: upgrade — usa extract_with_trafilatura (Readability-quality).
Fallback a regex strip se trafilatura non disponibile.
query opzionale per filtrare paragrafi rilevanti.
"""
try:
async with httpx.AsyncClient(timeout=15, follow_redirects=True) as c:
r = await c.get(
url,
headers={"User-Agent": "Mozilla/5.0 (compatible; AgentBot/1.0)"},
)
try:
from tools.content_cleaner import extract_with_trafilatura
result = extract_with_trafilatura(
html=r.text, url=url, query=query or None, max_chars=5000,
)
return {
"url": url,
"content": result["content"],
"status": r.status_code,
"extractor": result.get("extractor", "trafilatura"),
"chars": result.get("chars", 0),
}
except ImportError:
import re as _re
_c = _re.sub(r"<[^>]+>", " ", r.text)
_c = _re.sub(r"\s+", " ", _c).strip()
return {"url": url, "content": _c[:5000], "status": r.status_code, "extractor": "regex"}
except Exception as e:
return {"url": url, "content": "", "error": str(e)[:300]}
async def _get_weather(city: str) -> dict:
try:
async with httpx.AsyncClient(timeout=8) as c:
# S390-B-H: usa params= invece di f-string per URL encoding corretto.
# f-string non codifica spazi/accenti → API restituisce 0 risultati per "New York", "Reggio Emilia", ecc.
geo = await c.get(
"https://geocoding-api.open-meteo.com/v1/search",
params={"name": city, "count": 1, "language": "it", "format": "json"},
)
results = geo.json().get("results", [])
if not results:
return {"error": f"Città '{city}' non trovata"}
loc = results[0]
weather = await c.get(
f"https://api.open-meteo.com/v1/forecast?latitude={loc['latitude']}&longitude={loc['longitude']}"
f"¤t=temperature_2m,weather_code,wind_speed_10m&timezone=auto"
)
w = weather.json().get("current", {})
return {"city": loc["name"], "country": loc.get("country", ""), "temp_c": w.get("temperature_2m"), "wind_kmh": w.get("wind_speed_10m"), "code": w.get("weather_code")}
except Exception as e:
return {"error": str(e)}
async def _calculate(expression: str) -> dict:
try:
import ast, operator
# S390-B-M: aggiunto Mod (%) e FloorDiv (//) agli operator consentiti
allowed = {ast.Add: operator.add, ast.Sub: operator.sub, ast.Mult: operator.mul, ast.Div: operator.truediv, ast.Pow: operator.pow, ast.USub: operator.neg, ast.Mod: operator.mod, ast.FloorDiv: operator.floordiv}
def eval_node(node):
if isinstance(node, ast.Constant): return node.value
if isinstance(node, ast.BinOp): return allowed[type(node.op)](eval_node(node.left), eval_node(node.right))
if isinstance(node, ast.UnaryOp): return allowed[type(node.op)](eval_node(node.operand))
raise ValueError("Operazione non supportata")
tree = ast.parse(expression, mode='eval')
result = eval_node(tree.body)
return {"expression": expression, "result": result}
except Exception as e:
return {"expression": expression, "error": str(e)}
async def _run_python(code: str) -> dict:
"""S574/S749: run_python con backend-exec microservice (sandbox persistente per sessione).
Priorità:
1. backend-exec (EXEC_ENGINE_URL configurato) — sandbox persistente, pip install per sessione,
resource limits 512 MB, TTL cleaner. Usa AGENT_SESSION_ID env come session_id.
2. Fallback: exec_sandbox locale (S574) — tmpdir effimera, comportamento pre-S749.
Vantaggio chiave: se lo step precedente ha installato pandas, questo step lo trova.
"""
# S749-D: session_id da ContextVar (impostata da unified_loop per task isolation)
_session_id = _agent_session_id_var.get()
_remote = await _call_exec_engine({
"code": code,
"language": "python",
"timeout": 15,
"session_id": _session_id,
})
if _remote is not None:
# Normalizza formato: backend-exec → {exit_code, stdout, stderr}
# exec_sandbox → {returncode, stdout, stderr}
return {
"returncode": _remote.get("exit_code", -1),
"stdout": _remote.get("stdout", ""),
"stderr": _remote.get("stderr", ""),
}
# Fallback locale (S574)
from api.exec_sandbox import run_in_sandbox_async
return await run_in_sandbox_async(code, lang="python",
task_id=_session_id, timeout=15.0) # GAP-2: usa session_id per task isolation (no race condition)
async def _generate_image(prompt: str, width: int = 512, height: int = 512) -> dict:
"""
Genera immagine AI: FLUX.1-schnell via backend /api/vision/generate (HF Space, gratuito),
fallback Pollinations AI se FLUX non disponibile.
"""
import urllib.parse
# V001: Prova prima il backend interno FLUX (stesso processo, localhost)
_base_url = os.environ.get("BACKEND_BASE_URL", "http://localhost:7860")
_token = os.environ.get("INTERNAL_TOKEN", "")
try:
async with httpx.AsyncClient(timeout=30.0) as c:
_r = await c.post(
f"{_base_url}/api/vision/generate",
json={"prompt": prompt.strip()[:600], "width": width, "height": height},
headers={**({"X-Internal-Token": _token} if _token else {})},
)
if _r.status_code == 200:
_data = _r.json()
if _data.get("ok") and _data.get("image_url"):
return {
"url": _data["image_url"],
"prompt": prompt,
"width": width,
"height": height,
"ready": True,
"source": "flux",
"note": "Immagine generata con FLUX.1-schnell via backend.",
}
except Exception:
pass # FLUX non raggiungibile → Pollinations fallback
# Fallback: Pollinations AI (gratuito, nessuna API key)
encoded = urllib.parse.quote(prompt.strip(), safe="")
seed = sum(ord(c) for c in prompt) % 9999 + 1
url = (
f"https://image.pollinations.ai/prompt/{encoded}"
f"?width={width}&height={height}&seed={seed}&nologo=true&enhance=true"
)
return {
"url": url,
"prompt": prompt,
"width": width,
"height": height,
"ready": True,
"source": "pollinations",
"note": "Copia l\'URL nel browser o incollalo in un tag ![]()
per vedere l\'immagine.",
}
async def _browser_navigate(url: str, wait_ms: int = 2000, mobile: bool = False) -> dict:
"""
S403 — Limite 1 (criticità 10/10): Browser execution layer.
Naviga a una URL con Playwright headless, restituisce titolo + testo + DOM links/inputs.
Stateless (nessuna sessione persistente) — usa per lettura/scraping rapido.
Playwright è già installato nel backend (browser.py v5 S65).
"""
try:
from api.browser import (
_safe_url, _LAUNCH_ARGS, _make_context, _DOM_SCRIPT,
GOTO_TIMEOUT, MAX_LINKS, MAX_INPUTS, MAX_TEXT
)
if not _safe_url(url):
return {"error": "URL non consentita (localhost/intranet bloccato per sicurezza)"}
from playwright.async_api import async_playwright
async with async_playwright() as pw:
browser = await pw.chromium.launch(headless=True, args=_LAUNCH_ARGS)
ctx = await _make_context(browser, 1280, 800, mobile)
page = await ctx.new_page()
try:
await page.goto(url, wait_until="domcontentloaded", timeout=GOTO_TIMEOUT)
await page.wait_for_timeout(wait_ms)
title = await page.title()
dom_raw = await page.evaluate(_DOM_SCRIPT % (MAX_LINKS, MAX_INPUTS, MAX_TEXT))
text = (dom_raw.get("text") or "")[:2000] if isinstance(dom_raw, dict) else ""
links = (dom_raw.get("links") or [])[:15] if isinstance(dom_raw, dict) else []
inputs = (dom_raw.get("inputs") or [])[:10] if isinstance(dom_raw, dict) else []
return {
"url": page.url,
"title": title,
"text": text,
"links": links,
"inputs": inputs,
}
except Exception as e:
# S600: 300→500 — browser inner exception può contenere stack/path
return {"url": url, "error": str(e)[:500]}
finally:
await ctx.close()
await browser.close()
except ImportError:
return {"error": "Playwright non disponibile — usa read_page come alternativa"}
except Exception as e:
# S600: 300→500 — outer exception
return {"error": str(e)[:500]}
async def _browser_session_open(url: str, wait_ms: int = 1500, mobile: bool = False) -> dict:
"""
S403 — Apre sessione Playwright persistente (stateful).
Restituisce session_id da usare con browser_session_act e browser_session_close.
Max 2 sessioni simultanee (OOM guard HF free tier).
Usa questa modalità per task multi-step: login → naviga → compila → invia.
"""
try:
from api.browser import (
_safe_url, _LAUNCH_ARGS, _make_context, _DOM_SCRIPT,
_sessions, SESSION_LIMIT, GOTO_TIMEOUT, MAX_LINKS, MAX_INPUTS, MAX_TEXT,
_close_session
)
import uuid, time
if not _safe_url(url):
return {"error": "URL non consentita"}
if len(_sessions) >= SESSION_LIMIT:
oldest = min(_sessions, key=lambda sid: _sessions[sid]["last_used"])
await _close_session(oldest, "OOM guard from tool")
from playwright.async_api import async_playwright
pw = await async_playwright().start()
browser = await pw.chromium.launch(headless=True, args=_LAUNCH_ARGS)
ctx = await _make_context(browser, 1280, 800, mobile)
page = await ctx.new_page()
await page.goto(url, wait_until="domcontentloaded", timeout=GOTO_TIMEOUT)
await page.wait_for_timeout(wait_ms)
title = await page.title()
dom_raw = await page.evaluate(_DOM_SCRIPT % (MAX_LINKS, MAX_INPUTS, MAX_TEXT))
sid = uuid.uuid4().hex[:16]
_sessions[sid] = {
"pw": pw, "browser": browser, "context": ctx, "page": page,
"created_at": time.time(), "last_used": time.time(), "url": page.url,
}
return {
"session_id": sid,
"url": page.url,
"title": title,
"text": (dom_raw.get("text") or "")[:1500] if isinstance(dom_raw, dict) else "",
"links": (dom_raw.get("links") or [])[:15] if isinstance(dom_raw, dict) else [],
"inputs": (dom_raw.get("inputs") or [])[:10] if isinstance(dom_raw, dict) else [],
}
except ImportError:
return {"error": "Playwright non disponibile"}
except Exception as e:
# S600: 300→500 — registry browser session exception
return {"error": str(e)[:500]}
async def _browser_session_act(session_id: str, actions: list, wait_ms: int = 1000) -> dict:
"""
S403 — Esegue azioni su sessione Playwright esistente.
actions: lista di {type, selector?, value?, key?, ms?}
Tipi supportati: click, fill, select, press, hover, wait_for, wait, scroll.
Restituisce DOM aggiornato (titolo + testo + links + inputs).
"""
try:
from api.browser import (
_sessions, _execute_actions, _DOM_SCRIPT,
MAX_LINKS, MAX_INPUTS, MAX_TEXT
)
import time
sess = _sessions.get(session_id)
if not sess:
return {"error": f"Sessione {session_id} non trovata o scaduta (TTL 8 min)"}
sess["last_used"] = time.time()
page = sess["page"]
# Converti lista dict → oggetti con attributo .type, .selector, ecc.
class _A:
def __init__(self, d: dict):
self.type = d.get("type", "")
self.selector = d.get("selector")
self.value = d.get("value")
self.key = d.get("key")
self.ms = d.get("ms")
await _execute_actions(page, [_A(a) for a in (actions or [])])
await page.wait_for_timeout(wait_ms)
sess["url"] = page.url
title = await page.title()
dom_raw = await page.evaluate(_DOM_SCRIPT % (MAX_LINKS, MAX_INPUTS, MAX_TEXT))
return {
"session_id": session_id,
"url": page.url,
"title": title,
"text": (dom_raw.get("text") or "")[:1500] if isinstance(dom_raw, dict) else "",
"links": (dom_raw.get("links") or [])[:15] if isinstance(dom_raw, dict) else [],
"inputs": (dom_raw.get("inputs") or [])[:10] if isinstance(dom_raw, dict) else [],
}
except ImportError:
return {"error": "Playwright non disponibile"}
except Exception as e:
# S601: 300→500 — parity con altri session handler
return {"error": str(e)[:500]}
async def _browser_session_close(session_id: str) -> dict:
"""S403 — Chiude sessione Playwright e libera memoria (~300 MB/sessione)."""
try:
from api.browser import _sessions, _close_session
if session_id not in _sessions:
return {"ok": True, "note": "Sessione già chiusa o non trovata"}
await _close_session(session_id, "tool call")
return {"ok": True, "session_id": session_id}
except Exception as e:
return {"ok": False, "error": str(e)[:300]} # S589: 200→300
async def _get_news(query: str, max_results: int = 5) -> dict:
"""S378: get_news — alias di web_search con query ottimizzata per notizie recenti.
smolagents chiama questo tool per richieste di tipo 'notizie su X'.
"""
news_query = f"notizie recenti {query}" if not any(
kw in query.lower() for kw in ("notizie", "news", "ultime", "latest", "breaking")
) else query
return await _web_search(news_query, max_results=max_results)
# ─── S666: tool FS (apply_patch/write_file/read_file/execute_shell) ───────────
# Questi tool erano in unified_loop._TOOL_MAP (S659) ma NON in TOOL_REGISTRY.
# agent.py usa TOOL_REGISTRY per lookup → chiamate a questi tool falliscono con
# KeyError silenzioso. Aggiunte _fn + entries per copertura completa del registry.
import pathlib as _pathlib
async def _read_file(path: str, encoding: str = "utf-8") -> dict:
"""S666/SEC-FS-JAIL: Legge un file dal filesystem del backend, confinato a _fs_jail_root()."""
try:
_p, _err = _safe_fs_path(path)
if _err:
return {"ok": False, "error": _err}
if not _p.exists():
return {"ok": False, "error": f"File non trovato: {path}"}
if _p.stat().st_size > 10 * 1024 * 1024: # 10 MB limit
return {"ok": False, "error": f"File troppo grande (>{10}MB): {path}"}
text = _p.read_text(encoding=encoding, errors="replace")
return {"ok": True, "content": text, "path": path, "size": len(text)}
except PermissionError:
return {"ok": False, "error": f"Accesso negato: {path}"}
except Exception as exc:
return {"ok": False, "error": str(exc)}
async def _write_file(path: str, content: str, encoding: str = "utf-8") -> dict:
"""S666/GAP-2/SEC-FS-JAIL: Scrive/sovrascrive un file nel filesystem del backend, confinato a _fs_jail_root().
Manus Gap fix: read-back verifica completezza scrittura.
write_text() puo completare senza eccezione anche su FS con truncation silenziosa."""
try:
_p, _err = _safe_fs_path(path)
if _err:
return {"ok": False, "error": _err}
_p.parent.mkdir(parents=True, exist_ok=True)
_p.write_text(content, encoding=encoding)
# GAP-2: read-back — verifica che il file abbia la dimensione attesa
# apply_patch fa gia questo (+ rollback TS); ora anche _write_file e atomico
_written = _p.read_text(encoding=encoding)
if len(_written) != len(content):
return {
"ok": False,
"error": f"Write truncated: expected {len(content)} chars, got {len(_written)}",
"path": path,
}
return {"ok": True, "path": path, "size": len(_written)}
except PermissionError:
return {"ok": False, "error": f"Accesso negato: {path}"}
except Exception as exc:
return {"ok": False, "error": str(exc)}
async def _ts_syntax_check(path: str) -> tuple[bool, str]:
"""GAP-6: verifica sintassi TypeScript via node --check dopo apply_patch.
Ritorna (ok, error_msg). Non-bloccante: su node assente o timeout assume ok."""
try:
result = await asyncio.to_thread(
subprocess.run,
["node", "--check", path],
capture_output=True, text=True, timeout=10,
)
if result.returncode == 0:
return True, ""
return False, (result.stderr or result.stdout).strip()[:500]
except (FileNotFoundError, subprocess.TimeoutExpired):
return True, "" # node non disponibile o timeout → assume ok (fail-safe)
except Exception:
return True, "" # fail-safe assoluto — non deve mai bloccare il tool
async def _apply_patch(path: str, patch: str) -> dict:
"""S666/SEC-FS-JAIL: Applica una patch unified-diff a un file esistente, confinato a _fs_jail_root()."""
import io
try:
_p, _err = _safe_fs_path(path)
if _err:
return {"ok": False, "error": _err}
if not _p.exists():
return {"ok": False, "error": f"File non trovato: {path}"}
original = _p.read_text(encoding="utf-8", errors="replace")
# Prova con subprocess patch(1) se disponibile, altrimenti Python fallback
result = await asyncio.to_thread(
subprocess.run,
["patch", "-p0", "--input=-", str(_p)],
input=patch, capture_output=True, text=True, timeout=15,
)
if result.returncode == 0:
# GAP-6: typecheck post-patch su .ts/.tsx — rollback automatico se sintassi invalida
if path.endswith((".ts", ".tsx")):
_ts_ok, _ts_err = await _ts_syntax_check(path)
if not _ts_ok:
try: _p.write_text(original, encoding="utf-8")
except Exception: pass
return {"ok": False, "error": f"Rollback: sintassi TS invalida dopo patch — {_ts_err}", "path": path, "rolled_back": True}
return {"ok": True, "path": path, "applied": True, "output": result.stdout.strip()}
# Fallback: ricerca/sostituzione del blocco più semplice
lines = patch.splitlines()
removes = [l[1:] for l in lines if l.startswith("-") and not l.startswith("---")]
adds = [l[1:] for l in lines if l.startswith("+") and not l.startswith("+++")]
if removes and len(removes) == len(adds):
patched = original
for rem, add in zip(removes, adds):
patched = patched.replace(rem, add, 1)
if patched != original:
_p.write_text(patched, encoding="utf-8")
# GAP-6: typecheck post-patch su .ts/.tsx — rollback automatico se sintassi invalida
if path.endswith((".ts", ".tsx")):
_ts_ok, _ts_err = await _ts_syntax_check(path)
if not _ts_ok:
try: _p.write_text(original, encoding="utf-8")
except Exception: pass
return {"ok": False, "error": f"Rollback: sintassi TS invalida dopo patch (fallback) — {_ts_err}", "path": path, "rolled_back": True}
return {"ok": True, "path": path, "applied": True, "method": "fallback_replace"}
return {"ok": False, "error": result.stderr.strip() or "Patch non applicata", "path": path}
except FileNotFoundError:
return {"ok": False, "error": "patch(1) non trovato — solo fallback Python disponibile"}
except Exception as exc:
return {"ok": False, "error": str(exc)}
# S679/S749: execute_shell con backend-exec microservice (sessione persistente)
async def _execute_shell(command: str, timeout: int = 30, cwd: str = ".") -> dict:
"""S666/S749: Esegue un comando shell con timeout e cattura output.
Priorità:
1. backend-exec (EXEC_ENGINE_URL) — sessione persistente, pip install condiviso con run_python.
2. Fallback: subprocess locale con asyncio.to_thread.
"""
_shell_err = _validate_shell_cmd(command)
if _shell_err:
return {"ok":False,"error":_shell_err,"command":command}
# S749-D: session_id da ContextVar (impostata da unified_loop per task isolation)
_session_id = _agent_session_id_var.get()
_remote = await _call_exec_engine({
"command": command,
"timeout": min(timeout, 60),
"session_id": _session_id,
}, endpoint="/api/execute-shell")
if _remote is not None:
_ec = _remote.get("exit_code", -1)
return {
"ok": _ec == 0,
"stdout": _remote.get("stdout", "")[:8192],
"stderr": _remote.get("stderr", "")[:2048],
"code": _ec,
"command": command,
}
# Fallback locale — _run_shell_sync (NO shell=True, env pulita)
try:
_fb = await asyncio.to_thread(_run_shell_sync, command, cwd, min(timeout,120))
return {"ok":_fb["ok"],"stdout":_fb["stdout"],"stderr":_fb["stderr"],
"code":_fb.get("code",-1),"command":command,
}
except subprocess.TimeoutExpired:
return {"ok": False, "error": f"Timeout {timeout}s superato", "command": command}
except Exception as exc:
return {"ok": False, "error": str(exc), "command": command}
# ─── Registry ─────────────────────────────────────────────
# ── V002-V004: nuovi tool backend (web_research, send_email, database_query) ──
async def _web_research(topic: str, depth: int = 4, synthesize: bool = True) -> dict:
"""Ricerca web multi-fonte con sintesi AI via /api/web/research."""
_base = os.environ.get("BACKEND_BASE_URL", "http://localhost:7860")
_tok = os.environ.get("INTERNAL_TOKEN", "")
try:
async with httpx.AsyncClient(timeout=60.0) as c:
r = await c.post(
f"{_base}/api/web/research",
json={"topic": str(topic)[:400], "depth": min(int(depth), 8), "synthesize": synthesize},
headers={**({"X-Internal-Token": _tok} if _tok else {})},
)
if r.status_code == 200:
return r.json()
except Exception as exc:
return {"ok": False, "error": str(exc)[:200]}
return {"ok": False, "error": "web_research endpoint non disponibile"}
async def _send_email(to: str, subject: str, body: str, html: bool = False, from_name: str = "Agente AI") -> dict:
"""Invia email via /api/email/send (richiede RESEND_API_KEY)."""
_base = os.environ.get("BACKEND_BASE_URL", "http://localhost:7860")
_tok = os.environ.get("INTERNAL_TOKEN", "")
try:
async with httpx.AsyncClient(timeout=20.0) as c:
r = await c.post(
f"{_base}/api/email/send",
json={"to": to, "subject": subject, "body": body, "html": html, "from_name": from_name},
headers={**({"X-Internal-Token": _tok} if _tok else {})},
)
if r.status_code == 200:
return r.json()
except Exception as exc:
return {"ok": False, "message": str(exc)[:200]}
return {"ok": False, "message": "send_email endpoint non disponibile"}
async def _database_query(sql: str, params: list | None = None, read_only: bool = True) -> dict:
"""Esegue query SQL su DATABASE_URL configurato via /api/database/query."""
_base = os.environ.get("BACKEND_BASE_URL", "http://localhost:7860")
_tok = os.environ.get("INTERNAL_TOKEN", "")
try:
async with httpx.AsyncClient(timeout=30.0) as c:
r = await c.post(
f"{_base}/api/database/query",
json={"sql": str(sql)[:2000], "params": params or [], "read_only": read_only},
headers={**({"X-Internal-Token": _tok} if _tok else {})},
)
if r.status_code == 200:
return r.json()
except Exception as exc:
return {"ok": False, "error": str(exc)[:200]}
return {"ok": False, "error": "database_query endpoint non disponibile"}
async def _call_api(url: str, method: str = "GET", headers: dict | None = None,
body=None, auth_type: str = "none", auth_value: str = "",
timeout_ms: int = 15000) -> dict:
"""S601: Chiama qualsiasi endpoint REST esterno con metodo/headers/body/auth configurabili.
SEC-SSRF: valida scheme + risolve hostname, blocca IP privati/loopback/riservati.
follow_redirects=False per evitare bypass SSRF via redirect verso IP interni."""
import json as _j
_ssrf_err = _ssrf_check(url)
if _ssrf_err:
return {"ok": False, "error": _ssrf_err, "url": url}
_method = method.upper()
_headers = dict(headers or {})
_timeout = min(float(timeout_ms) / 1000, 30.0)
# auth injection
if auth_type == "bearer" and auth_value:
_headers["Authorization"] = f"Bearer {auth_value}"
elif auth_type == "basic" and auth_value:
import base64 as _b64
_headers["Authorization"] = "Basic " + _b64.b64encode(auth_value.encode()).decode()
elif auth_type == "api_key" and auth_value and ":" in auth_value:
k, v = auth_value.split(":", 1)
_headers[k.strip()] = v.strip()
try:
async with httpx.AsyncClient(timeout=_timeout, follow_redirects=False) as c:
kwargs = {"headers": _headers}
if body is not None:
if isinstance(body, (dict, list)):
kwargs["json"] = body
else:
kwargs["content"] = str(body).encode()
_headers.setdefault("Content-Type", "text/plain")
r = await c.request(_method, url, **kwargs)
_ct = r.headers.get("content-type", "")
if "json" in _ct:
try:
resp_body = r.json()
except Exception:
resp_body = r.text[:4000]
else:
resp_body = r.text[:4000]
return {
"ok": True, "status": r.status_code,
"body": resp_body, "headers": dict(r.headers),
}
except Exception as exc:
return {"ok": False, "status": 0, "error": str(exc)[:200]}
async def _execute_sql(sql: str, params: list | None = None, read_only: bool = True) -> dict:
"""S601: alias di _database_query — esegue SQL su DATABASE_URL (PostgreSQL/SQLite)."""
return await _database_query(sql=sql, params=params, read_only=read_only)
async def _create_pdf(content: str, filename: str = "documento.pdf",
format: str = "html") -> dict:
"""S601: Genera PDF da HTML/Markdown/testo via backend /api/vision/pdf."""
_base = os.environ.get("BACKEND_BASE_URL", "http://localhost:7860")
_tok = os.environ.get("INTERNAL_TOKEN", "")
try:
async with httpx.AsyncClient(timeout=30.0) as c:
r = await c.post(
f"{_base}/api/vision/pdf",
json={"content": str(content)[:50000], "filename": filename, "format": format},
headers={**({"X-Internal-Token": _tok} if _tok else {})},
)
if r.status_code == 200:
return r.json()
return {"ok": False, "error": f"HTTP {r.status_code}", "detail": r.text[:200]}
except Exception as exc:
return {"ok": False, "error": str(exc)[:200]}
# ─── 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 [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 []. 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 \n