#!/usr/bin/env python3 """ CyberBrain26 LAB - WebApp Kompletní verze s vlastními manažery """ import json import os import re import shutil import subprocess import uuid from datetime import datetime from pathlib import Path from typing import Any import uvicorn from fastapi import FastAPI, HTTPException from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import FileResponse from pydantic import BaseModel from cyberbrain26.paths import project_root as _project_root_fn # ============================================================================= # KONFIGURACE # ============================================================================= BASE_DIR = Path(str(_project_root_fn())) WEB_PORT = int(os.environ.get("CB26_WEB_PORT", os.environ.get("CB26_WEBAPP_PORT", "8180"))) WORKTREES_DIR = BASE_DIR / "workspaces" DATA_DIR = BASE_DIR / "data-lab" TASKS_FILE = DATA_DIR / "tasks.json" # Vytvoření adresářů WORKTREES_DIR.mkdir(parents=True, exist_ok=True) DATA_DIR.mkdir(parents=True, exist_ok=True) # ============================================================================= # WORKTREE MANAGER # ============================================================================= class WorktreeManager: """Správa Git worktree""" def __init__(self, base_repo: str): self.base_repo = Path(base_repo).resolve() self.worktrees_dir = self.base_repo / "workspaces" self.worktrees_dir.mkdir(parents=True, exist_ok=True) def _sanitize(self, name: str) -> str: """Sanitizace názvu pro Git branch a složku""" # Povolit pouze písmena, číslice a pomlčky name = re.sub(r'[^a-zA-Z0-9\-]', '-', name) # Odstranit více pomlček za sebou name = re.sub(r'-+', '-', name) # Odstranit pomlčky na začátku a konci name = name.strip('-') # Omezit délku if len(name) > 50: name = name[:50] return name def create_worktree(self, description: str) -> dict[str, Any]: """Vytvoří nový worktree""" # Sanitizace pro název složky folder_name = self._sanitize(description) # Sanitizace pro branch branch_name = self._sanitize(description) timestamp = datetime.utcnow().strftime("%Y%m%d%H%M%S") # Generování ID task_id = str(uuid.uuid4())[:8] # Cesta a branch path = self.worktrees_dir / task_id branch = f"task/{branch_name}-{timestamp}" # Pokud existuje, smažeme (pro jistotu) if path.exists(): shutil.rmtree(path, ignore_errors=True) # Git worktree add cmd = [ "git", "-C", str(self.base_repo), "worktree", "add", "-b", branch, str(path), "HEAD" ] result = subprocess.run(cmd, capture_output=True, text=True) if result.returncode != 0: raise RuntimeError(f"Git worktree add selhal: {result.stderr}") return { "task_id": task_id, "branch": branch, "path": str(path), "created_at": datetime.utcnow().isoformat() } def get_worktree_path(self, task_id: str) -> Path | None: """Vrátí cestu k worktree""" path = self.worktrees_dir / task_id return path if path.exists() else None def get_diff(self, task_id: str) -> str: """Vrátí diff worktree proti hlavní větvi (včetně nových souborů)""" path = self.get_worktree_path(task_id) if not path: raise FileNotFoundError(f"Worktree {task_id} neexistuje") # Zaregistrujeme nové (untracked) soubory jako "intent to add" # tím se objeví v git diff, ale nic se necommitne subprocess.run( ["git", "-C", str(path), "add", "-N", "."], capture_output=True, text=True ) # Nyní získáme diff (včetně nových souborů) cmd = ["git", "-C", str(path), "diff", "--no-color"] result = subprocess.run(cmd, capture_output=True, text=True) return result.stdout def apply_patch(self, task_id: str, patch_content: str) -> dict[str, Any]: """Aplikuje patch na worktree""" path = self.get_worktree_path(task_id) if not path: raise FileNotFoundError(f"Worktree {task_id} neexistuje") patch_file = path / "temp.patch" patch_file.write_text(patch_content) cmd = ["git", "-C", str(path), "apply", str(patch_file)] result = subprocess.run(cmd, capture_output=True, text=True) patch_file.unlink(missing_ok=True) if result.returncode != 0: raise RuntimeError(f"Patch selhal: {result.stderr}") return {"applied": True, "message": "Patch aplikován"} def merge_worktree(self, task_id: str, target_branch: str = "lab/phase-3a") -> dict[str, Any]: """Sloučí worktree do cílové větve (v hlavním repu)""" path = self.get_worktree_path(task_id) if not path: raise FileNotFoundError(f"Worktree {task_id} neexistuje") # Získáme název větve worktree branch_result = subprocess.run( ["git", "-C", str(path), "rev-parse", "--abbrev-ref", "HEAD"], capture_output=True, text=True, check=True ) task_branch = branch_result.stdout.strip() if not task_branch or task_branch == "HEAD": raise RuntimeError("Nelze zjistit větev worktree") # 1. Commitneme změny ve worktree subprocess.run( ["git", "-C", str(path), "add", "."], check=True, capture_output=True ) commit_result = subprocess.run( ["git", "-C", str(path), "commit", "-m", f"Task {task_id}: worktree changes"], capture_output=True, text=True ) # Pokud není co commitovat, není to chyba if commit_result.returncode != 0 and "nothing to commit" not in commit_result.stdout: raise RuntimeError(f"Commit selhal: {commit_result.stderr}") # 2. Merge v HLAVNÍM repu (ne ve worktree) # Přepneme se do hlavního repa # (předpokládáme, že hlavní repo je na target_branch) current_branch_result = subprocess.run( ["git", "-C", str(self.base_repo), "rev-parse", "--abbrev-ref", "HEAD"], capture_output=True, text=True, check=True ) current_branch = current_branch_result.stdout.strip() # Pokud nejsme na target_branch, přepneme se if current_branch != target_branch: subprocess.run( ["git", "-C", str(self.base_repo), "checkout", target_branch], check=True, capture_output=True, text=True ) # Merge větev tasku do cílové větve merge_result = subprocess.run( ["git", "-C", str(self.base_repo), "merge", task_branch, "--no-ff", "-m", f"Merge task {task_id}"], capture_output=True, text=True ) if merge_result.returncode != 0: raise RuntimeError(f"Merge selhal: {merge_result.stderr or merge_result.stdout}") return { "merged": True, "target_branch": target_branch, "task_branch": task_branch } def remove_worktree(self, task_id: str): """Odstraní worktree""" path = self.get_worktree_path(task_id) if not path: return cmd = ["git", "-C", str(self.base_repo), "worktree", "remove", str(path)] subprocess.run(cmd, check=True, capture_output=True) shutil.rmtree(path, ignore_errors=True) # ============================================================================= # TASK MANAGER # ============================================================================= class TaskManager: """Správa úkolů""" def __init__(self): self.tasks_file = TASKS_FILE self.tasks = self._load() def _load(self) -> dict[str, Any]: """Načte úkoly ze souboru""" if self.tasks_file.exists(): try: return json.loads(self.tasks_file.read_text()) except: return {} return {} def _save(self): """Uloží úkoly do souboru""" self.tasks_file.write_text(json.dumps(self.tasks, indent=2, ensure_ascii=False)) def create_task(self, description: str, worktree_info: dict[str, Any]) -> str: """Vytvoří nový úkol""" task_id = worktree_info["task_id"] self.tasks[task_id] = { "id": task_id, "description": description, "status": "created", "worktree": worktree_info, "diff": None, "review": None, "created_at": datetime.utcnow().isoformat(), "updated_at": datetime.utcnow().isoformat() } self._save() return task_id def get_task(self, task_id: str) -> dict[str, Any] | None: """Vrátí úkol podle ID""" return self.tasks.get(task_id) def update_task(self, task_id: str, data: dict[str, Any]): """Aktualizuje úkol""" if task_id in self.tasks: self.tasks[task_id].update(data) self.tasks[task_id]["updated_at"] = datetime.utcnow().isoformat() self._save() def set_diff(self, task_id: str, diff: str): """Nastaví diff úkolu""" self.update_task(task_id, {"diff": diff}) def set_review(self, task_id: str, review_result: dict[str, Any]): """Nastaví review úkolu""" self.update_task(task_id, {"review": review_result, "status": "reviewing"}) def approve_task(self, task_id: str): """Schválí úkol""" self.update_task(task_id, {"status": "approved"}) def merge_task(self, task_id: str): """Označí úkol jako sloučený""" self.update_task(task_id, {"status": "merged"}) def list_tasks(self, status: str | None = None) -> list[dict[str, Any]]: """Vrátí seznam úkolů""" items = list(self.tasks.values()) if status: items = [t for t in items if t.get("status") == status] return sorted(items, key=lambda x: x.get("created_at", ""), reverse=True) # ============================================================================= # REVIEWER AGENT (jednoduchá verze) # ============================================================================= class ReviewerAgent: """Recenzent kódu (jednoduchá verze)""" def review_diff(self, diff: str, description: str = "") -> dict[str, Any]: """Zkontroluje diff a vrátí hodnocení""" if not diff or not diff.strip(): return { "safe": True, "risk": "low", "comment": "Žádné změny k review" } # Pro testování automaticky schválíme return { "safe": True, "risk": "low", "comment": f"Automaticky schváleno: {description}" } # ============================================================================= # FASTAPI APLIKACE # ============================================================================= # Inicializace managerů wm = WorktreeManager(str(BASE_DIR)) tm = TaskManager() reviewer = ReviewerAgent() # FastAPI app app = FastAPI( title="CyberBrain26 LAB", version="3.1", description="Worktree management API" ) # CORS app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # ============================================================================= # MODELY # ============================================================================= class TaskCreate(BaseModel): description: str branch_name: str | None = None class TaskResponse(BaseModel): task_id: str worktree: dict[str, Any] class DiffResponse(BaseModel): diff: str class ReviewResponse(BaseModel): safe: bool risk: str comment: str # ============================================================================= # ENDPOINTY # ============================================================================= @app.get("/") def root(): """Kořenový endpoint""" return { "status": "ok", "service": "CyberBrain26 LAB", "version": "3.1" } @app.get("/health") def health(): """Health check""" return { "status": "ok", "service": "webapp", "port": WEB_PORT } # === WORKTREE ENDPOINTY === @app.post("/v1/worktree/tasks", response_model=TaskResponse) def create_task(data: TaskCreate): """Vytvoří nový úkol s worktree""" try: worktree_info = wm.create_worktree(data.description) task_id = tm.create_task(data.description, worktree_info) return {"task_id": task_id, "worktree": worktree_info} except Exception as e: raise HTTPException(status_code=400, detail=str(e)) @app.get("/v1/worktree/tasks") def list_tasks(status: str | None = None): """Vrátí seznam úkolů""" return {"tasks": tm.list_tasks(status)} @app.get("/v1/worktree/tasks/{task_id}") def get_task(task_id: str): """Vrátí detail úkolu""" task = tm.get_task(task_id) if not task: raise HTTPException(status_code=404, detail="Task not found") return task @app.get("/v1/worktree/tasks/{task_id}/diff", response_model=DiffResponse) def get_diff(task_id: str): """Vrátí diff úkolu""" task = tm.get_task(task_id) if not task: raise HTTPException(status_code=404, detail="Task not found") try: diff = wm.get_diff(task_id) tm.set_diff(task_id, diff) return {"diff": diff} except FileNotFoundError as e: raise HTTPException(status_code=404, detail=str(e)) except Exception as e: raise HTTPException(status_code=400, detail=str(e)) @app.post("/v1/worktree/tasks/{task_id}/review", response_model=ReviewResponse) def review_task(task_id: str): """Zahájí review úkolu""" task = tm.get_task(task_id) if not task: raise HTTPException(status_code=404, detail="Task not found") # Získat diff diff = task.get("diff") if not diff: try: diff = wm.get_diff(task_id) tm.set_diff(task_id, diff) except Exception as e: raise HTTPException(status_code=400, detail=f"Cannot get diff: {e!s}") if not diff or not diff.strip(): raise HTTPException(status_code=400, detail="No diff to review") # Review result = reviewer.review_diff(diff, task.get("description", "")) tm.set_review(task_id, result) return result @app.post("/v1/worktree/tasks/{task_id}/approve") def approve_task(task_id: str): """Schválí úkol""" task = tm.get_task(task_id) if not task: raise HTTPException(status_code=404, detail="Task not found") review = task.get("review") if not review: raise HTTPException(status_code=400, detail="Task must be reviewed first") if not review.get("safe", False): raise HTTPException(status_code=403, detail="Review did not pass safety") tm.approve_task(task_id) return {"status": "approved"} @app.post("/v1/worktree/tasks/{task_id}/merge") def merge_task(task_id: str): """Sloučí worktree do hlavní větve""" task = tm.get_task(task_id) if not task: raise HTTPException(status_code=404, detail="Task not found") if task.get("status") != "approved": raise HTTPException(status_code=403, detail="Task must be approved before merge") try: result = wm.merge_worktree(task_id) tm.merge_task(task_id) return result except Exception as e: raise HTTPException(status_code=400, detail=str(e)) @app.delete("/v1/worktree/tasks/{task_id}") def delete_task(task_id: str, force: bool = False): """Smaže úkol a worktree""" task = tm.get_task(task_id) if not task: raise HTTPException(status_code=404, detail="Task not found") if task.get("status") not in ["merged", "rejected"] and not force: raise HTTPException( status_code=400, detail="Task not finished. Use force=true to remove anyway." ) try: wm.remove_worktree(task_id) except: pass tm.tasks.pop(task_id, None) tm._save() return {"status": "removed"} # ============================================================================= # HLAVNÍ SPUŠTĚNÍ # ============================================================================= # ============================================================================= # 3D: TERMINAL & VSCODIUM ENDPOINTY # ============================================================================= from cyberbrain26.core.terminal_service import TerminalService ts = TerminalService() class ExecRequest(BaseModel): task_id: str command: str timeout: int = 30 class VSCodiumRequest(BaseModel): task_id: str @app.post("/v1/terminal/exec") def terminal_exec(data: ExecRequest): """Spustí příkaz v worktree daného tasku""" task = tm.get_task(data.task_id) if not task: raise HTTPException(status_code=404, detail="Task not found") worktree_path = task.get("worktree", {}).get("path") if not worktree_path: raise HTTPException(status_code=400, detail="Task nemá worktree") return ts.execute( task_id=data.task_id, worktree_path=worktree_path, command=data.command, timeout=data.timeout, ) @app.get("/v1/terminal/history/{task_id}") def terminal_history(task_id: str, limit: int = 50): """Vrátí historii příkazů pro task""" return {"history": ts.history(task_id, limit)} @app.post("/v1/terminal/stop/{execution_id}") def terminal_stop(execution_id: str): """Zastaví běžící příkaz""" return ts.stop(execution_id) @app.post("/v1/vscodium/open") def vscodium_open(data: VSCodiumRequest): """Otevře worktree ve VSCodium / code-serveru""" task = tm.get_task(data.task_id) if not task: raise HTTPException(status_code=404, detail="Task not found") worktree_path = task.get("worktree", {}).get("path") if not worktree_path: raise HTTPException(status_code=400, detail="Task nemá worktree") # Zkusíme různé možnosti # 1. code-server s URL import shutil if shutil.which("code-server"): # code-server obvykle běží na portu 8080 nebo 8443 # Vrátíme URL, kde si uživatel otevře složku return { "opened": False, "mode": "code-server", "url": f"http://0.0.0.0:8443/?folder={worktree_path}", "worktree_path": worktree_path, "message": "Otevřete URL v prohlížeči (přes SSH tunel)", } # 2. codium (desktop) if shutil.which("codium"): import subprocess try: subprocess.Popen(["codium", worktree_path]) return { "opened": True, "mode": "codium", "worktree_path": worktree_path, } except Exception as e: return {"opened": False, "error": str(e)} # 3. code (VSCode) if shutil.which("code"): import subprocess try: subprocess.Popen(["code", worktree_path]) return { "opened": True, "mode": "code", "worktree_path": worktree_path, } except Exception as e: return {"opened": False, "error": str(e)} # 4. Nic není k dispozici return { "opened": False, "error": "VSCodium ani code-server není nainstalován", "worktree_path": worktree_path, } @app.get("/ui") def ui(): return FileResponse(str(Path(__file__).resolve().parent / "static_ui.html")) # ============================================================================= # 3E: OPENCLAW ENDPOINTY # ============================================================================= from cyberbrain26.core.openclaw_adapter import OpenClawAdapter oa = OpenClawAdapter(str(_project_root_fn())) class OpenClawStart(BaseModel): task_id: str max_steps: int = 10 @app.post("/v1/openclaw/start") def openclaw_start(data: OpenClawStart): task = tm.get_task(data.task_id) if not task: raise HTTPException(status_code=404, detail="Task not found") wt = task.get("worktree", {}).get("path") if not wt: raise HTTPException(status_code=400, detail="Task nema worktree") return oa.start(data.task_id, wt, task.get("description", ""), data.max_steps) @app.get("/v1/openclaw/status/{run_id}") def openclaw_status(run_id: str): s = oa.status(run_id) if not s: raise HTTPException(status_code=404, detail="Run not found") return s @app.post("/v1/openclaw/stop/{run_id}") def openclaw_stop(run_id: str): return oa.stop(run_id) @app.get("/v1/openclaw/runs") def openclaw_runs(task_id: str | None = None): return {"runs": oa.list_runs(task_id)} @app.get("/v1/openclaw/result/{run_id}") def openclaw_result(run_id: str): r = oa.get_result(run_id) if not r: raise HTTPException(status_code=404, detail="Result not found") return r @app.get("/v1/openclaw/log/{run_id}") def openclaw_log(run_id: str, tail: int = 100): return {"log": oa.get_log(run_id, tail)} # === Faze 3F + 3G integrace (patch modul) === from cyberbrain26.core.webapp_phase3fg import register as register_phase3fg register_phase3fg(app) if __name__ == "__main__": print("=" * 60) print(" CyberBrain26 LAB - WebApp") print("=" * 60) print(f" Base dir: {BASE_DIR}") print(f" Worktrees: {WORKTREES_DIR}") print(f" Data: {DATA_DIR}") print(f" Tasks file: {TASKS_FILE}") print("-" * 60) print(f" Starting server on http://0.0.0.0:{WEB_PORT}") print(" Press Ctrl+C to stop") print("=" * 60) uvicorn.run( app, host="0.0.0.0", port=WEB_PORT, log_level="info" )