qwen3-6-lora/scripts/36_gate5_rescan_veto.py

"""Re-escanea resultados YA MEDIDOS de la puerta 5 con la lista de comportamiento prohibido
CORREGIDA (post hallazgo 54: `board.flex.appendChild` no esta roto, se saco del veto), sin volver
a correr el modelo ni tocar Penpot. Los payloads de `code` ya quedaron guardados en los JSON de
resultados, asi que el veto se puede recalcular offline con `scan_forbidden` / `score_prompt` del
propio scripts/35_gate5_penpot_design.py (fuente de verdad unica, nunca se reimplementa la logica).

    python3 scripts/36_gate5_rescan_veto.py

Escribe *-corrected.json junto a cada archivo de entrada. No modifica los originales.
"""
import importlib.util
import json
from pathlib import Path

REPO_ROOT = Path(__file__).resolve().parent.parent
GATE5_PATH = REPO_ROOT / "scripts" / "35_gate5_penpot_design.py"

FILES_HOLDOUT = [REPO_ROOT / "data" / "gate5_holdout_v2b-bf16-holdout.json"]
FILES_FULL = [
    REPO_ROOT / "data" / "gate5_results_v2b-bf16-full.json",
    REPO_ROOT / "data" / "gate5_results_v2b-bf16-generalization.json",
]


def load_gate5_module():
    spec = importlib.util.spec_from_file_location("gate5_penpot_design", GATE5_PATH)
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def rescan_holdout(gate5, path):
    data = json.loads(path.read_text(encoding="utf-8"))
    patrones = {}
    con_api_prohibida = 0
    cambios = []
    for r in data["resultados"]:
        veto_hits = []
        for code in r.get("code_payloads", []):
            veto, _otros = gate5.scan_forbidden(code)
            veto_hits.extend(veto)
        antes = bool(r.get("veto_hits"))
        r["veto_hits"] = veto_hits
        despues = bool(veto_hits)
        if antes != despues:
            cambios.append({"id": r.get("id"), "antes_vetado": antes, "despues_vetado": despues})
        if despues:
            con_api_prohibida += 1
            for hit in veto_hits:
                patrones[hit["patron"]] = patrones.get(hit["patron"], 0) + 1
    total = data["total"]
    tasa = con_api_prohibida / total if total else 0.0
    data["con_api_prohibida"] = con_api_prohibida
    data["tasa"] = tasa
    data["aprueba"] = tasa <= data["umbral"]
    data["patrones"] = patrones
    data["_rescan_veto_corregido"] = {
        "motivo": "hallazgo 54: board.flex.appendChild no esta roto, se sacio del veto de comportamiento prohibido",
        "cambios_por_prompt": cambios,
    }
    return data, cambios


def rescan_full(gate5, path):
    data = json.loads(path.read_text(encoding="utf-8"))
    cambios = []
    scores_antes = []
    scores_despues = []
    for p in data["prompts"]:
        score_antes = p["score"]
        scores_antes.append(score_antes)
        veto_hits = []
        for entry in p.get("code_payloads", []):
            code = entry["code"] if isinstance(entry, dict) else entry
            veto, _otros = gate5.scan_forbidden(code)
            veto_hits.extend(veto)
        p["veto_hits"] = veto_hits
        p["metricas_crudas"]["forbiddenBehavior"] = len(veto_hits)
        raw = p["metricas_crudas"]
        resultado = gate5.score_prompt(p["id"], raw)
        p["score"] = resultado["score"]
        p["pasadas"] = resultado["pasadas"]
        p["aplicables"] = resultado["aplicables"]
        p["veto_violado"] = resultado["veto_violado"]
        p["metricas"] = resultado["metricas"]
        scores_despues.append(resultado["score"])
        if resultado["score"] != score_antes:
            cambios.append({"id": p["id"], "score_antes": score_antes, "score_despues": resultado["score"]})
    score_medio_antes = sum(scores_antes) / len(scores_antes) if scores_antes else 0.0
    score_medio_despues = sum(scores_despues) / len(scores_despues) if scores_despues else 0.0
    veto_limpio_despues = sum(1 for p in data["prompts"] if not p["veto_violado"])
    prompts_60_despues = sum(1 for p in data["prompts"] if p["score"] >= 60)
    data["_rescan_veto_corregido"] = {
        "motivo": "hallazgo 54: board.flex.appendChild no esta roto, se sacio del veto de comportamiento prohibido",
        "score_medio_antes": round(score_medio_antes, 2),
        "score_medio_despues": round(score_medio_despues, 2),
        "veto_limpio_despues": f"{veto_limpio_despues}/{len(data['prompts'])}",
        "prompts_60_despues": f"{prompts_60_despues}/{len(data['prompts'])}",
        "cambios_por_prompt": cambios,
    }
    return data, cambios


def main():
    gate5 = load_gate5_module()

    for path in FILES_HOLDOUT:
        if not path.exists():
            print(f"[SKIP] {path} no existe")
            continue
        data, cambios = rescan_holdout(gate5, path)
        out = path.with_name(path.stem + "-corrected.json")
        out.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8")
        print(f"[HOLDOUT] {path.name}: con_api_prohibida={data['con_api_prohibida']} "
              f"tasa={data['tasa']:.4f} aprueba={data['aprueba']} patrones={data['patrones']}")
        print(f"          cambios: {cambios}")
        print(f"          -> {out}")

    for path in FILES_FULL:
        if not path.exists():
            print(f"[SKIP] {path} no existe")
            continue
        data, cambios = rescan_full(gate5, path)
        out = path.with_name(path.stem + "-corrected.json")
        out.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8")
        info = data["_rescan_veto_corregido"]
        print(f"[FULL] {path.name}: score_medio {info['score_medio_antes']} -> {info['score_medio_despues']} "
              f"veto_limpio_despues={info['veto_limpio_despues']} prompts_60_despues={info['prompts_60_despues']}")
        print(f"       cambios: {cambios}")
        print(f"       -> {out}")


if __name__ == "__main__":
    main()