aleleba
5f0ddd962c
Ran the full agent loop against vllm-qwen36 on port 8000 (read-only HTTP)
with the Penpot plugin live, before asking for any downtime. Without this
file "it improved" would be a claim rather than a measurement.
Result over the 8 graded prompts: mean score 14.9, zero prompts at or above
60, veto violated on 2 of 8, 66% of execute_code calls raised, and 7 of 8
prompts burned all 14 turns without producing a final message.
The plan predicted production would score near zero on distinct colours and
style richness while producing a high shape count - grey boxes. The shape
count is also zero. On a fresh page it creates nothing at all, so the
failure sits upstream of the grey boxes: the model invents a Figma-shaped
API wholesale and every call throws. From the captured turns:
penpot.currentPage() is a property, not a function
penpot.createRectangle(page, 200, 56) takes no arguments
penpot.createText(page, ...) takes one, the text
penpot.getPageById(...) lives on penpotUtils
fills = [{type:'solid', color:{r,g,b,a}}] is {fillColor, fillOpacity}
shadows = [{type:'drop', x, y, blur, ...}] is {style, offsetX, offsetY}
It then spends the remaining turns querying penpot_api_info without
recovering. So the reported symptom understated it.
Two robustness fixes the run itself forced, both after losing a completed
run to them:
- A ConnectionError does not just drop the request, it can drop the MCP
session, so retrying the same tools/call against a dead session fails
identically every time - which is exactly what the first attempt showed,
four retries and four identical ConnectionErrors. The client now redoes
the handshake before retrying, and that recovered two drops in this run.
- Results are written after every prompt. The first attempt died on prompt
4 and lost the three already measured, which is the expensive data
precisely because it requires production to be up.
1382 lines
58 KiB
Python
1382 lines
58 KiB
Python
"""Fase 6 -- Puerta 5: calidad de diseno en Penpot, medida contra el MCP en vivo.
|
|
|
|
A diferencia de las puertas 2-4, esta necesita un LOOP DE AGENTE REAL: modelo -> tool call ->
|
|
MCP de Penpot en vivo -> tool result -> modelo, hasta GATE5_MAX_TURNS turnos por prompt. La
|
|
calidad de diseno no existe hasta que el codigo se ejecuta: un payload de `execute_code`
|
|
perfectamente plausible puede dejar ocho rectangulos grises, y eso solo se ve en el arbol.
|
|
|
|
GATE5_BASE_URL=http://10.212.133.200:8004/v1 \
|
|
GATE5_MODEL=qwen3.6-35b-a3b-mcp-v2-nvfp4 \
|
|
PENPOT_MCP_URL=... PENPOT_MCP_TOKEN=... GATE5_TAG=v2-nvfp4 \
|
|
python3 scripts/35_gate5_penpot_design.py
|
|
|
|
Modo barato (disparador de fallback, corre ANTES de gastar un minuto en cuantizacion):
|
|
|
|
GATE5_HOLDOUT=1 GATE5_BASE_URL=... GATE5_MODEL=... GATE5_TAG=v2-bf16 \
|
|
python3 scripts/35_gate5_penpot_design.py
|
|
|
|
En modo holdout NO se toca el MCP: se genera una sola respuesta por prompt de
|
|
data/holdout_penpot_design.jsonl (~60) y se hace el grep estatico sobre los payloads de `code`
|
|
que produjo, sin ejecutarlos. Umbral: tasa de API prohibida <= 5% (<=3/60).
|
|
|
|
DOS ENDPOINTS, LOS DOS POR ENV, SIN DEFAULTS Y SIN IMPRIMIRSE NUNCA
|
|
------------------------------------------------------------------
|
|
- El modelo: API OpenAI-compatible de vLLM (GATE5_BASE_URL, GATE5_MODEL).
|
|
- El MCP de Penpot: servidor HTTP MCP (streamable HTTP / JSON-RPC 2.0) en PENPOT_MCP_URL con
|
|
header `Authorization: Bearer $PENPOT_MCP_TOKEN`. Ni la URL ni el token se imprimen jamas
|
|
-- la URL suele llevar el token embebido en el path.
|
|
|
|
PRECONDICION HUMANA: el plugin de Penpot tiene que estar abierto en el navegador y conectado al
|
|
archivo. Si no lo esta, el servidor responde "No Penpot plugin instances are currently
|
|
connected..." y la puerta ABORTA DE ENTRADA con exit code 2, antes de quemar los 8 prompts.
|
|
|
|
LA PUERTA NUNCA BORRA NADA. Crea una pagina fresca por prompt (`gate5/<tag>/<prompt-id>/<ts>`)
|
|
y la deja para inspeccion humana, con su PNG exportado.
|
|
|
|
LAS 17 METRICAS las computa un payload de auditoria que inyecta LA PUERTA (nunca el modelo)
|
|
sobre el arbol resultante. Ese payload respeta la misma API verificada que se le exige al
|
|
modelo -- se autochequea al arrancar contra los patrones prohibidos de
|
|
scripts/07_lint_penpot_code.py, que es la unica fuente de verdad de esos regex.
|
|
|
|
`placeholderGreys` y `forbiddenBehavior` son METRICAS DE VETO: cualquier violacion -> el prompt
|
|
saca 0. score = 100 * pasadas / aplicables.
|
|
Aprobacion: score medio >= 65, veto limpio en 8/8, >=6/8 prompts >= 60, y la landing de
|
|
pizzeria >= 60.
|
|
|
|
Exit 0 si aprueba, 1 si no, 2 si el plugin de Penpot no esta conectado.
|
|
"""
|
|
import base64
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
import time
|
|
import unicodedata
|
|
from collections import Counter
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
import requests
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parent.parent
|
|
SCHEMAS_DIR = REPO_ROOT / "data" / "schemas"
|
|
PROMPTS_PATH = REPO_ROOT / "data" / "gate5_prompts.jsonl"
|
|
HOLDOUT_PATH = REPO_ROOT / "data" / "holdout_penpot_design.jsonl"
|
|
PENPOT_TOOLS_PATH = SCHEMAS_DIR / "penpot.json"
|
|
SYSTEM_PROMPT_PATH = SCHEMAS_DIR / "penpot_system_prompt.md"
|
|
LINT_PATH = REPO_ROOT / "scripts" / "07_lint_penpot_code.py"
|
|
|
|
# Corpus contra el que los 8 prompts tienen que ser disjuntos (shingles de 6-gramas).
|
|
DISJOINT_AGAINST = [
|
|
REPO_ROOT / "data" / "train.jsonl",
|
|
REPO_ROOT / "data" / "eval.jsonl",
|
|
REPO_ROOT / "data" / "raw" / "seeds" / "penpot.jsonl",
|
|
]
|
|
SHINGLE_N = 6
|
|
|
|
# Reintento de transporte contra el MCP de Penpot. El servidor corta la conexion a mitad de
|
|
# llamada de forma intermitente; ya se habia visto en la puerta 4 de la Fase 4.
|
|
MCP_MAX_RETRIES = int(os.environ.get("PENPOT_MCP_RETRIES", "4"))
|
|
MCP_RETRY_BACKOFF = float(os.environ.get("PENPOT_MCP_RETRY_BACKOFF", "3"))
|
|
|
|
MAX_TURNS = int(os.environ.get("GATE5_MAX_TURNS", "14"))
|
|
MAX_TOKENS = int(os.environ.get("GATE5_MAX_TOKENS", "4096"))
|
|
HTTP_TIMEOUT = int(os.environ.get("GATE5_HTTP_TIMEOUT", "600"))
|
|
MCP_TIMEOUT = int(os.environ.get("GATE5_MCP_TIMEOUT", "180"))
|
|
|
|
# El prompt que reproduce el fallo exacto de produccion. Tiene su propia condicion de
|
|
# aprobacion: si este no llega a 60, la puerta no aprueba aunque el promedio alcance.
|
|
FLAGSHIP_PROMPT_ID = "g5-06-landing-pizzeria"
|
|
|
|
MEAN_SCORE_MIN = 65.0
|
|
PROMPT_SCORE_MIN = 60.0
|
|
MIN_PROMPTS_OVER_MIN = 6
|
|
HOLDOUT_FORBIDDEN_RATE_MAX = 0.05
|
|
|
|
PLUGIN_NOT_CONNECTED = "No Penpot plugin instances are currently connected"
|
|
EXIT_PLUGIN_NOT_CONNECTED = 2
|
|
|
|
|
|
class Gate5Error(RuntimeError):
|
|
"""Fallo de la puerta que no es un fallo del modelo (env, transporte, MCP)."""
|
|
|
|
|
|
class PluginNotConnected(Gate5Error):
|
|
"""El plugin de Penpot no esta abierto/conectado: no tiene sentido seguir."""
|
|
|
|
|
|
# ------------------------------------------------------------------------------------------
|
|
# Patrones prohibidos: se IMPORTAN del lint, nunca se duplican
|
|
# ------------------------------------------------------------------------------------------
|
|
def load_lint_module():
|
|
"""Importa scripts/07_lint_penpot_code.py (el nombre empieza con digito, no es importable
|
|
con `import`). La fuente de verdad de los regex prohibidos tiene que ser una sola."""
|
|
spec = importlib.util.spec_from_file_location("lint_penpot_code", LINT_PATH)
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
LINT = load_lint_module()
|
|
|
|
# El subconjunto de FORBIDDEN que constituye "comportamiento prohibido" a efectos de VETO,
|
|
# identificado por el nombre exacto con el que el lint lo declara. El resto de los patrones del
|
|
# lint se sigue registrando en el JSON de resultados (columna `otros_patrones`) porque es
|
|
# informacion util, pero no veta: la lista de veto es la del PLAN, ni mas ni menos.
|
|
VETO_PATTERN_NAMES = {
|
|
"findShapeById con 2 argumentos",
|
|
"propiedad .layout inexistente",
|
|
"flex.appendChild",
|
|
"fontSize/fontWeight/lineHeight/letterSpacing numerico",
|
|
"import_image / importImage / createImage / filePath",
|
|
}
|
|
|
|
# Argumentos que el schema de export_shape NO declara. Inventarlos es el mismo comportamiento
|
|
# prohibido, solo que del lado de los argumentos de la tool y no del texto del `code`.
|
|
EXPORT_SHAPE_ALLOWED_ARGS = {"shapeId", "format", "mode"}
|
|
|
|
|
|
def scan_forbidden(code):
|
|
"""Devuelve (veto_hits, otros_hits) sobre un payload de `code`."""
|
|
veto, otros = [], []
|
|
for pname, rx, _why in LINT.FORBIDDEN:
|
|
match = rx.search(code)
|
|
if not match:
|
|
continue
|
|
frag = code[max(0, match.start() - 40):match.end() + 40].replace("\n", " ")
|
|
hit = {"patron": pname, "fragmento": frag}
|
|
(veto if pname in VETO_PATTERN_NAMES else otros).append(hit)
|
|
return veto, otros
|
|
|
|
|
|
# ------------------------------------------------------------------------------------------
|
|
# Entorno
|
|
# ------------------------------------------------------------------------------------------
|
|
def require_env(name, why):
|
|
value = os.environ.get(name, "").strip()
|
|
if not value:
|
|
raise Gate5Error(f"falta la variable de entorno {name} ({why})")
|
|
return value
|
|
|
|
|
|
def load_env(need_mcp):
|
|
"""Lee la config. Nunca devuelve nada que se pueda imprimir sin pensar: la URL del MCP
|
|
lleva el token embebido en varios deployments."""
|
|
faltantes = []
|
|
cfg = {}
|
|
for name, why in (("GATE5_BASE_URL", "endpoint OpenAI-compatible de vLLM, ej. http://HOST:PUERTO/v1"),
|
|
("GATE5_MODEL", "nombre del modelo servido por vLLM")):
|
|
try:
|
|
cfg[name] = require_env(name, why)
|
|
except Gate5Error as e:
|
|
faltantes.append(str(e))
|
|
if need_mcp:
|
|
for name, why in (("PENPOT_MCP_URL", "endpoint HTTP MCP del servidor de Penpot"),
|
|
("PENPOT_MCP_TOKEN", "token bearer del servidor MCP de Penpot")):
|
|
try:
|
|
cfg[name] = require_env(name, why)
|
|
except Gate5Error as e:
|
|
faltantes.append(str(e))
|
|
if faltantes:
|
|
raise Gate5Error("configuracion incompleta:\n - " + "\n - ".join(faltantes))
|
|
cfg["GATE5_TAG"] = os.environ.get("GATE5_TAG", "").strip() or "sin-tag"
|
|
return cfg
|
|
|
|
|
|
# ------------------------------------------------------------------------------------------
|
|
# Cliente MCP sobre HTTP (streamable HTTP / JSON-RPC 2.0)
|
|
# ------------------------------------------------------------------------------------------
|
|
class PenpotMCP:
|
|
"""Cliente minimo de MCP sobre HTTP.
|
|
|
|
Handshake: initialize -> notifications/initialized -> tools/list -> tools/call.
|
|
La respuesta puede llegar como `application/json` o como `text/event-stream` (lineas
|
|
`data: {...}`); las dos se manejan. El `Mcp-Session-Id` que devuelve el initialize se
|
|
guarda y se reusa en todas las llamadas siguientes.
|
|
"""
|
|
|
|
PROTOCOL_VERSION = "2025-06-18"
|
|
|
|
def __init__(self, url, token, timeout=MCP_TIMEOUT):
|
|
self._url = url
|
|
self._token = token
|
|
self.timeout = timeout
|
|
self.session_id = None
|
|
self.tool_names = []
|
|
self._next_id = 0
|
|
self._in_handshake = False
|
|
self._http = requests.Session()
|
|
|
|
# -- transporte -------------------------------------------------------------------
|
|
def _headers(self):
|
|
headers = {
|
|
"Content-Type": "application/json",
|
|
"Accept": "application/json, text/event-stream",
|
|
"Authorization": f"Bearer {self._token}",
|
|
"MCP-Protocol-Version": self.PROTOCOL_VERSION,
|
|
}
|
|
if self.session_id:
|
|
headers["Mcp-Session-Id"] = self.session_id
|
|
return headers
|
|
|
|
@staticmethod
|
|
def _parse_body(response):
|
|
"""Devuelve la lista de mensajes JSON-RPC del cuerpo, venga como SSE o como JSON."""
|
|
ctype = (response.headers.get("Content-Type") or "").lower()
|
|
text = response.text or ""
|
|
if "text/event-stream" in ctype:
|
|
mensajes = []
|
|
for line in text.splitlines():
|
|
line = line.strip()
|
|
if not line.startswith("data:"):
|
|
continue
|
|
payload = line[len("data:"):].strip()
|
|
if not payload or payload == "[DONE]":
|
|
continue
|
|
try:
|
|
mensajes.append(json.loads(payload))
|
|
except json.JSONDecodeError:
|
|
continue
|
|
return mensajes
|
|
if not text.strip():
|
|
return []
|
|
try:
|
|
parsed = json.loads(text)
|
|
except json.JSONDecodeError as e:
|
|
raise Gate5Error(f"el servidor MCP devolvio un cuerpo no-JSON ({e})")
|
|
return parsed if isinstance(parsed, list) else [parsed]
|
|
|
|
def _post(self, payload, expect_response=True):
|
|
try:
|
|
response = self._http.post(
|
|
self._url, headers=self._headers(), json=payload, timeout=self.timeout
|
|
)
|
|
except requests.RequestException as e:
|
|
# El mensaje de requests incluye la URL -> se descarta a proposito.
|
|
raise Gate5Error(f"error de transporte contra el MCP de Penpot: {type(e).__name__}")
|
|
|
|
nueva_sesion = response.headers.get("Mcp-Session-Id") or response.headers.get("mcp-session-id")
|
|
if nueva_sesion:
|
|
self.session_id = nueva_sesion
|
|
|
|
if response.status_code >= 400:
|
|
cuerpo = (response.text or "")[:300]
|
|
if PLUGIN_NOT_CONNECTED in cuerpo:
|
|
raise PluginNotConnected(cuerpo.strip())
|
|
raise Gate5Error(f"el MCP de Penpot respondio HTTP {response.status_code}: {cuerpo}")
|
|
|
|
if not expect_response:
|
|
return None
|
|
|
|
mensajes = self._parse_body(response)
|
|
for msg in mensajes:
|
|
if msg.get("id") != payload.get("id"):
|
|
continue
|
|
if "error" in msg:
|
|
detalle = json.dumps(msg["error"], ensure_ascii=False)[:400]
|
|
if PLUGIN_NOT_CONNECTED in detalle:
|
|
raise PluginNotConnected(detalle)
|
|
raise Gate5Error(f"el MCP de Penpot devolvio un error JSON-RPC: {detalle}")
|
|
return msg.get("result", {})
|
|
raise Gate5Error(f"el MCP de Penpot no devolvio respuesta para el metodo {payload.get('method')}")
|
|
|
|
def _request(self, method, params=None):
|
|
"""Envia un request JSON-RPC, reintentando los fallos de TRANSPORTE.
|
|
|
|
El MCP de Penpot corta la conexion a mitad de llamada de forma intermitente
|
|
("transport dropped mid-call"); ya se habia visto en la puerta 4 de la Fase 4. Sin
|
|
reintento, un corte transitorio en el prompt 6 de 8 tira abajo una corrida entera de la
|
|
puerta, y peor: se diagnostica como un fallo del modelo cuando es de la red.
|
|
|
|
Se reintenta SOLO `Gate5Error` de transporte/protocolo. `PluginNotConnected` no se
|
|
reintenta -- ese es un estado real del mundo (el usuario no tiene el plugin abierto) y
|
|
reintentarlo solo demora el mensaje que hay que dar.
|
|
"""
|
|
ultimo = None
|
|
for intento in range(1, MCP_MAX_RETRIES + 1):
|
|
self._next_id += 1
|
|
payload = {"jsonrpc": "2.0", "id": self._next_id, "method": method}
|
|
if params is not None:
|
|
payload["params"] = params
|
|
try:
|
|
return self._post(payload)
|
|
except PluginNotConnected:
|
|
raise
|
|
except Gate5Error as e:
|
|
ultimo = e
|
|
if intento == MCP_MAX_RETRIES:
|
|
break
|
|
espera = MCP_RETRY_BACKOFF * (2 ** (intento - 1))
|
|
print(f"[MCP] fallo transitorio en '{method}' (intento {intento}/"
|
|
f"{MCP_MAX_RETRIES}): {e}. Reintento en {espera:.0f}s")
|
|
time.sleep(espera)
|
|
# Un ConnectionError no solo tira la request: puede tirar la SESION. Reintentar
|
|
# el mismo tools/call contra una sesion muerta falla siempre igual, que es lo
|
|
# que se vio en la primera corrida (4 intentos identicos, 4 ConnectionError).
|
|
# Hay que rehacer el handshake antes de reintentar.
|
|
if not self._in_handshake and method != "initialize":
|
|
try:
|
|
print("[MCP] rehaciendo el handshake (la sesion pudo haber muerto)")
|
|
self.session_id = None
|
|
self.handshake()
|
|
except Gate5Error as e2:
|
|
print(f"[MCP] el re-handshake tambien fallo: {e2}")
|
|
raise Gate5Error(f"el MCP de Penpot fallo {MCP_MAX_RETRIES} veces en '{method}': {ultimo}")
|
|
|
|
def _notify(self, method, params=None):
|
|
payload = {"jsonrpc": "2.0", "method": method}
|
|
if params is not None:
|
|
payload["params"] = params
|
|
self._post(payload, expect_response=False)
|
|
|
|
# -- protocolo --------------------------------------------------------------------
|
|
def handshake(self):
|
|
self._in_handshake = True
|
|
try:
|
|
return self._handshake_inner()
|
|
finally:
|
|
self._in_handshake = False
|
|
|
|
def _handshake_inner(self):
|
|
result = self._request("initialize", {
|
|
"protocolVersion": self.PROTOCOL_VERSION,
|
|
"capabilities": {},
|
|
"clientInfo": {"name": "gate5-penpot-design", "version": "1.0"},
|
|
})
|
|
self._notify("notifications/initialized")
|
|
listado = self._request("tools/list") or {}
|
|
self.tool_names = [t.get("name") for t in listado.get("tools", [])]
|
|
server = (result.get("serverInfo") or {}).get("name", "?")
|
|
print(f"[MCP] handshake ok con '{server}'; tools: {', '.join(self.tool_names)}")
|
|
return result
|
|
|
|
@staticmethod
|
|
def _flatten_content(result):
|
|
"""Aplana `content` de un tool result MCP a (texto, lista_de_binarios_base64)."""
|
|
partes, binarios = [], []
|
|
for item in (result or {}).get("content", []) or []:
|
|
tipo = item.get("type")
|
|
if tipo == "text":
|
|
partes.append(item.get("text") or "")
|
|
elif tipo in ("image", "audio") and item.get("data"):
|
|
binarios.append(item["data"])
|
|
partes.append(f"[{tipo}/{item.get('mimeType', '?')}, {len(item['data'])} bytes base64]")
|
|
elif tipo == "resource":
|
|
recurso = item.get("resource") or {}
|
|
if recurso.get("text"):
|
|
partes.append(recurso["text"])
|
|
elif recurso.get("blob"):
|
|
binarios.append(recurso["blob"])
|
|
return "\n".join(partes).strip(), binarios
|
|
|
|
def call_tool(self, name, arguments):
|
|
"""Devuelve (texto, binarios, es_error). Aborta si el plugin no esta conectado."""
|
|
result = self._request("tools/call", {"name": name, "arguments": arguments}) or {}
|
|
texto, binarios = self._flatten_content(result)
|
|
if PLUGIN_NOT_CONNECTED in texto:
|
|
raise PluginNotConnected(texto)
|
|
return texto, binarios, bool(result.get("isError"))
|
|
|
|
def execute_code(self, code):
|
|
texto, _binarios, es_error = self.call_tool("execute_code", {"code": code})
|
|
return texto, es_error
|
|
|
|
def execute_json(self, code, que):
|
|
"""execute_code cuyo resultado la puerta necesita parsear como JSON.
|
|
|
|
El servidor MCP no devuelve el `return` del payload pelado: lo envuelve en
|
|
`{"result": <lo-que-retornaste>, "log": "<lo-que-escribiste-por-console>"}`. La puerta
|
|
necesita el interior, asi que se desenvuelve ACA, en un solo lugar.
|
|
|
|
Ojo con no desenvolver de mas: si un payload de la puerta retornara a proposito un
|
|
objeto con una clave `result` propia, desenvolver a ciegas lo destruiria. Por eso se
|
|
exige la forma exacta del envoltorio (`result` presente y ninguna clave fuera de
|
|
{result, log}).
|
|
|
|
`execute_code` (el metodo de abajo) NO desenvuelve: lo que ve el modelo como tool result
|
|
tiene que ser byte a byte lo que le llegaria en produccion, envoltorio incluido.
|
|
"""
|
|
texto, es_error = self.execute_code(code)
|
|
try:
|
|
parsed = json.loads(texto)
|
|
except json.JSONDecodeError:
|
|
raise Gate5Error(
|
|
f"{que}: el MCP no devolvio JSON ({'error' if es_error else 'texto'}): {texto[:300]}"
|
|
)
|
|
if (isinstance(parsed, dict) and "result" in parsed
|
|
and set(parsed).issubset({"result", "log"})):
|
|
log = (parsed.get("log") or "").strip()
|
|
if log:
|
|
print(f"[MCP] console del payload de la puerta ({que}): {log[:300]}")
|
|
return parsed["result"]
|
|
return parsed
|
|
|
|
|
|
# ------------------------------------------------------------------------------------------
|
|
# Payloads de la puerta (nunca los escribe el modelo)
|
|
# ------------------------------------------------------------------------------------------
|
|
def js_string(value):
|
|
"""Literal JS seguro a partir de un valor Python (via JSON, que es un subconjunto)."""
|
|
return json.dumps(value, ensure_ascii=False)
|
|
|
|
|
|
SETUP_JS = """
|
|
const page = penpot.createPage();
|
|
page.name = %PAGE_NAME%;
|
|
penpot.openPage(page);
|
|
const seed = %SEED%;
|
|
let seedBoardId = null;
|
|
if (seed) {
|
|
const board = penpot.createBoard();
|
|
board.name = seed.nombre;
|
|
board.x = seed.x;
|
|
board.y = seed.y;
|
|
board.resize(seed.ancho, seed.alto);
|
|
board.fills = [{ fillColor: "#FFFFFF", fillOpacity: 1 }];
|
|
for (const spec of seed.rectangulos) {
|
|
const rect = penpot.createRectangle();
|
|
rect.name = spec.nombre;
|
|
rect.x = seed.x + spec.x;
|
|
rect.y = seed.y + spec.y;
|
|
rect.resize(spec.ancho, spec.alto);
|
|
rect.fills = [{ fillColor: seed.gris, fillOpacity: 1 }];
|
|
board.insertChild(board.children.length, rect);
|
|
}
|
|
seedBoardId = board.id;
|
|
}
|
|
return {
|
|
pageId: page.id,
|
|
pageName: page.name,
|
|
rootId: penpot.root.id,
|
|
seedBoardId: seedBoardId
|
|
};
|
|
"""
|
|
|
|
# Payload de auditoria. Lo inyecta la puerta, asi que tiene que respetar la misma API verificada
|
|
# que se le exige al modelo: findShapeById de 1 argumento, board.flex/board.grid (nunca .layout),
|
|
# includeChildren (nunca withChildren), sin asignaciones a propiedades read-only.
|
|
AUDIT_JS = """
|
|
const ROOT_ID = %ROOT_ID%;
|
|
const root = penpotUtils.findShapeById(ROOT_ID);
|
|
if (!root) { return { auditError: "root no encontrado" }; }
|
|
|
|
const nodes = [];
|
|
const collect = (shape, depth) => {
|
|
nodes.push({ shape: shape, depth: depth });
|
|
const kids = shape.children || [];
|
|
for (let i = 0; i < kids.length; i++) { collect(kids[i], depth + 1); }
|
|
};
|
|
const top = root.children || [];
|
|
for (let i = 0; i < top.length; i++) { collect(top[i], 1); }
|
|
|
|
const PUROS = ["#FFFFFF", "#000000"];
|
|
const esGrisPlaceholder = (hex) => {
|
|
const r = parseInt(hex.slice(1, 3), 16);
|
|
const g = parseInt(hex.slice(3, 5), 16);
|
|
const b = parseInt(hex.slice(5, 7), 16);
|
|
const mx = Math.max(r, g, b);
|
|
const mn = Math.min(r, g, b);
|
|
const sat = mx === 0 ? 0 : (100 * (mx - mn)) / mx;
|
|
return sat <= 10 && mx >= 100 && mx <= 220;
|
|
};
|
|
|
|
const hexes = [];
|
|
const grises = [];
|
|
let gradientes = 0, sombras = 0, radios = 0;
|
|
let textos = 0, textosConChars = 0, textosConFill = 0, textosConFont = 0;
|
|
const fontSizes = [];
|
|
let boards = 0, boardsConLayout = 0, hijosDeLayout = 0, hijosDimensionados = 0;
|
|
let profundidad = 0, secciones = 0;
|
|
|
|
for (const nodo of nodes) {
|
|
const s = nodo.shape;
|
|
if (nodo.depth > profundidad) { profundidad = nodo.depth; }
|
|
let fills = null;
|
|
try { fills = s.fills; } catch (e) { fills = null; }
|
|
const locales = [];
|
|
if (Array.isArray(fills)) {
|
|
for (const f of fills) {
|
|
if (f && typeof f.fillColor === "string") { locales.push(f.fillColor.toUpperCase()); }
|
|
const grad = f && f.fillColorGradient;
|
|
if (grad) {
|
|
gradientes++;
|
|
for (const stop of (grad.stops || [])) {
|
|
if (stop && typeof stop.color === "string") { locales.push(stop.color.toUpperCase()); }
|
|
}
|
|
}
|
|
}
|
|
}
|
|
for (const hex of locales) {
|
|
hexes.push(hex);
|
|
if (esGrisPlaceholder(hex)) { grises.push({ id: s.id, name: s.name, hex: hex }); }
|
|
}
|
|
try { if (Array.isArray(s.shadows) && s.shadows.length > 0) { sombras++; } } catch (e) {}
|
|
try { if (typeof s.borderRadius === "number" && s.borderRadius > 0) { radios++; } } catch (e) {}
|
|
|
|
if (s.type === "text") {
|
|
textos++;
|
|
const chars = (s.characters || "").trim();
|
|
if (chars.length > 0) { textosConChars++; }
|
|
if (Array.isArray(fills) && fills.length > 0) { textosConFill++; }
|
|
if (s.fontId && String(s.fontId).length > 0) { textosConFont++; }
|
|
const tam = parseFloat(s.fontSize);
|
|
if (!isNaN(tam)) { fontSizes.push(tam); }
|
|
}
|
|
|
|
if (s.type === "board") {
|
|
boards++;
|
|
const conLayout = !!(s.flex || s.grid);
|
|
if (conLayout) {
|
|
boardsConLayout++;
|
|
for (const kid of (s.children || [])) {
|
|
hijosDeLayout++;
|
|
let lc = null;
|
|
try { lc = kid.layoutChild; } catch (e) { lc = null; }
|
|
if (lc && lc.horizontalSizing) { hijosDimensionados++; }
|
|
}
|
|
}
|
|
if (nodo.depth <= 2 && (s.children || []).length >= 2) { secciones++; }
|
|
}
|
|
}
|
|
|
|
const distintos = [];
|
|
for (const hex of hexes) {
|
|
if (PUROS.indexOf(hex) === -1 && distintos.indexOf(hex) === -1) { distintos.push(hex); }
|
|
}
|
|
|
|
let contencion = [];
|
|
let contencionError = null;
|
|
try {
|
|
const analisis = penpotUtils.analyzeDescendants(root, (raiz, shape) => {
|
|
const padre = shape.parent;
|
|
if (!padre || padre.id === raiz.id) { return null; }
|
|
return penpotUtils.isContainedIn(shape, padre)
|
|
? null
|
|
: { id: shape.id, name: shape.name, parent: padre.name };
|
|
}, 12) || [];
|
|
for (const entrada of analisis) {
|
|
if (entrada && entrada.result) { contencion.push(entrada.result); }
|
|
}
|
|
} catch (e) {
|
|
contencionError = String((e && e.message) || e);
|
|
}
|
|
|
|
let principal = null;
|
|
for (const s of top) {
|
|
if (s.type !== "board") { continue; }
|
|
if (!principal || s.width * s.height > principal.width * principal.height) { principal = s; }
|
|
}
|
|
if (!principal && top.length > 0) { principal = top[0]; }
|
|
|
|
let desajusteAltura = null;
|
|
if (principal) {
|
|
let fondo = null;
|
|
for (const kid of (principal.children || [])) {
|
|
const limite = kid.y + kid.height;
|
|
if (fondo === null || limite > fondo) { fondo = limite; }
|
|
}
|
|
if (fondo !== null) {
|
|
desajusteAltura = Math.round(Math.abs(principal.y + principal.height - fondo));
|
|
}
|
|
}
|
|
|
|
let cssLen = 0, markupLen = 0, renderError = null;
|
|
try {
|
|
const objetivo = principal ? [principal] : top;
|
|
if (objetivo.length > 0) {
|
|
const css = penpot.generateStyle(objetivo, { type: "css", includeChildren: true }) || "";
|
|
const markup = penpot.generateMarkup(objetivo, { type: "html" }) || "";
|
|
cssLen = String(css).length;
|
|
markupLen = String(markup).length;
|
|
}
|
|
} catch (e) {
|
|
renderError = String((e && e.message) || e);
|
|
}
|
|
|
|
const tamsUnicos = [];
|
|
for (const tam of fontSizes) { if (tamsUnicos.indexOf(tam) === -1) { tamsUnicos.push(tam); } }
|
|
|
|
return {
|
|
shapeCount: nodes.length,
|
|
distinctFillColors: distintos.length,
|
|
fillColors: distintos.slice(0, 24),
|
|
placeholderGreys: grises.length,
|
|
greySamples: grises.slice(0, 8),
|
|
textsTotal: textos,
|
|
textsWithCharacters: textosConChars,
|
|
distinctFontSizes: tamsUnicos.length,
|
|
maxFontSize: tamsUnicos.length > 0 ? Math.max.apply(null, tamsUnicos) : 0,
|
|
textFillShare: textos > 0 ? textosConFill / textos : 0,
|
|
fontIdBoundShare: textos > 0 ? textosConFont / textos : 0,
|
|
boards: boards,
|
|
boardsWithLayout: boardsConLayout,
|
|
layoutChildren: hijosDeLayout,
|
|
layoutChildrenSized: hijosDimensionados,
|
|
layoutChildSizingShare: hijosDeLayout > 0 ? hijosDimensionados / hijosDeLayout : 0,
|
|
containmentViolations: contencion.length,
|
|
containmentSamples: contencion.slice(0, 5),
|
|
containmentError: contencionError,
|
|
nestingDepth: profundidad,
|
|
sectionCount: secciones,
|
|
radiusCount: radios,
|
|
shadowCount: sombras,
|
|
gradientCount: gradientes,
|
|
styleRichness: radios + sombras + gradientes,
|
|
mainBoardId: principal ? principal.id : null,
|
|
rootHeightMismatch: desajusteAltura,
|
|
cssLength: cssLen,
|
|
markupLength: markupLen,
|
|
renderError: renderError
|
|
};
|
|
"""
|
|
|
|
|
|
def build_setup_js(page_name, seed_board):
|
|
return (SETUP_JS
|
|
.replace("%PAGE_NAME%", js_string(page_name))
|
|
.replace("%SEED%", js_string(seed_board) if seed_board else "null"))
|
|
|
|
|
|
def build_audit_js(root_id):
|
|
return AUDIT_JS.replace("%ROOT_ID%", js_string(root_id))
|
|
|
|
|
|
def self_check_gate_payloads():
|
|
"""El payload de auditoria de la puerta tiene que respetar la misma API verificada que se le
|
|
exige al modelo. Si no, la puerta medirira sus propias excepciones."""
|
|
problemas = []
|
|
for nombre, code in (("SETUP_JS", SETUP_JS), ("AUDIT_JS", AUDIT_JS)):
|
|
veto, otros = scan_forbidden(code)
|
|
for hit in veto + otros:
|
|
problemas.append(f"{nombre}: [{hit['patron']}] ...{hit['fragmento']}...")
|
|
if problemas:
|
|
raise Gate5Error(
|
|
"los payloads de la propia puerta violan la API verificada:\n - "
|
|
+ "\n - ".join(problemas)
|
|
)
|
|
|
|
|
|
# ------------------------------------------------------------------------------------------
|
|
# Disjuncion contra train/eval/seeds via shingles de 6-gramas
|
|
# ------------------------------------------------------------------------------------------
|
|
def normalizar(texto):
|
|
sin_tildes = "".join(
|
|
c for c in unicodedata.normalize("NFD", texto.lower()) if unicodedata.category(c) != "Mn"
|
|
)
|
|
return re.findall(r"[a-z0-9]+", sin_tildes)
|
|
|
|
|
|
def shingles(texto, n=SHINGLE_N):
|
|
tokens = normalizar(texto)
|
|
if len(tokens) < n:
|
|
return {" ".join(tokens)} if tokens else set()
|
|
return {" ".join(tokens[i:i + n]) for i in range(len(tokens) - n + 1)}
|
|
|
|
|
|
def corpus_shingles(paths):
|
|
"""Shingles de todo el texto de usuario de los corpus de entrenamiento/eval/seeds."""
|
|
acumulado = set()
|
|
for path in paths:
|
|
if not path.exists():
|
|
print(f"[WARN] {path.name} no existe: no se puede verificar disjuncion contra el")
|
|
continue
|
|
with open(path, encoding="utf-8") as f:
|
|
for linea in f:
|
|
linea = linea.strip()
|
|
if not linea:
|
|
continue
|
|
try:
|
|
fila = json.loads(linea)
|
|
except json.JSONDecodeError:
|
|
continue
|
|
for msg in fila.get("messages", []):
|
|
if msg.get("role") != "user":
|
|
continue
|
|
acumulado |= shingles(msg.get("content") or "")
|
|
return acumulado
|
|
|
|
|
|
def check_disjunto(prompts):
|
|
corpus = corpus_shingles(DISJOINT_AGAINST)
|
|
if not corpus:
|
|
print("[WARN] corpus vacio: se saltea el chequeo de disjuncion")
|
|
return
|
|
problemas = []
|
|
for fila in prompts:
|
|
solapados = sorted(shingles(fila["prompt"]) & corpus)
|
|
if solapados:
|
|
problemas.append(f"{fila['id']}: {len(solapados)} shingle(s), p.ej. {solapados[0]!r}")
|
|
if problemas:
|
|
raise Gate5Error(
|
|
"los prompts de la puerta 5 NO son disjuntos de train/eval/seeds "
|
|
f"(shingles de {SHINGLE_N}-gramas):\n - " + "\n - ".join(problemas)
|
|
)
|
|
print(f"[OK] los {len(prompts)} prompts son disjuntos de train/eval/seeds "
|
|
f"({len(corpus)} shingles de referencia)")
|
|
|
|
|
|
# ------------------------------------------------------------------------------------------
|
|
# Las 17 metricas
|
|
# ------------------------------------------------------------------------------------------
|
|
# (clave, etiqueta, comparador). Comparadores:
|
|
# min -> pasa si valor >= umbral
|
|
# max -> pasa si valor <= umbral
|
|
# pair -> valor y umbral son tuplas; pasa si cada componente >= la suya
|
|
# bool -> pasa si valor es True (el umbral es True o None para "no aplica")
|
|
METRICS = [
|
|
("shapeCount", "conteo de shapes", "min"),
|
|
("distinctFillColors", "fillColor distintos (sin blanco/negro puros)", "min"),
|
|
("placeholderGreys", "grises de placeholder [VETO]", "max"),
|
|
("textsWithCharacters", "textos con characters no vacio", "min"),
|
|
("typeScale", "fontSize distintos y maximo", "pair"),
|
|
("textFillShare", "share de textos con fill", "min"),
|
|
("layoutBoards", "boards con flex/grid", "min"),
|
|
("layoutChildSizingShare", "share de hijos con layoutChild.horizontalSizing", "min"),
|
|
("containmentViolations", "violaciones de contencion", "max"),
|
|
("structure", "profundidad de anidamiento y secciones", "pair"),
|
|
("styleRichness", "riqueza de estilo (radios/sombras/gradientes)", "min"),
|
|
("rootHeightMismatch", "desajuste de altura root vs contenido", "max"),
|
|
("exportBeforeFinal", "export_shape antes del mensaje final", "bool"),
|
|
("forbiddenBehavior", "comportamiento prohibido [VETO]", "max"),
|
|
("exceptionRate", "tasa de excepciones de execute_code", "max"),
|
|
("fontIdBoundShare", "binding tipografico real (fontId)", "min"),
|
|
("rendersOk", "el arbol renderiza (generateStyle/generateMarkup)", "bool"),
|
|
]
|
|
|
|
VETO_METRICS = ("placeholderGreys", "forbiddenBehavior")
|
|
|
|
# Umbrales por prompt. `None` = la metrica NO aplica a ese prompt y no entra en el denominador
|
|
# del score. Estan aca y no en el .jsonl a proposito: el .jsonl describe la TAREA, el umbral es
|
|
# criterio de la puerta y se versiona junto al codigo que lo aplica.
|
|
THRESHOLDS = {
|
|
"g5-01-boton-primario": {
|
|
"shapeCount": 2, "distinctFillColors": 2, "placeholderGreys": 0,
|
|
"textsWithCharacters": 1, "typeScale": (1, 14), "textFillShare": 1.0,
|
|
"layoutBoards": None, "layoutChildSizingShare": None, "containmentViolations": 0,
|
|
"structure": None, "styleRichness": 2, "rootHeightMismatch": 24,
|
|
"exportBeforeFinal": True, "forbiddenBehavior": 0, "exceptionRate": 0.34,
|
|
"fontIdBoundShare": 1.0, "rendersOk": True,
|
|
},
|
|
"g5-02-navbar-flex": {
|
|
"shapeCount": 6, "distinctFillColors": 3, "placeholderGreys": 0,
|
|
"textsWithCharacters": 5, "typeScale": (1, 14), "textFillShare": 1.0,
|
|
"layoutBoards": 1, "layoutChildSizingShare": 0.5, "containmentViolations": 0,
|
|
"structure": (2, 1), "styleRichness": 1, "rootHeightMismatch": 24,
|
|
"exportBeforeFinal": True, "forbiddenBehavior": 0, "exceptionRate": 0.34,
|
|
"fontIdBoundShare": 1.0, "rendersOk": True,
|
|
},
|
|
"g5-03-card-producto": {
|
|
"shapeCount": 6, "distinctFillColors": 3, "placeholderGreys": 0,
|
|
"textsWithCharacters": 4, "typeScale": (3, 20), "textFillShare": 1.0,
|
|
"layoutBoards": None, "layoutChildSizingShare": None, "containmentViolations": 0,
|
|
"structure": (2, 1), "styleRichness": 3, "rootHeightMismatch": 32,
|
|
"exportBeforeFinal": True, "forbiddenBehavior": 0, "exceptionRate": 0.34,
|
|
"fontIdBoundShare": 1.0, "rendersOk": True,
|
|
},
|
|
"g5-04-tokens-biblioteca": {
|
|
"shapeCount": 8, "distinctFillColors": 6, "placeholderGreys": 0,
|
|
"textsWithCharacters": 4, "typeScale": (3, 24), "textFillShare": 1.0,
|
|
"layoutBoards": None, "layoutChildSizingShare": None, "containmentViolations": 0,
|
|
"structure": (2, 1), "styleRichness": 2, "rootHeightMismatch": 32,
|
|
"exportBeforeFinal": True, "forbiddenBehavior": 0, "exceptionRate": 0.34,
|
|
"fontIdBoundShare": 1.0, "rendersOk": True,
|
|
},
|
|
"g5-05-grid-tres-diferenciales": {
|
|
"shapeCount": 12, "distinctFillColors": 3, "placeholderGreys": 0,
|
|
"textsWithCharacters": 9, "typeScale": (2, 20), "textFillShare": 1.0,
|
|
"layoutBoards": 1, "layoutChildSizingShare": 0.5, "containmentViolations": 0,
|
|
"structure": (3, 3), "styleRichness": 3, "rootHeightMismatch": 32,
|
|
"exportBeforeFinal": True, "forbiddenBehavior": 0, "exceptionRate": 0.34,
|
|
"fontIdBoundShare": 1.0, "rendersOk": True,
|
|
},
|
|
"g5-06-landing-pizzeria": {
|
|
"shapeCount": 25, "distinctFillColors": 5, "placeholderGreys": 0,
|
|
"textsWithCharacters": 12, "typeScale": (4, 32), "textFillShare": 1.0,
|
|
"layoutBoards": 2, "layoutChildSizingShare": 0.5, "containmentViolations": 0,
|
|
"structure": (3, 5), "styleRichness": 5, "rootHeightMismatch": 48,
|
|
"exportBeforeFinal": True, "forbiddenBehavior": 0, "exceptionRate": 0.34,
|
|
"fontIdBoundShare": 1.0, "rendersOk": True,
|
|
},
|
|
"g5-07-onboarding-mobile": {
|
|
"shapeCount": 10, "distinctFillColors": 4, "placeholderGreys": 0,
|
|
"textsWithCharacters": 5, "typeScale": (3, 24), "textFillShare": 1.0,
|
|
"layoutBoards": 1, "layoutChildSizingShare": 0.5, "containmentViolations": 0,
|
|
"structure": (3, 2), "styleRichness": 3, "rootHeightMismatch": 32,
|
|
"exportBeforeFinal": True, "forbiddenBehavior": 0, "exceptionRate": 0.34,
|
|
"fontIdBoundShare": 1.0, "rendersOk": True,
|
|
},
|
|
"g5-08-reparacion-grises": {
|
|
"shapeCount": 8, "distinctFillColors": 4, "placeholderGreys": 0,
|
|
"textsWithCharacters": 5, "typeScale": (3, 24), "textFillShare": 1.0,
|
|
"layoutBoards": None, "layoutChildSizingShare": None, "containmentViolations": 0,
|
|
"structure": (2, 2), "styleRichness": 3, "rootHeightMismatch": 32,
|
|
"exportBeforeFinal": True, "forbiddenBehavior": 0, "exceptionRate": 0.34,
|
|
"fontIdBoundShare": 1.0, "rendersOk": True,
|
|
},
|
|
}
|
|
|
|
|
|
def build_raw_metrics(audit, runtime):
|
|
"""Aplana la auditoria del arbol + las señales del loop en las 17 claves de METRICS."""
|
|
audit = audit or {}
|
|
return {
|
|
"shapeCount": audit.get("shapeCount", 0),
|
|
"distinctFillColors": audit.get("distinctFillColors", 0),
|
|
"placeholderGreys": audit.get("placeholderGreys", 0),
|
|
"textsWithCharacters": audit.get("textsWithCharacters", 0),
|
|
"typeScale": (audit.get("distinctFontSizes", 0), audit.get("maxFontSize", 0)),
|
|
"textFillShare": audit.get("textFillShare", 0.0),
|
|
"layoutBoards": audit.get("boardsWithLayout", 0),
|
|
"layoutChildSizingShare": audit.get("layoutChildSizingShare", 0.0),
|
|
"containmentViolations": audit.get("containmentViolations", 0),
|
|
"structure": (audit.get("nestingDepth", 0), audit.get("sectionCount", 0)),
|
|
"styleRichness": audit.get("styleRichness", 0),
|
|
# Sin contenido no hay desajuste que medir: se penaliza como el peor caso posible.
|
|
"rootHeightMismatch": (audit.get("rootHeightMismatch")
|
|
if audit.get("rootHeightMismatch") is not None else 10 ** 6),
|
|
"exportBeforeFinal": runtime["exportBeforeFinal"],
|
|
"forbiddenBehavior": runtime["forbiddenBehavior"],
|
|
"exceptionRate": runtime["exceptionRate"],
|
|
"fontIdBoundShare": audit.get("fontIdBoundShare", 0.0),
|
|
"rendersOk": bool(audit.get("cssLength", 0) > 0 and audit.get("markupLength", 0) > 0
|
|
and not audit.get("renderError")),
|
|
}
|
|
|
|
|
|
def evaluar(clave, comparador, valor, umbral):
|
|
if umbral is None:
|
|
return None
|
|
if comparador == "min":
|
|
return valor >= umbral
|
|
if comparador == "max":
|
|
return valor <= umbral
|
|
if comparador == "bool":
|
|
return bool(valor) is bool(umbral)
|
|
if comparador == "pair":
|
|
return all(v >= u for v, u in zip(valor, umbral))
|
|
raise Gate5Error(f"comparador desconocido para {clave}: {comparador}")
|
|
|
|
|
|
def score_prompt(prompt_id, raw):
|
|
umbrales = THRESHOLDS[prompt_id]
|
|
detalle = []
|
|
pasadas = aplicables = 0
|
|
veto = []
|
|
for clave, etiqueta, comparador in METRICS:
|
|
umbral = umbrales.get(clave)
|
|
ok = evaluar(clave, comparador, raw[clave], umbral)
|
|
detalle.append({
|
|
"metrica": clave, "etiqueta": etiqueta, "valor": raw[clave],
|
|
"umbral": umbral, "aplica": ok is not None, "pasa": ok,
|
|
})
|
|
if ok is None:
|
|
continue
|
|
aplicables += 1
|
|
if ok:
|
|
pasadas += 1
|
|
elif clave in VETO_METRICS:
|
|
veto.append(clave)
|
|
score = 0.0 if veto else (100.0 * pasadas / aplicables if aplicables else 0.0)
|
|
return {
|
|
"score": round(score, 1), "pasadas": pasadas, "aplicables": aplicables,
|
|
"veto_violado": veto, "metricas": detalle,
|
|
}
|
|
|
|
|
|
# ------------------------------------------------------------------------------------------
|
|
# Cliente del modelo (vLLM, OpenAI-compatible)
|
|
# ------------------------------------------------------------------------------------------
|
|
def load_penpot_tools():
|
|
"""Las 4 tools de data/schemas/penpot.json, byte a byte lo que el modelo ve en produccion."""
|
|
tools = json.loads(PENPOT_TOOLS_PATH.read_text(encoding="utf-8"))
|
|
return [{
|
|
"type": "function",
|
|
"function": {
|
|
"name": t["name"],
|
|
"description": t.get("description", ""),
|
|
"parameters": t.get("parameters") or {"type": "object", "properties": {}},
|
|
},
|
|
} for t in tools]
|
|
|
|
|
|
def load_system_prompt():
|
|
"""El bloque `instructions` verbatim del servidor MCP. Medir con otro system prompt mediria
|
|
otra cosa: es exactamente lo que el modelo ve en produccion."""
|
|
return LINT.extract_system_block()
|
|
|
|
|
|
def chat_completion(cfg, messages, tools):
|
|
payload = {
|
|
"model": cfg["GATE5_MODEL"],
|
|
"messages": messages,
|
|
"tools": tools,
|
|
"tool_choice": "auto",
|
|
"max_tokens": MAX_TOKENS,
|
|
"temperature": 0.0,
|
|
}
|
|
url = cfg["GATE5_BASE_URL"].rstrip("/") + "/chat/completions"
|
|
response = requests.post(url, json=payload, timeout=HTTP_TIMEOUT)
|
|
response.raise_for_status()
|
|
return response.json()["choices"][0]
|
|
|
|
|
|
def parse_tool_arguments(tool_call):
|
|
raw = tool_call.get("function", {}).get("arguments")
|
|
if isinstance(raw, dict):
|
|
return raw, None
|
|
try:
|
|
return json.loads(raw or "{}"), None
|
|
except json.JSONDecodeError as e:
|
|
return {}, f"arguments no es JSON valido: {e}"
|
|
|
|
|
|
# ------------------------------------------------------------------------------------------
|
|
# Loop de agente sobre un prompt
|
|
# ------------------------------------------------------------------------------------------
|
|
def run_prompt(cfg, mcp, tools, system_prompt, fila, png_dir):
|
|
prompt_id = fila["id"]
|
|
tag = cfg["GATE5_TAG"]
|
|
ts = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
|
|
page_name = f"gate5/{tag}/{prompt_id}/{ts}"
|
|
|
|
print(f"\n--- {prompt_id} ({fila.get('dificultad', '?')}) -> pagina {page_name}")
|
|
setup = mcp.execute_json(build_setup_js(page_name, fila.get("seed_board")),
|
|
f"{prompt_id}: setup de la pagina")
|
|
root_id = setup.get("rootId")
|
|
if not root_id:
|
|
raise Gate5Error(f"{prompt_id}: el setup no devolvio rootId: {setup}")
|
|
|
|
contexto = [
|
|
f"Trabajás en la página «{page_name}», que ya está creada y abierta "
|
|
f"(id {setup.get('pageId')}). Todo lo que hagas va en esa página."
|
|
]
|
|
if setup.get("seedBoardId"):
|
|
contexto.append(
|
|
f"El board a intervenir es «{fila['seed_board']['nombre']}», "
|
|
f"id {setup['seedBoardId']}."
|
|
)
|
|
user_content = fila["prompt"] + "\n\n" + " ".join(contexto)
|
|
|
|
messages = [{"role": "system", "content": system_prompt},
|
|
{"role": "user", "content": user_content}]
|
|
|
|
code_payloads = []
|
|
export_calls = []
|
|
turnos = []
|
|
exec_calls = exec_exceptions = 0
|
|
veto_hits, otros_hits = [], []
|
|
final_content = None
|
|
finish_reason = None
|
|
|
|
for turno in range(1, MAX_TURNS + 1):
|
|
try:
|
|
choice = chat_completion(cfg, messages, tools)
|
|
except requests.RequestException as e:
|
|
turnos.append({"turno": turno, "error_modelo": f"{type(e).__name__}: {e}"})
|
|
break
|
|
message = choice.get("message", {})
|
|
finish_reason = choice.get("finish_reason")
|
|
tool_calls = message.get("tool_calls") or []
|
|
registro = {
|
|
"turno": turno,
|
|
"finish_reason": finish_reason,
|
|
"content": message.get("content") or "",
|
|
"reasoning": message.get("reasoning") or "",
|
|
"tool_calls": [],
|
|
}
|
|
|
|
if not tool_calls:
|
|
final_content = message.get("content") or ""
|
|
turnos.append(registro)
|
|
break
|
|
|
|
messages.append({
|
|
"role": "assistant",
|
|
"content": message.get("content") or "",
|
|
"tool_calls": tool_calls,
|
|
})
|
|
|
|
for tool_call in tool_calls:
|
|
nombre = tool_call.get("function", {}).get("name")
|
|
args, error_args = parse_tool_arguments(tool_call)
|
|
resultado = ""
|
|
if error_args:
|
|
resultado = error_args
|
|
elif nombre == "execute_code":
|
|
code = args.get("code", "")
|
|
code_payloads.append({"turno": turno, "code": code})
|
|
veto, otros = scan_forbidden(code)
|
|
for hit in veto:
|
|
hit["turno"] = turno
|
|
for hit in otros:
|
|
hit["turno"] = turno
|
|
veto_hits.extend(veto)
|
|
otros_hits.extend(otros)
|
|
exec_calls += 1
|
|
texto, es_error = mcp.execute_code(code)
|
|
if es_error or not texto.strip().startswith(("{", "[")):
|
|
exec_exceptions += 1
|
|
resultado = texto
|
|
elif nombre == "export_shape":
|
|
inventados = sorted(k for k in args if k not in EXPORT_SHAPE_ALLOWED_ARGS)
|
|
if inventados:
|
|
veto_hits.append({
|
|
"patron": "argumentos inventados en export_shape",
|
|
"fragmento": ", ".join(inventados), "turno": turno,
|
|
})
|
|
export_calls.append({"turno": turno, "args": args})
|
|
limpios = {k: v for k, v in args.items() if k in EXPORT_SHAPE_ALLOWED_ARGS}
|
|
texto, binarios, _ = mcp.call_tool("export_shape", limpios)
|
|
resultado = texto or f"[{len(binarios)} binario(s) devueltos]"
|
|
elif nombre in ("high_level_overview", "penpot_api_info"):
|
|
texto, _binarios, _ = mcp.call_tool(nombre, args)
|
|
resultado = texto
|
|
else:
|
|
resultado = f"Unknown tool: {nombre}"
|
|
|
|
registro["tool_calls"].append({
|
|
"name": nombre, "arguments": args,
|
|
"result_preview": resultado[:400],
|
|
})
|
|
messages.append({
|
|
"role": "tool",
|
|
"tool_call_id": tool_call.get("id"),
|
|
"name": nombre,
|
|
"content": resultado,
|
|
})
|
|
|
|
turnos.append(registro)
|
|
else:
|
|
print(f"[WARN] {prompt_id}: se agotaron los {MAX_TURNS} turnos sin mensaje final")
|
|
|
|
# --- auditoria inyectada por la puerta ------------------------------------------------
|
|
try:
|
|
audit = mcp.execute_json(build_audit_js(root_id), f"{prompt_id}: auditoria")
|
|
except Gate5Error as e:
|
|
print(f"[WARN] {prompt_id}: la auditoria fallo ({e})")
|
|
audit = {"auditError": str(e)}
|
|
|
|
# --- PNG de la puerta (nunca reemplaza al export del modelo, se suma) -----------------
|
|
png_path = None
|
|
shape_a_exportar = audit.get("mainBoardId") or root_id
|
|
try:
|
|
_texto, binarios, _ = mcp.call_tool(
|
|
"export_shape", {"shapeId": shape_a_exportar, "format": "png", "mode": "shape"})
|
|
if binarios:
|
|
png_dir.mkdir(parents=True, exist_ok=True)
|
|
png_path = png_dir / f"{prompt_id}.png"
|
|
png_path.write_bytes(base64.b64decode(binarios[0]))
|
|
print(f"[PNG] {png_path.relative_to(REPO_ROOT)}")
|
|
else:
|
|
print(f"[WARN] {prompt_id}: export_shape no devolvio binario")
|
|
except Gate5Error as e:
|
|
print(f"[WARN] {prompt_id}: no se pudo exportar el PNG ({e})")
|
|
|
|
runtime = {
|
|
"exportBeforeFinal": bool(export_calls),
|
|
"forbiddenBehavior": len(veto_hits),
|
|
"exceptionRate": (exec_exceptions / exec_calls) if exec_calls else 0.0,
|
|
}
|
|
raw = build_raw_metrics(audit, runtime)
|
|
resultado = score_prompt(prompt_id, raw)
|
|
|
|
print(f" score={resultado['score']:.1f} "
|
|
f"({resultado['pasadas']}/{resultado['aplicables']}) "
|
|
f"grises={raw['placeholderGreys']} prohibido={raw['forbiddenBehavior']} "
|
|
f"shapes={raw['shapeCount']} colores={raw['distinctFillColors']} "
|
|
f"textos={raw['textsWithCharacters']}")
|
|
|
|
return {
|
|
"id": prompt_id,
|
|
"dificultad": fila.get("dificultad"),
|
|
"prompt": fila["prompt"],
|
|
"pageId": setup.get("pageId"),
|
|
"pageName": page_name,
|
|
"rootId": root_id,
|
|
"seedBoardId": setup.get("seedBoardId"),
|
|
"turnos_usados": len(turnos),
|
|
"finish_reason": finish_reason,
|
|
"mensaje_final": final_content,
|
|
"auditoria": audit,
|
|
"metricas_crudas": {k: (list(v) if isinstance(v, tuple) else v) for k, v in raw.items()},
|
|
"execute_code_calls": exec_calls,
|
|
"execute_code_excepciones": exec_exceptions,
|
|
"code_payloads": code_payloads,
|
|
"export_shape_calls": export_calls,
|
|
"veto_hits": veto_hits,
|
|
"otros_patrones": otros_hits,
|
|
"png": str(png_path.relative_to(REPO_ROOT)) if png_path else None,
|
|
"turnos": turnos,
|
|
**resultado,
|
|
}
|
|
|
|
|
|
# ------------------------------------------------------------------------------------------
|
|
# Modo holdout: grep estatico barato, sin MCP y sin loop
|
|
# ------------------------------------------------------------------------------------------
|
|
def run_holdout(cfg, tools, system_prompt):
|
|
if not HOLDOUT_PATH.exists():
|
|
raise Gate5Error(
|
|
f"no existe {HOLDOUT_PATH.relative_to(REPO_ROOT)}; generalo con "
|
|
f"scripts/35_build_penpot_design_holdout.py antes de correr el modo holdout"
|
|
)
|
|
filas = []
|
|
with open(HOLDOUT_PATH, encoding="utf-8") as f:
|
|
for linea in f:
|
|
linea = linea.strip()
|
|
if linea:
|
|
filas.append(json.loads(linea))
|
|
print(f"[INFO] modo holdout: {len(filas)} prompts de diseno, sin tocar el MCP\n")
|
|
|
|
resultados = []
|
|
con_prohibido = 0
|
|
errores = 0
|
|
patrones = Counter()
|
|
for i, fila in enumerate(filas, start=1):
|
|
prompt = fila.get("prompt") or fila.get("user") or ""
|
|
messages = [{"role": "system", "content": system_prompt},
|
|
{"role": "user", "content": prompt}]
|
|
try:
|
|
choice = chat_completion(cfg, messages, tools)
|
|
except requests.RequestException as e:
|
|
# Una generacion que nunca ocurrio NO es una generacion limpia: se cuenta aparte y
|
|
# nunca entra en el denominador de la tasa (si no, un endpoint caido "aprueba").
|
|
errores += 1
|
|
resultados.append({"prompt": prompt, "error": f"{type(e).__name__}: {e}"})
|
|
continue
|
|
message = choice.get("message", {})
|
|
veto_total, otros_total, codes = [], [], []
|
|
for tool_call in message.get("tool_calls") or []:
|
|
nombre = tool_call.get("function", {}).get("name")
|
|
args, error_args = parse_tool_arguments(tool_call)
|
|
if error_args:
|
|
continue
|
|
if nombre == "execute_code":
|
|
code = args.get("code", "")
|
|
codes.append(code)
|
|
veto, otros = scan_forbidden(code)
|
|
veto_total.extend(veto)
|
|
otros_total.extend(otros)
|
|
elif nombre == "export_shape":
|
|
inventados = sorted(k for k in args if k not in EXPORT_SHAPE_ALLOWED_ARGS)
|
|
if inventados:
|
|
veto_total.append({"patron": "argumentos inventados en export_shape",
|
|
"fragmento": ", ".join(inventados)})
|
|
elif nombre not in ("high_level_overview", "penpot_api_info"):
|
|
veto_total.append({"patron": "tool inexistente", "fragmento": str(nombre)})
|
|
if veto_total:
|
|
con_prohibido += 1
|
|
for hit in veto_total:
|
|
patrones[hit["patron"]] += 1
|
|
resultados.append({
|
|
"prompt": prompt,
|
|
"id": fila.get("id"),
|
|
"tool_calls": [tc.get("function", {}).get("name")
|
|
for tc in (message.get("tool_calls") or [])],
|
|
"code_payloads": codes,
|
|
"veto_hits": veto_total,
|
|
"otros_patrones": otros_total,
|
|
"content": message.get("content") or "",
|
|
})
|
|
if i % 10 == 0:
|
|
print(f"[INFO] {i}/{len(filas)} prompts procesados")
|
|
|
|
total = len(resultados) - errores
|
|
if total <= 0:
|
|
raise Gate5Error(
|
|
f"ninguna de las {len(filas)} generaciones del holdout llego a completarse "
|
|
f"({errores} error(es) de request contra vLLM): no hay nada que medir"
|
|
)
|
|
tasa = con_prohibido / total
|
|
# Los errores de request bloquean la aprobacion a proposito: la puerta es el disparador de
|
|
# fallback antes de cuantizar, y un "aprueba" sobre generaciones faltantes no vale nada.
|
|
aprueba = tasa <= HOLDOUT_FORBIDDEN_RATE_MAX and errores == 0
|
|
|
|
print("\n=== Puerta 5 (modo holdout) -- tasa de API prohibida ===")
|
|
print(f" prompts medidos: {total} de {len(filas)}")
|
|
print(f" errores de request: {errores}")
|
|
print(f" con API prohibida: {con_prohibido}")
|
|
print(f" tasa: {100 * tasa:.1f}% (umbral <= {100 * HOLDOUT_FORBIDDEN_RATE_MAX:.0f}%)")
|
|
if patrones:
|
|
print(" desglose por patron:")
|
|
for patron, n in patrones.most_common():
|
|
print(f" - {patron}: {n}")
|
|
print(f"\n VEREDICTO: {'APRUEBA' if aprueba else 'NO APRUEBA -- disparador de fallback'}")
|
|
|
|
salida = REPO_ROOT / "data" / f"gate5_holdout_{cfg['GATE5_TAG']}.json"
|
|
salida.write_text(json.dumps({
|
|
"modo": "holdout",
|
|
"tag": cfg["GATE5_TAG"],
|
|
"modelo": cfg["GATE5_MODEL"],
|
|
"total": total,
|
|
"prompts_en_archivo": len(filas),
|
|
"errores_de_request": errores,
|
|
"con_api_prohibida": con_prohibido,
|
|
"tasa": tasa,
|
|
"umbral": HOLDOUT_FORBIDDEN_RATE_MAX,
|
|
"aprueba": aprueba,
|
|
"patrones": dict(patrones),
|
|
"resultados": resultados,
|
|
}, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
print(f" resultados en {salida.relative_to(REPO_ROOT)}")
|
|
return 0 if aprueba else 1
|
|
|
|
|
|
# ------------------------------------------------------------------------------------------
|
|
# Modo completo
|
|
# ------------------------------------------------------------------------------------------
|
|
def load_prompts():
|
|
filas = []
|
|
with open(PROMPTS_PATH, encoding="utf-8") as f:
|
|
for lineno, linea in enumerate(f, start=1):
|
|
linea = linea.strip()
|
|
if not linea:
|
|
continue
|
|
try:
|
|
filas.append(json.loads(linea))
|
|
except json.JSONDecodeError as e:
|
|
raise Gate5Error(f"{PROMPTS_PATH.name}:{lineno}: JSON invalido: {e}")
|
|
faltantes = [f["id"] for f in filas if f["id"] not in THRESHOLDS]
|
|
if faltantes:
|
|
raise Gate5Error(f"prompts sin umbrales declarados en THRESHOLDS: {faltantes}")
|
|
return filas
|
|
|
|
|
|
def resumen(resultados, tag):
|
|
scores = [r["score"] for r in resultados]
|
|
media = sum(scores) / len(scores) if scores else 0.0
|
|
veto_limpio = sum(1 for r in resultados if not r["veto_violado"])
|
|
sobre_minimo = sum(1 for r in resultados if r["score"] >= PROMPT_SCORE_MIN)
|
|
flagship = next((r for r in resultados if r["id"] == FLAGSHIP_PROMPT_ID), None)
|
|
flagship_score = flagship["score"] if flagship else 0.0
|
|
|
|
print(f"\n=== Puerta 5 -- calidad de diseno en Penpot (tag={tag}) ===\n")
|
|
print(f" {'prompt':32s} {'dif':6s} {'score':>6s} {'pasa':>7s} {'grises':>7s} "
|
|
f"{'prohib':>7s} {'shapes':>7s} {'colores':>8s} {'textos':>7s}")
|
|
for r in resultados:
|
|
crudas = r["metricas_crudas"]
|
|
print(f" {r['id']:32s} {str(r['dificultad'])[:6]:6s} {r['score']:6.1f} "
|
|
f"{r['pasadas']:3d}/{r['aplicables']:<3d} {crudas['placeholderGreys']:7d} "
|
|
f"{crudas['forbiddenBehavior']:7d} {crudas['shapeCount']:7d} "
|
|
f"{crudas['distinctFillColors']:8d} {crudas['textsWithCharacters']:7d}")
|
|
|
|
condiciones = [
|
|
(f"score medio >= {MEAN_SCORE_MIN:.0f}", media >= MEAN_SCORE_MIN, f"{media:.1f}"),
|
|
("veto limpio en 8/8", veto_limpio == len(resultados), f"{veto_limpio}/{len(resultados)}"),
|
|
(f">= {MIN_PROMPTS_OVER_MIN}/8 prompts >= {PROMPT_SCORE_MIN:.0f}",
|
|
sobre_minimo >= MIN_PROMPTS_OVER_MIN, f"{sobre_minimo}/{len(resultados)}"),
|
|
(f"landing de pizzeria >= {PROMPT_SCORE_MIN:.0f}",
|
|
flagship_score >= PROMPT_SCORE_MIN, f"{flagship_score:.1f}"),
|
|
]
|
|
print("\n condiciones de aprobacion:")
|
|
for etiqueta, ok, valor in condiciones:
|
|
print(f" [{'ok ' if ok else 'NO '}] {etiqueta:40s} -> {valor}")
|
|
|
|
aprueba = all(ok for _e, ok, _v in condiciones)
|
|
print(f"\n VEREDICTO GLOBAL: {'APRUEBA' if aprueba else 'NO APRUEBA'}")
|
|
return aprueba, {
|
|
"score_medio": round(media, 2),
|
|
"veto_limpio": veto_limpio,
|
|
"prompts_sobre_minimo": sobre_minimo,
|
|
"score_landing_pizzeria": flagship_score,
|
|
"aprueba": aprueba,
|
|
"condiciones": [{"condicion": e, "ok": ok, "valor": v} for e, ok, v in condiciones],
|
|
}
|
|
|
|
|
|
def run_full(cfg, tools, system_prompt):
|
|
prompts = load_prompts()
|
|
check_disjunto(prompts)
|
|
self_check_gate_payloads()
|
|
|
|
mcp = PenpotMCP(cfg["PENPOT_MCP_URL"], cfg["PENPOT_MCP_TOKEN"])
|
|
mcp.handshake()
|
|
# Sonda barata: si el plugin no esta conectado, abortar ANTES de quemar los 8 prompts.
|
|
sonda, _es_error = mcp.execute_code("return { gate5Probe: true };")
|
|
if PLUGIN_NOT_CONNECTED in sonda:
|
|
raise PluginNotConnected(sonda)
|
|
print(f"[MCP] sonda de conectividad del plugin: {sonda[:120]}")
|
|
|
|
png_dir = REPO_ROOT / "data" / f"gate5_png_{cfg['GATE5_TAG']}"
|
|
salida = REPO_ROOT / "data" / f"gate5_results_{cfg['GATE5_TAG']}.json"
|
|
|
|
def volcar(resultados_parciales, dt_parcial, veredicto_parcial=None, completo=False):
|
|
"""Escribe el JSON de resultados. Se llama DESPUES DE CADA PROMPT.
|
|
|
|
La primera corrida del baseline murio en el prompt 4 por una caida de conexion del MCP
|
|
y se perdio la medicion de los tres que ya habian terminado -- que era justamente el
|
|
dato caro, porque exige tener produccion arriba. Un fallo de transporte no puede costar
|
|
el trabajo ya hecho.
|
|
"""
|
|
salida.write_text(json.dumps({
|
|
"modo": "completo" if completo else "parcial",
|
|
"tag": cfg["GATE5_TAG"],
|
|
"modelo": cfg["GATE5_MODEL"],
|
|
"max_turnos": MAX_TURNS,
|
|
"max_tokens": MAX_TOKENS,
|
|
"timestamp": datetime.now(timezone.utc).isoformat(),
|
|
"duracion_s": round(dt_parcial, 1),
|
|
"prompts_completados": len(resultados_parciales),
|
|
"prompts_totales": len(prompts),
|
|
"metricas": [{"clave": k, "etiqueta": e, "comparador": c} for k, e, c in METRICS],
|
|
"veto": list(VETO_METRICS),
|
|
"umbrales": {pid: {k: (list(v) if isinstance(v, tuple) else v) for k, v in u.items()}
|
|
for pid, u in THRESHOLDS.items()},
|
|
"veredicto": veredicto_parcial,
|
|
"prompts": resultados_parciales,
|
|
}, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
|
|
t0 = time.time()
|
|
resultados = []
|
|
for fila in prompts:
|
|
try:
|
|
resultados.append(run_prompt(cfg, mcp, tools, system_prompt, fila, png_dir))
|
|
except PluginNotConnected:
|
|
volcar(resultados, time.time() - t0)
|
|
raise
|
|
except Gate5Error as e:
|
|
print(f"[ERROR] {fila['id']}: {e}")
|
|
print(f"[INFO] se guardan los {len(resultados)} prompt(s) ya medidos y se sigue con "
|
|
f"el siguiente; el JSON queda marcado como parcial")
|
|
resultados.append({"id": fila["id"], "error": str(e), "score": None})
|
|
volcar(resultados, time.time() - t0)
|
|
dt = time.time() - t0
|
|
|
|
aprueba, veredicto = resumen(resultados, cfg["GATE5_TAG"])
|
|
print(f"\n tiempo total: {dt / 60:.1f} min")
|
|
|
|
volcar(resultados, dt, veredicto_parcial=veredicto, completo=True)
|
|
print(f" resultados en {salida.relative_to(REPO_ROOT)}")
|
|
print(f" PNGs en {png_dir.relative_to(REPO_ROOT)}/ (la puerta no borra nada: las paginas "
|
|
f"quedan para inspeccion humana)")
|
|
return 0 if aprueba else 1
|
|
|
|
|
|
def main():
|
|
modo_holdout = os.environ.get("GATE5_HOLDOUT", "").strip() in ("1", "true", "yes")
|
|
try:
|
|
cfg = load_env(need_mcp=not modo_holdout)
|
|
tools = load_penpot_tools()
|
|
system_prompt = load_system_prompt()
|
|
if modo_holdout:
|
|
return run_holdout(cfg, tools, system_prompt)
|
|
return run_full(cfg, tools, system_prompt)
|
|
except PluginNotConnected as e:
|
|
print(
|
|
"\n[ABORTA] el servidor MCP de Penpot no tiene ningun plugin conectado.\n"
|
|
" Abri el archivo de Penpot en el navegador, arranca el plugin del MCP y espera a\n"
|
|
" que quede conectado; recien ahi volve a correr la puerta. Sin plugin conectado no\n"
|
|
" tiene sentido gastar los 8 prompts.\n"
|
|
f" Mensaje del servidor: {str(e)[:200]}",
|
|
file=sys.stderr,
|
|
)
|
|
return EXIT_PLUGIN_NOT_CONNECTED
|
|
except Gate5Error as e:
|
|
print(f"\n[ERROR] {e}", file=sys.stderr)
|
|
return 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|