From 8ea4572edd62839f8fbf27f6ddf3c13aed410fe9 Mon Sep 17 00:00:00 2001 From: Alejandro Lembke Barrientos Date: Thu, 30 Jul 2026 17:18:01 +0000 Subject: [PATCH] Phase 6.3: add gate 5, design quality in Penpot MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Unlike gates 2-4 this one needs a real agent loop - model, tool call, live MCP, result, up to 14 turns - because design quality only exists after the code executes. It talks to vLLM over the OpenAI API and to the Penpot MCP over HTTP (initialize, notifications/initialized, tools/list, tools/call), handling both application/json and text/event-stream responses. Endpoints and credentials come from env with no defaults and are never printed or stored; requests errors are reduced to the exception type because the requests message embeds the URL. Eight graded prompts, a fresh page per prompt named gate5///, and the gate never deletes anything. The 17 metrics are computed by an audit payload the gate injects, not the model. placeholderGreys and forbidden behaviour are veto metrics: any hit scores that prompt 0. Two things worth calling out. The forbidden-pattern regexes are imported from 07_lint_penpot_code.py rather than duplicated, and the gate runs those same regexes over its own setup and audit payloads at startup - a gate that violated the API it is grading would be measuring its own bug. The holdout mode had a silent failure that is exactly the kind this phase exists to catch: with the endpoint down, all 60 generations failed, each entered the denominator with zero forbidden patterns found, and the gate reported 0% forbidden API and APPROVED. Since that number is the fallback trigger, a false pass there would have launched the quantization run. Request errors are now counted separately, never enter the denominator, and block approval outright. Known issue, resolved separately: gate prompt 6 is the exact production failure ("hazme una landing page de una pizzería con colores vibrantes"), and the flagship B6 seed was written to the same wording. Shingle overlap measures 20%, under the 34% threshold, but the seed prompt is a literal substring of the gate prompt - the threshold is too loose for prompts this short. The seed's domain gets changed rather than the gate's, so gate 5 measures transfer instead of memorisation; the real pizzeria prompt still runs in the human acceptance test, which is the criterion that decides. --- data/gate5_prompts.jsonl | 8 + scripts/35_gate5_penpot_design.py | 1285 +++++++++++++++++++++++++++++ 2 files changed, 1293 insertions(+) create mode 100644 data/gate5_prompts.jsonl create mode 100644 scripts/35_gate5_penpot_design.py diff --git a/data/gate5_prompts.jsonl b/data/gate5_prompts.jsonl new file mode 100644 index 0000000..78fed99 --- /dev/null +++ b/data/gate5_prompts.jsonl @@ -0,0 +1,8 @@ +{"id": "g5-01-boton-primario", "dificultad": "baja", "prompt": "Diseñá un botón primario que un diseñador aprobaría: fondo con un color de marca elegido por vos, esquinas redondeadas, una sombra sutil y la etiqueta «Reservar mesa» centrada con su propio peso y tamaño. Que se vea terminado, no un rectángulo con letras encima."} +{"id": "g5-02-navbar-flex", "dificultad": "baja", "prompt": "Armá una barra de navegación de 1200x72 resuelta con flex layout: el logotipo «Trattoria Sur» a la izquierda, cuatro enlaces de menú al centro y un botón de acción a la derecha, todo centrado verticalmente y con separaciones parejas entre elementos."} +{"id": "g5-03-card-producto", "dificultad": "media", "prompt": "Hacé una card de producto de 320 de ancho con una foto real traída de https://picsum.photos/seed/espresso/640/480, un título, una bajada de dos líneas, el precio destacado y un botón de compra. La foto tiene que ser la imagen de verdad: nada de rectángulos grises simulando la imagen."} +{"id": "g5-04-tokens-biblioteca", "dificultad": "media", "prompt": "Creá en la biblioteca local del archivo una paleta de 6 colores nombrados por rol y una escala tipográfica de 5 pasos, y después aplicá esos tokens a una muestra visible: un encabezado, un párrafo de apoyo y tres chips que muestren los colores aplicados desde la biblioteca."} +{"id": "g5-05-grid-tres-diferenciales", "dificultad": "media", "prompt": "Componé una sección de 3 diferenciales de 1140 de ancho resuelta con un grid de 3 columnas: cada columna con una figura geométrica propia a modo de ícono, un titular corto y dos líneas de texto de apoyo, todo con la misma alineación y respiración entre columnas."} +{"id": "g5-06-landing-pizzeria", "dificultad": "alta", "prompt": "hazme una landing page de una pizzería napolitana con colores vibrantes, de 1200 de ancho"} +{"id": "g5-07-onboarding-mobile", "dificultad": "alta", "prompt": "Diseñá una pantalla mobile de onboarding de 375x812 para una app de reservas: ilustración o figura de bienvenida arriba, titular, párrafo explicativo, indicador de tres pasos y dos acciones al pie (una primaria y una secundaria)."} +{"id": "g5-08-reparacion-grises", "dificultad": "alta", "prompt": "Este board está hecho enteramente de rectángulos grises y no tiene una sola línea de texto. Auditalo, definí una paleta y una escala tipográfica, y rehacelo hasta que sea una pantalla de verdad, con contenido legible y jerarquía visual.", "seed_board": {"nombre": "Wireframe sin disenar", "x": 0, "y": 0, "ancho": 960, "alto": 640, "gris": "#B1B2B5", "rectangulos": [{"nombre": "header", "x": 32, "y": 32, "ancho": 896, "alto": 88}, {"nombre": "hero", "x": 32, "y": 144, "ancho": 896, "alto": 208}, {"nombre": "card-1", "x": 32, "y": 376, "ancho": 288, "alto": 160}, {"nombre": "card-2", "x": 336, "y": 376, "ancho": 288, "alto": 160}, {"nombre": "card-3", "x": 640, "y": 376, "ancho": 288, "alto": 160}, {"nombre": "footer", "x": 32, "y": 560, "ancho": 896, "alto": 48}]}} diff --git a/scripts/35_gate5_penpot_design.py b/scripts/35_gate5_penpot_design.py new file mode 100644 index 0000000..5bf1c36 --- /dev/null +++ b/scripts/35_gate5_penpot_design.py @@ -0,0 +1,1285 @@ +"""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///`) +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 + +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._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): + self._next_id += 1 + payload = {"jsonrpc": "2.0", "id": self._next_id, "method": method} + if params is not None: + payload["params"] = params + return self._post(payload) + + 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): + 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.""" + texto, es_error = self.execute_code(code) + try: + return json.loads(texto) + except json.JSONDecodeError: + raise Gate5Error(f"{que}: el MCP no devolvio JSON ({'error' if es_error else 'texto'}): {texto[:300]}") + + +# ------------------------------------------------------------------------------------------ +# 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']}" + t0 = time.time() + resultados = [run_prompt(cfg, mcp, tools, system_prompt, fila, png_dir) for fila in prompts] + dt = time.time() - t0 + + aprueba, veredicto = resumen(resultados, cfg["GATE5_TAG"]) + print(f"\n tiempo total: {dt / 60:.1f} min") + + salida = REPO_ROOT / "data" / f"gate5_results_{cfg['GATE5_TAG']}.json" + salida.write_text(json.dumps({ + "modo": "completo", + "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, 1), + "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, + "prompts": resultados, + }, ensure_ascii=False, indent=2), encoding="utf-8") + 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())