aleleba/qwen3-6-lora
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
8094183940ff5678bd1c629baf465605bf7b9331
qwen3-6-lora/scripts/35_gate5_penpot_design.py
T
aleleba 8094183940 Phase 6.3.17: fix a harness fidelity bug and add a vibrancy metric 
The first baseline measured production with only 2290 of the 16392
characters of the server's instructions block - 14%. The missing 86% is
exactly the API grounding: Core Shape Properties and Methods, Layout
Systems, Text Elements, and The penpot and penpotUtils Objects, which is
where insertChild, resize(), the layouts and penpotUtils are documented.

That was worth catching, because the discrepancy had a visible signature:
the measurement said production creates nothing, while the user's real
Claude Code session produced grey boxes, i.e. shapes greater than zero.
When a harness and reality disagree, the harness is the first suspect. In
phase 5 a low max_tokens manufactured an apparent regression the same way.

The gate now injects the full document, minus the trailing "You have hereby
read the Penpot High-Level Overview" line, which is framing of the tool
response rather than part of the instructions block and would otherwise
tell the model it had already read something.

The finding survives the fix. Across the five prompts measured cleanly
under the corrected condition, shapeCount is still zero on every one. So
the API invention is not an artefact of withholding documentation from the
model - it happens with the documentation present.

Also adds the vibrancy requirement the user raised as first-class scope:
given an ambiguous brief the model must choose and justify a palette rather
than ask or fall back to defaults. Neither distinctFillColors nor
placeholderGreys distinguishes a vibrant palette from a muted but
technically non-grey one, so four metrics are added: chromaticFills,
meanChromaticSaturation, paletteStructured (a dominant brand hue, an accent
at least 30 degrees away, and neutrals), and finalMessageListsHex, because
a palette chosen in silence cannot be adjusted by the user.

The saturation floor of 45 is derived, not asserted: measured over the 325
non-neutral fills of this phase's hand-authored corpus, median HSL
saturation is 75, p25 is 48 and p10 is 35. A floor of 45 sits just under
the first quartile and is cleared by 79% of those fills, so it is a floor
the target behaviour already clears rather than an aspiration. The
lightness band of 15 to 85 excludes near-blacks and near-whites, which can
compute as highly saturated while reading as neutral.

Gate prompt 6 becomes the user's literal failing sentence, and two
ambiguous-brief prompts are added. One of them had to be re-domained after
the disjointness check found it shared a 6-gram with a seed - the check
fails on a single shared shingle, which is what makes it useful.

Results so far are partial: prompts 1-5 measured cleanly, 6 has a timed-out
audit and 7-10 hit the MCP outage, so those get re-measured. Both runs are
kept, the 14% one renamed to record what it was.
2026-07-30 18:38:24 +00:00

1548 lines
67 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"
PROMPTS_PATH = REPO_ROOT / "data" / "gate5_prompts.jsonl"
HOLDOUT_PATH = REPO_ROOT / "data" / "holdout_penpot_design.jsonl"
PENPOT_TOOLS_PATH = SCHEMAS_DIR / "penpot.json"
SYSTEM_PROMPT_PATH = SCHEMAS_DIR / "penpot_system_prompt.md"
LINT_PATH = REPO_ROOT / "scripts" / "07_lint_penpot_code.py"
# Corpus contra el que los 8 prompts tienen que ser disjuntos (shingles de 6-gramas).
DISJOINT_AGAINST = [
REPO_ROOT / "data" / "train.jsonl",
REPO_ROOT / "data" / "eval.jsonl",
REPO_ROOT / "data" / "raw" / "seeds" / "penpot.jsonl",
]
SHINGLE_N = 6
# Reintento de transporte contra el MCP de Penpot. El servidor corta la conexion a mitad de
# llamada de forma intermitente; ya se habia visto en la puerta 4 de la Fase 4.
MCP_MAX_RETRIES = int(os.environ.get("PENPOT_MCP_RETRIES", "4"))
MCP_RETRY_BACKOFF = float(os.environ.get("PENPOT_MCP_RETRY_BACKOFF", "3"))
MAX_TURNS = int(os.environ.get("GATE5_MAX_TURNS", "14"))
MAX_TOKENS = int(os.environ.get("GATE5_MAX_TOKENS", "4096"))
HTTP_TIMEOUT = int(os.environ.get("GATE5_HTTP_TIMEOUT", "600"))
MCP_TIMEOUT = int(os.environ.get("GATE5_MCP_TIMEOUT", "180"))
# El prompt que reproduce el fallo exacto de produccion. Tiene su propia condicion de
# aprobacion: si este no llega a 60, la puerta no aprueba aunque el promedio alcance.
FLAGSHIP_PROMPT_ID = "g5-06-landing-pizzeria"
MEAN_SCORE_MIN = 65.0
PROMPT_SCORE_MIN = 60.0
MIN_PROMPTS_OVER_MIN = 6
HOLDOUT_FORBIDDEN_RATE_MAX = 0.05
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
class Gate5Error(RuntimeError):
"""Fallo de la puerta que no es un fallo del modelo (env, transporte, MCP)."""
class PluginNotConnected(Gate5Error):
"""El plugin de Penpot no esta abierto/conectado: no tiene sentido seguir."""
# ------------------------------------------------------------------------------------------
# Patrones prohibidos: se IMPORTAN del lint, nunca se duplican
# ------------------------------------------------------------------------------------------
def load_lint_module():
"""Importa scripts/07_lint_penpot_code.py (el nombre empieza con digito, no es importable
con `import`). La fuente de verdad de los regex prohibidos tiene que ser una sola."""
spec = importlib.util.spec_from_file_location("lint_penpot_code", LINT_PATH)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
LINT = load_lint_module()
# El subconjunto de FORBIDDEN que constituye "comportamiento prohibido" a efectos de VETO,
# identificado por el nombre exacto con el que el lint lo declara. El resto de los patrones del
# lint se sigue registrando en el JSON de resultados (columna `otros_patrones`) porque es
# informacion util, pero no veta: la lista de veto es la del PLAN, ni mas ni menos.
VETO_PATTERN_NAMES = {
"findShapeById con 2 argumentos",
"propiedad .layout inexistente",
"flex.appendChild",
"fontSize/fontWeight/lineHeight/letterSpacing numerico",
"import_image / importImage / createImage / filePath",
}
# Argumentos que el schema de export_shape NO declara. Inventarlos es el mismo comportamiento
# prohibido, solo que del lado de los argumentos de la tool y no del texto del `code`.
EXPORT_SHAPE_ALLOWED_ARGS = {"shapeId", "format", "mode"}
def scan_forbidden(code):
"""Devuelve (veto_hits, otros_hits) sobre un payload de `code`."""
veto, otros = [], []
for pname, rx, _why in LINT.FORBIDDEN:
match = rx.search(code)
if not match:
continue
frag = code[max(0, match.start() - 40):match.end() + 40].replace("\n", " ")
hit = {"patron": pname, "fragmento": frag}
(veto if pname in VETO_PATTERN_NAMES else otros).append(hit)
return veto, otros
# ------------------------------------------------------------------------------------------
# Entorno
# ------------------------------------------------------------------------------------------
def require_env(name, why):
value = os.environ.get(name, "").strip()
if not value:
raise Gate5Error(f"falta la variable de entorno {name} ({why})")
return value
def load_env(need_mcp):
"""Lee la config. Nunca devuelve nada que se pueda imprimir sin pensar: la URL del MCP
lleva el token embebido en varios deployments."""
faltantes = []
cfg = {}
for name, why in (("GATE5_BASE_URL", "endpoint OpenAI-compatible de vLLM, ej. http://HOST:PUERTO/v1"),
("GATE5_MODEL", "nombre del modelo servido por vLLM")):
try:
cfg[name] = require_env(name, why)
except Gate5Error as e:
faltantes.append(str(e))
if need_mcp:
for name, why in (("PENPOT_MCP_URL", "endpoint HTTP MCP del servidor de Penpot"),
("PENPOT_MCP_TOKEN", "token bearer del servidor MCP de Penpot")):
try:
cfg[name] = require_env(name, why)
except Gate5Error as e:
faltantes.append(str(e))
if faltantes:
raise Gate5Error("configuracion incompleta:\n - " + "\n - ".join(faltantes))
cfg["GATE5_TAG"] = os.environ.get("GATE5_TAG", "").strip() or "sin-tag"
return cfg
# ------------------------------------------------------------------------------------------
# Cliente MCP sobre HTTP (streamable HTTP / JSON-RPC 2.0)
# ------------------------------------------------------------------------------------------
class PenpotMCP:
"""Cliente minimo de MCP sobre HTTP.
Handshake: initialize -> notifications/initialized -> tools/list -> tools/call.
La respuesta puede llegar como `application/json` o como `text/event-stream` (lineas
`data: {...}`); las dos se manejan. El `Mcp-Session-Id` que devuelve el initialize se
guarda y se reusa en todas las llamadas siguientes.
"""
PROTOCOL_VERSION = "2025-06-18"
def __init__(self, url, token, timeout=MCP_TIMEOUT):
self._url = url
self._token = token
self.timeout = timeout
self.session_id = None
self.tool_names = []
self._next_id = 0
self._in_handshake = False
self._http = requests.Session()
# -- transporte -------------------------------------------------------------------
def _headers(self):
headers = {
"Content-Type": "application/json",
"Accept": "application/json, text/event-stream",
"Authorization": f"Bearer {self._token}",
"MCP-Protocol-Version": self.PROTOCOL_VERSION,
}
if self.session_id:
headers["Mcp-Session-Id"] = self.session_id
return headers
@staticmethod
def _parse_body(response):
"""Devuelve la lista de mensajes JSON-RPC del cuerpo, venga como SSE o como JSON."""
ctype = (response.headers.get("Content-Type") or "").lower()
text = response.text or ""
if "text/event-stream" in ctype:
mensajes = []
for line in text.splitlines():
line = line.strip()
if not line.startswith("data:"):
continue
payload = line[len("data:"):].strip()
if not payload or payload == "[DONE]":
continue
try:
mensajes.append(json.loads(payload))
except json.JSONDecodeError:
continue
return mensajes
if not text.strip():
return []
try:
parsed = json.loads(text)
except json.JSONDecodeError as e:
raise Gate5Error(f"el servidor MCP devolvio un cuerpo no-JSON ({e})")
return parsed if isinstance(parsed, list) else [parsed]
def _post(self, payload, expect_response=True):
try:
response = self._http.post(
self._url, headers=self._headers(), json=payload, timeout=self.timeout
)
except requests.RequestException as e:
# El mensaje de requests incluye la URL -> se descarta a proposito.
raise Gate5Error(f"error de transporte contra el MCP de Penpot: {type(e).__name__}")
nueva_sesion = response.headers.get("Mcp-Session-Id") or response.headers.get("mcp-session-id")
if nueva_sesion:
self.session_id = nueva_sesion
if response.status_code >= 400:
cuerpo = (response.text or "")[:300]
if PLUGIN_NOT_CONNECTED in cuerpo:
raise PluginNotConnected(cuerpo.strip())
raise Gate5Error(f"el MCP de Penpot respondio HTTP {response.status_code}: {cuerpo}")
if not expect_response:
return None
mensajes = self._parse_body(response)
for msg in mensajes:
if msg.get("id") != payload.get("id"):
continue
if "error" in msg:
detalle = json.dumps(msg["error"], ensure_ascii=False)[:400]
if PLUGIN_NOT_CONNECTED in detalle:
raise PluginNotConnected(detalle)
raise Gate5Error(f"el MCP de Penpot devolvio un error JSON-RPC: {detalle}")
return msg.get("result", {})
raise Gate5Error(f"el MCP de Penpot no devolvio respuesta para el metodo {payload.get('method')}")
def _request(self, method, params=None):
"""Envia un request JSON-RPC, reintentando los fallos de TRANSPORTE.
El MCP de Penpot corta la conexion a mitad de llamada de forma intermitente
("transport dropped mid-call"); ya se habia visto en la puerta 4 de la Fase 4. Sin
reintento, un corte transitorio en el prompt 6 de 8 tira abajo una corrida entera de la
puerta, y peor: se diagnostica como un fallo del modelo cuando es de la red.
Se reintenta SOLO `Gate5Error` de transporte/protocolo. `PluginNotConnected` no se
reintenta -- ese es un estado real del mundo (el usuario no tiene el plugin abierto) y
reintentarlo solo demora el mensaje que hay que dar.
"""
ultimo = None
for intento in range(1, MCP_MAX_RETRIES + 1):
self._next_id += 1
payload = {"jsonrpc": "2.0", "id": self._next_id, "method": method}
if params is not None:
payload["params"] = params
try:
return self._post(payload)
except PluginNotConnected:
raise
except Gate5Error as e:
ultimo = e
if intento == MCP_MAX_RETRIES:
break
espera = MCP_RETRY_BACKOFF * (2 ** (intento - 1))
print(f"[MCP] fallo transitorio en '{method}' (intento {intento}/"
f"{MCP_MAX_RETRIES}): {e}. Reintento en {espera:.0f}s")
time.sleep(espera)
# Un ConnectionError no solo tira la request: puede tirar la SESION. Reintentar
# el mismo tools/call contra una sesion muerta falla siempre igual, que es lo
# que se vio en la primera corrida (4 intentos identicos, 4 ConnectionError).
# Hay que rehacer el handshake antes de reintentar.
if not self._in_handshake and method != "initialize":
try:
print("[MCP] rehaciendo el handshake (la sesion pudo haber muerto)")
self.session_id = None
self.handshake()
except Gate5Error as e2:
print(f"[MCP] el re-handshake tambien fallo: {e2}")
raise Gate5Error(f"el MCP de Penpot fallo {MCP_MAX_RETRIES} veces en '{method}': {ultimo}")
def _notify(self, method, params=None):
payload = {"jsonrpc": "2.0", "method": method}
if params is not None:
payload["params"] = params
self._post(payload, expect_response=False)
# -- protocolo --------------------------------------------------------------------
def handshake(self):
self._in_handshake = True
try:
return self._handshake_inner()
finally:
self._in_handshake = False
def _handshake_inner(self):
result = self._request("initialize", {
"protocolVersion": self.PROTOCOL_VERSION,
"capabilities": {},
"clientInfo": {"name": "gate5-penpot-design", "version": "1.0"},
})
self._notify("notifications/initialized")
listado = self._request("tools/list") or {}
self.tool_names = [t.get("name") for t in listado.get("tools", [])]
server = (result.get("serverInfo") or {}).get("name", "?")
print(f"[MCP] handshake ok con '{server}'; tools: {', '.join(self.tool_names)}")
return result
@staticmethod
def _flatten_content(result):
"""Aplana `content` de un tool result MCP a (texto, lista_de_binarios_base64)."""
partes, binarios = [], []
for item in (result or {}).get("content", []) or []:
tipo = item.get("type")
if tipo == "text":
partes.append(item.get("text") or "")
elif tipo in ("image", "audio") and item.get("data"):
binarios.append(item["data"])
partes.append(f"[{tipo}/{item.get('mimeType', '?')}, {len(item['data'])} bytes base64]")
elif tipo == "resource":
recurso = item.get("resource") or {}
if recurso.get("text"):
partes.append(recurso["text"])
elif recurso.get("blob"):
binarios.append(recurso["blob"])
return "\n".join(partes).strip(), binarios
def call_tool(self, name, arguments):
"""Devuelve (texto, binarios, es_error). Aborta si el plugin no esta conectado."""
result = self._request("tools/call", {"name": name, "arguments": arguments}) or {}
texto, binarios = self._flatten_content(result)
if PLUGIN_NOT_CONNECTED in texto:
raise PluginNotConnected(texto)
return texto, binarios, bool(result.get("isError"))
def execute_code(self, code):
texto, _binarios, es_error = self.call_tool("execute_code", {"code": code})
return texto, es_error
def execute_json(self, code, que):
"""execute_code cuyo resultado la puerta necesita parsear como JSON.
El servidor MCP no devuelve el `return` del payload pelado: lo envuelve en
`{"result": <lo-que-retornaste>, "log": "<lo-que-escribiste-por-console>"}`. La puerta
necesita el interior, asi que se desenvuelve ACA, en un solo lugar.
Ojo con no desenvolver de mas: si un payload de la puerta retornara a proposito un
objeto con una clave `result` propia, desenvolver a ciegas lo destruiria. Por eso se
exige la forma exacta del envoltorio (`result` presente y ninguna clave fuera de
{result, log}).
`execute_code` (el metodo de abajo) NO desenvuelve: lo que ve el modelo como tool result
tiene que ser byte a byte lo que le llegaria en produccion, envoltorio incluido.
"""
texto, es_error = self.execute_code(code)
try:
parsed = json.loads(texto)
except json.JSONDecodeError:
raise Gate5Error(
f"{que}: el MCP no devolvio JSON ({'error' if es_error else 'texto'}): {texto[:300]}"
)
if (isinstance(parsed, dict) and "result" in parsed
and set(parsed).issubset({"result", "log"})):
log = (parsed.get("log") or "").strip()
if log:
print(f"[MCP] console del payload de la puerta ({que}): {log[:300]}")
return parsed["result"]
return parsed
# ------------------------------------------------------------------------------------------
# Payloads de la puerta (nunca los escribe el modelo)
# ------------------------------------------------------------------------------------------
def js_string(value):
"""Literal JS seguro a partir de un valor Python (via JSON, que es un subconjunto)."""
return json.dumps(value, ensure_ascii=False)
SETUP_JS = """
const page = penpot.createPage();
page.name = %PAGE_NAME%;
penpot.openPage(page);
const seed = %SEED%;
let seedBoardId = null;
if (seed) {
const board = penpot.createBoard();
board.name = seed.nombre;
board.x = seed.x;
board.y = seed.y;
board.resize(seed.ancho, seed.alto);
board.fills = [{ fillColor: "#FFFFFF", fillOpacity: 1 }];
for (const spec of seed.rectangulos) {
const rect = penpot.createRectangle();
rect.name = spec.nombre;
rect.x = seed.x + spec.x;
rect.y = seed.y + spec.y;
rect.resize(spec.ancho, spec.alto);
rect.fills = [{ fillColor: seed.gris, fillOpacity: 1 }];
board.insertChild(board.children.length, rect);
}
seedBoardId = board.id;
}
return {
pageId: page.id,
pageName: page.name,
rootId: penpot.root.id,
seedBoardId: seedBoardId
};
"""
# Payload de auditoria. Lo inyecta la puerta, asi que tiene que respetar la misma API verificada
# que se le exige al modelo: findShapeById de 1 argumento, board.flex/board.grid (nunca .layout),
# includeChildren (nunca withChildren), sin asignaciones a propiedades read-only.
AUDIT_JS = """
const ROOT_ID = %ROOT_ID%;
const root = penpotUtils.findShapeById(ROOT_ID);
if (!root) { return { auditError: "root no encontrado" }; }
const nodes = [];
const collect = (shape, depth) => {
nodes.push({ shape: shape, depth: depth });
const kids = shape.children || [];
for (let i = 0; i < kids.length; i++) { collect(kids[i], depth + 1); }
};
const top = root.children || [];
for (let i = 0; i < top.length; i++) { collect(top[i], 1); }
const PUROS = ["#FFFFFF", "#000000"];
const esGrisPlaceholder = (hex) => {
const r = parseInt(hex.slice(1, 3), 16);
const g = parseInt(hex.slice(3, 5), 16);
const b = parseInt(hex.slice(5, 7), 16);
const mx = Math.max(r, g, b);
const mn = Math.min(r, g, b);
const sat = mx === 0 ? 0 : (100 * (mx - mn)) / mx;
return sat <= 10 && mx >= 100 && mx <= 220;
};
const hexes = [];
const grises = [];
let gradientes = 0, sombras = 0, radios = 0;
let textos = 0, textosConChars = 0, textosConFill = 0, textosConFont = 0;
const fontSizes = [];
let boards = 0, boardsConLayout = 0, hijosDeLayout = 0, hijosDimensionados = 0;
let profundidad = 0, secciones = 0;
for (const nodo of nodes) {
const s = nodo.shape;
if (nodo.depth > profundidad) { profundidad = nodo.depth; }
let fills = null;
try { fills = s.fills; } catch (e) { fills = null; }
const locales = [];
if (Array.isArray(fills)) {
for (const f of fills) {
if (f && typeof f.fillColor === "string") { locales.push(f.fillColor.toUpperCase()); }
const grad = f && f.fillColorGradient;
if (grad) {
gradientes++;
for (const stop of (grad.stops || [])) {
if (stop && typeof stop.color === "string") { locales.push(stop.color.toUpperCase()); }
}
}
}
}
for (const hex of locales) {
hexes.push(hex);
if (esGrisPlaceholder(hex)) { grises.push({ id: s.id, name: s.name, hex: hex }); }
}
try { if (Array.isArray(s.shadows) && s.shadows.length > 0) { sombras++; } } catch (e) {}
try { if (typeof s.borderRadius === "number" && s.borderRadius > 0) { radios++; } } catch (e) {}
if (s.type === "text") {
textos++;
const chars = (s.characters || "").trim();
if (chars.length > 0) { textosConChars++; }
if (Array.isArray(fills) && fills.length > 0) { textosConFill++; }
if (s.fontId && String(s.fontId).length > 0) { textosConFont++; }
const tam = parseFloat(s.fontSize);
if (!isNaN(tam)) { fontSizes.push(tam); }
}
if (s.type === "board") {
boards++;
const conLayout = !!(s.flex || s.grid);
if (conLayout) {
boardsConLayout++;
for (const kid of (s.children || [])) {
hijosDeLayout++;
let lc = null;
try { lc = kid.layoutChild; } catch (e) { lc = null; }
if (lc && lc.horizontalSizing) { hijosDimensionados++; }
}
}
if (nodo.depth <= 2 && (s.children || []).length >= 2) { secciones++; }
}
}
const distintos = [];
for (const hex of hexes) {
if (PUROS.indexOf(hex) === -1 && distintos.indexOf(hex) === -1) { distintos.push(hex); }
}
let contencion = [];
let contencionError = null;
try {
const analisis = penpotUtils.analyzeDescendants(root, (raiz, shape) => {
const padre = shape.parent;
if (!padre || padre.id === raiz.id) { return null; }
return penpotUtils.isContainedIn(shape, padre)
? null
: { id: shape.id, name: shape.name, parent: padre.name };
}, 12) || [];
for (const entrada of analisis) {
if (entrada && entrada.result) { contencion.push(entrada.result); }
}
} catch (e) {
contencionError = String((e && e.message) || e);
}
let principal = null;
for (const s of top) {
if (s.type !== "board") { continue; }
if (!principal || s.width * s.height > principal.width * principal.height) { principal = s; }
}
if (!principal && top.length > 0) { principal = top[0]; }
let desajusteAltura = null;
if (principal) {
let fondo = null;
for (const kid of (principal.children || [])) {
const limite = kid.y + kid.height;
if (fondo === null || limite > fondo) { fondo = limite; }
}
if (fondo !== null) {
desajusteAltura = Math.round(Math.abs(principal.y + principal.height - fondo));
}
}
let cssLen = 0, markupLen = 0, renderError = null;
try {
const objetivo = principal ? [principal] : top;
if (objetivo.length > 0) {
const css = penpot.generateStyle(objetivo, { type: "css", includeChildren: true }) || "";
const markup = penpot.generateMarkup(objetivo, { type: "html" }) || "";
cssLen = String(css).length;
markupLen = String(markup).length;
}
} catch (e) {
renderError = String((e && e.message) || e);
}
const tamsUnicos = [];
for (const tam of fontSizes) { if (tamsUnicos.indexOf(tam) === -1) { tamsUnicos.push(tam); } }
// 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 };
}
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;
return {
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
};
"""
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"),
("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")
# 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}
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,
},
"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,
},
"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,
},
"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,
},
"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,
},
"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,
},
"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,
},
"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,
},
"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,
},
"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,
},
}
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"],
"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),
"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` 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):
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})")
# 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,
"rootId": root_id,
"seedBoardId": setup.get("seedBoardId"),
"turnos_usados": len(turnos),
"finish_reason": finish_reason,
"mensaje_final": final_content,
"auditoria": audit,
"metricas_crudas": {k: (list(v) if isinstance(v, tuple) else v) for k, v in raw.items()},
"execute_code_calls": exec_calls,
"execute_code_excepciones": exec_exceptions,
"code_payloads": code_payloads,
"export_shape_calls": export_calls,
"veto_hits": veto_hits,
"otros_patrones": otros_hits,
"png": str(png_path.relative_to(REPO_ROOT)) if png_path else None,
"turnos": turnos,
**resultado,
}
# ------------------------------------------------------------------------------------------
# Modo holdout: grep estatico barato, sin MCP y sin loop
# ------------------------------------------------------------------------------------------
def run_holdout(cfg, tools, system_prompt):
if not HOLDOUT_PATH.exists():
raise Gate5Error(
f"no existe {HOLDOUT_PATH.relative_to(REPO_ROOT)}; generalo con "
f"scripts/35_build_penpot_design_holdout.py antes de correr el modo holdout"
)
filas = []
with open(HOLDOUT_PATH, encoding="utf-8") as f:
for linea in f:
linea = linea.strip()
if linea:
filas.append(json.loads(linea))
print(f"[INFO] modo holdout: {len(filas)} prompts de diseno, sin tocar el MCP\n")
resultados = []
con_prohibido = 0
errores = 0
patrones = Counter()
for i, fila in enumerate(filas, start=1):
prompt = fila.get("prompt") or fila.get("user") or ""
messages = [{"role": "system", "content": system_prompt},
{"role": "user", "content": prompt}]
try:
choice = chat_completion(cfg, messages, tools)
except requests.RequestException as e:
# Una generacion que nunca ocurrio NO es una generacion limpia: se cuenta aparte y
# nunca entra en el denominador de la tasa (si no, un endpoint caido "aprueba").
errores += 1
resultados.append({"prompt": prompt, "error": f"{type(e).__name__}: {e}"})
continue
message = choice.get("message", {})
veto_total, otros_total, codes = [], [], []
for tool_call in message.get("tool_calls") or []:
nombre = tool_call.get("function", {}).get("name")
args, error_args = parse_tool_arguments(tool_call)
if error_args:
continue
if nombre == "execute_code":
code = args.get("code", "")
codes.append(code)
veto, otros = scan_forbidden(code)
veto_total.extend(veto)
otros_total.extend(otros)
elif nombre == "export_shape":
inventados = sorted(k for k in args if k not in EXPORT_SHAPE_ALLOWED_ARGS)
if inventados:
veto_total.append({"patron": "argumentos inventados en export_shape",
"fragmento": ", ".join(inventados)})
elif nombre not in ("high_level_overview", "penpot_api_info"):
veto_total.append({"patron": "tool inexistente", "fragmento": str(nombre)})
if veto_total:
con_prohibido += 1
for hit in veto_total:
patrones[hit["patron"]] += 1
resultados.append({
"prompt": prompt,
"id": fila.get("id"),
"tool_calls": [tc.get("function", {}).get("name")
for tc in (message.get("tool_calls") or [])],
"code_payloads": codes,
"veto_hits": veto_total,
"otros_patrones": otros_total,
"content": message.get("content") or "",
})
if i % 10 == 0:
print(f"[INFO] {i}/{len(filas)} prompts procesados")
total = len(resultados) - errores
if total <= 0:
raise Gate5Error(
f"ninguna de las {len(filas)} generaciones del holdout llego a completarse "
f"({errores} error(es) de request contra vLLM): no hay nada que medir"
)
tasa = con_prohibido / total
# Los errores de request bloquean la aprobacion a proposito: la puerta es el disparador de
# fallback antes de cuantizar, y un "aprueba" sobre generaciones faltantes no vale nada.
aprueba = tasa <= HOLDOUT_FORBIDDEN_RATE_MAX and errores == 0
print("\n=== Puerta 5 (modo holdout) -- tasa de API prohibida ===")
print(f" prompts medidos: {total} de {len(filas)}")
print(f" errores de request: {errores}")
print(f" con API prohibida: {con_prohibido}")
print(f" tasa: {100 * tasa:.1f}% (umbral <= {100 * HOLDOUT_FORBIDDEN_RATE_MAX:.0f}%)")
if patrones:
print(" desglose por patron:")
for patron, n in patrones.most_common():
print(f" - {patron}: {n}")
print(f"\n VEREDICTO: {'APRUEBA' if aprueba else 'NO APRUEBA -- disparador de fallback'}")
salida = REPO_ROOT / "data" / f"gate5_holdout_{cfg['GATE5_TAG']}.json"
salida.write_text(json.dumps({
"modo": "holdout",
"tag": cfg["GATE5_TAG"],
"modelo": cfg["GATE5_MODEL"],
"total": total,
"prompts_en_archivo": len(filas),
"errores_de_request": errores,
"con_api_prohibida": con_prohibido,
"tasa": tasa,
"umbral": HOLDOUT_FORBIDDEN_RATE_MAX,
"aprueba": aprueba,
"patrones": dict(patrones),
"resultados": resultados,
}, ensure_ascii=False, indent=2), encoding="utf-8")
print(f" resultados en {salida.relative_to(REPO_ROOT)}")
return 0 if aprueba else 1
# ------------------------------------------------------------------------------------------
# Modo completo
# ------------------------------------------------------------------------------------------
def load_prompts():
filas = []
with open(PROMPTS_PATH, encoding="utf-8") as f:
for lineno, linea in enumerate(f, start=1):
linea = linea.strip()
if not linea:
continue
try:
filas.append(json.loads(linea))
except json.JSONDecodeError as e:
raise Gate5Error(f"{PROMPTS_PATH.name}:{lineno}: JSON invalido: {e}")
faltantes = [f["id"] for f in filas if f["id"] not in THRESHOLDS]
if faltantes:
raise Gate5Error(f"prompts sin umbrales declarados en THRESHOLDS: {faltantes}")
return filas
def resumen(resultados, tag):
scores = [r["score"] for r in resultados]
media = sum(scores) / len(scores) if scores else 0.0
veto_limpio = sum(1 for r in resultados if not r["veto_violado"])
sobre_minimo = sum(1 for r in resultados if r["score"] >= PROMPT_SCORE_MIN)
flagship = next((r for r in resultados if r["id"] == FLAGSHIP_PROMPT_ID), None)
flagship_score = flagship["score"] if flagship else 0.0
print(f"\n=== Puerta 5 -- calidad de diseno en Penpot (tag={tag}) ===\n")
print(f" {'prompt':32s} {'dif':6s} {'score':>6s} {'pasa':>7s} {'grises':>7s} "
f"{'prohib':>7s} {'shapes':>7s} {'colores':>8s} {'textos':>7s}")
for r in resultados:
crudas = r["metricas_crudas"]
print(f" {r['id']:32s} {str(r['dificultad'])[:6]:6s} {r['score']:6.1f} "
f"{r['pasadas']:3d}/{r['aplicables']:<3d} {crudas['placeholderGreys']:7d} "
f"{crudas['forbiddenBehavior']:7d} {crudas['shapeCount']:7d} "
f"{crudas['distinctFillColors']:8d} {crudas['textsWithCharacters']:7d}")
condiciones = [
(f"score medio >= {MEAN_SCORE_MIN:.0f}", media >= MEAN_SCORE_MIN, f"{media:.1f}"),
("veto limpio en 8/8", veto_limpio == len(resultados), f"{veto_limpio}/{len(resultados)}"),
(f">= {MIN_PROMPTS_OVER_MIN}/8 prompts >= {PROMPT_SCORE_MIN:.0f}",
sobre_minimo >= MIN_PROMPTS_OVER_MIN, f"{sobre_minimo}/{len(resultados)}"),
(f"landing de pizzeria >= {PROMPT_SCORE_MIN:.0f}",
flagship_score >= PROMPT_SCORE_MIN, f"{flagship_score:.1f}"),
]
print("\n condiciones de aprobacion:")
for etiqueta, ok, valor in condiciones:
print(f" [{'ok ' if ok else 'NO '}] {etiqueta:40s} -> {valor}")
aprueba = all(ok for _e, ok, _v in condiciones)
print(f"\n VEREDICTO GLOBAL: {'APRUEBA' if aprueba else 'NO APRUEBA'}")
return aprueba, {
"score_medio": round(media, 2),
"veto_limpio": veto_limpio,
"prompts_sobre_minimo": sobre_minimo,
"score_landing_pizzeria": flagship_score,
"aprueba": aprueba,
"condiciones": [{"condicion": e, "ok": ok, "valor": v} for e, ok, v in condiciones],
}
def run_full(cfg, tools, system_prompt):
prompts = load_prompts()
check_disjunto(prompts)
self_check_gate_payloads()
mcp = PenpotMCP(cfg["PENPOT_MCP_URL"], cfg["PENPOT_MCP_TOKEN"])
mcp.handshake()
# Sonda barata: si el plugin no esta conectado, abortar ANTES de quemar los 8 prompts.
sonda, _es_error = mcp.execute_code("return { gate5Probe: true };")
if PLUGIN_NOT_CONNECTED in sonda:
raise PluginNotConnected(sonda)
print(f"[MCP] sonda de conectividad del plugin: {sonda[:120]}")
png_dir = REPO_ROOT / "data" / f"gate5_png_{cfg['GATE5_TAG']}"
salida = REPO_ROOT / "data" / f"gate5_results_{cfg['GATE5_TAG']}.json"
def volcar(resultados_parciales, dt_parcial, veredicto_parcial=None, completo=False):
"""Escribe el JSON de resultados. Se llama DESPUES DE CADA PROMPT.
La primera corrida del baseline murio en el prompt 4 por una caida de conexion del MCP
y se perdio la medicion de los tres que ya habian terminado -- que era justamente el
dato caro, porque exige tener produccion arriba. Un fallo de transporte no puede costar
el trabajo ya hecho.
"""
salida.write_text(json.dumps({
"modo": "completo" if completo else "parcial",
"tag": cfg["GATE5_TAG"],
"modelo": cfg["GATE5_MODEL"],
"max_turnos": MAX_TURNS,
"max_tokens": MAX_TOKENS,
"timestamp": datetime.now(timezone.utc).isoformat(),
"duracion_s": round(dt_parcial, 1),
"prompts_completados": len(resultados_parciales),
"prompts_totales": len(prompts),
"metricas": [{"clave": k, "etiqueta": e, "comparador": c} for k, e, c in METRICS],
"veto": list(VETO_METRICS),
"umbrales": {pid: {k: (list(v) if isinstance(v, tuple) else v) for k, v in u.items()}
for pid, u in THRESHOLDS.items()},
"veredicto": veredicto_parcial,
"prompts": resultados_parciales,
}, ensure_ascii=False, indent=2), encoding="utf-8")
t0 = time.time()
resultados = []
for fila in prompts:
try:
resultados.append(run_prompt(cfg, mcp, tools, system_prompt, fila, png_dir))
except PluginNotConnected:
volcar(resultados, time.time() - t0)
raise
except Gate5Error as e:
print(f"[ERROR] {fila['id']}: {e}")
print(f"[INFO] se guardan los {len(resultados)} prompt(s) ya medidos y se sigue con "
f"el siguiente; el JSON queda marcado como parcial")
resultados.append({"id": fila["id"], "error": str(e), "score": None})
volcar(resultados, time.time() - t0)
dt = time.time() - t0
aprueba, veredicto = resumen(resultados, cfg["GATE5_TAG"])
print(f"\n tiempo total: {dt / 60:.1f} min")
volcar(resultados, dt, veredicto_parcial=veredicto, completo=True)
print(f" resultados en {salida.relative_to(REPO_ROOT)}")
print(f" PNGs en {png_dir.relative_to(REPO_ROOT)}/ (la puerta no borra nada: las paginas "
f"quedan para inspeccion humana)")
return 0 if aprueba else 1
def main():
modo_holdout = os.environ.get("GATE5_HOLDOUT", "").strip() in ("1", "true", "yes")
try:
cfg = load_env(need_mcp=not modo_holdout)
tools = load_penpot_tools()
system_prompt = load_system_prompt()
if modo_holdout:
return run_holdout(cfg, tools, system_prompt)
return run_full(cfg, tools, system_prompt)
except PluginNotConnected as e:
print(
"\n[ABORTA] el servidor MCP de Penpot no tiene ningun plugin conectado.\n"
" Abri el archivo de Penpot en el navegador, arranca el plugin del MCP y espera a\n"
" que quede conectado; recien ahi volve a correr la puerta. Sin plugin conectado no\n"
" tiene sentido gastar los 8 prompts.\n"
f" Mensaje del servidor: {str(e)[:200]}",
file=sys.stderr,
)
return EXIT_PLUGIN_NOT_CONNECTED
except Gate5Error as e:
print(f"\n[ERROR] {e}", file=sys.stderr)
return 1
if __name__ == "__main__":
sys.exit(main())
Reference in New Issue View Git Blame Copy Permalink