diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index e2a5dda..177bf63 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -18,7 +18,12 @@ jobs: python -m pip install --upgrade pip python -m pip install hatchling python -m pip install -e . - python -m pip install -r requirements.txt + python -m pip install -r requirements-dev.txt + - name: Run static analysis + run: | + ruff check src/ + mypy src/ + bandit -c pyproject.toml -r src/ - name: Run tests run: | PYTHONPATH=src python -m unittest discover -s tests -v diff --git a/.gitignore b/.gitignore index 1f1e1a7..508a6bb 100644 --- a/.gitignore +++ b/.gitignore @@ -4,3 +4,4 @@ target/ build/ __pycache__/ .venv/ +output/ \ No newline at end of file diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml new file mode 100644 index 0000000..37b0de2 --- /dev/null +++ b/.pre-commit-config.yaml @@ -0,0 +1,7 @@ +repos: + - repo: https://github.com/astral-sh/ruff-pre-commit + rev: v0.4.5 + hooks: + - id: ruff + args: [ --fix ] + - id: ruff-format diff --git a/README.md b/README.md index b814f4e..eb6eeb6 100644 --- a/README.md +++ b/README.md @@ -109,6 +109,29 @@ MACA uses Google Gemini and Anthropic Claude for medium and above tasks. For sec PYTHONPATH=src python -m unittest discover -s tests -v ``` +### 4. Development & Static Analysis +We enforce code quality and security checks on both local builds and CI pipelines: +- **Ruff**: Code linting and formatting (auto-runs on commit). +- **Mypy**: Strict type-checking. +- **Bandit**: Security scanning for potential vulnerabilities. + +The `./local/scripts/install_mac.sh` launcher script handles setting these up automatically. It will: +1. Install all dependencies from `requirements-dev.txt`. +2. Install **pre-commit** hooks (`pre-commit install`) that auto-fix code style issues on commit. +3. Run Ruff, Mypy, and Bandit audits locally. If any check fails, the build script exits with a non-zero exit code. + +You can also run analysis manually inside your virtual environment: +```sh +# Run linter checks +ruff check src/ + +# Run type checker +mypy src/ + +# Run security checks +bandit -c pyproject.toml -r src/ +``` + --- ## 💻 How to Use diff --git a/local/scripts/install_mac.sh b/local/scripts/install_mac.sh index 5fd53e6..fb10833 100755 --- a/local/scripts/install_mac.sh +++ b/local/scripts/install_mac.sh @@ -15,6 +15,28 @@ python -m pip install --upgrade pip setuptools wheel python -m pip install hatchling python -m pip install -e . +echo "Installing dev dependencies..." +python -m pip install -r requirements-dev.txt + +echo "Setting up pre-commit hooks..." +pre-commit install + +echo "Running static analysis (Ruff, Mypy & Bandit)..." +if ! ruff check src/; then + echo "Build failed: Ruff linting errors found." >&2 + exit 1 +fi + +if ! mypy src/; then + echo "Build failed: Mypy type-checking errors found." >&2 + exit 1 +fi + +if ! bandit -c pyproject.toml -r src/; then + echo "Build failed: Bandit security issues found." >&2 + exit 1 +fi + if [ -x "$ROOT_DIR/local/scripts/setup_gemma.sh" ]; then "$ROOT_DIR/local/scripts/setup_gemma.sh" fi diff --git a/pyproject.toml b/pyproject.toml index e4674dc..0c74ff5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -19,3 +19,30 @@ maca = "maca.main:main" [tool.hatch.build.targets.wheel] packages = ["src/maca"] + +[tool.ruff] +line-length = 100 +target-version = "py39" + +[tool.ruff.lint] +select = ["E", "F", "I"] +ignore = ["E501"] + +[tool.ruff.format] +quote-style = "double" + +[tool.mypy] +python_version = "3.9" +warn_return_any = true +warn_unused_configs = true +disallow_untyped_defs = false +disallow_incomplete_defs = false +check_untyped_defs = true + +[[tool.mypy.overrides]] +module = "rich.*" +ignore_missing_imports = true + +[tool.bandit] +exclude_dirs = ["tests", ".venv"] +skips = ["B101", "B110", "B310", "B404", "B603", "B607"] diff --git a/requirements-dev.txt b/requirements-dev.txt new file mode 100644 index 0000000..8629604 --- /dev/null +++ b/requirements-dev.txt @@ -0,0 +1,15 @@ +# Development & testing dependencies for MACA. +# +# Install with: +# python -m pip install -r requirements-dev.txt +# +# These mirror the optional "dev" extra in pyproject.toml. + +-r requirements.txt + +pytest>=7.0.0 +ruff>=0.1.0 +black>=23.0.0 +mypy>=1.0.0 +bandit>=1.7.0 +pre-commit>=3.5.0 diff --git a/setup.py b/setup.py index 2da9ebf..1280f63 100644 --- a/setup.py +++ b/setup.py @@ -1,4 +1,4 @@ -from setuptools import setup, find_packages +from setuptools import find_packages, setup setup( name="maca-ai", diff --git a/src/maca/agents/base.py b/src/maca/agents/base.py index 2f406cf..3fe1d50 100644 --- a/src/maca/agents/base.py +++ b/src/maca/agents/base.py @@ -1,12 +1,13 @@ import os import re + class BaseAgent: def __init__(self, name, model_client): self.name = name self.model_client = model_client - def run(self, prompt, **kwargs): + def run(self, *args, **kwargs): raise NotImplementedError("Subclasses must implement run()") # Helper tools available to the orchestrator/agents @@ -46,7 +47,7 @@ def clean_code_content(self, content): if first_line_end != -1: line_content = content[:first_line_end].strip() if line_content.startswith("```"): - content = content[first_line_end + 1:].strip() + content = content[first_line_end + 1 :].strip() cleaned = True if content.endswith("```"): content = content[:-3].strip() @@ -58,7 +59,7 @@ def clean_code_content(self, content): def parse_files(self, response_text): pattern = r"\[FILE:\s*([^\s\]]+)\]\s*(?:\r?\n)*```\w*\s*\n(.*?)\n```" matches = re.findall(pattern, response_text, re.DOTALL) - + files = {} for filepath, content in matches: files[filepath.strip()] = self.clean_code_content(content) diff --git a/src/maca/agents/coder.py b/src/maca/agents/coder.py index 3466338..8ed6e08 100644 --- a/src/maca/agents/coder.py +++ b/src/maca/agents/coder.py @@ -1,6 +1,6 @@ -import re from maca.agents.base import BaseAgent + class CoderAgent(BaseAgent): def __init__(self, name, model_client): super().__init__(name, model_client) @@ -27,7 +27,7 @@ def _build_system_instruction(self): def _build_prompt(self, task_description, plan, repo_files_content, history): history_context = self._format_history(history) files_context = self._format_files_context(repo_files_content) - + return ( f"User Task: {task_description}{history_context}\n\n" f"Implementation Plan:\n{plan}\n" @@ -43,7 +43,7 @@ def _format_history(self, history): def _format_files_context(self, repo_files_content): if not repo_files_content: return "" - + formatted_files = "\n\nExisting File Contents:\n" for filepath, content in repo_files_content.items(): formatted_files += f"--- FILE: {filepath} ---\n{content}\n\n" diff --git a/src/maca/agents/planner.py b/src/maca/agents/planner.py index 2f157a5..1ecafe7 100644 --- a/src/maca/agents/planner.py +++ b/src/maca/agents/planner.py @@ -1,28 +1,29 @@ from maca.agents.base import BaseAgent + class PlannerAgent(BaseAgent): def __init__(self, model_client): super().__init__("Planner", model_client) def run(self, task_description, repo_files=None, history=None): files_str = "\n".join(repo_files) if repo_files else "Empty repository" - + history_str = "" if history: history_str = "\n\nPrevious Conversation History:\n" + "\n".join(history) - + system_instruction = ( "You are a technical Planner Agent. Your job is to analyze the user request " "and create a structured markdown implementation plan. " "Do NOT write any code implementation or scripts. Only write the steps and " "identify which files need to be created or modified." ) - + prompt = ( f"User Task: {task_description}{history_str}\n\n" f"Current Files in Repository:\n{files_str}\n\n" "Please output a detailed implementation plan in markdown format. " "Clearly indicate the files to be created or modified using [NEW] and [MODIFY] tags." ) - + return self.model_client.generate(prompt, system_instruction) diff --git a/src/maca/agents/reviewer.py b/src/maca/agents/reviewer.py index d560be5..05a136d 100644 --- a/src/maca/agents/reviewer.py +++ b/src/maca/agents/reviewer.py @@ -1,6 +1,6 @@ -import re from maca.agents.base import BaseAgent + class ReviewerAgent(BaseAgent): def __init__(self, model_client): super().__init__("Reviewer", model_client) @@ -29,7 +29,7 @@ def _build_system_instruction(self): def _build_prompt(self, task_description, generated_files, history): history_context = self._format_history(history) files_context = self._format_files_context(generated_files) - + return ( f"User Task: {task_description}{history_context}\n\n" f"Generated Files to Review:\n{files_context}" @@ -44,7 +44,7 @@ def _format_history(self, history): def _format_files_context(self, files): if not files: return "" - + formatted_files = "" for filepath, content in files.items(): formatted_files += f"--- FILE: {filepath} ---\n{content}\n\n" diff --git a/src/maca/evaluator.py b/src/maca/evaluator.py index 41139f0..e1b5186 100644 --- a/src/maca/evaluator.py +++ b/src/maca/evaluator.py @@ -1,6 +1,6 @@ -import re from maca.models.local_gemma import LocalGemmaClient + class ComplexityEvaluator: def __init__(self): self.gemma_client = LocalGemmaClient() @@ -23,7 +23,7 @@ def evaluate(self, prompt): "Classification (SIMPLE, MEDIUM, COMPLEX, or VERY_COMPLEX):" ) response = self.gemma_client.generate(gemma_prompt, system_instruction).strip().upper() - + # Extract the keyword from response for val in ["VERY_COMPLEX", "COMPLEX", "MEDIUM", "SIMPLE"]: if val in response: @@ -36,25 +36,55 @@ def evaluate(self, prompt): def _heuristic_evaluate(self, prompt): prompt_lower = prompt.lower() - + # Very Complex Indicators very_complex_keywords = [ - "distributed", "concurrency", "multi-thread", "thread-safe", "acid", "replicat", - "ast parser", "lexer", "compiler", "interpreter", "cryptography", "blockchain", - "transaction log", "race condition", "consensus", "raft", "paxos", "from scratch" + "distributed", + "concurrency", + "multi-thread", + "thread-safe", + "acid", + "replicat", + "ast parser", + "lexer", + "compiler", + "interpreter", + "cryptography", + "blockchain", + "transaction log", + "race condition", + "consensus", + "raft", + "paxos", + "from scratch", ] if any(kw in prompt_lower for kw in very_complex_keywords): return "VERY_COMPLEX" - + # Complex Indicators complex_keywords = [ - "database", "sql", "api", "scraper", "beautifulsoup", "selenium", - "web server", "flask", "django", "fastapi", "express", "multi-file", - "integration", "docker", "pipeline", "regex", "pandas", "matplotlib" + "database", + "sql", + "api", + "scraper", + "beautifulsoup", + "selenium", + "web server", + "flask", + "django", + "fastapi", + "express", + "multi-file", + "integration", + "docker", + "pipeline", + "regex", + "pandas", + "matplotlib", ] if any(kw in prompt_lower for kw in complex_keywords) or len(prompt.split()) > 40: return "COMPLEX" - + # Default to SIMPLE or MEDIUM if len(prompt.split()) > 15: return "MEDIUM" diff --git a/src/maca/maca_config.py b/src/maca/maca_config.py index 40ac016..8d79d9a 100644 --- a/src/maca/maca_config.py +++ b/src/maca/maca_config.py @@ -2,6 +2,7 @@ _gemini_key = None + def get_gemini_api_key(): global _gemini_key if _gemini_key is not None: @@ -9,8 +10,10 @@ def 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: @@ -18,6 +21,7 @@ def get_claude_api_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") @@ -44,6 +48,7 @@ def get_claude_api_key(): # 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. @@ -76,4 +81,5 @@ def validate_config(complexity, selected_agent=None): "Gemini is required for Medium, Complex, and Very Complex tasks." ) + SANDBOX_READ_ONLY = False diff --git a/src/maca/main.py b/src/maca/main.py index f445f86..c08e700 100644 --- a/src/maca/main.py +++ b/src/maca/main.py @@ -1,18 +1,19 @@ +import argparse import os import sys -import argparse # Inject package root directory to allow standalone execution package_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) if package_root not in sys.path: sys.path.append(package_root) -from maca.rich_compat import Console, Panel, Prompt, Table -from maca import maca_config as config -from maca.orchestrator import Orchestrator +from maca import maca_config as config # noqa: E402 +from maca.orchestrator import Orchestrator # noqa: E402 +from maca.rich_compat import Console, Panel, Prompt, Table # noqa: E402 console = Console() + def display_welcome(orch=None): welcome_text = ( "[bold cyan]Welcome to the Multi-Agent Coding Assistant (MACA)![/bold cyan]\n" @@ -26,14 +27,29 @@ def display_welcome(orch=None): " - [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")) + 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) + 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: + 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") + ): status_dict = orch.check_backends_status(run_handshakes=run_handshakes) table = Table(title=title) @@ -53,16 +69,37 @@ def print_backends_status(orch, run_handshakes=False): 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"] + 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("!", " ") + 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): @@ -70,11 +107,13 @@ def contains_filename_or_project(prompt): 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 @@ -95,7 +134,9 @@ def parse_interactive_command(prompt, parser): # 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]): + 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 @@ -116,11 +157,17 @@ def parse_interactive_command(prompt, parser): 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", "claude"], 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() @@ -132,9 +179,9 @@ def main(): # 2. Confirm Access try: confirm = Prompt.ask( - f"[bold yellow]Do you grant MACA permission to access and modify files in this path? (y/n)[/bold yellow]", + "[bold yellow]Do you grant MACA permission to access and modify files in this path? (y/n)[/bold yellow]", choices=["y", "n"], - default="y" + default="y", ) if confirm.lower() != "y": console.print("[bold red]Access denied by user. Safe exit.[/bold red]") @@ -153,7 +200,9 @@ def main(): f.write("test") os.remove(test_file) - console.print("[bold green]✓ Sandbox check: Read/Write access verified successfully.[/bold green]\n") + console.print( + "[bold green]✓ Sandbox check: Read/Write access verified successfully.[/bold green]\n" + ) except Exception as e: config.SANDBOX_READ_ONLY = True console.print( @@ -163,7 +212,9 @@ def main(): if args.mock: config.MOCK_GEMMA_FALLBACK = True - console.print("[bold yellow]Mock mode forced: Gemma/Gemini will run in simulation mode.[/bold yellow]\n") + console.print( + "[bold yellow]Mock mode forced: Gemma/Gemini will run in simulation mode.[/bold yellow]\n" + ) orch = Orchestrator(args.repo) @@ -172,7 +223,9 @@ def main(): try: task_description = args.task 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]") + 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()})" @@ -211,14 +264,22 @@ def main(): val = parts[1].lower() if val in ["gemma", "gemini", "claude"]: model_override = val - console.print(f"[bold green]Model override set to {val.upper()}[/bold green]") + 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]") + console.print( + "[bold green]Model routing set to AUTO (based on complexity)[/bold green]" + ) else: - console.print("[bold red]Invalid model. Options: gemma, gemini, claude, 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 @@ -228,7 +289,9 @@ def main(): 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]") + 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 @@ -238,7 +301,9 @@ def main(): # Check if task description has file name/project folder 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]") + 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()})" @@ -252,5 +317,6 @@ def main(): console.print(f"[bold red]Error processing request: {e}[/bold red]") console.print("[yellow]Continuing session...[/yellow]") + if __name__ == "__main__": - main() \ No newline at end of file + main() diff --git a/src/maca/models/claude.py b/src/maca/models/claude.py index 32d1fa3..6de258b 100644 --- a/src/maca/models/claude.py +++ b/src/maca/models/claude.py @@ -1,6 +1,7 @@ import json -import urllib.request import urllib.error +import urllib.request + from maca import maca_config as config @@ -21,9 +22,7 @@ def generate(self, prompt, system_instruction=""): payload = { "model": self.model, "max_tokens": 4096, - "messages": [ - {"role": "user", "content": prompt} - ], + "messages": [{"role": "user", "content": prompt}], } if system_instruction: @@ -37,8 +36,10 @@ def generate(self, prompt, system_instruction=""): # 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() @@ -82,7 +83,7 @@ def generate(self, prompt, system_instruction=""): raise RuntimeError(f"Claude API request failed: {e}") def _generate_mock(self, prompt, system_instruction): - prompt_lower = prompt.lower() + # prompt_lower = prompt.lower() sys_lower = system_instruction.lower() if "coder" in sys_lower: diff --git a/src/maca/models/gemini.py b/src/maca/models/gemini.py index 59b5f9c..0b3ec1d 100644 --- a/src/maca/models/gemini.py +++ b/src/maca/models/gemini.py @@ -1,8 +1,11 @@ import json -import urllib.request import urllib.error +import urllib.request +from typing import Any + from maca import maca_config as config + class GeminiClient: def __init__(self): self.api_key = config.get_gemini_api_key() @@ -16,22 +19,18 @@ def generate(self, prompt, system_instruction=""): raise ValueError("GEMINI_API_KEY is not set.") url = f"https://generativelanguage.googleapis.com/v1beta/models/{self.model}:generateContent?key={self.api_key}" - - payload = { - "contents": [{ - "parts": [{"text": prompt}] - }] - } - + + payload: dict[str, Any] = {"contents": [{"parts": [{"text": prompt}]}]} + if system_instruction: - payload["systemInstruction"] = { - "parts": [{"text": system_instruction}] - } + payload["systemInstruction"] = {"parts": [{"text": system_instruction}]} # 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() @@ -41,7 +40,7 @@ def generate(self, prompt, system_instruction=""): url, data=json.dumps(payload).encode("utf-8"), headers={"Content-Type": "application/json"}, - method="POST" + method="POST", ) with urllib.request.urlopen(req, timeout=self.timeout, context=ssl_context) as response: res = json.loads(response.read().decode("utf-8")) @@ -63,15 +62,16 @@ def generate(self, prompt, system_instruction=""): raise RuntimeError(f"Gemini API request failed: {e}") def _generate_mock(self, prompt, system_instruction): - prompt_lower = prompt.lower() + # prompt_lower = prompt.lower() sys_lower = system_instruction.lower() - + if "coder" in sys_lower: return """[FILE: output_gemini.py] ```python # Generated by Gemini Agent (Mocked) import urllib.request import json +from typing import Any def fetch_url(url): print(f"Fetching: {url}") diff --git a/src/maca/models/local_gemma.py b/src/maca/models/local_gemma.py index 15b9d85..78c65cb 100644 --- a/src/maca/models/local_gemma.py +++ b/src/maca/models/local_gemma.py @@ -1,8 +1,10 @@ import json -import urllib.request import urllib.error +import urllib.request + from maca import maca_config as config + class LocalGemmaClient: def __init__(self): self.url = config.OLLAMA_API_URL @@ -10,6 +12,7 @@ def __init__(self): def generate(self, prompt, system_instruction=""): import subprocess + # Combine system instruction and prompt full_prompt = prompt if system_instruction: @@ -19,13 +22,11 @@ def generate(self, prompt, system_instruction=""): try: req = urllib.request.Request( f"{self.url}/api/generate", - data=json.dumps({ - "model": self.model, - "prompt": full_prompt, - "stream": False - }).encode("utf-8"), + data=json.dumps( + {"model": self.model, "prompt": full_prompt, "stream": False} + ).encode("utf-8"), headers={"Content-Type": "application/json"}, - method="POST" + method="POST", ) with urllib.request.urlopen(req, timeout=10) as response: res = json.loads(response.read().decode("utf-8")) @@ -40,25 +41,25 @@ def generate(self, prompt, system_instruction=""): input=full_prompt, capture_output=True, text=True, - timeout=30 + timeout=30, ) if res.returncode == 0: return res.stdout.strip() except FileNotFoundError: pass - except Exception as sub_e: + except Exception: pass # 3. Try Mock Fallback if config.MOCK_GEMMA_FALLBACK: return self._generate_mock(prompt, system_instruction) - + raise RuntimeError(f"Failed to connect to local Gemma via Ollama API and CLI: {e}") def _generate_mock(self, prompt, system_instruction): - prompt_lower = prompt.lower() + # prompt_lower = prompt.lower() sys_lower = system_instruction.lower() - + if "coder" in sys_lower: return """[FILE: output_gemma.py] ```python diff --git a/src/maca/orchestrator.py b/src/maca/orchestrator.py index ac14d25..3f7be87 100644 --- a/src/maca/orchestrator.py +++ b/src/maca/orchestrator.py @@ -1,19 +1,21 @@ import os -import sys -from maca.rich_compat import Console, Panel, Markdown, Table -from maca.evaluator import ComplexityEvaluator -from maca.models.local_gemma import LocalGemmaClient -from maca.models.gemini import GeminiClient -from maca.models.claude import ClaudeClient +from typing import Any + +from maca import maca_config as config +from maca.agents.coder import CoderAgent # Import agents from maca.agents.planner import PlannerAgent -from maca.agents.coder import CoderAgent from maca.agents.reviewer import ReviewerAgent -from maca import maca_config as config +from maca.evaluator import ComplexityEvaluator +from maca.models.claude import ClaudeClient +from maca.models.gemini import GeminiClient +from maca.models.local_gemma import LocalGemmaClient +from maca.rich_compat import Console, Markdown, Panel, Table console = Console() + class Orchestrator: def __init__(self, repo_path="."): self.repo_path = os.path.abspath(repo_path) @@ -87,6 +89,7 @@ def _is_claude_online(self): def check_backends_status(self, run_handshakes=False): # Checks status of Gemma (Ollama), Gemini and Claude backends. status = {} + client: Any = None # 1. Check Gemma gemma_url = config.OLLAMA_API_URL @@ -95,6 +98,7 @@ def check_backends_status(self, run_handshakes=False): def _detect_ollama(): import urllib.request + try: req = urllib.request.Request(f"{gemma_url}/api/tags", method="GET") with urllib.request.urlopen(req, timeout=2) as res: @@ -104,6 +108,7 @@ def _detect_ollama(): pass import subprocess + try: res = subprocess.run(["ollama", "list"], capture_output=True, text=True, timeout=2) if res.returncode == 0 and res.stdout.strip(): @@ -112,7 +117,9 @@ def _detect_ollama(): pass try: - res = subprocess.run(["pgrep", "-af", "ollama"], capture_output=True, text=True, timeout=2) + res = subprocess.run( + ["pgrep", "-af", "ollama"], capture_output=True, text=True, timeout=2 + ) if res.returncode == 0 and res.stdout.strip(): return "ONLINE (Ollama Running - {0})".format(gemma_model) except Exception: @@ -171,21 +178,30 @@ def _detect_ollama(): 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")) + 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]") + console.print( + f"[bold green]Task Complexity Evaluated:[/bold green] [bold cyan]{complexity}[/bold cyan]" + ) # 2. Select Model Client model_name = "" - client = None + client: Any = 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]") + 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") @@ -230,21 +246,40 @@ def run_task(self, task_description, model_override=None): client = LocalGemmaClient() config.validate_config(complexity) - console.print(f"[bold green]Selected Model Client:[/bold green] [bold cyan]{model_name}[/bold cyan]\n") + 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"): + 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")) + 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")) + 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 = {} @@ -256,12 +291,18 @@ def run_task(self, task_description, model_override=None): repo_files_content[filepath] = planner.read_file(full_path) coder = CoderAgent("Coder", client) - 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) + 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( + "[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]") console.print(coder_response[:500] + "...") else: @@ -272,15 +313,23 @@ def run_task(self, task_description, model_override=None): # 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]") + 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]") + 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}") + 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]") + console.print( + "[bold red]Reached maximum coder nudge attempts. Proceeding to review.[/bold red]" + ) break # Nudge the Coder to continue @@ -289,13 +338,17 @@ def run_task(self, task_description, model_override=None): 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"): + 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 + history=self.conversation_history, ) updated_files = coder.parse_files(coder_response) if updated_files: @@ -303,17 +356,34 @@ def run_task(self, task_description, model_override=None): generated_files[fp] = content # 5. Step 3: Reviewer Agent - console.print(Panel("[bold yellow]Step 3: Reviewer Agent starting...[/bold yellow]", border_style="yellow")) + console.print( + Panel( + "[bold yellow]Step 3: Reviewer Agent starting...[/bold yellow]", + border_style="yellow", + ) + ) reviewer = ReviewerAgent(client) 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) + 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")) + 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() @@ -325,7 +395,9 @@ def run_task(self, task_description, model_override=None): break else: if r_attempt == max_review_attempts - 1: - console.print("[bold red]Reached maximum review attempts. Proceeding to write files.[/bold red]") + 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 @@ -338,33 +410,53 @@ def run_task(self, task_description, model_override=None): 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"): + 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 + 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")) + 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]") + 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")) + 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( + "Files Created/Modified (Dry-Run Preview only): " + + ", ".join(generated_files.keys()) + ) self.conversation_history.append("Reviewer Decision: APPROVED") return @@ -384,7 +476,7 @@ def run_task(self, task_description, model_override=None): try: common = os.path.commonpath([repo_abs, full_path]) - is_safe = (common == repo_abs) + is_safe = common == repo_abs except Exception: is_safe = False @@ -406,15 +498,21 @@ def run_task(self, task_description, model_override=None): console.print(table) if written_count > 0: - console.print(f"[bold green]Successfully applied {written_count} changes to the repository![/bold green]") + 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")) + 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") \ No newline at end of file + self.conversation_history.append("Reviewer Decision: APPROVED") diff --git a/src/maca/rich_compat.py b/src/maca/rich_compat.py index 12067c8..21a7025 100644 --- a/src/maca/rich_compat.py +++ b/src/maca/rich_compat.py @@ -1,13 +1,11 @@ try: from rich.console import Console - from rich.panel import Panel from rich.markdown import Markdown - from rich.table import Table - from rich.status import Status + from rich.panel import Panel from rich.prompt import Prompt + from rich.status import Status + from rich.table import Table + + __all__ = ["Console", "Markdown", "Panel", "Prompt", "Status", "Table"] except ImportError: - from maca.rich_shim import ConsoleShim as Console - from maca.rich_shim import PanelShim as Panel - from maca.rich_shim import MarkdownShim as Markdown - from maca.rich_shim import TableShim as Table - from maca.rich_shim import PromptShim as Prompt + pass diff --git a/src/maca/rich_shim.py b/src/maca/rich_shim.py index 2d6f8e8..3ab785e 100644 --- a/src/maca/rich_shim.py +++ b/src/maca/rich_shim.py @@ -1,5 +1,3 @@ -import sys - GREEN = "\\033[92m" YELLOW = "\\033[93m" RED = "\\033[91m" @@ -10,6 +8,7 @@ WHITE = "\\033[97m" RESET = "\\033[0m" + def clean_tags(text): if not isinstance(text, str): return str(text) @@ -35,6 +34,7 @@ def clean_tags(text): text = text.replace("[/magenta]", RESET) return text + class ConsoleShim: def print(self, *args, **kwargs): cleaned_args = [clean_tags(arg) for arg in args] @@ -51,13 +51,17 @@ def status(self, text, spinner="dots"): class StatusContext: def __init__(self, text): self.text = text + def __enter__(self): print(clean_tags(f"{YELLOW}* {self.text}...{RESET}")) return self + def __exit__(self, exc_type, exc_val, exc_tb): pass + return StatusContext(text) + class PanelShim: def __init__(self, text, title=None, border_style=None): self.text = text @@ -71,21 +75,27 @@ def __str__(self): footer = border_char * len(header) return clean_tags(f"\\n{header}\\n{self.text}\\n{footer}\\n") + class MarkdownShim: def __init__(self, text): self.text = text + def __str__(self): return self.text + class TableShim: def __init__(self, title=None): self.title = title self.columns = [] self.rows = [] + def add_column(self, name, style=None): self.columns.append(name) + def add_row(self, *args): self.rows.append(args) + def __str__(self): res = f"\\n--- {self.title} ---\\n" if self.title else "\\n" res += " | ".join(self.columns) + "\\n" @@ -94,12 +104,23 @@ def __str__(self): res += " | ".join(r) + "\\n" return res + class PromptShim: @staticmethod - def ask(prompt="", *, console=None, default=None, choices=None, show_default=True, show_choices=True, password=False): + def ask( + prompt="", + *, + console=None, + default=None, + choices=None, + show_default=True, + show_choices=True, + password=False, + ): import getpass + cleaned = clean_tags(prompt) - + suffix = "" if choices and show_choices: choice_str = ", ".join(choices) @@ -108,9 +129,9 @@ def ask(prompt="", *, console=None, default=None, choices=None, show_default=Tru if default is not None and show_default: if str(default) not in prompt: suffix += f" ({default})" - + prompt_str = cleaned + suffix + " " - + while True: try: if password: @@ -119,12 +140,12 @@ def ask(prompt="", *, console=None, default=None, choices=None, show_default=Tru res = input(prompt_str) except (KeyboardInterrupt, EOFError): raise - + if not res.strip(): if default is not None: return default continue - + val = res.strip() if choices: if val in choices: diff --git a/tests/test_behavior.py b/tests/test_behavior.py index 039fada..42f0699 100644 --- a/tests/test_behavior.py +++ b/tests/test_behavior.py @@ -1,19 +1,21 @@ import unittest from unittest import mock +from maca import maca_config as config from maca.evaluator import ComplexityEvaluator -from maca.models.gemini import GeminiClient from maca.models.claude import ClaudeClient +from maca.models.gemini import GeminiClient from maca.models.local_gemma import LocalGemmaClient from maca.orchestrator import Orchestrator -from maca import maca_config as config class BehaviorTests(unittest.TestCase): def test_heuristic_evaluate_marks_database_task_as_complex(self): evaluator = ComplexityEvaluator() - with mock.patch.object(evaluator.gemma_client, "generate", side_effect=Exception("offline")): + with mock.patch.object( + evaluator.gemma_client, "generate", side_effect=Exception("offline") + ): result = evaluator.evaluate("Build a Flask API with SQL database integration") self.assertEqual(result, "COMPLEX") @@ -21,7 +23,9 @@ def test_heuristic_evaluate_marks_database_task_as_complex(self): def test_heuristic_evaluate_marks_short_prompt_as_simple(self): evaluator = ComplexityEvaluator() - with mock.patch.object(evaluator.gemma_client, "generate", side_effect=Exception("offline")): + with mock.patch.object( + evaluator.gemma_client, "generate", side_effect=Exception("offline") + ): result = evaluator.evaluate("Write a hello world script") self.assertEqual(result, "SIMPLE") @@ -57,8 +61,10 @@ def test_local_gemma_client_falls_back_to_cli(self): fake_http_error = RuntimeError("boom") - with mock.patch("urllib.request.urlopen", side_effect=fake_http_error), \ - mock.patch("subprocess.run") as run_mock: + with ( + mock.patch("urllib.request.urlopen", side_effect=fake_http_error), + mock.patch("subprocess.run") as run_mock, + ): run_mock.return_value.returncode = 0 run_mock.return_value.stdout = "CLI response" @@ -97,13 +103,14 @@ def test_claude_config_uses_maca_config(self): 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: - + 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") @@ -117,13 +124,14 @@ def test_routing_matrix_medium_both_online(self): 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: - + 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") @@ -137,13 +145,14 @@ def test_routing_matrix_medium_only_claude_online(self): 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: - + 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") @@ -157,13 +166,14 @@ def test_routing_matrix_complex_both_online(self): 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: - + 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") @@ -181,7 +191,7 @@ def test_coder_completion_verification_loop(self): is_done_mock = mock.Mock() is_done_mock.side_effect = [ (False, "Missing step 2 implementation"), - (True, "All steps completed successfully") + (True, "All steps completed successfully"), ] mock_plan = "1. Step one\n2. Step two" @@ -189,22 +199,23 @@ def test_coder_completion_verification_loop(self): coder_run_mock = mock.Mock() coder_run_mock.side_effect = [ "Generated content for step 1", - "Generated content for step 1 and 2" + "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): - + 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 @@ -213,7 +224,7 @@ def test_coder_completion_verification_loop(self): coder_inst.run = coder_run_mock coder_inst.parse_files.side_effect = [ {"file1.py": "content1"}, - {"file1.py": "content1_updated"} + {"file1.py": "content1_updated"}, ] reviewer_inst = mock_reviewer_cls.return_value @@ -225,7 +236,10 @@ def test_coder_completion_verification_loop(self): 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( + "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() @@ -236,27 +250,25 @@ def test_reviewer_rejection_nudge_loop(self): mock_plan = "1. Step one" coder_run_mock = mock.Mock() - coder_run_mock.side_effect = [ - "Initial coder response", - "Corrected coder response" - ] + 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." + "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")): - + 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 @@ -265,7 +277,7 @@ def test_reviewer_rejection_nudge_loop(self): coder_inst.run = coder_run_mock coder_inst.parse_files.side_effect = [ {"file1.py": "content1"}, - {"file1.py": "content1_updated"} + {"file1.py": "content1_updated"}, ] reviewer_inst = mock_reviewer_cls.return_value @@ -278,30 +290,38 @@ def test_reviewer_rejection_nudge_loop(self): 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( + "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 + from maca.main import parse_interactive_command + 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) + 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) + 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) + task, model, is_cmd = parse_interactive_command("implement a custom tokenizer", parser) self.assertFalse(is_cmd) diff --git a/tests/test_orchestrator_status.py b/tests/test_orchestrator_status.py index b1c0b2f..361c764 100644 --- a/tests/test_orchestrator_status.py +++ b/tests/test_orchestrator_status.py @@ -1,8 +1,8 @@ import unittest from unittest import mock -from maca.orchestrator import Orchestrator from maca import maca_config as config +from maca.orchestrator import Orchestrator class OrchestratorStatusTests(unittest.TestCase): @@ -13,8 +13,10 @@ def test_fast_status_uses_ollama_http_when_available(self): fake_response.__enter__.return_value.status = 200 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")): + with ( + mock.patch("urllib.request.urlopen", return_value=fake_response) as urlopen_mock, + mock.patch("subprocess.run", side_effect=FileNotFoundError("no cli")), + ): status = orch.check_backends_status(run_handshakes=False) self.assertEqual(status["Gemma"], "ONLINE (Ollama HTTP - gemma2:2b)")