File size: 4,539 Bytes
97df2ae | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 | #!/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())
|