aleleba/qwen3-6-lora
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
c9792c5c40dbef69d17012718f02a4a0487ccfce
qwen3-6-lora/scripts/35_gate5_penpot_design.py
T
aleleba c65d309719 Phase 6.4: make the gates fail when they cannot verify something 
A code review found seven ways these gates could pass green with something
actually wrong. All are the same family: a missing value was treated as OK.
The rule now written into all three files is that absent is not OK, absent
is "could not verify", and that either fails or is reported as an explicit
SKIP - it never slips through as green.

30_eval_suite.py:
- A bucket with no baseline of its own fell back to the global 0.2750 and
  printed it in a column headed "baseline", as if it were that bucket's
  number. Measured against the real eval.jsonl buckets: negativos going
  from 0.12 to 0.33 is a real +0.21 regression, but the computed delta was
  +0.055 and it PASSED; manejo_errores sitting unchanged at 0.42 produced
  a fabricated +0.145 FAIL that would have discarded a healthy candidate
  mid-downtime. Now such buckets print SKIP and the verdict reports how
  many went unverified.
- "VEREDICTO: FAIL" exited 0, so a runbook chaining the gate into
  quantization would have carried on to write 24 GB. Now exits 1.
- A typo in BASELINE_BUCKET_LOSSES silently matched nothing; now aborts.
- The penpot exemption is labelled honestly: those 11 rows are pre-existing
  LoRA #1 tool-calling, not new capability, so gate 1 has no regression
  coverage there and the log says so.

20_merge_lora.py dry-run (merge path untouched, verified by AST diff):
- adapter_config.get("use_rslora", False) meant a missing key passed AND
  the log printed use_rslora=False, asserting it had checked something that
  was never there. A different PEFT version omitting a key was enough.
- lora_bias was not checked at all, only bias. They are different fields:
  lora_bias puts a bias inside lora_B, which W + scaling * (B @ A) ignores.
- The 620 keys were printed but never asserted, so an adapter with extra
  tensors printed "310 + 310 = 930" and passed.
- rank_pattern/alpha_pattern were not checked. They set r per module, so
  scaling is not uniformly alpha/r while both the dry-run and the merge
  apply a single 2.0 to all 310 tensors.
- A missing family was invisible: swap linear_attn for 150 mlp.gate targets
  and the total is still 310, no norm is zero because the family is simply
  gone, and it passed. Now presence and per-family counts are asserted,
  derived from the real adapter: linear_attn 150, shared_expert 120,
  attention_qkvo 40, otros 0.
Verified against seven synthetic adapters plus the real phase 3 one; only
the correct adapter passes.

21_quantize_nvfp4.py (recipe and oneshot untouched): the calibration cache
now carries a provenance.json recording the training file's sha256, the
recipe numbers and the bucket distribution, and loading aborts on mismatch.
This is the phase's number one risk and it had no mechanical defence: the
phase 5 cache on disk has exactly 512 rows, the same as the v2 recipe, so
the only existing check could not tell them apart and reusing it would have
calibrated with zero design data and washed out the new capability
silently. Verified: that cache now aborts.

gate 5: retry transport failures against the Penpot MCP, which drops
connections mid-call intermittently (seen before in phase 4's gate 4).
Without it a blip on prompt 6 of 8 kills a whole run and reads like a model
failure. PluginNotConnected is deliberately not retried - that is a real
state of the world. Also unwrap the {"result":..., "log":...} envelope the
server wraps execute_code returns in; the gate was reading keys off the
outer object and rejecting a valid page setup.
2026-07-30 17:37:46 +00:00

1338 lines
56 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
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):
"""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 * intento
print(f"[MCP] fallo transitorio en '{method}' (intento {intento}/"
f"{MCP_MAX_RETRIES}): {e}. Reintento en {espera:.0f}s")
time.sleep(espera)
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):
result = self._request("initialize", {
"protocolVersion": self.PROTOCOL_VERSION,
"capabilities": {},
"clientInfo": {"name": "gate5-penpot-design", "version": "1.0"},
})
self._notify("notifications/initialized")
listado = self._request("tools/list") or {}
self.tool_names = [t.get("name") for t in listado.get("tools", [])]
server = (result.get("serverInfo") or {}).get("name", "?")
print(f"[MCP] handshake ok con '{server}'; tools: {', '.join(self.tool_names)}")
return result
@staticmethod
def _flatten_content(result):
"""Aplana `content` de un tool result MCP a (texto, lista_de_binarios_base64)."""
partes, binarios = [], []
for item in (result or {}).get("content", []) or []:
tipo = item.get("type")
if tipo == "text":
partes.append(item.get("text") or "")
elif tipo in ("image", "audio") and item.get("data"):
binarios.append(item["data"])
partes.append(f"[{tipo}/{item.get('mimeType', '?')}, {len(item['data'])} bytes base64]")
elif tipo == "resource":
recurso = item.get("resource") or {}
if recurso.get("text"):
partes.append(recurso["text"])
elif recurso.get("blob"):
binarios.append(recurso["blob"])
return "\n".join(partes).strip(), binarios
def call_tool(self, name, arguments):
"""Devuelve (texto, binarios, es_error). Aborta si el plugin no esta conectado."""
result = self._request("tools/call", {"name": name, "arguments": arguments}) or {}
texto, binarios = self._flatten_content(result)
if PLUGIN_NOT_CONNECTED in texto:
raise PluginNotConnected(texto)
return texto, binarios, bool(result.get("isError"))
def execute_code(self, code):
texto, _binarios, es_error = self.call_tool("execute_code", {"code": code})
return texto, es_error
def execute_json(self, code, que):
"""execute_code cuyo resultado la puerta necesita parsear como JSON.
El servidor MCP no devuelve el `return` del payload pelado: lo envuelve en
`{"result": <lo-que-retornaste>, "log": "<lo-que-escribiste-por-console>"}`. La puerta
necesita el interior, asi que se desenvuelve ACA, en un solo lugar.
Ojo con no desenvolver de mas: si un payload de la puerta retornara a proposito un
objeto con una clave `result` propia, desenvolver a ciegas lo destruiria. Por eso se
exige la forma exacta del envoltorio (`result` presente y ninguna clave fuera de
{result, log}).
`execute_code` (el metodo de abajo) NO desenvuelve: lo que ve el modelo como tool result
tiene que ser byte a byte lo que le llegaria en produccion, envoltorio incluido.
"""
texto, es_error = self.execute_code(code)
try:
parsed = json.loads(texto)
except json.JSONDecodeError:
raise Gate5Error(
f"{que}: el MCP no devolvio JSON ({'error' if es_error else 'texto'}): {texto[:300]}"
)
if (isinstance(parsed, dict) and "result" in parsed
and set(parsed).issubset({"result", "log"})):
log = (parsed.get("log") or "").strip()
if log:
print(f"[MCP] console del payload de la puerta ({que}): {log[:300]}")
return parsed["result"]
return parsed
# ------------------------------------------------------------------------------------------
# Payloads de la puerta (nunca los escribe el modelo)
# ------------------------------------------------------------------------------------------
def js_string(value):
"""Literal JS seguro a partir de un valor Python (via JSON, que es un subconjunto)."""
return json.dumps(value, ensure_ascii=False)
SETUP_JS = """
const page = penpot.createPage();
page.name = %PAGE_NAME%;
penpot.openPage(page);
const seed = %SEED%;
let seedBoardId = null;
if (seed) {
const board = penpot.createBoard();
board.name = seed.nombre;
board.x = seed.x;
board.y = seed.y;
board.resize(seed.ancho, seed.alto);
board.fills = [{ fillColor: "#FFFFFF", fillOpacity: 1 }];
for (const spec of seed.rectangulos) {
const rect = penpot.createRectangle();
rect.name = spec.nombre;
rect.x = seed.x + spec.x;
rect.y = seed.y + spec.y;
rect.resize(spec.ancho, spec.alto);
rect.fills = [{ fillColor: seed.gris, fillOpacity: 1 }];
board.insertChild(board.children.length, rect);
}
seedBoardId = board.id;
}
return {
pageId: page.id,
pageName: page.name,
rootId: penpot.root.id,
seedBoardId: seedBoardId
};
"""
# Payload de auditoria. Lo inyecta la puerta, asi que tiene que respetar la misma API verificada
# que se le exige al modelo: findShapeById de 1 argumento, board.flex/board.grid (nunca .layout),
# includeChildren (nunca withChildren), sin asignaciones a propiedades read-only.
AUDIT_JS = """
const ROOT_ID = %ROOT_ID%;
const root = penpotUtils.findShapeById(ROOT_ID);
if (!root) { return { auditError: "root no encontrado" }; }
const nodes = [];
const collect = (shape, depth) => {
nodes.push({ shape: shape, depth: depth });
const kids = shape.children || [];
for (let i = 0; i < kids.length; i++) { collect(kids[i], depth + 1); }
};
const top = root.children || [];
for (let i = 0; i < top.length; i++) { collect(top[i], 1); }
const PUROS = ["#FFFFFF", "#000000"];
const esGrisPlaceholder = (hex) => {
const r = parseInt(hex.slice(1, 3), 16);
const g = parseInt(hex.slice(3, 5), 16);
const b = parseInt(hex.slice(5, 7), 16);
const mx = Math.max(r, g, b);
const mn = Math.min(r, g, b);
const sat = mx === 0 ? 0 : (100 * (mx - mn)) / mx;
return sat <= 10 && mx >= 100 && mx <= 220;
};
const hexes = [];
const grises = [];
let gradientes = 0, sombras = 0, radios = 0;
let textos = 0, textosConChars = 0, textosConFill = 0, textosConFont = 0;
const fontSizes = [];
let boards = 0, boardsConLayout = 0, hijosDeLayout = 0, hijosDimensionados = 0;
let profundidad = 0, secciones = 0;
for (const nodo of nodes) {
const s = nodo.shape;
if (nodo.depth > profundidad) { profundidad = nodo.depth; }
let fills = null;
try { fills = s.fills; } catch (e) { fills = null; }
const locales = [];
if (Array.isArray(fills)) {
for (const f of fills) {
if (f && typeof f.fillColor === "string") { locales.push(f.fillColor.toUpperCase()); }
const grad = f && f.fillColorGradient;
if (grad) {
gradientes++;
for (const stop of (grad.stops || [])) {
if (stop && typeof stop.color === "string") { locales.push(stop.color.toUpperCase()); }
}
}
}
}
for (const hex of locales) {
hexes.push(hex);
if (esGrisPlaceholder(hex)) { grises.push({ id: s.id, name: s.name, hex: hex }); }
}
try { if (Array.isArray(s.shadows) && s.shadows.length > 0) { sombras++; } } catch (e) {}
try { if (typeof s.borderRadius === "number" && s.borderRadius > 0) { radios++; } } catch (e) {}
if (s.type === "text") {
textos++;
const chars = (s.characters || "").trim();
if (chars.length > 0) { textosConChars++; }
if (Array.isArray(fills) && fills.length > 0) { textosConFill++; }
if (s.fontId && String(s.fontId).length > 0) { textosConFont++; }
const tam = parseFloat(s.fontSize);
if (!isNaN(tam)) { fontSizes.push(tam); }
}
if (s.type === "board") {
boards++;
const conLayout = !!(s.flex || s.grid);
if (conLayout) {
boardsConLayout++;
for (const kid of (s.children || [])) {
hijosDeLayout++;
let lc = null;
try { lc = kid.layoutChild; } catch (e) { lc = null; }
if (lc && lc.horizontalSizing) { hijosDimensionados++; }
}
}
if (nodo.depth <= 2 && (s.children || []).length >= 2) { secciones++; }
}
}
const distintos = [];
for (const hex of hexes) {
if (PUROS.indexOf(hex) === -1 && distintos.indexOf(hex) === -1) { distintos.push(hex); }
}
let contencion = [];
let contencionError = null;
try {
const analisis = penpotUtils.analyzeDescendants(root, (raiz, shape) => {
const padre = shape.parent;
if (!padre || padre.id === raiz.id) { return null; }
return penpotUtils.isContainedIn(shape, padre)
? null
: { id: shape.id, name: shape.name, parent: padre.name };
}, 12) || [];
for (const entrada of analisis) {
if (entrada && entrada.result) { contencion.push(entrada.result); }
}
} catch (e) {
contencionError = String((e && e.message) || e);
}
let principal = null;
for (const s of top) {
if (s.type !== "board") { continue; }
if (!principal || s.width * s.height > principal.width * principal.height) { principal = s; }
}
if (!principal && top.length > 0) { principal = top[0]; }
let desajusteAltura = null;
if (principal) {
let fondo = null;
for (const kid of (principal.children || [])) {
const limite = kid.y + kid.height;
if (fondo === null || limite > fondo) { fondo = limite; }
}
if (fondo !== null) {
desajusteAltura = Math.round(Math.abs(principal.y + principal.height - fondo));
}
}
let cssLen = 0, markupLen = 0, renderError = null;
try {
const objetivo = principal ? [principal] : top;
if (objetivo.length > 0) {
const css = penpot.generateStyle(objetivo, { type: "css", includeChildren: true }) || "";
const markup = penpot.generateMarkup(objetivo, { type: "html" }) || "";
cssLen = String(css).length;
markupLen = String(markup).length;
}
} catch (e) {
renderError = String((e && e.message) || e);
}
const tamsUnicos = [];
for (const tam of fontSizes) { if (tamsUnicos.indexOf(tam) === -1) { tamsUnicos.push(tam); } }
return {
shapeCount: nodes.length,
distinctFillColors: distintos.length,
fillColors: distintos.slice(0, 24),
placeholderGreys: grises.length,
greySamples: grises.slice(0, 8),
textsTotal: textos,
textsWithCharacters: textosConChars,
distinctFontSizes: tamsUnicos.length,
maxFontSize: tamsUnicos.length > 0 ? Math.max.apply(null, tamsUnicos) : 0,
textFillShare: textos > 0 ? textosConFill / textos : 0,
fontIdBoundShare: textos > 0 ? textosConFont / textos : 0,
boards: boards,
boardsWithLayout: boardsConLayout,
layoutChildren: hijosDeLayout,
layoutChildrenSized: hijosDimensionados,
layoutChildSizingShare: hijosDeLayout > 0 ? hijosDimensionados / hijosDeLayout : 0,
containmentViolations: contencion.length,
containmentSamples: contencion.slice(0, 5),
containmentError: contencionError,
nestingDepth: profundidad,
sectionCount: secciones,
radiusCount: radios,
shadowCount: sombras,
gradientCount: gradientes,
styleRichness: radios + sombras + gradientes,
mainBoardId: principal ? principal.id : null,
rootHeightMismatch: desajusteAltura,
cssLength: cssLen,
markupLength: markupLen,
renderError: renderError
};
"""
def build_setup_js(page_name, seed_board):
return (SETUP_JS
.replace("%PAGE_NAME%", js_string(page_name))
.replace("%SEED%", js_string(seed_board) if seed_board else "null"))
def build_audit_js(root_id):
return AUDIT_JS.replace("%ROOT_ID%", js_string(root_id))
def self_check_gate_payloads():
"""El payload de auditoria de la puerta tiene que respetar la misma API verificada que se le
exige al modelo. Si no, la puerta medirira sus propias excepciones."""
problemas = []
for nombre, code in (("SETUP_JS", SETUP_JS), ("AUDIT_JS", AUDIT_JS)):
veto, otros = scan_forbidden(code)
for hit in veto + otros:
problemas.append(f"{nombre}: [{hit['patron']}] ...{hit['fragmento']}...")
if problemas:
raise Gate5Error(
"los payloads de la propia puerta violan la API verificada:\n - "
+ "\n - ".join(problemas)
)
# ------------------------------------------------------------------------------------------
# Disjuncion contra train/eval/seeds via shingles de 6-gramas
# ------------------------------------------------------------------------------------------
def normalizar(texto):
sin_tildes = "".join(
c for c in unicodedata.normalize("NFD", texto.lower()) if unicodedata.category(c) != "Mn"
)
return re.findall(r"[a-z0-9]+", sin_tildes)
def shingles(texto, n=SHINGLE_N):
tokens = normalizar(texto)
if len(tokens) < n:
return {" ".join(tokens)} if tokens else set()
return {" ".join(tokens[i:i + n]) for i in range(len(tokens) - n + 1)}
def corpus_shingles(paths):
"""Shingles de todo el texto de usuario de los corpus de entrenamiento/eval/seeds."""
acumulado = set()
for path in paths:
if not path.exists():
print(f"[WARN] {path.name} no existe: no se puede verificar disjuncion contra el")
continue
with open(path, encoding="utf-8") as f:
for linea in f:
linea = linea.strip()
if not linea:
continue
try:
fila = json.loads(linea)
except json.JSONDecodeError:
continue
for msg in fila.get("messages", []):
if msg.get("role") != "user":
continue
acumulado |= shingles(msg.get("content") or "")
return acumulado
def check_disjunto(prompts):
corpus = corpus_shingles(DISJOINT_AGAINST)
if not corpus:
print("[WARN] corpus vacio: se saltea el chequeo de disjuncion")
return
problemas = []
for fila in prompts:
solapados = sorted(shingles(fila["prompt"]) & corpus)
if solapados:
problemas.append(f"{fila['id']}: {len(solapados)} shingle(s), p.ej. {solapados[0]!r}")
if problemas:
raise Gate5Error(
"los prompts de la puerta 5 NO son disjuntos de train/eval/seeds "
f"(shingles de {SHINGLE_N}-gramas):\n - " + "\n - ".join(problemas)
)
print(f"[OK] los {len(prompts)} prompts son disjuntos de train/eval/seeds "
f"({len(corpus)} shingles de referencia)")
# ------------------------------------------------------------------------------------------
# Las 17 metricas
# ------------------------------------------------------------------------------------------
# (clave, etiqueta, comparador). Comparadores:
# min -> pasa si valor >= umbral
# max -> pasa si valor <= umbral
# pair -> valor y umbral son tuplas; pasa si cada componente >= la suya
# bool -> pasa si valor es True (el umbral es True o None para "no aplica")
METRICS = [
("shapeCount", "conteo de shapes", "min"),
("distinctFillColors", "fillColor distintos (sin blanco/negro puros)", "min"),
("placeholderGreys", "grises de placeholder [VETO]", "max"),
("textsWithCharacters", "textos con characters no vacio", "min"),
("typeScale", "fontSize distintos y maximo", "pair"),
("textFillShare", "share de textos con fill", "min"),
("layoutBoards", "boards con flex/grid", "min"),
("layoutChildSizingShare", "share de hijos con layoutChild.horizontalSizing", "min"),
("containmentViolations", "violaciones de contencion", "max"),
("structure", "profundidad de anidamiento y secciones", "pair"),
("styleRichness", "riqueza de estilo (radios/sombras/gradientes)", "min"),
("rootHeightMismatch", "desajuste de altura root vs contenido", "max"),
("exportBeforeFinal", "export_shape antes del mensaje final", "bool"),
("forbiddenBehavior", "comportamiento prohibido [VETO]", "max"),
("exceptionRate", "tasa de excepciones de execute_code", "max"),
("fontIdBoundShare", "binding tipografico real (fontId)", "min"),
("rendersOk", "el arbol renderiza (generateStyle/generateMarkup)", "bool"),
]
VETO_METRICS = ("placeholderGreys", "forbiddenBehavior")
# Umbrales por prompt. `None` = la metrica NO aplica a ese prompt y no entra en el denominador
# del score. Estan aca y no en el .jsonl a proposito: el .jsonl describe la TAREA, el umbral es
# criterio de la puerta y se versiona junto al codigo que lo aplica.
THRESHOLDS = {
"g5-01-boton-primario": {
"shapeCount": 2, "distinctFillColors": 2, "placeholderGreys": 0,
"textsWithCharacters": 1, "typeScale": (1, 14), "textFillShare": 1.0,
"layoutBoards": None, "layoutChildSizingShare": None, "containmentViolations": 0,
"structure": None, "styleRichness": 2, "rootHeightMismatch": 24,
"exportBeforeFinal": True, "forbiddenBehavior": 0, "exceptionRate": 0.34,
"fontIdBoundShare": 1.0, "rendersOk": True,
},
"g5-02-navbar-flex": {
"shapeCount": 6, "distinctFillColors": 3, "placeholderGreys": 0,
"textsWithCharacters": 5, "typeScale": (1, 14), "textFillShare": 1.0,
"layoutBoards": 1, "layoutChildSizingShare": 0.5, "containmentViolations": 0,
"structure": (2, 1), "styleRichness": 1, "rootHeightMismatch": 24,
"exportBeforeFinal": True, "forbiddenBehavior": 0, "exceptionRate": 0.34,
"fontIdBoundShare": 1.0, "rendersOk": True,
},
"g5-03-card-producto": {
"shapeCount": 6, "distinctFillColors": 3, "placeholderGreys": 0,
"textsWithCharacters": 4, "typeScale": (3, 20), "textFillShare": 1.0,
"layoutBoards": None, "layoutChildSizingShare": None, "containmentViolations": 0,
"structure": (2, 1), "styleRichness": 3, "rootHeightMismatch": 32,
"exportBeforeFinal": True, "forbiddenBehavior": 0, "exceptionRate": 0.34,
"fontIdBoundShare": 1.0, "rendersOk": True,
},
"g5-04-tokens-biblioteca": {
"shapeCount": 8, "distinctFillColors": 6, "placeholderGreys": 0,
"textsWithCharacters": 4, "typeScale": (3, 24), "textFillShare": 1.0,
"layoutBoards": None, "layoutChildSizingShare": None, "containmentViolations": 0,
"structure": (2, 1), "styleRichness": 2, "rootHeightMismatch": 32,
"exportBeforeFinal": True, "forbiddenBehavior": 0, "exceptionRate": 0.34,
"fontIdBoundShare": 1.0, "rendersOk": True,
},
"g5-05-grid-tres-diferenciales": {
"shapeCount": 12, "distinctFillColors": 3, "placeholderGreys": 0,
"textsWithCharacters": 9, "typeScale": (2, 20), "textFillShare": 1.0,
"layoutBoards": 1, "layoutChildSizingShare": 0.5, "containmentViolations": 0,
"structure": (3, 3), "styleRichness": 3, "rootHeightMismatch": 32,
"exportBeforeFinal": True, "forbiddenBehavior": 0, "exceptionRate": 0.34,
"fontIdBoundShare": 1.0, "rendersOk": True,
},
"g5-06-landing-pizzeria": {
"shapeCount": 25, "distinctFillColors": 5, "placeholderGreys": 0,
"textsWithCharacters": 12, "typeScale": (4, 32), "textFillShare": 1.0,
"layoutBoards": 2, "layoutChildSizingShare": 0.5, "containmentViolations": 0,
"structure": (3, 5), "styleRichness": 5, "rootHeightMismatch": 48,
"exportBeforeFinal": True, "forbiddenBehavior": 0, "exceptionRate": 0.34,
"fontIdBoundShare": 1.0, "rendersOk": True,
},
"g5-07-onboarding-mobile": {
"shapeCount": 10, "distinctFillColors": 4, "placeholderGreys": 0,
"textsWithCharacters": 5, "typeScale": (3, 24), "textFillShare": 1.0,
"layoutBoards": 1, "layoutChildSizingShare": 0.5, "containmentViolations": 0,
"structure": (3, 2), "styleRichness": 3, "rootHeightMismatch": 32,
"exportBeforeFinal": True, "forbiddenBehavior": 0, "exceptionRate": 0.34,
"fontIdBoundShare": 1.0, "rendersOk": True,
},
"g5-08-reparacion-grises": {
"shapeCount": 8, "distinctFillColors": 4, "placeholderGreys": 0,
"textsWithCharacters": 5, "typeScale": (3, 24), "textFillShare": 1.0,
"layoutBoards": None, "layoutChildSizingShare": None, "containmentViolations": 0,
"structure": (2, 2), "styleRichness": 3, "rootHeightMismatch": 32,
"exportBeforeFinal": True, "forbiddenBehavior": 0, "exceptionRate": 0.34,
"fontIdBoundShare": 1.0, "rendersOk": True,
},
}
def build_raw_metrics(audit, runtime):
"""Aplana la auditoria del arbol + las señales del loop en las 17 claves de METRICS."""
audit = audit or {}
return {
"shapeCount": audit.get("shapeCount", 0),
"distinctFillColors": audit.get("distinctFillColors", 0),
"placeholderGreys": audit.get("placeholderGreys", 0),
"textsWithCharacters": audit.get("textsWithCharacters", 0),
"typeScale": (audit.get("distinctFontSizes", 0), audit.get("maxFontSize", 0)),
"textFillShare": audit.get("textFillShare", 0.0),
"layoutBoards": audit.get("boardsWithLayout", 0),
"layoutChildSizingShare": audit.get("layoutChildSizingShare", 0.0),
"containmentViolations": audit.get("containmentViolations", 0),
"structure": (audit.get("nestingDepth", 0), audit.get("sectionCount", 0)),
"styleRichness": audit.get("styleRichness", 0),
# Sin contenido no hay desajuste que medir: se penaliza como el peor caso posible.
"rootHeightMismatch": (audit.get("rootHeightMismatch")
if audit.get("rootHeightMismatch") is not None else 10 ** 6),
"exportBeforeFinal": runtime["exportBeforeFinal"],
"forbiddenBehavior": runtime["forbiddenBehavior"],
"exceptionRate": runtime["exceptionRate"],
"fontIdBoundShare": audit.get("fontIdBoundShare", 0.0),
"rendersOk": bool(audit.get("cssLength", 0) > 0 and audit.get("markupLength", 0) > 0
and not audit.get("renderError")),
}
def evaluar(clave, comparador, valor, umbral):
if umbral is None:
return None
if comparador == "min":
return valor >= umbral
if comparador == "max":
return valor <= umbral
if comparador == "bool":
return bool(valor) is bool(umbral)
if comparador == "pair":
return all(v >= u for v, u in zip(valor, umbral))
raise Gate5Error(f"comparador desconocido para {clave}: {comparador}")
def score_prompt(prompt_id, raw):
umbrales = THRESHOLDS[prompt_id]
detalle = []
pasadas = aplicables = 0
veto = []
for clave, etiqueta, comparador in METRICS:
umbral = umbrales.get(clave)
ok = evaluar(clave, comparador, raw[clave], umbral)
detalle.append({
"metrica": clave, "etiqueta": etiqueta, "valor": raw[clave],
"umbral": umbral, "aplica": ok is not None, "pasa": ok,
})
if ok is None:
continue
aplicables += 1
if ok:
pasadas += 1
elif clave in VETO_METRICS:
veto.append(clave)
score = 0.0 if veto else (100.0 * pasadas / aplicables if aplicables else 0.0)
return {
"score": round(score, 1), "pasadas": pasadas, "aplicables": aplicables,
"veto_violado": veto, "metricas": detalle,
}
# ------------------------------------------------------------------------------------------
# Cliente del modelo (vLLM, OpenAI-compatible)
# ------------------------------------------------------------------------------------------
def load_penpot_tools():
"""Las 4 tools de data/schemas/penpot.json, byte a byte lo que el modelo ve en produccion."""
tools = json.loads(PENPOT_TOOLS_PATH.read_text(encoding="utf-8"))
return [{
"type": "function",
"function": {
"name": t["name"],
"description": t.get("description", ""),
"parameters": t.get("parameters") or {"type": "object", "properties": {}},
},
} for t in tools]
def load_system_prompt():
"""El bloque `instructions` verbatim del servidor MCP. Medir con otro system prompt mediria
otra cosa: es exactamente lo que el modelo ve en produccion."""
return LINT.extract_system_block()
def chat_completion(cfg, messages, tools):
payload = {
"model": cfg["GATE5_MODEL"],
"messages": messages,
"tools": tools,
"tool_choice": "auto",
"max_tokens": MAX_TOKENS,
"temperature": 0.0,
}
url = cfg["GATE5_BASE_URL"].rstrip("/") + "/chat/completions"
response = requests.post(url, json=payload, timeout=HTTP_TIMEOUT)
response.raise_for_status()
return response.json()["choices"][0]
def parse_tool_arguments(tool_call):
raw = tool_call.get("function", {}).get("arguments")
if isinstance(raw, dict):
return raw, None
try:
return json.loads(raw or "{}"), None
except json.JSONDecodeError as e:
return {}, f"arguments no es JSON valido: {e}"
# ------------------------------------------------------------------------------------------
# Loop de agente sobre un prompt
# ------------------------------------------------------------------------------------------
def run_prompt(cfg, mcp, tools, system_prompt, fila, png_dir):
prompt_id = fila["id"]
tag = cfg["GATE5_TAG"]
ts = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
page_name = f"gate5/{tag}/{prompt_id}/{ts}"
print(f"\n--- {prompt_id} ({fila.get('dificultad', '?')}) -> pagina {page_name}")
setup = mcp.execute_json(build_setup_js(page_name, fila.get("seed_board")),
f"{prompt_id}: setup de la pagina")
root_id = setup.get("rootId")
if not root_id:
raise Gate5Error(f"{prompt_id}: el setup no devolvio rootId: {setup}")
contexto = [
f"Trabajás en la página «{page_name}», que ya está creada y abierta "
f"(id {setup.get('pageId')}). Todo lo que hagas va en esa página."
]
if setup.get("seedBoardId"):
contexto.append(
f"El board a intervenir es «{fila['seed_board']['nombre']}», "
f"id {setup['seedBoardId']}."
)
user_content = fila["prompt"] + "\n\n" + " ".join(contexto)
messages = [{"role": "system", "content": system_prompt},
{"role": "user", "content": user_content}]
code_payloads = []
export_calls = []
turnos = []
exec_calls = exec_exceptions = 0
veto_hits, otros_hits = [], []
final_content = None
finish_reason = None
for turno in range(1, MAX_TURNS + 1):
try:
choice = chat_completion(cfg, messages, tools)
except requests.RequestException as e:
turnos.append({"turno": turno, "error_modelo": f"{type(e).__name__}: {e}"})
break
message = choice.get("message", {})
finish_reason = choice.get("finish_reason")
tool_calls = message.get("tool_calls") or []
registro = {
"turno": turno,
"finish_reason": finish_reason,
"content": message.get("content") or "",
"reasoning": message.get("reasoning") or "",
"tool_calls": [],
}
if not tool_calls:
final_content = message.get("content") or ""
turnos.append(registro)
break
messages.append({
"role": "assistant",
"content": message.get("content") or "",
"tool_calls": tool_calls,
})
for tool_call in tool_calls:
nombre = tool_call.get("function", {}).get("name")
args, error_args = parse_tool_arguments(tool_call)
resultado = ""
if error_args:
resultado = error_args
elif nombre == "execute_code":
code = args.get("code", "")
code_payloads.append({"turno": turno, "code": code})
veto, otros = scan_forbidden(code)
for hit in veto:
hit["turno"] = turno
for hit in otros:
hit["turno"] = turno
veto_hits.extend(veto)
otros_hits.extend(otros)
exec_calls += 1
texto, es_error = mcp.execute_code(code)
if es_error or not texto.strip().startswith(("{", "[")):
exec_exceptions += 1
resultado = texto
elif nombre == "export_shape":
inventados = sorted(k for k in args if k not in EXPORT_SHAPE_ALLOWED_ARGS)
if inventados:
veto_hits.append({
"patron": "argumentos inventados en export_shape",
"fragmento": ", ".join(inventados), "turno": turno,
})
export_calls.append({"turno": turno, "args": args})
limpios = {k: v for k, v in args.items() if k in EXPORT_SHAPE_ALLOWED_ARGS}
texto, binarios, _ = mcp.call_tool("export_shape", limpios)
resultado = texto or f"[{len(binarios)} binario(s) devueltos]"
elif nombre in ("high_level_overview", "penpot_api_info"):
texto, _binarios, _ = mcp.call_tool(nombre, args)
resultado = texto
else:
resultado = f"Unknown tool: {nombre}"
registro["tool_calls"].append({
"name": nombre, "arguments": args,
"result_preview": resultado[:400],
})
messages.append({
"role": "tool",
"tool_call_id": tool_call.get("id"),
"name": nombre,
"content": resultado,
})
turnos.append(registro)
else:
print(f"[WARN] {prompt_id}: se agotaron los {MAX_TURNS} turnos sin mensaje final")
# --- auditoria inyectada por la puerta ------------------------------------------------
try:
audit = mcp.execute_json(build_audit_js(root_id), f"{prompt_id}: auditoria")
except Gate5Error as e:
print(f"[WARN] {prompt_id}: la auditoria fallo ({e})")
audit = {"auditError": str(e)}
# --- PNG de la puerta (nunca reemplaza al export del modelo, se suma) -----------------
png_path = None
shape_a_exportar = audit.get("mainBoardId") or root_id
try:
_texto, binarios, _ = mcp.call_tool(
"export_shape", {"shapeId": shape_a_exportar, "format": "png", "mode": "shape"})
if binarios:
png_dir.mkdir(parents=True, exist_ok=True)
png_path = png_dir / f"{prompt_id}.png"
png_path.write_bytes(base64.b64decode(binarios[0]))
print(f"[PNG] {png_path.relative_to(REPO_ROOT)}")
else:
print(f"[WARN] {prompt_id}: export_shape no devolvio binario")
except Gate5Error as e:
print(f"[WARN] {prompt_id}: no se pudo exportar el PNG ({e})")
runtime = {
"exportBeforeFinal": bool(export_calls),
"forbiddenBehavior": len(veto_hits),
"exceptionRate": (exec_exceptions / exec_calls) if exec_calls else 0.0,
}
raw = build_raw_metrics(audit, runtime)
resultado = score_prompt(prompt_id, raw)
print(f" score={resultado['score']:.1f} "
f"({resultado['pasadas']}/{resultado['aplicables']}) "
f"grises={raw['placeholderGreys']} prohibido={raw['forbiddenBehavior']} "
f"shapes={raw['shapeCount']} colores={raw['distinctFillColors']} "
f"textos={raw['textsWithCharacters']}")
return {
"id": prompt_id,
"dificultad": fila.get("dificultad"),
"prompt": fila["prompt"],
"pageId": setup.get("pageId"),
"pageName": page_name,
"rootId": root_id,
"seedBoardId": setup.get("seedBoardId"),
"turnos_usados": len(turnos),
"finish_reason": finish_reason,
"mensaje_final": final_content,
"auditoria": audit,
"metricas_crudas": {k: (list(v) if isinstance(v, tuple) else v) for k, v in raw.items()},
"execute_code_calls": exec_calls,
"execute_code_excepciones": exec_exceptions,
"code_payloads": code_payloads,
"export_shape_calls": export_calls,
"veto_hits": veto_hits,
"otros_patrones": otros_hits,
"png": str(png_path.relative_to(REPO_ROOT)) if png_path else None,
"turnos": turnos,
**resultado,
}
# ------------------------------------------------------------------------------------------
# Modo holdout: grep estatico barato, sin MCP y sin loop
# ------------------------------------------------------------------------------------------
def run_holdout(cfg, tools, system_prompt):
if not HOLDOUT_PATH.exists():
raise Gate5Error(
f"no existe {HOLDOUT_PATH.relative_to(REPO_ROOT)}; generalo con "
f"scripts/35_build_penpot_design_holdout.py antes de correr el modo holdout"
)
filas = []
with open(HOLDOUT_PATH, encoding="utf-8") as f:
for linea in f:
linea = linea.strip()
if linea:
filas.append(json.loads(linea))
print(f"[INFO] modo holdout: {len(filas)} prompts de diseno, sin tocar el MCP\n")
resultados = []
con_prohibido = 0
errores = 0
patrones = Counter()
for i, fila in enumerate(filas, start=1):
prompt = fila.get("prompt") or fila.get("user") or ""
messages = [{"role": "system", "content": system_prompt},
{"role": "user", "content": prompt}]
try:
choice = chat_completion(cfg, messages, tools)
except requests.RequestException as e:
# Una generacion que nunca ocurrio NO es una generacion limpia: se cuenta aparte y
# nunca entra en el denominador de la tasa (si no, un endpoint caido "aprueba").
errores += 1
resultados.append({"prompt": prompt, "error": f"{type(e).__name__}: {e}"})
continue
message = choice.get("message", {})
veto_total, otros_total, codes = [], [], []
for tool_call in message.get("tool_calls") or []:
nombre = tool_call.get("function", {}).get("name")
args, error_args = parse_tool_arguments(tool_call)
if error_args:
continue
if nombre == "execute_code":
code = args.get("code", "")
codes.append(code)
veto, otros = scan_forbidden(code)
veto_total.extend(veto)
otros_total.extend(otros)
elif nombre == "export_shape":
inventados = sorted(k for k in args if k not in EXPORT_SHAPE_ALLOWED_ARGS)
if inventados:
veto_total.append({"patron": "argumentos inventados en export_shape",
"fragmento": ", ".join(inventados)})
elif nombre not in ("high_level_overview", "penpot_api_info"):
veto_total.append({"patron": "tool inexistente", "fragmento": str(nombre)})
if veto_total:
con_prohibido += 1
for hit in veto_total:
patrones[hit["patron"]] += 1
resultados.append({
"prompt": prompt,
"id": fila.get("id"),
"tool_calls": [tc.get("function", {}).get("name")
for tc in (message.get("tool_calls") or [])],
"code_payloads": codes,
"veto_hits": veto_total,
"otros_patrones": otros_total,
"content": message.get("content") or "",
})
if i % 10 == 0:
print(f"[INFO] {i}/{len(filas)} prompts procesados")
total = len(resultados) - errores
if total <= 0:
raise Gate5Error(
f"ninguna de las {len(filas)} generaciones del holdout llego a completarse "
f"({errores} error(es) de request contra vLLM): no hay nada que medir"
)
tasa = con_prohibido / total
# Los errores de request bloquean la aprobacion a proposito: la puerta es el disparador de
# fallback antes de cuantizar, y un "aprueba" sobre generaciones faltantes no vale nada.
aprueba = tasa <= HOLDOUT_FORBIDDEN_RATE_MAX and errores == 0
print("\n=== Puerta 5 (modo holdout) -- tasa de API prohibida ===")
print(f" prompts medidos: {total} de {len(filas)}")
print(f" errores de request: {errores}")
print(f" con API prohibida: {con_prohibido}")
print(f" tasa: {100 * tasa:.1f}% (umbral <= {100 * HOLDOUT_FORBIDDEN_RATE_MAX:.0f}%)")
if patrones:
print(" desglose por patron:")
for patron, n in patrones.most_common():
print(f" - {patron}: {n}")
print(f"\n VEREDICTO: {'APRUEBA' if aprueba else 'NO APRUEBA -- disparador de fallback'}")
salida = REPO_ROOT / "data" / f"gate5_holdout_{cfg['GATE5_TAG']}.json"
salida.write_text(json.dumps({
"modo": "holdout",
"tag": cfg["GATE5_TAG"],
"modelo": cfg["GATE5_MODEL"],
"total": total,
"prompts_en_archivo": len(filas),
"errores_de_request": errores,
"con_api_prohibida": con_prohibido,
"tasa": tasa,
"umbral": HOLDOUT_FORBIDDEN_RATE_MAX,
"aprueba": aprueba,
"patrones": dict(patrones),
"resultados": resultados,
}, ensure_ascii=False, indent=2), encoding="utf-8")
print(f" resultados en {salida.relative_to(REPO_ROOT)}")
return 0 if aprueba else 1
# ------------------------------------------------------------------------------------------
# Modo completo
# ------------------------------------------------------------------------------------------
def load_prompts():
filas = []
with open(PROMPTS_PATH, encoding="utf-8") as f:
for lineno, linea in enumerate(f, start=1):
linea = linea.strip()
if not linea:
continue
try:
filas.append(json.loads(linea))
except json.JSONDecodeError as e:
raise Gate5Error(f"{PROMPTS_PATH.name}:{lineno}: JSON invalido: {e}")
faltantes = [f["id"] for f in filas if f["id"] not in THRESHOLDS]
if faltantes:
raise Gate5Error(f"prompts sin umbrales declarados en THRESHOLDS: {faltantes}")
return filas
def resumen(resultados, tag):
scores = [r["score"] for r in resultados]
media = sum(scores) / len(scores) if scores else 0.0
veto_limpio = sum(1 for r in resultados if not r["veto_violado"])
sobre_minimo = sum(1 for r in resultados if r["score"] >= PROMPT_SCORE_MIN)
flagship = next((r for r in resultados if r["id"] == FLAGSHIP_PROMPT_ID), None)
flagship_score = flagship["score"] if flagship else 0.0
print(f"\n=== Puerta 5 -- calidad de diseno en Penpot (tag={tag}) ===\n")
print(f" {'prompt':32s} {'dif':6s} {'score':>6s} {'pasa':>7s} {'grises':>7s} "
f"{'prohib':>7s} {'shapes':>7s} {'colores':>8s} {'textos':>7s}")
for r in resultados:
crudas = r["metricas_crudas"]
print(f" {r['id']:32s} {str(r['dificultad'])[:6]:6s} {r['score']:6.1f} "
f"{r['pasadas']:3d}/{r['aplicables']:<3d} {crudas['placeholderGreys']:7d} "
f"{crudas['forbiddenBehavior']:7d} {crudas['shapeCount']:7d} "
f"{crudas['distinctFillColors']:8d} {crudas['textsWithCharacters']:7d}")
condiciones = [
(f"score medio >= {MEAN_SCORE_MIN:.0f}", media >= MEAN_SCORE_MIN, f"{media:.1f}"),
("veto limpio en 8/8", veto_limpio == len(resultados), f"{veto_limpio}/{len(resultados)}"),
(f">= {MIN_PROMPTS_OVER_MIN}/8 prompts >= {PROMPT_SCORE_MIN:.0f}",
sobre_minimo >= MIN_PROMPTS_OVER_MIN, f"{sobre_minimo}/{len(resultados)}"),
(f"landing de pizzeria >= {PROMPT_SCORE_MIN:.0f}",
flagship_score >= PROMPT_SCORE_MIN, f"{flagship_score:.1f}"),
]
print("\n condiciones de aprobacion:")
for etiqueta, ok, valor in condiciones:
print(f" [{'ok ' if ok else 'NO '}] {etiqueta:40s} -> {valor}")
aprueba = all(ok for _e, ok, _v in condiciones)
print(f"\n VEREDICTO GLOBAL: {'APRUEBA' if aprueba else 'NO APRUEBA'}")
return aprueba, {
"score_medio": round(media, 2),
"veto_limpio": veto_limpio,
"prompts_sobre_minimo": sobre_minimo,
"score_landing_pizzeria": flagship_score,
"aprueba": aprueba,
"condiciones": [{"condicion": e, "ok": ok, "valor": v} for e, ok, v in condiciones],
}
def run_full(cfg, tools, system_prompt):
prompts = load_prompts()
check_disjunto(prompts)
self_check_gate_payloads()
mcp = PenpotMCP(cfg["PENPOT_MCP_URL"], cfg["PENPOT_MCP_TOKEN"])
mcp.handshake()
# Sonda barata: si el plugin no esta conectado, abortar ANTES de quemar los 8 prompts.
sonda, _es_error = mcp.execute_code("return { gate5Probe: true };")
if PLUGIN_NOT_CONNECTED in sonda:
raise PluginNotConnected(sonda)
print(f"[MCP] sonda de conectividad del plugin: {sonda[:120]}")
png_dir = REPO_ROOT / "data" / f"gate5_png_{cfg['GATE5_TAG']}"
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())
Reference in New Issue View Git Blame Copy Permalink