Blame for scripts/70_suite_v090.py

#!/usr/bin/env python3
"""70_suite_v090.py — v0.9.0: VEREDITO CONSOLIDADO das críticas Qwen3.8 (lote 2).

Herda os gates da v0.8.0 (suite_v080_final.json) e adiciona os blocos v0.9.0
(análise/prova/teste em r-34):
  C1. Configs tipadas nas demos — erro de tipo agora é ERRO, não default mudo
  C2. Oráculo candle_nn::optim::AdamW (cargo) + medição de custo relativo
  C3. Entrada numpy (rust-numpy) — medida de fronteira lista vs array
  C4. SIMD despacho em runtime: bit-identidade (cargo) + bench dedicado
  B3. Distâncias: paisagem medida (escalar/SoA/RM/matmul) — gate de honestidade
  B1. GNG por entropia: portão com piso (não-fome) + telemetria
  C.  Thompson Sampling vs UCB1 (regret) + determinismo
  C.  JEPA fração adaptativa: 1 braço = fixa bit-a-bit; multi-braço explora
"""
import json, os, subprocess, sys, time


def main():
    pasta = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
    R = os.path.join(pasta, "resultados")
    rel = {"versao": "0.9.0", "blocos": {}, "gates": {}, "medicoes": {}}

    def carrega(arq):
        p = os.path.join(R, arq)
        if os.path.exists(p):
            with open(p) as f:
                return json.load(f)
        return None

    # ---- legado v0.8.0
    legado = carrega("suite_v080_final.json")
    if legado:
        g_legado = legado.get("gates", legado)
        ok = sum(1 for v in g_legado.values() if v is True)
        rel["blocos"]["legado_v080"] = f"{ok}/{len(g_legado)}"
        for k, v in g_legado.items():
            rel["gates"][f"legado_{k}"] = v

    import khtst_rust as k
    import numpy as np

    # ================= C1: configs tipadas =================
    rng = np.random.default_rng(11)
    dados4d = rng.normal(0, 1, (200, 4)).tolist()
    # cfg VÁLIDO continua funcionando (comportamento preservado)
    r_ok = k.treinar_som4d(dados4d, json.dumps({"nx": 3, "ny": 3, "nz": 2, "nt": 2, "epocas": 5}))
    rel["gates"]["c1_cfg_valido_funciona"] = r_ok["celulas"] == 36
    # cfg com TIPO ERRADO agora é erro explícito (antes: default silencioso)
    try:
        k.treinar_som4d(dados4d, '{"nx": "grande"}')
        rel["gates"]["c1_tipo_errado_rejeitado"] = False
    except Exception:
        rel["gates"]["c1_tipo_errado_rejeitado"] = True
    rel["blocos"]["c1_cfg_tipada"] = "2/2" if (
        rel["gates"]["c1_cfg_valido_funciona"] and rel["gates"]["c1_tipo_errado_rejeitado"]
    ) else "0/2"

    # ================= C2/C4: bench dedicado (binário) =================
    bench_path = os.path.join(pasta, "target", "release", "bench_v090")
    bench = None
    if os.path.exists(bench_path):
        try:
            res = subprocess.run([bench_path], capture_output=True, text=True, timeout=900)
            bench = json.loads(res.stdout)["bench_v090"]
        except Exception as e:
            print(f"bench falhou: {e}")
    if bench:
        rel["medicoes"]["bench_v090"] = bench
        rel["gates"]["c4_simd_bit_identico"] = bench.get("adamw_bit_identico") is True
        rel["gates"]["c4_avx512_detectado"] = bench.get("avx512_disponivel") is True
        rel["gates"]["c2_nosso_adamw_nao_pior_candle"] = (
            bench.get("nosso_vs_candle_nn", 0.0) >= 1.0
        )
        rel["gates"]["c4_alinhamento_garantido_64b"] = bench.get("alinhado_endereco_mod64") is True
        rel["gates"]["d3_bf16_nao_e_gratis"] = bench.get("bf16_vs_f32", 1.0) < 1.0
        rel["blocos"]["c2c4_bench"] = "5/5" if all(
            rel["gates"][g] for g in (
                "c4_simd_bit_identico", "c4_avx512_detectado",
                "c2_nosso_adamw_nao_pior_candle", "c4_alinhamento_garantido_64b",
                "d3_bf16_nao_e_gratis")
        ) else "0/5"
        # B3: paisagem de distâncias — o gate é a PRESENÇA da medição honesta
        # (o próprio resultado refuta o "~8×" da proposta; não exigimos ganho)
        rel["gates"]["b3_paisagem_medi_da"] = all(
            x in bench for x in ("bmu_vecvec_ms", "bmu_soa_ms", "bmu_rm_ms", "bmu_matmul_ms")
        )
        rel["gates"]["b3_escalar_argmin_nao_perde"] = (
            bench.get("bmu_vecvec_ms", 1e9) <= min(
                bench.get("bmu_soa_ms", 1e9), bench.get("bmu_rm_ms", 1e9))
        )
    else:
        rel["blocos"]["c2c4_bench"] = "bench ausente"
        for g in ("c4_simd_bit_identico", "c4_avx512_detectado",
                  "c2_nosso_adamw_nao_pior_candle", "c4_alinhamento_garantido_64b",
                  "d3_bf16_nao_e_gratis", "b3_paisagem_medi_da", "b3_escalar_argmin_nao_perde"):
            rel["gates"][g] = False

    # ================= C3: fronteira Python: lista vs numpy =================
    # método ISOLADO (v090-7): funções de diagnóstico _fronteira_* medem SÓ a
    # conversão na fronteira (o método anterior via diferença de treinos com
    # teto por tempo media ruído — marginal negativa, sem sentido).
    n_seq, n_tok = 2048, 256
    seqs_list = [[(i * 31 + j * 7) % 16384 for j in range(n_tok)] for i in range(n_seq)]
    seqs_arr = np.array(seqs_list, dtype=np.uint32)
    # aquecimento
    k._fronteira_lista(seqs_list[:8]); k._fronteira_np(seqs_arr[:8])
    t0 = time.perf_counter()
    for _ in range(5):
        total_l = k._fronteira_lista(seqs_list)
    t_lista = (time.perf_counter() - t0) / 5
    t0 = time.perf_counter()
    for _ in range(5):
        total_n = k._fronteira_np(seqs_arr)
    t_np = (time.perf_counter() - t0) / 5
    assert total_l == total_n == n_seq * n_tok, "totais divergem"
    rel["medicoes"]["c3_fronteira"] = {
        "lista_ms_524k_tokens": t_lista * 1e3,
        "numpy_ms_524k_tokens": t_np * 1e3,
        "razao_lista_vs_numpy": (t_lista / t_np) if t_np > 0 else float("inf"),
        "metodo": "funções de diagnóstico isoladas (_fronteira_lista/_fronteira_np)",
    }
    rel["gates"]["c3_numpy_nao_pior"] = t_np <= t_lista * 1.05
    rel["blocos"]["c3_numpy"] = "1/1" if rel["gates"]["c3_numpy_nao_pior"] else "0/1"

    # ================= B1: GNG entropia (não-fome) =================
    dados_gng = np.vstack([
        rng.normal(0, 0.4, (300, 4)),
        rng.normal(3, 0.4, (300, 4)),
    ]).tolist()
    cfg_gng = json.dumps({"passos": 3000, "max_vertices": 30})
    g_pad = k.treinar_gng(dados_gng, cfg_gng)
    g_piso = k.treinar_gng(dados_gng, json.dumps({
        "passos": 3000, "max_vertices": 30, "criterio": "entropia_com_piso"}))
    rel["medicoes"]["b1_gng"] = {
        "padrao": {"vertices": g_pad["vertices"], "eq": g_pad["eq_medio"], "insercoes": g_pad["insercoes"]},
        "entropia_piso": {"vertices": g_piso["vertices"], "eq": g_piso["eq_medio"],
                          "insercoes": g_piso["insercoes"],
                          "bloqueadas": g_piso["insercoes_bloqueadas_entropia"]},
    }
    # não-fome: o piso garante crescimento quando QE ainda é alto
    rel["gates"]["b1_gng_piso_insere"] = g_piso["insercoes"] >= 5
    # e o bloqueio existe (o portão de entropia atua)
    rel["gates"]["b1_gng_portao_ativo"] = g_piso["insercoes_bloqueadas_entropia"] > 0
    rel["blocos"]["b1_gng_entropia"] = "2/2" if (
        rel["gates"]["b1_gng_piso_insere"] and rel["gates"]["b1_gng_portao_ativo"]
    ) else "0/2"

    # ================= C: Thompson Sampling =================
    ts = k.demo_thompson(semente=101, passos=4000)
    rel["medicoes"]["thompson"] = ts["arrependimento_medio"]
    for g in ("ts_aprende", "ucb1_aprende", "melhor_braço_domina_ts"):
        rel["gates"][f"c_ts_{g}"] = ts["gates"][g]
    rel["blocos"]["c_thompson"] = "3/3" if all(
        rel["gates"][f"c_ts_{g}"] for g in ("ts_aprende", "ucb1_aprende", "melhor_braço_domina_ts")
    ) else "0/3"

    # ================= C: JEPA fração adaptativa =================
    dados_j = rng.normal(0, 1, (128, 8)).tolist()
    # (1) um braço == fração fixa: bit-a-bit (cargo test cobre; aqui smoke)
    adapt = k.treinar_jepa_adaptativa(
        dados_j, [16, 8], "image", [0.35],
        json.dumps({"epocas": 15, "d_emb": 16, "semente": 42}))
    fixo = k.treinar_jepa(dados_j, [16, 8], "image",
                          json.dumps({"epocas": 15, "d_emb": 16, "semente": 42}))
    dif = abs(adapt["perda_pred_final"] - fixo["perda_pred_final"])
    rel["gates"]["c_jepa_um_braco_bit_a_bit"] = dif < 1e-12
    # (2) multi-braço explora todos
    adapt4 = k.treinar_jepa_adaptativa(
        dados_j, [16, 8], "image", [0.2, 0.35, 0.5, 0.65],
        json.dumps({"epocas": 60, "d_emb": 16, "semente": 42}))
    rel["gates"]["c_jepa_multibraco_explora"] = all(j > 0 for j in adapt4["jogadas_por_braco"])
    rel["gates"]["c_jepa_sem_colapso"] = adapt4["colapso_evitado"]
    rel["medicoes"]["c_jepa_adaptativa"] = {
        "jogadas": adapt4["jogadas_por_braco"],
        "posterior": adapt4["posterior"],
        "queda_pct": adapt4["queda_pct"],
    }
    rel["blocos"]["c_jepa_adaptativa"] = "3/3" if all(
        rel["gates"][g] for g in (
            "c_jepa_um_braco_bit_a_bit", "c_jepa_multibraco_explora", "c_jepa_sem_colapso")
    ) else "0/3"

    # ================= painel v090 =================
    rel["medicoes"]["simd_relatorio"] = k.simd_relatorio()

    # ================= cargo (contagem) =================
    try:
        res = subprocess.run(
            ["cargo", "test", "--release", "--manifest-path", os.path.join(pasta, "Cargo.toml")],
            capture_output=True, text=True, timeout=1800,
            env={**os.environ, "PATH": os.environ["PATH"] + ":" + os.path.expanduser("~/.cargo/bin")},
        )
        saida = res.stdout + res.stderr
        linhas = [l for l in saida.splitlines() if "test result:" in l]
        totais = []
        for l in linhas:
            import re as _re
            m = _re.search(r"(\d+) passed; (\d+) failed", l)
            if m:
                totais.append((int(m.group(1)), int(m.group(2))))
        p = sum(t[0] for t in totais)
        f = sum(t[1] for t in totais)
        rel["blocos"]["cargo"] = f"{p}/{p + f}"
        rel["gates"]["cargo_todos_verdes"] = f == 0
    except Exception as e:
        rel["blocos"]["cargo"] = f"erro: {e}"
        rel["gates"]["cargo_todos_verdes"] = False

    # ================= veredito =================
    total = len(rel["gates"])
    ok = sum(1 for v in rel["gates"].values() if v is True)
    rel["veredito"] = f"{ok}/{total}"
    with open(os.path.join(R, "suite_v090_final.json"), "w") as fo:
        json.dump(rel, fo, indent=1, ensure_ascii=False)
    print(json.dumps({"blocos": rel["blocos"], "veredito": rel["veredito"]},
                     indent=1, ensure_ascii=False))
    sys.exit(0 if ok == total else 1)


if __name__ == "__main__":
    main()