Upload nanobot_tools.py with huggingface_hub

nanobot_tools.py

@@ -0,0 +1,509 @@
+#!/usr/bin/env python3
+"""
+NANOBOT TOOL SWARM
+Each nanobot is a specialized tool. No cloud. No APIs. Local execution only.
+Coordinator routes tasks to specialist nanobots. They run in parallel.
+"""
+import os, sys, json, subprocess, re, hashlib, time
+from concurrent.futures import ThreadPoolExecutor, as_completed
+from pathlib import Path
+from datetime import datetime
+
+# ═══════════════════════════════════════════════════════════════
+# BASE NANOBOT CLASS
+# ═══════════════════════════════════════════════════════════════
+class Nanobot:
+    def __init__(self, name, specialty):
+        self.name = name
+        self.specialty = specialty
+        self.calls = 0
+        self.errors = 0
+        self.last_result = None
+    
+    def execute(self, task):
+        """Override in subclasses. Return (success, result, error)."""
+        raise NotImplementedError
+    
+    def run(self, task):
+        self.calls += 1
+        try:
+            success, result, error = self.execute(task)
+            self.last_result = result
+            if not success:
+                self.errors += 1
+            return success, result, error
+        except Exception as e:
+            self.errors += 1
+            return False, None, str(e)
+
+# ═══════════════════════════════════════════════════════════════
+# SPECIALIST NANOBOTS
+# ═══════════════════════════════════════════════════════════════
+
+class FileSystemNanobot(Nanobot):
+    """Handles file read/write/move/delete/list."""
+    def __init__(self):
+        super().__init__("fs_bot", "filesystem")
+    
+    def execute(self, task):
+        action = task.get("action")
+        path = task.get("path", "")
+        
+        if action == "read":
+            if not os.path.exists(path):
+                return False, None, f"File not found: {path}"
+            with open(path, 'r', errors='ignore') as f:
+                return True, f.read(), None
+        
+        elif action == "write":
+            content = task.get("content", "")
+            os.makedirs(os.path.dirname(path), exist_ok=True)
+            with open(path, 'w') as f:
+                f.write(content)
+            return True, f"Wrote {len(content)} chars to {path}", None
+        
+        elif action == "list":
+            if not os.path.exists(path):
+                return False, None, f"Path not found: {path}"
+            items = []
+            for entry in sorted(os.listdir(path)):
+                full = os.path.join(path, entry)
+                items.append({
+                    "name": entry,
+                    "is_dir": os.path.isdir(full),
+                    "size": os.path.getsize(full) if os.path.isfile(full) else 0
+                })
+            return True, items, None
+        
+        elif action == "exists":
+            return True, os.path.exists(path), None
+        
+        elif action == "delete":
+            if os.path.isfile(path):
+                os.remove(path)
+                return True, f"Deleted {path}", None
+            elif os.path.isdir(path):
+                import shutil
+                shutil.rmtree(path)
+                return True, f"Deleted dir {path}", None
+            return False, None, f"Not found: {path}"
+        
+        elif action == "mkdir":
+            os.makedirs(path, exist_ok=True)
+            return True, f"Created {path}", None
+        
+        else:
+            return False, None, f"Unknown action: {action}"
+
+class GitNanobot(Nanobot):
+    """Handles git operations: status, diff, commit, branch, log."""
+    def __init__(self):
+        super().__init__("git_bot", "git")
+    
+    def _git(self, args, cwd=None):
+        cmd = ['git'] + args
+        result = subprocess.run(cmd, capture_output=True, text=True, cwd=cwd, timeout=30)
+        return result.returncode == 0, result.stdout, result.stderr
+    
+    def execute(self, task):
+        action = task.get("action")
+        cwd = task.get("cwd", "/tmp/fsi_felon")
+        
+        if action == "status":
+            ok, out, err = self._git(['status', '--short'], cwd)
+            return ok, out, err
+        
+        elif action == "diff":
+            ok, out, err = self._git(['diff'], cwd)
+            return ok, out, err
+        
+        elif action == "add":
+            files = task.get("files", ['.'])
+            ok, out, err = self._git(['add'] + files, cwd)
+            return ok, out, err
+        
+        elif action == "commit":
+            msg = task.get("message", "FSI auto-commit")
+            ok, out, err = self._git(['commit', '-m', msg], cwd)
+            return ok, out, err
+        
+        elif action == "log":
+            n = task.get("n", 5)
+            ok, out, err = self._git(['log', f'-n{n}', '--oneline'], cwd)
+            return ok, out, err
+        
+        elif action == "branch":
+            ok, out, err = self._git(['branch'], cwd)
+            return ok, out, err
+        
+        elif action == "checkout":
+            branch = task.get("branch", "main")
+            ok, out, err = self._git(['checkout', branch], cwd)
+            return ok, out, err
+        
+        else:
+            return False, None, f"Unknown git action: {action}"
+
+class CompileNanobot(Nanobot):
+    """Compile-checks Python code. Returns pass/fail with error details."""
+    def __init__(self):
+        super().__init__("compile_bot", "compile")
+    
+    def execute(self, task):
+        code = task.get("code", "")
+        path = task.get("path", "")
+        
+        if path and os.path.exists(path):
+            # Compile existing file
+            result = subprocess.run(
+                ['python3', '-m', 'py_compile', path],
+                capture_output=True, timeout=10
+            )
+        elif code:
+            # Compile string
+            test_path = f"/tmp/compile_bot_{hashlib.md5(code[:200].encode()).hexdigest()[:8]}.py"
+            with open(test_path, 'w') as f:
+                f.write(code)
+            result = subprocess.run(
+                ['python3', '-m', 'py_compile', test_path],
+                capture_output=True, timeout=10
+            )
+            if os.path.exists(test_path):
+                os.remove(test_path)
+        else:
+            return False, None, "No code or path provided"
+        
+        if result.returncode == 0:
+            return True, "COMPILE PASS", None
+        else:
+            err = result.stderr.decode()[:200]
+            return False, "COMPILE FAIL", err
+
+class TestNanobot(Nanobot):
+    """Runs pytest and returns results."""
+    def __init__(self):
+        super().__init__("test_bot", "test")
+    
+    def execute(self, task):
+        path = task.get("path", "/tmp/fsi_felon")
+        try:
+            result = subprocess.run(
+                ['python3', '-m', 'pytest', path, '-q', '--tb=short'],
+                capture_output=True, text=True, timeout=60, cwd=path
+            )
+            passed = result.returncode == 0
+            summary = self._parse_summary(result.stdout + result.stderr)
+            return passed, summary, None if passed else result.stderr[:300]
+        except Exception as e:
+            return False, None, str(e)
+    
+    def _parse_summary(self, output):
+        # Extract "X passed, Y failed" or similar
+        match = re.search(r'(\d+)\s+passed.*?(\d+)\s+failed', output, re.DOTALL)
+        if match:
+            return f"{match.group(1)} passed, {match.group(2)} failed"
+        if "passed" in output:
+            return "All passed"
+        return output[:200]
+
+class SearchNanobot(Nanobot):
+    """Local code search. Greps files, finds patterns. No web, no cloud."""
+    def __init__(self):
+        super().__init__("search_bot", "search")
+    
+    def execute(self, task):
+        query = task.get("query", "")
+        path = task.get("path", "/tmp/fsi_felon")
+        ext = task.get("ext", ".py")
+        
+        matches = []
+        for root, dirs, files in os.walk(path):
+            # Skip hidden and cache
+            dirs[:] = [d for d in dirs if not d.startswith('.') and d != '__pycache__']
+            for f in files:
+                if f.endswith(ext):
+                    fpath = os.path.join(root, f)
+                    try:
+                        with open(fpath, 'r', errors='ignore') as fh:
+                            for i, line in enumerate(fh, 1):
+                                if query.lower() in line.lower():
+                                    matches.append({
+                                        "file": fpath,
+                                        "line": i,
+                                        "content": line.strip()[:100]
+                                    })
+                                    if len(matches) >= 20:
+                                        break
+                    except:
+                        pass
+                    if len(matches) >= 20:
+                        break
+            if len(matches) >= 20:
+                break
+        
+        return True, matches, None
+
+class ShellNanobot(Nanobot):
+    """Runs shell commands. Sandboxed to /tmp/fsi_felon by default."""
+    def __init__(self):
+        super().__init__("shell_bot", "shell")
+    
+    def execute(self, task):
+        cmd = task.get("command", "")
+        cwd = task.get("cwd", "/tmp/fsi_felon")
+        timeout = task.get("timeout", 30)
+        
+        # Safety: block dangerous commands
+        dangerous = ['rm -rf /', 'mkfs', 'dd if=', '>:', 'shutdown', 'reboot']
+        for d in dangerous:
+            if d in cmd:
+                return False, None, f"Blocked dangerous command: {cmd}"
+        
+        try:
+            result = subprocess.run(
+                cmd, shell=True, capture_output=True, text=True,
+                timeout=timeout, cwd=cwd
+            )
+            return result.returncode == 0, result.stdout, result.stderr
+        except Exception as e:
+            return False, None, str(e)
+
+
+# ═══════════════════════════════════════════════════════════════
+# KNOWLEDGE LIBRARY NANOBOT — RAG for SE textbooks
+# ═══════════════════════════════════════════════════════════════
+class LibraryNanobot(Nanobot):
+    """Query the SE knowledge library. Model learns from results."""
+
+    def __init__(self):
+        super().__init__("library", "se_knowledge_query")
+        from knowledge_library import KnowledgeLibrary
+        self.lib = KnowledgeLibrary()
+        self.lib.load()
+    
+    def execute(self, task):
+        query = task.get("query", "")
+        top_k = task.get("top_k", 3)
+        if not query:
+            return False, None, "Missing 'query' parameter"
+        
+        results = self.lib.query(query, top_k=top_k)
+        if not results:
+            return True, "No matching knowledge found", ""
+        
+        output = []
+        for r in results:
+            output.append(f"[{r['topic']}] {r['content'][:500]}")
+        return True, "\n\n".join(output), ""
+
+
+class WebSearchNanobot(Nanobot):
+    """Controlled web search — allowlisted domains only."""
+
+    ALLOWED_DOMAINS = [
+        "docs.python.org",
+        "stackoverflow.com",
+        "github.com",
+        "pypi.org",
+        "realpython.com",
+        "w3schools.com/python",
+        "geeksforgeeks.org/python",
+    ]
+
+    def __init__(self):
+        super().__init__("web_search", "controlled_internet_search")
+    
+    def execute(self, task):
+        query = task.get("query", "")
+        if not query:
+            return False, None, "Missing 'query' parameter"
+        
+        import urllib.request, urllib.parse
+        results = []
+        
+        for domain in self.ALLOWED_DOMAINS:
+            try:
+                encoded = urllib.parse.quote(f"site:{domain} python {query}")
+                url = f"https://www.google.com/search?q={encoded}"
+                results.append(f"[{domain}] Search URL: {url}")
+            except Exception as e:
+                results.append(f"[{domain}] Error: {e}")
+        
+        return True, "\n".join(results[:5]), ""
+
+
+# ═══════════════════════════════════════════════════════════════
+# SWARM COORDINATOR
+# ═══════════════════════════════════════════════════════════════
+class SwarmCoordinator:
+    """Routes tasks to specialist nanobots. Runs parallel when possible."""
+    
+    def __init__(self):
+        self.bots = {
+            "fs": FileSystemNanobot(),
+            "git": GitNanobot(),
+            "compile": CompileNanobot(),
+            "test": TestNanobot(),
+            "search": SearchNanobot(),
+            "shell": ShellNanobot(),
+            "library": LibraryNanobot(),
+            "web_search": WebSearchNanobot(),
+        }
+        self.executor = ThreadPoolExecutor(max_workers=8)
+        self.history = []
+    
+    def dispatch(self, bot_name, task):
+        """Send task to one nanobot. Return result."""
+        bot = self.bots.get(bot_name)
+        if not bot:
+            return False, None, f"Unknown bot: {bot_name}"
+        
+        success, result, error = bot.run(task)
+        self.history.append({
+            "bot": bot_name,
+            "task": task,
+            "success": success,
+            "time": time.time()
+        })
+        return success, result, error
+    
+    def dispatch_parallel(self, tasks):
+        """
+        tasks = [("bot_name", {"action": "...", ...}), ...]
+        Returns list of (success, result, error) in same order.
+        """
+        futures = []
+        for bot_name, task in tasks:
+            bot = self.bots.get(bot_name)
+            if bot:
+                future = self.executor.submit(bot.run, task)
+                futures.append(future)
+            else:
+                futures.append(None)
+        
+        results = []
+        for future in futures:
+            if future is None:
+                results.append((False, None, "Unknown bot"))
+            else:
+                results.append(future.result())
+        return results
+    
+    def swarm_status(self):
+        """Report on all nanobots."""
+        status = {}
+        for name, bot in self.bots.items():
+            status[name] = {
+                "specialty": bot.specialty,
+                "calls": bot.calls,
+                "errors": bot.errors,
+                "health": "HEALTHY" if bot.errors < bot.calls * 0.5 else "DEGRADED" if bot.calls > 0 else "IDLE"
+            }
+        return status
+
+# ═══════════════════════════════════════════════════════════════
+# INTEGRATION WITH FELON
+# ═══════════════════════════════════════════════════════════════
+class FELONSwarmInterface:
+    """High-level interface: FELON agent calls this, swarm handles execution."""
+    
+    def __init__(self):
+        self.coordinator = SwarmCoordinator()
+    
+    def generate_and_verify(self, prompt, app_type):
+        """
+        Full pipeline: generate app → compile → test → git commit.
+        All via nanobot swarm.
+        """
+        print(f"\n[SWARM] Generating {app_type}...")
+        
+        # Step 1: Generate (via FELON agent for now, Keli later)
+        # This would call agent.generate_app() — keeping it simple
+        from agent.agent import FelonAgent
+        from quantum.engine import QNFREConfig, QNFREEngine
+        
+        agent = FelonAgent(QNFREEngine(QNFREConfig.full_config()))
+        result = agent.generate_app(prompt, app_type=app_type)
+        project_dir = result.get("project_dir", "")
+        
+        if not project_dir or not os.path.exists(project_dir):
+            return False, "Generation failed", None
+        
+        print(f"[SWARM] Project: {project_dir}")
+        
+        # Step 2: Compile check (parallel with test discovery)
+        py_files = list(Path(project_dir).rglob("*.py"))
+        compile_tasks = [("compile", {"path": str(f)}) for f in py_files[:5]]
+        compile_results = self.coordinator.dispatch_parallel(compile_tasks)
+        
+        compile_pass = sum(1 for s, r, e in compile_results if s)
+        compile_fail = len(compile_results) - compile_pass
+        
+        print(f"[SWARM] Compile: {compile_pass}/{len(compile_results)} pass")
+        
+        # Step 3: Run tests
+        ok, test_result, test_err = self.coordinator.dispatch("test", {"path": project_dir})
+        print(f"[SWARM] Tests: {test_result if ok else 'FAIL'}")
+        
+        # Step 4: Git commit if all pass
+        if compile_pass == len(compile_results) and ok:
+            self.coordinator.dispatch("git", {"action": "add", "files": ['.'], "cwd": project_dir})
+            self.coordinator.dispatch("git", {
+                "action": "commit",
+                "message": f"FSI: {app_type} generated and verified",
+                "cwd": project_dir
+            })
+            print(f"[SWARM] Committed to git")
+            return True, result, None
+        
+        return False, result, f"Compile: {compile_fail} fail, Tests: {test_result if ok else test_err}"
+
+# ═══════════════════════════════════════════════════════════════
+# CLI INTERFACE
+# ═══════════════════════════════════════════════════════════════
+def main():
+    import argparse
+    parser = argparse.ArgumentParser(description="FSI Nanobot Tool Swarm")
+    parser.add_argument("command", choices=["status", "test", "generate", "search", "git"])
+    parser.add_argument("--path", default="/tmp/fsi_felon")
+    parser.add_argument("--query", default="")
+    parser.add_argument("--app-type", default="web_app")
+    parser.add_argument("--prompt", default="build a web app")
+    args = parser.parse_args()
+    
+    swarm = SwarmCoordinator()
+    
+    if args.command == "status":
+        status = swarm.swarm_status()
+        print("\n  ╔══════════════════════════════════════╗")
+        print("  ║      NANOBOT SWARM STATUS            ║")
+        print("  ╚══════════════════════════════════════╝\n")
+        for name, info in status.items():
+            health_color = "\x1b[32m" if info["health"] == "HEALTHY" else "\x1b[33m" if info["health"] == "DEGRADED" else "\x1b[90m"
+            print(f"  {name:12s} | {info['specialty']:15s} | calls: {info['calls']:4d} | errors: {info['errors']:3d} | {health_color}{info['health']}\x1b[0m")
+        print()
+    
+    elif args.command == "test":
+        ok, result, err = swarm.dispatch("test", {"path": args.path})
+        print(f"Tests: {result if ok else err}")
+    
+    elif args.command == "search":
+        ok, result, err = swarm.dispatch("search", {"query": args.query, "path": args.path})
+        if ok:
+            for match in result[:10]:
+                print(f"  {match['file']}:{match['line']} | {match['content']}")
+    
+    elif args.command == "git":
+        ok, result, err = swarm.dispatch("git", {"action": "status", "cwd": args.path})
+        print(result if ok else err)
+    
+    elif args.command == "generate":
+        interface = FELONSwarmInterface()
+        ok, result, err = interface.generate_and_verify(args.prompt, args.app_type)
+        if ok:
+            print(f"\n✓ Generated and verified: {result.get('project_name')}")
+        else:
+            print(f"\n✗ Failed: {err}")
+
+if __name__ == "__main__":
+    main()