khtst-rust / scripts /76_telemetria_v010.py
PowerMachine's picture
v0.10.0: análise/prova/teste das críticas de engenharia Qwen3.8 lote 3 (r-35) — divisão SRP dos monólitos jepa/pensamento/moe/python em 26 submódulos (API estável via re-exports; 'Amdahl de compilação' refutada por experimento controlado, builds frios neutros), checkpoints atômicos SHA-256 puro FIPS 180-4 + fsync + rename + manifesto (CP-1..CP-3; correções da proposta: sem destruir estado anterior, manifesto com barreira, sem bincode/unwrap), TemperaturaInversa T-V3 (fronteira tipada da convexidade N-2; alegação compile-time corrigida), trait CamadaKhtst+CamadaHomeostatica (premissa do controlador refutada no fonte; Vec<Var>; adaptador MoeComPapel bit-a-bit), CLI Typer 75 (fachada aditiva com resume verificado por hash) — 123/123 cargo, 107/107 gates, treino real Wikipédia PT-BR com checkpoints íntegros
06e4b6f verified
Raw History Blame Contribute Delete
6.4 kB
#!/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()