#!/usr/bin/env python3 import threading as _threading import uuid as _uuid_pg from cyberbrain26.services.safe_oi import progress as _progress_mod _thread_local = _threading.local() """ CyberBrain26 SAFE-OI / FULL-OI API – LiteLLM Phase 1. Běžná konverzace jde přímo přes LiteLLM, nikoli přes Open Interpreter. SAFE i FULL generují návrhy; provádění kódu je v této fázi záměrně zakázané. """ import os import sys import uuid import logging import re from datetime import datetime from pathlib import Path from typing import Any, Dict, List, Optional import sys from pathlib import Path from . import runner as safe_runner import httpx from fastapi import FastAPI, HTTPException from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel import uvicorn from cyberbrain26.memory.auto import remember_if_relevant from cyberbrain26.memory.core import remember, retrieve_active_context from cyberbrain26.services.safe_oi.web_search import ( web_search as _web_search_fn, WEB_SEARCH_TOOL as _WEB_SEARCH_TOOL, ) from cyberbrain26.paths import logs_dir def _safe_oi_log_path(): p = logs_dir() / "safe_oi_api.log" p.parent.mkdir(parents=True, exist_ok=True) return p logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", handlers=[ logging.FileHandler(str(_safe_oi_log_path())), logging.StreamHandler(), ], ) logger = logging.getLogger("safe_oi_api") from cyberbrain26.paths import project_root as _project_root_fn PROJECT_ROOT_STR = str(_project_root_fn()) SANDBOX_PATH = "/tmp/cb26-sandbox" RUNNER_WORKSPACE_PATH = f"{PROJECT_ROOT_STR}/runner-workspace" BACKUPS_PATH = f"{PROJECT_ROOT_STR}/backups" PROXY_URL = os.getenv("CB26_PROXY_URL", "http://127.0.0.1:4000").rstrip("/") PROXY_KEY = os.getenv( "CB26_PROXY_KEY", os.getenv("LITELLM_MASTER_KEY", "sk-cb26-local"), ) DEFAULT_MODEL = os.getenv("CB26_DEFAULT_MODEL", "groq-gpt-oss-120b") HOST = "127.0.0.1" PORT = int(os.getenv("CB26_SAFE_OI_PORT", "7682")) MAX_TOKENS = 1024 HISTORY_LIMIT = 12 MEMORY_LIMIT = 8 REQUEST_TIMEOUT = 90.0 MODEL_ALIASES = { "default": "groq-gpt-oss-120b", "fast": "gemini-flash-lite", "gemini": "gemini-flash", "code": "mistral-codestral", "long": "openrouter-nemotron-3-ultra-550b-free", "vision": "openrouter-gemma-4-31b-free", "code-free": "openrouter-north-mini-code-free", } app = FastAPI(title="CyberBrain26 SAFE-OI API", version="0.6-phase1") app.add_middleware( CORSMiddleware, allow_origins=["http://localhost:8080", "http://127.0.0.1:8080"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) sessions: Dict[str, dict] = {} pending_approvals: Dict[str, dict] = {} executed_approvals: Dict[str, dict] = {} class ChatRequest(BaseModel): thread_id: Optional[str] = None message: str model: str = DEFAULT_MODEL mode: str = "SAFE" depth: str = "normal" confirm_full: bool = False class ApproveRequest(BaseModel): action: str = "approve" def resolve_model(model: str) -> str: return MODEL_ALIASES.get(model, model) def get_or_create_session(thread_id: Optional[str], model: str, mode: str) -> str: if not thread_id or thread_id not in sessions: thread_id = thread_id or str(uuid.uuid4()) sessions[thread_id] = { "history": [], "mode": mode.upper(), "model": model, "pending_approval": None, } logger.info("Nová session: %s | mode=%s", thread_id, mode) return thread_id def mode_instruction(mode: str) -> str: mode = (mode or "SAFE").upper() common = f""" Jsi CyberBrain26, lokální AI asistent pro projekt v {PROJECT_ROOT_STR}. Komunikuj česky, prakticky a stručně. Nikdy nevyzrazuj ani nevypisuj hodnoty z .env, API klíče, tokeny, hesla, SSH privátní klíče, cookies ani jiné secrets. Pokud je nutná kontrola .env, pracuj pouze se jmény proměnných bez hodnot. Před akcí stručně řekni plán. Po akci jasně uveď skutečný výsledek, návratový kód, relevantní výstup a další bezpečný krok. Neprohlašuj, že se něco provedlo, pokud nemáš skutečný výsledek nástroje. """ safe = """ Režim SAFE: - Nic automaticky nespouštěj a nic neměň. - Připrav návrhy příkazů nebo kroků v samostatném bash bloku. - Vysvětli, co by příkaz provedl a jaké může mít dopady. """ full = f""" Režim FULL MAX: - Máš oprávnění autonomně provádět read-only diagnostiku lokálního projektu CyberBrain26, služeb a Linux VM. - Smíš samostatně procházet necitlivé soubory, hledat v projektu, číst logy, kontrolovat procesy, porty, health endpointy, modely, Git stav, testy, Python syntax, konfigurace bez secrets, konverzace a metadata paměti. - Smíš autonomně vytvářet, upravovat, přesouvat a mazat běžné pracovní soubory pouze uvnitř: {PROJECT_ROOT_STR} {RUNNER_WORKSPACE_PATH} /tmp/cb26-sandbox - U změny existujícího zdrojového nebo konfiguračního souboru nejdříve vytvoř timestampovanou zálohu do {BACKUPS_PATH}. - Po změně Pythonu proveď py_compile; po změně služby proveď relevantní health-check; po změně logiky proveď cílený funkční test. - Při selhání pokračuj bezpečnou read-only diagnostikou, nalezni konkrétní chybu a navrhni nebo proveď vratelnou opravu v povoleném projektu. - Diagnostika vlastního lokálního programu a jeho funkcí je explicitně povolena. - I ve FULL MAX si vždy vyžádej samostatné potvrzení před: sudo, su, změnou /etc, /usr, /boot, /var/lib nebo domovských dat mimo projekt; apt, pip, npm či jinou instalací; Dockerem; změnou firewallu, sítě, Tailscale nebo Cloudflare Tunnel; veřejným odesláním dat; Git push/force push; git reset --hard; git clean; mazáním databází, konverzací, záloh nebo virtuálních prostředí; reboot, shutdown, kill hromadných procesů; dd, mkfs, mount, chmod/chown -R; a akcemi se secrets. """ return common + (safe if mode == "SAFE" else full) def extract_code_from_response(text: str) -> tuple[Optional[str], Optional[str], list]: code_blocks = re.findall(r"```(\w+)?\n(.*?)```", text, re.DOTALL) if not code_blocks: return None, None, [] language, code = code_blocks[-1] return language or "text", code.strip(), code_blocks def classify_action(code: str, language: str): """ Vrací (category, reasons). Kategorie: - read: bezpečné čtení a diagnostika, automaticky ve FULL. - project_write: běžná vratelná práce v povoleném projektu/sandboxu, automaticky ve FULL. - write: citlivá/riziková činnost, vyžaduje approval. """ code = (code or "").strip() language = (language or "bash").strip().lower() lowered = code.lower() # --- WHITELIST: .env priklady/sablony jsou bezpecne -> read --- # MUSI byt PRED vsemi deny kontrolami! import re as _re_env_wl if _re_env_wl.search( r"\.env(\.(lab|local|production|secrets|prod|staging|test))?\.(example|sample|template|dist)(?![\w.])", code, ): return "read", ["priklad nebo sablona .env souboru"] # --- HARD DENY (nikdy nespustit, ani s approval) --- hard_deny = [ "rm -rf /", "rm -rf /*", "rm -rf /home", "rm -rf /etc", "rm -rf /usr", "rm -rf /var", "rm -rf /boot", "rm -rf ~", "dd if=", "dd of=", "mkfs", ":(){:|:&};:", "> /dev/", "chmod -r 000 /", "chown -r root", "/etc/passwd", "/etc/shadow", ".ssh/id_rsa", ".ssh/id_ed25519", "authorized_keys", "gemini_api_key", "litellm_master_key", "openrouter_api_key", "/home/oem/.env", "gemini_api_key", "litellm_master_key", "openrouter_api_key", "groq_api_key", "mistral_api_key", "cerebras_api_key", "/.ssh/", "id_rsa", "id_ed25519", "authorized_keys", "known_hosts", ".netrc", ".pgpass", "credentials.json", "service_account", ".aws/", ".gcloud/", ".config/gcloud", ".docker/config.json", "chroma.sqlite3", ] for pat in hard_deny: if pat in lowered: return "deny", [f"zakazany pattern: {pat}"] # --- Cteni .env souboru (regex, bezpecne) --- import re as _re_env # Matchuje: /.env, /.env , " ' ` pred .env, konec radku/mezera za .env # NEmatCHuje: .env.lab, .env.example (tecky za) if _re_env.search(r"(? deny --- import re as _re for _m in _re.finditer(r'>>?\s*(/\S+)', code): target = _m.group(1).rstrip(';|&') allowed = ( "/home/oem/CyberBrain26-lab/", "/tmp/cb26-sandbox/", ) if not any(target.startswith(a) for a in allowed): return "deny", [f"zapis mimo povoleny prostor: {target}"] if not code: return "write", ["prázdný nebo nejednoznačný příkaz"] always_approval = [ "sudo", "su ", "doas", "apt ", "apt-get", "dpkg", "snap ", "flatpak ", "pip install", "pip uninstall", "npm install", "npm uninstall", "docker ", "podman ", "reboot", "shutdown", "poweroff", "halt", "mkfs", "fdisk", "parted", "dd ", "mount ", "umount ", "iptables", "nft ", "ufw ", "firewall", "tailscale", "cloudflared", "ngrok", "curl ", "wget ", "ssh ", "scp ", "rsync ", "git push", "git reset --hard", "git clean", "git checkout --", "rm -rf", "rm -r ", "rmdir ", "chmod -r", "chown -r", # .env, id_rsa, atd. presunuty do hard_deny vyse "/etc/", "/usr/", "/boot/", "/var/lib/", "chroma.sqlite3", "workspace/chroma_db", "data/conversations", "venv/", "venv-", ] for token in always_approval: if token in lowered: return "write", [f"citlivá nebo systémová operace: {token}"] shell_operators = [";", "&&", "||", "`", "$(", "\n"] if language in {"bash", "sh", "shell", "zsh"}: for operator in shell_operators: if operator in code: return "write", ["složený shell příkaz nebo řetězení"] read_prefixes = ( "pwd", "ls", "find ", "grep ", "rg ", "cat ", "head ", "tail ", "sed -n", "awk ", "cut ", "sort ", "uniq ", "wc ", "stat ", "file ", "du ", "df ", "free", "ps ", "pgrep ", "ss ", "git status", "git diff", "git log", "git branch", "python -m py_compile", "python3 -m py_compile", "pytest", "curl -s http://127.0.0.1", "curl -sS http://127.0.0.1", ) if language in {"bash", "sh", "shell", "zsh"} and lowered.startswith(read_prefixes): if ">" not in code and ">>" not in code: return "read", ["bezpečný čtecí nebo diagnostický příkaz"] allowed_paths = ( PROJECT_ROOT_STR.lower(), SANDBOX_PATH, RUNNER_WORKSPACE_PATH.lower(), ) is_allowed_path = any(p in lowered for p in allowed_paths) project_write_markers = ( "touch ", "mkdir ", "mkdir -p", "cp ", "mv ", "python ", "python3 ", "pytest ", "git add ", "git commit ", "tee ", ">", ) if language in {"bash", "sh", "shell", "zsh"}: if is_allowed_path and lowered.startswith(project_write_markers): return "project_write", ["vratelná práce v povoleném projektu nebo sandboxu"] if lowered.startswith(("touch ", "mkdir ", "mkdir -p")) and "/tmp/cb26-sandbox" in lowered: return "project_write", ["vratelná práce v sandboxu"] if language in {"python", "python3"}: dangerous_python = ( "subprocess", "os.system", "shutil.rmtree", "os.remove", "pathlib.path.unlink", "requests.", "urllib", "socket", ) if any(token in lowered for token in dangerous_python): return "write", ["Python kód obsahuje citlivou operaci"] if any(path in lowered for path in allowed_paths): return "project_write", ["Python práce v povoleném projektu nebo sandboxu"] return "read", ["neprivilegovaný Python diagnostický kód"] return "write", ["neznámý nebo nejednoznačný typ akce"] def build_messages(session: dict, normalized_message: str, mode: str, depth: str) -> List[dict]: messages: List[dict] = [{"role": "system", "content": mode_instruction(mode)}] try: records = retrieve_active_context(query=normalized_message, limit=MEMORY_LIMIT) if records: memory_text = "\n".join(f"- {item['content']}" for item in records) messages.append({ "role": "system", "content": ( "Relevantní dlouhodobá paměť CyberBrain26. Použij ji jako kontext. " "Netvrď, že je něco uložené, pokud to není v tomto seznamu:\n" + memory_text ), }) except Exception as exc: logger.warning("Nelze načíst paměť: %s", exc) for item in session["history"][-HISTORY_LIMIT:]: role = item.get("role") content = item.get("content") if role in {"user", "assistant"} and isinstance(content, str) and content: messages.append({"role": role, "content": content}) prefix = "[Přemýšlej důkladně] " if depth == "deep" else "" messages.append({"role": "user", "content": prefix + normalized_message}) return messages def call_proxy(model: str, messages: List[dict]) -> str: try: with httpx.Client(timeout=REQUEST_TIMEOUT) as client: response = client.post( f"{PROXY_URL}/v1/chat/completions", headers={ "Authorization": f"Bearer {PROXY_KEY}", "Content-Type": "application/json", }, json={ "model": model, "messages": messages, "max_tokens": MAX_TOKENS, }, ) except httpx.RequestError as exc: raise HTTPException(status_code=502, detail=f"Nelze se spojit s LiteLLM proxy: {exc}") from exc if response.status_code != 200: raise HTTPException( status_code=502, detail=f"Proxy HTTP {response.status_code}: {response.text[:1000]}", ) try: data = response.json() return data["choices"][0]["message"]["content"] except (ValueError, KeyError, IndexError, TypeError) as exc: raise HTTPException(status_code=502, detail=f"Neočekávaná odpověď proxy: {exc}") from exc @app.get("/health") def health(): return { "status": "ok", "service": "safe_oi_api", "version": "0.6-phase1", "sessions": len(sessions), "proxy": PROXY_URL, "execution": "disabled-phase1", } # === Tool integration (patched) === import json as _json_wst from cyberbrain26.services.safe_oi.web_search import ( web_search as _web_search_fn, WEB_SEARCH_TOOL as _WEB_SEARCH_TOOL, ) from cyberbrain26.services.safe_oi.file_tools import ( read_file as _read_file_fn, list_dir as _list_dir_fn, find_files as _find_files_fn, READ_FILE_TOOL as _READ_FILE_TOOL, LIST_DIR_TOOL as _LIST_DIR_TOOL, FIND_FILES_TOOL as _FIND_FILES_TOOL, ) from cyberbrain26.services.safe_oi.system_tools import ( system_status as _system_status_fn, SYSTEM_STATUS_TOOL as _SYSTEM_STATUS_TOOL, ) from cyberbrain26.services.safe_oi.shell_tools import ( run_shell as _run_shell_fn, RUN_SHELL_TOOL as _RUN_SHELL_TOOL, ) _ALL_TOOLS = [ _WEB_SEARCH_TOOL, _READ_FILE_TOOL, _LIST_DIR_TOOL, _FIND_FILES_TOOL, _SYSTEM_STATUS_TOOL, _RUN_SHELL_TOOL, ] def _dispatch_tool(name: str, args: dict): try: if name == "web_search": return _web_search_fn(query=args.get("query", ""), max_results=args.get("max_results", 5)) if name == "read_file": return _read_file_fn(path=args.get("path", ""), max_bytes=args.get("max_bytes", 50000)) if name == "list_dir": return _list_dir_fn(path=args.get("path", ".")) if name == "find_files": return _find_files_fn(pattern=args.get("pattern", ""), path=args.get("path", "."), max_results=args.get("max_results", 50)) if name == "system_status": return _system_status_fn() if name == "run_shell": return _run_shell_fn(command=args.get("command", ""), cwd=args.get("cwd", "."), timeout=args.get("timeout", 30)) return {"status": "error", "error": f"neznamy tool: {name}"} except Exception as e: return {"status": "error", "error": f"{type(e).__name__}: {e}"} def _call_with_tools(model, messages): try: return _call_with_tools_inner(model, messages) except Exception as exc: import logging logging.getLogger(__name__).warning( "Tool calling selhalo (%s), fallback na plain: %s", type(exc).__name__, exc) return call_proxy(model, messages) def _force_tool_call(model, working, tools, _json_wst, httpx, PROXY_URL, PROXY_KEY): """Posle modelu znovu zpravu, aby zavolal tool.""" working2 = working + [{ "role": "user", "content": ( "NEPIS PLAN. ZAVOLEJ TOOL HNED. " "Pouzij tools API (tool_calls), ne text. " "Pokud potrebujes vice informaci, zavolej nekolik toolu za sebou v jednom tahu." ), }] data = _litellm_call_fallback(model, working2, tools, "required") return data # === Model fallback (patch) === _FALLBACK_MODELS = [ "gemini-flash", "mistral-codestral", "groq-gpt-oss-120b", "gemini-flash-lite", ] def _litellm_call_fallback(model, messages, tools, tool_choice="auto"): """Zkusi zadany model, pri chybe (404/429/5xx) zkusi dalsi z fallback seznamu.""" import logging as _lg logger = _lg.getLogger(__name__) candidates = [model] + [m for m in _FALLBACK_MODELS if m != model] tried = [] for m in candidates: try: with httpx.Client(timeout=60.0) as client: r = client.post( f"{PROXY_URL}/v1/chat/completions", headers={"Authorization": f"Bearer {PROXY_KEY}", "Content-Type": "application/json"}, json={"model": m, "messages": messages, "tools": tools, "tool_choice": tool_choice, "max_tokens": 3000, "temperature": 0.1}, ) if r.status_code == 200: data = r.json() data["_used_model"] = m if m != model: logger.warning("Fallback: %s -> %s (puvodni selhal)", model, m) return data if r.status_code in (400, 404, 429, 500, 502, 503, 504): tried.append(f"{m}:HTTP{r.status_code}") logger.warning("Model %s selhal HTTP %s, zkousim dalsi...", m, r.status_code) continue raise RuntimeError(f"HTTP {r.status_code}: {r.text[:200]}") except Exception as e: tried.append(f"{m}:{type(e).__name__}") logger.warning("Model %s vyjimka %s, zkousim dalsi...", m, type(e).__name__) continue raise RuntimeError(f"Vsechny modely selhaly: {tried}") # === konec model fallback === def _call_with_tools_inner(model, messages): working = list(messages) last_text = "" seen = [] _jid = getattr(_thread_local, "job_id", None) if "_thread_local" in globals() else None for it in range(8): if _jid: pct = min(25 + it * 8, 85) _progress_mod.set_phase(_jid, "llm", pct, f"Model pracuje (iterace {it+1}/8)...") data = _litellm_call_fallback(model, working, _ALL_TOOLS, "auto") _used = data.get("_used_model", model) if _jid and _used != model: _progress_mod.set_phase(_jid, "llm", 30, "Fallback: " + _used) choices = data.get("choices") or [] if not choices: raise RuntimeError("zadne choices") msg = choices[0].get("message", {}) calls = msg.get("tool_calls") or [] text = msg.get("content") or "" # Detekce opakovani if text and text in seen: break if text: seen.append(text) seen = seen[-5:] # Pokud model v prvni iteraci neposlal tool a jen popsal plan: # Vynutime tool_choice="required" if not calls and it == 0 and text and len(text) > 30: if _jid: _progress_mod.set_phase(_jid, "llm", 30, "Model popsal plan, vynucuji tool...") try: data2 = _force_tool_call(model, working, _ALL_TOOLS, _json_wst, httpx, PROXY_URL, PROXY_KEY) choices2 = data2.get("choices") or [] if choices2: msg2 = choices2[0].get("message", {}) calls2 = msg2.get("tool_calls") or [] if calls2: calls = calls2 text = msg2.get("content") or "" except Exception: pass # Zpracovani tool callu if calls: working.append({"role": "assistant", "content": text, "tool_calls": calls}) for tc in calls: fn = tc.get("function", {}) or {} tn = fn.get("name", "") try: ta = _json_wst.loads(fn.get("arguments") or "{}") except Exception: ta = {} if _jid: _progress_mod.add_tool(_jid, tn) res = _dispatch_v2(tn, ta) if "_dispatch_v2" in globals() else _dispatch_tool(tn, ta) working.append({"role": "tool", "tool_call_id": tc.get("id", ""), "name": tn, "content": _json_wst.dumps(res, ensure_ascii=False)}) last_text = text continue # Fallback: text-embedded tool tt = _extract_text_tool_v2(text) if "_extract_text_tool_v2" in globals() else None if tt: tn, ta = tt if _jid: _progress_mod.add_tool(_jid, tn) res = _dispatch_v2(tn, ta) if "_dispatch_v2" in globals() else _dispatch_tool(tn, ta) working.append({"role": "assistant", "content": text}) working.append({"role": "user", "content": "Vysledek toolu: " + _json_wst.dumps(res, ensure_ascii=False) + "\n\nNyni odpovez uzivateli."}) continue if text: return text return last_text or "(Model neposkytl odpoved)" @app.post("/chat/start") def chat_start(req: ChatRequest) -> dict: _progress_mod.cleanup_old() job_id = str(_uuid_pg.uuid4()) _progress_mod.create(job_id) _progress_mod.set_phase(job_id, "init", 5, "Prijimam zpravu...") _threading.Thread(target=_chat_worker, args=(job_id, req), daemon=True).start() return {"job_id": job_id} @app.get("/chat/progress/{job_id}") def chat_progress(job_id: str) -> dict: snap = _progress_mod.snapshot(job_id) if not snap: raise HTTPException(404, "job nenalezen") return snap def _chat_worker(job_id: str, req: ChatRequest) -> None: _thread_local.job_id = job_id try: _progress_mod.set_phase(job_id, "context", 15, "Nacitam pamet a kontext...") result = chat(req) _progress_mod.finish(job_id, result=result) except Exception as exc: import traceback _progress_mod.finish(job_id, error=f"{type(exc).__name__}: {exc}") traceback.print_exc() finally: _thread_local.job_id = None # === konec progress === # === fix-all v2: proposals + loop === import re as _re_v2 from cyberbrain26.services.safe_oi.proposals import ( save_proposal as _save_prop_fn, list_proposals as _list_props_fn, read_proposal as _read_prop_fn, update_status as _upd_prop_status, SAVE_PROPOSAL_TOOL as _SAVE_PROPOSAL_TOOL) _TEXT_TOOL_RE_V2 = _re_v2.compile( r"```(?:bash|shell|sh|python)?\s*\n\s*" r"(system_status|read_file|list_dir|find_files|run_shell|web_search|save_proposal)" r"(?:\s*\(|\s+|\n)", _re_v2.IGNORECASE) def _extract_text_tool_v2(text): if not text: return None m = _TEXT_TOOL_RE_V2.search(text) if not m: return None name = m.group(1).lower() args = {} block = _re_v2.search(r"```[^`]*\n(.*?)```", text, _re_v2.DOTALL) if block: body = block.group(1) for k in ("query", "path", "pattern", "command", "title", "description"): vm = _re_v2.search(rf"{k}\s*[:=]\s*(.+?)(?:\n|$)", body) if vm: args[k] = vm.group(1).strip()[:500] return name, args _ALL_TOOLS_V2 = _ALL_TOOLS + [_SAVE_PROPOSAL_TOOL] def _dispatch_v2(name, args): if name == "save_proposal": return _save_prop_fn(title=args.get("title", ""), description=args.get("description", ""), code_changes=args.get("code_changes", ""), files_affected=args.get("files_affected") or []) result = _dispatch_tool(name, args) if name == "run_shell" and isinstance(result, dict) and result.get("status") == "needs_approval": _thread_local.pending_approval_request = {"command": result.get("command", ""), "reason": result.get("reason", "")} return result def _call_with_tools_inner_v2(model, messages): working = list(messages) last_text = "" seen = [] _jid = getattr(_thread_local, "job_id", None) for it in range(8): if _jid: pct = min(25 + it * 8, 85) _progress_mod.set_phase(_jid, "llm", pct, f"Model pracuje (iterace {it+1}/8)...") with httpx.Client(timeout=60.0) as client: r = client.post(f"{PROXY_URL}/v1/chat/completions", headers={"Authorization": f"Bearer {PROXY_KEY}", "Content-Type": "application/json"}, json={"model": model, "messages": working, "tools": _ALL_TOOLS_V2, "tool_choice": "auto", "max_tokens": 3000, "temperature": 0.1}) if r.status_code != 200: raise RuntimeError(f"HTTP {r.status_code}: {r.text[:300]}") data = r.json() choices = data.get("choices") or [] if not choices: raise RuntimeError("zadne choices") msg = choices[0].get("message", {}) calls = msg.get("tool_calls") or [] text = msg.get("content") or "" if text and text in seen: if _jid: _progress_mod.set_phase(_jid, "stop", 85, "Detekovano opakovani, ukoncuji.") break if text: seen.append(text) seen = seen[-5:] if calls: working.append({"role": "assistant", "content": text, "tool_calls": calls}) for tc in calls: fn = tc.get("function", {}) or {} tn = fn.get("name", "") try: ta = _json_wst.loads(fn.get("arguments") or "{}") except Exception: ta = {} if _jid: _progress_mod.add_tool(_jid, tn) res = _dispatch_v2(tn, ta) working.append({"role": "tool", "tool_call_id": tc.get("id", ""), "name": tn, "content": _json_wst.dumps(res, ensure_ascii=False)}) last_text = text continue tt = _extract_text_tool_v2(text) if tt: tn, ta = tt if _jid: _progress_mod.add_tool(_jid, tn) res = _dispatch_v2(tn, ta) working.append({"role": "assistant", "content": text}) working.append({"role": "user", "content": f"Vysledek toolu {tn}: " + _json_wst.dumps(res, ensure_ascii=False) + "\n\nNyni odpovez uzivateli finalni odpovedi."}) continue if text: if _jid: _progress_mod.set_phase(_jid, "writing", 90, "Finalizuji...") return text return last_text or "(Model neposkytl odpoved)" def _call_with_tools_v2(model, messages): try: return _call_with_tools_inner_v2(model, messages) except Exception as exc: import logging logging.getLogger(__name__).warning("V2 selhal (%s): %s", type(exc).__name__, exc) return call_proxy(model, messages) @app.get("/proposals") def _list_props_endpoint(status: str | None = None): return {"proposals": _list_props_fn(status=status)} @app.get("/proposals/{pid}") def _get_prop_endpoint(pid: str): p = _read_prop_fn(pid) if not p: raise HTTPException(404, "proposal nenalezen") return p @app.post("/proposals/{pid}/reject") def _reject_prop_endpoint(pid: str): if not _upd_prop_status(pid, "rejected"): raise HTTPException(404, "proposal nenalezen") return {"status": "rejected", "id": pid} # === konec fix-all v2 === @app.post("/chat") def chat(req: ChatRequest): mode = req.mode.upper() if mode not in {"SAFE", "FULL"}: raise HTTPException(status_code=400, detail="mode musí být SAFE nebo FULL") if mode == "FULL" and not req.confirm_full: raise HTTPException(status_code=403, detail="FULL-OI vyžaduje confirm_full=true") normalized_message = req.message.strip() if not normalized_message: raise HTTPException(status_code=400, detail="Zadej zprávu.") resolved_model = resolve_model(req.model) thread_id = get_or_create_session(req.thread_id, resolved_model, mode) session = sessions[thread_id] session["mode"] = mode session["model"] = resolved_model memory_saved = None if normalized_message.casefold().startswith("zapamatuj si"): memory_content = normalized_message[len("zapamatuj si"):].lstrip(" ,:-") if not memory_content: raise HTTPException( status_code=400, detail="Za příkazem „zapamatuj si“ musí následovat text k uložení.", ) try: record = remember( content=memory_content, kind="fact", source=f"{mode.lower()}-oi-explicit", confidence="user_confirmed", ) memory_saved = { "id": record["id"], "content": record["content"], "kind": record["metadata"]["kind"], } except Exception as exc: raise HTTPException(status_code=500, detail=f"Nepodařilo se uložit paměť: {exc}") from exc messages = build_messages(session, normalized_message, mode, req.depth) logger.info("[%s] %s | %s | %s", thread_id, mode, resolved_model, normalized_message[:120]) reply = _call_with_tools_v2(resolved_model, messages) _pending_req = getattr(_thread_local, "pending_approval_request", None) session["history"].append({"role": "user", "content": normalized_message}) session["history"].append({"role": "assistant", "content": reply}) auto_memory_saved = None if not memory_saved: try: auto_memory_saved = remember_if_relevant( normalized_message, source=f"{mode.lower()}-oi-auto", ) except Exception as exc: logger.warning("Automatické uložení paměti selhalo: %s", exc) approval_id = None has_approval = False auto_run = False run_result = None if _pending_req: _thread_local.pending_approval_request = None approval_id = str(uuid.uuid4()) pending_approvals[approval_id] = {"thread_id": thread_id, "code": _pending_req.get("command", ""), "language": "bash", "category": "write", "reasons": [_pending_req.get("reason", "")], "mode": mode, "status": "pending", "created_at": datetime.now().isoformat()} session["pending_approval"] = approval_id has_approval = True reply = "Ceka na schvaleni: " + _pending_req.get("command", "") language, code, _ = extract_code_from_response(reply) if code: category, reasons = classify_action(code, language) if mode == "FULL" and category == "read": try: run_result = safe_runner.run_approved_code(code, language) auto_run = True result_text = ( f"Status: {run_result.get('status', 'unknown')}\n" f"Return code: {run_result.get('returncode', 'N/A')}\n" ) if run_result.get("stdout"): result_text += f"STDOUT:\n{run_result['stdout']}\n" if run_result.get("stderr"): result_text += f"STDERR:\n{run_result['stderr']}\n" reply += ( "\n\n---\n" "**Automaticky proveden bezpečný čtecí příkaz**\n\n" f"Typ: {category}\n" f"Důvod: {', '.join(reasons)}\n\n" f"```text\n{result_text.strip()}\n```" ) session["history"].append({ "role": "system", "content": ( "Bezpečný čtecí příkaz byl proveden. Výsledek:\n" + result_text ), }) except Exception as exc: reply += ( "\n\n---\n" "**Chyba při automatickém spuštění bezpečného čtecího příkazu**\n\n" f"`{type(exc).__name__}: {exc}`" ) elif category == "deny": # Audit log try: import json as _json audit_path = Path(logs_dir()) / "terminal_audit.jsonl" audit_path.parent.mkdir(parents=True, exist_ok=True) with open(audit_path, "a", encoding="utf-8") as _f: _f.write(_json.dumps({ "ts": datetime.now().isoformat(), "event": "deny", "thread_id": thread_id, "code": code[:500], "language": language, "reasons": reasons, }, ensure_ascii=False) + "\n") except Exception as _e: logger.warning("Audit log selhal: %s", _e) reply += ( "\n\n---\n**⛔ ODMÍTNUTO**\n\n" f"Důvod: {', '.join(reasons)}\n\n" "Tento příkaz je na blacklistu a nebyl spuštěn." ) elif category != "read": approval_id = str(uuid.uuid4()) pending_approvals[approval_id] = { "thread_id": thread_id, "code": code, "language": language, "category": category, "reasons": reasons, "mode": mode, "status": "pending", "created_at": datetime.now().isoformat(), } session["pending_approval"] = approval_id has_approval = True reply += ( f"\n\n---\n**Čeká na potvrzení**\n\nTyp: {category}" f"\nDůvod: {', '.join(reasons)}" "\n\nTento příkaz může měnit systém nebo data. " "Pro provedení vyžaduje samostatné potvrzení." ) result = { "thread_id": thread_id, "reply": reply, "mode": mode, "model": req.model, "auto_run": auto_run, "run_result": run_result, "has_approval": has_approval, "approval_id": approval_id, } if memory_saved: result["memory_saved"] = memory_saved if auto_memory_saved: result["auto_memory_saved"] = auto_memory_saved return result @app.get("/approvals") def list_approvals(): return { "count": len(pending_approvals), "approvals": [ { "approval_id": aid, "category": item.get("category"), "status": item.get("status"), "mode": item.get("mode"), "created_at": item.get("created_at"), } for aid, item in pending_approvals.items() ], } @app.get("/approvals/{approval_id}") def get_approval(approval_id: str): approval = pending_approvals.get(approval_id) if not approval: raise HTTPException(status_code=404, detail="Approval not found") return approval @app.post("/approvals/{approval_id}") def approve_action(approval_id: str, req: ApproveRequest): approval = pending_approvals.get(approval_id) if not approval: raise HTTPException(status_code=404, detail="Approval not found") thread_id = approval["thread_id"] session = sessions.get(thread_id) if req.action == "deny": approval["status"] = "denied" if session: session["pending_approval"] = None executed_approvals[approval_id] = { "code": approval["code"], "language": approval["language"], "result": "Akce byla uživatelem zamítnuta.", "executed_at": datetime.now().isoformat(), "execution": {"status": "denied"}, } pending_approvals.pop(approval_id, None) return { "status": "denied", "approval_id": approval_id, "reply": "❌ Akce byla zamítnuta.", "pending_approval": None, "events": [ { "id": int(datetime.now().timestamp() * 1000), "type": "approval.denied", "payload": {"approval_id": approval_id}, } ], } if req.action != "approve": raise HTTPException(status_code=400, detail="Neznámá akce") approval["status"] = "approved-pending-runner" if session: session["pending_approval"] = None try: run_result = safe_runner.run_approved_code( approval["code"], approval["language"], ) execution_status = run_result.get("status", "unknown") final_status = "executed" if execution_status not in {"error", "failed"} else "failed" result_text = ( f"Status: {execution_status}\n" f"Return code: {run_result.get('returncode', 'N/A')}\n" ) if run_result.get("stdout"): result_text += f"STDOUT:\n{run_result['stdout']}\n" if run_result.get("stderr"): result_text += f"STDERR:\n{run_result['stderr']}\n" except Exception as exc: final_status = "failed" result_text = f"Chyba při provádění: {type(exc).__name__}: {exc}" run_result = { "status": "error", "error": str(exc), } approval["status"] = final_status executed_approvals[approval_id] = { "code": approval["code"], "language": approval["language"], "result": result_text, "executed_at": datetime.now().isoformat(), "execution": run_result, } pending_approvals.pop(approval_id, None) event_type = "approval.executed" if final_status == "executed" else "run.failed" reply_prefix = "✅ Schválená akce byla provedena." if final_status == "executed" else "⛔ Schválená akce selhala." return { "status": final_status, "approval_id": approval_id, "thread_id": thread_id, "reply": f"{reply_prefix}\n\n{result_text}", "result": result_text, "pending_approval": None, "events": [ { "id": int(datetime.now().timestamp() * 1000), "type": event_type, "payload": { "approval_id": approval_id, "status": final_status, }, } ], } @app.get("/executed/{approval_id}") def get_executed(approval_id: str): result = executed_approvals.get(approval_id) if not result: raise HTTPException(status_code=404, detail="Execution not found") return result @app.get("/sessions") def list_sessions(): return { "count": len(sessions), "sessions": [ { "thread_id": thread_id, "mode": session.get("mode"), "model": session.get("model"), "messages": len(session.get("history", [])), "has_pending": session.get("pending_approval") is not None, } for thread_id, session in sessions.items() ], } @app.delete("/sessions/{thread_id}") def delete_session(thread_id: str): if thread_id in sessions: del sessions[thread_id] return {"status": "deleted"} raise HTTPException(status_code=404, detail="Session not found") if __name__ == "__main__": logger.info("Starting SAFE/FULL LiteLLM Phase 1 API on http://%s:%s", HOST, PORT) uvicorn.run(app, host=HOST, port=PORT, log_level="info")