first commit

This commit is contained in:
neoricalex 2026-03-01 00:51:56 +01:00
commit 377bc51a5b
28 changed files with 1529 additions and 0 deletions

59
.gitignore vendored Normal file
View File

@ -0,0 +1,59 @@
# === sub-módulos clonados ===
src/modules/
# === Autotools build artefacts ===
Makefile
Makefile.in
aclocal.m4
autom4te.cache/
config.log
config.status
configure
depcomp
install-sh
missing
py-compile
stamp-h1
# === Build & dist directories ===
/build/
/dist/
!/dist/releases/
!/dist/releases/*
*.tar.gz
*.tar.bz2
*.zip
# === Python cache & venv ===
__pycache__/
*.pyc
*.pyo
*.pyd
*.egg-info/
.eggs/
venv/
.env/
.venv/
# === Editor / OS junk ===
*.swp
*.swo
*.bak
*.tmp
*~
.DS_Store
Thumbs.db
# === Logs ===
*.log
nohup.out
# === IDE / workspace ===
.vscode/
.idea/
*.iml
# === Backup copies ===
*.old
*.orig
*.rej

125
Makefile.am Normal file
View File

@ -0,0 +1,125 @@
SUBDIRS = src
# ===========================
# Caminhos e artefactos
# ===========================
TOP_DIR = $(shell pwd)
DIST_DIR = $(TOP_DIR)/dist
BUILD_DIR = $(TOP_DIR)/build
GIT_VER = $(shell git describe --tags --always --dirty 2>/dev/null || echo "0.1-dev")
SRC_TAR = $(DIST_DIR)/neuro-$(GIT_VER)-src.tar.gz
# ===========================
# Configurações de Git
# ===========================
GIT_USER ?= "neo.webmaster.2@gmail.com"
GIT_EMAIL ?= "neo.webmaster.2@gmail.com"
GIT_REMOTE ?= "origin"
GIT_BRANCH ?= "main"
COMMIT_MSG ?= "Auto-commit via make git"
# ===========================
# Alvos principais
# ===========================
.PHONY: all tarball git release run clean-local check-remote
all: $(DIST_DIR)
$(DIST_DIR):
@mkdir -p $(DIST_DIR)
# ===========================
# Empacotamento do código-fonte
# ===========================
tarball: $(SRC_TAR)
$(SRC_TAR):
@echo "[TAR] Empacotando código-fonte (versão $(GIT_VER))..."
@mkdir -p "$(DIST_DIR)" "$(BUILD_DIR)"
cd "$(TOP_DIR)" && tar \
--exclude="$(notdir $(SRC_TAR))" \
--exclude="$(DIST_DIR)" \
--exclude="$(BUILD_DIR)" \
--exclude='*/__pycache__' \
--exclude='*/.venv' \
--exclude='*/venv' \
--exclude='*.pyc' \
--exclude='*.pyo' \
--exclude='*.o' \
--exclude='*.a' \
--exclude='*.so' \
-czf "$(SRC_TAR)" .
@echo "[✔] Tarball gerado em $(SRC_TAR)"
# ===========================
# Git (commit + push)
# ===========================
git: check-remote
@echo "📦 Commit automático → Gitea"
@git config user.name $(GIT_USER)
@git config user.email $(GIT_EMAIL)
@git rev-parse --abbrev-ref HEAD >/dev/null 2>&1 || true
@git add -A
@git commit -m "$$(echo '$(COMMIT_MSG)')" || echo "Nenhuma modificação para commitar."
@git push $(GIT_REMOTE) $(GIT_BRANCH)
# ===========================
# Release (Tarball + Tag)
# ===========================
release: tarball check-remote
@echo "🚀 Publicando build em dist/releases (versão: $(GIT_VER))"
@mkdir -p $(DIST_DIR)/releases
@cp $(SRC_TAR) $(DIST_DIR)/releases/ 2>/dev/null || echo "⚠️ Nenhum tarball encontrado. Execute 'make tarball' primeiro."
@echo "📦 Adicionando releases ignoradas (forçado)"
@git add -f $(DIST_DIR)/releases/* 2>/dev/null || echo "⚠️ Nenhum artefato novo para adicionar."
@git commit -m "Build automático: release $(GIT_VER)" || echo "Nenhum ficheiro novo para commitar."
@git push origin main
@TAG="$(GIT_VER)"; \
if git rev-parse "$$TAG" >/dev/null 2>&1; then \
echo "⚠️ Tag $$TAG já existente."; \
else \
echo "🏷 Criando tag $$TAG"; \
git tag -a "$$TAG" -m "Release automática $$TAG"; \
git push origin "$$TAG"; \
fi
# ===========================
# Git Remote (HTTPS → SSH Auto-Fix)
# ===========================
check-remote:
@REMOTE_URL=$$(git remote get-url $(GIT_REMOTE)); \
if echo $$REMOTE_URL | grep -q '^https://gitea\.neoricalex\.com'; then \
echo "⚠️ Repositório configurado com HTTPS:"; \
echo " $$REMOTE_URL"; \
echo "🔄 Convertendo para SSH (porta 2222)..."; \
SSH_URL=$$(echo $$REMOTE_URL | sed -E 's|https://gitea\.neoricalex\.com[:/]+|ssh://git@gitea.neoricalex.com:2222/|'); \
git remote set-url $(GIT_REMOTE) $$SSH_URL; \
echo "✅ Remote atualizado para:"; \
git remote -v; \
else \
echo "✅ Remote SSH já configurado:"; \
git remote -v | grep $(GIT_REMOTE); \
fi; \
echo "🔍 Testando conectividade SSH com Gitea..."; \
if ssh -T git@gitea.neoricalex.com -p 2222 2>&1 | grep -q "successfully authenticated"; then \
echo "✅ Conexão SSH funcional com Gitea."; \
else \
echo "❌ Falha na autenticação SSH com Gitea."; \
echo " Verifique a chave em ~/.ssh/id_ed25519.pub e nas SSH Keys do Gitea."; \
exit 1; \
fi
# ===========================
# Teste
# ===========================
run:
@python3 src/root/__main__.py
# ===========================
# Limpeza
# ===========================
clean-local:
@echo "[CLEAN] Removendo diretórios temporários..."
@rm -rf $(BUILD_DIR)
@find $(DIST_DIR) -maxdepth 1 -type f ! -path "$(DIST_DIR)/releases/*" -delete 2>/dev/null || true
@echo "[✔] Limpeza concluída (releases preservadas)"

37
configure.ac Normal file
View File

@ -0,0 +1,37 @@
AC_INIT([NEURO], [NEO_VERSION], [https://gitea.neoricalex.com/neo/neuro.git])
# ===========================
# Diretórios base
# ===========================
AC_CONFIG_AUX_DIR([.])
AC_SUBST([TOP_DIR], ['$(CURDIR)'])
AC_SUBST([BUILD_DIR], ['$(CURDIR)/build'])
AC_SUBST([DIST_DIR], ['$(CURDIR)/dist'])
# ===========================
# Versão dinâmica (Git)
# ===========================
m4_define([NEO_VERSION],
m4_esyscmd_s([git describe --tags --always --dirty 2>/dev/null || echo "0.1-dev"]))
AC_SUBST([NEO_VERSION])
# Caminho do tarball dinâmico (protegido contra M4 expansion)
AC_SUBST([SRC_TAR],
['$(CURDIR)/dist/neuro-@NEO_VERSION@-src.tar.gz'])
# ===========================
# Automake + Python
# ===========================
AM_INIT_AUTOMAKE([foreign dist-bzip2 no-dist-gzip])
AM_PATH_PYTHON([3.0])
# ===========================
# Arquivos Makefile
# ===========================
AC_CONFIG_FILES([
Makefile
src/Makefile
src/root/Makefile
])
AC_OUTPUT

18
src/Makefile.am Normal file
View File

@ -0,0 +1,18 @@
SUBDIRS = root
bin_SCRIPTS = neuro
CLEANFILES = $(bin_SCRIPTS)
EXTRA_DIST = neuro.in
# ===========================
# Substituição dinâmica
# ===========================
neuro: neuro.in Makefile
@which git >/dev/null || { echo "⚠️ Git não encontrado — instale-o manualmente."; exit 1; }
sed \
-e 's,[@]pythondir[@],$(pythondir),g' \
-e 's,[@]PACKAGE[@],$(PACKAGE),g' \
-e 's,[@]VERSION[@],$(VERSION),g' \
< $(srcdir)/neuro.in > neuro
chmod +x neuro

40
src/bootstrap.py Normal file
View File

@ -0,0 +1,40 @@
from pathlib import Path
import subprocess
import sys
import os
class Application(object):
def __init__(self, *args, **kwargs):
for key in kwargs:
setattr(self, key, kwargs[key])
self.base_path = Path(__file__).resolve().parents[1]
self.venv_path = self.base_path / "venv"
self.python = self.venv_path / "bin" / "python"
self._hello()
def _hello(self):
print(f"NEURO {getattr(self, 'version', '')} — inicializador de ambiente")
def run(self):
self._ensure_venv()
self.launch_neuro()
def _ensure_venv(self):
os.chdir(self.base_path)
if not self.venv_path.exists():
print("[+] Criando ambiente Python...")
subprocess.run(
[sys.executable, "-m", "venv", str(self.venv_path)],
check=True
)
else:
print("[=] Ambiente Python já existente.")
def launch_neuro(self):
neuro = self.base_path / "src" / "root" / "__main__.py"
subprocess.run([str(self.python), str(neuro)])

11
src/neuro Executable file
View File

@ -0,0 +1,11 @@
#!/usr/bin/env python
import sys
sys.path.insert(1, '/usr/local/local/lib/python3.12/dist-packages')
from bootstrap import Application
if __name__ == "__main__":
app = Application(package="neuro", version="0.1-dev")
app.run()

11
src/neuro.in Normal file
View File

@ -0,0 +1,11 @@
#!/usr/bin/env python
import sys
sys.path.insert(1, '@pythondir@')
from bootstrap import Application
if __name__ == "__main__":
app = Application(package="@PACKAGE@", version="@VERSION@")
app.run()

1
src/requirements.lock Normal file
View File

@ -0,0 +1 @@
ec72420df5dfbdce4111f715c96338df3b7cb75f58e478d2449c9720e560de8c

1
src/requirements.txt Normal file
View File

@ -0,0 +1 @@
requests

16
src/root/Makefile.am Normal file
View File

@ -0,0 +1,16 @@
# ===========================
# Módulos Python do neoricalex
# ===========================
# Diretório de instalação Python
neurodir = $(pythondir)
# Fontes do root
dist_neuro_PYTHON = \
__main__.py \
__init__.py
# Inclui o bootstrap (nível acima) como parte deste pacote
EXTRA_DIST = ../bootstrap.py

215
src/root/README.md Normal file
View File

@ -0,0 +1,215 @@
Boa — agora tens o **Codex como “motor de edição/execução”**. A IA que vai interagir com ele pode ser “from scratch” *no teu estilo*: pequena no core, mas completa como sistema.
A minha sugestão: pensa nela como um **cérebro (orquestrador)** + **ferramentas (Codex + shell + git)** + **memória (estado)** + **protocolo de prompts**.
A arquitetura abaixo já nasce “Neuro-like” e dá para crescer sem refazer tudo.
---
## 1) Papel do Codex no sistema
O Codex (CLI) é o teu **executor especializado** para:
* ler ficheiros / contexto do repo
* propor alterações
* aplicar patches
* (opcional) correr comandos/testes
Ou seja: o teu “LLM principal” (ou regras) decide *o que fazer*; o Codex faz *o trabalho pesado no repo*.
---
## 2) Mínimo completo (mas bem fundacional)
### Componentes
1. **Kernel de loop** (observe → think → act → rest)
* lê intenção do utilizador
* monta contexto
* escolhe ação (Codex / shell / git / responder)
2. **State** (memória curta e longa)
* curta: última tarefa, decisões, paths tocados, erros
* longa: “project rules”, convenções do repo, preferências
3. **Tool layer** (wrappers)
* `CodexTool`: chama o binário do codex e captura output
* `GitTool`: status/diff/commit/branch
* `ShellTool`: executar comandos (com allowlist)
* `FileTool`: read/write/scan (se precisares fora do codex)
4. **Prompt Protocol** (muito importante)
Define um “contrato” estável entre o teu orquestrador e o Codex:
* **Objective**: o que queremos mudar
* **Constraints**: não quebrar build, seguir estilo, não tocar em X
* **Context**: ficheiros relevantes + output de testes + diffs
* **Expected Output**: patch/diff + comandos sugeridos
5. **Policy & Safety**
* allowlist de comandos
* limites de paths editáveis
* nunca executar coisas destrutivas sem “modo explícito”
---
## 3) Um blueprint que funciona mesmo sem “LLM externo”
Tu disseste “vamos fazer através de prompts” — perfeito.
Dá para ter 2 modos:
### Modo A — “IA determinística” (sem API)
* heurísticas + templates de prompt
* decisões baseadas em regras (ex.: se falha teste → pedir ao codex para corrigir)
* ótimo para bootstrap
### Modo B — “IA híbrida”
* o mesmo kernel, mas o “think/planning” pode usar OpenAI API depois (quando quiseres)
* o Codex continua sendo o executor
A sacada: **não acoplas o core à API**. A API vira só mais um “ThoughtProvider”.
---
## 4) Fluxo operacional recomendado
### 1) Observe
* `git status --porcelain`
* (opcional) ler ficheiros-chave (`README`, `pyproject`, `system/`)
* recolher logs/erros se houver
### 2) Think (planner)
* classificar tarefa: “refactor”, “bugfix”, “feature”, “docs”
* gerar “plan steps” pequeno (15 passos)
* escolher ferramenta: codex vs shell
### 3) Act (codex)
* chama codex com prompt estruturado
* exige output em formato “patch + rationale + commands”
* aplica patch
* corre comandos (pytest/mypy/etc) se permitido
### 4) Verify
* `git diff` + `git status`
* executa testes
* se falhar: loop volta com “erro + diff + logs” como contexto
### 5) Rest
* grava “worklog” e “memory frame” (json)
* atualiza “project_rules.json”
---
## 5) Estrutura de diretórios sugerida (bem limpa)
```text
src/root/
ai/
__init__.py
kernel.py # loop observe/think/act
state.py # state + persistence
prompts.py # templates de prompt
tools/
codex.py # wrapper do binário
git.py
shell.py
files.py
policies.py # allowlist/paths
system/
env.py # ensure_codex já está
```
---
## 6) O “prompt contract” com o Codex (o segredo)
Usa sempre um prompt com esta estrutura (mesmo sem LLM externo):
* **ROLE**: “You are Codex, repo editor”
* **TASK**: objetivo claro
* **CONSTRAINTS**: estilo/paths/comandos permitidos
* **CONTEXT**: diffs, erros, ficheiros relevantes (mínimo necessário)
* **OUTPUT FORMAT**: obrigatório
Exemplo de output format (super útil):
* `PLAN:` bullets
* `PATCH:` unified diff
* `COMMANDS:` lista de comandos para validar
Isto torna o sistema “programável”.
---
## 7) Como “bem completa” sem virar monstro
Para ficar completa de verdade, adiciona 4 capacidades logo cedo:
1. **Controle de escopo**
* “só tocar em src/root/ai e system/”
* evitar alterações fora do perímetro
2. **Verificação automática**
* sempre rodar pelo menos `python -m compileall` ou `pytest -q` quando houver
3. **Journaling**
* `data/ai_runs/YYYYMMDD_HHMMSS.json`
* guarda prompts, diffs, outputs, comandos
4. **Rollback**
* se algo der ruim: `git checkout -- .` ou `git reset --hard HEAD` (somente em modo explícito)
---
## 8) Próximo passo prático (eu sugiro)
Vamos implementar em 3 “camadas” rápidas:
### Passo 1 — Tool wrapper (CodexTool)
* função `run(prompt, cwd, timeout)`
* captura stdout/stderr
* retorna objeto `{ok, text, rc}`
### Passo 2 — Kernel mínimo (1 loop)
* lê input do user
* monta prompt
* chama codex
* mostra diff/result
### Passo 3 — Verify loop
* roda testes
* se falha: chama codex de novo com logs + diff
Com isso, tens uma “IA” que já age como agente de manutenção do repo.
---
## Escolha que muda tudo (preciso que decidas já, sem drama)
Queres que essa IA:
1. **Aplique mudanças automaticamente** (auto-patch + auto-test), ou
2. **Só proponha e tu confirmas** (human-in-the-loop)?
Eu recomendo começar em **(2)** por segurança e depois habilitar (1) por feature flag.
Se me disseres “auto” ou “confirm”, eu já te devolvo um esqueleto inicial (arquivos + classes + prompts) pronto para colares no repo.

0
src/root/__init__.py Normal file
View File

25
src/root/__main__.py Normal file
View File

@ -0,0 +1,25 @@
#!/usr/bin/env python3
from email.mime import base
from pathlib import Path
from system.env import ensure_codex
from system.requirements import ensure_requirements
from ai.cli import run_ai
def main():
base = Path(__file__).resolve().parents[2]
python_bin = base / "venv" / "bin" / "python"
# 1. Dependências Python
ensure_requirements(python_bin)
codex_bin = ensure_codex(base)
print("codex:", codex_bin)
run_ai(base, codex_bin)
if __name__ == "__main__":
main()

0
src/root/ai/__init__.py Normal file
View File

43
src/root/ai/cli.py Normal file
View File

@ -0,0 +1,43 @@
# src/root/ai/cli.py
from __future__ import annotations
import argparse
from pathlib import Path
from .kernel import Kernel
from .policies import policies_from_env
from .tools.codex import CodexTool
from .tools.git import GitTool
from .tools.shell import ShellTool
def run_ai(repo_root: Path, codex_bin: Path, argv: list[str] | None = None) -> None:
ap = argparse.ArgumentParser(prog="neuro-ai")
ap.add_argument("--auto", action="store_true", help="auto-apply patches (dangerous)")
ap.add_argument("--once", action="store_true", help="run a single turn and exit")
ap.add_argument("--model", default=None, help="codex model (e.g. gpt-5.3-codex medium)")
args = ap.parse_args(argv)
policies = policies_from_env(repo_root, auto_flag=args.auto)
kernel = Kernel(
repo_root=repo_root,
policies=policies,
codex=CodexTool(codex_bin, model=args.model),
git=GitTool(repo_root),
shell=ShellTool(policies),
)
print(f"[ai] mode={'AUTO' if policies.auto_apply else 'CONFIRM'}")
print("[ai] Enter a task (or empty to exit):")
while True:
try:
user_input = input("> ").strip()
except EOFError:
break
if not user_input:
break
kernel.run_once(user_input)
if args.once:
break

137
src/root/ai/kernel.py Normal file
View File

@ -0,0 +1,137 @@
# src/root/ai/kernel.py
from __future__ import annotations
import re
from dataclasses import dataclass
from pathlib import Path
from .policies import Policies
from .prompts import build_codex_prompt
from .state import RunLog, save_runlog
from .tools.codex import CodexTool
from .tools.git import GitTool
from .tools.shell import ShellTool
@dataclass
class Kernel:
repo_root: Path
policies: Policies
codex: CodexTool
git: GitTool
shell: ShellTool
def _constraints_text(self) -> str:
return (
f"- Only modify these paths: {', '.join(self.policies.allowed_paths)}\n"
"- Output unified diff inside ```diff fences.\n"
"- Do not delete large files.\n"
"- Keep changes minimal and buildable.\n"
)
def _context_text(self) -> str:
snap = self.git.snapshot()
ctx = [
"Repo status (porcelain):",
snap.status_porcelain or "(clean)",
"",
"Current diff:",
snap.diff or "(no diff)",
]
return "\n".join(ctx)
def _extract_diff(self, codex_output: str) -> str:
# pega conteúdo dentro de ```diff ... ```
m = re.search(r"```diff\s*(.*?)```", codex_output, flags=re.DOTALL)
return (m.group(1).strip() + "\n") if m else ""
def run_once(self, user_input: str) -> None:
before = self.git.snapshot()
prompt = build_codex_prompt(
task=user_input,
constraints=self._constraints_text(),
context=self._context_text(),
)
codex_res = self.codex.run(prompt.task, cwd=self.repo_root)
if not codex_res.ok:
# tentativa fallback via stdin (se quiseres automático)
codex_res = self.codex.run_exec_stdin(prompt.task, cwd=self.repo_root)
print("[ai] codex failed (rc=%d)" % codex_res.rc)
print(codex_res.text)
# log mesmo assim
after = self.git.snapshot()
path = save_runlog(self.repo_root, RunLog(
ts="now",
user_input=user_input,
codex_prompt=prompt.task,
codex_output=codex_res.text,
before_status=before.status_porcelain,
before_diff=before.diff,
after_status=after.status_porcelain,
after_diff=after.diff,
applied=False,
notes="codex returned non-zero",
))
print(f"[ai] runlog: {path}")
return
diff = self._extract_diff(codex_res.text)
print("\n========== CODEX OUTPUT ==========\n")
print(codex_res.text)
if not diff.strip():
print("\n[ai] No PATCH found in codex output.")
after = self.git.snapshot()
path = save_runlog(self.repo_root, RunLog(
ts="now",
user_input=user_input,
codex_prompt=prompt.task,
codex_output=codex_res.text,
before_status=before.status_porcelain,
before_diff=before.diff,
after_status=after.status_porcelain,
after_diff=after.diff,
applied=False,
notes="no diff extracted",
))
print(f"[ai] runlog: {path}")
return
# Human-in-the-loop gate
applied = False
if self.policies.auto_apply:
print("\n[ai] AUTO mode enabled -> applying patch.")
self.git.apply_patch(diff)
applied = True
else:
print("\n[ai] Proposed PATCH extracted. Apply? [y/N] ", end="", flush=True)
ans = input().strip().lower()
if ans in ("y", "yes"):
self.git.apply_patch(diff)
applied = True
else:
print("[ai] Not applied.")
after = self.git.snapshot()
path = save_runlog(self.repo_root, RunLog(
ts="now",
user_input=user_input,
codex_prompt=prompt.task,
codex_output=codex_res.text,
before_status=before.status_porcelain,
before_diff=before.diff,
after_status=after.status_porcelain,
after_diff=after.diff,
applied=applied,
))
print(f"[ai] runlog: {path}")
if applied:
print("\n[ai] git status --porcelain:")
print(after.status_porcelain or "(clean)")
print("\n[ai] git diff:")
print(after.diff or "(no diff)")

34
src/root/ai/policies.py Normal file
View File

@ -0,0 +1,34 @@
# src/root/ai/policies.py
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
import os
@dataclass(frozen=True)
class Policies:
auto_apply: bool
repo_root: Path
allowed_paths: tuple[str, ...]
allowed_shell_prefixes: tuple[str, ...]
def policies_from_env(repo_root: Path, auto_flag: bool | None = None) -> Policies:
env_auto = os.getenv("NEURO_AUTO", "").strip().lower() in ("1", "true", "yes", "on")
auto_apply = env_auto if auto_flag is None else auto_flag
return Policies(
auto_apply=auto_apply,
repo_root=repo_root,
# limita onde a IA pode mexer (ajusta depois)
allowed_paths=(
"src/root/ai/",
),
# allowlist de comandos (mínimo)
allowed_shell_prefixes=(
"python ",
"python3 ",
"pytest",
"git ",
),
)

35
src/root/ai/prompts.py Normal file
View File

@ -0,0 +1,35 @@
# src/root/ai/prompts.py
from __future__ import annotations
from dataclasses import dataclass
@dataclass(frozen=True)
class PromptPack:
system: str
task: str
def build_codex_prompt(
task: str,
context: str,
constraints: str = ("- Do NOT run shell commands.\n"
"- Do NOT use network.\n"
"- Output unified diff inside ```diff fences.\n"
),
) -> PromptPack:
system = (
"You are Codex CLI acting as a repo editor.\n"
"Return output in EXACT sections:\n"
"PLAN:\n- ...\n"
"PATCH:\n```diff\n...unified diff...\n```\n"
"COMMANDS:\n- command\n"
"NOTES:\n...\n"
"If you cannot produce a patch, leave PATCH empty but explain in NOTES.\n"
)
full = (
f"{system}\n"
f"TASK:\n{task}\n\n"
f"CONSTRAINTS:\n{constraints}\n\n"
f"CONTEXT:\n{context}\n"
)
return PromptPack(system=system, task=full)

30
src/root/ai/state.py Normal file
View File

@ -0,0 +1,30 @@
# src/root/ai/state.py
from __future__ import annotations
import json
from dataclasses import dataclass, asdict
from datetime import datetime
from pathlib import Path
@dataclass
class RunLog:
ts: str
user_input: str
codex_prompt: str
codex_output: str
before_status: str
before_diff: str
after_status: str
after_diff: str
applied: bool
notes: str = ""
def save_runlog(repo_root: Path, log: RunLog) -> Path:
outdir = repo_root / "data" / "ai_runs"
outdir.mkdir(parents=True, exist_ok=True)
fname = datetime.utcnow().strftime("%Y%m%d_%H%M%S") + ".json"
path = outdir / fname
path.write_text(json.dumps(asdict(log), ensure_ascii=False, indent=2), encoding="utf-8")
return path

View File

102
src/root/ai/tools/codex.py Normal file
View File

@ -0,0 +1,102 @@
# src/root/ai/tools/codex.py
from __future__ import annotations
import subprocess
from dataclasses import dataclass
from pathlib import Path
@dataclass
class CodexResult:
ok: bool
rc: int
text: str
class CodexTool:
def __init__(self, codex_bin: Path, model: str | None = None):
self.codex_bin = codex_bin
self.model = model
def run_exec(
self,
prompt: str,
cwd: Path,
sandbox: str = "read-only",
approval: str = "never",
timeout_s: int = 1800,
) -> CodexResult:
"""
Usa: codex exec -C <DIR> -s read-only -a never [--model ...] "<PROMPT>"
"""
base_cmd = [
str(self.codex_bin),
"-C", str(cwd),
"-s", sandbox,
"-a", approval,
]
if self.model:
base_cmd += ["-m", self.model]
# 1) tenta passar prompt como argumento (mais comum)
cmd = base_cmd + ["exec", prompt]
try:
p = subprocess.Popen(
cmd,
cwd=str(cwd),
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
)
assert p.stdout is not None
out, _ = p.communicate(timeout=timeout_s)
rc = p.returncode or 0
if rc == 0:
return CodexResult(True, rc, out)
# fallback se falhar
fallback_needed = True
return CodexResult(False, rc, out + ("\n[exec-arg failed]\n" if fallback_needed else ""))
except subprocess.TimeoutExpired:
return CodexResult(False, 124, "[TIMEOUT]\n")
def run_exec_stdin(
self,
prompt: str,
cwd: Path,
sandbox: str = "read-only",
approval: str = "never",
timeout_s: int = 1800,
) -> CodexResult:
"""
Fallback: tenta enviar prompt via stdin (alguns CLIs aceitam quando prompt é omitido)
"""
base_cmd = [
str(self.codex_bin),
"-C", str(cwd),
"-s", sandbox,
"-a", approval,
"exec",
]
if self.model:
base_cmd = [
str(self.codex_bin),
"-C", str(cwd),
"-s", sandbox,
"-a", approval,
] + (["-m", self.model] if self.model else []) + ["exec"]
try:
p = subprocess.Popen(
base_cmd,
cwd=str(cwd),
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
)
assert p.stdin and p.stdout
out, _ = p.communicate(prompt, timeout=timeout_s)
rc = p.returncode or 0
return CodexResult(rc == 0, rc, out)
except subprocess.TimeoutExpired:
return CodexResult(False, 124, "[TIMEOUT]\n")

View File

52
src/root/ai/tools/git.py Normal file
View File

@ -0,0 +1,52 @@
# src/root/ai/tools/git.py
from __future__ import annotations
import subprocess
from dataclasses import dataclass
from pathlib import Path
def _run_git(args: list[str], cwd: Path) -> str:
out = subprocess.check_output(["git"] + args, cwd=str(cwd), text=True, stderr=subprocess.STDOUT)
return out.strip()
@dataclass
class GitSnapshot:
status_porcelain: str
diff: str
class GitTool:
def __init__(self, repo_root: Path):
self.repo_root = repo_root
def status_porcelain(self) -> str:
return _run_git(["status", "--porcelain"], self.repo_root)
def diff(self) -> str:
return _run_git(["diff"], self.repo_root)
def snapshot(self) -> GitSnapshot:
return GitSnapshot(
status_porcelain=self.status_porcelain(),
diff=self.diff(),
)
def apply_patch(self, patch_text: str) -> None:
# Aplica via stdin. Patch deve ser unified diff.
p = subprocess.Popen(
["git", "apply", "--whitespace=nowarn", "-"],
cwd=str(self.repo_root),
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
)
assert p.stdin
out, _ = p.communicate(patch_text)
if p.returncode != 0:
raise RuntimeError(f"git apply failed:\n{out}")
def reset_hard(self) -> None:
_run_git(["reset", "--hard", "HEAD"], self.repo_root)

View File

@ -0,0 +1,41 @@
# src/root/ai/tools/shell.py
from __future__ import annotations
import subprocess
from dataclasses import dataclass
from pathlib import Path
from ..policies import Policies
@dataclass
class ShellResult:
ok: bool
rc: int
out: str
class ShellTool:
def __init__(self, policies: Policies):
self.policies = policies
def run(self, cmd: str, cwd: Path | None = None, timeout_s: int = 1200) -> ShellResult:
cmd = cmd.strip()
if not any(cmd.startswith(p) for p in self.policies.allowed_shell_prefixes):
return ShellResult(False, 126, f"Blocked command by policy: {cmd}")
p = subprocess.Popen(
["bash", "-lc", cmd],
cwd=str(cwd or self.policies.repo_root),
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
)
assert p.stdout is not None
try:
out, _ = p.communicate(timeout=timeout_s)
rc = p.returncode or 0
return ShellResult(ok=(rc == 0), rc=rc, out=out)
except subprocess.TimeoutExpired:
p.kill()
out = p.stdout.read() if p.stdout else ""
return ShellResult(False, 124, out + "\n[TIMEOUT]\n")

View File

279
src/root/system/env.py Normal file
View File

@ -0,0 +1,279 @@
# system/env.py
from __future__ import annotations
import os
import shutil
import subprocess
import sys
from pathlib import Path
CODEX_REPO = "https://github.com/openai/codex.git"
MIN_RUSTC = (1, 82, 0)
def _print(msg: str):
print(f"[codex] {msg}", flush=True)
def _run(cmd: list[str], cwd: Path | None = None, env: dict | None = None):
_print(f"RUN: {' '.join(cmd)}")
p = subprocess.Popen(
cmd,
cwd=str(cwd) if cwd else None,
env=env,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
)
assert p.stdout is not None
for line in p.stdout:
print(f"[codex] {line.rstrip()}", flush=True)
rc = p.wait()
if rc != 0:
raise RuntimeError(f"Command failed: {' '.join(cmd)}")
def _run_capture(cmd: list[str], env: dict | None = None) -> str:
out = subprocess.check_output(cmd, env=env, text=True, stderr=subprocess.STDOUT)
return out.strip()
def _paths(base: Path):
cache = base / ".cache" / "codex"
root = cache / "repo"
workspace = root / "codex-rs"
exe = "codex.exe" if sys.platform.startswith("win") else "codex"
target_bin = workspace / "target" / "debug" / exe
return cache, root, workspace, target_bin
def _parse_rustc_version(line: str) -> tuple[int, int, int]:
# rustc 1.76.0 (....)
parts = line.split()
ver = parts[1].split(".")
return (int(ver[0]), int(ver[1]), int(ver[2]))
def _version_ge(a: tuple[int, int, int], b: tuple[int, int, int]) -> bool:
return a >= b
def _ensure_rust_min(env: dict) -> dict:
_print("Checking Rust toolchain...")
cargo_bin = Path.home() / ".cargo" / "bin"
env["PATH"] = f"{cargo_bin}:{env.get('PATH','')}"
if shutil.which("rustup") is None or shutil.which("cargo") is None or shutil.which("rustc") is None:
_print("Rust toolchain not fully found. Installing rustup...")
_run(
["bash", "-lc", "curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y"],
env=env,
)
env["PATH"] = f"{cargo_bin}:{env.get('PATH','')}"
cur_line = _run_capture(["bash", "-lc", "rustc --version"], env=env)
cur = _parse_rustc_version(cur_line)
_print(f"rustc version: {cur[0]}.{cur[1]}.{cur[2]} (min required 1.82.0)")
if not _version_ge(cur, MIN_RUSTC):
_print("rustc is too old. Updating stable toolchain...")
_run(["bash", "-lc", "rustup toolchain install stable"], env=env)
_run(["bash", "-lc", "rustup default stable"], env=env)
_run(["bash", "-lc", "rustup update stable"], env=env)
cur_line2 = _run_capture(["bash", "-lc", "rustc --version"], env=env)
cur2 = _parse_rustc_version(cur_line2)
_print(f"rustc version after update: {cur2[0]}.{cur2[1]}.{cur2[2]}")
if not _version_ge(cur2, MIN_RUSTC):
raise RuntimeError(f"rustc still < 1.82.0 after update: {cur_line2}")
_print("Ensuring rustfmt + clippy...")
_run(["bash", "-lc", "rustup component add rustfmt"], env=env)
_run(["bash", "-lc", "rustup component add clippy"], env=env)
return env
def _have_sudo() -> bool:
return shutil.which("sudo") is not None
def _sudo_noprompt_ok() -> bool:
if not _have_sudo():
return False
try:
subprocess.check_call(["sudo", "-n", "true"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
return True
except Exception:
return False
def _detect_pkg_manager() -> str | None:
for pm in ("apt-get", "dnf", "pacman", "zypper"):
if shutil.which(pm) is not None:
return pm
return None
def _install_system_deps_for_linux_sandbox() -> None:
"""
Codex linux-sandbox precisa de:
- pkg-config
- libcap (dev headers + libcap.pc)
"""
pm = _detect_pkg_manager()
if pm is None:
raise RuntimeError(
"Missing system deps (pkg-config/libcap). No supported package manager found.\n"
"Install manually: pkg-config + libcap development package (contains libcap.pc)."
)
if not _have_sudo():
raise RuntimeError(
"Missing system deps (pkg-config/libcap) and sudo not found.\n"
"Install manually: pkg-config + libcap development package (contains libcap.pc)."
)
if not _sudo_noprompt_ok():
# sem prompt: a ensure_codex não pode ficar parada pedindo senha
raise RuntimeError(
"Missing system deps (pkg-config/libcap). Need sudo privileges.\n"
"Run manually (then re-run neuro):\n"
" - Debian/Ubuntu: sudo apt-get update && sudo apt-get install -y pkg-config libcap-dev\n"
" - Fedora: sudo dnf install -y pkgconf-pkg-config libcap-devel\n"
" - Arch: sudo pacman -S --needed pkgconf libcap"
)
_print(f"Installing system dependencies via {pm} (sudo -n)...")
if pm == "apt-get":
_run(["sudo", "-n", "apt-get", "update"])
_run(["sudo", "-n", "apt-get", "install", "-y", "pkg-config", "libcap-dev"])
elif pm == "dnf":
_run(["sudo", "-n", "dnf", "install", "-y", "pkgconf-pkg-config", "libcap-devel"])
elif pm == "pacman":
_run(["sudo", "-n", "pacman", "-S", "--needed", "--noconfirm", "pkgconf", "libcap"])
elif pm == "zypper":
_run(["sudo", "-n", "zypper", "--non-interactive", "install", "pkg-config", "libcap-devel"])
else:
raise RuntimeError(f"Unsupported package manager: {pm}")
def _ensure_linux_sandbox_deps(env: dict):
# Só faz sentido no Linux
if not sys.platform.startswith("linux"):
_print("Non-Linux platform detected; skipping linux-sandbox deps check.")
return
_print("Checking system deps for codex-linux-sandbox (pkg-config + libcap)...")
# pkg-config
if shutil.which("pkg-config") is None:
_print("pkg-config not found.")
_install_system_deps_for_linux_sandbox()
return
# libcap via pkg-config
try:
subprocess.check_call(["pkg-config", "--exists", "libcap"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
_print("libcap found via pkg-config.")
except Exception:
_print("libcap not found via pkg-config (libcap.pc missing).")
_install_system_deps_for_linux_sandbox()
def _ensure_just(env: dict):
if shutil.which("just") is None:
_print("Installing just...")
_run(["bash", "-lc", "cargo install just"], env=env)
else:
_print("just already installed.")
def _ensure_nextest(env: dict):
if shutil.which("cargo-nextest") is None:
_print("Installing cargo-nextest (optional)...")
_run(["bash", "-lc", "cargo install --locked cargo-nextest"], env=env)
else:
_print("cargo-nextest already installed.")
def ensure_codex(base: Path) -> Path:
_print("Ensuring Codex build...")
cache, root, workspace, target_bin = _paths(base)
cache.mkdir(parents=True, exist_ok=True)
env = dict(os.environ)
# Repo
if not root.exists():
_print("Cloning Codex repository...")
_run(["git", "clone", CODEX_REPO, str(root)], cwd=cache, env=env)
else:
_print("Repository already exists. Checking for updates...")
_run(["git", "fetch"], cwd=root, env=env)
_run(["git", "pull"], cwd=root, env=env)
if not workspace.exists():
raise RuntimeError(f"Expected workspace not found: {workspace}")
# Rust
env = _ensure_rust_min(env)
# System deps (Linux sandbox)
_ensure_linux_sandbox_deps(env)
# Helper tools
_ensure_just(env)
_ensure_nextest(env)
# Build
_print("Building Codex...")
_run(["bash", "-lc", "cargo build"], cwd=workspace, env=env)
if not target_bin.exists():
raise RuntimeError(f"Codex build failed: binary not found at {target_bin}")
_print(f"Build complete: {target_bin}")
return target_bin
def ensure_ollama():
if shutil.which("ollama"):
print("[=] Ollama encontrado.")
return
print("[+] Ollama não encontrado. A instalar...")
subprocess.run(
["bash", "-c", "curl -fsSL https://ollama.com/install.sh | sh"],
check=True
)
def _ollama_running():
try:
subprocess.run(
["ollama", "list"],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
timeout=2,
)
return True
except Exception:
return False
def ensure_ollama_serve():
if _ollama_running():
print("[=] Ollama serve já ativo.")
return
print("[+] A iniciar ollama serve...")
def _serve():
subprocess.run(["ollama", "serve"])
t = threading.Thread(target=_serve, daemon=True)
t.start()
time.sleep(2)

View File

@ -0,0 +1,46 @@
import hashlib
import subprocess
from pathlib import Path
def _hash_file(path: Path) -> str:
h = hashlib.sha256()
with path.open("rb") as f:
for chunk in iter(lambda: f.read(8192), b""):
h.update(chunk)
return h.hexdigest()
def ensure_requirements(python_bin: Path):
"""
Instala dependências apenas se requirements.txt mudou.
Usa SEMPRE o Python do venv.
"""
base_path = Path(__file__).resolve().parents[2]
req_file = base_path / "requirements.txt"
lock_file = base_path / "requirements.lock"
if not req_file.exists():
print("[!] requirements.txt não encontrado. A ignorar.")
return
current_hash = _hash_file(req_file)
if lock_file.exists() and lock_file.read_text().strip() == current_hash:
print("[=] Dependências Python já atualizadas.")
return
print("[+] Atualizando dependências Python...")
subprocess.run(
[str(python_bin), "-m", "pip", "install", "--upgrade", "pip"],
check=True
)
subprocess.run(
[str(python_bin), "-m", "pip", "install", "-r", str(req_file)],
check=True
)
lock_file.write_text(current_hash)
print("[✓] Dependências instaladas com sucesso.")

171
src/root/teste.py Normal file
View File

@ -0,0 +1,171 @@
# mini_ai_3d_xor.py — MLP do zero (numpy) + Adam, softmax e backprop
# Dataset 3D XOR (paridade) para classificação binária em 2 classes.
import numpy as np
# ---------- util ----------
def accuracy(logits, y):
return (np.argmax(logits, axis=1) == y).mean()
# ---------- dataset (3D XOR / parity) ----------
def make_xor3d(n=4000, noise=0.25, seed=0):
"""
Gera XOR 3D por paridade:
label = 1 se (x>0) XOR (y>0) XOR (z>0) for True (ímpar de positivos)
label = 0 caso contrário (par de positivos)
X ~ (±1, ±1, ±1) com ruído gaussiano.
"""
rng = np.random.default_rng(seed)
# bits 0/1 -> sinais -1/+1
bits = rng.integers(0, 2, size=(n, 3), dtype=np.int64) # (n,3) em {0,1}
signs = (bits * 2 - 1).astype(np.float32) # -> {-1, +1}
X = signs + rng.normal(0, noise, size=signs.shape).astype(np.float32)
# paridade (XOR) dos 3 bits
y = (bits[:, 0] ^ bits[:, 1] ^ bits[:, 2]).astype(np.int64) # (n,) em {0,1}
# normalização leve ajuda (opcional)
X = X.astype(np.float32)
return X, y
# ---------- layers ----------
class Linear:
def __init__(self, in_dim, out_dim, seed=0):
rng = np.random.default_rng(seed)
self.W = (rng.normal(0, 1, (in_dim, out_dim)).astype(np.float32) * np.sqrt(2.0 / in_dim))
self.b = np.zeros((1, out_dim), dtype=np.float32)
self.x = None
self.dW = np.zeros_like(self.W)
self.db = np.zeros_like(self.b)
def forward(self, x):
self.x = x
return x @ self.W + self.b
def backward(self, grad_out):
self.dW = self.x.T @ grad_out
self.db = grad_out.sum(axis=0, keepdims=True)
return grad_out @ self.W.T
class ReLU:
def __init__(self):
self.mask = None
def forward(self, x):
self.mask = (x > 0).astype(np.float32)
return x * self.mask
def backward(self, grad_out):
return grad_out * self.mask
# ---------- loss: softmax cross-entropy ----------
def softmax(logits):
z = logits - logits.max(axis=1, keepdims=True)
expz = np.exp(z)
return expz / expz.sum(axis=1, keepdims=True)
def softmax_cross_entropy(logits, y):
probs = softmax(logits)
N = logits.shape[0]
loss = -np.log(probs[np.arange(N), y] + 1e-12).mean()
grad = probs
grad[np.arange(N), y] -= 1.0
grad /= N
return loss, grad
# ---------- optimizer: Adam ----------
class Adam:
def __init__(self, params, lr=2e-2, betas=(0.9, 0.999), eps=1e-8, weight_decay=1e-4):
self.params = params
self.lr = lr
self.b1, self.b2 = betas
self.eps = eps
self.wd = weight_decay
self.t = 0
self.m = [np.zeros_like(p) for (p, _) in params]
self.v = [np.zeros_like(p) for (p, _) in params]
def step(self):
self.t += 1
for i, (p, g) in enumerate(self.params):
g = g + self.wd * p # weight decay (tipo AdamW simples)
self.m[i] = self.b1 * self.m[i] + (1 - self.b1) * g
self.v[i] = self.b2 * self.v[i] + (1 - self.b2) * (g * g)
mhat = self.m[i] / (1 - self.b1 ** self.t)
vhat = self.v[i] / (1 - self.b2 ** self.t)
p -= self.lr * mhat / (np.sqrt(vhat) + self.eps)
# ---------- model ----------
class MLP:
def __init__(self, in_dim, hidden_dim, out_dim, seed=0):
self.l1 = Linear(in_dim, hidden_dim, seed=seed + 1)
self.a1 = ReLU()
self.l2 = Linear(hidden_dim, hidden_dim, seed=seed + 2)
self.a2 = ReLU()
self.l3 = Linear(hidden_dim, out_dim, seed=seed + 3)
def forward(self, x):
x = self.l1.forward(x)
x = self.a1.forward(x)
x = self.l2.forward(x)
x = self.a2.forward(x)
return self.l3.forward(x)
def backward(self, grad_logits):
g = self.l3.backward(grad_logits)
g = self.a2.backward(g)
g = self.l2.backward(g)
g = self.a1.backward(g)
_ = self.l1.backward(g)
def parameters(self):
return [
(self.l1.W, self.l1.dW), (self.l1.b, self.l1.db),
(self.l2.W, self.l2.dW), (self.l2.b, self.l2.db),
(self.l3.W, self.l3.dW), (self.l3.b, self.l3.db),
]
# ---------- train ----------
def train(seed=42):
X, y = make_xor3d(n=6000, noise=0.30, seed=seed)
# split treino/val
n = X.shape[0]
p = np.random.default_rng(seed).permutation(n)
X, y = X[p], y[p]
n_train = int(0.8 * n)
Xtr, ytr = X[:n_train], y[:n_train]
Xva, yva = X[n_train:], y[n_train:]
model = MLP(in_dim=3, hidden_dim=64, out_dim=2, seed=seed)
opt = Adam(model.parameters(), lr=2e-2, weight_decay=1e-4)
rng = np.random.default_rng(seed + 123)
batch = 256
steps = 2000
for step in range(1, steps + 1):
idx = rng.integers(0, Xtr.shape[0], size=batch)
xb, yb = Xtr[idx], ytr[idx]
logits = model.forward(xb)
loss, grad = softmax_cross_entropy(logits, yb)
model.backward(grad)
opt.step()
if step % 200 == 0 or step == 1:
tr_logits = model.forward(Xtr)
va_logits = model.forward(Xva)
tr_acc = accuracy(tr_logits, ytr)
va_acc = accuracy(va_logits, yva)
print(f"step {step:4d} | loss {loss:.4f} | train_acc {tr_acc:.3f} | val_acc {va_acc:.3f}")
if __name__ == "__main__":
train()