"""OpenClaw Worker - LLM coding agent running as isolated subprocess.""" import argparse import json import os import re import shlex import subprocess import sys import urllib.error import urllib.request from datetime import datetime from pathlib import Path LITELLM_URL = os.environ.get("OPENCLAW_LITELLM_URL", "http://127.0.0.1:4100") LITELLM_KEY = os.environ.get("LITELLM_MASTER_KEY", "sk-dummy") MODEL = os.environ.get("OPENCLAW_MODEL", "gemini-flash") ALLOWED_COMMANDS = { "ls", "cat", "head", "tail", "grep", "find", "wc", "pwd", "echo", "git", "python3", "pip", "pip3", "pytest", "tree", "file", "du", "df", } FORBIDDEN_PATTERNS = [ r"\brm\s+-rf\b", r"\bsudo\b", r"\bsu\b", r"\bcurl\b", r"\bwget\b", r"\bssh\b", r"\bmkfs\b", r"\bdd\b", r"\bshutdown\b", r"\breboot\b", r"\bkill\b", r"\bpkill\b", r">\s*/dev/", r">\s*/etc/", ] SYSTEM_PROMPT = """Jsi OpenClaw, autonomni kodovaci agent v izolovanem worktree. Nastroje (odpovidej POUZE jednim JSON objektem): - {"tool": "read_file", "path": "relativni/cesta"} - {"tool": "write_file", "path": "relativni/cesta", "content": "..."} - {"tool": "run_command", "command": "ls -la"} - {"tool": "list_files", "path": "."} - {"tool": "done", "summary": "Co jsi udelal"} Pravidla: - Vzdy odpovidej pouze jednim JSON objektem, nic jineho. - Nepouzivej zakazane prikazy (rm -rf, sudo, curl, atd.). - Zustan v ramci worktree. - Az budes hotov, zavolej {"tool": "done", "summary": "..."}. """ class OpenClawWorker: def __init__(self, task_id, worktree_path, description, max_steps=10): self.task_id = task_id self.worktree = Path(worktree_path).resolve() self.description = description self.max_steps = max_steps self.transcript = [] if not self.worktree.exists(): raise FileNotFoundError(f"Worktree neexistuje: {worktree_path}") def _log(self, entry): entry["timestamp"] = datetime.utcnow().isoformat() entry["task_id"] = self.task_id self.transcript.append(entry) t = entry.get("type", "?") print(f"[{t}] {json.dumps({k:v for k,v in entry.items() if k not in ('type','timestamp','task_id')}, ensure_ascii=False)[:300]}", flush=True) def _check_path(self, p): full = (self.worktree / p).resolve() if not str(full).startswith(str(self.worktree)): raise PermissionError(f"Cesta mimo worktree: {p}") return full def _check_command(self, cmd): for pat in FORBIDDEN_PATTERNS: if re.search(pat, cmd, re.IGNORECASE): raise PermissionError(f"Zakazany vzor: {pat}") try: parts = shlex.split(cmd) except ValueError as e: raise ValueError(f"Chyba syntaxe: {e}") if not parts: raise ValueError("Prazdny prikaz") if parts[0] not in ALLOWED_COMMANDS: raise PermissionError(f"Prikaz '{parts[0]}' neni povolen") def tool_read_file(self, path): p = self._check_path(path) if not p.is_file(): return f"CHYBA: soubor neexistuje: {path}" return p.read_text(encoding="utf-8", errors="replace")[:5000] def tool_write_file(self, path, content): p = self._check_path(path) p.parent.mkdir(parents=True, exist_ok=True) p.write_text(content, encoding="utf-8") return f"OK: zapsano {len(content)} bajtu do {path}" def tool_list_files(self, path="."): p = self._check_path(path) if not p.is_dir(): return f"CHYBA: neni adresar: {path}" items = sorted(p.iterdir())[:100] return "\n".join(f"{'[D]' if x.is_dir() else '[F]'} {x.name}" for x in items) def tool_run_command(self, command): self._check_command(command) try: r = subprocess.run( shlex.split(command), shell=False, cwd=str(self.worktree), capture_output=True, text=True, timeout=30 ) out = (r.stdout or "")[:3000] err = (r.stderr or "")[:1000] return f"exit={r.returncode}\nSTDOUT:\n{out}\nSTDERR:\n{err}" except subprocess.TimeoutExpired: return "CHYBA: timeout (30s)" def _call_llm(self, messages): url = f"{LITELLM_URL}/v1/chat/completions" payload = json.dumps({ "model": MODEL, "messages": messages, "temperature": 0.2, "max_tokens": 1000, }).encode("utf-8") req = urllib.request.Request(url, data=payload, method="POST") req.add_header("Content-Type", "application/json") req.add_header("Authorization", f"Bearer {LITELLM_KEY}") try: with urllib.request.urlopen(req, timeout=60) as resp: data = json.loads(resp.read().decode("utf-8")) return data["choices"][0]["message"]["content"] except urllib.error.HTTPError as e: body = e.read().decode("utf-8", errors="replace")[:500] raise RuntimeError(f"LLM HTTP {e.code}: {body}") except Exception as e: raise RuntimeError(f"LLM chyba: {e}") def _parse_tool(self, text): text = text.strip() if text.startswith("```"): lines = text.split("\n") text = "\n".join(lines[1:-1]) if len(lines) > 2 else text start = text.find("{") end = text.rfind("}") if start < 0 or end < 0: raise ValueError(f"Nelze najit JSON: {text[:200]}") obj = json.loads(text[start:end+1]) if "tool" not in obj: raise ValueError("Chybi 'tool' v odpovedi") return obj def run(self): self._log({"type": "start", "description": self.description, "model": MODEL}) messages = [ {"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": f"Ukol: {self.description}\n\nWorktree: {self.worktree}\n\nZacni pruzkumem souboru."} ] success = False summary = "" for step in range(1, self.max_steps + 1): try: response = self._call_llm(messages) self._log({"type": "llm_response", "step": step, "content": response[:500]}) tool_call = self._parse_tool(response) except Exception as e: self._log({"type": "error", "step": step, "error": str(e)}) break tool_name = tool_call.get("tool") messages.append({"role": "assistant", "content": json.dumps(tool_call)}) try: if tool_name == "read_file": result = self.tool_read_file(tool_call["path"]) elif tool_name == "write_file": result = self.tool_write_file(tool_call["path"], tool_call.get("content", "")) elif tool_name == "list_files": result = self.tool_list_files(tool_call.get("path", ".")) elif tool_name == "run_command": result = self.tool_run_command(tool_call["command"]) elif tool_name == "done": success = True summary = tool_call.get("summary", "") self._log({"type": "done", "step": step, "summary": summary}) break else: result = f"CHYBA: neznamy nastroj '{tool_name}'" except Exception as e: result = f"CHYBA: {e}" self._log({"type": "tool_result", "step": step, "tool": tool_name, "result": result[:500]}) messages.append({"role": "user", "content": f"Vysledek:\n{result}"}) return { "success": success, "steps": len(self.transcript), "summary": summary, "transcript": self.transcript, "worktree": str(self.worktree), "task_id": self.task_id, } if __name__ == "__main__": parser = argparse.ArgumentParser() parser.add_argument("--task-id", required=True) parser.add_argument("--worktree", required=True) parser.add_argument("--description", required=True) parser.add_argument("--max-steps", type=int, default=10) parser.add_argument("--output", required=True) args = parser.parse_args() try: worker = OpenClawWorker(args.task_id, args.worktree, args.description, args.max_steps) result = worker.run() except Exception as e: result = {"success": False, "error": str(e), "transcript": [], "task_id": args.task_id} Path(args.output).write_text(json.dumps(result, indent=2, ensure_ascii=False)) sys.exit(0 if result.get("success") else 1)