File size: 4,539 Bytes
97df2ae
e26f4cc
 
 
 
97df2ae
 
 
e26f4cc
 
97df2ae
 
 
 
 
e26f4cc
97df2ae
 
e26f4cc
97df2ae
 
 
e26f4cc
 
97df2ae
e26f4cc
 
 
97df2ae
e26f4cc
97df2ae
 
 
 
 
 
e26f4cc
 
97df2ae
e26f4cc
 
 
97df2ae
 
 
 
 
e26f4cc
 
 
 
 
 
 
 
97df2ae
 
e26f4cc
97df2ae
 
 
e26f4cc
97df2ae
 
 
 
 
 
 
 
 
 
e26f4cc
 
97df2ae
e26f4cc
97df2ae
 
 
 
 
 
e26f4cc
97df2ae
 
 
 
 
 
 
 
 
e26f4cc
97df2ae
 
 
e26f4cc
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())