From 4f369af5e2972411f56059643dd546e29a25a7c2 Mon Sep 17 00:00:00 2001 From: GSSwain <12575773+GSSwain@users.noreply.github.com> Date: Sun, 14 Jun 2026 07:14:04 +1000 Subject: [PATCH] adds support for claude --- README.md | 42 +++-- docs/keys_setup.md | 30 ++- src/maca/maca_config.py | 65 ++++++- src/maca/main.py | 112 +++++++++--- src/maca/models/__init__.py | 3 + src/maca/models/claude.py | 117 ++++++++++++ src/maca/orchestrator.py | 291 ++++++++++++++++++++++++------ tests/test_behavior.py | 236 ++++++++++++++++++++++++ tests/test_orchestrator_status.py | 15 +- 9 files changed, 803 insertions(+), 108 deletions(-) create mode 100644 src/maca/models/claude.py diff --git a/README.md b/README.md index a479656..b814f4e 100644 --- a/README.md +++ b/README.md @@ -18,7 +18,8 @@ MACA is a **local-first, hybrid AI coding assistant** that routes work by comple - **Smart task routing:** analyzes each request and sends it to the best backend for the job. - **Simple tasks:** local Gemma (`gemma2:2b`) via Ollama. - - **Medium / complex / very complex tasks:** Google Gemini (`gemini-3.5-flash`) via REST API when a key is configured. + - **Medium complexity tasks:** Google Gemini (`gemini-3.5-flash`) via REST API. If Gemini is offline/unconfigured, falls back to Anthropic Claude, then local Gemma. + - **Complex / Very Complex tasks:** Anthropic Claude (default `claude-opus-4-8` or custom model) via Messages API. If Claude is offline/unconfigured, falls back to Google Gemini, then local Gemma. - **Multi-agent orchestration:** coordinates a planner, coder, and reviewer to work through tasks in sequence. - **Planner Agent:** inspects the codebase and creates an implementation plan. - **Coder Agent:** writes and applies the actual code changes. @@ -33,21 +34,35 @@ MACA is a **local-first, hybrid AI coding assistant** that routes work by comple ```mermaid graph TD User([User Prompt]) --> Router[Task Router] - Router -->|SIMPLE| Gemma[Local Gemma Client] - Router -->|MEDIUM / COMPLEX / VERY_COMPLEX| Gemini[Gemini Client] - - Gemma & Gemini --> Orchestrator[Multi-Agent Orchestrator] - + Router -.->|1. Classify Task| Gemma[Local Gemma Client] + Gemma -.->|2. Complexity Result| Router + + Router -->|SIMPLE| Gemma + Router -->|MEDIUM| Decider1{Backend Availability} + Router -->|COMPLEX / VERY_COMPLEX| Decider2{Backend Availability} + + Decider1 -->|Gemini Available| Gemini[Gemini Client] + Decider1 -->|Fallback| Claude[Claude Client] + Decider1 -->|None Available| Gemma + + Decider2 -->|Claude Available| Claude + Decider2 -->|Fallback| Gemini + Decider2 -->|None Available| Gemma + + Gemma & Gemini & Claude --> Orchestrator[Multi-Agent Orchestrator] + subgraph Agents Orchestrator --> Planner[Planner Agent] Planner --> Coder[Coder Agent] Coder --> Reviewer[Reviewer Agent] end - + Reviewer --> Write[File System Writer] Write --> Workspace[(Workspace)] ``` +> **Note on Complexity Classification**: The Task Router queries the local Gemma model (`gemma2:2b`) to classify prompt complexity. If Gemma is offline, it falls back to a regex-based keyword density and word count heuristic classifier. + --- ## 🛠️ Setup Instructions @@ -65,17 +80,21 @@ If you only need the Gemma/Ollama setup, use: ``` ### 2. Configure API Keys (Environment or macOS Keychain) -MACA currently uses Gemini for medium and above tasks. For secure setup details, see [docs/keys_setup.md](docs/keys_setup.md). +MACA uses Google Gemini and Anthropic Claude for medium and above tasks. For secure setup details, see [docs/keys_setup.md](docs/keys_setup.md). * **Quick Env Setup**: ```bash export GEMINI_API_KEY="your-gemini-api-key" + export CLAUDE_API_KEY="your-claude-api-key" ``` - - You can optionally customize the Gemini model and request timeout (in seconds): + + You can optionally customize the models and request timeouts (in seconds): ```bash export GEMINI_MODEL="gemini-3.5-flash" - export GEMINI_TIMEOUT_SECONDS="45" + export GEMINI_TIMEOUT_SECONDS="120" + + export CLAUDE_MODEL="claude-opus-4-8" + export CLAUDE_TIMEOUT_SECONDS="120" ``` * **Recommended macOS Keychain + .zshenv Setup**: ```bash @@ -110,6 +129,7 @@ maca "write a simple hello world script in output.py" Override default task routing: ```sh maca --model gemini "implement a custom tokenizer" +maca --model claude "implement a custom tokenizer" ``` ### Run in Mock Mode (Simulated Run) diff --git a/docs/keys_setup.md b/docs/keys_setup.md index 359ac15..ff4c262 100644 --- a/docs/keys_setup.md +++ b/docs/keys_setup.md @@ -1,6 +1,6 @@ # Secure Key Setup Guide for MACA -This guide documents how to acquire your **Google Gemini** API key and store it securely using the **macOS Keychain** so that the Multi-Agent Coding Assistant (MACA) can access it without exposing the secret in plain text. +This guide documents how to acquire your **Google Gemini** and **Anthropic Claude** API keys and store them securely using the **macOS Keychain** so that the Multi-Agent Coding Assistant (MACA) can access them without exposing the secrets in plain text. --- @@ -13,6 +13,12 @@ This guide documents how to acquire your **Google Gemini** API key and store it 4. Click **Create API key** (select a Google Cloud project or create a new one). 5. Copy your generated key. +### B. Anthropic Claude API Key +1. Visit the [Anthropic Console](https://console.anthropic.com/). +2. Sign in with your account. +3. Navigate to **API Keys**. +4. Click **Create Key**, name it, and copy the generated key. + --- ## 2. Secure Configuration via macOS Keychain @@ -25,31 +31,39 @@ Open your terminal and run the following command (replace `YOUR_GEMINI_API_KEY` security add-generic-password -a "$USER" -s "GEMINI_API_KEY" -w "YOUR_GEMINI_API_KEY" ``` +### B. Store Claude API Key +Open your terminal and run the following command (replace `YOUR_CLAUDE_API_KEY` with your actual key): +```bash +security add-generic-password -a "$USER" -s "CLAUDE_API_KEY" -w "YOUR_CLAUDE_API_KEY" +``` + --- -## 3. Add the Keychain Secret to ~/.zshenv +## 3. Add the Keychain Secrets to ~/.zshenv -To make the secret available automatically in every new zsh session, add this line to your shell startup file: +To make the secrets available automatically in every new zsh session, add these lines to your shell startup file: ```bash -printf '%s\n' 'export GEMINI_API_KEY="$(security find-generic-password -a "$USER" -s "GEMINI_API_KEY" -w 2>/dev/null)"' >> ~/.zshenv +printf "%s\n" "export GEMINI_API_KEY=\"\$(security find-generic-password -a \"\$USER\" -s \"GEMINI_API_KEY\" -w 2>/dev/null)\"" >> ~/.zshenv +printf "%s\n" "export CLAUDE_API_KEY=\"\$(security find-generic-password -a \"\$USER\" -s \"CLAUDE_API_KEY\" -w 2>/dev/null)\"" >> ~/.zshenv source ~/.zshenv ``` -This uses the Keychain value for `GEMINI_API_KEY` without storing the raw secret in `.zshrc` or other shell config files. +This uses the Keychain values for `GEMINI_API_KEY` and `CLAUDE_API_KEY` without storing the raw secrets in `.zshrc` or other shell config files. --- ## 4. Verifying Keychain Storage -You can verify that the key was successfully written to your macOS Keychain by retrieving it: +You can verify that the keys were successfully written to your macOS Keychain by retrieving them: ```bash security find-generic-password -a "$USER" -s "GEMINI_API_KEY" -w +security find-generic-password -a "$USER" -s "CLAUDE_API_KEY" -w ``` --- -## 5. How MACA Accesses the Key +## 5. How MACA Accesses the Keys -MACA reads `GEMINI_API_KEY` from your shell environment when it starts. The recommended macOS setup is to export that value from your Keychain in `~/.zshenv`, which makes the key available to the CLI in every new zsh session. +MACA reads `GEMINI_API_KEY` and `CLAUDE_API_KEY` from your shell environment when it starts. The recommended macOS setup is to export those values from your Keychain in `~/.zshenv`, which makes the keys available to the CLI in every new zsh session. diff --git a/src/maca/maca_config.py b/src/maca/maca_config.py index a9c3e6e..40ac016 100644 --- a/src/maca/maca_config.py +++ b/src/maca/maca_config.py @@ -6,25 +6,70 @@ def get_gemini_api_key(): global _gemini_key if _gemini_key is not None: return _gemini_key - _gemini_key = os.environ.get("GEMINI_API_KEY", "") + _gemini_key = os.environ.get("GEMINI_API_KEY", "").strip() return _gemini_key +_claude_key = None + +def get_claude_api_key(): + global _claude_key + if _claude_key is not None: + return _claude_key + _claude_key = os.environ.get("CLAUDE_API_KEY", "").strip() + return _claude_key + OLLAMA_API_URL = os.environ.get("OLLAMA_API_URL", "http://localhost:11434") OLLAMA_MODEL = os.environ.get("OLLAMA_MODEL", "gemma2:2b") -GEMINI_MODEL = os.environ.get("GEMINI_MODEL", "gemini-3.5-flash") +# Configurable timeouts (seconds) for cloud model API calls try: - GEMINI_TIMEOUT_SECONDS = float(os.environ.get("GEMINI_TIMEOUT_SECONDS", "45")) + GEMINI_TIMEOUT_SECONDS = float(os.environ.get("GEMINI_TIMEOUT_SECONDS", "90")) except ValueError: - GEMINI_TIMEOUT_SECONDS = 45.0 + GEMINI_TIMEOUT_SECONDS = 90.0 + +try: + CLAUDE_TIMEOUT_SECONDS = float(os.environ.get("CLAUDE_TIMEOUT_SECONDS", "90")) +except ValueError: + CLAUDE_TIMEOUT_SECONDS = 90.0 + +GEMINI_TIMEOUT = GEMINI_TIMEOUT_SECONDS +CLAUDE_TIMEOUT = CLAUDE_TIMEOUT_SECONDS + +# Claude model to use for API calls +CLAUDE_MODEL = os.environ.get("CLAUDE_MODEL", "claude-opus-4-8") + +# Gemini model to use for API calls +GEMINI_MODEL = os.environ.get("GEMINI_MODEL", "gemini-3.5-flash") + +# If set to True, will mock Gemma/Gemini/Claude calls if unavailable/unconfigured +MOCK_GEMMA_FALLBACK = os.environ.get("MOCK_GEMMA_FALLBACK", "False").lower() == "true" + +def validate_config(complexity, selected_agent=None): + """Validate that the selected agent has the required configuration. -# If set to True, will mock Gemma/Gemini calls if unavailable/unconfigured -MOCK_GEMMA_FALLBACK = os.environ.get("MOCK_GEMMA_FALLBACK", "True").lower() == "true" + Args: + complexity: The evaluated task complexity level. + selected_agent: The agent name that was selected for this task + (e.g. "GEMINI", "CLAUDE"). If None, falls back to + legacy behaviour of requiring Gemini for non-SIMPLE tasks. + """ + if complexity == "SIMPLE": + return # Local Gemma, no API key needed -def validate_config(complexity): - # Only SIMPLE tasks run Gemma (which is local and does not require API keys). - # All other tasks (MEDIUM, COMPLEX, VERY_COMPLEX) route to Gemini. - if complexity != "SIMPLE": + if selected_agent == "CLAUDE": + if not get_claude_api_key(): + raise ValueError( + "CLAUDE_API_KEY is not set in environment variables. " + "Claude is required for this task's complexity level." + ) + elif selected_agent == "GEMINI": + if not get_gemini_api_key(): + raise ValueError( + "GEMINI_API_KEY is not set in environment variables. " + "Gemini is required for this task's complexity level." + ) + else: + # Legacy fallback: require Gemini for any non-SIMPLE task if not get_gemini_api_key(): raise ValueError( "GEMINI_API_KEY is not set in environment variables. " diff --git a/src/maca/main.py b/src/maca/main.py index 9a1cebd..f445f86 100644 --- a/src/maca/main.py +++ b/src/maca/main.py @@ -18,15 +18,16 @@ def display_welcome(orch=None): "[bold cyan]Welcome to the Multi-Agent Coding Assistant (MACA)![/bold cyan]\n" "MACA classifies your coding task complexity and routes it to the best model:\n" " - [bold green]Simple Complexity[/bold green] -> [bold cyan]Local Gemma[/bold cyan] (via Ollama)\n" - " - [bold yellow]Medium/Complex/Very Complex Complexity[/bold yellow] -> [bold cyan]Google Gemini[/bold cyan] (via API)\n\n" + " - [bold yellow]Medium Complexity[/bold yellow] -> [bold cyan]Google Gemini[/bold cyan] (via API)\n" + " - [bold red]Complex / Very Complex Complexity[/bold red] -> [bold cyan]Anthropic Claude[/bold cyan] (via API)\n\n" "Type your task prompt directly, or use special slash commands:\n" " - [bold magenta]/exit[/bold magenta] or [bold magenta]/quit[/bold magenta]: Exit the assistant\n" " - [bold magenta]/status[/bold magenta]: Run live connection checks on all models\n" - " - [bold magenta]/model [/bold magenta]: Override default model (options: gemma, gemini, auto)\n" + " - [bold magenta]/model [/bold magenta]: Override default model (options: gemma, gemini, claude, auto)\n" " - [bold magenta]/help[/bold magenta]: Show this help message" ) console.print(Panel(welcome_text, title="[bold white]MACA CLI v1.0.0-alpha[/bold white]", border_style="cyan")) - + if orch: print_backends_status(orch, run_handshakes=False) @@ -34,41 +35,92 @@ def print_backends_status(orch, run_handshakes=False): title = "Backend Connectivity (Live Handshakes)" if run_handshakes else "Backend Connectivity (Fast Check)" with console.status("[bold yellow]Checking backends...", spinner="dots") if run_handshakes else console.status("[bold yellow]Checking config...", spinner="dots") as s: status_dict = orch.check_backends_status(run_handshakes=run_handshakes) - + table = Table(title=title) table.add_column("Backend Model", style="cyan") table.add_column("Connectivity Status", style="magenta") - + for name, stat in status_dict.items(): style_color = "green" if "ONLINE" in stat or "CONFIGURED" in stat else "red" if "OFFLINE" in stat: style_color = "yellow" table.add_row(name, f"[{style_color}]{stat}[/{style_color}]") - + console.print(table) + # Print full error details for any failed connections + for name, stat in status_dict.items(): + if "CONNECTION FAILED" in stat: + console.print(f"[bold red]Full Error for {name}:[/bold red]\n{stat}\n") + def contains_filename_or_project(prompt): prompt_lower = prompt.lower() - + # Check for common extensions extensions = [".py", ".js", ".ts", ".html", ".css", ".json", ".md", ".sh", ".java", ".cpp", ".h", ".cs", ".go", ".rs", ".yml", ".yaml", ".txt"] if any(ext in prompt_lower for ext in extensions): return True - + # Clean up punctuation and split clean_prompt = prompt_lower.replace(",", " ").replace(".", " ").replace("?", " ").replace("!", " ") words = clean_prompt.split() keywords = {"file", "project", "folder", "directory", "repo", "repository"} if any(w in keywords for w in words): return True - + return False +def parse_interactive_command(prompt, parser): + """Parse a full command line if pasted/entered into the interactive prompt. + Returns (task_description, model_override, is_command_line) or (None, None, False) + """ + import shlex + prompt_stripped = prompt.strip() + + # Check if the prompt starts with a command invocation prefix + prefixes = ["maca ", "./maca ", "python ", "python3 "] + starts_with_prefix = any(prompt_stripped.startswith(p) for p in prefixes) + + if not starts_with_prefix: + return None, None, False + + try: + tokens = shlex.split(prompt_stripped) + except Exception: + return None, None, False + + if not tokens: + return None, None, False + + # Determine where the arguments start + start_idx = 0 + if tokens[0] in ("python", "python3"): + if len(tokens) > 1 and (tokens[1].endswith(".py") or "main.py" in tokens[1] or "maca" in tokens[1]): + start_idx = 2 + else: + start_idx = 1 + elif tokens[0] in ("maca", "./maca"): + start_idx = 1 + else: + if "maca" in tokens[0]: + start_idx = 1 + else: + return None, None, False + + try: + sub_args, unknown = parser.parse_known_args(tokens[start_idx:]) + # If mock flag is specified, apply it globally + if getattr(sub_args, "mock", False): + config.MOCK_GEMMA_FALLBACK = True + return sub_args.task, sub_args.model, True + except Exception: + return None, None, False + def main(): parser = argparse.ArgumentParser(description="Multi-Agent Coding Assistant (MACA)") parser.add_argument("task", nargs="?", default=None, help="The coding task description") parser.add_argument("--repo", default=".", help="Target repository directory path") - parser.add_argument("--model", default=None, choices=["gemma", "gemini"], help="Force a specific model") + parser.add_argument("--model", default=None, choices=["gemma", "gemini", "claude"], help="Force a specific model") parser.add_argument("--mock", action="store_true", help="Force local Gemma simulated mode") args = parser.parse_args() @@ -100,7 +152,7 @@ def main(): with open(test_file, "w") as f: f.write("test") os.remove(test_file) - + console.print("[bold green]✓ Sandbox check: Read/Write access verified successfully.[/bold green]\n") except Exception as e: config.SANDBOX_READ_ONLY = True @@ -123,7 +175,7 @@ def main(): target = Prompt.ask("[bold yellow]No file name or project specified. Target file/folder name:[/bold yellow]") if target.strip(): task_description += f" (Target: {target.strip()})" - + orch.run_task(task_description, model_override=args.model) except Exception as e: console.print(f"[bold red]Error running task: {e}[/bold red]") @@ -137,16 +189,16 @@ def main(): try: model_indicator = f" ({model_override})" if model_override else " (auto)" prompt = Prompt.ask(f"[bold green]MACA{model_indicator} >[/bold green]") - + # Check empty prompt if not prompt.strip(): continue - + # Handle commands if prompt.strip().startswith("/"): parts = prompt.strip().split() cmd = parts[0].lower() - + if cmd in ["/exit", "/quit"]: console.print("[bold cyan]Goodbye![/bold cyan]") break @@ -157,30 +209,42 @@ def main(): elif cmd == "/model": if len(parts) > 1: val = parts[1].lower() - if val in ["gemma", "gemini"]: + if val in ["gemma", "gemini", "claude"]: model_override = val console.print(f"[bold green]Model override set to {val.upper()}[/bold green]") elif val == "auto": model_override = None console.print("[bold green]Model routing set to AUTO (based on complexity)[/bold green]") else: - console.print("[bold red]Invalid model. Options: gemma, gemini, auto[/bold red]") + console.print("[bold red]Invalid model. Options: gemma, gemini, claude, auto[/bold red]") else: - console.print("[bold red]Usage: /model [/bold red]") + console.print("[bold red]Usage: /model [/bold red]") else: console.print(f"[bold red]Unknown command: {cmd}[/bold red]") continue - + + # Check if it is a full command line pasted/typed in prompt + sub_task, sub_model, is_cmd = parse_interactive_command(prompt, parser) + + if is_cmd: + if not sub_task or not sub_task.strip(): + console.print("[bold red]Error: No task description provided in command line.[/bold red]") + continue + task_description = sub_task + current_model_override = sub_model if sub_model else model_override + else: + task_description = prompt + current_model_override = model_override + # Check if task description has file name/project folder - task_description = prompt if not contains_filename_or_project(task_description): target = Prompt.ask("[bold yellow]No file name or project specified. Target file/folder name:[/bold yellow]") if target.strip(): task_description += f" (Target: {target.strip()})" - + # Run the task - orch.run_task(task_description, model_override=model_override) - + orch.run_task(task_description, model_override=current_model_override) + except (KeyboardInterrupt, EOFError): console.print("\n[bold cyan]Goodbye![/bold cyan]") break @@ -189,4 +253,4 @@ def main(): console.print("[yellow]Continuing session...[/yellow]") if __name__ == "__main__": - main() + main() \ No newline at end of file diff --git a/src/maca/models/__init__.py b/src/maca/models/__init__.py index e69de29..8a571d1 100644 --- a/src/maca/models/__init__.py +++ b/src/maca/models/__init__.py @@ -0,0 +1,3 @@ +from maca.models.gemini import GeminiClient + +__all__ = ["GeminiClient"] diff --git a/src/maca/models/claude.py b/src/maca/models/claude.py new file mode 100644 index 0000000..32d1fa3 --- /dev/null +++ b/src/maca/models/claude.py @@ -0,0 +1,117 @@ +import json +import urllib.request +import urllib.error +from maca import maca_config as config + + +class ClaudeClient: + def __init__(self): + self.api_key = config.get_claude_api_key() + self.model = config.CLAUDE_MODEL + self.timeout = config.CLAUDE_TIMEOUT_SECONDS + + def generate(self, prompt, system_instruction=""): + if not self.api_key: + if config.MOCK_GEMMA_FALLBACK: + return self._generate_mock(prompt, system_instruction) + raise ValueError("CLAUDE_API_KEY is not set.") + + url = "https://api.anthropic.com/v1/messages" + + payload = { + "model": self.model, + "max_tokens": 4096, + "messages": [ + {"role": "user", "content": prompt} + ], + } + + if system_instruction: + payload["system"] = system_instruction + + headers = { + "Content-Type": "application/json", + "x-api-key": self.api_key, + "anthropic-version": "2023-06-01", + } + + # Setup secure SSL context using certifi if available + import ssl + try: + import certifi + ssl_context = ssl.create_default_context(cafile=certifi.where()) + except ImportError: + ssl_context = ssl.create_default_context() + + try: + req = urllib.request.Request( + url, + data=json.dumps(payload).encode("utf-8"), + headers=headers, + method="POST", + ) + with urllib.request.urlopen(req, timeout=self.timeout, context=ssl_context) as response: + res = json.loads(response.read().decode("utf-8")) + # Extract response text from the Messages API response + content_blocks = res.get("content", []) + if content_blocks: + # Concatenate all text blocks + texts = [ + block.get("text", "") + for block in content_blocks + if block.get("type") == "text" + ] + if texts: + return "\n".join(texts) + return "Error: Empty response from Claude." + except urllib.error.HTTPError as e: + error_body = e.read().decode("utf-8") + try: + err_json = json.loads(error_body) + msg = err_json.get("error", {}).get("message", error_body) + err_type = err_json.get("error", {}).get("type", "") + err_msg = f"{err_type}: {msg}" + except Exception: + err_msg = error_body + if config.MOCK_GEMMA_FALLBACK: + return self._generate_mock(prompt, system_instruction) + raise RuntimeError(f"Claude API request failed (HTTP {e.code}): {err_msg}") + except Exception as e: + if config.MOCK_GEMMA_FALLBACK: + return self._generate_mock(prompt, system_instruction) + raise RuntimeError(f"Claude API request failed: {e}") + + def _generate_mock(self, prompt, system_instruction): + prompt_lower = prompt.lower() + sys_lower = system_instruction.lower() + + if "coder" in sys_lower: + return """[FILE: output_claude.py] +```python +# Generated by Claude Agent (Mocked) +import urllib.request +import json + + +def fetch_url(url): + print(f"Fetching: {url}") + try: + with urllib.request.urlopen(url) as res: + return res.read().decode("utf-8") + except Exception as e: + return f"Error: {e}" + + +if __name__ == "__main__": + content = fetch_url("https://httpbin.org/get") + print(content[:200]) +```""" + elif "reviewer" in sys_lower: + return """# Code Review (Claude Mocked) +- Syntax checks: PASSED +APPROVED.""" + else: + return """# Implementation Plan (Claude Mocked) +1. [NEW] Create `output_claude.py` with advanced network utility logic. +2. Verify connection. +""" diff --git a/src/maca/orchestrator.py b/src/maca/orchestrator.py index f56c2ab..ac14d25 100644 --- a/src/maca/orchestrator.py +++ b/src/maca/orchestrator.py @@ -4,6 +4,7 @@ from maca.evaluator import ComplexityEvaluator from maca.models.local_gemma import LocalGemmaClient from maca.models.gemini import GeminiClient +from maca.models.claude import ClaudeClient # Import agents from maca.agents.planner import PlannerAgent @@ -18,11 +19,75 @@ def __init__(self, repo_path="."): self.repo_path = os.path.abspath(repo_path) self.evaluator = ComplexityEvaluator() self.conversation_history = [] - + + def _is_coder_done(self, client, task_description, plan, generated_files): + """Ask the model if the coder has completed all steps in the plan.""" + if config.MOCK_GEMMA_FALLBACK and not client.api_key: + return True, "Mock Coder finished." + + files_str = "" + for filepath, content in generated_files.items(): + files_str += f"--- FILE: {filepath} ---\n{content}\n\n" + + system_instruction = ( + "You are a Quality Assurance validator. Compare the implementation plan " + "with the generated files to see if all planned tasks/steps are fully completed. " + "Respond with 'YES' if everything is completely implemented. " + "Otherwise, respond with 'NO' followed by a detailed list of what is missing." + ) + prompt = ( + f"Task: {task_description}\n\n" + f"Plan:\n{plan}\n\n" + f"Generated Files:\n{files_str}\n\n" + "Are all steps in the plan completely implemented? (Start your response with YES or NO)" + ) + try: + response = client.generate(prompt, system_instruction).strip() + is_done = response.upper().startswith("YES") + return is_done, response + except Exception as e: + return True, f"Error validating completion: {e}" + + def _is_gemini_online(self): + key = config.get_gemini_api_key() + if not key: + return False + if config.MOCK_GEMMA_FALLBACK: + return True + try: + old_timeout = config.GEMINI_TIMEOUT + config.GEMINI_TIMEOUT = 3 + try: + client = GeminiClient() + res = client.generate("Reply with only the word OK.") + return bool(res) + finally: + config.GEMINI_TIMEOUT = old_timeout + except Exception: + return False + + def _is_claude_online(self): + key = config.get_claude_api_key() + if not key: + return False + if config.MOCK_GEMMA_FALLBACK: + return True + try: + old_timeout = config.CLAUDE_TIMEOUT + config.CLAUDE_TIMEOUT = 3 + try: + client = ClaudeClient() + res = client.generate("Reply with only the word OK.") + return bool(res) + finally: + config.CLAUDE_TIMEOUT = old_timeout + except Exception: + return False + def check_backends_status(self, run_handshakes=False): - # Checks status of Gemma (Ollama) and Gemini backends. + # Checks status of Gemma (Ollama), Gemini and Claude backends. status = {} - + # 1. Check Gemma gemma_url = config.OLLAMA_API_URL gemma_model = config.OLLAMA_MODEL @@ -56,9 +121,9 @@ def _detect_ollama(): return "OFFLINE (Mock Mode Fallback)" gemma_status = _detect_ollama() - + status["Gemma"] = gemma_status - + # 2. Check Gemini gemini_key = config.get_gemini_api_key() if not gemini_key: @@ -75,61 +140,112 @@ def _detect_ollama(): else: status["Gemini"] = f"ONLINE (Unexpected response: {res.strip()})" except Exception as e: - status["Gemini"] = f"CONNECTION FAILED: {str(e)[:80]}" + status["Gemini"] = f"CONNECTION FAILED: {str(e)}" finally: config.MOCK_GEMMA_FALLBACK = old_mock else: status["Gemini"] = "CONFIGURED (Key Present)" - + + # 3. Check Claude + claude_key = config.get_claude_api_key() + if not claude_key: + status["Claude"] = "UNCONFIGURED (Missing API Key)" + else: + if run_handshakes: + old_mock = config.MOCK_GEMMA_FALLBACK + config.MOCK_GEMMA_FALLBACK = False + try: + client = ClaudeClient() + res = client.generate("Reply with only the word OK.") + if "OK" in res.upper(): + status["Claude"] = "ONLINE (Connected)" + else: + status["Claude"] = f"ONLINE (Unexpected response: {res.strip()})" + except Exception as e: + status["Claude"] = f"CONNECTION FAILED: {str(e)}" + finally: + config.MOCK_GEMMA_FALLBACK = old_mock + else: + status["Claude"] = "CONFIGURED (Key Present)" + return status - + def run_task(self, task_description, model_override=None): console.print(Panel(f"[bold blue]Multi-Agent Coding Assistant[/bold blue]\n[bold white]Repo Path:[/bold white] {self.repo_path}\n[bold white]Task:[/bold white] {task_description}", border_style="blue")) - + # 1. Evaluate Complexity with console.status("[bold yellow]Evaluating task complexity...", spinner="dots"): complexity = self.evaluator.evaluate(task_description) - + console.print(f"[bold green]Task Complexity Evaluated:[/bold green] [bold cyan]{complexity}[/bold cyan]") - + # 2. Select Model Client model_name = "" client = None - + if model_override: model_name = model_override.upper() console.print(f"[bold yellow]Model override active:[/bold yellow] [bold cyan]{model_name}[/bold cyan]") + if model_name.startswith("GEMINI"): + client = GeminiClient() + config.validate_config(complexity, selected_agent="GEMINI") + elif model_name.startswith("CLAUDE"): + client = ClaudeClient() + config.validate_config(complexity, selected_agent="CLAUDE") + elif model_name.startswith("GEMMA"): + client = LocalGemmaClient() + config.validate_config(complexity) else: if complexity == "SIMPLE": model_name = "GEMMA (LOCAL)" client = LocalGemmaClient() else: - model_name = "GEMINI (REMOTE)" - client = GeminiClient() - - # Validate config for selected model - if model_name.startswith("GEMINI"): - client = GeminiClient() - config.validate_config(complexity) - elif model_name.startswith("GEMMA") and client is None: - client = LocalGemmaClient() + gemini_online = self._is_gemini_online() + claude_online = self._is_claude_online() + + if complexity == "MEDIUM": + if gemini_online: + model_name = "GEMINI (REMOTE)" + client = GeminiClient() + config.validate_config(complexity, selected_agent="GEMINI") + elif claude_online: + model_name = "CLAUDE (REMOTE)" + client = ClaudeClient() + config.validate_config(complexity, selected_agent="CLAUDE") + else: + model_name = "GEMMA (LOCAL)" + client = LocalGemmaClient() + config.validate_config(complexity) + else: # COMPLEX and VERY_COMPLEX + if claude_online: + model_name = "CLAUDE (REMOTE)" + client = ClaudeClient() + config.validate_config(complexity, selected_agent="CLAUDE") + elif gemini_online: + model_name = "GEMINI (REMOTE)" + client = GeminiClient() + config.validate_config(complexity, selected_agent="GEMINI") + else: + model_name = "GEMMA (LOCAL)" + client = LocalGemmaClient() + config.validate_config(complexity) console.print(f"[bold green]Selected Model Client:[/bold green] [bold cyan]{model_name}[/bold cyan]\n") # Get existing files in the repo planner = PlannerAgent(client) repo_files = planner.list_files(self.repo_path) - + # 3. Step 1: Planning Agent console.print(Panel("[bold yellow]Step 1: Planner Agent starting...[/bold yellow]", border_style="yellow")) with console.status("[bold yellow]Planner Agent is generating the implementation plan...", spinner="dots"): plan = planner.run(task_description, repo_files, history=self.conversation_history) - + console.print(Panel(Markdown(plan), title="[bold green]Implementation Plan[/bold green]", border_style="green")) - + # 4. Step 2: Coder Agent console.print(Panel("[bold yellow]Step 2: Coder Agent starting...[/bold yellow]", border_style="yellow")) - + # Read contents of files mentioned in plan to provide context to Coder if they exist repo_files_content = {} for filepath in repo_files: @@ -143,7 +259,7 @@ def run_task(self, task_description, model_override=None): with console.status("[bold yellow]Coder Agent is implementing the changes...", spinner="dots"): coder_response = coder.run(task_description, plan, repo_files_content, history=self.conversation_history) generated_files = coder.parse_files(coder_response) - + if not generated_files: console.print("[bold red]Warning: Coder did not output any files in the expected format [FILE: path]...[/bold red]") console.print("[yellow]Raw coder response structure check:[/yellow]") @@ -152,65 +268,132 @@ def run_task(self, task_description, model_override=None): console.print(f"[bold green]Coder generated {len(generated_files)} files:[/bold green]") for fp in generated_files.keys(): console.print(f" - [cyan]{fp}[/cyan]") - + + # Coder Verification Loop + max_nudge_attempts = 10 + for attempt in range(max_nudge_attempts): + console.print(f"[bold yellow]Checking if Coder completed all planned steps (Attempt {attempt + 1})...[/bold yellow]") + is_done, feedback = self._is_coder_done(client, task_description, plan, generated_files) + if is_done: + console.print("[bold green]Coder confirmed all planned tasks are complete![/bold green]") + break + else: + console.print(f"[bold red]Coder has NOT completed all steps. Feedback:[/bold red]\n{feedback}") + if attempt == max_nudge_attempts - 1: + console.print("[bold red]Reached maximum coder nudge attempts. Proceeding to review.[/bold red]") + break + + # Nudge the Coder to continue + nudge_prompt = ( + f"You have not completed all the steps in the plan. Here is the feedback on what is missing:\n\n" + f"{feedback}\n\n" + f"Please continue implementing the missing parts and output the complete updated files." + ) + console.print("[bold yellow]Nudging Coder Agent to finish the task...[/bold yellow]") + with console.status("[bold yellow]Coder Agent is continuing implementation...", spinner="dots"): + coder_response = coder.run( + task_description=task_description + f"\n\nNudge: {nudge_prompt}", + plan=plan, + repo_files_content={**repo_files_content, **generated_files}, + history=self.conversation_history + ) + updated_files = coder.parse_files(coder_response) + if updated_files: + for fp, content in updated_files.items(): + generated_files[fp] = content + # 5. Step 3: Reviewer Agent console.print(Panel("[bold yellow]Step 3: Reviewer Agent starting...[/bold yellow]", border_style="yellow")) reviewer = ReviewerAgent(client) - with console.status("[bold yellow]Reviewer Agent is auditing the generated code...", spinner="dots"): - reviewer_response = reviewer.run(task_description, generated_files, history=self.conversation_history) - reviewed_files = reviewer.parse_files(reviewer_response) - - console.print(Panel(Markdown(reviewer_response), title="[bold green]Reviewer Report[/bold green]", border_style="green")) - - # Merge changes from reviewer if any - if reviewed_files: - console.print("[bold yellow]Reviewer suggested corrections for the following files:[/bold yellow]") - for fp, content in reviewed_files.items(): - console.print(f" - [cyan]{fp}[/cyan] (Updated)") - generated_files[fp] = content - + + max_review_attempts = 10 + for r_attempt in range(max_review_attempts): + console.print(f"[bold yellow]Running Reviewer Agent (Attempt {r_attempt + 1})...[/bold yellow]") + with console.status("[bold yellow]Reviewer Agent is auditing the generated code...", spinner="dots"): + reviewer_response = reviewer.run(task_description, generated_files, history=self.conversation_history) + reviewed_files = reviewer.parse_files(reviewer_response) + + console.print(Panel(Markdown(reviewer_response), title=f"[bold green]Reviewer Report (Attempt {r_attempt + 1})[/bold green]", border_style="green")) + + is_approved = "APPROVED" in reviewer_response.upper() + + if is_approved: + if reviewed_files: + for fp, content in reviewed_files.items(): + generated_files[fp] = content + console.print("[bold green]Reviewer APPROVED the implementation![/bold green]") + break + else: + if r_attempt == max_review_attempts - 1: + console.print("[bold red]Reached maximum review attempts. Proceeding to write files.[/bold red]") + if reviewed_files: + for fp, content in reviewed_files.items(): + generated_files[fp] = content + break + + # Nudge the Coder to address Reviewer concerns + nudge_prompt = ( + f"The Reviewer has audited your code and raised issues / corrections. " + f"Here is the Reviewer's report:\n\n" + f"{reviewer_response}\n\n" + f"Please address all these issues and output the complete updated files." + ) + console.print("[bold yellow]Nudging Coder Agent to address Reviewer concerns...[/bold yellow]") + with console.status("[bold yellow]Coder Agent is applying corrections...", spinner="dots"): + coder_response = coder.run( + task_description=task_description + f"\n\nNudge: {nudge_prompt}", + plan=plan, + repo_files_content={**repo_files_content, **generated_files}, + history=self.conversation_history + ) + updated_files = coder.parse_files(coder_response) + if updated_files: + for fp, content in updated_files.items(): + generated_files[fp] = content + + # 6. Step 4: Writing Changes to Disk console.print(Panel("[bold yellow]Step 4: Writing files to repository...[/bold yellow]", border_style="yellow")) - + if config.SANDBOX_READ_ONLY: console.print("[bold red]Sandbox Protection Active: Current directory is read-only. Bypassing writes.[/bold red]") for rel_path, content in generated_files.items(): console.print(Panel(content, title=f"[cyan]File Preview: {rel_path}[/cyan] (Read-Only Mode)")) console.print(Panel("[bold yellow]MACA completed the run, but did not write to disk due to sandbox permissions.[/bold yellow]", border_style="yellow")) - + # Record dry run summary self.conversation_history.append(f"User Request: {task_description}") self.conversation_history.append(f"Planner Implementation Steps:\n{plan}") self.conversation_history.append("Files Created/Modified (Dry-Run Preview only): " + ", ".join(generated_files.keys())) self.conversation_history.append("Reviewer Decision: APPROVED") return - + table = Table(title="File Writing Summary") table.add_column("File Path", style="cyan") table.add_column("Action", style="green") table.add_column("Size (chars)", style="magenta") - + written_count = 0 for rel_path, content in generated_files.items(): # Clean leading/trailing spaces and leading slashes to prevent absolute path escapes clean_rel_path = rel_path.strip().lstrip("/") - + # Form absolute paths and verify they reside strictly inside self.repo_path (sandbox safety) repo_abs = os.path.abspath(self.repo_path) full_path = os.path.abspath(os.path.join(repo_abs, clean_rel_path)) - + try: common = os.path.commonpath([repo_abs, full_path]) is_safe = (common == repo_abs) except Exception: is_safe = False - + if not is_safe: table.add_row(rel_path, "[bold red]Failed: Sandbox escape blocked[/bold red]", "0") continue - + action = "Modified" if os.path.exists(full_path) else "Created" - + # Write file using standard agent call helper res = planner.write_file(full_path, content) if "Successfully" in res: @@ -218,20 +401,20 @@ def run_task(self, task_description, model_override=None): written_count += 1 else: table.add_row(clean_rel_path, f"[bold red]Failed: {res}[/bold red]", "0") - + if len(generated_files) > 0: console.print(table) - + if written_count > 0: console.print(f"[bold green]Successfully applied {written_count} changes to the repository![/bold green]") else: console.print("[bold red]No changes were applied to the repository.[/bold red]") - + console.print(Panel("[bold green]Coding Task Completed successfully![/bold green]", border_style="green")) - + # 7. Record to conversation history self.conversation_history.append(f"User Request: {task_description}") self.conversation_history.append(f"Planner Implementation Steps:\n{plan}") files_written = ", ".join(generated_files.keys()) if written_count > 0 else "None" self.conversation_history.append(f"Files Modified/Created: {files_written}") - self.conversation_history.append(f"Reviewer Decision: APPROVED") + self.conversation_history.append(f"Reviewer Decision: APPROVED") \ No newline at end of file diff --git a/tests/test_behavior.py b/tests/test_behavior.py index 2ab70e8..039fada 100644 --- a/tests/test_behavior.py +++ b/tests/test_behavior.py @@ -3,7 +3,9 @@ from maca.evaluator import ComplexityEvaluator from maca.models.gemini import GeminiClient +from maca.models.claude import ClaudeClient from maca.models.local_gemma import LocalGemmaClient +from maca.orchestrator import Orchestrator from maca import maca_config as config @@ -36,6 +38,18 @@ def test_gemini_client_returns_mock_response_without_api_key(self): self.assertIn("output_gemini.py", response) + def test_claude_client_returns_mock_response_without_api_key(self): + old_mock = config.MOCK_GEMMA_FALLBACK + config.MOCK_GEMMA_FALLBACK = True + try: + client = ClaudeClient() + with mock.patch.object(client, "api_key", ""): + response = client.generate("Create a utility module", "You are a coder") + finally: + config.MOCK_GEMMA_FALLBACK = old_mock + + self.assertIn("output_claude.py", response) + def test_local_gemma_client_falls_back_to_cli(self): client = LocalGemmaClient() old_mock = config.MOCK_GEMMA_FALLBACK @@ -68,6 +82,228 @@ def test_gemini_config_uses_maca_config(self): config.GEMINI_MODEL = old_model config.GEMINI_TIMEOUT_SECONDS = old_timeout + def test_claude_config_uses_maca_config(self): + old_model = config.CLAUDE_MODEL + old_timeout = config.CLAUDE_TIMEOUT_SECONDS + try: + config.CLAUDE_MODEL = "claude-test-model" + config.CLAUDE_TIMEOUT_SECONDS = 15.5 + client = ClaudeClient() + self.assertEqual(client.model, "claude-test-model") + self.assertEqual(client.timeout, 15.5) + finally: + config.CLAUDE_MODEL = old_model + config.CLAUDE_TIMEOUT_SECONDS = old_timeout + + def test_routing_matrix_medium_both_online(self): + orch = Orchestrator(".") + with mock.patch.object(orch, "_is_gemini_online", return_value=True), \ + mock.patch.object(orch, "_is_claude_online", return_value=True), \ + mock.patch.object(orch.evaluator, "evaluate", return_value="MEDIUM"), \ + mock.patch.object(config, "get_gemini_api_key", return_value="fake_key"), \ + mock.patch.object(config, "get_claude_api_key", return_value="fake_key"), \ + mock.patch("maca.orchestrator.PlannerAgent") as mock_planner: + + mock_planner.side_effect = RuntimeError("abort_task") + try: + orch.run_task("dummy task") + except RuntimeError as e: + if str(e) != "abort_task": + raise + + mock_planner.assert_called_once() + called_client = mock_planner.call_args[0][0] + self.assertIsInstance(called_client, GeminiClient) + + def test_routing_matrix_medium_only_claude_online(self): + orch = Orchestrator(".") + with mock.patch.object(orch, "_is_gemini_online", return_value=False), \ + mock.patch.object(orch, "_is_claude_online", return_value=True), \ + mock.patch.object(orch.evaluator, "evaluate", return_value="MEDIUM"), \ + mock.patch.object(config, "get_gemini_api_key", return_value=""), \ + mock.patch.object(config, "get_claude_api_key", return_value="fake_key"), \ + mock.patch("maca.orchestrator.PlannerAgent") as mock_planner: + + mock_planner.side_effect = RuntimeError("abort_task") + try: + orch.run_task("dummy task") + except RuntimeError as e: + if str(e) != "abort_task": + raise + + mock_planner.assert_called_once() + called_client = mock_planner.call_args[0][0] + self.assertIsInstance(called_client, ClaudeClient) + + def test_routing_matrix_complex_both_online(self): + orch = Orchestrator(".") + with mock.patch.object(orch, "_is_gemini_online", return_value=True), \ + mock.patch.object(orch, "_is_claude_online", return_value=True), \ + mock.patch.object(orch.evaluator, "evaluate", return_value="COMPLEX"), \ + mock.patch.object(config, "get_gemini_api_key", return_value="fake_key"), \ + mock.patch.object(config, "get_claude_api_key", return_value="fake_key"), \ + mock.patch("maca.orchestrator.PlannerAgent") as mock_planner: + + mock_planner.side_effect = RuntimeError("abort_task") + try: + orch.run_task("dummy task") + except RuntimeError as e: + if str(e) != "abort_task": + raise + + mock_planner.assert_called_once() + called_client = mock_planner.call_args[0][0] + self.assertIsInstance(called_client, ClaudeClient) + + def test_routing_matrix_complex_only_gemini_online(self): + orch = Orchestrator(".") + with mock.patch.object(orch, "_is_gemini_online", return_value=True), \ + mock.patch.object(orch, "_is_claude_online", return_value=False), \ + mock.patch.object(orch.evaluator, "evaluate", return_value="COMPLEX"), \ + mock.patch.object(config, "get_gemini_api_key", return_value="fake_key"), \ + mock.patch.object(config, "get_claude_api_key", return_value=""), \ + mock.patch("maca.orchestrator.PlannerAgent") as mock_planner: + + mock_planner.side_effect = RuntimeError("abort_task") + try: + orch.run_task("dummy task") + except RuntimeError as e: + if str(e) != "abort_task": + raise + + mock_planner.assert_called_once() + called_client = mock_planner.call_args[0][0] + self.assertIsInstance(called_client, GeminiClient) + + def test_coder_completion_verification_loop(self): + orch = Orchestrator(".") + + is_done_mock = mock.Mock() + is_done_mock.side_effect = [ + (False, "Missing step 2 implementation"), + (True, "All steps completed successfully") + ] + + mock_plan = "1. Step one\n2. Step two" + + coder_run_mock = mock.Mock() + coder_run_mock.side_effect = [ + "Generated content for step 1", + "Generated content for step 1 and 2" + ] + + reviewer_run_mock = mock.Mock() + reviewer_run_mock.return_value = "APPROVED" + + with mock.patch.object(orch, "_is_gemini_online", return_value=True), \ + mock.patch.object(orch, "_is_claude_online", return_value=False), \ + mock.patch.object(orch.evaluator, "evaluate", return_value="MEDIUM"), \ + mock.patch.object(config, "get_gemini_api_key", return_value="fake_key"), \ + mock.patch.object(config, "SANDBOX_READ_ONLY", True), \ + mock.patch("maca.orchestrator.PlannerAgent") as mock_planner_cls, \ + mock.patch("maca.orchestrator.CoderAgent") as mock_coder_cls, \ + mock.patch("maca.orchestrator.ReviewerAgent") as mock_reviewer_cls, \ + mock.patch.object(orch, "_is_coder_done", is_done_mock): + + planner_inst = mock_planner_cls.return_value + planner_inst.list_files.return_value = [] + planner_inst.run.return_value = mock_plan + + coder_inst = mock_coder_cls.return_value + coder_inst.run = coder_run_mock + coder_inst.parse_files.side_effect = [ + {"file1.py": "content1"}, + {"file1.py": "content1_updated"} + ] + + reviewer_inst = mock_reviewer_cls.return_value + reviewer_inst.run = reviewer_run_mock + reviewer_inst.parse_files.return_value = {} + + orch.run_task("Implement task") + + self.assertEqual(coder_run_mock.call_count, 2) + + second_call_args = coder_run_mock.call_args_list[1] + self.assertIn("Nudge: You have not completed all the steps in the plan", second_call_args[1]["task_description"]) + self.assertIn("Missing step 2 implementation", second_call_args[1]["task_description"]) + + reviewer_run_mock.assert_called_once() + + def test_reviewer_rejection_nudge_loop(self): + orch = Orchestrator(".") + + mock_plan = "1. Step one" + + coder_run_mock = mock.Mock() + coder_run_mock.side_effect = [ + "Initial coder response", + "Corrected coder response" + ] + + reviewer_run_mock = mock.Mock() + reviewer_run_mock.side_effect = [ + "Issues found: missing docstring. REJECTED.", + "Looks perfect. APPROVED." + ] + + with mock.patch.object(orch, "_is_gemini_online", return_value=True), \ + mock.patch.object(orch, "_is_claude_online", return_value=False), \ + mock.patch.object(orch.evaluator, "evaluate", return_value="MEDIUM"), \ + mock.patch.object(config, "get_gemini_api_key", return_value="fake_key"), \ + mock.patch.object(config, "SANDBOX_READ_ONLY", True), \ + mock.patch("maca.orchestrator.PlannerAgent") as mock_planner_cls, \ + mock.patch("maca.orchestrator.CoderAgent") as mock_coder_cls, \ + mock.patch("maca.orchestrator.ReviewerAgent") as mock_reviewer_cls, \ + mock.patch.object(orch, "_is_coder_done", return_value=(True, "Done")): + + planner_inst = mock_planner_cls.return_value + planner_inst.list_files.return_value = [] + planner_inst.run.return_value = mock_plan + + coder_inst = mock_coder_cls.return_value + coder_inst.run = coder_run_mock + coder_inst.parse_files.side_effect = [ + {"file1.py": "content1"}, + {"file1.py": "content1_updated"} + ] + + reviewer_inst = mock_reviewer_cls.return_value + reviewer_inst.run = reviewer_run_mock + reviewer_inst.parse_files.return_value = {} + + orch.run_task("Implement task") + + self.assertEqual(coder_run_mock.call_count, 2) + self.assertEqual(reviewer_run_mock.call_count, 2) + + second_call_args = coder_run_mock.call_args_list[1] + self.assertIn("Nudge: The Reviewer has audited your code and raised issues", second_call_args[1]["task_description"]) + self.assertIn("missing docstring. REJECTED.", second_call_args[1]["task_description"]) + + def test_interactive_command_line_parsing(self): + from maca.main import parse_interactive_command + import argparse + + parser = argparse.ArgumentParser() + parser.add_argument("task", nargs="?", default=None) + parser.add_argument("--repo", default=".") + parser.add_argument("--model", default=None) + parser.add_argument("--mock", action="store_true") + + task, model, is_cmd = parse_interactive_command('maca --model claude "implement a custom tokenizer"', parser) + self.assertTrue(is_cmd) + self.assertEqual(task, "implement a custom tokenizer") + self.assertEqual(model, "claude") + + task, model, is_cmd = parse_interactive_command('python3 src/maca/main.py --model gemini "do something"', parser) + self.assertTrue(is_cmd) + self.assertEqual(task, "do something") + self.assertEqual(model, "gemini") + + task, model, is_cmd = parse_interactive_command('implement a custom tokenizer', parser) + self.assertFalse(is_cmd) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_orchestrator_status.py b/tests/test_orchestrator_status.py index 3ceccf5..b1c0b2f 100644 --- a/tests/test_orchestrator_status.py +++ b/tests/test_orchestrator_status.py @@ -2,6 +2,7 @@ from unittest import mock from maca.orchestrator import Orchestrator +from maca import maca_config as config class OrchestratorStatusTests(unittest.TestCase): @@ -10,7 +11,7 @@ def test_fast_status_uses_ollama_http_when_available(self): fake_response = mock.MagicMock() fake_response.__enter__.return_value.status = 200 - fake_response.__enter__.return_value.read.return_value = b'{}' + fake_response.__enter__.return_value.read.return_value = b"{}" with mock.patch("urllib.request.urlopen", return_value=fake_response) as urlopen_mock, \ mock.patch("subprocess.run", side_effect=FileNotFoundError("no cli")): @@ -19,6 +20,18 @@ def test_fast_status_uses_ollama_http_when_available(self): self.assertEqual(status["Gemma"], "ONLINE (Ollama HTTP - gemma2:2b)") urlopen_mock.assert_called_once() + def test_status_shows_claude_unconfigured(self): + orch = Orchestrator(".") + with mock.patch.object(config, "get_claude_api_key", return_value=""): + status = orch.check_backends_status(run_handshakes=False) + self.assertEqual(status["Claude"], "UNCONFIGURED (Missing API Key)") + + def test_status_shows_claude_configured(self): + orch = Orchestrator(".") + with mock.patch.object(config, "get_claude_api_key", return_value="fake_key"): + status = orch.check_backends_status(run_handshakes=False) + self.assertEqual(status["Claude"], "CONFIGURED (Key Present)") + if __name__ == "__main__": unittest.main()