aleleba/qwen3-6-lora
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
cc252d70f5a465c24ea50a354854d5b185efc2ba
qwen3-6-lora/scripts/35_gate5_penpot_design.py
T
aleleba cc252d70f5 Phase 6.4.25b: gate5 -- allow explicit ack of known train/prompt overlap (GATE5_KNOWN_OVERLAP_IDS) 
The 4 uploadMediaUrl reinforcement seeds added in the corrective round (134-137) share a
6-gram with g5-03-card-producto ('foto real traida de https picsum'). v2b-bf16 is already
trained on that text -- editing the seed corpus now cannot retroactively decontaminate it.
This lets the disjoint check acknowledge that specific, already-baked-in overlap explicitly
(never silently) instead of hard-aborting gate5 full mode entirely.
2026-08-04 05:22:32 +00:00

1980 lines
92 KiB
Python
Raw Blame History

"""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"
# GATE5_PROMPTS_PATH (Fase 6.4.25b): permite apuntar el mismo harness (17 metricas, veto,
# manejo de paginas/reintentos MCP) a un archivo de prompts distinto de los 10 oficiales de
# la puerta 5 -- p.ej. la sonda de generalizacion de rubros nunca vistos. La puerta 5 oficial
# (data/gate5_prompts.jsonl) queda intacta: sin la env var, el comportamiento es identico.
PROMPTS_PATH = Path(os.environ["GATE5_PROMPTS_PATH"]) if os.environ.get("GATE5_PROMPTS_PATH") \
else 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
# Rubros que usan los prompts de la puerta. Ningun seed puede compartirlos.
#
# Los shingles de 6-gramas NO atrapan esto: "necesito la home de una escuela de musica para
# chicos" (puerta) y "estoy redisenando el sitio de la escuela de musica" (seed) no comparten
# ningun 6-grama literal y sin embargo son el MISMO negocio. Entrenar sobre el dominio que
# despues se evalua infla el resultado sin que se note en ninguna metrica -- es la misma fuga
# que la de las palabras, un nivel mas arriba. Los sinonimos y las parafrasis del mismo rubro
# solo se atrapan nombrando el rubro.
RUBROS_DE_LA_PUERTA = [
"pizzeria", "pizzería",
"escuela de musica", "escuela de música",
"notaria", "notaría",
"trattoria",
]
# 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", "240"))
# Fallos que vuelven como respuesta HTTP EXITOSA con texto de error adentro. El limite de 30
# segundos NO es del cliente: es el del servidor MCP para una tarea del plugin, asi que no se
# puede subir desde aca -- solo reintentar. Y hay que reintentarlos explicitamente porque, al no
# ser errores de transporte, el reintento de `_request` no los ve: la request fue un exito.
# La auditoria es la llamada mas pesada de la puerta y es la que se cayo asi en el prompt
# critico de la fase.
REINTENTABLES_EN_RESULTADO = re.compile(
r"timed out after \d+ seconds|Tool execution failed: Error: Task ", re.I)
AUDIT_MAX_RETRIES = int(os.environ.get("GATE5_AUDIT_RETRIES", "5"))
# Vaciar la pagina de la puerta despues de medirla y exportar su PNG. Se puede apagar para
# inspeccion manual con GATE5_KEEP_PAGES=1.
LIMPIAR_PAGINAS = os.environ.get("GATE5_KEEP_PAGES", "").lower() not in ("1", "true", "yes")
# Tamano de lote. Ver el comentario del bucle en run_full(): el plugin se degrada tras 5 o 6
# prompts pesados, asi que 4 deja margen y da un punto de re-handshake antes del deterioro.
BATCH_SIZE = int(os.environ.get("GATE5_BATCH_SIZE", "4"))
# CORRIDA UNICA. El MCP de Penpot se cae seguido y cada ventana en que funciona es cara, asi que
# la corrida tiene que medir los 10 prompts de una y no rendirse por el fallo de uno.
# - Un prompt que falla NO aborta la corrida: se anota y se reintenta en una pasada posterior.
# - Tope por prompt, para que un cuelgue no coma la ventana entera.
# - Tope de pared, para que la corrida termine si o si y deje resultados.
MAX_PASADAS = int(os.environ.get("GATE5_MAX_PASSES", "3"))
PROMPT_TIMEOUT_S = int(os.environ.get("GATE5_PROMPT_TIMEOUT_S", "240"))
WALL_CLOCK_S = int(os.environ.get("GATE5_WALL_CLOCK_MIN", "45")) * 60
# 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
HEX_EN_TEXTO = re.compile(r"#[0-9a-fA-F]{6}\b")
PLUGIN_NOT_CONNECTED = "No Penpot plugin instances are currently connected"
EXIT_PLUGIN_NOT_CONNECTED = 2
# El plugin respondio pero se degrado a mitad de corrida (timeouts en createPage). No es lo mismo
# que "no esta conectado": hay que pedirle al usuario que lo RECARGUE y retomar los que faltan.
EXIT_PLUGIN_DEGRADADO = 3
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."""
class PromptTimeout(Gate5Error):
"""Se agoto el tope de tiempo de ESTE prompt. No dice nada del resto de la corrida."""
# ------------------------------------------------------------------------------------------
# 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 = None
for intento in range(1, AUDIT_MAX_RETRIES + 1):
texto, es_error = self.execute_code(code)
if not REINTENTABLES_EN_RESULTADO.search(texto or ""):
break
if intento == AUDIT_MAX_RETRIES:
break
espera = MCP_RETRY_BACKOFF * intento
print(f"[MCP] '{que}' fallo con timeout de tarea del servidor "
f"(intento {intento}/{AUDIT_MAX_RETRIES}). Reintento en {espera:.0f}s")
time.sleep(espera)
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,
// NO se devuelve penpot.root.id. `penpot.root` es el root de la pagina ACTIVA, y despues de
// `createPage` + `openPage` puede no haberse actualizado todavia: en la primera corrida con
// borrado de paginas activado devolvio "" en el segundo prompt. Ademas todas las paginas
// comparten el mismo id de root, asi que ese valor nunca sirvio para identificar nada. El
// pageId es lo unico unico y estable, y es lo que usa la auditoria.
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.
DELETE_PAGE_JS = """
// Borra la pagina que creo LA PUERTA, nunca una del usuario: se exige que el nombre empiece con
// el prefijo gate5/ y se compara el id, asi que no hay forma de que apunte a otra cosa.
const PAGE_ID = %PAGE_ID%;
const page = penpotUtils.getPageById(PAGE_ID);
if (!page) { return { borrada: false, motivo: "no existe" }; }
if (String(page.name).indexOf("gate5/") !== 0) {
return { borrada: false, motivo: "no es una pagina de la puerta: " + page.name };
}
const nombre = page.name;
const shapes = penpotUtils.findShapes(function () { return true; }, page.root);
for (let i = shapes.length - 1; i >= 0; i--) { shapes[i].remove(); }
return { borrada: true, nombre: nombre, shapesBorrados: shapes.length };
"""
AUDIT_JS = """
// El root se resuelve por PAGE_ID, nunca por rootId.
// Por que: en Penpot TODAS las paginas nuevas comparten el mismo id de root frame,
// "00000000-0000-0000-0000-000000000000". Y `penpotUtils.findShapeById` busca GLOBALMENTE, asi
// que buscar por ese id devuelve el root de la PRIMERA pagina del archivo -- la del usuario,
// vacia -- y no la que la puerta acaba de crear. Esa fue la causa de que el primer baseline
// midiera shapeCount=0 en los 10 prompts mientras el archivo real tenia 28 shapes y 20 textos.
// El pageId si es unico.
const PAGE_ID = %PAGE_ID%;
const page = penpotUtils.getPageById(PAGE_ID);
if (!page) { return { auditError: "pagina no encontrada: " + PAGE_ID }; }
const root = page.root;
if (!root) { return { auditError: "la pagina no tiene root" }; }
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));
}
}
// generateStyle con includeChildren y generateMarkup son, de lejos, lo mas caro de la auditoria:
// serializan el subarbol entero. El servidor MCP corta cualquier tarea del plugin a los 30
// segundos -- ese limite es SUYO, no del cliente, asi que no se puede subir desde aca; lo unico
// que se puede hacer es no acercarse. Por encima de MAX_NODOS_RENDER se saltea y se deja
// constancia, en vez de arriesgar que se caiga la auditoria ENTERA y con ella la medicion del
// prompt. `rendersOk` queda en null (no aplica) y no entra al denominador del score.
const MAX_NODOS_RENDER = 400;
let cssLen = 0, markupLen = 0, renderError = null, renderSalteado = false;
if (nodes.length > MAX_NODOS_RENDER) {
renderSalteado = true;
} else {
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); } }
// VIBRACION. Ni `distinctFillColors` ni `placeholderGreys` distinguen una paleta vibrante de
// una apagada pero tecnicamente no gris: seis tonos polvorientos pasan las dos. Esto mide
// saturacion de verdad, en HSL.
function aHsl(hex) {
const r = parseInt(hex.slice(1, 3), 16) / 255;
const g = parseInt(hex.slice(3, 5), 16) / 255;
const b = parseInt(hex.slice(5, 7), 16) / 255;
const mx = Math.max(r, g, b), mn = Math.min(r, g, b), d = mx - mn;
const l = (mx + mn) / 2;
let h = 0, sat = 0;
if (d !== 0) {
sat = d / (1 - Math.abs(2 * l - 1));
if (mx === r) { h = ((g - b) / d) % 6; }
else if (mx === g) { h = (b - r) / d + 2; }
else { h = (r - g) / d + 4; }
h = h * 60; if (h < 0) { h += 360; }
}
return { h: h, s: sat * 100, l: l * 100 };
}
// FILL EXPLICITO. Los tres defaults de Penpot: board blanco, rectangulo #B1B2B5, texto negro.
// Un diseno cuyos UNICOS colores son esos tres no es un diseno: son shapes que nunca recibieron
// fill. Esta metrica sola habria diagnosticado el bug de produccion, que las de "colores
// distintos" y "cero grises" no ven -- #FFFFFF y #000000 no son gris medio, asi que pasan.
const DEFAULTS = ["#FFFFFF", "#B1B2B5", "#000000"];
let conFill = 0, conFillExplicito = 0;
for (const n of nodes) {
const fs = Array.isArray(n.shape.fills) ? n.shape.fills : [];
const hexes = fs.filter(function (f) { return f && typeof f.fillColor === "string"; })
.map(function (f) { return f.fillColor.toUpperCase(); });
if (!hexes.length) { continue; }
conFill++;
if (hexes.some(function (h) { return DEFAULTS.indexOf(h) === -1; })) { conFillExplicito++; }
}
const soloDefaults = distintos.length > 0
&& distintos.every(function (h) { return DEFAULTS.indexOf(h) !== -1; });
const cromaticos = [];
const neutrales = [];
for (const hex of distintos) {
const c = aHsl(hex);
// Cromatico: saturacion >= 45 y luminosidad en 15..85. La banda de luminosidad excluye los
// casi-negros y casi-blancos, que pueden dar saturacion alta y sin embargo leerse neutros.
if (c.s >= 45 && c.l >= 15 && c.l <= 85) { cromaticos.push({ hex: hex, h: c.h, s: c.s, l: c.l }); }
else if (c.s <= 15) { neutrales.push(hex); }
}
// Estructura de paleta: una marca dominante, un acento con tono claramente distinto, y neutrales.
let separacionDeTono = 0;
for (let i = 0; i < cromaticos.length; i++) {
for (let j = i + 1; j < cromaticos.length; j++) {
let d = Math.abs(cromaticos[i].h - cromaticos[j].h);
if (d > 180) { d = 360 - d; }
if (d > separacionDeTono) { separacionDeTono = d; }
}
}
const saturacionMedia = cromaticos.length
? cromaticos.reduce(function (a, c) { return a + c.s; }, 0) / cromaticos.length : 0;
const resultado = {
explicitFillShare: conFill > 0 ? conFillExplicito / conFill : 0,
shapesWithFill: conFill,
shapesWithExplicitFill: conFillExplicito,
onlyDefaultColors: soloDefaults,
chromaticFills: cromaticos.length,
chromaticSamples: cromaticos.slice(0, 8).map(function (c) {
return { hex: c.hex, s: Math.round(c.s), l: Math.round(c.l) }; }),
meanChromaticSaturation: Math.round(saturacionMedia),
neutralFills: neutrales.length,
hueSeparation: Math.round(separacionDeTono),
paletteStructured: cromaticos.length >= 2 && neutrales.length >= 1 && separacionDeTono >= 30,
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,
renderSkipped: renderSalteado
};
// El servidor MCP valida el resultado contra un schema y rechaza la llamada ENTERA si algun
// campo no es serializable ("Invalid input: expected string, received function"). Un solo objeto
// de Penpot que se cuele en el return -- todos traen metodos -- tira abajo la auditoria completa
// y con ella la medicion de ese prompt. Forzar datos planos es mas barato y mas robusto que
// cazar el campo culpable cada vez que se agrega una metrica.
return JSON.parse(JSON.stringify(resultado));
"""
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(page_id):
return AUDIT_JS.replace("%PAGE_ID%", js_string(page_id))
def build_delete_page_js(page_id):
return DELETE_PAGE_JS.replace("%PAGE_ID%", js_string(page_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_rubros(prompts):
"""Ningun seed puede compartir el RUBRO de un prompt de la puerta. Ver RUBROS_DE_LA_PUERTA."""
import unicodedata as _ud
def plano(t):
t = _ud.normalize("NFKD", (t or "").lower())
return "".join(c for c in t if not _ud.combining(c))
rubros = sorted({plano(r) for r in RUBROS_DE_LA_PUERTA})
problemas = []
seeds_path = REPO_ROOT / "data" / "raw" / "seeds" / "penpot.jsonl"
if not seeds_path.exists():
print("[WARN] no hay corpus de seeds: se saltea el chequeo de rubros")
return
with open(seeds_path, encoding="utf-8") as f:
for lineno, linea in enumerate(f, start=1):
linea = linea.strip()
if not linea:
continue
texto = plano(linea)
for rubro in rubros:
if rubro in texto:
problemas.append(f"seed {lineno}: usa el rubro {rubro!r} de la puerta 5")
break
if problemas:
raise Gate5Error(
"hay seeds que comparten RUBRO con los prompts de la puerta 5 (fuga de dominio: los "
"shingles no la ven porque no hace falta compartir palabras para compartir negocio):"
"\n - " + "\n - ".join(problemas[:20])
)
print(f"[OK] ningun seed usa los rubros de la puerta ({', '.join(rubros)})")
# Hallazgo 6.4.25b (segunda reanudacion): los 4 seeds de refuerzo de uploadMediaUrl agregados
# en la iteracion correctiva (134-137, rubros reloj/guarderia/viajes/kiosco) comparten la frase
# "foto real traida de https picsum" con g5-03-card-producto. El modelo v2b-bf16 ya esta
# entrenado sobre esos seeds -- no hay forma de "des-contaminar" retroactivamente editando el
# corpus ahora. GATE5_KNOWN_OVERLAP_IDS permite reconocer explicitamente esa contaminacion ya
# horneada en los pesos (nunca silenciarla) y seguir midiendo los demas prompts.
KNOWN_OVERLAP_IDS = {x.strip() for x in os.environ.get("GATE5_KNOWN_OVERLAP_IDS", "").split(",") if x.strip()}
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:
if fila["id"] in KNOWN_OVERLAP_IDS:
print(f"[WARN] {fila['id']}: {len(solapados)} shingle(s) conocido(s) y aceptado "
f"via GATE5_KNOWN_OVERLAP_IDS, p.ej. {solapados[0]!r} -- resultado de este "
"prompt se interpreta con cautela, no como generalizacion limpia")
continue
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"),
("explicitFillShare", "share de shapes con fill NO-default", "min"),
("onlyDefaultColors", "la paleta son SOLO los 3 defaults de Penpot [VETO]", "max"),
("chromaticFills", "fills cromaticos (sat HSL >= 45, L 15-85)", "min"),
("meanChromaticSaturation", "saturacion media de los fills cromaticos", "min"),
("paletteStructured", "paleta estructurada (marca + acento + neutrales)", "bool"),
("finalMessageListsHex", "el mensaje final enumera los hex elegidos", "bool"),
("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", "onlyDefaultColors")
# 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.
# ------------------------------------------------------------------------------------------
# Umbrales de VIBRACION, solo para los prompts de brief ambiguo
# ------------------------------------------------------------------------------------------
# Requisito explicito del usuario: ante un brief ambiguo el modelo tiene que ELEGIR una paleta y
# justificarla, no preguntar ni caer en defaults. Ni `distinctFillColors` ni `placeholderGreys`
# distinguen una paleta vibrante de una apagada pero tecnicamente no gris.
#
# El umbral de saturacion 45 NO es una opinion: sale de medir las 325 paletas no neutrales del
# corpus escrito a mano de esta fase. Su saturacion HSL mediana es 75, con p25 en 48 y p10 en 35.
# Un piso de 45 queda apenas por debajo del primer cuartil -- lo pasa el 79% de esos fills -- asi
# que es un suelo que el comportamiento objetivo despeja comodo, no una aspiracion. La banda de
# luminosidad 15..85 excluye casi-negros y casi-blancos, que pueden dar saturacion alta y aun asi
# leerse neutros.
UMBRALES_VIBRACION = {
"chromaticFills": 2,
"meanChromaticSaturation": 45,
"paletteStructured": True,
"finalMessageListsHex": True,
}
UMBRALES_VIBRACION_NO_APLICA = {k: None for k in UMBRALES_VIBRACION}
# Fill explicito. `onlyDefaultColors` es VETO y aplica a TODOS los prompts: un diseno cuyos unicos
# colores son los tres defaults de Penpot es el bug de produccion, no un diseno sobrio.
#
# La distincion que pedia el usuario -- blanco y negro son legitimos cuando el modelo los ELIGE --
# se resuelve por CO-PRESENCIA, no penalizando el hex: #FFFFFF y #000000 solo son sospechosos si
# son lo UNICO que hay. Si conviven con colores de marca elegidos, `onlyDefaultColors` es false y
# ademas cuentan como neutrales legitimos en `paletteStructured`. Por eso son dos metricas y no
# una: `explicitFillShare` mide cuanto del diseno recibio color deliberado, y `onlyDefaultColors`
# atrapa el caso degenerado donde no recibio ninguno.
#
# El umbral de share se grada por dificultad: un boton primario (baja) tiene pocos shapes y algun
# neutral estructural pesa mucho en la proporcion; una landing (alta) no tiene excusa.
UMBRALES_FILL_EXPLICITO_BAJA = {"explicitFillShare": 0.50, "onlyDefaultColors": False}
UMBRALES_FILL_EXPLICITO_MEDIA = {"explicitFillShare": 0.65, "onlyDefaultColors": False}
UMBRALES_FILL_EXPLICITO = {"explicitFillShare": 0.80, "onlyDefaultColors": False}
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,
**UMBRALES_VIBRACION_NO_APLICA,
**UMBRALES_FILL_EXPLICITO_BAJA,
},
"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,
**UMBRALES_VIBRACION_NO_APLICA,
**UMBRALES_FILL_EXPLICITO_BAJA,
},
"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,
**UMBRALES_VIBRACION_NO_APLICA,
**UMBRALES_FILL_EXPLICITO_MEDIA,
},
"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,
**UMBRALES_VIBRACION_NO_APLICA,
**UMBRALES_FILL_EXPLICITO_MEDIA,
},
"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,
**UMBRALES_VIBRACION_NO_APLICA,
**UMBRALES_FILL_EXPLICITO_MEDIA,
},
"g5-09-brief-ambiguo-escuela": {
"shapeCount": 30, "distinctFillColors": 6, "placeholderGreys": 0,
"textsWithCharacters": 12, "typeScale": (3, 32), "textFillShare": 1.0,
"layoutBoards": 3, "layoutChildSizingShare": 0.5, "containmentViolations": 0,
"structure": (3, 4), "styleRichness": 3, "rootHeightMismatch": 48,
"exportBeforeFinal": True, "forbiddenBehavior": 0, "exceptionRate": 0.34,
"fontIdBoundShare": 1.0, "rendersOk": True,
**UMBRALES_VIBRACION,
**UMBRALES_FILL_EXPLICITO,
},
"g5-10-brief-ambiguo-notaria": {
"shapeCount": 30, "distinctFillColors": 5, "placeholderGreys": 0,
"textsWithCharacters": 12, "typeScale": (3, 32), "textFillShare": 1.0,
"layoutBoards": 3, "layoutChildSizingShare": 0.5, "containmentViolations": 0,
"structure": (3, 4), "styleRichness": 3, "rootHeightMismatch": 48,
"exportBeforeFinal": True, "forbiddenBehavior": 0, "exceptionRate": 0.34,
"fontIdBoundShare": 1.0, "rendersOk": True,
# "elegante" no pide estridencia: alcanza con UN cromatico bien elegido sobre neutrales,
# pero la paleta tiene que estar estructurada y declarada igual.
"chromaticFills": 1, "meanChromaticSaturation": 30,
"paletteStructured": True, "finalMessageListsHex": True,
**UMBRALES_FILL_EXPLICITO,
},
"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,
**UMBRALES_VIBRACION,
**UMBRALES_FILL_EXPLICITO,
},
"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,
**UMBRALES_VIBRACION_NO_APLICA,
**UMBRALES_FILL_EXPLICITO,
},
"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,
**UMBRALES_VIBRACION_NO_APLICA,
**UMBRALES_FILL_EXPLICITO,
},
}
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"],
"explicitFillShare": audit.get("explicitFillShare", 0.0),
# Sin shapes no hay paleta que juzgar, pero "no hice nada" no puede pasar el veto: se
# trata como el caso malo, igual que rootHeightMismatch.
"onlyDefaultColors": bool(audit.get("onlyDefaultColors", True)),
"chromaticFills": audit.get("chromaticFills", 0),
"meanChromaticSaturation": audit.get("meanChromaticSaturation", 0),
"paletteStructured": bool(audit.get("paletteStructured", False)),
"finalMessageListsHex": runtime["finalMessageListsHex"],
"fontIdBoundShare": audit.get("fontIdBoundShare", 0.0),
# Si la auditoria salteo el render por tamano, la metrica NO APLICA (None): no se puede
# afirmar que el arbol renderiza ni que no renderiza, y contarla como fallo castigaria
# justo a los disenos grandes, que es lo contrario de lo que la puerta quiere premiar.
"rendersOk": (None if audit.get("renderSkipped")
else 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)
# Una metrica NO APLICA si su umbral es None (el prompt no la exige) o si su VALOR es
# None (la auditoria no la pudo computar, p.ej. el render salteado por tamano). Contar
# "no se pudo medir" como fallo castigaria al diseno grande, que es lo contrario de lo
# que la puerta premia.
ok = None if raw[clave] is None else 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` COMPLETO del servidor MCP, verbatim.
Medir con otro system prompt mide otra cosa. Y esto se equivoco una vez, caro:
La primera corrida del baseline inyectaba solo la seccion inicial del documento (la que
`penpot_system_prompt.md` guarda como "Bloque completo, verbatim"), que son 2290 de los
16396 caracteres reales -- el 14%. El 86% que faltaba es justo el grounding de API:
"Core Shape Properties and Methods", "Layout Systems", "Text Elements" y "The `penpot` and
`penpotUtils` Objects", o sea insertChild, resize(), los layouts y penpotUtils.
Con ese recorte, produccion media shapes=0 en los 8 prompts: sin ninguna documentacion de
API, el modelo se inventaba una con forma de Figma y todas las llamadas lanzaban. Pero el
sintoma que el usuario reporto en su sesion real de Claude Code era otro y menos severo
("se conecta y crea cajas grises"), o sea shapes>0. Esa discrepancia era la firma de que el
harness, y no el modelo, estaba produciendo el resultado.
En una sesion real el servidor MCP entrega el documento entero como `instructions`, que es
tambien lo que devuelve `high_level_overview`. Asi que la fuente correcta es la Captura 1 de
penpot_api_docs.md, completa.
Nota deliberada sobre los seeds: los 34 seeds que llevan mensaje `system` cargan solo la
seccion inicial, no las 4554 tokens del documento entero -- no entrarian en el presupuesto
de 3300 tokens por ejemplo. Eso es aceptable porque la contramedida que los seeds tienen que
ensenar (la regla de "don't pick your own colours" y el razonamiento que la desambigua) vive
entera en esa seccion inicial.
"""
if os.environ.get("GATE5_SYSTEM_PROMPT_SECTION_ONLY", "").lower() in ("1", "true", "yes"):
print("[SYSTEM] usando SOLO la seccion inicial del documento (modo de diagnostico)")
return LINT.extract_system_block()
lineas = (SCHEMAS_DIR / "penpot_api_docs.md").read_text(encoding="utf-8").splitlines()
try:
inicio = next(n for n, l in enumerate(lineas) if l.strip() == "Salida verbatim:") + 3
fin = next(n for n in range(inicio + 5, len(lineas)) if lineas[n].rstrip() == "```")
except StopIteration:
raise Gate5Error(
"no se pudo extraer la Captura 1 (high_level_overview) de penpot_api_docs.md; "
"ese bloque es el system prompt real del servidor y sin el la medicion no es fiel"
)
completo = "\n".join(lineas[inicio:fin])
# La captura viene de `high_level_overview`, que cierra con un separador y la frase
# "You have hereby read the 'Penpot High-Level Overview' and need not use a tool to read it
# again." Eso es framing de la RESPUESTA DE LA TOOL, no parte del bloque `instructions` que
# el servidor inyecta. Dejarlo adentro le estaria diciendo al modelo que ya leyo un documento
# por haberlo recibido, lo que altera su decision de llamar o no a la tool.
corte = completo.rfind("\n--\nYou have hereby read")
if corte != -1:
completo = completo[:corte]
seccion = LINT.extract_system_block()
if not completo.startswith(seccion[:200]):
raise Gate5Error(
"la Captura 1 de penpot_api_docs.md no arranca como el bloque de "
"penpot_system_prompt.md: uno de los dos archivos se desincronizo"
)
print(f"[SYSTEM] documento completo del servidor: {len(completo)} chars "
f"({len(completo.splitlines())} lineas)")
return completo
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, deadline=None):
prompt_id = fila["id"]
def queda_tiempo(que):
if deadline is not None and time.time() > deadline:
raise PromptTimeout(
f"{prompt_id}: se agoto el tope de {PROMPT_TIMEOUT_S}s en '{que}'")
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")
page_id = setup.get("pageId")
if not page_id:
raise Gate5Error(f"{prompt_id}: el setup no devolvio pageId: {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):
queda_tiempo(f"turno {turno}")
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(page_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
# Sin mainBoardId no hay nada util que exportar: el root de la pagina no se puede exportar
# por id (todas las paginas comparten el mismo), asi que se omite el PNG en vez de exportar
# el arbol equivocado.
shape_a_exportar = audit.get("mainBoardId")
if not shape_a_exportar:
print(f"[WARN] {prompt_id}: la auditoria no encontro un board principal; no hay PNG")
try:
if not shape_a_exportar:
raise Gate5Error("sin board principal que exportar")
_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})")
# --- vaciar la pagina de la puerta ---------------------------------------------------------
# La evidencia que queda es el PNG + las metricas del JSON, no la pagina viva. Dejarlas
# acumular tiene un costo medible: el plugin de Penpot se degrada de forma progresiva a
# medida que crece el archivo -- en tres corridas seguidas aguanto 5 o 6 prompts pesados y
# despues empezo a dar timeouts de 30 s en `createPage`. Con 26 paginas acumuladas eso deja
# de ser flakiness del MCP y pasa a ser un problema que la propia puerta se fabrica.
# Ademas le deja el archivo limpio al usuario.
if LIMPIAR_PAGINAS and png_path is not None:
try:
info = mcp.execute_json(build_delete_page_js(page_id),
f"{prompt_id}: limpieza de la pagina")
if info.get("borrada"):
print(f"[CLEAN] {info['nombre']}: {info['shapesBorrados']} shape(s) borrados "
f"(la evidencia queda en el PNG y en el JSON)")
else:
print(f"[WARN] {prompt_id}: no se limpio la pagina ({info.get('motivo')})")
except Gate5Error as e:
print(f"[WARN] {prompt_id}: no se pudo limpiar la pagina ({e})")
elif png_path is None:
print(f"[CLEAN] {prompt_id}: la pagina se CONSERVA porque no hubo PNG, para poder "
f"inspeccionar que paso")
# Ante un brief ambiguo el modelo tiene que ELEGIR una paleta y DECIRSELA al usuario, para
# que la pueda ajustar. Una paleta elegida en silencio no es colaborable.
hex_en_final = set(HEX_EN_TEXTO.findall(final_content or ""))
runtime = {
"exportBeforeFinal": bool(export_calls),
"forbiddenBehavior": len(veto_hits),
"exceptionRate": (exec_exceptions / exec_calls) if exec_calls else 0.0,
"finalMessageListsHex": len(hex_en_final) >= 3,
}
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,
"pageId_auditado": page_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, esperados=None):
"""Resume la corrida.
Un prompt SIN medicion no cuenta como 0 en la media. "No pude medir" y "el modelo lo hizo
mal" son dos cosas distintas y promediarlas juntas produce un numero que no significa nada:
un corte del plugin en el prompt 6 abarataria artificialmente el promedio y despues se
compararia contra el modelo nuevo como si fuera calidad. Los no medidos se reportan aparte y
**invalidan el veredicto**: un baseline con 5 de 10 medidos no es un baseline.
"""
medidos = [r for r in resultados if r.get("score") is not None]
no_medidos = [r for r in resultados if r.get("score") is None]
scores = [r["score"] for r in medidos]
media = sum(scores) / len(scores) if scores else 0.0
veto_limpio = sum(1 for r in medidos if not r["veto_violado"])
sobre_minimo = sum(1 for r in medidos if r["score"] >= PROMPT_SCORE_MIN)
flagship = next((r for r in medidos if r["id"] == FLAGSHIP_PROMPT_ID), None)
flagship_score = flagship["score"] if flagship else 0.0
total_esperado = esperados if esperados is not None else len(resultados)
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 medidos:
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}")
if no_medidos:
print(f"\n NO MEDIDOS ({len(no_medidos)}), no cuentan como 0 en la media:")
for r in no_medidos:
print(f" {r['id']:32s} {str(r.get('error'))[:110]}")
condiciones = [
("todos los prompts medidos", len(medidos) == total_esperado,
f"{len(medidos)}/{total_esperado}"),
(f"score medio >= {MEAN_SCORE_MIN:.0f}", media >= MEAN_SCORE_MIN, f"{media:.1f}"),
(f"veto limpio en {total_esperado}/{total_esperado}", veto_limpio == total_esperado,
f"{veto_limpio}/{total_esperado}"),
(f">= {MIN_PROMPTS_OVER_MIN}/{total_esperado} prompts >= {PROMPT_SCORE_MIN:.0f}",
sobre_minimo >= MIN_PROMPTS_OVER_MIN, f"{sobre_minimo}/{total_esperado}"),
(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)
if no_medidos:
print(f"\n VEREDICTO GLOBAL: INVALIDO -- faltan {len(no_medidos)} prompt(s) por medir. "
f"Re-medilos con GATE5_ONLY={','.join(r['id'] for r in no_medidos)}")
else:
print(f"\n VEREDICTO GLOBAL: {'APRUEBA' if aprueba else 'NO APRUEBA'}")
return aprueba, {
"valido": not no_medidos,
"no_medidos": [r["id"] for r in no_medidos],
"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()
todos_los_prompts = list(prompts)
check_disjunto(prompts)
check_rubros(prompts)
# Re-medir un subconjunto sin quemar los 10. Existe porque el MCP de Penpot se cae: cuando
# una corrida pierde 5 prompts por un corte de transporte, repetir los 10 desperdicia media
# hora de produccion arriba Y descarta mediciones que eran validas.
solo = [x.strip() for x in os.environ.get("GATE5_ONLY", "").split(",") if x.strip()]
previos = []
if solo:
conocidos = {f["id"] for f in prompts}
desconocidos = [x for x in solo if x not in conocidos]
if desconocidos:
raise Gate5Error(f"GATE5_ONLY nombra prompts que no existen: {desconocidos}")
# Los resultados ya medidos que NO se vuelven a correr se conservan y se fusionan.
anterior = REPO_ROOT / "data" / f"gate5_results_{cfg['GATE5_TAG']}.json"
if anterior.exists():
try:
previos = [r for r in json.loads(anterior.read_text(encoding="utf-8")).get("prompts", [])
if r.get("id") not in solo and r.get("score") is not None]
print(f"[INFO] se conservan {len(previos)} prompt(s) ya medidos de "
f"{anterior.name}: {[r['id'] for r in previos]}")
except (json.JSONDecodeError, OSError) as e:
print(f"[WARN] no se pudo leer el resultado anterior ({e}); se re-mide todo lo pedido")
prompts = [f for f in prompts if f["id"] in solo]
print(f"[INFO] GATE5_ONLY: se miden {len(prompts)} prompt(s): {[f['id'] for f in prompts]}")
self_check_gate_payloads()
mcp = PenpotMCP(cfg["PENPOT_MCP_URL"], cfg["PENPOT_MCP_TOKEN"])
mcp.handshake()
# Sonda barata. Ya NO aborta: la corrida hace hasta MAX_PASADAS pasadas con handshake limpio,
# asi que un plugin que no responde ahora puede responder en la pasada 2. Abortar aca
# desperdiciaria la ventana entera por un estado transitorio.
try:
sonda, _es_error = mcp.execute_code("return { gate5Probe: true };")
if PLUGIN_NOT_CONNECTED in sonda:
print("[MCP] AVISO: el plugin no figura conectado. Se intenta igual: las pasadas "
"posteriores rehacen el handshake.")
else:
print(f"[MCP] sonda de conectividad del plugin: {sonda[:120]}")
except Gate5Error as e:
print(f"[MCP] AVISO: la sonda fallo ({str(e)[:140]}). Se intenta igual.")
png_dir = REPO_ROOT / "data" / f"gate5_png_{cfg['GATE5_TAG']}"
salida = REPO_ROOT / "data" / f"gate5_results_{cfg['GATE5_TAG']}.json"
def fusionar(resultados_parciales):
"""Los recien medidos + los conservados, en el orden canonico del .jsonl."""
por_id = {r["id"]: r for r in previos}
por_id.update({r["id"]: r for r in resultados_parciales})
orden = [f["id"] for f in todos_los_prompts]
return [por_id[i] for i in orden if i in por_id]
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(fusionar(resultados_parciales)),
"prompts_totales": len(todos_los_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": fusionar(resultados_parciales),
}, ensure_ascii=False, indent=2), encoding="utf-8")
t0 = time.time()
fin_de_pared = t0 + WALL_CLOCK_S
medidos = {}
ultimo_error = {}
# CORRIDA UNICA, EN PASADAS. El MCP de Penpot se cae seguido y cada ventana en que funciona
# es cara, asi que la corrida no puede rendirse por el fallo de uno: recorre todos los
# prompts, anota los que fallaron, y los reintenta en la pasada siguiente con un handshake
# limpio en el medio. Los cortes anteriores nos costaron tres ventanas del usuario.
por_medir = list(prompts)
for pasada in range(1, MAX_PASADAS + 1):
if not por_medir:
break
if time.time() > fin_de_pared:
print(f"\n[PARED] se agoto el tope de {WALL_CLOCK_S // 60} min; quedan "
f"{len(por_medir)} sin medir")
break
if pasada > 1:
print(f"\n[PASADA {pasada}/{MAX_PASADAS}] handshake limpio y reintento de "
f"{len(por_medir)}: {[f['id'] for f in por_medir]}")
mcp.session_id = None
try:
mcp.handshake()
sonda, _ = mcp.execute_code("return 1;")
if PLUGIN_NOT_CONNECTED in sonda:
print("[PASADA] el plugin no esta conectado; no tiene sentido reintentar")
break
except Gate5Error as e:
print(f"[PASADA] el re-handshake fallo ({e}); se corta")
break
else:
print(f"\n[PASADA 1/{MAX_PASADAS}] {len(por_medir)} prompt(s)")
fallaron = []
for fila in por_medir:
if time.time() > fin_de_pared:
print(f"[PARED] tope alcanzado; {fila['id']} y los siguientes quedan sin medir")
fallaron.append(fila)
continue
deadline = min(time.time() + PROMPT_TIMEOUT_S, fin_de_pared)
try:
medidos[fila["id"]] = run_prompt(cfg, mcp, tools, system_prompt, fila,
png_dir, deadline=deadline)
except PluginNotConnected as e:
# Sin plugin no hay nada que medir en NINGUN prompt: se corta la pasada, pero la
# corrida NO aborta -- lo ya medido queda en disco y la pasada siguiente
# reintenta con handshake limpio.
print(f"[ERROR] {fila['id']}: plugin no conectado ({str(e)[:90]})")
ultimo_error[fila["id"]] = f"PluginNotConnected: {str(e)[:200]}"
fallaron.append(fila)
fallaron.extend(f for f in por_medir[por_medir.index(fila) + 1:])
break
except Gate5Error as e:
# Un fallo de un prompt NO aborta la corrida ni quema el prompt: se reintenta.
clase = "timeout del prompt" if isinstance(e, PromptTimeout) else "fallo"
print(f"[ERROR] {fila['id']}: {clase} -> {str(e)[:220]}")
ultimo_error[fila["id"]] = str(e)[:400]
fallaron.append(fila)
except Exception as e: # noqa: BLE001 - una excepcion inesperada tampoco puede
print(f"[ERROR] {fila['id']}: excepcion inesperada {type(e).__name__}: "
f"{str(e)[:200]}") # tirar abajo los 9 prompts restantes
ultimo_error[fila["id"]] = f"{type(e).__name__}: {str(e)[:300]}"
fallaron.append(fila)
# Linea de progreso legible desde afuera, sin parsear el JSON.
r = medidos.get(fila["id"])
if r:
c = r.get("metricas_crudas") or {}
print(f"PROGRESO {len(medidos)} de {len(todos_los_prompts)}, {fila['id']}, "
f"score {r.get('score')}, shapes {c.get('shapeCount')}, "
f"colores {c.get('distinctFillColors')}", flush=True)
volcar(list(medidos.values()), time.time() - t0)
por_medir = fallaron
resultados = list(medidos.values())
for fila in por_medir:
resultados.append({"id": fila["id"], "score": None,
"error": ultimo_error.get(fila["id"], "no medido")})
pendientes = [f["id"] for f in por_medir]
dt = time.time() - t0
aprueba, veredicto = resumen(fusionar(resultados), cfg["GATE5_TAG"],
esperados=len(todos_los_prompts))
print(f"\n tiempo total: {dt / 60:.1f} min")
volcar(resultados, dt, veredicto_parcial=veredicto,
completo=not pendientes)
# Inventario de las paginas gate5 que quedaron de corridas ANTERIORES. No se borran solas:
# es el archivo del usuario y la decision es suya. Se listan para poder ofrecerselo.
try:
inv = mcp.execute_json(
"const ps = penpotUtils.getPages().filter(function(p){"
" return p.name && p.name.indexOf('gate5/') === 0; });"
"return { total: ps.length, nombres: ps.slice(0, 60).map(function(p){ return p.name; }) };",
"inventario de paginas gate5")
print(f"\n[PAGINAS] el archivo tiene {inv.get('total')} pagina(s) gate5/ acumuladas de "
f"esta corrida y de anteriores.")
print(" Las de ESTA corrida ya quedaron vaciadas tras exportar su PNG. Las viejas se "
"dejan como estan: es el archivo del usuario y se le ofrece borrarlas en el C1.")
(REPO_ROOT / "data" / f"gate5_paginas_{cfg['GATE5_TAG']}.json").write_text(
json.dumps(inv, ensure_ascii=False, indent=2), encoding="utf-8")
except Gate5Error as e:
print(f"[WARN] no se pudo inventariar las paginas gate5 ({e})")
sin_medir = sorted(set(pendientes) |
{r["id"] for r in fusionar(resultados) if r.get("score") is None})
if sin_medir:
# NO culpar al plugin por defecto. Un mensaje que manda al usuario a recargar el navegador
# cuando la causa es un bug de la puerta cuesta tiempo de las dos partes, y ya paso una
# vez: el setup fallo por depender de `penpot.root` y el script igual imprimio "plugin
# degradado". Se clasifica por la evidencia del error, no por el hecho de que falte algo.
errores = " ".join(str(r.get("error", "")) for r in fusionar(resultados)
if r.get("score") is None)
parece_plugin = bool(re.search(
r"timed out after \d+ seconds|No Penpot plugin instances|ConnectionError|"
r"error de transporte", errores, re.I))
print(f"\n[INCOMPLETO] {len(sin_medir)} prompt(s) no se pudieron medir.")
if parece_plugin:
print(" Causa probable: el plugin de Penpot. Los errores son timeouts de tarea o "
"cortes de transporte.")
print(" Pedile al usuario que RECARGUE el plugin en el navegador y retoma con:")
else:
print(" Causa probable: un bug de la PUERTA, no del plugin -- los errores no son "
"timeouts ni cortes de transporte. Revisar el mensaje de error antes de "
"pedirle nada al usuario.")
print(" Cuando este arreglado, retoma con:")
print(f" GATE5_ONLY={','.join(sin_medir)}")
return EXIT_PLUGIN_DEGRADADO if parece_plugin else 1
print(f" resultados en {salida.relative_to(REPO_ROOT)}")
print(f" PNGs en {png_dir.relative_to(REPO_ROOT)}/ (la pagina de cada prompt se vacia tras "
f"exportar su PNG; solo se conserva la de un prompt cuyo PNG haya fallado)")
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())
Reference in New Issue View Git Blame Copy Permalink