v0.9.0: análise/prova/teste das críticas Qwen3.8 lote 2 (r-34) — SIMD com despacho em runtime bit-a-bit (X-1), oráculo candle_nn::optim::AdamW (nosso 1,53× mais rápido), Thompson Sampling (regret 0,0019 vs UCB1 0,0156), GNG por entropia com piso (T-G1'), JEPA de fração adaptativa (bit-a-bit com 1 braço), configs tipadas, entrada numpy (2008× na fronteira), avaliação com dados reais — 104/104 gates, 111/111 cargo
d444358 verified | #!/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() | |