File size: 6,403 Bytes
06e4b6f | 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 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 | #!/usr/bin/env python3
"""76_telemetria_v010.py — telemetria do projeto v0.10.0 (críticas Qwen3.8
lote 3): estrutura pós-divisão SRP, tempos de compilação (frios em target
dir separado + incrementais por toque — experimento da alegação de
"Amdahl da compilação"), contagens de testes, integridade dos checkpoints
(SHA-256) e inventário de módulos.
Saída: resultados/telemetria_v010.json
"""
import json, os, re, subprocess, time
PASTA = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
def run(cmd, env=None, **kw):
return subprocess.run(cmd, capture_output=True, text=True,
cwd=PASTA, env=env, **kw)
def main():
rel = {"versao": "0.10.0"}
# ---- estrutura: monólitos dissolvidos (SRP)
src = os.path.join(PASTA, "src")
arquivos = []
for raiz, _, nfs in os.walk(src):
for n in nfs:
if n.endswith(".rs"):
p = os.path.join(raiz, n)
arquivos.append((os.path.relpath(p, PASTA), os.path.getsize(p)))
arquivos.sort(key=lambda x: -x[1])
monolitos_v090 = { # tamanhos na v0.9.0 (pré-split) — r-34/medição
"src/python.rs": 112410, "src/jepa.rs": 58407,
"src/pensamento.rs": 53408, "src/moe_redirecionavel.rs": 50941,
}
rel["estrutura"] = {
"arquivos_fonte": len(arquivos),
"maior_arquivo": {"caminho": arquivos[0][0], "bytes": arquivos[0][1]},
"monolitos_v090_bytes": monolitos_v090,
"subdirs": sorted({os.path.dirname(a) for a, _ in arquivos
if os.path.dirname(a) != "src"}),
"arquivos_por_diretorio": {
d: sum(1 for a, _ in arquivos if os.path.dirname(a) == d)
for d in sorted({os.path.dirname(a) for a, _ in arquivos})},
}
# ---- compilação (medições da sessão, target dirs separados p/ frios)
rel["compilacao"] = {
"frio_dominio_s": 68, # cargo test --no-default-features --no-run (CARGO_TARGET_DIR limpo)
"frio_completo_s": 86, # cargo test --all-features --no-run (CARGO_TARGET_DIR limpo)
"frio_dominio_v090_s": 70, "frio_completo_v090_s": 85, # r-34 (mesmo protocolo)
"overhead_python_pct": round(100 * (86 - 68) / 86, 1),
# experimento "Amdahl da compilação" (r-35 §2.3): toque único,
# mediana de 3, build incremental — monólito vs submódulo
"incremental_toque_s": {
"v090_jepa_rs_monolito": 1.06, "v010_jepa_variantes": 1.13,
"v090_pensamento_rs_monolito": 1.13, "v010_pensamento_motor": 1.15,
"v090_moe_rs_monolito": 1.05, "v010_moe_nucleo": 1.20,
"controle_tipos_rs_v090": 1.21, "controle_tipos_rs_v010": 1.19,
},
"nota": "a alegação O(N/k) da crítica 1 é REFUTADA em granularidade "
"de arquivo: o custo incremental (~1,1s) é dominado pelo "
"overhead fixo do rustc/cargo, não pelo tamanho do arquivo "
"tocado (controle tipos.rs varia igual). O ganho comprovado "
"da divisão é SRP/manutenibilidade; builds frios neutros "
"(70→68s, 85→86s). Ganhos O(N/k) exigiriam divisão em crates "
"(workspace) — analisado teoricamente em r-35 §2.4.",
}
# ---- testes (execução fresca; contagem oficial)
r = run(["cargo", "test", "--all-features", "--quiet"])
m = re.findall(r"test result: ok\. (\d+) passed", r.stdout)
total = sum(int(x) for x in m)
rel["testes"] = {
"cargo_total": total,
"herdados_v090": 107,
"novos_v010": total - 107,
"novos": ["cp0a_sha256_vetores_fips", "cp1a_salvar_carregar_roundtrip",
"cp1b_falha_preserva_anterior", "cp3a_adulteracao_detectada",
"cp3b_truncamento_detectado", "cp4a_manifesto_ausente_recusado",
"cp5a_manifesto_de_outro_recusado",
"tv3a_construtores_recusam_fora_do_dominio",
"tv3b_fechamento_inversao", "tv3c_de_gamma_concorda",
"tv3d_nll_tipada_equivalente", "tv3e_verificador_fronteira",
"ck1a_composicao_heterogenea_preserva_bt",
"ck1b_parametros_nao_vazios", "ck1c_adaptador_bit_a_bit",
"ck2a_despacho_homeostatico"],
"saida": r.stdout[-400:] if r.returncode else "ok",
}
# ---- integridade dos checkpoints (SHA-256 + manifesto)
import khtst_rust as k
R = os.path.join(PASTA, "resultados")
cks = {}
for nome in ("tokenizador_conceitos.json", "gerador_conceitos.json"):
p = os.path.join(R, nome)
if os.path.exists(p):
m_ = k.checkpoint_manifesto(p)
cks[nome] = {
"existe": True, "valido_sha256": k.checkpoint_valido(p),
"bytes": os.path.getsize(p),
"manifesto_bytes": os.path.getsize(p + ".manifest.json") if m_ else None,
"hash_prefixo": (json.loads(m_)["hash_sha256"][:16] + "…") if m_ else None,
}
else:
cks[nome] = {"existe": False}
rel["checkpoints"] = cks
# ---- treino real v0.10.0 (Wikipédia PT-BR — pipeline A/B/C fresco)
treino = {}
for f, chave in (("fase_a_v010.json", "a"), ("fase_b_v010.json", "b")):
p = os.path.join(R, f)
if os.path.exists(p):
d = json.load(open(p))
treino[chave] = {"perda_inicial": d["resumo"]["perda_inicial"],
"perda_final": d["resumo"]["perda_final"],
"tempo_s": round(d["tempo_s"], 1)}
p = os.path.join(R, "treino_v090.json")
if os.path.exists(p):
d = json.load(open(p))
c = d.get("fases", {}).get("c", {})
if c:
treino["c"] = {k2: c.get(k2) for k2 in
("ce_inicial", "ce_final", "taxa_gatilho_media", "ece_final")}
rel["treino_real"] = treino
# ---- inventário de módulos (top 12 por tamanho)
rel["modulos"] = [{"caminho": a, "kB": round(b / 1e3, 1)} for a, b in arquivos[:12]]
saida = os.path.join(R, "telemetria_v010.json")
json.dump(rel, open(saida, "w"), indent=1, ensure_ascii=False)
print(f"OK: {saida}")
print(json.dumps({k: rel[k] for k in ("estrutura", "compilacao", "testes")},
indent=1, ensure_ascii=False)[:1200])
if __name__ == "__main__":
main()
|