diff --git a/codeframe/cli/app.py b/codeframe/cli/app.py index 51f22391..f062ffef 100644 --- a/codeframe/cli/app.py +++ b/codeframe/cli/app.py @@ -1971,7 +1971,7 @@ def work_start( False, "--execute", "-x", - help="Run the agent to execute the task (requires ANTHROPIC_API_KEY)", + help="Run the agent to execute the task (builtin engines require ANTHROPIC_API_KEY)", ), dry_run: bool = typer.Option( False, @@ -1997,7 +1997,7 @@ def work_start( engine: str = typer.Option( "react", "--engine", - help="Agent engine: 'react' (default, ReAct tool-use loop) or 'plan' (legacy step-based)", + help="Agent engine: react (default), plan (legacy), claude-code, opencode, or built-in", ), stall_timeout: int = typer.Option( 300, @@ -2049,9 +2049,12 @@ def work_start( task = matching[0] # Validate API key before creating run record (avoids dangling IN_PROGRESS state) + # External engines (claude-code, opencode) manage their own authentication if execute: - from codeframe.cli.validators import require_anthropic_api_key - require_anthropic_api_key() + from codeframe.core.engine_registry import is_external_engine + if not is_external_engine(engine): + from codeframe.cli.validators import require_anthropic_api_key + require_anthropic_api_key() # Start the run run = runtime.start_task_run(workspace, task.id) @@ -2875,7 +2878,7 @@ def batch_run( engine: str = typer.Option( "react", "--engine", - help="Agent engine: 'react' (default, ReAct tool-use loop) or 'plan' (legacy step-based)", + help="Agent engine: react (default), plan (legacy), claude-code, opencode, or built-in", ), stall_timeout: int = typer.Option( 300, @@ -2967,8 +2970,11 @@ def batch_run( return # Validate API key before batch execution - from codeframe.cli.validators import require_anthropic_api_key - require_anthropic_api_key() + # External engines (claude-code, opencode) manage their own authentication + from codeframe.core.engine_registry import is_external_engine + if not is_external_engine(engine): + from codeframe.cli.validators import require_anthropic_api_key + require_anthropic_api_key() # Execute batch if max_retries > 0: diff --git a/codeframe/core/adapters/__init__.py b/codeframe/core/adapters/__init__.py new file mode 100644 index 00000000..3051ec27 --- /dev/null +++ b/codeframe/core/adapters/__init__.py @@ -0,0 +1,25 @@ +from codeframe.core.adapters.agent_adapter import ( + AgentAdapter, + AgentEvent, + AgentResult, +) +from codeframe.core.adapters.builtin import ( + BuiltinPlanAdapter, + BuiltinReactAdapter, +) +from codeframe.core.adapters.claude_code import ClaudeCodeAdapter +from codeframe.core.adapters.opencode import OpenCodeAdapter +from codeframe.core.adapters.subprocess_adapter import SubprocessAdapter +from codeframe.core.adapters.verification_wrapper import VerificationWrapper + +__all__ = [ + "AgentAdapter", + "AgentEvent", + "AgentResult", + "BuiltinPlanAdapter", + "BuiltinReactAdapter", + "ClaudeCodeAdapter", + "OpenCodeAdapter", + "SubprocessAdapter", + "VerificationWrapper", +] diff --git a/codeframe/core/adapters/agent_adapter.py b/codeframe/core/adapters/agent_adapter.py new file mode 100644 index 00000000..683d087c --- /dev/null +++ b/codeframe/core/adapters/agent_adapter.py @@ -0,0 +1,60 @@ +"""Agent adapter protocol for delegating task execution to external coding agents.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from pathlib import Path +from typing import Callable, Literal, Protocol, runtime_checkable + + +@dataclass +class AgentResult: + """Result from an agent adapter execution.""" + + status: Literal["completed", "failed", "blocked"] + output: str = "" + modified_files: list[str] = field(default_factory=list) + error: str | None = None + blocker_question: str | None = None + + +@dataclass +class AgentEvent: + """Progress event emitted during agent execution.""" + + type: str # "progress", "tool_call", "output", "error" + data: dict = field(default_factory=dict) + + +@runtime_checkable +class AgentAdapter(Protocol): + """Protocol for agent execution engines. + + Any coding agent (Claude Code, OpenCode, Codex, etc.) must implement + this interface to be used as a CodeFRAME execution engine. + """ + + @property + def name(self) -> str: + """Engine name (e.g., 'claude-code', 'opencode').""" + ... + + def run( + self, + task_id: str, + prompt: str, + workspace_path: Path, + on_event: Callable[[AgentEvent], None] | None = None, + ) -> AgentResult: + """Execute a task and return the result. + + Args: + task_id: CodeFRAME task identifier + prompt: Rich context prompt assembled by TaskContextPackager + workspace_path: Path to the workspace/repo root + on_event: Optional callback for streaming progress events + + Returns: + AgentResult with status, output, and modified files + """ + ... diff --git a/codeframe/core/adapters/builtin.py b/codeframe/core/adapters/builtin.py new file mode 100644 index 00000000..da34e521 --- /dev/null +++ b/codeframe/core/adapters/builtin.py @@ -0,0 +1,197 @@ +"""Builtin adapter shims wrapping ReactAgent and Agent behind AgentAdapter protocol.""" + +from __future__ import annotations + +from pathlib import Path +from typing import TYPE_CHECKING, Callable, Optional + +from codeframe.core.adapters.agent_adapter import AgentEvent, AgentResult + +if TYPE_CHECKING: + from codeframe.adapters.llm.base import LLMProvider + from codeframe.core.conductor import GlobalFixCoordinator + from codeframe.core.stall_detector import StallAction + from codeframe.core.streaming import EventPublisher, RunOutputLogger + from codeframe.core.workspace import Workspace + + +class BuiltinReactAdapter: + """Wraps the existing ReactAgent behind the AgentAdapter interface. + + This shim allows the ReactAgent to be used through the unified + engine registry without modifying the ReactAgent class itself. + """ + + def __init__( + self, + workspace: Workspace, + llm_provider: LLMProvider, + *, + stall_timeout_s: float = 300, + stall_action: Optional[StallAction] = None, + event_publisher: Optional[EventPublisher] = None, + dry_run: bool = False, + verbose: bool = False, + debug: bool = False, + output_logger: Optional[RunOutputLogger] = None, + fix_coordinator: Optional[GlobalFixCoordinator] = None, + ) -> None: + self._workspace = workspace + self._llm_provider = llm_provider + self._stall_timeout_s = stall_timeout_s + self._stall_action = stall_action + self._event_publisher = event_publisher + self._dry_run = dry_run + self._verbose = verbose + self._debug = debug + self._output_logger = output_logger + self._fix_coordinator = fix_coordinator + + @property + def name(self) -> str: + return "react" + + def run( + self, + task_id: str, + prompt: str, + workspace_path: Path, + on_event: Callable[[AgentEvent], None] | None = None, + ) -> AgentResult: + """Run the ReactAgent and map its AgentStatus to AgentResult.""" + from codeframe.core.react_agent import ReactAgent + + def _bridge_event(event_type: str, data: dict) -> None: + if on_event: + on_event(AgentEvent(type=event_type, data=data)) + + kwargs: dict = { + "workspace": self._workspace, + "llm_provider": self._llm_provider, + "stall_timeout_s": self._stall_timeout_s, + "event_publisher": self._event_publisher, + "dry_run": self._dry_run, + "verbose": self._verbose, + "on_event": _bridge_event, + "debug": self._debug, + "output_logger": self._output_logger, + "fix_coordinator": self._fix_coordinator, + } + if self._stall_action is not None: + kwargs["stall_action"] = self._stall_action + + agent = ReactAgent(**kwargs) + status = agent.run(task_id) + return self._map_status(status) + + @staticmethod + def _map_status(status: object) -> AgentResult: + """Map AgentStatus enum to AgentResult.""" + from codeframe.core.agent import AgentStatus + + status_map = { + AgentStatus.COMPLETED: "completed", + AgentStatus.FAILED: "failed", + AgentStatus.BLOCKED: "blocked", + } + result_status = status_map.get(status, "failed") # type: ignore[arg-type] + return AgentResult( + status=result_status, + output=f"ReactAgent finished with status: {status.value}", # type: ignore[union-attr] + ) + + +class BuiltinPlanAdapter: + """Wraps the existing plan-based Agent behind the AgentAdapter interface.""" + + def __init__( + self, + workspace: Workspace, + llm_provider: LLMProvider, + *, + dry_run: bool = False, + verbose: bool = False, + debug: bool = False, + output_logger: Optional[RunOutputLogger] = None, + fix_coordinator: Optional[GlobalFixCoordinator] = None, + event_publisher: Optional[EventPublisher] = None, + ) -> None: + self._workspace = workspace + self._llm_provider = llm_provider + self._dry_run = dry_run + self._verbose = verbose + self._debug = debug + self._output_logger = output_logger + self._fix_coordinator = fix_coordinator + self._event_publisher = event_publisher + + @property + def name(self) -> str: + return "plan" + + def run( + self, + task_id: str, + prompt: str, + workspace_path: Path, + on_event: Callable[[AgentEvent], None] | None = None, + ) -> AgentResult: + """Run the plan-based Agent and map its AgentState to AgentResult.""" + from codeframe.core.agent import Agent + + def _bridge_event(event_type: str, data: dict) -> None: + if on_event: + on_event(AgentEvent(type=event_type, data=data)) + + agent = Agent( + workspace=self._workspace, + llm_provider=self._llm_provider, + dry_run=self._dry_run, + on_event=_bridge_event, + debug=self._debug, + verbose=self._verbose, + fix_coordinator=self._fix_coordinator, + output_logger=self._output_logger, + event_publisher=self._event_publisher, + ) + state = agent.run(task_id) + return self._map_state(state) + + @staticmethod + def _map_state(state: object) -> AgentResult: + """Map AgentState dataclass to AgentResult.""" + from codeframe.core.agent import AgentStatus + + status_map = { + AgentStatus.COMPLETED: "completed", + AgentStatus.FAILED: "failed", + AgentStatus.BLOCKED: "blocked", + } + result_status = status_map.get(state.status, "failed") # type: ignore[union-attr] + + blocker_question = None + if state.status == AgentStatus.BLOCKED: # type: ignore[union-attr] + blocker = getattr(state, "blocker", None) + if blocker: + blocker_question = getattr(blocker, "question", None) or getattr( + blocker, "reason", None + ) + + error = None + if state.status == AgentStatus.FAILED: # type: ignore[union-attr] + gate_results = getattr(state, "gate_results", None) or [] + error_parts: list[str] = [] + for gr in gate_results: + for check in getattr(gr, "checks", []): + output = getattr(check, "output", None) + if output: + error_parts.append(output) + if error_parts: + error = "\n".join(error_parts[:3]) + + return AgentResult( + status=result_status, + output=f"PlanAgent finished with status: {state.status.value}", # type: ignore[union-attr] + blocker_question=blocker_question, + error=error, + ) diff --git a/codeframe/core/adapters/claude_code.py b/codeframe/core/adapters/claude_code.py new file mode 100644 index 00000000..20eae714 --- /dev/null +++ b/codeframe/core/adapters/claude_code.py @@ -0,0 +1,62 @@ +"""Claude Code adapter for delegating task execution to the claude CLI.""" + +from __future__ import annotations + +from pathlib import Path + +from codeframe.core.adapters.subprocess_adapter import SubprocessAdapter + + +class ClaudeCodeAdapter(SubprocessAdapter): + """Adapter that delegates code execution to Claude Code CLI. + + Invokes ``claude`` with ``--print`` flag for non-interactive output. + The prompt is piped via stdin. + + Requires the Claude Code CLI to be installed: + https://docs.anthropic.com/en/docs/claude-code + """ + + def __init__(self, allowlist: list[str] | None = None) -> None: + """Initialize the Claude Code adapter. + + Args: + allowlist: Optional list of allowed tools/permissions. + If provided, uses ``--allowedTools`` flag for each tool. + When omitted, no permission flags are added (the caller + is responsible for configuring permissions externally). + """ + cli_args = ["--print"] + if allowlist: + for tool in allowlist: + cli_args.extend(["--allowedTools", tool]) + + super().__init__(binary="claude", cli_args=cli_args) + self._allowlist = allowlist + + @property + def name(self) -> str: # noqa: D102 + return "claude-code" + + def build_command(self, prompt: str, workspace_path: Path) -> list[str]: + """Build claude CLI command. + + Args: + prompt: The task prompt (sent via stdin, not in the command). + workspace_path: Workspace root (cwd is set by the base class). + + Returns: + Command list for subprocess.Popen. + """ + return [self._binary_path, *self._cli_args] + + def get_stdin(self, prompt: str) -> str | None: + """Send prompt via stdin. + + Args: + prompt: The task prompt to pipe into the claude process. + + Returns: + The prompt string. + """ + return prompt diff --git a/codeframe/core/adapters/opencode.py b/codeframe/core/adapters/opencode.py new file mode 100644 index 00000000..86ffb2bc --- /dev/null +++ b/codeframe/core/adapters/opencode.py @@ -0,0 +1,48 @@ +"""OpenCode adapter for delegating task execution to the opencode CLI.""" + +from __future__ import annotations + +from pathlib import Path + +from codeframe.core.adapters.subprocess_adapter import SubprocessAdapter + + +class OpenCodeAdapter(SubprocessAdapter): + """Adapter that delegates code execution to OpenCode CLI. + + Invokes ``opencode`` with ``--non-interactive`` flag for headless execution. + The prompt is piped via stdin. + + Requires OpenCode to be installed: + https://github.com/opencode-ai/opencode + """ + + def __init__(self) -> None: + super().__init__(binary="opencode", cli_args=["--non-interactive"]) + + @property + def name(self) -> str: # noqa: D102 + return "opencode" + + def build_command(self, prompt: str, workspace_path: Path) -> list[str]: + """Build opencode CLI command. + + Args: + prompt: The task prompt (sent via stdin, not in the command). + workspace_path: Workspace root (cwd is set by the base class). + + Returns: + Command list for subprocess.Popen. + """ + return [self._binary_path, *self._cli_args] + + def get_stdin(self, prompt: str) -> str | None: + """Send prompt via stdin. + + Args: + prompt: The task prompt to pipe into the opencode process. + + Returns: + The prompt string. + """ + return prompt diff --git a/codeframe/core/adapters/subprocess_adapter.py b/codeframe/core/adapters/subprocess_adapter.py new file mode 100644 index 00000000..e3be6d4a --- /dev/null +++ b/codeframe/core/adapters/subprocess_adapter.py @@ -0,0 +1,204 @@ +"""Base subprocess adapter for external coding agents.""" + +from __future__ import annotations + +import shutil +import subprocess +import threading +from pathlib import Path +from typing import Callable + +from codeframe.core.adapters.agent_adapter import AgentEvent, AgentResult +from codeframe.core.blocker_detection import classify_error_for_blocker + + +class SubprocessAdapter: + """Base adapter for coding agents invoked as subprocesses. + + Provides shared infrastructure: binary availability check, subprocess + execution with stdout streaming, and exit code to AgentResult mapping. + + Subclasses override build_command() to customize CLI invocation. + """ + + # Default timeout: 30 minutes (coding agents can be long-running) + DEFAULT_TIMEOUT_S = 1800 + + def __init__( + self, + binary: str, + cli_args: list[str] | None = None, + timeout_s: int | None = None, + ) -> None: + """Initialize with the binary name and default CLI args. + + Args: + binary: Name of the CLI binary (e.g., 'claude', 'opencode') + cli_args: Default CLI arguments appended to every invocation + timeout_s: Max execution time in seconds (default: 1800, None = no limit) + + Raises: + EnvironmentError: If the binary is not found on PATH + """ + self._binary = binary + self._cli_args = cli_args or [] + self._timeout_s = timeout_s if timeout_s is not None else self.DEFAULT_TIMEOUT_S + + resolved = shutil.which(binary) + if resolved is None: + raise EnvironmentError( + f"'{binary}' not found on PATH. " + f"Install it or ensure it is available in your environment." + ) + self._binary_path = resolved + + @property + def name(self) -> str: + """Engine name derived from the binary.""" + return self._binary + + def build_command(self, prompt: str, workspace_path: Path) -> list[str]: + """Build the subprocess command list. + + Override in subclasses for custom CLI invocation. + Default: [binary, *cli_args] with prompt on stdin. + + Args: + prompt: The task prompt to send to the agent + workspace_path: Path to the workspace root + + Returns: + Command list for subprocess.Popen + """ + return [self._binary_path, *self._cli_args] + + def get_stdin(self, prompt: str) -> str | None: + """Return stdin content for the subprocess, or None to not pipe stdin. + + Override in subclasses if the agent reads prompt from a file instead. + Default: returns the prompt string (piped via stdin). + """ + return prompt + + def run( + self, + task_id: str, + prompt: str, + workspace_path: Path, + on_event: Callable[[AgentEvent], None] | None = None, + ) -> AgentResult: + """Execute the agent subprocess and return the result.""" + cmd = self.build_command(prompt, workspace_path) + stdin_content = self.get_stdin(prompt) + + stdout_lines: list[str] = [] + stderr_chunks: list[str] = [] + + try: + process = subprocess.Popen( + cmd, + stdin=subprocess.PIPE if stdin_content else None, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + cwd=str(workspace_path), + text=True, + ) + + # Write stdin and close + if stdin_content and process.stdin: + process.stdin.write(stdin_content) + process.stdin.close() + + # Drain stderr in a background thread to prevent deadlock. + # Without this, if the child fills the stderr pipe buffer (~64KB) + # before finishing stdout, both processes block indefinitely. + def _drain_stderr() -> None: + if process.stderr: + stderr_chunks.append(process.stderr.read()) + + stderr_thread = threading.Thread(target=_drain_stderr, daemon=True) + stderr_thread.start() + + # Stream stdout line-by-line + if process.stdout: + for line in process.stdout: + stripped = line.rstrip("\n") + stdout_lines.append(stripped) + if on_event: + on_event(AgentEvent(type="output", data={"line": stripped})) + + # Wait for stderr thread and process to finish + stderr_thread.join(timeout=10) + try: + process.wait(timeout=self._timeout_s) + except subprocess.TimeoutExpired: + process.kill() + process.wait() + return AgentResult( + status="failed", + output="\n".join(stdout_lines), + error=f"Process timed out after {self._timeout_s}s", + ) + + except FileNotFoundError: + return AgentResult( + status="failed", + error=f"Binary '{self._binary}' not found during execution", + ) + except OSError as e: + return AgentResult( + status="failed", + error=f"Failed to start '{self._binary}': {e}", + ) + + stderr_output = "".join(stderr_chunks) + + return self._map_result( + exit_code=process.returncode, + stdout="\n".join(stdout_lines), + stderr=stderr_output, + workspace_path=workspace_path, + ) + + def _map_result( + self, + exit_code: int, + stdout: str, + stderr: str, + workspace_path: Path, + ) -> AgentResult: + """Map subprocess exit code and output to AgentResult. + + Override in subclasses for custom exit code interpretation. + Default: 0 = completed, non-zero = failed. + Uses blocker_detection.classify_error_for_blocker for blocker detection. + """ + if exit_code == 0: + return AgentResult( + status="completed", + output=stdout, + ) + + # Check if the error looks like a blocker using the shared classifier + combined_output = f"{stdout}\n{stderr}".strip() + category = classify_error_for_blocker(combined_output) + if category is not None: + return AgentResult( + status="blocked", + output=stdout, + error=stderr or None, + blocker_question=self._extract_blocker_question(combined_output), + ) + + return AgentResult( + status="failed", + output=stdout, + error=stderr or f"Process exited with code {exit_code}", + ) + + def _extract_blocker_question(self, output: str) -> str: + """Extract a meaningful blocker question from output.""" + lines = [line.strip() for line in output.splitlines() if line.strip()] + if lines: + return lines[-1] + return "Agent encountered a blocker but no details were provided." diff --git a/codeframe/core/adapters/verification_wrapper.py b/codeframe/core/adapters/verification_wrapper.py new file mode 100644 index 00000000..d76ab972 --- /dev/null +++ b/codeframe/core/adapters/verification_wrapper.py @@ -0,0 +1,138 @@ +"""Verification gate wrapper for agent adapters.""" + +from __future__ import annotations + +from pathlib import Path +from typing import Callable, Optional + +from codeframe.core.adapters.agent_adapter import AgentAdapter, AgentEvent, AgentResult +from codeframe.core.gates import GateStatus +from codeframe.core.gates import run as run_gates +from codeframe.core.workspace import Workspace + + +class VerificationWrapper: + """Wraps any AgentAdapter with post-execution verification gates. + + After the inner adapter completes, runs verification gates (pytest, ruff, etc.). + If gates fail, re-invokes the adapter with error context for self-correction, + up to max_correction_rounds times. + + This is the same self-correction loop that ReactAgent._run_final_verification() + uses, but decoupled from any specific engine so it wraps any adapter. + """ + + def __init__( + self, + inner: AgentAdapter, + workspace: Workspace, + max_correction_rounds: int = 3, + gate_names: Optional[list[str]] = None, + verbose: bool = False, + ) -> None: + self._inner = inner + self._workspace = workspace + self._max_correction_rounds = max_correction_rounds + self._gate_names = gate_names # None = use default gates + self._verbose = verbose + + @property + def name(self) -> str: + return f"verified-{self._inner.name}" + + def run( + self, + task_id: str, + prompt: str, + workspace_path: Path, + on_event: Callable[[AgentEvent], None] | None = None, + ) -> AgentResult: + """Run the inner adapter, then verify with gates. Self-correct on failure.""" + + # Initial run + result = self._inner.run(task_id, prompt, workspace_path, on_event) + + # Only verify if the adapter reported success + if result.status != "completed": + return result + + # Run verification gates with self-correction loop + for round_num in range(self._max_correction_rounds): + if on_event: + on_event(AgentEvent( + type="verification", + data={ + "round": round_num + 1, + "max_rounds": self._max_correction_rounds, + }, + )) + + gate_result = run_gates( + self._workspace, + gates=self._gate_names, + verbose=self._verbose, + ) + + if gate_result.passed: + if on_event: + on_event(AgentEvent(type="verification_passed", data={})) + return result + + # Gates failed -- build correction prompt and re-invoke + if on_event: + on_event(AgentEvent( + type="verification_failed", + data={ + "round": round_num + 1, + "checks": len(gate_result.checks), + }, + )) + + error_summary = self._format_gate_errors(gate_result) + correction_prompt = ( + f"{prompt}\n\n" + f"## Verification Gate Failures (Correction Round {round_num + 1})\n\n" + f"Your previous changes failed the following verification gates. " + f"Fix these issues:\n\n{error_summary}" + ) + + result = self._inner.run( + task_id, correction_prompt, workspace_path, on_event, + ) + + if result.status != "completed": + return result + + # Final gate check after all correction rounds + gate_result = run_gates( + self._workspace, + gates=self._gate_names, + verbose=self._verbose, + ) + + if gate_result.passed: + return result + + # All rounds exhausted, gates still failing + error_summary = self._format_gate_errors(gate_result) + return AgentResult( + status="failed", + output=result.output, + error=( + f"Verification gates still failing after " + f"{self._max_correction_rounds} correction rounds:\n{error_summary}" + ), + ) + + @staticmethod + def _format_gate_errors(gate_result) -> str: + """Format gate check failures into a readable summary.""" + + lines: list[str] = [] + for check in gate_result.checks: + if check.status == GateStatus.FAILED: + lines.append(f"### {check.name}: FAILED") + if check.output: + lines.append(f"```\n{check.output[:2000]}\n```") + lines.append("") + return "\n".join(lines) if lines else "Gate checks failed (no details available)" diff --git a/codeframe/core/context_packager.py b/codeframe/core/context_packager.py new file mode 100644 index 00000000..135e335c --- /dev/null +++ b/codeframe/core/context_packager.py @@ -0,0 +1,99 @@ +"""Task context packager for assembling rich prompts for agent adapters.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Optional + +from codeframe.core.context import ContextLoader, TaskContext +from codeframe.core.workspace import Workspace + + +@dataclass +class PackagedContext: + """Packaged task context ready for an agent adapter.""" + + prompt: str + context: TaskContext + + +class TaskContextPackager: + """Assembles rich task prompts from CodeFRAME context for any agent adapter. + + Wraps ContextLoader and appends gate requirements so external agents + know what verification criteria they must satisfy. + """ + + def __init__(self, workspace: Workspace) -> None: + self._workspace = workspace + self._loader = ContextLoader(workspace) + + def build( + self, task_id: str, gate_names: Optional[list[str]] = None + ) -> PackagedContext: + """Build a rich prompt and context for the given task. + + Args: + task_id: CodeFRAME task identifier. + gate_names: Optional list of gate names that will run post-execution. + If None, uses default gates (pytest, ruff). + + Returns: + PackagedContext with assembled prompt and raw TaskContext. + """ + context = self._loader.load(task_id) + + prompt_parts = [context.to_prompt_context()] + + gates = gate_names or ["pytest", "ruff"] + prompt_parts.append(self._build_gate_section(gates)) + prompt_parts.append(self._build_instructions_section()) + + return PackagedContext( + prompt="\n".join(prompt_parts), + context=context, + ) + + def _build_gate_section(self, gate_names: list[str]) -> str: + """Build the gate requirements section.""" + lines = [ + "", + "## Verification Gates", + "", + "After you complete the task, the following verification gates " + "will run automatically:", + "", + ] + for gate in gate_names: + if gate == "pytest": + lines.append( + "- **pytest**: All tests must pass. " + "Run `pytest` to verify before finishing." + ) + elif gate == "ruff": + lines.append( + "- **ruff**: Code must pass linting. " + "Run `ruff check .` to verify." + ) + else: + lines.append(f"- **{gate}**: Must pass.") + lines.append("") + lines.append( + "Ensure your changes satisfy ALL gates before reporting completion." + ) + lines.append("") + return "\n".join(lines) + + def _build_instructions_section(self) -> str: + """Build general execution instructions for the agent.""" + return "\n".join( + [ + "## Execution Instructions", + "", + "- Make only the changes necessary to complete the task", + "- Do not modify unrelated files", + "- Follow existing code patterns and conventions", + "- If you encounter a blocker you cannot resolve, report it clearly", + "", + ] + ) diff --git a/codeframe/core/engine_registry.py b/codeframe/core/engine_registry.py new file mode 100644 index 00000000..1c421c66 --- /dev/null +++ b/codeframe/core/engine_registry.py @@ -0,0 +1,167 @@ +"""Engine registry for adapter lookup and resolution.""" + +from __future__ import annotations + +import os +from typing import Any + +from codeframe.core.adapters.agent_adapter import AgentAdapter + + +# All valid engine names +VALID_ENGINES = frozenset({ + "react", + "plan", + "claude-code", + "opencode", + "built-in", # Alias for "react" +}) + +# External engines that use subprocess adapters (no LLM provider needed) +EXTERNAL_ENGINES = frozenset({ + "claude-code", + "opencode", +}) + +# Builtin engines that need workspace + LLM provider +BUILTIN_ENGINES = frozenset({ + "react", + "plan", + "built-in", +}) + + +def resolve_engine(cli_engine: str | None = None) -> str: + """Resolve which engine to use. + + Priority: + 1. CLI --engine flag (explicit wins) + 2. CODEFRAME_ENGINE environment variable + 3. Default: "react" + + Args: + cli_engine: Engine name from CLI flag, or None. + + Returns: + Resolved engine name. + + Raises: + ValueError: If resolved engine is not valid. + """ + engine = cli_engine or os.environ.get("CODEFRAME_ENGINE") or "react" + + # Normalize alias + if engine == "built-in": + engine = "react" + + if engine not in VALID_ENGINES: + raise ValueError( + f"Invalid engine '{engine}'. " + f"Must be one of: {', '.join(sorted(VALID_ENGINES))}" + ) + + return engine + + +def is_external_engine(engine: str) -> bool: + """Check if an engine uses an external subprocess adapter.""" + return engine in EXTERNAL_ENGINES + + +def get_external_adapter(engine: str, **kwargs: Any) -> AgentAdapter: + """Get an adapter instance for an external engine. + + Args: + engine: Engine name (must be in EXTERNAL_ENGINES). + **kwargs: Adapter-specific keyword arguments. + + Returns: + AgentAdapter instance. + + Raises: + ValueError: If engine is not an external engine. + EnvironmentError: If the required binary is not installed. + """ + if engine == "claude-code": + from codeframe.core.adapters.claude_code import ClaudeCodeAdapter + + return ClaudeCodeAdapter(**kwargs) + elif engine == "opencode": + from codeframe.core.adapters.opencode import OpenCodeAdapter + + return OpenCodeAdapter() + else: + raise ValueError( + f"Unknown external engine '{engine}'. " + f"Valid external engines: {', '.join(sorted(EXTERNAL_ENGINES))}" + ) + + +def get_builtin_adapter( + engine: str, + workspace: Any, + llm_provider: Any, + **kwargs: Any, +) -> AgentAdapter: + """Get an adapter instance for a builtin engine. + + Args: + engine: Engine name (must be in BUILTIN_ENGINES). + workspace: Workspace instance. + llm_provider: LLM provider instance. + **kwargs: Additional keyword arguments passed to the adapter. + + Returns: + AgentAdapter instance. + + Raises: + ValueError: If engine is not a builtin engine. + """ + resolved = "react" if engine == "built-in" else engine + + if resolved == "react": + from codeframe.core.adapters.builtin import BuiltinReactAdapter + + return BuiltinReactAdapter(workspace, llm_provider, **kwargs) + elif resolved == "plan": + from codeframe.core.adapters.builtin import BuiltinPlanAdapter + + return BuiltinPlanAdapter(workspace, llm_provider, **kwargs) + else: + raise ValueError( + f"Unknown builtin engine '{engine}'. " + f"Valid builtin engines: {', '.join(sorted(BUILTIN_ENGINES))}" + ) + + +def get_adapter( + engine: str, + workspace: Any = None, + llm_provider: Any = None, + **kwargs: Any, +) -> AgentAdapter: + """Get an adapter for any engine type. + + Unified factory that handles both external and builtin engines. + + Args: + engine: Engine name. + workspace: Workspace (required for builtin engines). + llm_provider: LLM provider (required for builtin engines). + **kwargs: Additional adapter-specific arguments. + + Returns: + AgentAdapter instance. + + Raises: + ValueError: If engine is invalid or builtin engine missing required args. + EnvironmentError: If external engine binary is not installed. + """ + if is_external_engine(engine): + return get_external_adapter(engine, **kwargs) + else: + if workspace is None or llm_provider is None: + raise ValueError( + f"Builtin engine '{engine}' requires workspace and llm_provider" + ) + return get_builtin_adapter(engine, workspace, llm_provider, **kwargs) diff --git a/codeframe/core/runtime.py b/codeframe/core/runtime.py index 988f77cd..d2943627 100644 --- a/codeframe/core/runtime.py +++ b/codeframe/core/runtime.py @@ -613,7 +613,7 @@ def execute_agent( verbose: If True, print detailed progress to stdout fix_coordinator: Optional coordinator for global fixes (for parallel execution) event_publisher: Optional EventPublisher for SSE streaming (real-time events) - engine: Agent engine to use ("react" for ReactAgent (default), "plan" for legacy Agent) + engine: Agent engine to use ("react", "plan", "claude-code", "opencode", "built-in") stall_timeout_s: Seconds without tool activity before stall detection (0 = disabled) stall_action: Recovery action on stall ("blocker", "retry", or "fail") @@ -621,28 +621,29 @@ def execute_agent( Final AgentState after execution Raises: - ValueError: If ANTHROPIC_API_KEY is not set or engine is invalid + ValueError: If ANTHROPIC_API_KEY is not set (for builtin engines) or engine is invalid """ import os from codeframe.core.agent import Agent, AgentState, AgentStatus from codeframe.adapters.llm import get_provider from codeframe.core.diagnostics import RunLogger, LogCategory + from codeframe.core.engine_registry import VALID_ENGINES, is_external_engine # Validate engine parameter - valid_engines = ("plan", "react") - if engine not in valid_engines: + if engine not in VALID_ENGINES: raise ValueError( - f"Invalid engine '{engine}'. Must be one of: {', '.join(valid_engines)}" + f"Invalid engine '{engine}'. Must be one of: {', '.join(sorted(VALID_ENGINES))}" ) - # Get LLM provider - if not os.getenv("ANTHROPIC_API_KEY"): + # External engines manage their own authentication + if not is_external_engine(engine) and not os.getenv("ANTHROPIC_API_KEY"): raise ValueError( "ANTHROPIC_API_KEY environment variable is required for agent execution. " "Set it with: export ANTHROPIC_API_KEY=your-key" ) - provider = get_provider("anthropic") + # Only create LLM provider for builtin engines (external engines manage their own) + provider = get_provider("anthropic") if not is_external_engine(engine) else None # Create run logger for structured logging run_logger = RunLogger(workspace, run.id, run.task_id) @@ -672,7 +673,69 @@ def on_agent_event(event_type: str, data: dict) -> None: run_logger.info(category, f"Agent event: {event_type}", data) # Create and run agent based on engine selection - if engine == "react": + # --- External engine path (claude-code, opencode, etc.) --- + if is_external_engine(engine): + from codeframe.core.engine_registry import get_external_adapter + from codeframe.core.context_packager import TaskContextPackager + from codeframe.core.adapters.verification_wrapper import VerificationWrapper + from codeframe.core.adapters.agent_adapter import AgentEvent + + run_logger.info( + LogCategory.AGENT_ACTION, + f"Using external engine: {engine}", + {"engine": engine}, + ) + + # Build rich context prompt for the external agent + packager = TaskContextPackager(workspace) + packaged = packager.build(run.task_id) + + # Get the adapter and wrap with verification gates + adapter = get_external_adapter(engine) + wrapper = VerificationWrapper( + adapter, + workspace, + max_correction_rounds=3, + verbose=verbose, + ) + + # Bridge AgentEvent callbacks to workspace event system + def on_adapter_event(event: AgentEvent) -> None: + on_agent_event(event.type, event.data) + + result = wrapper.run( + run.task_id, + packaged.prompt, + workspace.repo_path, + on_event=on_adapter_event, + ) + + # Map AgentResult to AgentState for compatibility with rest of runtime + status_map = { + "completed": AgentStatus.COMPLETED, + "failed": AgentStatus.FAILED, + "blocked": AgentStatus.BLOCKED, + } + agent_status = status_map.get(result.status, AgentStatus.FAILED) + state = AgentState(status=agent_status) + + # Create blocker if adapter reported one + if result.status == "blocked" and result.blocker_question: + from codeframe.core import blockers + blockers.create( + workspace, + task_id=run.task_id, + question=result.blocker_question, + ) + + run_logger.info( + LogCategory.AGENT_ACTION, + f"External engine completed: {result.status}", + {"engine": engine, "output_length": len(result.output)}, + ) + + # --- Builtin engine paths (react, plan) --- + elif engine == "react": # ReactAgent has a simpler interface — it handles its own # retries and verification internally. from codeframe.core.react_agent import ReactAgent diff --git a/tests/core/adapters/__init__.py b/tests/core/adapters/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/core/adapters/test_agent_adapter.py b/tests/core/adapters/test_agent_adapter.py new file mode 100644 index 00000000..bfd7ce18 --- /dev/null +++ b/tests/core/adapters/test_agent_adapter.py @@ -0,0 +1,116 @@ +"""Tests for agent adapter protocol and data types.""" + +from pathlib import Path + +from codeframe.core.adapters.agent_adapter import AgentAdapter, AgentEvent, AgentResult + + +class TestAgentResult: + """Tests for AgentResult dataclass.""" + + def test_completed_result(self): + result = AgentResult(status="completed", output="Task done") + assert result.status == "completed" + assert result.output == "Task done" + assert result.modified_files == [] + assert result.error is None + assert result.blocker_question is None + + def test_failed_result_with_error(self): + result = AgentResult(status="failed", error="Tests failed") + assert result.status == "failed" + assert result.error == "Tests failed" + + def test_blocked_result_with_question(self): + result = AgentResult( + status="blocked", + blocker_question="Which database should I use?", + ) + assert result.status == "blocked" + assert result.blocker_question == "Which database should I use?" + + def test_result_with_modified_files(self): + result = AgentResult( + status="completed", + modified_files=["src/main.py", "tests/test_main.py"], + ) + assert len(result.modified_files) == 2 + + def test_default_values(self): + result = AgentResult(status="completed") + assert result.output == "" + assert result.modified_files == [] + assert result.error is None + assert result.blocker_question is None + + +class TestAgentEvent: + """Tests for AgentEvent dataclass.""" + + def test_event_creation(self): + event = AgentEvent(type="progress", data={"step": 1}) + assert event.type == "progress" + assert event.data == {"step": 1} + + def test_event_default_data(self): + event = AgentEvent(type="output") + assert event.data == {} + + +class TestAgentAdapterProtocol: + """Tests for AgentAdapter protocol compliance.""" + + def test_protocol_is_runtime_checkable(self): + """Verify the protocol can be checked at runtime.""" + + class MockAdapter: + @property + def name(self) -> str: + return "mock" + + def run(self, task_id, prompt, workspace_path, on_event=None): + return AgentResult(status="completed") + + adapter = MockAdapter() + assert isinstance(adapter, AgentAdapter) + + def test_non_conforming_class_fails_check(self): + """A class without the right methods should not match the protocol.""" + + class NotAnAdapter: + pass + + assert not isinstance(NotAnAdapter(), AgentAdapter) + + def test_partial_implementation_fails_check(self): + """A class with only some methods should not match.""" + + class PartialAdapter: + @property + def name(self) -> str: + return "partial" + + # Missing run() + + assert not isinstance(PartialAdapter(), AgentAdapter) + + def test_adapter_can_stream_events(self): + """Verify on_event callback works during execution.""" + events_received: list[AgentEvent] = [] + + class StreamingAdapter: + @property + def name(self) -> str: + return "streaming" + + def run(self, task_id, prompt, workspace_path, on_event=None): + if on_event: + on_event(AgentEvent(type="progress", data={"step": 1})) + on_event(AgentEvent(type="output", data={"line": "hello"})) + return AgentResult(status="completed") + + adapter = StreamingAdapter() + adapter.run("task-1", "do stuff", Path("/tmp"), on_event=events_received.append) + assert len(events_received) == 2 + assert events_received[0].type == "progress" + assert events_received[1].type == "output" diff --git a/tests/core/adapters/test_builtin.py b/tests/core/adapters/test_builtin.py new file mode 100644 index 00000000..12392d84 --- /dev/null +++ b/tests/core/adapters/test_builtin.py @@ -0,0 +1,242 @@ +"""Tests for builtin adapter shims.""" + +import pytest +from pathlib import Path +from unittest.mock import MagicMock, patch + +from codeframe.core.adapters.agent_adapter import AgentAdapter, AgentEvent +from codeframe.core.adapters.builtin import BuiltinPlanAdapter, BuiltinReactAdapter +from codeframe.core.agent import AgentState, AgentStatus + +# Patch targets at the source modules, since builtin.py uses lazy imports. +_REACT_AGENT_CLS = "codeframe.core.react_agent.ReactAgent" +_PLAN_AGENT_CLS = "codeframe.core.agent.Agent" + + +@pytest.fixture +def mock_workspace(): + ws = MagicMock() + ws.repo_path = Path("/tmp/test-repo") + return ws + + +@pytest.fixture +def mock_provider(): + return MagicMock() + + +# --------------------------------------------------------------------------- +# BuiltinReactAdapter +# --------------------------------------------------------------------------- + + +class TestBuiltinReactAdapter: + def test_name(self, mock_workspace, mock_provider): + adapter = BuiltinReactAdapter(mock_workspace, mock_provider) + assert adapter.name == "react" + + def test_conforms_to_protocol(self, mock_workspace, mock_provider): + adapter = BuiltinReactAdapter(mock_workspace, mock_provider) + assert isinstance(adapter, AgentAdapter) + + @pytest.mark.parametrize( + "agent_status, expected", + [ + (AgentStatus.COMPLETED, "completed"), + (AgentStatus.FAILED, "failed"), + (AgentStatus.BLOCKED, "blocked"), + ], + ) + def test_status_mapping( + self, mock_workspace, mock_provider, agent_status, expected + ): + with patch(_REACT_AGENT_CLS) as mock_cls: + mock_cls.return_value.run.return_value = agent_status + adapter = BuiltinReactAdapter(mock_workspace, mock_provider) + result = adapter.run("task-1", "prompt", Path("/tmp")) + assert result.status == expected + + def test_unknown_status_maps_to_failed(self, mock_workspace, mock_provider): + with patch(_REACT_AGENT_CLS) as mock_cls: + mock_cls.return_value.run.return_value = AgentStatus.IDLE + adapter = BuiltinReactAdapter(mock_workspace, mock_provider) + result = adapter.run("task-1", "prompt", Path("/tmp")) + assert result.status == "failed" + + def test_forwards_events(self, mock_workspace, mock_provider): + received: list[AgentEvent] = [] + + def capture_constructor(**kwargs): + cb = kwargs.get("on_event") + if cb: + cb("step_started", {"step": 1}) + inst = MagicMock() + inst.run.return_value = AgentStatus.COMPLETED + return inst + + with patch(_REACT_AGENT_CLS) as mock_cls: + mock_cls.side_effect = capture_constructor + adapter = BuiltinReactAdapter(mock_workspace, mock_provider) + adapter.run("task-1", "prompt", Path("/tmp"), on_event=received.append) + + assert len(received) == 1 + assert received[0].type == "step_started" + assert received[0].data == {"step": 1} + + def test_no_event_callback_does_not_error(self, mock_workspace, mock_provider): + """Calling with on_event=None should not raise.""" + + def trigger_event(**kwargs): + cb = kwargs.get("on_event") + if cb: + cb("some_event", {}) + inst = MagicMock() + inst.run.return_value = AgentStatus.COMPLETED + return inst + + with patch(_REACT_AGENT_CLS) as mock_cls: + mock_cls.side_effect = trigger_event + adapter = BuiltinReactAdapter(mock_workspace, mock_provider) + result = adapter.run("task-1", "prompt", Path("/tmp")) + assert result.status == "completed" + + def test_stall_action_forwarded(self, mock_workspace, mock_provider): + from codeframe.core.stall_detector import StallAction + + with patch(_REACT_AGENT_CLS) as mock_cls: + mock_cls.return_value.run.return_value = AgentStatus.COMPLETED + adapter = BuiltinReactAdapter( + mock_workspace, mock_provider, stall_action=StallAction.FAIL + ) + adapter.run("task-1", "prompt", Path("/tmp")) + _, kwargs = mock_cls.call_args + assert kwargs["stall_action"] == StallAction.FAIL + + def test_stall_action_omitted_when_none(self, mock_workspace, mock_provider): + with patch(_REACT_AGENT_CLS) as mock_cls: + mock_cls.return_value.run.return_value = AgentStatus.COMPLETED + adapter = BuiltinReactAdapter(mock_workspace, mock_provider) + adapter.run("task-1", "prompt", Path("/tmp")) + _, kwargs = mock_cls.call_args + assert "stall_action" not in kwargs + + +# --------------------------------------------------------------------------- +# BuiltinPlanAdapter +# --------------------------------------------------------------------------- + + +class TestBuiltinPlanAdapter: + def test_name(self, mock_workspace, mock_provider): + adapter = BuiltinPlanAdapter(mock_workspace, mock_provider) + assert adapter.name == "plan" + + def test_conforms_to_protocol(self, mock_workspace, mock_provider): + adapter = BuiltinPlanAdapter(mock_workspace, mock_provider) + assert isinstance(adapter, AgentAdapter) + + def _make_state(self, status, blocker=None, gate_results=None): + state = MagicMock(spec=AgentState) + state.status = status + state.blocker = blocker + state.gate_results = gate_results or [] + return state + + @pytest.mark.parametrize( + "agent_status, expected", + [ + (AgentStatus.COMPLETED, "completed"), + (AgentStatus.FAILED, "failed"), + (AgentStatus.BLOCKED, "blocked"), + ], + ) + def test_status_mapping( + self, mock_workspace, mock_provider, agent_status, expected + ): + with patch(_PLAN_AGENT_CLS) as mock_cls: + mock_cls.return_value.run.return_value = self._make_state(agent_status) + adapter = BuiltinPlanAdapter(mock_workspace, mock_provider) + result = adapter.run("task-1", "prompt", Path("/tmp")) + assert result.status == expected + + def test_blocked_extracts_blocker_question(self, mock_workspace, mock_provider): + blocker = MagicMock() + blocker.question = "Which database should I use?" + blocker.reason = None + + with patch(_PLAN_AGENT_CLS) as mock_cls: + mock_cls.return_value.run.return_value = self._make_state( + AgentStatus.BLOCKED, blocker=blocker + ) + adapter = BuiltinPlanAdapter(mock_workspace, mock_provider) + result = adapter.run("task-1", "prompt", Path("/tmp")) + assert result.status == "blocked" + assert result.blocker_question == "Which database should I use?" + + def test_blocked_falls_back_to_reason(self, mock_workspace, mock_provider): + blocker = MagicMock() + blocker.question = None + blocker.reason = "Missing dependency" + + with patch(_PLAN_AGENT_CLS) as mock_cls: + mock_cls.return_value.run.return_value = self._make_state( + AgentStatus.BLOCKED, blocker=blocker + ) + adapter = BuiltinPlanAdapter(mock_workspace, mock_provider) + result = adapter.run("task-1", "prompt", Path("/tmp")) + assert result.blocker_question == "Missing dependency" + + def test_failed_extracts_gate_errors(self, mock_workspace, mock_provider): + check = MagicMock() + check.output = "ruff: E501 line too long" + gate = MagicMock() + gate.checks = [check] + + with patch(_PLAN_AGENT_CLS) as mock_cls: + mock_cls.return_value.run.return_value = self._make_state( + AgentStatus.FAILED, gate_results=[gate] + ) + adapter = BuiltinPlanAdapter(mock_workspace, mock_provider) + result = adapter.run("task-1", "prompt", Path("/tmp")) + assert result.status == "failed" + assert "E501" in result.error + + def test_failed_limits_error_output(self, mock_workspace, mock_provider): + checks = [] + for i in range(10): + c = MagicMock() + c.output = f"error {i}" + checks.append(c) + gate = MagicMock() + gate.checks = checks + + with patch(_PLAN_AGENT_CLS) as mock_cls: + mock_cls.return_value.run.return_value = self._make_state( + AgentStatus.FAILED, gate_results=[gate] + ) + adapter = BuiltinPlanAdapter(mock_workspace, mock_provider) + result = adapter.run("task-1", "prompt", Path("/tmp")) + assert result.error.count("\n") == 2 # 3 errors, 2 newlines + + def test_forwards_events(self, mock_workspace, mock_provider): + received: list[AgentEvent] = [] + + def capture_constructor(**kwargs): + cb = kwargs.get("on_event") + if cb: + cb("planning", {"phase": "start"}) + inst = MagicMock() + state = MagicMock(spec=AgentState) + state.status = AgentStatus.COMPLETED + state.blocker = None + state.gate_results = [] + inst.run.return_value = state + return inst + + with patch(_PLAN_AGENT_CLS) as mock_cls: + mock_cls.side_effect = capture_constructor + adapter = BuiltinPlanAdapter(mock_workspace, mock_provider) + adapter.run("task-1", "prompt", Path("/tmp"), on_event=received.append) + + assert len(received) == 1 + assert received[0].type == "planning" diff --git a/tests/core/adapters/test_claude_code.py b/tests/core/adapters/test_claude_code.py new file mode 100644 index 00000000..5affaaa2 --- /dev/null +++ b/tests/core/adapters/test_claude_code.py @@ -0,0 +1,126 @@ +"""Tests for Claude Code adapter.""" + +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest + +from codeframe.core.adapters.agent_adapter import AgentAdapter +from codeframe.core.adapters.claude_code import ClaudeCodeAdapter + + +class TestClaudeCodeAdapter: + """Unit tests for ClaudeCodeAdapter.""" + + def test_name(self) -> None: + with patch("shutil.which", return_value="/usr/bin/claude"): + adapter = ClaudeCodeAdapter() + assert adapter.name == "claude-code" + + def test_conforms_to_protocol(self) -> None: + with patch("shutil.which", return_value="/usr/bin/claude"): + adapter = ClaudeCodeAdapter() + assert isinstance(adapter, AgentAdapter) + + def test_raises_if_claude_not_installed(self) -> None: + with patch("shutil.which", return_value=None): + with pytest.raises(EnvironmentError, match="not found on PATH"): + ClaudeCodeAdapter() + + def test_build_command_includes_print_flag(self) -> None: + with patch("shutil.which", return_value="/usr/bin/claude"): + adapter = ClaudeCodeAdapter() + cmd = adapter.build_command("prompt", Path("/tmp")) + assert cmd[0] == "/usr/bin/claude" + assert "--print" in cmd + + def test_build_command_without_allowlist(self) -> None: + with patch("shutil.which", return_value="/usr/bin/claude"): + adapter = ClaudeCodeAdapter() + cmd = adapter.build_command("prompt", Path("/tmp")) + assert "--allowedTools" not in cmd + + def test_build_command_with_allowlist(self) -> None: + with patch("shutil.which", return_value="/usr/bin/claude"): + adapter = ClaudeCodeAdapter(allowlist=["Edit", "Write"]) + cmd = adapter.build_command("prompt", Path("/tmp")) + assert "--allowedTools" in cmd + # Verify both tools are present after their respective flags + idx_edit = cmd.index("Edit") + idx_write = cmd.index("Write") + assert cmd[idx_edit - 1] == "--allowedTools" + assert cmd[idx_write - 1] == "--allowedTools" + + def test_sends_prompt_via_stdin(self) -> None: + with patch("shutil.which", return_value="/usr/bin/claude"): + adapter = ClaudeCodeAdapter() + assert adapter.get_stdin("my prompt") == "my prompt" + + def test_successful_execution(self) -> None: + with patch("shutil.which", return_value="/usr/bin/claude"): + adapter = ClaudeCodeAdapter() + + mock_process = MagicMock() + mock_process.stdout = iter( + ["Created file src/main.py\n", "All tests pass\n"] + ) + mock_process.stderr = MagicMock() + mock_process.stderr.read.return_value = "" + mock_process.stdin = MagicMock() + mock_process.returncode = 0 + mock_process.wait.return_value = None + + with patch("subprocess.Popen", return_value=mock_process): + result = adapter.run("task-1", "fix the bug", Path("/tmp/repo")) + + assert result.status == "completed" + assert "All tests pass" in result.output + + def test_failed_execution(self) -> None: + with patch("shutil.which", return_value="/usr/bin/claude"): + adapter = ClaudeCodeAdapter() + + mock_process = MagicMock() + mock_process.stdout = iter([]) + mock_process.stderr = MagicMock() + mock_process.stderr.read.return_value = "Traceback: syntax error in config" + mock_process.stdin = MagicMock() + mock_process.returncode = 1 + mock_process.wait.return_value = None + + with patch("subprocess.Popen", return_value=mock_process): + result = adapter.run("task-1", "fix the bug", Path("/tmp/repo")) + + assert result.status == "failed" + + def test_event_callback_receives_output_lines(self) -> None: + with patch("shutil.which", return_value="/usr/bin/claude"): + adapter = ClaudeCodeAdapter() + + events: list = [] + mock_process = MagicMock() + mock_process.stdout = iter(["line one\n", "line two\n"]) + mock_process.stderr = MagicMock() + mock_process.stderr.read.return_value = "" + mock_process.stdin = MagicMock() + mock_process.returncode = 0 + mock_process.wait.return_value = None + + with patch("subprocess.Popen", return_value=mock_process): + adapter.run( + "task-1", "do work", Path("/tmp/repo"), on_event=events.append + ) + + assert len(events) == 2 + assert events[0].data["line"] == "line one" + assert events[1].data["line"] == "line two" + + def test_allowlist_stored(self) -> None: + with patch("shutil.which", return_value="/usr/bin/claude"): + adapter = ClaudeCodeAdapter(allowlist=["Read", "Bash"]) + assert adapter._allowlist == ["Read", "Bash"] + + def test_no_allowlist_defaults_to_none(self) -> None: + with patch("shutil.which", return_value="/usr/bin/claude"): + adapter = ClaudeCodeAdapter() + assert adapter._allowlist is None diff --git a/tests/core/adapters/test_opencode.py b/tests/core/adapters/test_opencode.py new file mode 100644 index 00000000..89518e51 --- /dev/null +++ b/tests/core/adapters/test_opencode.py @@ -0,0 +1,97 @@ +"""Tests for OpenCode adapter.""" + +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest + +from codeframe.core.adapters.agent_adapter import AgentAdapter +from codeframe.core.adapters.opencode import OpenCodeAdapter + + +class TestOpenCodeAdapter: + """Unit tests for OpenCodeAdapter.""" + + def test_name(self) -> None: + with patch("shutil.which", return_value="/usr/bin/opencode"): + adapter = OpenCodeAdapter() + assert adapter.name == "opencode" + + def test_conforms_to_protocol(self) -> None: + with patch("shutil.which", return_value="/usr/bin/opencode"): + adapter = OpenCodeAdapter() + assert isinstance(adapter, AgentAdapter) + + def test_raises_if_opencode_not_installed(self) -> None: + with patch("shutil.which", return_value=None): + with pytest.raises(EnvironmentError, match="not found on PATH"): + OpenCodeAdapter() + + def test_build_command_includes_non_interactive(self) -> None: + with patch("shutil.which", return_value="/usr/bin/opencode"): + adapter = OpenCodeAdapter() + cmd = adapter.build_command("prompt", Path("/tmp")) + assert cmd[0] == "/usr/bin/opencode" + assert "--non-interactive" in cmd + + def test_sends_prompt_via_stdin(self) -> None: + with patch("shutil.which", return_value="/usr/bin/opencode"): + adapter = OpenCodeAdapter() + assert adapter.get_stdin("my prompt") == "my prompt" + + def test_successful_execution(self) -> None: + with patch("shutil.which", return_value="/usr/bin/opencode"): + adapter = OpenCodeAdapter() + + mock_process = MagicMock() + mock_process.stdout = iter(["Updated main.py\n"]) + mock_process.stderr = MagicMock() + mock_process.stderr.read.return_value = "" + mock_process.stdin = MagicMock() + mock_process.returncode = 0 + mock_process.wait.return_value = None + + with patch("subprocess.Popen", return_value=mock_process): + result = adapter.run("task-1", "implement feature", Path("/tmp/repo")) + + assert result.status == "completed" + assert "Updated main.py" in result.output + + def test_failed_execution(self) -> None: + with patch("shutil.which", return_value="/usr/bin/opencode"): + adapter = OpenCodeAdapter() + + mock_process = MagicMock() + mock_process.stdout = iter([]) + mock_process.stderr = MagicMock() + mock_process.stderr.read.return_value = "Fatal error" + mock_process.stdin = MagicMock() + mock_process.returncode = 1 + mock_process.wait.return_value = None + + with patch("subprocess.Popen", return_value=mock_process): + result = adapter.run("task-1", "implement feature", Path("/tmp/repo")) + + assert result.status == "failed" + + def test_event_callback_receives_output_lines(self) -> None: + with patch("shutil.which", return_value="/usr/bin/opencode"): + adapter = OpenCodeAdapter() + + events: list = [] + mock_process = MagicMock() + mock_process.stdout = iter(["line one\n", "line two\n"]) + mock_process.stderr = MagicMock() + mock_process.stderr.read.return_value = "" + mock_process.stdin = MagicMock() + mock_process.returncode = 0 + mock_process.wait.return_value = None + + with patch("subprocess.Popen", return_value=mock_process): + adapter.run( + "task-1", "do work", Path("/tmp/repo"), on_event=events.append + ) + + assert len(events) == 2 + assert events[0].data["line"] == "line one" + assert events[1].data["line"] == "line two" diff --git a/tests/core/adapters/test_subprocess_adapter.py b/tests/core/adapters/test_subprocess_adapter.py new file mode 100644 index 00000000..b998aa32 --- /dev/null +++ b/tests/core/adapters/test_subprocess_adapter.py @@ -0,0 +1,246 @@ +"""Tests for SubprocessAdapter base class.""" + +import subprocess + +import pytest +from pathlib import Path +from unittest.mock import patch, MagicMock + +from codeframe.core.adapters.subprocess_adapter import SubprocessAdapter +from codeframe.core.adapters.agent_adapter import AgentAdapter, AgentEvent + + +class TestSubprocessAdapterInit: + """Tests for adapter initialization and binary detection.""" + + def test_init_with_available_binary(self): + with patch("shutil.which", return_value="/usr/bin/test-agent"): + adapter = SubprocessAdapter("test-agent") + assert adapter.name == "test-agent" + + def test_init_raises_on_missing_binary(self): + with patch("shutil.which", return_value=None): + with pytest.raises(EnvironmentError, match="not found on PATH"): + SubprocessAdapter("nonexistent-agent") + + def test_init_stores_cli_args(self): + with patch("shutil.which", return_value="/usr/bin/agent"): + adapter = SubprocessAdapter("agent", cli_args=["--auto", "--quiet"]) + assert adapter._cli_args == ["--auto", "--quiet"] + + def test_init_defaults_cli_args_to_empty(self): + with patch("shutil.which", return_value="/usr/bin/agent"): + adapter = SubprocessAdapter("agent") + assert adapter._cli_args == [] + + def test_init_stores_resolved_path(self): + with patch("shutil.which", return_value="/opt/bin/my-agent"): + adapter = SubprocessAdapter("my-agent") + assert adapter._binary_path == "/opt/bin/my-agent" + + +class TestSubprocessAdapterRun: + """Tests for subprocess execution.""" + + @pytest.fixture + def adapter(self): + with patch("shutil.which", return_value="/usr/bin/test-agent"): + return SubprocessAdapter("test-agent", cli_args=["--print"]) + + def _make_mock_process( + self, stdout_lines=None, stderr_text="", returncode=0 + ): + mock = MagicMock() + mock.stdout = iter(stdout_lines or []) + mock.stderr = MagicMock() + mock.stderr.read.return_value = stderr_text + mock.stdin = MagicMock() + mock.returncode = returncode + mock.wait.return_value = None + return mock + + def test_successful_execution(self, adapter): + mock_process = self._make_mock_process( + stdout_lines=["line 1\n", "line 2\n"], returncode=0 + ) + with patch("subprocess.Popen", return_value=mock_process): + result = adapter.run("task-1", "fix the bug", Path("/tmp/repo")) + + assert result.status == "completed" + assert "line 1" in result.output + assert "line 2" in result.output + + def test_failed_execution(self, adapter): + mock_process = self._make_mock_process( + stderr_text="Error: tests failed", returncode=1 + ) + with patch("subprocess.Popen", return_value=mock_process): + result = adapter.run("task-1", "fix the bug", Path("/tmp/repo")) + + assert result.status == "failed" + assert result.error == "Error: tests failed" + + def test_failed_with_empty_stderr_uses_exit_code(self, adapter): + mock_process = self._make_mock_process(returncode=42) + with patch("subprocess.Popen", return_value=mock_process): + result = adapter.run("task-1", "fix", Path("/tmp")) + + assert result.status == "failed" + assert "42" in result.error + + def test_blocked_on_permission_denied(self, adapter): + mock_process = self._make_mock_process( + stdout_lines=["permission denied: cannot access repo\n"], returncode=1 + ) + with patch("subprocess.Popen", return_value=mock_process): + result = adapter.run("task-1", "fix the bug", Path("/tmp/repo")) + + assert result.status == "blocked" + assert result.blocker_question is not None + + def test_blocked_on_credentials_required(self, adapter): + mock_process = self._make_mock_process( + stdout_lines=["credentials required to access the service\n"], + returncode=1, + ) + with patch("subprocess.Popen", return_value=mock_process): + result = adapter.run("task-1", "deploy", Path("/tmp/repo")) + + assert result.status == "blocked" + + def test_blocked_on_api_key_missing(self, adapter): + mock_process = self._make_mock_process( + stderr_text="Error: API key not configured", returncode=1 + ) + with patch("subprocess.Popen", return_value=mock_process): + result = adapter.run("task-1", "call api", Path("/tmp")) + + assert result.status == "blocked" + + def test_streams_events(self, adapter): + events: list[AgentEvent] = [] + mock_process = self._make_mock_process( + stdout_lines=["hello\n", "world\n"], returncode=0 + ) + with patch("subprocess.Popen", return_value=mock_process): + adapter.run("task-1", "fix", Path("/tmp"), on_event=events.append) + + assert len(events) == 2 + assert events[0].type == "output" + assert events[0].data["line"] == "hello" + assert events[1].data["line"] == "world" + + def test_no_events_when_callback_is_none(self, adapter): + mock_process = self._make_mock_process( + stdout_lines=["hello\n"], returncode=0 + ) + with patch("subprocess.Popen", return_value=mock_process): + result = adapter.run("task-1", "fix", Path("/tmp"), on_event=None) + + assert result.status == "completed" + + def test_passes_cwd_to_popen(self, adapter): + mock_process = self._make_mock_process(returncode=0) + with patch("subprocess.Popen", return_value=mock_process) as mock_popen: + adapter.run("task-1", "fix", Path("/my/repo")) + mock_popen.assert_called_once() + assert mock_popen.call_args.kwargs["cwd"] == "/my/repo" + + def test_sends_prompt_via_stdin(self, adapter): + mock_process = self._make_mock_process(returncode=0) + with patch("subprocess.Popen", return_value=mock_process): + adapter.run("task-1", "my prompt text", Path("/tmp")) + mock_process.stdin.write.assert_called_once_with("my prompt text") + mock_process.stdin.close.assert_called_once() + + def test_handles_oserror(self, adapter): + with patch("subprocess.Popen", side_effect=OSError("spawn failed")): + result = adapter.run("task-1", "fix", Path("/tmp")) + assert result.status == "failed" + assert "spawn failed" in result.error + + def test_handles_file_not_found_error(self, adapter): + with patch("subprocess.Popen", side_effect=FileNotFoundError()): + result = adapter.run("task-1", "fix", Path("/tmp")) + assert result.status == "failed" + assert "not found during execution" in result.error + + def test_conforms_to_agent_adapter_protocol(self, adapter): + assert isinstance(adapter, AgentAdapter) + + def test_timeout_kills_process(self): + """Process should be killed when timeout expires.""" + with patch("shutil.which", return_value="/usr/bin/agent"): + adapter = SubprocessAdapter("agent", timeout_s=1) + + mock_process = MagicMock() + mock_process.stdout = iter(["working...\n"]) + mock_process.stderr = MagicMock() + mock_process.stderr.read.return_value = "" + mock_process.stdin = MagicMock() + mock_process.wait.side_effect = [ + subprocess.TimeoutExpired(cmd="agent", timeout=1), + None, + ] + mock_process.returncode = -9 + + with patch("subprocess.Popen", return_value=mock_process): + result = adapter.run("task-1", "fix", Path("/tmp")) + + assert result.status == "failed" + assert "timed out" in result.error + mock_process.kill.assert_called_once() + + +class TestSubprocessAdapterBuildCommand: + """Tests for command building.""" + + def test_default_build_command(self): + with patch("shutil.which", return_value="/usr/bin/agent"): + adapter = SubprocessAdapter("agent", cli_args=["--auto"]) + cmd = adapter.build_command("prompt", Path("/tmp")) + assert cmd == ["/usr/bin/agent", "--auto"] + + def test_build_command_without_args(self): + with patch("shutil.which", return_value="/usr/bin/agent"): + adapter = SubprocessAdapter("agent") + cmd = adapter.build_command("prompt", Path("/tmp")) + assert cmd == ["/usr/bin/agent"] + + +class TestSubprocessAdapterGetStdin: + """Tests for stdin content generation.""" + + def test_default_returns_prompt(self): + with patch("shutil.which", return_value="/usr/bin/agent"): + adapter = SubprocessAdapter("agent") + assert adapter.get_stdin("hello world") == "hello world" + + def test_default_returns_empty_prompt(self): + with patch("shutil.which", return_value="/usr/bin/agent"): + adapter = SubprocessAdapter("agent") + assert adapter.get_stdin("") == "" + + +class TestSubprocessAdapterBlockerExtraction: + """Tests for blocker question extraction.""" + + def test_extracts_last_line(self): + with patch("shutil.which", return_value="/usr/bin/agent"): + adapter = SubprocessAdapter("agent") + question = adapter._extract_blocker_question( + "Starting...\nChecking...\nPermission denied for /secret/file" + ) + assert question == "Permission denied for /secret/file" + + def test_handles_empty_output(self): + with patch("shutil.which", return_value="/usr/bin/agent"): + adapter = SubprocessAdapter("agent") + question = adapter._extract_blocker_question("") + assert "no details" in question.lower() + + def test_handles_blank_lines_only(self): + with patch("shutil.which", return_value="/usr/bin/agent"): + adapter = SubprocessAdapter("agent") + question = adapter._extract_blocker_question("\n\n \n") + assert "no details" in question.lower() diff --git a/tests/core/adapters/test_verification_wrapper.py b/tests/core/adapters/test_verification_wrapper.py new file mode 100644 index 00000000..2e5d8b20 --- /dev/null +++ b/tests/core/adapters/test_verification_wrapper.py @@ -0,0 +1,180 @@ +"""Tests for VerificationWrapper.""" + +import pytest +from pathlib import Path +from unittest.mock import MagicMock, patch + +from codeframe.core.adapters.verification_wrapper import VerificationWrapper +from codeframe.core.adapters.agent_adapter import AgentAdapter, AgentEvent, AgentResult +from codeframe.core.gates import GateStatus + + +@pytest.fixture +def mock_workspace(): + ws = MagicMock() + ws.repo_path = Path("/tmp/test-repo") + return ws + + +@pytest.fixture +def mock_inner_adapter(): + adapter = MagicMock(spec=AgentAdapter) + adapter.name = "mock" + adapter.run.return_value = AgentResult(status="completed", output="done") + return adapter + + +@pytest.fixture +def passing_gate_result(): + result = MagicMock() + result.passed = True + result.checks = [] + return result + + +@pytest.fixture +def failing_gate_result(): + check = MagicMock() + check.name = "pytest" + check.status = GateStatus.FAILED + check.output = "FAILED test_main.py::test_foo - AssertionError" + result = MagicMock() + result.passed = False + result.checks = [check] + return result + + +class TestVerificationWrapper: + def test_name_includes_inner(self, mock_inner_adapter, mock_workspace): + wrapper = VerificationWrapper(mock_inner_adapter, mock_workspace) + assert wrapper.name == "verified-mock" + + def test_passes_through_non_completed(self, mock_inner_adapter, mock_workspace): + """If inner adapter fails/blocks, skip verification entirely.""" + mock_inner_adapter.run.return_value = AgentResult( + status="failed", error="crash", + ) + wrapper = VerificationWrapper(mock_inner_adapter, mock_workspace) + result = wrapper.run("t1", "prompt", Path("/tmp")) + assert result.status == "failed" + + def test_runs_gates_on_completed( + self, mock_inner_adapter, mock_workspace, passing_gate_result, + ): + with patch( + "codeframe.core.adapters.verification_wrapper.run_gates", + return_value=passing_gate_result, + ): + wrapper = VerificationWrapper(mock_inner_adapter, mock_workspace) + result = wrapper.run("t1", "prompt", Path("/tmp")) + assert result.status == "completed" + + def test_self_correction_on_gate_failure( + self, + mock_inner_adapter, + mock_workspace, + failing_gate_result, + passing_gate_result, + ): + """After gate failure, re-invoke adapter with error context, then pass.""" + with patch( + "codeframe.core.adapters.verification_wrapper.run_gates", + ) as mock_gates: + # First call: gates fail. Second call (after correction): gates pass. + mock_gates.side_effect = [failing_gate_result, passing_gate_result] + + wrapper = VerificationWrapper( + mock_inner_adapter, mock_workspace, max_correction_rounds=3, + ) + result = wrapper.run("t1", "prompt", Path("/tmp")) + + assert result.status == "completed" + # Inner adapter called twice: initial + 1 correction + assert mock_inner_adapter.run.call_count == 2 + # Second call should include error context in prompt + second_prompt = mock_inner_adapter.run.call_args_list[1][0][1] + assert "Verification Gate Failures" in second_prompt + + def test_exhausted_correction_rounds( + self, mock_inner_adapter, mock_workspace, failing_gate_result, + ): + """If all correction rounds fail, return failed result.""" + with patch( + "codeframe.core.adapters.verification_wrapper.run_gates", + return_value=failing_gate_result, + ): + wrapper = VerificationWrapper( + mock_inner_adapter, mock_workspace, max_correction_rounds=2, + ) + result = wrapper.run("t1", "prompt", Path("/tmp")) + + assert result.status == "failed" + assert "still failing after 2 correction rounds" in result.error + + def test_emits_verification_events( + self, mock_inner_adapter, mock_workspace, passing_gate_result, + ): + events: list[AgentEvent] = [] + with patch( + "codeframe.core.adapters.verification_wrapper.run_gates", + return_value=passing_gate_result, + ): + wrapper = VerificationWrapper(mock_inner_adapter, mock_workspace) + wrapper.run("t1", "prompt", Path("/tmp"), on_event=events.append) + + types = [e.type for e in events] + assert "verification" in types + assert "verification_passed" in types + + def test_correction_stops_on_inner_failure( + self, mock_inner_adapter, mock_workspace, failing_gate_result, + ): + """If inner adapter fails during correction, stop immediately.""" + mock_inner_adapter.run.side_effect = [ + AgentResult(status="completed", output="v1"), + AgentResult(status="failed", error="crash on retry"), + ] + with patch( + "codeframe.core.adapters.verification_wrapper.run_gates", + return_value=failing_gate_result, + ): + wrapper = VerificationWrapper( + mock_inner_adapter, mock_workspace, max_correction_rounds=3, + ) + result = wrapper.run("t1", "prompt", Path("/tmp")) + assert result.status == "failed" + assert result.error == "crash on retry" + + def test_conforms_to_agent_adapter_protocol( + self, mock_inner_adapter, mock_workspace, + ): + wrapper = VerificationWrapper(mock_inner_adapter, mock_workspace) + assert isinstance(wrapper, AgentAdapter) + + def test_format_gate_errors_with_output(self, failing_gate_result): + summary = VerificationWrapper._format_gate_errors(failing_gate_result) + assert "pytest" in summary + assert "FAILED" in summary + assert "test_main.py" in summary + + def test_format_gate_errors_no_failures(self, passing_gate_result): + summary = VerificationWrapper._format_gate_errors(passing_gate_result) + assert "no details available" in summary + + def test_custom_gate_names( + self, mock_inner_adapter, mock_workspace, passing_gate_result, + ): + """Gate names are forwarded to run_gates.""" + with patch( + "codeframe.core.adapters.verification_wrapper.run_gates", + return_value=passing_gate_result, + ) as mock_gates: + wrapper = VerificationWrapper( + mock_inner_adapter, + mock_workspace, + gate_names=["ruff"], + ) + wrapper.run("t1", "prompt", Path("/tmp")) + mock_gates.assert_called_once_with( + mock_workspace, gates=["ruff"], verbose=False, + ) diff --git a/tests/core/test_context_packager.py b/tests/core/test_context_packager.py new file mode 100644 index 00000000..3f176a15 --- /dev/null +++ b/tests/core/test_context_packager.py @@ -0,0 +1,138 @@ +"""Tests for TaskContextPackager.""" + +import pytest +from pathlib import Path +from unittest.mock import MagicMock, patch + +from codeframe.core.context_packager import TaskContextPackager, PackagedContext +from codeframe.core.context import TaskContext + + +@pytest.fixture +def mock_workspace(): + ws = MagicMock() + ws.repo_path = Path("/tmp/test-repo") + ws.state_dir = Path("/tmp/test-repo/.codeframe") + return ws + + +@pytest.fixture +def mock_task_context(): + ctx = MagicMock(spec=TaskContext) + ctx.to_prompt_context.return_value = ( + "## Task\n**Title:** Fix the bug\n**Description:** Fix it\n" + ) + return ctx + + +class TestPackagedContext: + """Tests for the PackagedContext dataclass.""" + + def test_stores_prompt_and_context(self, mock_task_context): + pc = PackagedContext(prompt="hello", context=mock_task_context) + assert pc.prompt == "hello" + assert pc.context is mock_task_context + + +class TestTaskContextPackager: + """Tests for TaskContextPackager.""" + + def test_build_returns_packaged_context(self, mock_workspace, mock_task_context): + with patch("codeframe.core.context_packager.ContextLoader") as MockLoader: + MockLoader.return_value.load.return_value = mock_task_context + + packager = TaskContextPackager(mock_workspace) + result = packager.build("task-1") + + assert isinstance(result, PackagedContext) + assert isinstance(result.prompt, str) + assert result.context is mock_task_context + + def test_build_includes_base_context(self, mock_workspace, mock_task_context): + with patch("codeframe.core.context_packager.ContextLoader") as MockLoader: + MockLoader.return_value.load.return_value = mock_task_context + + packager = TaskContextPackager(mock_workspace) + result = packager.build("task-1") + + assert "Fix the bug" in result.prompt + + def test_build_includes_default_gates(self, mock_workspace, mock_task_context): + with patch("codeframe.core.context_packager.ContextLoader") as MockLoader: + MockLoader.return_value.load.return_value = mock_task_context + + packager = TaskContextPackager(mock_workspace) + result = packager.build("task-1") + + assert "pytest" in result.prompt + assert "ruff" in result.prompt + assert "Verification Gates" in result.prompt + + def test_build_with_custom_gates(self, mock_workspace, mock_task_context): + with patch("codeframe.core.context_packager.ContextLoader") as MockLoader: + MockLoader.return_value.load.return_value = mock_task_context + + packager = TaskContextPackager(mock_workspace) + result = packager.build("task-1", gate_names=["pytest", "ruff", "mypy"]) + + assert "mypy" in result.prompt + assert "Must pass" in result.prompt + + def test_build_with_only_custom_gates_omits_defaults( + self, mock_workspace, mock_task_context + ): + with patch("codeframe.core.context_packager.ContextLoader") as MockLoader: + MockLoader.return_value.load.return_value = mock_task_context + + packager = TaskContextPackager(mock_workspace) + result = packager.build("task-1", gate_names=["mypy"]) + + assert "mypy" in result.prompt + # Default gates should NOT appear since we overrode + assert "ruff" not in result.prompt + + def test_build_includes_execution_instructions( + self, mock_workspace, mock_task_context + ): + with patch("codeframe.core.context_packager.ContextLoader") as MockLoader: + MockLoader.return_value.load.return_value = mock_task_context + + packager = TaskContextPackager(mock_workspace) + result = packager.build("task-1") + + assert "Execution Instructions" in result.prompt + assert "Do not modify unrelated files" in result.prompt + + def test_build_calls_loader_with_task_id(self, mock_workspace, mock_task_context): + with patch("codeframe.core.context_packager.ContextLoader") as MockLoader: + MockLoader.return_value.load.return_value = mock_task_context + + packager = TaskContextPackager(mock_workspace) + packager.build("task-42") + + MockLoader.return_value.load.assert_called_once_with("task-42") + + def test_prompt_ordering(self, mock_workspace, mock_task_context): + """Verify the prompt sections appear in the correct order.""" + with patch("codeframe.core.context_packager.ContextLoader") as MockLoader: + MockLoader.return_value.load.return_value = mock_task_context + + packager = TaskContextPackager(mock_workspace) + result = packager.build("task-1") + + base_pos = result.prompt.index("Fix the bug") + gates_pos = result.prompt.index("Verification Gates") + instr_pos = result.prompt.index("Execution Instructions") + + assert base_pos < gates_pos < instr_pos + + def test_empty_gate_list(self, mock_workspace, mock_task_context): + """An empty gate list should still produce a valid prompt.""" + with patch("codeframe.core.context_packager.ContextLoader") as MockLoader: + MockLoader.return_value.load.return_value = mock_task_context + + packager = TaskContextPackager(mock_workspace) + result = packager.build("task-1", gate_names=[]) + + assert "Verification Gates" in result.prompt + assert isinstance(result.prompt, str) diff --git a/tests/core/test_engine_registry.py b/tests/core/test_engine_registry.py new file mode 100644 index 00000000..a596750a --- /dev/null +++ b/tests/core/test_engine_registry.py @@ -0,0 +1,151 @@ +"""Tests for engine registry.""" + +import os + +import pytest +from unittest.mock import MagicMock, patch + +from codeframe.core.adapters.agent_adapter import AgentAdapter +from codeframe.core.engine_registry import ( + BUILTIN_ENGINES, + EXTERNAL_ENGINES, + VALID_ENGINES, + get_adapter, + get_builtin_adapter, + get_external_adapter, + is_external_engine, + resolve_engine, +) + + +class TestResolveEngine: + def test_cli_flag_wins(self): + assert resolve_engine("claude-code") == "claude-code" + + def test_env_var_fallback(self): + with patch.dict(os.environ, {"CODEFRAME_ENGINE": "opencode"}): + assert resolve_engine(None) == "opencode" + + def test_default_is_react(self): + with patch.dict(os.environ, {}, clear=True): + os.environ.pop("CODEFRAME_ENGINE", None) + assert resolve_engine(None) == "react" + + def test_cli_overrides_env(self): + with patch.dict(os.environ, {"CODEFRAME_ENGINE": "opencode"}): + assert resolve_engine("plan") == "plan" + + def test_built_in_alias(self): + assert resolve_engine("built-in") == "react" + + def test_invalid_engine_raises(self): + with pytest.raises(ValueError, match="Invalid engine"): + resolve_engine("nonexistent") + + def test_all_valid_engines_resolve(self): + for engine in VALID_ENGINES: + result = resolve_engine(engine) + assert result in VALID_ENGINES + + +class TestIsExternalEngine: + def test_claude_code_is_external(self): + assert is_external_engine("claude-code") is True + + def test_opencode_is_external(self): + assert is_external_engine("opencode") is True + + def test_react_is_not_external(self): + assert is_external_engine("react") is False + + def test_plan_is_not_external(self): + assert is_external_engine("plan") is False + + +class TestGetExternalAdapter: + def test_claude_code_adapter(self): + with patch("shutil.which", return_value="/usr/bin/claude"): + adapter = get_external_adapter("claude-code") + assert adapter.name == "claude-code" + assert isinstance(adapter, AgentAdapter) + + def test_opencode_adapter(self): + with patch("shutil.which", return_value="/usr/bin/opencode"): + adapter = get_external_adapter("opencode") + assert adapter.name == "opencode" + assert isinstance(adapter, AgentAdapter) + + def test_invalid_engine_raises(self): + with pytest.raises(ValueError, match="Unknown external engine"): + get_external_adapter("react") + + def test_missing_binary_raises(self): + with patch("shutil.which", return_value=None): + with pytest.raises(EnvironmentError): + get_external_adapter("claude-code") + + +class TestGetBuiltinAdapter: + def test_react_adapter(self): + ws = MagicMock() + provider = MagicMock() + adapter = get_builtin_adapter("react", ws, provider) + assert adapter.name == "react" + assert isinstance(adapter, AgentAdapter) + + def test_plan_adapter(self): + ws = MagicMock() + provider = MagicMock() + adapter = get_builtin_adapter("plan", ws, provider) + assert adapter.name == "plan" + assert isinstance(adapter, AgentAdapter) + + def test_built_in_alias_maps_to_react(self): + ws = MagicMock() + provider = MagicMock() + adapter = get_builtin_adapter("built-in", ws, provider) + assert adapter.name == "react" + + def test_invalid_engine_raises(self): + with pytest.raises(ValueError, match="Unknown builtin engine"): + get_builtin_adapter("claude-code", MagicMock(), MagicMock()) + + +class TestGetAdapter: + def test_external_engine_no_workspace_needed(self): + with patch("shutil.which", return_value="/usr/bin/claude"): + adapter = get_adapter("claude-code") + assert adapter.name == "claude-code" + + def test_builtin_requires_workspace(self): + with pytest.raises(ValueError, match="requires workspace"): + get_adapter("react") + + def test_builtin_with_workspace(self): + ws = MagicMock() + provider = MagicMock() + adapter = get_adapter("react", workspace=ws, llm_provider=provider) + assert adapter.name == "react" + + def test_passes_kwargs_to_external(self): + with patch("shutil.which", return_value="/usr/bin/claude"): + adapter = get_adapter("claude-code", allowlist=["Edit"]) + assert adapter.name == "claude-code" + + def test_passes_kwargs_to_builtin(self): + ws = MagicMock() + provider = MagicMock() + adapter = get_adapter( + "react", workspace=ws, llm_provider=provider, dry_run=True + ) + assert adapter.name == "react" + + +class TestConstants: + def test_external_and_builtin_cover_all_valid(self): + """Every valid engine is either external or builtin.""" + assert VALID_ENGINES == EXTERNAL_ENGINES | BUILTIN_ENGINES + + def test_no_overlap(self): + """External and builtin sets are disjoint.""" + assert EXTERNAL_ENGINES & BUILTIN_ENGINES == frozenset() diff --git a/tests/core/test_runtime_adapters.py b/tests/core/test_runtime_adapters.py new file mode 100644 index 00000000..08c809b2 --- /dev/null +++ b/tests/core/test_runtime_adapters.py @@ -0,0 +1,138 @@ +"""Tests for runtime execute_agent with external engine adapters.""" + +import pytest +from unittest.mock import MagicMock, patch + +from codeframe.core.adapters.agent_adapter import AgentResult + + +@pytest.fixture +def mock_workspace(tmp_path): + """Create a minimal workspace for testing.""" + ws = MagicMock() + ws.repo_path = tmp_path + ws.state_dir = tmp_path / ".codeframe" + ws.state_dir.mkdir() + ws.id = "test-ws" + ws.tech_stack = None + ws.db_path = str(tmp_path / "test.db") + return ws + + +@pytest.fixture +def mock_run(): + """Create a minimal run record.""" + run = MagicMock() + run.id = "run-1" + run.task_id = "task-1" + return run + + +def _runtime_patches(): + """Common patches for runtime tests that touch the database/filesystem.""" + return [ + patch("codeframe.core.runtime.get_db_connection"), + patch("codeframe.core.runtime.events"), + patch("codeframe.core.diagnostics.get_db_connection"), + patch("codeframe.core.streaming.RunOutputLogger"), + ] + + +class TestExecuteAgentExternalEngine: + """Tests for external engine path in execute_agent.""" + + def test_invalid_engine_raises(self, mock_workspace, mock_run): + from codeframe.core.runtime import execute_agent + + with pytest.raises(ValueError, match="Invalid engine"): + execute_agent(mock_workspace, mock_run, engine="nonexistent") + + def test_external_engine_skips_api_key_check(self, mock_workspace, mock_run): + """External engines should not require ANTHROPIC_API_KEY.""" + from codeframe.core.runtime import execute_agent + + patches = _runtime_patches() + [ + patch( + "codeframe.core.runtime.get_external_adapter", + create=True, + ), + patch("codeframe.core.context_packager.ContextLoader"), + patch("codeframe.core.runtime.complete_run"), + patch("codeframe.core.adapters.verification_wrapper.run_gates"), + ] + + with patch.dict("os.environ", {}, clear=True): + import os + os.environ.pop("ANTHROPIC_API_KEY", None) + + # Apply all patches + mocks = {} + for p in patches: + m = p.start() + mocks[p.attribute or ""] = m + + try: + # Set up context loader mock + mock_context = MagicMock() + mock_context.to_prompt_context.return_value = "test context" + from codeframe.core.context_packager import TaskContextPackager + with patch.object( + TaskContextPackager, "build" + ) as mock_build: + from codeframe.core.context_packager import PackagedContext + mock_build.return_value = PackagedContext( + prompt="test prompt", context=mock_context + ) + + # Mock the adapter returned by get_external_adapter + mock_adapter = MagicMock() + mock_adapter.name = "claude-code" + mock_adapter.run.return_value = AgentResult( + status="completed", output="done" + ) + + # Patch at the point of import inside runtime + with patch( + "codeframe.core.engine_registry.get_external_adapter", + return_value=mock_adapter, + ): + # Mock gate passing + with patch( + "codeframe.core.adapters.verification_wrapper.run_gates" + ) as mock_gates: + mock_gate_result = MagicMock() + mock_gate_result.passed = True + mock_gates.return_value = mock_gate_result + + result = execute_agent( + mock_workspace, + mock_run, + engine="claude-code", + ) + assert result.status.value == "completed" + finally: + for p in patches: + p.stop() + + def test_builtin_engine_still_requires_api_key( + self, mock_workspace, mock_run + ): + """Builtin engines should still require ANTHROPIC_API_KEY.""" + from codeframe.core.runtime import execute_agent + + with patch.dict("os.environ", {}, clear=True): + import os + os.environ.pop("ANTHROPIC_API_KEY", None) + + with pytest.raises(ValueError, match="ANTHROPIC_API_KEY"): + execute_agent(mock_workspace, mock_run, engine="react") + + def test_valid_engines_include_external(self): + """VALID_ENGINES should include external engine names.""" + from codeframe.core.engine_registry import VALID_ENGINES + + assert "claude-code" in VALID_ENGINES + assert "opencode" in VALID_ENGINES + assert "react" in VALID_ENGINES + assert "plan" in VALID_ENGINES + assert "built-in" in VALID_ENGINES