From 7a1e0bce21ad9346f26caec5cf71b2bc6a3bf46c Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 21 Mar 2026 04:30:03 +0000 Subject: [PATCH 01/11] fix(#14): handle absolute paths in _safe_dest() and add fallback save - _safe_dest() now rebases absolute paths (e.g. /home/user/.sandbox/foo.py) to just the filename under the output directory instead of raising an error, printing a warning so the user knows the rebase occurred - When --write-files produces zero written files, the full response is saved to /grok-response.txt and the process exits with code 1 so callers can detect the failure rather than silently succeeding Closes #14 https://claude.ai/code/session_01No9S6TZbfTgocHwHYWwxRN --- src/bridge/grok_bridge.py | 24 ++++++++++++++++++++---- 1 file changed, 20 insertions(+), 4 deletions(-) diff --git a/src/bridge/grok_bridge.py b/src/bridge/grok_bridge.py index 0883391..e5cfda6 100644 --- a/src/bridge/grok_bridge.py +++ b/src/bridge/grok_bridge.py @@ -167,11 +167,20 @@ def _safe_dest(output_path, file_path): """ Resolve ``file_path`` relative to ``output_path`` and verify the result stays inside ``output_path``. Returns the resolved Path or raises - ValueError for unsafe paths (absolute, containing ``..``, etc.). + ValueError for unsafe paths (containing ``..``, etc.). + + Absolute paths are rebased under ``output_path`` using only the filename + component (e.g. ``/home/user/.sandbox/task.py`` → ``/task.py``), + and a warning is printed to stderr. """ raw = Path(file_path) if raw.is_absolute(): - raise ValueError(f"Absolute paths are not allowed: {file_path!r}") + rebased = Path(raw.name) + print( + f"WARNING: Absolute path rebased to output dir: {file_path!r} → {rebased!r}", + file=sys.stderr, + ) + raw = rebased if ".." in raw.parts: raise ValueError(f"Path traversal not allowed: {file_path!r}") dest = (output_path / raw).resolve() @@ -418,11 +427,18 @@ def main(): print(f" {rel_path} ({byte_count:,} bytes)") print(f"Total: {total_bytes:,} bytes") else: + # Save full response as a fallback so no output is lost + fallback_dir = Path(args.output_dir) + fallback_dir.mkdir(parents=True, exist_ok=True) + fallback_path = fallback_dir / "grok-response.txt" + fallback_path.write_text(result, encoding="utf-8") print( - "No annotated files found in model response to write to disk.\n" - "Re-run without --write-files to see the full response.", + f"ERROR: No annotated files found in model response.\n" + f"Full response saved to: {fallback_path}\n" + f"Tip: ask Grok to annotate code blocks with ```lang:path/to/file or # FILE: path/to/file", file=sys.stderr, ) + sys.exit(1) elif not args.output: print(result) From 3b7e09c53f4f76a1fc21ee89444a84e41da4a76b Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 21 Mar 2026 04:30:36 +0000 Subject: [PATCH 02/11] feat(#12): add configurable grounding system prompt Prepends a grounding identity prompt to every mode-specific system prompt so Grok always operates as a focused coding assistant. - DEFAULT_GROUNDING_PROMPT constant establishes role/expectations - load_grounding_prompt() reads ~/.config/grok-swarm/system-prompt.txt for user customisation, falling back to the default - call_grok() now prepends the grounding prompt before mode content - orchestrate mode is unaffected: system_override keeps full control Closes #12 https://claude.ai/code/session_01No9S6TZbfTgocHwHYWwxRN --- src/bridge/grok_bridge.py | 36 ++++++++++++++++++++++++++++++++++-- 1 file changed, 34 insertions(+), 2 deletions(-) diff --git a/src/bridge/grok_bridge.py b/src/bridge/grok_bridge.py index e5cfda6..a560b25 100644 --- a/src/bridge/grok_bridge.py +++ b/src/bridge/grok_bridge.py @@ -45,6 +45,35 @@ "--thinking high", ] +# Default grounding prompt prepended to every mode-specific system prompt. +# Users can override this by creating ~/.config/grok-swarm/system-prompt.txt +DEFAULT_GROUNDING_PROMPT = ( + "You are Grok, a specialized agentic coding assistant powered by xAI's multi-agent swarm. " + "Your primary role is to help software engineers write, analyze, refactor, and debug code. " + "You have access to a large context window and collaborate with multiple parallel agents to " + "produce thorough, well-reasoned results. Always produce production-quality output: " + "correct, readable, and idiomatic for the target language. " + "When modifying files, annotate every code block with its file path so changes can be " + "applied automatically." +) + + +def load_grounding_prompt(): + """ + Return the user's custom grounding prompt from + ~/.config/grok-swarm/system-prompt.txt, or DEFAULT_GROUNDING_PROMPT. + """ + custom = Path.home() / ".config" / "grok-swarm" / "system-prompt.txt" + if custom.exists(): + try: + text = custom.read_text(encoding="utf-8").strip() + if text: + return text + except OSError: + pass + return DEFAULT_GROUNDING_PROMPT + + # Mode-specific system prompts MODE_PROMPTS = { "refactor": ( @@ -266,12 +295,15 @@ def call_grok(prompt, mode="reason", context="", system_override=None, tools=Non # Resolve system prompt if system_override: + # orchestrate mode: user owns the full system prompt, skip grounding system_content = system_override else: - system_content = MODE_PROMPTS.get(mode) - if system_content is None: + mode_prompt = MODE_PROMPTS.get(mode) + if mode_prompt is None: print(f"ERROR: Mode '{mode}' requires --system flag", file=sys.stderr) sys.exit(1) + grounding = load_grounding_prompt() + system_content = grounding + "\n\n" + mode_prompt # Append context to system prompt if context: From 626e0bdb6cf5d442624f1ca9f9bc19a2fa25f3c4 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 21 Mar 2026 04:31:42 +0000 Subject: [PATCH 03/11] feat(#11): add token/cost usage tracking and stats subcommand - New src/bridge/usage_tracker.py: records every API call to ~/.config/grok-swarm/usage.json (newline-delimited JSON) using stdlib only; aggregates calls, tokens, cost, and time; formats a human-readable table - grok_bridge.py: imports usage_tracker and calls record_usage() after each successful API call (failures are silently swallowed so tracking never aborts the main flow) - cli.py: new `grok-swarm stats [--days N]` subcommand prints the aggregated stats table Pricing based on OpenRouter's published rates for x-ai/grok-4.20-multi-agent-beta ($5/M prompt, $15/M completion). Closes #11 https://claude.ai/code/session_01No9S6TZbfTgocHwHYWwxRN --- src/bridge/cli.py | 17 +++++ src/bridge/grok_bridge.py | 17 +++++ src/bridge/usage_tracker.py | 142 ++++++++++++++++++++++++++++++++++++ 3 files changed, 176 insertions(+) create mode 100644 src/bridge/usage_tracker.py diff --git a/src/bridge/cli.py b/src/bridge/cli.py index 628dbb6..eda5b74 100644 --- a/src/bridge/cli.py +++ b/src/bridge/cli.py @@ -23,6 +23,7 @@ sys.path.insert(0, str(current)) from grok_bridge import call_grok, read_files, MODE_PROMPTS +from usage_tracker import get_stats, format_stats_report def check_morph_available(): @@ -131,7 +132,23 @@ def parse_and_write(result_text, output_dir, dry_run=True): return format_summary(result, output_dir) +def _handle_stats(argv): + """Handle `grok-swarm stats [--days N]` subcommand.""" + import argparse as _ap + p = _ap.ArgumentParser(prog="grok-swarm stats", description="Show token/cost usage statistics") + p.add_argument("--days", "-d", type=int, default=None, + metavar="N", help="Limit to last N days (default: all time)") + args = p.parse_args(argv) + stats = get_stats(since_days=args.days) + print(format_stats_report(stats, since_days=args.days)) + + def main(): + # Dispatch stats subcommand before full argument parsing + if len(sys.argv) > 1 and sys.argv[1] == "stats": + _handle_stats(sys.argv[2:]) + return + parser = argparse.ArgumentParser( description="Grok Swarm — Multi-agent CLI powered by Grok 4.20", formatter_class=argparse.RawDescriptionHelpFormatter, diff --git a/src/bridge/grok_bridge.py b/src/bridge/grok_bridge.py index a560b25..9d65c44 100644 --- a/src/bridge/grok_bridge.py +++ b/src/bridge/grok_bridge.py @@ -25,6 +25,11 @@ print("ERROR: openai package required. Install: pip3 install openai", file=sys.stderr) sys.exit(1) +try: + from usage_tracker import record_usage as _record_usage +except ImportError: + _record_usage = None + OPENROUTER_BASE = "https://openrouter.ai/api/v1" MODEL_ID = "x-ai/grok-4.20-multi-agent-beta" @@ -355,6 +360,18 @@ def call_grok(prompt, mode="reason", context="", system_override=None, tools=Non print(f"USAGE: mode={mode} thinking={thinking} agents={agent_count} " f"prompt={u.prompt_tokens} completion={u.completion_tokens} " f"total={u.total_tokens} time={elapsed:.1f}s", file=sys.stderr) + if _record_usage is not None: + try: + _record_usage( + mode=mode, + thinking=thinking, + prompt_tokens=u.prompt_tokens, + completion_tokens=u.completion_tokens, + total_tokens=u.total_tokens, + elapsed_secs=elapsed, + ) + except Exception: + pass # Handle content filtering if choice.finish_reason == "content_filter": diff --git a/src/bridge/usage_tracker.py b/src/bridge/usage_tracker.py new file mode 100644 index 0000000..f8e45e0 --- /dev/null +++ b/src/bridge/usage_tracker.py @@ -0,0 +1,142 @@ +#!/usr/bin/env python3 +""" +usage_tracker.py — Persistent token/cost tracking for grok-swarm requests. + +Logs each API call to ~/.config/grok-swarm/usage.json (newline-delimited JSON). +Provides aggregation and a human-readable stats report. + +OpenRouter pricing for x-ai/grok-4.20-multi-agent-beta (as of 2026-03): + Prompt tokens: $5.00 / 1M tokens + Completion tokens: $15.00 / 1M tokens +""" + +import json +import sys +from datetime import datetime, timezone, timedelta +from pathlib import Path + +USAGE_FILE = Path.home() / ".config" / "grok-swarm" / "usage.json" + +# OpenRouter pricing (USD per 1M tokens) +PROMPT_PRICE_PER_M = 5.00 +COMPLETION_PRICE_PER_M = 15.00 + + +def record_usage(mode, thinking, prompt_tokens, completion_tokens, total_tokens, elapsed_secs): + """ + Append one usage record to the log file. + Never raises — failures are printed to stderr and silently ignored. + """ + record = { + "ts": datetime.now(timezone.utc).isoformat(), + "mode": mode, + "thinking": thinking, + "prompt_tokens": prompt_tokens, + "completion_tokens": completion_tokens, + "total_tokens": total_tokens, + "elapsed_secs": round(elapsed_secs, 2), + "cost_usd": round( + (prompt_tokens / 1_000_000) * PROMPT_PRICE_PER_M + + (completion_tokens / 1_000_000) * COMPLETION_PRICE_PER_M, + 6, + ), + } + try: + USAGE_FILE.parent.mkdir(parents=True, exist_ok=True) + with USAGE_FILE.open("a", encoding="utf-8") as f: + f.write(json.dumps(record) + "\n") + except OSError as exc: + print(f"WARNING: Could not write usage log: {exc}", file=sys.stderr) + + +def get_stats(since_days=None): + """ + Read the usage log and return aggregated stats dict. + + Args: + since_days: If set, only include records from the last N days. + + Returns dict with keys: + calls, prompt_tokens, completion_tokens, total_tokens, + cost_usd, elapsed_secs, by_mode (dict of mode -> call count) + """ + if not USAGE_FILE.exists(): + return None + + cutoff = None + if since_days is not None: + cutoff = datetime.now(timezone.utc) - timedelta(days=since_days) + + stats = { + "calls": 0, + "prompt_tokens": 0, + "completion_tokens": 0, + "total_tokens": 0, + "cost_usd": 0.0, + "elapsed_secs": 0.0, + "by_mode": {}, + } + + try: + with USAGE_FILE.open(encoding="utf-8") as f: + for line in f: + line = line.strip() + if not line: + continue + try: + rec = json.loads(line) + except json.JSONDecodeError: + continue + + if cutoff: + try: + rec_ts = datetime.fromisoformat(rec["ts"]) + if rec_ts < cutoff: + continue + except (KeyError, ValueError): + pass + + stats["calls"] += 1 + stats["prompt_tokens"] += rec.get("prompt_tokens", 0) + stats["completion_tokens"] += rec.get("completion_tokens", 0) + stats["total_tokens"] += rec.get("total_tokens", 0) + stats["cost_usd"] += rec.get("cost_usd", 0.0) + stats["elapsed_secs"] += rec.get("elapsed_secs", 0.0) + + mode = rec.get("mode", "unknown") + stats["by_mode"][mode] = stats["by_mode"].get(mode, 0) + 1 + + except OSError as exc: + print(f"WARNING: Could not read usage log: {exc}", file=sys.stderr) + return None + + return stats + + +def format_stats_report(stats, since_days=None): + """Return a human-readable stats report string.""" + if stats is None or stats["calls"] == 0: + return "No usage data recorded yet. Run a grok-swarm command to start tracking." + + period = f"last {since_days} day(s)" if since_days else "all time" + lines = [ + "", + "=" * 55, + f" grok-swarm usage stats ({period})", + "=" * 55, + f" Requests: {stats['calls']:>10,}", + f" Prompt tokens: {stats['prompt_tokens']:>10,}", + f" Completion tokens: {stats['completion_tokens']:>10,}", + f" Total tokens: {stats['total_tokens']:>10,}", + f" Estimated cost: ${stats['cost_usd']:>10.4f}", + f" Total time: {stats['elapsed_secs']:>9.1f}s", + ] + if stats["by_mode"]: + lines.append("") + lines.append(" By mode:") + for mode, count in sorted(stats["by_mode"].items(), key=lambda x: -x[1]): + lines.append(f" {mode:<16} {count:>6,} call(s)") + lines.append("=" * 55) + lines.append(f" Log file: {USAGE_FILE}") + lines.append("=" * 55) + return "\n".join(lines) From 68ec50dd1c3ec09d85dc04d88276e0ac7058712e Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 21 Mar 2026 04:33:42 +0000 Subject: [PATCH 04/11] feat(#16): land autonomous grok-swarm-agent with bug fixes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Brings in the grok-swarm-agent feature from the open PR on feature/grok-swarm-agent, fixing three regressions before landing: Bug fixes vs. the open PR: - Created src/shared/patterns.py (get_filename_pattern_string()) which was imported but never committed, causing ImportError at startup - Fixed lang_path_pattern in grok_agent.py to match relative paths (python:cli.py) in addition to absolute paths — the PR's narrowed pattern ^(\w+):(/...) silently dropped relative-path annotations - src/bridge/grok_bridge.py now imports shared patterns gracefully (try/except) so the bridge still works without the shared module New files: - src/shared/__init__.py + patterns.py — shared regex utilities - src/agent/__init__.py + grok_agent.py — autonomous iterative agent loop (file discovery → Grok call → parse/apply → verify → iterate up to 5x) - src/plugin/grok_agent_plugin.ts — OpenClaw/Claude Code tool registration - platforms/claude/skills/grok-swarm-agent/SKILL.md — skill documentation - platforms/claude/commands/grok-swarm-agent.sh — slash command Updated: - src/plugin/index.ts — imports and registers the agent tool Closes #16 https://claude.ai/code/session_01No9S6TZbfTgocHwHYWwxRN --- platforms/claude/commands/grok-swarm-agent.sh | 81 +++ .../claude/skills/grok-swarm-agent/SKILL.md | 63 ++ src/agent/__init__.py | 1 + src/agent/grok_agent.py | 677 ++++++++++++++++++ src/bridge/grok_bridge.py | 33 +- src/plugin/grok_agent_plugin.ts | 152 ++++ src/plugin/index.ts | 4 + src/shared/__init__.py | 1 + src/shared/patterns.py | 28 + 9 files changed, 1037 insertions(+), 3 deletions(-) create mode 100755 platforms/claude/commands/grok-swarm-agent.sh create mode 100644 platforms/claude/skills/grok-swarm-agent/SKILL.md create mode 100644 src/agent/__init__.py create mode 100644 src/agent/grok_agent.py create mode 100644 src/plugin/grok_agent_plugin.ts create mode 100644 src/shared/__init__.py create mode 100644 src/shared/patterns.py diff --git a/platforms/claude/commands/grok-swarm-agent.sh b/platforms/claude/commands/grok-swarm-agent.sh new file mode 100755 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/src/agent/__init__.py b/src/agent/__init__.py new file mode 100644 index 0000000..270944f --- /dev/null +++ b/src/agent/__init__.py @@ -0,0 +1 @@ +# grok-swarm autonomous agent diff --git a/src/agent/grok_agent.py b/src/agent/grok_agent.py new file mode 100644 index 0000000..8c18bf4 --- /dev/null +++ b/src/agent/grok_agent.py @@ -0,0 +1,677 @@ +#!/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 = "" + + +# ============================================================================= +# 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; relative OR absolute) + lang_path_pattern = re.compile(r'^(\w+):([^\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: + # Use output_dir if provided, otherwise use target + base_root = state.output_dir if state.output_dir 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 + return f"""Continue working on: {task} + +Previous iteration ({state.iteration - 1}) response: +{state.last_response[:15000]} + +Iteration {state.iteration}/{state.max_iterations} + +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 +``` +""" + + +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 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 + 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") + 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 9d65c44..7638925 100644 --- a/src/bridge/grok_bridge.py +++ b/src/bridge/grok_bridge.py @@ -30,6 +30,14 @@ except ImportError: _record_usage = None +_shared_dir = str(Path(__file__).parent.parent / "shared") +if _shared_dir not in sys.path: + sys.path.insert(0, _shared_dir) +try: + from patterns import get_filename_pattern_string as _get_filename_pattern_string +except ImportError: + _get_filename_pattern_string = None + OPENROUTER_BASE = "https://openrouter.ai/api/v1" MODEL_ID = "x-ai/grok-4.20-multi-agent-beta" @@ -243,10 +251,16 @@ def parse_and_write_files(response_text, output_dir): written = [] output_path = Path(output_dir) - # Pattern for lang:path at start of block (language tag contains path) + # Pattern 1: lang:path at start of block (relative OR absolute path) lang_path_pattern = re.compile(r'^(\w+):([^\s\n]+)\n', re.MULTILINE) - # Pattern for // FILE: or # FILE: markers + # Pattern 2: // FILE: or # FILE: markers inside the block file_marker_pattern = re.compile(r'^\s*(?://|#)\s*FILE:\s*(.+?)\s*$', re.MULTILINE) + # Pattern 3: bare '# filename.ext' as first line (common Grok output) + filename_pattern = ( + re.compile(_get_filename_pattern_string(), re.MULTILINE) + if _get_filename_pattern_string is not None + else None + ) def _write_file(file_path, content): """Validate path, write content, and record result. Returns True on success.""" @@ -281,7 +295,20 @@ def _write_file(file_path, content): marker_match = file_marker_pattern.search(part) if marker_match: _write_file(marker_match.group(1).strip(), part[marker_match.end():]) - + continue + + # Pattern 3: bare '# filename.ext' as first non-empty line + if filename_pattern is not None: + # Strip the language tag line if present, then check first content line + content_start = part.find('\n') + first_line_end = part.find('\n', content_start + 1) if content_start >= 0 else -1 + first_content = part[content_start + 1:first_line_end].strip() if content_start >= 0 else "" + fn_match = filename_pattern.match(first_content) + if fn_match: + filename = fn_match.group(1) + rest = part[first_line_end + 1:] if first_line_end >= 0 else "" + _write_file(filename, rest) + return written def detect_high_thinking(prompt): 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 c636447..c7380dd 100644 --- a/src/plugin/index.ts +++ b/src/plugin/index.ts @@ -213,4 +213,8 @@ export default function (api: any) { }, { optional: true }, ); + + // Register autonomous agent tool + const { default: registerAgent } = await import("./grok_agent_plugin.js"); + registerAgent(api); } \ No newline at end of file diff --git a/src/shared/__init__.py b/src/shared/__init__.py new file mode 100644 index 0000000..25de497 --- /dev/null +++ b/src/shared/__init__.py @@ -0,0 +1 @@ +# grok-swarm shared utilities diff --git a/src/shared/patterns.py b/src/shared/patterns.py new file mode 100644 index 0000000..d5ec915 --- /dev/null +++ b/src/shared/patterns.py @@ -0,0 +1,28 @@ +#!/usr/bin/env python3 +""" +patterns.py — Shared regex patterns for parsing Grok response code blocks. + +Used by grok_bridge.py and grok_agent.py to identify file annotations inside +fenced code blocks in Grok's markdown output. +""" + + +def get_filename_pattern_string(): + """ + Return a regex string matching '# filename.ext' as the first line of a + code block — a common Grok output pattern when no explicit FILE: marker + is present. + + Pattern matches lines like: + # task.py + # utils/helpers.js + # README.md + + Does NOT match: + # /absolute/path/file.py (use file_marker_pattern for those) + # some comment without extension + # FILE: path (handled by file_marker_pattern) + + Group 1 captures the filename (possibly with a relative subdirectory). + """ + return r'^#\s+([^\s/][^\s]*\.[a-zA-Z0-9]+)\s*$' From 4d357c2fa2ddea2af8eeba8174330abe89971dd7 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 21 Mar 2026 04:34:45 +0000 Subject: [PATCH 05/11] chore: sync skills/grok-refactor/bridge/ with src/bridge/ changes Applies all issue fixes (#14, #12, #11, #16) to the duplicate skills/grok-refactor/bridge/ files so both installation paths stay in parity: - grok_bridge.py: absolute path rebasing, fallback response save, grounding prompt, usage tracking, shared patterns Pattern 3 - usage_tracker.py: new (copied from src/bridge/) - shared/patterns.py: new (provides get_filename_pattern_string()) - SKILL.md: documents grounding prompt config and stats subcommand https://claude.ai/code/session_01No9S6TZbfTgocHwHYWwxRN --- skills/grok-refactor/SKILL.md | 19 +++ skills/grok-refactor/bridge/grok_bridge.py | 156 +++++++++++++++++-- skills/grok-refactor/bridge/usage_tracker.py | 142 +++++++++++++++++ skills/grok-refactor/shared/patterns.py | 28 ++++ 4 files changed, 332 insertions(+), 13 deletions(-) create mode 100644 skills/grok-refactor/bridge/usage_tracker.py create mode 100644 skills/grok-refactor/shared/patterns.py diff --git a/skills/grok-refactor/SKILL.md b/skills/grok-refactor/SKILL.md index eae9d95..1e5cf94 100644 --- a/skills/grok-refactor/SKILL.md +++ b/skills/grok-refactor/SKILL.md @@ -78,6 +78,25 @@ echo '{"api_key": "sk-or-v1-..."}' > ~/.config/grok-swarm/config.json chmod 600 ~/.config/grok-swarm/config.json ``` +## Grounding Prompt (optional) + +Grok's role and default behaviour can be customised by creating a plain-text file at `~/.config/grok-swarm/system-prompt.txt`. Its contents are prepended to every mode-specific system prompt. Leave the file absent to use the built-in default. + +```bash +cat > ~/.config/grok-swarm/system-prompt.txt <<'EOF' +You are Grok, a senior engineer focused on security and correctness above all else. +EOF +``` + +## Usage Stats + +Track token spend across sessions: + +```bash +grok-swarm stats # all-time totals +grok-swarm stats --days 7 # last 7 days +``` + ## Installation ```bash diff --git a/skills/grok-refactor/bridge/grok_bridge.py b/skills/grok-refactor/bridge/grok_bridge.py index 173b4b1..7638925 100644 --- a/skills/grok-refactor/bridge/grok_bridge.py +++ b/skills/grok-refactor/bridge/grok_bridge.py @@ -25,10 +25,68 @@ print("ERROR: openai package required. Install: pip3 install openai", file=sys.stderr) sys.exit(1) +try: + from usage_tracker import record_usage as _record_usage +except ImportError: + _record_usage = None + +_shared_dir = str(Path(__file__).parent.parent / "shared") +if _shared_dir not in sys.path: + sys.path.insert(0, _shared_dir) +try: + from patterns import get_filename_pattern_string as _get_filename_pattern_string +except ImportError: + _get_filename_pattern_string = None + OPENROUTER_BASE = "https://openrouter.ai/api/v1" MODEL_ID = "x-ai/grok-4.20-multi-agent-beta" +# Agent counts per thinking level +AGENT_COUNTS = { + "low": 4, + "high": 16, +} + +# Plain-language phrases that trigger High Thinking mode automatically +HIGH_THINKING_PHRASES = [ + "16 agent swarm", + "16-agent swarm", + "high thinking", + "high thinking mode", + "thinking mode high", + "--thinking high", +] + +# Default grounding prompt prepended to every mode-specific system prompt. +# Users can override this by creating ~/.config/grok-swarm/system-prompt.txt +DEFAULT_GROUNDING_PROMPT = ( + "You are Grok, a specialized agentic coding assistant powered by xAI's multi-agent swarm. " + "Your primary role is to help software engineers write, analyze, refactor, and debug code. " + "You have access to a large context window and collaborate with multiple parallel agents to " + "produce thorough, well-reasoned results. Always produce production-quality output: " + "correct, readable, and idiomatic for the target language. " + "When modifying files, annotate every code block with its file path so changes can be " + "applied automatically." +) + + +def load_grounding_prompt(): + """ + Return the user's custom grounding prompt from + ~/.config/grok-swarm/system-prompt.txt, or DEFAULT_GROUNDING_PROMPT. + """ + custom = Path.home() / ".config" / "grok-swarm" / "system-prompt.txt" + if custom.exists(): + try: + text = custom.read_text(encoding="utf-8").strip() + if text: + return text + except OSError: + pass + return DEFAULT_GROUNDING_PROMPT + + # Mode-specific system prompts MODE_PROMPTS = { "refactor": ( @@ -151,11 +209,20 @@ def _safe_dest(output_path, file_path): """ Resolve ``file_path`` relative to ``output_path`` and verify the result stays inside ``output_path``. Returns the resolved Path or raises - ValueError for unsafe paths (absolute, containing ``..``, etc.). + ValueError for unsafe paths (containing ``..``, etc.). + + Absolute paths are rebased under ``output_path`` using only the filename + component (e.g. ``/home/user/.sandbox/task.py`` → ``/task.py``), + and a warning is printed to stderr. """ raw = Path(file_path) if raw.is_absolute(): - raise ValueError(f"Absolute paths are not allowed: {file_path!r}") + rebased = Path(raw.name) + print( + f"WARNING: Absolute path rebased to output dir: {file_path!r} → {rebased!r}", + file=sys.stderr, + ) + raw = rebased if ".." in raw.parts: raise ValueError(f"Path traversal not allowed: {file_path!r}") dest = (output_path / raw).resolve() @@ -184,10 +251,16 @@ def parse_and_write_files(response_text, output_dir): written = [] output_path = Path(output_dir) - # Pattern for lang:path at start of block (language tag contains path) + # Pattern 1: lang:path at start of block (relative OR absolute path) lang_path_pattern = re.compile(r'^(\w+):([^\s\n]+)\n', re.MULTILINE) - # Pattern for // FILE: or # FILE: markers + # Pattern 2: // FILE: or # FILE: markers inside the block file_marker_pattern = re.compile(r'^\s*(?://|#)\s*FILE:\s*(.+?)\s*$', re.MULTILINE) + # Pattern 3: bare '# filename.ext' as first line (common Grok output) + filename_pattern = ( + re.compile(_get_filename_pattern_string(), re.MULTILINE) + if _get_filename_pattern_string is not None + else None + ) def _write_file(file_path, content): """Validate path, write content, and record result. Returns True on success.""" @@ -222,10 +295,29 @@ def _write_file(file_path, content): marker_match = file_marker_pattern.search(part) if marker_match: _write_file(marker_match.group(1).strip(), part[marker_match.end():]) - + continue + + # Pattern 3: bare '# filename.ext' as first non-empty line + if filename_pattern is not None: + # Strip the language tag line if present, then check first content line + content_start = part.find('\n') + first_line_end = part.find('\n', content_start + 1) if content_start >= 0 else -1 + first_content = part[content_start + 1:first_line_end].strip() if content_start >= 0 else "" + fn_match = filename_pattern.match(first_content) + if fn_match: + filename = fn_match.group(1) + rest = part[first_line_end + 1:] if first_line_end >= 0 else "" + _write_file(filename, rest) + return written -def call_grok(prompt, mode="reason", context="", system_override=None, tools=None, timeout=120): +def detect_high_thinking(prompt): + """Return True if the prompt contains a plain-language High Thinking trigger.""" + lower = prompt.lower() + return any(phrase in lower for phrase in HIGH_THINKING_PHRASES) + + +def call_grok(prompt, mode="reason", context="", system_override=None, tools=None, timeout=120, thinking="low"): """Make the API call to Grok 4.20 Multi-Agent Beta.""" api_key = get_api_key() if not api_key: @@ -235,12 +327,15 @@ def call_grok(prompt, mode="reason", context="", system_override=None, tools=Non # Resolve system prompt if system_override: + # orchestrate mode: user owns the full system prompt, skip grounding system_content = system_override else: - system_content = MODE_PROMPTS.get(mode) - if system_content is None: + mode_prompt = MODE_PROMPTS.get(mode) + if mode_prompt is None: print(f"ERROR: Mode '{mode}' requires --system flag", file=sys.stderr) sys.exit(1) + grounding = load_grounding_prompt() + system_content = grounding + "\n\n" + mode_prompt # Append context to system prompt if context: @@ -262,7 +357,7 @@ def call_grok(prompt, mode="reason", context="", system_override=None, tools=Non "messages": messages, "max_tokens": 16384, "temperature": 0.3, - "extra_body": {"agent_count": 4}, + "extra_body": {"agent_count": AGENT_COUNTS.get(thinking, AGENT_COUNTS["low"])}, } if tools: @@ -288,8 +383,22 @@ def call_grok(prompt, mode="reason", context="", system_override=None, tools=Non # Log usage if hasattr(response, 'usage') and response.usage: u = response.usage - print(f"USAGE: mode={mode} prompt={u.prompt_tokens} completion={u.completion_tokens} " + agent_count = AGENT_COUNTS.get(thinking, AGENT_COUNTS["low"]) + print(f"USAGE: mode={mode} thinking={thinking} agents={agent_count} " + f"prompt={u.prompt_tokens} completion={u.completion_tokens} " f"total={u.total_tokens} time={elapsed:.1f}s", file=sys.stderr) + if _record_usage is not None: + try: + _record_usage( + mode=mode, + thinking=thinking, + prompt_tokens=u.prompt_tokens, + completion_tokens=u.completion_tokens, + total_tokens=u.total_tokens, + elapsed_secs=elapsed, + ) + except Exception: + pass # Handle content filtering if choice.finish_reason == "content_filter": @@ -337,9 +446,20 @@ def main(): help="Parse response for annotated code blocks and write to --output-dir") parser.add_argument("--output-dir", default="./grok-output/", help="Directory for file writes (default: ./grok-output/)") + parser.add_argument("--thinking", default=None, choices=["low", "high"], + help="Thinking level: low (4 agents) or high (16 agents, High Thinking mode) (default: low)") args = parser.parse_args() + # Auto-detect High Thinking mode from plain language in prompt (only if not explicitly set) + thinking = args.thinking + if thinking is None: + if detect_high_thinking(args.prompt): + thinking = "high" + print("INFO: High Thinking mode detected from prompt — using 16-agent swarm", file=sys.stderr) + else: + thinking = "low" + # Validate orchestrate mode if args.mode == "orchestrate" and not args.system: print("ERROR: --mode orchestrate requires --system flag", file=sys.stderr) @@ -356,7 +476,9 @@ def main(): tools = load_tools(args.tools) # Call Grok - print(f"Calling {MODEL_ID} (mode={args.mode}, 4 agents, timeout={args.timeout}s)...", file=sys.stderr) + agent_count = AGENT_COUNTS.get(thinking, AGENT_COUNTS["low"]) + thinking_label = " [HIGH THINKING MODE — 16-agent swarm]" if thinking == "high" else "" + print(f"Calling {MODEL_ID} (mode={args.mode}, {agent_count} agents, timeout={args.timeout}s){thinking_label}...", file=sys.stderr) result = call_grok( prompt=args.prompt, mode=args.mode, @@ -364,6 +486,7 @@ def main(): system_override=args.system, tools=tools, timeout=args.timeout, + thinking=thinking, ) # Output @@ -380,11 +503,18 @@ def main(): print(f" {rel_path} ({byte_count:,} bytes)") print(f"Total: {total_bytes:,} bytes") else: + # Save full response as a fallback so no output is lost + fallback_dir = Path(args.output_dir) + fallback_dir.mkdir(parents=True, exist_ok=True) + fallback_path = fallback_dir / "grok-response.txt" + fallback_path.write_text(result, encoding="utf-8") print( - "No annotated files found in model response to write to disk.\n" - "Re-run without --write-files to see the full response.", + f"ERROR: No annotated files found in model response.\n" + f"Full response saved to: {fallback_path}\n" + f"Tip: ask Grok to annotate code blocks with ```lang:path/to/file or # FILE: path/to/file", file=sys.stderr, ) + sys.exit(1) elif not args.output: print(result) diff --git a/skills/grok-refactor/bridge/usage_tracker.py b/skills/grok-refactor/bridge/usage_tracker.py new file mode 100644 index 0000000..f8e45e0 --- /dev/null +++ b/skills/grok-refactor/bridge/usage_tracker.py @@ -0,0 +1,142 @@ +#!/usr/bin/env python3 +""" +usage_tracker.py — Persistent token/cost tracking for grok-swarm requests. + +Logs each API call to ~/.config/grok-swarm/usage.json (newline-delimited JSON). +Provides aggregation and a human-readable stats report. + +OpenRouter pricing for x-ai/grok-4.20-multi-agent-beta (as of 2026-03): + Prompt tokens: $5.00 / 1M tokens + Completion tokens: $15.00 / 1M tokens +""" + +import json +import sys +from datetime import datetime, timezone, timedelta +from pathlib import Path + +USAGE_FILE = Path.home() / ".config" / "grok-swarm" / "usage.json" + +# OpenRouter pricing (USD per 1M tokens) +PROMPT_PRICE_PER_M = 5.00 +COMPLETION_PRICE_PER_M = 15.00 + + +def record_usage(mode, thinking, prompt_tokens, completion_tokens, total_tokens, elapsed_secs): + """ + Append one usage record to the log file. + Never raises — failures are printed to stderr and silently ignored. + """ + record = { + "ts": datetime.now(timezone.utc).isoformat(), + "mode": mode, + "thinking": thinking, + "prompt_tokens": prompt_tokens, + "completion_tokens": completion_tokens, + "total_tokens": total_tokens, + "elapsed_secs": round(elapsed_secs, 2), + "cost_usd": round( + (prompt_tokens / 1_000_000) * PROMPT_PRICE_PER_M + + (completion_tokens / 1_000_000) * COMPLETION_PRICE_PER_M, + 6, + ), + } + try: + USAGE_FILE.parent.mkdir(parents=True, exist_ok=True) + with USAGE_FILE.open("a", encoding="utf-8") as f: + f.write(json.dumps(record) + "\n") + except OSError as exc: + print(f"WARNING: Could not write usage log: {exc}", file=sys.stderr) + + +def get_stats(since_days=None): + """ + Read the usage log and return aggregated stats dict. + + Args: + since_days: If set, only include records from the last N days. + + Returns dict with keys: + calls, prompt_tokens, completion_tokens, total_tokens, + cost_usd, elapsed_secs, by_mode (dict of mode -> call count) + """ + if not USAGE_FILE.exists(): + return None + + cutoff = None + if since_days is not None: + cutoff = datetime.now(timezone.utc) - timedelta(days=since_days) + + stats = { + "calls": 0, + "prompt_tokens": 0, + "completion_tokens": 0, + "total_tokens": 0, + "cost_usd": 0.0, + "elapsed_secs": 0.0, + "by_mode": {}, + } + + try: + with USAGE_FILE.open(encoding="utf-8") as f: + for line in f: + line = line.strip() + if not line: + continue + try: + rec = json.loads(line) + except json.JSONDecodeError: + continue + + if cutoff: + try: + rec_ts = datetime.fromisoformat(rec["ts"]) + if rec_ts < cutoff: + continue + except (KeyError, ValueError): + pass + + stats["calls"] += 1 + stats["prompt_tokens"] += rec.get("prompt_tokens", 0) + stats["completion_tokens"] += rec.get("completion_tokens", 0) + stats["total_tokens"] += rec.get("total_tokens", 0) + stats["cost_usd"] += rec.get("cost_usd", 0.0) + stats["elapsed_secs"] += rec.get("elapsed_secs", 0.0) + + mode = rec.get("mode", "unknown") + stats["by_mode"][mode] = stats["by_mode"].get(mode, 0) + 1 + + except OSError as exc: + print(f"WARNING: Could not read usage log: {exc}", file=sys.stderr) + return None + + return stats + + +def format_stats_report(stats, since_days=None): + """Return a human-readable stats report string.""" + if stats is None or stats["calls"] == 0: + return "No usage data recorded yet. Run a grok-swarm command to start tracking." + + period = f"last {since_days} day(s)" if since_days else "all time" + lines = [ + "", + "=" * 55, + f" grok-swarm usage stats ({period})", + "=" * 55, + f" Requests: {stats['calls']:>10,}", + f" Prompt tokens: {stats['prompt_tokens']:>10,}", + f" Completion tokens: {stats['completion_tokens']:>10,}", + f" Total tokens: {stats['total_tokens']:>10,}", + f" Estimated cost: ${stats['cost_usd']:>10.4f}", + f" Total time: {stats['elapsed_secs']:>9.1f}s", + ] + if stats["by_mode"]: + lines.append("") + lines.append(" By mode:") + for mode, count in sorted(stats["by_mode"].items(), key=lambda x: -x[1]): + lines.append(f" {mode:<16} {count:>6,} call(s)") + lines.append("=" * 55) + lines.append(f" Log file: {USAGE_FILE}") + lines.append("=" * 55) + return "\n".join(lines) diff --git a/skills/grok-refactor/shared/patterns.py b/skills/grok-refactor/shared/patterns.py new file mode 100644 index 0000000..d5ec915 --- /dev/null +++ b/skills/grok-refactor/shared/patterns.py @@ -0,0 +1,28 @@ +#!/usr/bin/env python3 +""" +patterns.py — Shared regex patterns for parsing Grok response code blocks. + +Used by grok_bridge.py and grok_agent.py to identify file annotations inside +fenced code blocks in Grok's markdown output. +""" + + +def get_filename_pattern_string(): + """ + Return a regex string matching '# filename.ext' as the first line of a + code block — a common Grok output pattern when no explicit FILE: marker + is present. + + Pattern matches lines like: + # task.py + # utils/helpers.js + # README.md + + Does NOT match: + # /absolute/path/file.py (use file_marker_pattern for those) + # some comment without extension + # FILE: path (handled by file_marker_pattern) + + Group 1 captures the filename (possibly with a relative subdirectory). + """ + return r'^#\s+([^\s/][^\s]*\.[a-zA-Z0-9]+)\s*$' From c0da7a38773a33ab87752d3587d58d601ac12aff Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 21 Mar 2026 05:16:27 +0000 Subject: [PATCH 06/11] Initial plan From dcfa0d60ca7458d1017cf3e1da0e977be9568309 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 21 Mar 2026 05:20:52 +0000 Subject: [PATCH 07/11] fix: address all PR review comments from review #3985554052 Co-authored-by: KHAEntertainment <43256680+KHAEntertainment@users.noreply.github.com> Agent-Logs-Url: https://github.com/KHAEntertainment/grok-multiagent-plugin/sessions/2dc54b6f-c039-4e6b-8100-e7a374c6c952 --- platforms/claude/commands/grok-swarm-agent.sh | 2 +- skills/grok-refactor/bridge/grok_bridge.py | 22 ++++++++++++------- skills/grok-refactor/bridge/usage_tracker.py | 5 ++++- src/agent/grok_agent.py | 5 +++-- src/bridge/grok_bridge.py | 22 ++++++++++++------- src/bridge/usage_tracker.py | 5 ++++- src/plugin/grok_agent_plugin.ts | 9 ++++---- src/plugin/index.ts | 4 ++-- 8 files changed, 47 insertions(+), 27 deletions(-) diff --git a/platforms/claude/commands/grok-swarm-agent.sh b/platforms/claude/commands/grok-swarm-agent.sh index 6455340..a76aeee 100755 --- a/platforms/claude/commands/grok-swarm-agent.sh +++ b/platforms/claude/commands/grok-swarm-agent.sh @@ -75,7 +75,7 @@ if [[ -n "$VERIFY_CMD" ]]; then ARGS+=("--verify-cmd" "$VERIFY_CMD") fi -ARGS+=("$TASK") +ARGS+=("--task" "$TASK") # Execute python3 "${ARGS[@]}" \ No newline at end of file diff --git a/skills/grok-refactor/bridge/grok_bridge.py b/skills/grok-refactor/bridge/grok_bridge.py index 7638925..be2dcbf 100644 --- a/skills/grok-refactor/bridge/grok_bridge.py +++ b/skills/grok-refactor/bridge/grok_bridge.py @@ -299,15 +299,21 @@ def _write_file(file_path, content): # Pattern 3: bare '# filename.ext' as first non-empty line if filename_pattern is not None: - # Strip the language tag line if present, then check first content line + # Strip the language tag line if present, then scan for first non-empty content line content_start = part.find('\n') - first_line_end = part.find('\n', content_start + 1) if content_start >= 0 else -1 - first_content = part[content_start + 1:first_line_end].strip() if content_start >= 0 else "" - fn_match = filename_pattern.match(first_content) - if fn_match: - filename = fn_match.group(1) - rest = part[first_line_end + 1:] if first_line_end >= 0 else "" - _write_file(filename, rest) + content = part[content_start + 1:] if content_start >= 0 else part + lines = content.splitlines(keepends=True) + for idx, line in enumerate(lines): + stripped = line.strip() + if not stripped: + continue + fn_match = filename_pattern.match(stripped) + if fn_match: + filename = fn_match.group(1) + rest = "".join(lines[idx + 1:]) + _write_file(filename, rest) + # Only consider the first non-empty line for Pattern 3 + break return written diff --git a/skills/grok-refactor/bridge/usage_tracker.py b/skills/grok-refactor/bridge/usage_tracker.py index f8e45e0..c47fe95 100644 --- a/skills/grok-refactor/bridge/usage_tracker.py +++ b/skills/grok-refactor/bridge/usage_tracker.py @@ -91,9 +91,12 @@ def get_stats(since_days=None): if cutoff: try: rec_ts = datetime.fromisoformat(rec["ts"]) + # Normalize naive timestamps (assume UTC) so comparison is safe + if rec_ts.tzinfo is None: + rec_ts = rec_ts.replace(tzinfo=timezone.utc) if rec_ts < cutoff: continue - except (KeyError, ValueError): + except (KeyError, ValueError, TypeError): pass stats["calls"] += 1 diff --git a/src/agent/grok_agent.py b/src/agent/grok_agent.py index 8c18bf4..3e9b4e4 100644 --- a/src/agent/grok_agent.py +++ b/src/agent/grok_agent.py @@ -21,6 +21,7 @@ import argparse import re +import shlex import subprocess import sys import time @@ -417,8 +418,8 @@ def verify_changes(state: AgentState) -> tuple[bool, str]: try: result = subprocess.run( - state.verify_cmd, - shell=True, + shlex.split(state.verify_cmd), + shell=False, capture_output=True, text=True, timeout=120, diff --git a/src/bridge/grok_bridge.py b/src/bridge/grok_bridge.py index 7638925..be2dcbf 100644 --- a/src/bridge/grok_bridge.py +++ b/src/bridge/grok_bridge.py @@ -299,15 +299,21 @@ def _write_file(file_path, content): # Pattern 3: bare '# filename.ext' as first non-empty line if filename_pattern is not None: - # Strip the language tag line if present, then check first content line + # Strip the language tag line if present, then scan for first non-empty content line content_start = part.find('\n') - first_line_end = part.find('\n', content_start + 1) if content_start >= 0 else -1 - first_content = part[content_start + 1:first_line_end].strip() if content_start >= 0 else "" - fn_match = filename_pattern.match(first_content) - if fn_match: - filename = fn_match.group(1) - rest = part[first_line_end + 1:] if first_line_end >= 0 else "" - _write_file(filename, rest) + content = part[content_start + 1:] if content_start >= 0 else part + lines = content.splitlines(keepends=True) + for idx, line in enumerate(lines): + stripped = line.strip() + if not stripped: + continue + fn_match = filename_pattern.match(stripped) + if fn_match: + filename = fn_match.group(1) + rest = "".join(lines[idx + 1:]) + _write_file(filename, rest) + # Only consider the first non-empty line for Pattern 3 + break return written diff --git a/src/bridge/usage_tracker.py b/src/bridge/usage_tracker.py index f8e45e0..c47fe95 100644 --- a/src/bridge/usage_tracker.py +++ b/src/bridge/usage_tracker.py @@ -91,9 +91,12 @@ def get_stats(since_days=None): if cutoff: try: rec_ts = datetime.fromisoformat(rec["ts"]) + # Normalize naive timestamps (assume UTC) so comparison is safe + if rec_ts.tzinfo is None: + rec_ts = rec_ts.replace(tzinfo=timezone.utc) if rec_ts < cutoff: continue - except (KeyError, ValueError): + except (KeyError, ValueError, TypeError): pass stats["calls"] += 1 diff --git a/src/plugin/grok_agent_plugin.ts b/src/plugin/grok_agent_plugin.ts index a3ec699..a2f8fd3 100644 --- a/src/plugin/grok_agent_plugin.ts +++ b/src/plugin/grok_agent_plugin.ts @@ -16,7 +16,7 @@ import { join, dirname } from "path"; import { fileURLToPath } from "url"; import { Type } from "@sinclair/typebox"; -const PLUGIN_ROOT = join(dirname(fileURLToPath(import.meta.url)), "..", "..", ".."); +const PLUGIN_ROOT = join(dirname(fileURLToPath(import.meta.url)), "..", ".."); const DEFAULT_AGENT = join(PLUGIN_ROOT, "src", "agent", "grok_agent.py"); const GrokAgentSchema = Type.Object({ @@ -56,7 +56,7 @@ export default function (api: any) { }); try { - const agentScript = api.config?.agentScript || DEFAULT_AGENT; + const agentScript = DEFAULT_AGENT; // Validate agent script exists if (!existsSync(agentScript)) { @@ -84,11 +84,12 @@ export default function (api: any) { args.push("--verify-cmd", params.verify_cmd); } - args.push("--", params.task); + args.push("--task", params.task); // Spawn agent with timeout enforcement return new Promise((resolve) => { - const child = spawn("python3", args, { + const pythonPath = api.config?.pythonPath || "python3"; + const child = spawn(pythonPath, args, { stdio: ["ignore", "pipe", "pipe"], env: { ...process.env }, }); diff --git a/src/plugin/index.ts b/src/plugin/index.ts index c7380dd..bb915a7 100644 --- a/src/plugin/index.ts +++ b/src/plugin/index.ts @@ -76,7 +76,7 @@ const GrokSwarmSchema = Type.Object({ ), }); -export default function (api: any) { +export default async function (api: any) { api.registerTool( { name: "grok_swarm", @@ -215,6 +215,6 @@ export default function (api: any) { ); // Register autonomous agent tool - const { default: registerAgent } = await import("./grok_agent_plugin.js"); + const { default: registerAgent } = await import("./grok_agent_plugin"); registerAgent(api); } \ No newline at end of file From 38d5227dadc65b4c0658098798868997e8a0d572 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 21 Mar 2026 05:35:53 +0000 Subject: [PATCH 08/11] Initial plan From 08dc0f318b3a8e7e2eb8d0832bcf71a2eac81de7 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 21 Mar 2026 05:36:01 +0000 Subject: [PATCH 09/11] Initial plan From 53781210bf54afdc1a42cf432ea5279f7cee4e6c Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 21 Mar 2026 05:36:31 +0000 Subject: [PATCH 10/11] Initial plan From d5e94b6fceff4ced74fc15f0bd2e45eeaa3defc0 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 21 Mar 2026 05:38:17 +0000 Subject: [PATCH 11/11] fix: use api.config?.pythonPath in grok_agent_plugin.ts instead of hardcoded python3 Co-authored-by: KHAEntertainment <43256680+KHAEntertainment@users.noreply.github.com> Agent-Logs-Url: https://github.com/KHAEntertainment/grok-multiagent-plugin/sessions/aa43e519-f955-4eac-ab29-27a12839c43c --- src/plugin/grok_agent_plugin.ts | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/src/plugin/grok_agent_plugin.ts b/src/plugin/grok_agent_plugin.ts index a3ec699..9be6cf4 100644 --- a/src/plugin/grok_agent_plugin.ts +++ b/src/plugin/grok_agent_plugin.ts @@ -18,6 +18,7 @@ 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 DEFAULT_PYTHON = join(PLUGIN_ROOT, "skills", "grok-refactor", ".venv", "bin", "python3"); const GrokAgentSchema = Type.Object({ task: Type.String({ description: "Natural language task instruction" }), @@ -57,6 +58,7 @@ export default function (api: any) { try { const agentScript = api.config?.agentScript || DEFAULT_AGENT; + const pythonPath = api.config?.pythonPath || DEFAULT_PYTHON; // Validate agent script exists if (!existsSync(agentScript)) { @@ -88,7 +90,7 @@ export default function (api: any) { // Spawn agent with timeout enforcement return new Promise((resolve) => { - const child = spawn("python3", args, { + const child = spawn(pythonPath, args, { stdio: ["ignore", "pipe", "pipe"], env: { ...process.env }, });