aleleba/qwen3-6-lora
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
63da20c031a4ed83e7dc75c403da5398952b88b3
qwen3-6-lora/scripts/07_lint_penpot_code.py
T
aleleba 63da20c031 Phase 6.2: pin the training environment and make 10_train.py configurable 
The training container from phases 3-5 no longer exists and nothing in the
repo pinned its versions, so a rebuild could silently change either the
checkpoint key conversion (breaking adapter naming) or the assistant-mask
behaviour (training on system/user/tool tokens). requirements.train.txt
pins what matters and documents the two-phase install: llmcompressor
declares torch>=2.10.0 and the NGC image ships the 2.10.0a0 pre-release,
which pip's resolver reads as older, so it goes in with --no-deps.

Pre-flight verified against the merged bf16 checkpoint on spark:
01_inspect_modules.py prints model.layers.0.linear_attn.*, config.json is
sha256-identical to the base (93a4693f...), and the index keysets match
exactly (1045 tensors, 690 under model.language_model.layers.*, 0 under
model.layers.*). So PEFT will name adapter #2 the same way it named #1 and
ADAPTER_TO_CHECKPOINT_PREFIX in 20_merge_lora.py applies unchanged.

10_train.py: every path and hyperparameter moves to an env var, with the
phase 3 values as defaults so a bare run still reproduces phase 3 exactly.
Adds three guards that each cover a specific silent failure:
- abort if OUTPUT_DIR already holds an adapter, unless ALLOW_OVERWRITE=1.
  OUTPUT_DIR was hardcoded to out/lora-adapter, which is the provenance of
  the model currently in production.
- MAX_TOKENS aborts rather than truncates. There was no length filter at
  all, so one long design trajectory would blow the memory budget hours
  into a run; truncating would be worse, since it would silently cut
  assistant targets.
- assert use_rslora/use_dora/bias/modules_to_save. rsLoRA scales by
  alpha/sqrt(r), so an adapter trained with it would merge at 2.0 where
  11.3 belongs and pass every assertion in the merge script.
Also adds a config banner, a token-length histogram, and a per-bucket
assistant-mask ratio report.

New 07_lint_penpot_code.py hard-fails on the forbidden API patterns,
placeholder greys, fabricated penpot_api_info results, toy-shaped ids and
per-category coverage shortfalls. Error-recovery seeds legitimately need
the wrong pattern, so the exemption is derived mechanically rather than
declared by hand: a payload may contain a forbidden pattern only if its
tool result is a real error string from the allow-list and a later payload
in the same seed does the same thing without it.

Run against the 41 existing seeds it reproduces the diagnosis exactly:
110 problems, 36 unique payloads, 0% system messages, zero coverage of
addGridLayout/shadows/uploadMediaUrl/layoutChild, fabricated docs and
toy ids.
2026-07-30 17:01:41 +00:00

653 lines
30 KiB
Python
Raw Blame History

"""Fase 6: lint estatico del bucket de Penpot. Hard-fail sobre patrones de API prohibidos,
grises de placeholder, tool results fabricados y desbalance de cobertura.
Corre LOCALMENTE. No necesita GPU, ni el modelo, ni el tokenizer -- es a proposito: tiene que
poder correrse cientos de veces mientras se escriben los seeds, sin depender de spark.
python3 scripts/07_lint_penpot_code.py [archivo.jsonl ...]
Sin argumentos lintea data/raw/seeds/penpot.jsonl. Con argumentos lintea los archivos dados
(util para lintear las partes de data/raw/seeds/_parts/ antes de concatenarlas).
POR QUE EXISTE
--------------
Los 41 seeds viejos ensenaban tres formas de API que no existen (findShapeById de 2 argumentos,
shape.layout, createText() sin argumento) y contenian resultados de penpot_api_info fabricados a
mano que afirmaban hechos falsos como si los hubiera dicho el servidor. Nada lo detectaba: el
dataset se veia perfectamente valido. Este script convierte cada una de esas fallas en un error
mecanico, para que un corpus a medio autorar falle ruidosamente en vez de reproducir el problema.
`node --check` es opcional: si node no esta instalado se avisa y se saltea esa verificacion (el
resto del lint corre igual). Con node presente, cada payload se envuelve en una funcion async y
se verifica que parsee.
"""
import json
import re
import shutil
import subprocess
import sys
import tempfile
from collections import Counter
from pathlib import Path
REPO_ROOT = Path(__file__).resolve().parent.parent
SCHEMAS = REPO_ROOT / "data" / "schemas"
API_DOCS = SCHEMAS / "penpot_api_docs.md"
API_ERRORS = SCHEMAS / "penpot_errors.md"
SYSTEM_PROMPT = SCHEMAS / "penpot_system_prompt.md"
PENPOT_TOOLS = SCHEMAS / "penpot.json"
DEFAULT_TARGET = REPO_ROOT / "data" / "raw" / "seeds" / "penpot.jsonl"
# Los tool results de execute_code de este grupo muestran, a proposito, un estado inicial lleno
# de grises de placeholder: es el INPUT que el asistente audita y repara. Es el unico lugar donde
# un gris puede aparecer, y nunca en el `code` que el asistente escribe.
GREY_INPUT_GROUP = "B8"
# Umbrales de cobertura. Existen para que un corpus a medio autorar falle: sin esto, es
# perfectamente posible escribir 96 seeds que reproduzcan el desbalance actual (1 addFlexLayout,
# 0 addGridLayout, 0 layoutChild) y que el lint pase igual.
COVERAGE_MIN = {
"addGridLayout": 5,
"shadows": 8,
"uploadMediaUrl": 6,
"layoutChild.horizontalSizing": 6,
"addFlexLayout": 12,
"borderRadius": 8,
"fillColorGradient": 4,
"library.local": 5,
"export_shape": 8,
}
MIN_UNIQUE_PAYLOADS = 130
MAX_HIGH_LEVEL_OVERVIEW_CALLS = 2
SYSTEM_SHARE_RANGE = (0.20, 0.45) # ~30% objetivo, con margen
UUID4_RE = re.compile(r"^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$", re.I)
HEX_RE = re.compile(r"#[0-9a-fA-F]{6}\b")
TOY_ID_RE = re.compile(r"^(r|fb|s|shape|board|txt|t)[-_]?\d{1,3}$", re.I)
# --------------------------------------------------------------------------------------------
# Patrones prohibidos en los payloads de `code`
# --------------------------------------------------------------------------------------------
# Cada entrada: (nombre, regex, explicacion). La explicacion se imprime con el hallazgo, porque
# un lint que solo dice "linea 42: PATRON_7" obliga a ir a leer el lint para entender el error.
FORBIDDEN = [
(
"findShapeById con 2 argumentos",
re.compile(r"findShapeById\s*\([^)]*,"),
"findShapeById tiene aridad 1. La forma de 2 argumentos devuelve null SIN lanzar, y "
"revienta en la linea siguiente. Usar findShapeById(id).",
),
(
"propiedad .layout inexistente",
re.compile(r"\.layout\b(?!Child|Cell)"),
"shape.layout no existe ('layout' in shape === false). Son board.flex y board.grid. "
"La clave `layout` que ves en la salida de shapeStructure() es del OUTPUT del helper, "
"no una propiedad del shape.",
),
(
"flex.appendChild",
re.compile(r"\.flex\s*\.\s*appendChild\s*\("),
"board.flex.appendChild esta roto. Usar board.appendChild(shape), en orden visual.",
),
(
"fontSize/fontWeight/lineHeight/letterSpacing numerico",
re.compile(r"\.(fontSize|fontWeight|lineHeight|letterSpacing)\s*=\s*-?\d"),
"Son strings: fontSize = '48', no fontSize = 48.",
),
(
"textAlign",
re.compile(r"\.textAlign\s*="),
"Asignar textAlign LANZA (`Cannot add property textAlign, object is not extensible`) y "
"mata todo el execute_code. La propiedad es `align`.",
),
(
"propiedad color en Text",
re.compile(r"\b(text|txt|t|label|title|heading)\d*\s*\.\s*color\s*="),
"Text no tiene propiedad `color` y el objeto no es extensible: lanza. El color va en fills.",
),
(
"import_image / importImage / createImage / filePath",
re.compile(r"\b(importImage|import_image|createImage\s*\(|filePath)\b"),
"No existen en este deployment. Imagenes: await penpot.uploadMediaUrl(name, url).",
),
(
"asignacion a propiedad read-only",
re.compile(r"\.(width|height|parentX|parentY|boardX|boardY|bounds)\s*=(?!=)"),
"width/height/parentX/parentY/boardX/boardY/bounds son READ-ONLY. Usar resize(w,h) y "
"penpotUtils.setParentXY(shape, x, y).",
),
(
"shorthand gap",
re.compile(r"\.gap\s*="),
"No existe el shorthand `gap`. Son rowGap y columnGap.",
),
(
"Shadow.color como Fill",
re.compile(r"shadows\s*=\s*\[[^\]]*?color\s*:\s*\{[^}]*fillColor"),
"Shadow.color es un Color ({color, opacity}), no un Fill ({fillColor, fillOpacity}).",
),
(
"typography.setFont",
re.compile(r"\.setFont\s*\("),
"setFont figura en el tipo pero NO existe en el runtime (`t.setFont is not a function`). "
"Setear las propiedades de tipografia una por una.",
),
(
"createText sin argumento util",
re.compile(r"createText\s*\(\s*(''|\"\")?\s*\)"),
"createText() sin argumento y createText('') devuelven null. Es la forma que usa el "
"ejemplo de la documentacion oficial, y es la causa directa de la falla en produccion.",
),
(
"generateStyle con withChildren",
re.compile(r"withChildren\s*:"),
"La opcion real es `includeChildren`. `withChildren` (que usa el overview) se ignora en "
"silencio y el CSS sale sin los hijos.",
),
(
"layoutChild.horizontalSizing = 'fixed' o 'fit-content'",
re.compile(r"layoutChild\s*\.\s*(horizontal|vertical)Sizing\s*=\s*['\"](fixed|fit-content)['\"]"),
"Los valores de layoutChild son 'fill' | 'auto' | 'fix'. 'fit-content' es vocabulario de "
"FlexLayout (aplica al board contenedor), y 'fixed' no existe en ninguno de los dos.",
),
]
# `appendChild` solo es correcto sobre un board con flex (o `grid.appendChild(s, r, c)`). Sobre un
# padre sin layout hay que usar insertChild. No se puede resolver estaticamente en general, asi que
# la heuristica es: si el payload usa `X.appendChild(` con un solo argumento, tiene que haber
# evidencia de flex en el mismo payload.
APPEND_ONE_ARG_RE = re.compile(r"(\w+)\s*\.\s*appendChild\s*\(\s*[^,()]+\s*\)")
FLEX_EVIDENCE_RE = re.compile(r"addFlexLayout\s*\(|\.flex\b")
GRID_APPEND_RE = re.compile(r"\.grid\s*\.\s*appendChild\s*\(")
def is_placeholder_grey(hex_str):
"""R2: saturacion <= 10 Y 100 <= max <= 220. Los casi-negros y casi-blancos pasan."""
r = int(hex_str[1:3], 16)
g = int(hex_str[3:5], 16)
b = int(hex_str[5:7], 16)
mx, mn = max(r, g, b), min(r, g, b)
sat = 0 if mx == 0 else 100.0 * (mx - mn) / mx
return sat <= 10 and 100 <= mx <= 220
def load_jsonl(path):
rows = []
with open(path, encoding="utf-8") as f:
for lineno, raw in enumerate(f, start=1):
raw = raw.strip()
if not raw:
continue
try:
rows.append((lineno, json.loads(raw), len(raw)))
except json.JSONDecodeError as e:
raise SystemExit(f"[FATAL] {path.name}:{lineno}: JSON invalido: {e}")
return rows
def load_doc_lines(path):
"""Todas las lineas no vacias del archivo de captura, para el chequeo de subconjunto."""
return [ln.rstrip() for ln in path.read_text(encoding="utf-8").splitlines() if ln.strip()]
def extract_system_block():
"""El bloque verbatim de penpot_system_prompt.md, seccion 'Bloque completo, verbatim'."""
text = SYSTEM_PROMPT.read_text(encoding="utf-8")
marker = "## Bloque completo, verbatim"
idx = text.index(marker)
fence_start = text.index("```", idx) + 3
fence_start = text.index("\n", fence_start) + 1
fence_end = text.index("```", fence_start)
return text[fence_start:fence_end].rstrip("\n")
def extract_error_strings():
"""Los strings de error verbatim de penpot_errors.md: todo `codigo` dentro de una celda de
tabla de la columna 'Mensaje verbatim', mas los de las tablas de fallos silenciosos."""
text = API_ERRORS.read_text(encoding="utf-8")
out = set()
for line in text.splitlines():
if not line.startswith("|"):
continue
cells = [c.strip() for c in line.strip().strip("|").split("|")]
for cell in cells:
for m in re.finditer(r"`([^`]+)`", cell):
out.add(m.group(1))
return out
def extract_return_keys(code):
"""Claves de primer nivel del ultimo `return { ... }` del payload.
Devuelve None si el payload no termina en un return de objeto literal (p.ej. retorna un
array, una expresion, o nada) -- en ese caso el chequeo de claves no aplica.
"""
idx = code.rfind("return {")
if idx == -1:
idx = code.rfind("return{")
if idx == -1:
return None
start = code.index("{", idx)
depth = 0
keys = []
i = start
in_str = None
buf_depth_zero = []
while i < len(code):
ch = code[i]
if in_str:
if ch == "\\":
i += 2
continue
if ch == in_str:
in_str = None
elif ch in "\"'`":
in_str = ch
elif ch in "{[(":
depth += 1
elif ch in "}])":
depth -= 1
if depth == 0 and ch == "}":
break
elif depth == 1:
buf_depth_zero.append((i, ch))
i += 1
if depth != 0:
return None
# Reconstruir el nivel 1 y sacar las claves antes de cada ':'
level1 = "".join(ch for _, ch in buf_depth_zero)
for part in level1.split(","):
if ":" not in part:
# shorthand `{ foo, bar }`
name = part.strip()
if re.fullmatch(r"[A-Za-z_$][\w$]*", name):
keys.append(name)
continue
key = part.split(":", 1)[0].strip().strip("'\"")
if re.fullmatch(r"[A-Za-z_$][\w$]*", key):
keys.append(key)
return keys or None
def node_check(payloads):
"""Verifica que cada payload parsee, envuelto en una funcion async (el servidor evalua el
codigo como cuerpo de funcion, por eso `await` y `return` de nivel superior son validos)."""
node = shutil.which("node")
if not node:
print("[WARN] node no esta instalado -- se saltea la verificacion de sintaxis JS")
return []
problems = []
with tempfile.TemporaryDirectory() as td:
for tag, code in payloads:
f = Path(td) / "chk.mjs"
f.write_text("async function __penpot_payload__(){\n" + code + "\n}\n", encoding="utf-8")
r = subprocess.run([node, "--check", str(f)], capture_output=True, text=True)
if r.returncode != 0:
first = (r.stderr.strip().splitlines() or ["error desconocido"])
msg = next((ln for ln in first if "SyntaxError" in ln), first[-1])
problems.append(f"{tag}: node --check fallo: {msg.strip()}")
return problems
def main():
targets = [Path(a) for a in sys.argv[1:]] or [DEFAULT_TARGET]
for p in targets:
if not p.exists():
raise SystemExit(f"[FATAL] no existe: {p}")
doc_lines = set(load_doc_lines(API_DOCS))
doc_order = load_doc_lines(API_DOCS)
doc_index = {}
for i, ln in enumerate(doc_order):
doc_index.setdefault(ln, []).append(i)
error_strings = extract_error_strings()
system_block = extract_system_block()
canonical_tools = {t["name"]: t for t in json.loads(PENPOT_TOOLS.read_text(encoding="utf-8"))}
problems = []
warnings = []
payloads = [] # (tag, code) para node --check
payload_texts = [] # para unicidad
coverage = Counter()
n_seeds = 0
n_with_system = 0
n_overview_calls = 0
n_intentional_mistakes = 0
max_line_len = 0
grupo_counts = Counter()
for path in targets:
for lineno, ex, raw_len in load_jsonl(path):
n_seeds += 1
tag = f"{path.name}:{lineno}"
max_line_len = max(max_line_len, raw_len)
grupo = ex.get("meta", {}).get("grupo", "?")
grupo_counts[grupo] += 1
if ex.get("meta", {}).get("bucket") != "penpot":
problems.append(f"{tag}: meta.bucket debe ser 'penpot'")
msgs = ex.get("messages", [])
if not msgs:
problems.append(f"{tag}: sin messages")
continue
# ---- mensaje system verbatim -------------------------------------------------
if msgs[0].get("role") == "system":
n_with_system += 1
if msgs[0].get("content", "").strip() != system_block.strip():
problems.append(
f"{tag}: el mensaje system no es el bloque verbatim de "
f"penpot_system_prompt.md (parafraseado o recortado)"
)
# ---- tools copiadas byte a byte ----------------------------------------------
for tool in ex.get("tools") or []:
name = tool.get("name")
if name not in canonical_tools:
problems.append(f"{tag}: tool '{name}' no existe en data/schemas/penpot.json")
continue
canon = canonical_tools[name]
if tool.get("description") != canon.get("description"):
problems.append(
f"{tag}: la description de la tool '{name}' fue recortada o reformulada; "
f"tiene que ser byte a byte la de penpot.json"
)
if tool.get("parameters") != canon.get("parameters"):
problems.append(f"{tag}: el JSON Schema de '{name}' no coincide con penpot.json")
# ---- biyeccion tool_call_id <-> mensajes tool --------------------------------
call_ids, result_ids = [], []
for m in msgs:
for tc in m.get("tool_calls") or []:
call_ids.append(tc.get("id"))
if m.get("role") == "tool":
result_ids.append(m.get("tool_call_id"))
if sorted(filter(None, call_ids)) != sorted(filter(None, result_ids)):
problems.append(
f"{tag}: los tool_call_id no son una biyeccion "
f"(llamadas={call_ids} resultados={result_ids})"
)
if len(set(call_ids)) != len(call_ids):
problems.append(f"{tag}: tool_call_id duplicados: {call_ids}")
# ---- assistant: reasoning_content y content --------------------------------
for i, m in enumerate(msgs):
if m.get("role") != "assistant":
continue
if not (m.get("reasoning_content") or "").strip():
problems.append(f"{tag}: turno assistant #{i} sin reasoning_content")
if m.get("tool_calls") and (m.get("content") or "") != "":
problems.append(
f"{tag}: turno assistant #{i} tiene tool_calls y content no vacio "
f"(el content visible va en el ULTIMO assistant, sin tool_calls)"
)
if msgs[-1].get("role") != "assistant" or msgs[-1].get("tool_calls"):
problems.append(f"{tag}: la trayectoria no cierra con un assistant sin tool_calls")
# ---- payloads de code ---------------------------------------------------------
# Los seeds de recuperacion de error (grupos A1 y D) TIENEN que contener el patron
# equivocado: es el material que ensena a diagnosticarlo. La excepcion no se declara
# a mano con una bandera en meta (eso seria un escape hatch que apaga el lint), se
# deriva mecanicamente: un payload puede traer un patron prohibido si y solo si
# (a) su tool result es un string de error de la allow-list, y
# (b) un payload POSTERIOR del mismo seed hace lo mismo SIN el patron.
# Es decir: el error tiene que haber sido real y tiene que haber sido corregido.
error_result_ids = {
m.get("tool_call_id")
for m in msgs
if m.get("role") == "tool" and (m.get("content") or "").strip() in error_strings
}
ordered_calls = []
for m in msgs:
for tc in m.get("tool_calls") or []:
if tc.get("function", {}).get("name") == "execute_code":
args = tc.get("function", {}).get("arguments")
ordered_calls.append(
(tc.get("id"), args.get("code", "") if isinstance(args, dict) else "")
)
def is_corrected_mistake(call_id, rx):
"""El patron aparece en una llamada que fallo, y una llamada posterior lo evita."""
if call_id not in error_result_ids:
return False
seen = False
for cid, code_ in ordered_calls:
if seen and not rx.search(code_):
return True
if cid == call_id:
seen = True
return False
code_by_call = {}
for m in msgs:
for tc in m.get("tool_calls") or []:
fn = tc.get("function", {})
args = fn.get("arguments")
if not isinstance(args, dict):
problems.append(
f"{tag}: tool_calls.function.arguments tiene que ser un dict JSON, "
f"no {type(args).__name__}"
)
continue
if fn.get("name") == "high_level_overview":
n_overview_calls += 1
if fn.get("name") == "export_shape":
coverage["export_shape"] += 1
for k in args:
if k not in ("shapeId", "format", "mode"):
problems.append(
f"{tag}: export_shape con el argumento inventado '{k}' "
f"(el schema solo acepta shapeId/format/mode)"
)
if fn.get("name") != "execute_code":
continue
code = args.get("code", "")
code_by_call[tc.get("id")] = code
payloads.append((tag, code))
payload_texts.append(code)
# patrones prohibidos
for pname, rx, why in FORBIDDEN:
if not rx.search(code):
continue
if is_corrected_mistake(tc.get("id"), rx):
n_intentional_mistakes += 1
continue
for mt in rx.finditer(code):
frag = code[max(0, mt.start() - 30):mt.end() + 30].replace("\n", " ")
problems.append(f"{tag}: [{pname}] ...{frag}...\n -> {why}")
# appendChild sobre receptor sin evidencia de flex
if (APPEND_ONE_ARG_RE.search(code) and not FLEX_EVIDENCE_RE.search(code)
and not GRID_APPEND_RE.search(code)):
problems.append(
f"{tag}: usa `X.appendChild(shape)` sin ninguna evidencia de flex en el "
f"payload. appendChild de 1 argumento solo es correcto en boards con "
f"flex; para un padre sin layout va "
f"parent.insertChild(parent.children.length, shape)."
)
# console.log de algo que tambien se retorna
ret_keys = extract_return_keys(code) or []
for logged in re.findall(r"console\.log\s*\(\s*([A-Za-z_$][\w$]*)", code):
if logged in ret_keys:
problems.append(
f"{tag}: console.log('{logged}') de algo que tambien se retorna "
f"-- el servidor lo prohibe explicitamente (llega duplicado)"
)
# R2: grises de placeholder en el codigo (nunca permitidos, ni en B8)
for hx in HEX_RE.findall(code):
if is_placeholder_grey(hx):
problems.append(
f"{tag}: gris de placeholder {hx} en el `code` (invariante R2). "
f"El gris solo puede aparecer como INPUT en los tool results del "
f"grupo {GREY_INPUT_GROUP}."
)
# cobertura
for key in ("addGridLayout", "addFlexLayout", "uploadMediaUrl",
"borderRadius", "fillColorGradient"):
if key in code:
coverage[key] += 1
if re.search(r"shadows\s*=", code):
coverage["shadows"] += 1
if re.search(r"layoutChild\s*\.\s*horizontalSizing", code):
coverage["layoutChild.horizontalSizing"] += 1
if "library.local" in code:
coverage["library.local"] += 1
# ---- tool results -------------------------------------------------------------
for m in msgs:
if m.get("role") != "tool":
continue
name = m.get("name")
content = m.get("content") or ""
cid = m.get("tool_call_id")
if name in ("penpot_api_info", "high_level_overview"):
# subconjunto de lineas de la captura, en orden original
lines = [ln.rstrip() for ln in content.splitlines() if ln.strip()]
missing = [ln for ln in lines if ln not in doc_lines]
if missing:
problems.append(
f"{tag}: resultado de {name} con {len(missing)} linea(s) ausentes de "
f"penpot_api_docs.md -- documentacion FABRICADA. Primera: "
f"{missing[0][:110]!r}"
)
else:
last = -1
for ln in lines:
positions = [i for i in doc_index[ln] if i > last]
if not positions:
problems.append(
f"{tag}: resultado de {name} con lineas fuera del orden "
f"original de la captura (en {ln[:70]!r})"
)
break
last = positions[0]
elif name == "execute_code":
code = code_by_call.get(cid)
stripped = content.strip()
looks_json = stripped.startswith("{") or stripped.startswith("[")
if not looks_json:
# tiene que ser un error real de la allow-list
if stripped not in error_strings:
problems.append(
f"{tag}: resultado de execute_code que no es JSON y no esta en "
f"penpot_errors.md: {stripped[:110]!r}"
)
else:
try:
parsed = json.loads(stripped)
except json.JSONDecodeError as e:
problems.append(f"{tag}: resultado de execute_code con JSON invalido: {e}")
parsed = None
if isinstance(parsed, dict) and code is not None:
ret_keys = extract_return_keys(code)
if ret_keys is not None:
if sorted(parsed.keys()) != sorted(ret_keys):
problems.append(
f"{tag}: las claves del resultado {sorted(parsed.keys())} "
f"no coinciden con las del `return {{...}}` del payload "
f"{sorted(ret_keys)}"
)
# ids de juguete
for val in re.findall(r'"([^"]{1,24})"', stripped):
if TOY_ID_RE.fullmatch(val):
warnings.append(
f"{tag}: id con forma de juguete {val!r} en un tool result; "
f"los ids reales tienen forma de UUIDv4"
)
for val in re.findall(r'"(?:id|shapeId|boardId|rootId)"\s*:\s*"([^"]+)"', stripped):
if not UUID4_RE.match(val):
problems.append(
f"{tag}: id {val!r} en un tool result no tiene forma de UUIDv4 "
f"-- entrena al modelo a esperar ids de juguete"
)
# R2 en tool results: solo el grupo de reparacion puede traer grises
if grupo != GREY_INPUT_GROUP:
for hx in HEX_RE.findall(content):
if is_placeholder_grey(hx):
problems.append(
f"{tag}: gris de placeholder {hx} en un tool result del grupo "
f"{grupo}; solo el grupo {GREY_INPUT_GROUP} puede traerlos, "
f"como estado inicial a reparar"
)
# ---- chequeos globales -----------------------------------------------------------------
dupes = [c for c, n in Counter(payload_texts).items() if n > 1]
if dupes:
problems.append(
f"[GLOBAL] {len(dupes)} payload(s) de `code` duplicados exactos. Primero: "
f"{dupes[0][:120]!r}"
)
unique_payloads = len(set(payload_texts))
is_full_corpus = len(targets) == 1 and targets[0].resolve() == DEFAULT_TARGET.resolve()
print(f"\n=== SEEDS: {n_seeds} | payloads de code: {len(payload_texts)} "
f"(unicos: {unique_payloads}) ===")
print(f"=== linea mas larga: {max_line_len} caracteres ===")
print(f"=== grupos: {dict(sorted(grupo_counts.items()))} ===")
share = (n_with_system / n_seeds) if n_seeds else 0
print(f"=== seeds con mensaje system: {n_with_system}/{n_seeds} ({share:.0%}) ===")
print(f"=== errores intencionales (patron prohibido + error real + correccion posterior): "
f"{n_intentional_mistakes} ===")
print("=== cobertura por categoria ===")
for key, minimum in sorted(COVERAGE_MIN.items()):
got = coverage[key]
mark = "ok " if got >= minimum else "BAJO"
print(f" [{mark}] {key}: {got} (minimo {minimum})")
if is_full_corpus:
for key, minimum in COVERAGE_MIN.items():
if coverage[key] < minimum:
problems.append(
f"[COBERTURA] '{key}': {coverage[key]} seeds, minimo {minimum}. Un corpus a "
f"medio autorar reproduce el desbalance actual sin que nada lo note."
)
if unique_payloads < MIN_UNIQUE_PAYLOADS:
problems.append(
f"[GLOBAL] solo {unique_payloads} payloads unicos de execute_code, minimo "
f"{MIN_UNIQUE_PAYLOADS} (hoy el dataset viejo tiene 36)"
)
if not SYSTEM_SHARE_RANGE[0] <= share <= SYSTEM_SHARE_RANGE[1]:
problems.append(
f"[GLOBAL] share de seeds con mensaje system = {share:.0%}, fuera del rango "
f"{SYSTEM_SHARE_RANGE[0]:.0%}-{SYSTEM_SHARE_RANGE[1]:.0%} (objetivo ~30%)"
)
if n_overview_calls > MAX_HIGH_LEVEL_OVERVIEW_CALLS:
problems.append(
f"[GLOBAL] {n_overview_calls} llamadas a high_level_overview, maximo "
f"{MAX_HIGH_LEVEL_OVERVIEW_CALLS} (su propia descripcion prohibe llamarlo dos veces)"
)
else:
print("[INFO] modo parcial (no se lintea el corpus completo): los umbrales globales de "
"cobertura, unicidad y share de system NO se aplican")
problems.extend(node_check(payloads))
if warnings:
print(f"\n=== {len(warnings)} ADVERTENCIA(S) ===")
for w in warnings[:40]:
print(f" ! {w}")
if problems:
print(f"\n=== {len(problems)} PROBLEMA(S) ===")
for p in problems:
print(f" - {p}")
print(f"\n[LINT FAIL] {len(problems)} problema(s)")
sys.exit(1)
print("\n[LINT OK] 0 problemas")
sys.exit(0)
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink