diff --git a/platforms/claude/commands/grok-swarm-agent.sh b/platforms/claude/commands/grok-swarm-agent.sh new file mode 100644 index 0000000..6455340 --- /dev/null +++ b/platforms/claude/commands/grok-swarm-agent.sh @@ -0,0 +1,81 @@ +#!/bin/bash +# grok-swarm-agent command for Claude Code +# Invokes the grok_agent.py Python script with Claude Code context + +set -e + +# Find the plugin root (3 levels up from commands/) +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +PLUGIN_ROOT="$(cd "$SCRIPT_DIR/../../.." && pwd)" +AGENT_SCRIPT="$PLUGIN_ROOT/src/agent/grok_agent.py" + +# Default: preview mode +APPLY_FLAG="" +MAX_ITERATIONS="5" +VERIFY_CMD="" +TARGET="." + +# Parse arguments +TASK="" +while [[ $# -gt 0 ]]; do + case $1 in + --apply) + APPLY_FLAG="--apply" + shift + ;; + --target) + TARGET="$2" + shift 2 + ;; + --max-iterations) + MAX_ITERATIONS="$2" + shift 2 + ;; + --verify-cmd) + VERIFY_CMD="$2" + shift 2 + ;; + -*) + echo "Unknown option: $1" >&2 + echo "Usage: grok-swarm-agent [task description] [--apply] [--target DIR] [--max-iterations N] [--verify-cmd CMD]" >&2 + exit 1 + ;; + *) + # First non-flag is the task + if [[ -z "$TASK" ]]; then + TASK="$1" + fi + shift + ;; + esac +done + +if [[ -z "$TASK" ]]; then + echo "Usage: grok-swarm-agent [task description] [--apply] [--target DIR] [--max-iterations N] [--verify-cmd CMD]" + echo "" + echo "Example:" + echo " grok-swarm-agent refactor the auth module" + echo " grok-swarm-agent add tests --apply --verify-cmd pytest" + exit 1 +fi + +# Build argument array +ARGS=( + "$AGENT_SCRIPT" + "--platform" "claude" + "--target" "$TARGET" + "--max-iterations" "$MAX_ITERATIONS" +) + +if [[ -n "$APPLY_FLAG" ]]; then + ARGS+=("$APPLY_FLAG") +fi + +if [[ -n "$VERIFY_CMD" ]]; then + ARGS+=("--verify-cmd" "$VERIFY_CMD") +fi + +ARGS+=("$TASK") + +# Execute +python3 "${ARGS[@]}" \ No newline at end of file diff --git a/platforms/claude/skills/grok-swarm-agent/SKILL.md b/platforms/claude/skills/grok-swarm-agent/SKILL.md new file mode 100644 index 0000000..d3fabbb --- /dev/null +++ b/platforms/claude/skills/grok-swarm-agent/SKILL.md @@ -0,0 +1,63 @@ +--- +name: grok-swarm-agent +description: Spawn an autonomous Grok agent to accomplish tasks. Use when asked to "use grok agent to refactor X", "let grok agent handle this", "grok agent mode", "autonomous grok". Triggers: "grok agent", "agent mode", "autonomous grok", "grok-swarm-agent" +author: OpenClaw +version: 1.0.0 +--- + +# Grok Swarm Agent + +Spawn an autonomous agent powered by Grok 4.20 Multi-Agent Beta that iteratively refactors, analyzes, or modifies your codebase. + +## Usage + +``` +use grok agent to refactor src/auth/ +grok agent mode: improve error handling in lib/ +let grok swarm agent add tests to the backend +``` + +## How It Works + +1. **Discover**: Agent finds relevant files in target directory +2. **Plan**: Agent creates modification plan using Grok 4.20 +3. **Apply**: Agent writes changes using file tools +4. **Verify**: Agent validates changes (syntax check, tests) +5. **Iterate**: Agent refines until satisfied or max iterations reached + +## Options + +| Option | Description | +|--------|-------------| +| `--apply` | Actually write files (default is preview mode) | +| `--max-iterations N` | Max agent iterations (default: 5) | +| `--verify-cmd CMD` | Command to verify changes work | + +## Examples + +``` +# Preview mode - shows what would change +grok agent refactor the auth module + +# Actually apply changes +grok agent refactor the auth module --apply + +# With verification +grok agent add tests --apply --verify-cmd "pytest tests/" + +# Analyze with agent +grok agent analyze security vulnerabilities --target ./src +``` + +## Requirements + +- Grok Swarm plugin installed and configured +- OpenRouter API key set up (run `/grok-swarm:setup` if needed) + +## Output + +The agent reports: +- Status (success, max iterations, or errors) +- Number of iterations used +- List of files changed +- Verification results if applicable diff --git a/skills/grok-refactor/openclaw.plugin.json b/skills/grok-refactor/openclaw.plugin.json index 99f41db..5eef67e 100644 --- a/skills/grok-refactor/openclaw.plugin.json +++ b/skills/grok-refactor/openclaw.plugin.json @@ -1,7 +1,7 @@ { "id": "grok-swarm", "name": "Grok Multi-Agent Swarm", - "description": "Bridge to xAI Grok 4.20 Multi-Agent Beta (4-agent swarm) for codebase analysis, refactoring, reasoning, and code generation.", + "description": "Bridge to xAI Grok 4.20 Multi-Agent Beta (4-agent swarm) for codebase analysis, refactoring, reasoning, and code generation. Also supports autonomous agent mode for iterative multi-file modifications.", "version": "1.0.0", "configSchema": { "type": "object", @@ -19,6 +19,10 @@ "type": "string", "description": "Path to grok_bridge.py script" }, + "agentScript": { + "type": "string", + "description": "Path to grok_agent.py script (for autonomous agent mode)" + }, "defaultOutputDir": { "type": "string", "description": "Default output directory for file writes (default: ./grok-output/)" diff --git a/src/agent/__init__.py b/src/agent/__init__.py new file mode 100644 index 0000000..7b3369e --- /dev/null +++ b/src/agent/__init__.py @@ -0,0 +1,17 @@ +""" +grok-swarm-agent - Autonomous agent wrapper for Grok 4.20. + +This module provides an iterative agent loop that: +1. Receives natural language tasks +2. Discovers relevant files +3. Calls Grok 4.20 via grok_bridge +4. Parses code blocks and applies changes +5. Verifies changes with tests +6. Iterates until done or max iterations reached + +Cross-platform: works with both Claude Code and OpenClaw. +""" + +from .grok_agent import main, run_agent_loop, AgentState, Platform, AgentStatus + +__all__ = ["main", "run_agent_loop", "AgentState", "Platform", "AgentStatus"] diff --git a/src/agent/grok_agent.py b/src/agent/grok_agent.py new file mode 100644 index 0000000..e7b87a1 --- /dev/null +++ b/src/agent/grok_agent.py @@ -0,0 +1,700 @@ +#!/usr/bin/env python3 +""" +grok_agent.py — Autonomous agent loop powered by Grok 4.20. + +Minimal viable agent loop: +1. Receive task + target +2. Discover files +3. Call grok_bridge with context +4. Parse response for file operations +5. Apply changes (or preview) +6. Verify (if verify_cmd provided) +7. Iterate or report + +Cross-platform: --platform claude or --platform openclaw + +Usage: + python3 grok_agent.py --task "refactor auth module" --target ./src/auth + python3 grok_agent.py --task "analyze security" --target ./src --apply + python3 grok_agent.py --task "add tests" --target . --apply --verify-cmd "pytest" +""" + +import argparse +import re +import subprocess +import sys +import time +from dataclasses import dataclass, field +from enum import Enum +from pathlib import Path +from typing import Optional + +# Import existing bridge (sibling to agent/ directory) +sys.path.insert(0, str(Path(__file__).parent.parent / "bridge")) +from grok_bridge import call_grok, read_files + +# Import shared patterns +sys.path.insert(0, str(Path(__file__).parent.parent / "shared")) +from patterns import get_filename_pattern_string + + +class Platform(Enum): + CLAUDE = "claude" + OPENCLAW = "openclaw" + + +class AgentStatus(Enum): + RUNNING = "running" + SUCCESS = "success" + FAILED = "failed" + MAX_ITERATIONS = "max_iterations" + NO_FILES = "no_files" + + +@dataclass +class AgentState: + """Agent execution state.""" + task: str + target: str + platform: Platform + apply: bool = False + max_iterations: int = 5 + verify_cmd: Optional[str] = None + output_dir: Optional[str] = None + + iteration: int = 0 + status: AgentStatus = AgentStatus.RUNNING + changes: list = field(default_factory=list) + errors: list = field(default_factory=list) + + # Shared context across iterations + file_context: str = "" + last_response: str = "" + last_verification_output: str = "" + + +# ============================================================================= +# Path Sanitization +# ============================================================================= + +def sanitize_target_path(path_hint: str, base_root: str) -> Path: + """ + Sanitize a path hint to prevent directory traversal and absolute path attacks. + + Args: + path_hint: The path provided by the LLM (should be relative) + base_root: The base directory to write to (target or output_dir) + + Returns: + A safe resolved Path within base_root + + Raises: + ValueError: If the path is unsafe (absolute, escapes root, etc.) + """ + # Reject or strip leading "/" and "~" + hint = path_hint.strip() + if hint.startswith("/"): + hint = hint.lstrip("/") + if hint.startswith("~"): + raise ValueError(f"Cannot use home directory paths: {path_hint}") + + # Convert to Path and check for absolute + raw_path = Path(hint) + if raw_path.is_absolute(): + raise ValueError(f"Cannot use absolute paths: {path_hint}") + + # Get the base root and resolve it + root = Path(base_root) + if root.is_file(): + # If target is a file, use its parent as root + root = root.parent + root = root.resolve() + + # Build the target path and resolve it + target_path = (root / raw_path).resolve() + + # Check if resolved path is within the root + try: + target_path.relative_to(root) + except ValueError: + raise ValueError(f"Path escapes target directory: {path_hint} -> {target_path}") + + return target_path + + +# ============================================================================= +# File Discovery +# ============================================================================= + +def discover_files(target: str, max_files: int = 50) -> list[str]: + """ + Discover relevant code files in target directory. + + Supports: .py, .js, .ts, .tsx, .jsx, .go, .rs, .java, .c, .cpp, .h, .hpp + """ + path = Path(target) + if not path.exists(): + return [] + + if path.is_file(): + return [str(path)] + + # Language extensions to search for + extensions = { + ".py", ".js", ".ts", ".tsx", ".jsx", + ".go", ".rs", ".java", ".c", ".cpp", ".h", ".hpp", + ".cs", ".rb", ".php", ".swift", ".kt", ".scala", + } + + files = [] + for ext in extensions: + files.extend([str(p) for p in path.glob(f"**/*{ext}")]) + + # Sort by path length (shorter = more likely root-level) and limit + files.sort(key=lambda f: (len(Path(f).parts), f)) + return files[:max_files] + + +# ============================================================================= +# Code Block Parsing (Fixed from grok_bridge.py issues) +# ============================================================================= + +def parse_code_blocks(response_text: str) -> list[dict]: + """ + Parse code blocks from Grok response. + + Supports multiple annotation formats: + ```python:/path/to/file.py + # content + ``` + + ```python + // FILE: /path/to/file.py + # content + ``` + + ```python + # FILE: /path/to/file.py + # content + ``` + + ```python + # filename.py + # content + ``` + + Returns list of dicts with keys: language, path_hint, content, inferred_path + """ + blocks = [] + + # Pattern 1: lang:/path/to/file (language tag contains path) + lang_path_pattern = re.compile(r'^(\w+):(/[^/\s\n]+(?:/[^/\s\n]+)*)\n', re.MULTILINE) + + # Pattern 2: // FILE: /path or # FILE: /path + file_marker_pattern = re.compile( + r'^\s*(?:(?://|#)\s*)FILE:\s*(.+?)\s*$', + re.MULTILINE + ) + + # Pattern 3: # filename.py (just filename as first line comment) + filename_pattern = re.compile(get_filename_pattern_string(), re.MULTILINE) + + # Split into code blocks by ``` fences + parts = re.split(r'```', response_text) + + for i, part in enumerate(parts): + if i % 2 == 0: + continue + + # Get first line (may contain language or markers) + first_line_end = part.find('\n') + if first_line_end == -1: + first_line_end = len(part) + + first_line = part[:first_line_end] + rest = part[first_line_end + 1:] + + language = "" + path_hint = "" + content = rest + + # Check pattern 1: lang:/path + lang_match = lang_path_pattern.match(part) + if lang_match: + language = lang_match.group(1) + path_hint = lang_match.group(2) + content = part[lang_match.end():] + blocks.append({ + "language": language, + "path_hint": path_hint, + "content": content.strip(), + "inferred_path": path_hint.split('/')[-1] if '/' in path_hint else path_hint + }) + continue + + # Check for language in first line (e.g., "python") + lang_candidate = first_line.strip() + if lang_candidate and lang_candidate.isalpha(): + language = lang_candidate + + # Check pattern 1b: lang:path on first line (e.g., "python:cli.py" with no trailing newline) + lang_path_on_line = re.compile(r'^(\w+):([^\s\n]+)') + line_match = lang_path_on_line.match(first_line) + if line_match: + language = line_match.group(1) + path_hint = line_match.group(2) + content = rest + blocks.append({ + "language": language, + "path_hint": path_hint, + "content": content.strip(), + "inferred_path": path_hint.split('/')[-1] if '/' in path_hint else path_hint + }) + continue + + # Check pattern 2: // FILE: or # FILE: + # Only search the first non-empty line of rest + rest_lines = rest.split('\n') + first_non_lang_line = "" + first_line_idx = 0 + for idx, line in enumerate(rest_lines): + if line.strip(): + first_non_lang_line = line + first_line_idx = idx + break + + marker_match = file_marker_pattern.search(first_non_lang_line) if first_non_lang_line else None + if marker_match: + path_hint = marker_match.group(1).strip() + # Remove the marker line from content + content = '\n'.join(rest_lines[first_line_idx + 1:]) + blocks.append({ + "language": language, + "path_hint": path_hint, + "content": content.strip(), + "inferred_path": path_hint.split('/')[-1] if '/' in path_hint else path_hint + }) + continue + + # Check pattern 3: # filename.py + # Only search the first non-empty line of rest + filename_match = filename_pattern.search(first_non_lang_line) if first_non_lang_line else None + if filename_match: + filename = filename_match.group(1) + path_hint = filename + # Remove the filename line from content + content = '\n'.join(rest_lines[first_line_idx + 1:]) + blocks.append({ + "language": language, + "path_hint": path_hint, + "content": content.strip(), + "inferred_path": filename + }) + + return blocks + + +def parse_and_write_files(response_text: str, output_dir: str) -> list[tuple]: + """ + Parse code blocks and write files to output_dir. + + Returns list of (relative_path, byte_count) tuples. + """ + written = [] + output_path = Path(output_dir) + + blocks = parse_code_blocks(response_text) + + for block in blocks: + path_hint = block.get("path_hint", "") + content = block.get("content", "") + + if not path_hint or not content: + continue + + # Sanitize path + try: + dest = sanitize_target_path(path_hint, output_dir) + dest.parent.mkdir(parents=True, exist_ok=True) + encoded = content.strip().encode("utf-8", errors="replace") + dest.write_bytes(encoded) + written.append((str(Path(path_hint)), len(encoded))) + except ValueError as e: + print(f"WARNING: Skipping unsafe path: {e}", file=sys.stderr) + except Exception as e: + print(f"WARNING: Failed to write {path_hint}: {e}", file=sys.stderr) + + return written + + +# ============================================================================= +# File Application +# ============================================================================= + +def apply_file_change(file_path: str, content: str, dry_run: bool = True) -> bool: + """ + Apply a file change to the actual target directory. + + Args: + file_path: Relative path within target + content: File content to write + dry_run: If True, just preview; if False, actually write + """ + if dry_run: + print(f"[PREVIEW] Would write {file_path} ({len(content)} chars)") + return True + + try: + Path(file_path).parent.mkdir(parents=True, exist_ok=True) + Path(file_path).write_text(content) + print(f"[WROTE] {file_path}", file=sys.stderr) + return True + except Exception as e: + print(f"[ERROR] Failed to write {file_path}: {e}", file=sys.stderr) + return False + + +def apply_changes_from_response(state: AgentState, response: str) -> list[str]: + """ + Parse response for code blocks and apply to target directory. + + Returns list of files that were (or would be) written. + """ + blocks = parse_code_blocks(response) + applied = [] + + if not blocks: + # Check if there's any content that looks like code without annotations + if "```" in response: + print("[WARNING] Code blocks found but no file annotations - cannot apply", file=sys.stderr) + return applied + + for block in blocks: + path_hint = block.get("path_hint", "") + content = block.get("content", "") + + if not path_hint: + # Try to infer from language + lang = block.get("language", "") + if lang: + ext = {"python": "py", "javascript": "js", "typescript": "ts", "go": "go"}.get(lang.lower(), lang.lower()) + path_hint = f"generated.{ext}" + print(f"[WARNING] No path for {lang} block, using {path_hint}", file=sys.stderr) + else: + continue + + # Sanitize and apply in target directory + try: + # Determine if this is a new file by checking existence in target directory + # First resolve path_hint relative to state.target to check if file exists + temp_target_path = sanitize_target_path(path_hint, state.target) + is_new_file = not temp_target_path.exists() + + # Use output_dir only for new files, otherwise use target + base_root = state.output_dir if (state.output_dir and is_new_file) else state.target + target_path = sanitize_target_path(path_hint, base_root) + + if state.apply: + success = apply_file_change(str(target_path), content, dry_run=False) + if success: + applied.append(str(target_path)) + else: + apply_file_change(str(target_path), content, dry_run=True) + applied.append(str(target_path)) + except ValueError as e: + print(f"[ERROR] Skipping unsafe path: {e}", file=sys.stderr) + continue + + return applied + + +# ============================================================================= +# Verification +# ============================================================================= + +def verify_changes(state: AgentState) -> tuple[bool, str]: + """ + Run verification command. + + Returns (success, output). + """ + if not state.verify_cmd: + return True, "No verification command" + + try: + result = subprocess.run( + state.verify_cmd, + shell=True, + capture_output=True, + text=True, + timeout=120, + cwd=state.target, + ) + success = result.returncode == 0 + output = result.stdout + result.stderr + return success, output + except subprocess.TimeoutExpired: + return False, "Verification timed out (>120s)" + except Exception as e: + return False, str(e) + + +# ============================================================================= +# Agent Loop +# ============================================================================= + +def build_agent_prompt(state: AgentState) -> str: + """Build the prompt for Grok based on current state.""" + task = state.task + + if state.iteration == 1: + # First iteration: discover and plan + files = discover_files(state.target) + state.file_context = read_files(files) if files else "" + + file_count = len(files) if files else 0 + + # Get just the first 30K chars to avoid overwhelming Grok + context_preview = state.file_context[:30000] if state.file_context else "" + + return f"""You are an autonomous coding agent. Your task: {task} + +Target: {state.target} ({file_count} files) + +{context_preview} + +CRITICAL FORMAT - Write files using this EXACT format: +```python:cli.py +# full content here +``` +or: +```python +// FILE: cli.py +# full content here +``` + +Do NOT use just `# filename.py`. Do NOT use no annotation. +""" + else: + # Subsequent iterations: refine based on previous + prompt = f"""Continue working on: {task} + +Previous iteration ({state.iteration - 1}) response: +{state.last_response[:15000]} + +Iteration {state.iteration}/{state.max_iterations}""" + + # Include verification output if available + if state.last_verification_output: + prompt += f""" + +Verifier output / failure: +{state.last_verification_output[:5000]} +""" + + prompt += """ + +If the previous changes had errors or could be improved, refine them. Otherwise, continue with the next set of changes. + +Use the same annotation format: +```python:/path/to/file.py +# content +``` +""" + return prompt + + +def run_iteration(state: AgentState) -> bool: + """ + Run a single agent iteration. + + Returns True if agent should stop (done or success). + """ + state.iteration += 1 + print(f"\n=== Iteration {state.iteration}/{state.max_iterations} ===", file=sys.stderr) + + # Build prompt + prompt = build_agent_prompt(state) + + # Call Grok + print("Calling Grok 4.20 (refactor mode)...", file=sys.stderr) + try: + response = call_grok( + prompt=prompt, + mode="refactor", + timeout=180, + ) + state.last_response = response + except SystemExit as se: + error_msg = f"Grok call failed with exit code {se.code}" + state.errors.append(error_msg) + print(f"[ERROR] {error_msg}", file=sys.stderr) + return False + except Exception as e: + state.errors.append(f"Grok call failed: {e}") + print(f"[ERROR] Grok call failed: {e}", file=sys.stderr) + return False + + # Parse and apply changes + if state.apply: + applied = apply_changes_from_response(state, response) + state.changes.extend(applied) + print(f"Applied {len(applied)} files", file=sys.stderr) + else: + # Preview mode + blocks = parse_code_blocks(response) + print(f"[PREVIEW] Would modify {len(blocks)} blocks", file=sys.stderr) + + # Verify if command provided + verification_succeeded = True # Default to True if no verification + if state.verify_cmd and state.apply: + success, output = verify_changes(state) + verification_succeeded = success + state.last_verification_output = output # Store for next iteration + if success: + print("[VERIFY] Passed", file=sys.stderr) + else: + state.errors.append(f"Verification failed: {output[:500]}") + print(f"[VERIFY] Failed: {output[:500]}", file=sys.stderr) + # Continue anyway - Grok can fix in next iteration + + # Check if done - but only if verification passed + response_lower = response.lower() + done_markers = ["done", "complete", "finished", "all changes made", "successfully"] + if verification_succeeded and any(marker in response_lower for marker in done_markers): + return True + + # Check if no changes were made + blocks = parse_code_blocks(response) + if not blocks and state.iteration > 1: + return True + + return False + + +def run_agent_loop(state: AgentState) -> AgentState: + """Run the full agent loop until completion or max iterations.""" + print("[AGENT] Starting agent loop", file=sys.stderr) + print(f"[AGENT] Task: {state.task}", file=sys.stderr) + print(f"[AGENT] Target: {state.target}", file=sys.stderr) + print(f"[AGENT] Apply mode: {state.apply}", file=sys.stderr) + + # Check target exists + if not Path(state.target).exists(): + state.status = AgentStatus.NO_FILES + state.errors.append(f"Target does not exist: {state.target}") + return state + + # Check files exist + files = discover_files(state.target) + if not files: + state.status = AgentStatus.NO_FILES + state.errors.append(f"No code files found in: {state.target}") + return state + + while state.iteration < state.max_iterations: + done = run_iteration(state) + if done: + state.status = AgentStatus.SUCCESS + break + else: + state.status = AgentStatus.MAX_ITERATIONS + + return state + + +# ============================================================================= +# CLI Entry Point +# ============================================================================= + +def main(): + parser = argparse.ArgumentParser( + description="Grok Swarm Agent - Autonomous agent powered by Grok 4.20", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=""" +Examples: + # Preview mode (default) - shows what would change + python3 grok_agent.py --task "refactor auth module" --target ./src/auth + + # Apply changes + python3 grok_agent.py --task "refactor auth module" --target ./src/auth --apply + + # With verification + python3 grok_agent.py --task "add tests" --target ./src --apply --verify-cmd "pytest" + + # OpenClaw platform + python3 grok_agent.py --platform openclaw --task "analyze security" --target . + """ + ) + parser.add_argument("--platform", choices=["claude", "openclaw"], default="claude", + help="Platform (default: claude)") + parser.add_argument("--target", default=".", + help="Target directory or file (default: .)") + parser.add_argument("--apply", action="store_true", + help="Actually apply changes (default: preview mode)") + parser.add_argument("--max-iterations", type=int, default=5, + help="Max agent iterations (default: 5)") + parser.add_argument("--verify-cmd", + help="Command to run for verification (e.g., pytest)") + parser.add_argument("--output-dir", + help="Output directory for new files (existing files remain in target)") + parser.add_argument("--task", "-t", required=True, dest="task", + help="Natural language task instruction") + + args = parser.parse_args() + + # Create state + state = AgentState( + task=args.task, + target=args.target, + platform=Platform(args.platform), + apply=args.apply, + max_iterations=args.max_iterations, + verify_cmd=args.verify_cmd, + output_dir=args.output_dir, + ) + + # Run agent + result = run_agent_loop(state) + + # Output summary + print("\n" + "=" * 60, file=sys.stderr) + print("GROK-SWARM-AGENT SUMMARY", file=sys.stderr) + print("=" * 60, file=sys.stderr) + print(f"Status: {result.status.value}", file=sys.stderr) + print(f"Iterations: {result.iteration}/{result.max_iterations}", file=sys.stderr) + print(f"Files: {len(result.changes)} changed", file=sys.stderr) + + if result.changes: + print("\nChanged files:", file=sys.stderr) + for f in result.changes: + print(f" - {f}", file=sys.stderr) + + if result.errors: + print(f"\nErrors ({len(result.errors)}):", file=sys.stderr) + for err in result.errors: + print(f" - {err[:200]}", file=sys.stderr) + + print("=" * 60, file=sys.stderr) + + if result.apply: + # Show human-readable summary to stdout + if result.status == AgentStatus.SUCCESS: + print(f"✓ Completed in {result.iteration} iteration(s)") + print(f"✓ Changed {len(result.changes)} file(s)") + elif result.status == AgentStatus.MAX_ITERATIONS: + print(f"⚠ Max iterations ({result.max_iterations}) reached") + print(f" Changed {len(result.changes)} file(s) - may need more work") + elif result.status == AgentStatus.NO_FILES: + print(f"✗ No files found in target: {result.target}") + else: + print(f"✗ Failed: {result.errors[0] if result.errors else 'Unknown error'}") + else: + # Preview mode + print("\n[PREVIEW MODE] Re-run with --apply to actually write changes") + + sys.exit(0 if result.status == AgentStatus.SUCCESS else 1) + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/src/bridge/grok_bridge.py b/src/bridge/grok_bridge.py index 173b4b1..c070091 100644 --- a/src/bridge/grok_bridge.py +++ b/src/bridge/grok_bridge.py @@ -19,6 +19,10 @@ import time from pathlib import Path +# Import shared patterns +sys.path.insert(0, str(Path(__file__).parent.parent / "shared")) +from patterns import get_filename_pattern_string + try: from openai import OpenAI except ImportError: @@ -170,25 +174,43 @@ def _safe_dest(output_path, file_path): def parse_and_write_files(response_text, output_dir): """ Scan response for fenced code blocks with filename annotations and write to disk. - - Supports patterns: - ```lang:path/to/file ... ``` - ```lang - // FILE: path/to/file - ... - ``` - + + Supports multiple annotation formats: + ```python:/path/to/file.py ... ``` (lang:/path format) + ```python + // FILE: /path/to/file.py + ... content ... + ``` (C-style comment marker) + ```python + # FILE: /path/to/file.py + ... content ... + ``` (Python-style comment marker) + ```python + # filename.py + ... content ... + ``` (Just filename as comment - common Grok output) + Returns list of (relative_path, byte_count) tuples written, where byte_count is the number of UTF-8 bytes written. """ written = [] output_path = Path(output_dir) - - # Pattern for lang:path at start of block (language tag contains path) - lang_path_pattern = re.compile(r'^(\w+):([^\s\n]+)\n', re.MULTILINE) - # Pattern for // FILE: or # FILE: markers - file_marker_pattern = re.compile(r'^\s*(?://|#)\s*FILE:\s*(.+?)\s*$', re.MULTILINE) - + + # Pattern 1: lang:/path/to/file (language tag contains path) + lang_path_pattern = re.compile(r'^(\w+):(/[^\s\n]+(?:/[^\s\n]+)*)\n', re.MULTILINE) + + # Pattern 2: // FILE: /path or # FILE: /path + file_marker_pattern = re.compile( + r'^\s*(?:(?://|#)\s*)FILE:\s*(.+?)\s*$', + re.MULTILINE + ) + + # Pattern 3: # filename.py (just filename as first line - common Grok output) + filename_pattern = re.compile( + get_filename_pattern_string(), + re.MULTILINE + ) + def _write_file(file_path, content): """Validate path, write content, and record result. Returns True on success.""" try: @@ -206,23 +228,44 @@ def _write_file(file_path, content): # Even indices are fence markers or text between fences; skip them. # Odd indices are the actual code block contents. parts = re.split(r'```', response_text) - + for i, part in enumerate(parts): if i % 2 == 0: # Skip even-indexed parts (fences/text between fences) continue - - # Check for lang:path at start (language tag contains the path) + + # Check for lang:/path at start (language tag contains the path) lang_match = lang_path_pattern.match(part) if lang_match: _write_file(lang_match.group(2), part[lang_match.end():]) continue - + # Check for // FILE: or # FILE: marker within the block marker_match = file_marker_pattern.search(part) if marker_match: - _write_file(marker_match.group(1).strip(), part[marker_match.end():]) - + path = marker_match.group(1).strip() + # Remove the marker line from content + marker_end = part.find('\n', marker_match.start()) + if marker_end != -1: + content = part[marker_end + 1:] + else: + content = "" + _write_file(path, content) + continue + + # Check for # filename.py pattern (common Grok output) + filename_match = filename_pattern.match(part) + if filename_match: + filename = filename_match.group(1) + # Remove the filename line from content + filename_end = part.find('\n', filename_match.end()) + if filename_end != -1: + content = part[filename_end + 1:] + else: + content = "" + _write_file(filename, content) + continue + return written def call_grok(prompt, mode="reason", context="", system_override=None, tools=None, timeout=120): diff --git a/src/plugin/grok_agent_plugin.ts b/src/plugin/grok_agent_plugin.ts new file mode 100644 index 0000000..a3ec699 --- /dev/null +++ b/src/plugin/grok_agent_plugin.ts @@ -0,0 +1,152 @@ +/** + * grok-swarm-agent plugin — registers `grok_swarm_agent` as an autonomous agent tool. + * + * Bridges to xAI Grok 4.20 Multi-Agent Beta via OpenRouter with an iterative agent loop. + * + * Features: + * - Automatic file discovery + * - Iterative refinement + * - Verification commands + * - Cross-platform (Claude Code + OpenClaw) + */ + +import { spawn } from "child_process"; +import { existsSync } from "fs"; +import { join, dirname } from "path"; +import { fileURLToPath } from "url"; +import { Type } from "@sinclair/typebox"; + +const PLUGIN_ROOT = join(dirname(fileURLToPath(import.meta.url)), "..", "..", ".."); +const DEFAULT_AGENT = join(PLUGIN_ROOT, "src", "agent", "grok_agent.py"); + +const GrokAgentSchema = Type.Object({ + task: Type.String({ description: "Natural language task instruction" }), + target: Type.Optional( + Type.String({ description: "Target directory or file (default: .)" }), + ), + apply: Type.Optional( + Type.Boolean({ description: "Actually apply changes (default: preview mode)" }), + ), + max_iterations: Type.Optional( + Type.Number({ description: "Max agent iterations (default: 5)" }), + ), + verify_cmd: Type.Optional( + Type.String({ description: "Command to run for verification (e.g., pytest)" }), + ), +}); + +export default function (api: any) { + api.registerTool( + { + name: "grok_swarm_agent", + label: "Grok Swarm Agent", + description: + "Spawn an autonomous agent powered by Grok 4.20 Multi-Agent Beta. " + + "The agent iteratively discovers files, calls Grok 4.20 for modifications, " + + "applies changes, and verifies results. " + + "Use for complex refactoring, test generation, or multi-file modifications. " + + "Use --apply to actually write files (default is preview mode).", + parameters: GrokAgentSchema, + async execute(_toolCallId: string, params: any) { + const json = (payload: unknown) => ({ + content: [ + { type: "text" as const, text: typeof payload === "string" ? payload : JSON.stringify(payload, null, 2) }, + ], + details: payload, + }); + + try { + const agentScript = api.config?.agentScript || DEFAULT_AGENT; + + // Validate agent script exists + if (!existsSync(agentScript)) { + return json({ + error: `Agent script not found: ${agentScript}. Ensure grok-swarm plugin is properly installed.`, + }); + } + + const maxIterations = params.max_iterations || 5; + const timeout = Math.max(maxIterations * 200, 600); // At least 10min, more for higher iterations + + // Build args + const args = [ + agentScript, + "--platform", "openclaw", + "--target", params.target || ".", + "--max-iterations", String(maxIterations), + ]; + + if (params.apply) { + args.push("--apply"); + } + + if (params.verify_cmd) { + args.push("--verify-cmd", params.verify_cmd); + } + + args.push("--", params.task); + + // Spawn agent with timeout enforcement + return new Promise((resolve) => { + const child = spawn("python3", args, { + stdio: ["ignore", "pipe", "pipe"], + env: { ...process.env }, + }); + + let stdout = ""; + let stderr = ""; + let timedOut = false; + + const timer = setTimeout(() => { + timedOut = true; + child.kill("SIGTERM"); + setTimeout(() => child.kill("SIGKILL"), 5000); + }, timeout * 1000); + + child.stdout.on("data", (data: Buffer) => { + stdout += data.toString(); + }); + + child.stderr.on("data", (data: Buffer) => { + stderr += data.toString(); + }); + + child.on("close", (code: number | null) => { + clearTimeout(timer); + + if (timedOut) { + resolve( + json({ + error: `Agent timed out after ${timeout}s`, + stderr: stderr.slice(-500), + }), + ); + return; + } + + // Parse the output - stdout has the summary, stderr has debug info + const summary = stdout.trim() || "(no output)"; + const debug = stderr.trim(); + + resolve(json({ + result: summary, + debug: debug.slice(-1000), + exitCode: code, + })); + }); + + child.on("error", (err: Error) => { + clearTimeout(timer); + resolve(json({ error: `Failed to spawn agent: ${err.message}` })); + }); + }); + } catch (err) { + return json({ + error: err instanceof Error ? err.message : String(err), + }); + } + }, + }, + { optional: true }, + ); +} \ No newline at end of file diff --git a/src/plugin/index.ts b/src/plugin/index.ts index 40d77fa..1ca1b99 100644 --- a/src/plugin/index.ts +++ b/src/plugin/index.ts @@ -10,10 +10,11 @@ import { spawn } from "child_process"; import { existsSync } from "fs"; import { join, dirname } from "path"; +import { fileURLToPath } from "url"; import { Type } from "@sinclair/typebox"; const SKILL_DIR = join( - dirname(new URL(import.meta.url).pathname), + dirname(fileURLToPath(import.meta.url)), "..", "..", "..", diff --git a/src/shared/__init__.py b/src/shared/__init__.py new file mode 100644 index 0000000..ad5da11 --- /dev/null +++ b/src/shared/__init__.py @@ -0,0 +1 @@ +# shared package diff --git a/src/shared/patterns.py b/src/shared/patterns.py new file mode 100644 index 0000000..9fbdc47 --- /dev/null +++ b/src/shared/patterns.py @@ -0,0 +1,28 @@ +""" +Shared regex patterns used by both grok_bridge.py and grok_agent.py. + +Centralizing these patterns ensures consistent parsing across all modules. +""" + +# Canonical list of supported file extensions for filename detection. +SUPPORTED_EXTENSIONS = ( + "py|js|ts|jsx|tsx|go|rs|java|c|cpp|h|hpp|cs|rb|php|swift|kt|scala|sh|bash" +) + +# Regex pattern string that matches a comment-style filename header, e.g.: +# # filename.py +_FILENAME_PATTERN_STRING = ( + r"^\s*#\s*([a-zA-Z_][a-zA-Z0-9_]*\.(?:" + + SUPPORTED_EXTENSIONS + + r"))\s*$" +) + + +def get_filename_pattern_string() -> str: + """Return the canonical regex string for matching comment-style filename headers. + + Both grok_bridge.py and grok_agent.py compile this string (with re.MULTILINE) + so that any change to the supported extension list is automatically reflected + in every consumer. + """ + return _FILENAME_PATTERN_STRING