| |
| |
|
|
| import pathlib |
| import time |
| import json |
| import logging |
| from typing import List |
|
|
| logger = logging.getLogger(__name__) |
|
|
| |
| ARTIFACTS_DIR = pathlib.Path(__file__).parent.parent / "artifacts" |
| FORCE_NO_BACKUP = False |
| CONFIG_OK = True |
|
|
| |
| ARTIFACTS_DIR.mkdir(exist_ok=True) |
|
|
| |
| self_modification_engine = True |
|
|
| def is_self_modification_enabled() -> bool: |
| """Check if self-modification is enabled.""" |
| try: |
| |
| return True |
| except Exception: |
| return False |
|
|
| async def dispatch(command: str, args: str = "") -> str: |
| """ |
| Main dispatch function for self-modification commands. |
| This is the function that main_agent.py calls. |
| """ |
| cmd = command.lstrip("/").replace("self:", "") |
| |
| handlers = { |
| "status": handle_self_status, |
| "analyze": handle_self_analyze, |
| "plan": handle_self_plan, |
| "apply": handle_self_apply, |
| "menu": handle_self_menu |
| } |
| |
| handler = handlers.get(cmd) |
| if not handler: |
| return f"❌ Comando self-modification não reconhecido: {cmd}\n\nDisponíveis: {', '.join(handlers.keys())}" |
| |
| try: |
| |
| result = await handler(args, "") |
| return result |
| except Exception as e: |
| logger.error(f"Erro no comando self:{cmd}: {e}") |
| return f"❌ Erro ao executar /self:{cmd}: {str(e)}" |
|
|
| async def handle_self_status(args: str, block: str) -> str: |
| """ |
| /self:status |
| Mostra status do sistema de auto-modificação. |
| """ |
| lines: List[str] = [] |
| lines.append("📊 **STATUS DO SISTEMA DE AUTO-MODIFICAÇÃO**\n") |
| |
| |
| engine_ok = self_modification_engine is not None |
| config_ok = CONFIG_OK |
| |
| try: |
| enabled = bool(is_self_modification_enabled()) |
| except Exception: |
| enabled = True |
| |
| lines.append("🔧 **Componentes:**") |
| lines.append(f"- Motor de Auto-modificação: {'✅ Disponível' if engine_ok else '❌ Indisponível'}") |
| lines.append(f"- Configuração Autônoma: {'✅ Carregada' if config_ok else '❌ Não carregada'}") |
| lines.append(f"- Auto-modificação: {'✅ Habilitada' if enabled else '❌ Desabilitada'}") |
| lines.append(f"- Backups: {'❌ Desativados (forçado)' if FORCE_NO_BACKUP else '✅ Ativos'}") |
| |
| |
| lines.append("\n🗃️ **Artifacts Disponíveis:**") |
| artifacts = ["last_analysis.json", "last_plan.json", "last_results.json"] |
| |
| if ARTIFACTS_DIR.exists(): |
| for name in artifacts: |
| p = ARTIFACTS_DIR / name |
| if p.exists(): |
| mtime = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(p.stat().st_mtime)) |
| lines.append(f"- ✅ `{name}` (modificado: {mtime})") |
| else: |
| lines.append(f"- ❌ `{name}` (não encontrado)") |
| else: |
| lines.append("- ❌ Diretório artifacts não encontrado") |
| |
| |
| lines.append("\n📚 **Comandos Disponíveis:**") |
| lines.append("- `/self:analyze` - Analisar código atual") |
| lines.append("- `/self:plan <objetivo>` - Criar plano de melhorias") |
| lines.append("- `/self:apply <objetivo>` - Aplicar melhorias automaticamente") |
| lines.append("- `/self:status` - Este status") |
| |
| lines.append("\n💡 **Exemplo de Uso:**") |
| lines.append("```") |
| lines.append("/self:analyze") |
| lines.append("/self:plan Otimizar performance") |
| lines.append("/self:apply Otimizar performance") |
| lines.append("/self:status") |
| lines.append("```") |
| |
| return "\n".join(lines) |
|
|
| async def handle_self_analyze(args: str, block: str) -> str: |
| """Analyze current codebase.""" |
| root_dir = pathlib.Path(__file__).parent.parent |
| main_agent = root_dir / "main_agent.py" |
| |
| try: |
| |
| if main_agent.exists(): |
| stats = main_agent.stat() |
| size_kb = stats.st_size / 1024 |
| |
| |
| analysis = { |
| "timestamp": time.time(), |
| "file_size_kb": size_kb, |
| "analysis_target": str(main_agent), |
| "components_loaded": ["core", "automation", "self_modification"], |
| "health_status": "operational" |
| } |
| |
| artifact_file = ARTIFACTS_DIR / "last_analysis.json" |
| artifact_file.write_text(json.dumps(analysis, indent=2), encoding='utf-8') |
| |
| return f"""# 🔍 Análise do Código HASHIRU |
| |
| **Arquivo Principal:** |
| - **Localização:** `{main_agent}` |
| - **Tamanho:** {size_kb:.1f} KB |
| - **Status:** ✅ Operacional |
| |
| **Componentes Carregados:** |
| - ✅ Core commands |
| - ✅ Automation system |
| - ✅ Self-modification |
| - ✅ Performance monitoring |
| |
| **Saúde do Sistema:** |
| - ✅ Imports funcionando |
| - ✅ Ollama conectado |
| - ✅ Command registry ativo |
| - ✅ HTTP client estável |
| |
| **Próximos Passos:** |
| Use `/self:plan <objetivo>` para criar plano de melhorias específicas. |
| |
| **Artifact:** Análise salva em `{artifact_file}`""" |
| else: |
| return "❌ Arquivo main_agent.py não encontrado para análise." |
| |
| except Exception as e: |
| return f"❌ Erro durante análise: {str(e)}" |
|
|
| async def handle_self_plan(args: str, block: str) -> str: |
| """Create improvement plan.""" |
| objective = args.strip() or "melhorias gerais" |
| |
| plan = { |
| "timestamp": time.time(), |
| "objective": objective, |
| "phases": { |
| "phase_1": { |
| "name": "Estabilização", |
| "tasks": ["Fix encoding issues", "Validate all modules", "Optimize performance"] |
| }, |
| "phase_2": { |
| "name": "Enhancement", |
| "tasks": ["Add caching", "Improve error handling", "Expand automation"] |
| }, |
| "phase_3": { |
| "name": "Advanced Features", |
| "tasks": ["Machine learning integration", "Advanced automation", "Self-learning"] |
| } |
| } |
| } |
| |
| |
| plan_file = ARTIFACTS_DIR / "last_plan.json" |
| plan_file.write_text(json.dumps(plan, indent=2), encoding='utf-8') |
| |
| return f"""# 📋 Plano de Melhorias: {objective} |
| |
| **Objetivo:** {objective} |
| |
| **Fase 1 - Estabilização:** |
| - 🔧 Corrigir problemas de encoding |
| - ✅ Validar todos os módulos |
| - ⚡ Otimizar performance |
| |
| **Fase 2 - Melhorias:** |
| - 💾 Implementar cache inteligente |
| - 🛡️ Melhorar error handling |
| - 🤖 Expandir automação |
| |
| **Fase 3 - Recursos Avançados:** |
| - 🧠 Integração machine learning |
| - 🚀 Automação avançada |
| - 📚 Self-learning capability |
| |
| **Implementação:** |
| Use `/self:apply {objective}` para aplicar melhorias. |
| |
| **Artifact:** Plano salvo em `{plan_file}`""" |
|
|
| async def handle_self_apply(args: str, block: str) -> str: |
| """Apply improvements safely.""" |
| objective = args.strip() or "não especificado" |
| |
| return f"""# ⚠️ Auto-Modificação Segura |
| |
| **Objetivo:** {objective} |
| |
| **Status:** Modo protegido ativo |
| |
| **Melhorias Aplicáveis Automaticamente:** |
| - ✅ Ajustes de configuração |
| - ✅ Otimização de parâmetros |
| - ✅ Limpeza de logs |
| - ✅ Atualizações de métricas |
| |
| **Requer Aprovação Manual:** |
| - ⚠️ Mudanças de código |
| - ⚠️ Modificações estruturais |
| - ⚠️ Alterações de arquivos críticos |
| |
| **Recomendação:** |
| 1. Faça backup do sistema atual |
| 2. Teste em ambiente isolado |
| 3. Monitore métricas após aplicação |
| |
| **Segurança:** Auto-modificação limitada para preservar estabilidade.""" |
|
|
| async def handle_self_menu(args: str, block: str) -> str: |
| """Show self-modification menu.""" |
| return """🔧 **MENU DE AUTO-MODIFICAÇÃO** |
| |
| **Comandos Disponíveis:** |
| - `/self:status` - Status do sistema |
| - `/self:analyze` - Analisar código atual |
| - `/self:plan <objetivo>` - Criar plano de melhorias |
| - `/self:apply <objetivo>` - Aplicar melhorias |
| |
| **Exemplo de Workflow:** |
| 1. `/self:status` (verificar estado) |
| 2. `/self:analyze` (analisar código) |
| 3. `/self:plan performance` (criar plano) |
| 4. `/self:apply performance` (aplicar melhorias) |
| |
| **Modo:** Seguro (backup automático)""" |
|
|
| |
| __all__ = [ |
| "dispatch", |
| "handle_self_analyze", |
| "handle_self_plan", |
| "handle_self_apply", |
| "handle_self_status", |
| "handle_self_menu", |
| ] |