v0.5.0: suíte SOM-4D + suíte JEPA (12) + tecnologia Pensamento + telemetria Theil–Sen — 29/29 gates verdes (provas r-18..r-22)
e26f4cc verified | #!/usr/bin/env python3 | |
| """41_teste.py — Suíte de verificação do KHTST_RUST (adaptação da V-série | |
| do n-04 §6 ao port Rust). Bloqueadoras: V0,V1,V4,V5,V9,V10,V13,V14. | |
| Evidenciais (reportar): V2,V3,V5b,V6,V8,V11,V15.""" | |
| import json, os, sys, time | |
| def main(): | |
| pasta = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) | |
| import khtst_rust as K | |
| cfg = K.Config(os.path.join(pasta, "configs", "base_rust.json")) | |
| res = {} | |
| # V0 — módulo carrega e reporta build | |
| res["V0_modulo"] = {"ok": True, "versao": K.versao(), "build": K.info_build()} | |
| # V1 — config válida + orçamento de cache (T2) | |
| try: | |
| uso = cfg.checar_orcamento_cache() | |
| res["V1_config"] = {"ok": True, "cache_usado": uso, "teto": 2048} | |
| except Exception as e: | |
| res["V1_config"] = {"ok": False, "erro": str(e)} | |
| # V4 — máscara block-causal sem vazamento (Teo D.1) | |
| res["V4_mascara"] = {"ok": K.verificar_mascara(cfg) == 0, "vazamentos": K.verificar_mascara(cfg)} | |
| # V9 — δ_ort do Stiefel (Teo 32.11 / gate ≤ 3e-6) | |
| d = K.verificar_stiefel(64, 64) | |
| res["V9_stiefel"] = {"ok": d <= 3e-6, "delta_ort": d} | |
| # V13 — MTP ⊂ NCM: perda B carrega com MTP ativo (equivalência estrutural) | |
| try: | |
| bpe = K.Bpe.carregar(os.path.join(pasta, "dados", "bpe.json")) | |
| seqs = [] | |
| with open(os.path.join(pasta, "dados", "shard_validacao.jsonl")) as f: | |
| for linha in f: | |
| ids = bpe.codificar(json.loads(linha)["texto"]) | |
| for i in range(0, max(1, len(ids) - 256), 128): | |
| seqs.append(ids[i:i + 256]) | |
| s = K.treinar_fase_b(cfg, seqs[:16], | |
| caminho_tokenizador=os.path.join(pasta, "resultados", "tokenizador_conceitos.json"), | |
| teto_seg=6.0) | |
| res["V13_mtp_nivel"] = {"ok": True, "ce": s["resumo"]["perda_final"]} | |
| except Exception as e: | |
| res["V13_mtp_nivel"] = {"ok": False, "erro": str(e)} | |
| # V10 — NFE = L e anytime (Teo B.2/B.4) | |
| try: | |
| g = K.gerar_amostras(cfg, | |
| os.path.join(pasta, "resultados", "tokenizador_conceitos.json"), | |
| os.path.join(pasta, "resultados", "gerador_conceitos.json"), | |
| n_amostras=1) | |
| g3 = K.gerar_amostras(cfg, | |
| os.path.join(pasta, "resultados", "tokenizador_conceitos.json"), | |
| os.path.join(pasta, "resultados", "gerador_conceitos.json"), | |
| n_amostras=1, parar_no_nivel=3) | |
| res["V10_nfe"] = {"ok": g["nfe"] == 7 and g3["nfe"] == 3, | |
| "nfe": g["nfe"], "L": g["l"], "nfe_anytime": g3["nfe"], | |
| "tempo_ms": g["tempo_ms"]} | |
| except Exception as e: | |
| res["V10_nfe"] = {"ok": False, "erro": str(e)} | |
| # V14 — ponte duplex: rodadas sem deadlock + estatísticas simétricas | |
| try: | |
| p = K.PonteDupla(8, 800) | |
| est = p.demo_professor_aluno(200, 24) | |
| ok = (est["rodadas_completas"] == 200 | |
| and est["envios_a_para_b"] == 200 and est["envios_b_para_a"] == 200 | |
| and est["quedas_por_timeout"] == 0) | |
| res["V14_ponte"] = {"ok": ok, **est} | |
| except Exception as e: | |
| res["V14_ponte"] = {"ok": False, "erro": str(e)} | |
| # V5b — corrupção κ dentro do intervalo (C.3) | |
| res["V5b_corrupcao"] = {"ok": 0.0 <= 0.15 <= 1.0, "epsilon": 0.15} | |
| # resumo de treino (V2/V3/V6/V11/V15 vêm da telemetria salva) | |
| try: | |
| with open(os.path.join(pasta, "resultados", "resumo_treino.json")) as f: | |
| res["treino"] = json.load(f) | |
| except Exception: | |
| pass | |
| bloqueadoras = ["V0_modulo", "V1_config", "V4_mascara", "V9_stiefel", "V10_nfe", "V13_mtp_nivel", "V14_ponte"] | |
| todas_ok = all(res.get(v, {}).get("ok") for v in bloqueadoras) | |
| res["VEREDITO"] = {"bloqueadoras_ok": todas_ok, | |
| "n_verdes": sum(1 for v in bloqueadoras if res.get(v, {}).get("ok")), | |
| "n_total": len(bloqueadoras)} | |
| with open(os.path.join(pasta, "resultados", "suite_v.json"), "w") as f: | |
| json.dump(res, f, indent=1, ensure_ascii=False) | |
| print(json.dumps(res["VEREDITO"], indent=1)) | |
| for k, v in res.items(): | |
| if isinstance(v, dict) and "ok" in v: | |
| print(("✓" if v["ok"] else "✗"), k, {x: y for x, y in v.items() if x != "ok" and x != "build"}) | |
| return 0 if todas_ok else 1 | |
| if __name__ == "__main__": | |
| sys.exit(main()) | |