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 | #!/usr/bin/env python3 | |
| """KHTST-RUST — CLI MLOps unificada (v0.10.0, crítica Qwen3.8 lote 3 nº 2). | |
| **Por que FAÇADA e não substituição:** a proposta original ("substituir | |
| por um CLI declarativo") violaria duas regras do projeto: (a) os scripts | |
| numerados são a trilha de rastreabilidade das memórias r-XX (cada prova | |
| cita "scripts 39-74"); (b) aprimoramentos não removem funcionalidade | |
| eficaz. O CLI unifica o ENTRY POINT (um só comando para todo o pipeline) | |
| delegando via subprocess aos scripts canônicos — zero duplicação de | |
| lógica, zero remoção — e acrescenta o que não existia: validação de | |
| integridade criptográfica dos checkpoints ANTES do resume (Teo CP-1..3, | |
| r-35 §4) e gestão de estados locais. | |
| Uso: | |
| python scripts/75_cli.py versao | |
| python scripts/75_cli.py treinar --validar-checkpoints | |
| python scripts/75_cli.py verificar-checkpoints | |
| python scripts/75_cli.py estados [--apagar] | |
| python scripts/75_cli.py coletar|suite|telemetria|publicar [--confirmar] | |
| """ | |
| from __future__ import annotations | |
| import json | |
| import os | |
| import subprocess | |
| import sys | |
| from pathlib import Path | |
| from typing import Optional | |
| import typer | |
| app = typer.Typer(help="Pipeline KHTST-RUST v0.10.0 — fachada unificada (r-35 §6)") | |
| RAIZ = Path(__file__).resolve().parent.parent | |
| RESULTADOS = RAIZ / "resultados" | |
| CHECKPOINTS = [ | |
| RESULTADOS / "tokenizador_conceitos.json", | |
| RESULTADOS / "gerador_conceitos.json", | |
| ] | |
| # scripts canônicos por estágio (trilha r-XX preservada) | |
| DELEGACAO = { | |
| "coletar": "70_coleta_shards_v090.py", | |
| "treinar": "71_treino_v090.py", | |
| "suite": "77_suite_v010.py", | |
| "telemetria": "76_telemetria_v010.py", | |
| "publicar": "78_publicar_v010.py", | |
| } | |
| # padrões de estados de treino (mesma lista da varredura pós-publicação) | |
| PADROES_ESTADO = [ | |
| "*.pt", "*.npz", "*.ckpt", "*.safetensors", "*.estado.json", | |
| "**/*.pt", "**/*.npz", "**/*.ckpt", "**/*.safetensors", "**/*.estado.json", | |
| ] | |
| def _delega(estagio: str, extra: list[str] | None = None) -> int: | |
| script = DELEGACAO[estagio] | |
| cmd = [sys.executable, str(RAIZ / "scripts" / script)] + (extra or []) | |
| typer.echo(f"→ {estagio}: {' '.join(cmd)}", err=True) | |
| return subprocess.run(cmd, cwd=RAIZ).returncode | |
| def versao() -> None: | |
| """Versão do núcleo Rust e informação de build.""" | |
| import khtst_rust as k | |
| typer.echo(k.versao()) | |
| typer.echo(k.info_build()) | |
| def coletar() -> None: | |
| """Coleta shards reais (Wikipédia PT-BR, streaming controlado).""" | |
| raise typer.Exit(_delega("coletar")) | |
| def _valida_ou_regenera(ck: Path, k) -> str: | |
| """Política de resume com prova de integridade (CP-1..CP-3): | |
| - não existe → 'ausente' (treina) | |
| - manifesto bate (SHA256) → 'válido' (reusa — rápido) | |
| - manifesto divergente → 'corrompido' (APAGA e retreina: nunca | |
| reusa estado em silêncio) | |
| - sem manifesto (legado) → 'legado' (reusa; o carregador legado | |
| falha alto em JSON truncado — detecção por parse) | |
| """ | |
| if not ck.exists(): | |
| return "ausente" | |
| if k.checkpoint_valido(str(ck)): | |
| return "válido" | |
| tem_manifesto = k.checkpoint_manifesto(str(ck)) is not None | |
| if tem_manifesto: | |
| ck.unlink(missing_ok=True) | |
| Path(str(ck) + ".manifest.json").unlink(missing_ok=True) | |
| return "corrompido (apagado — será retreinado)" | |
| return "legado (sem manifesto; parse do carregador detecta truncamento)" | |
| def treinar( | |
| validar_checkpoints: bool = typer.Option( | |
| True, "--validar-checkpoints/--nao-validar", | |
| help="Valida SHA-256 dos checkpoints antes do resume (Teo CP-3)."), | |
| ) -> None: | |
| """Pipeline A/B/C (script 71) com pré-validação de integridade.""" | |
| import khtst_rust as k | |
| if validar_checkpoints: | |
| for ck in CHECKPOINTS: | |
| estado = _valida_ou_regenera(ck, k) | |
| typer.echo(f"checkpoint {ck.name}: {estado}") | |
| raise typer.Exit(_delega("treinar")) | |
| def verificar_checkpoints() -> None: | |
| """Relatório de integridade (SHA-256 + tamanho) dos checkpoints.""" | |
| import khtst_rust as k | |
| achou = False | |
| for ck in RESULTADOS.rglob("*.json"): | |
| if ck.name.endswith(".manifest.json"): | |
| continue | |
| manifesto = k.checkpoint_manifesto(str(ck)) | |
| if manifesto is None: | |
| continue | |
| achou = True | |
| m = json.loads(manifesto) | |
| valido = k.checkpoint_valido(str(ck)) | |
| typer.echo( | |
| f"{'✓' if valido else '✗'} {ck.name}: sha256={m['hash_sha256'][:16]}… " | |
| f"tamanho={m['tamanho']}B algoritmo={m['algoritmo']} " | |
| f"versão={m['versao']}") | |
| if not achou: | |
| typer.echo("nenhum checkpoint com manifesto encontrado") | |
| def suite() -> None: | |
| """Suíte consolidada de gates v0.10.0 (script 77).""" | |
| raise typer.Exit(_delega("suite")) | |
| def telemetria() -> None: | |
| """Coleta telemetria do projeto v0.10.0 (script 76).""" | |
| raise typer.Exit(_delega("telemetria")) | |
| def publicar( | |
| confirmar: bool = typer.Option(False, "--confirmar", | |
| help="Executa o upload único ao HF (script 74)."), | |
| ) -> None: | |
| """Publicação organizada no HF — exige --confirmar.""" | |
| if not confirmar: | |
| typer.echo("publicação não confirmada: use --confirmar (upload único, " | |
| "HF_TOKEN apenas via variável de ambiente)") | |
| raise typer.Exit(2) | |
| if not os.environ.get("HF_TOKEN"): | |
| typer.echo("HF_TOKEN ausente no ambiente — defina antes de publicar") | |
| raise typer.Exit(2) | |
| raise typer.Exit(_delega("publicar")) | |
| def estados( | |
| apagar: bool = typer.Option(False, "--apagar", | |
| help="Apaga os estados encontrados (regeneráveis)."), | |
| ) -> None: | |
| """Varre (e opcionalmente apaga) estados de treino do ambiente local.""" | |
| achados: list[Path] = [] | |
| for padrao in PADROES_ESTADO: | |
| achados.extend(RAIZ.rglob(padrao)) | |
| achados = sorted(set(p for p in achados if "target" not in p.parts | |
| and ".git" not in p.parts)) | |
| if not achados: | |
| typer.echo("0 estados de treino no ambiente local") | |
| return | |
| total = 0 | |
| for p in achados: | |
| tam = p.stat().st_size | |
| total += tam | |
| typer.echo(f"{'apagando' if apagar else 'achado'}: {p.relative_to(RAIZ)} " | |
| f"({tam / 1e6:.1f} MB)") | |
| if apagar: | |
| p.unlink() | |
| typer.echo(f"total: {len(achados)} arquivos, {total / 1e6:.1f} MB" | |
| + (" — apagados" if apagar else " (use --apagar para remover)")) | |
| if __name__ == "__main__": | |
| app() | |