diff --git a/src/deep_code_agent/code_agent.py b/src/deep_code_agent/code_agent.py index 564b806..c89a85d 100644 --- a/src/deep_code_agent/code_agent.py +++ b/src/deep_code_agent/code_agent.py @@ -9,6 +9,7 @@ - Code refactoring and optimization """ +import os from pathlib import Path from typing import Any @@ -36,71 +37,57 @@ def create_code_agent( It sets up the necessary backend and tools based on the specified backend type. Args: - codebase_dir (str): Absolute or relative path to the codebase directory. + codebase_dir: Absolute or relative path to the codebase directory. The directory will be created if it does not exist. - model (BaseChatModel | None, optional): Language model instance to power - the agent. If None, a default chat model will be created. - checkpointer (Any | None, optional): Optional checkpointer for persisting - agent state across sessions. - backend_type (str, optional): Backend type to use. Must be either "state" - (StateBackend) or "filesystem" (FilesystemBackend). Defaults to "state". - interrupt_on (dict[str, bool | Any] | None, optional): Configuration for - human-in-the-loop approval. Maps tool names to approval settings. - Defaults to DEFAULT_INTERRUPT_ON which enables approvals for - write_file, edit_file, execute, and terminal. Set to None to disable - all approvals. + model: Language model instance to power the agent. If None, a default + chat model will be created. + checkpointer: Optional checkpointer for persisting agent state across sessions. + backend_type: Backend type to use. Must be either "state" or "filesystem". + interrupt_on: Configuration for human-in-the-loop approval. Maps tool names + to approval settings. Set to None to disable all approvals. Returns: - DeepAgent: A fully configured Code Agent instance ready to handle - software development tasks. + A fully configured Code Agent instance ready to handle software development tasks. Raises: - PermissionError: If the agent lacks permission to create the codebase directory. - OSError: If an OS-level error occurs while creating the codebase directory. RuntimeError: If the DeepAgent creation fails for any reason. + ValueError: If an unsupported backend type is provided. """ - # Ensure codebase directory exists - try: - path = Path(codebase_dir).absolute() - if backend_type == "filesystem" and not path.exists(): - path.mkdir(parents=True, exist_ok=True) - codebase_dir = path.as_posix() - except PermissionError as e: - raise PermissionError(f"Permission denied creating directory '{codebase_dir}'") from e - except OSError as e: - raise OSError(f"Error creating directory '{codebase_dir}': {str(e)}") from e - - # Get system prompt - system_prompt = get_system_prompt() - - # Create subagent configurations + path = Path(codebase_dir) + if backend_type == "filesystem" and not path.exists(): + path.mkdir(parents=True, exist_ok=True) + + codebase_dir = path.absolute().as_posix() + system_prompt = get_system_prompt().format(codebase_dir=codebase_dir) + model = model or create_chat_model() subagents = create_subagent_configurations() - # Create and configure backend if backend_type == "filesystem": from deepagents.backends.filesystem import FilesystemBackend backend = FilesystemBackend(root_dir=codebase_dir) tools = [terminal] - else: + os.environ["DEEP_CODE_AGENT_TERMINAL_CWD"] = codebase_dir + elif backend_type == "state": from deepagents.backends.state import StateBackend - def _backend_factory(rt): + def backend(rt): return StateBackend(rt) - backend = _backend_factory tools = [] + os.environ.pop("DEEP_CODE_AGENT_TERMINAL_CWD", None) + else: + raise ValueError(f"Unsupported backend_type: {backend_type}") - # Create and return agent try: return create_deep_agent( - model=model or create_chat_model(), - system_prompt=system_prompt.format(codebase_dir=codebase_dir), + model=model, tools=tools, - subagents=list(subagents), + subagents=subagents, + system_prompt=system_prompt, checkpointer=checkpointer, backend=backend, interrupt_on=interrupt_on, ) - except Exception as e: - raise RuntimeError(f"Error creating DeepAgent: {str(e)}") from e + except Exception as exc: + raise RuntimeError(f"Error creating DeepAgent: {exc}") from exc diff --git a/src/deep_code_agent/tools/terminal.py b/src/deep_code_agent/tools/terminal.py index 0a9ef4d..ee945dd 100644 --- a/src/deep_code_agent/tools/terminal.py +++ b/src/deep_code_agent/tools/terminal.py @@ -1,74 +1,96 @@ """Terminal command execution tool.""" import os +import shlex import subprocess from langchain_core.tools import tool from deep_code_agent.config import DEFAULT_TIMEOUT, MAX_TIMEOUT +DISALLOWED_SHELL_SYNTAX = ("&&", "||", "|", ";", "$(", "`", ">", "<", "\n", "\r") +DANGEROUS_COMMAND_SNIPPETS = ( + "rm -rf /", + "format", + "del /q", + "mkfs", + "shutdown", + "reboot", + "diskutil erasedisk", +) + + +def _get_command_workdir() -> str: + """Resolve the agent-scoped working directory for terminal commands.""" + return os.environ.get("DEEP_CODE_AGENT_TERMINAL_CWD", os.getcwd()) + + +def _contains_disallowed_shell_syntax(command: str) -> bool: + """Reject shell-only syntax so commands can run without a shell.""" + return any(token in command for token in DISALLOWED_SHELL_SYNTAX) + @tool def terminal(command: str, timeout: int = DEFAULT_TIMEOUT) -> str: """Execute terminal commands with timeout protection. Args: - command (str): Terminal command to execute. - timeout (int): Command timeout in seconds, defaults to 30. + command: Terminal command to execute. + timeout: Command timeout in seconds, defaults to 30. Returns: - str: Command execution result. + Command execution result. """ - # Validate timeout if timeout <= 0: return f"Error: Timeout must be positive, got {timeout}" if timeout > MAX_TIMEOUT: return f"Error: Timeout {timeout} exceeds maximum allowed {MAX_TIMEOUT} seconds" + if not command or not command.strip(): + return "Error: Command cannot be empty" + + normalized_command = command.strip() + lower_command = normalized_command.lower() + + for dangerous in DANGEROUS_COMMAND_SNIPPETS: + if dangerous in lower_command: + return f"Error: Command contains potentially dangerous operation: {dangerous}" + + if _contains_disallowed_shell_syntax(normalized_command): + return "Error: Command contains disallowed shell control operators." + try: - # Validate command - if not command or not command.strip(): - return "Error: Command cannot be empty" + argv = shlex.split(normalized_command) + except ValueError as exc: + return f"Error: Invalid command syntax: {exc}" - # Security check - block dangerous commands - dangerous_commands = ["rm -rf /", "format", "del /q"] - for dangerous in dangerous_commands: - if dangerous in command.lower(): - return f"Error: Command contains potentially dangerous operation: {dangerous}" + if not argv: + return "Error: Command cannot be empty" - # Execute command + try: result = subprocess.run( - command, - shell=True, + argv, capture_output=True, text=True, timeout=timeout, - cwd=os.getcwd(), + cwd=_get_command_workdir(), ) - # Build output output_parts = [] - if result.stdout: output_parts.append(result.stdout) if result.stderr: output_parts.append(f"STDERR:\n{result.stderr}") - # Add status info status_info = f"Command executed with exit code: {result.returncode}" if result.returncode != 0: status_info += " (non-zero exit code indicates potential error)" output_parts.append(status_info) - - return "\n".join(output_parts) if output_parts else "Command executed successfully with no output." + return "\n".join(output_parts) except subprocess.TimeoutExpired: return f"Error: Command timed out after {timeout} seconds." - except FileNotFoundError: - return "Error: Command shell not found. Please ensure shell is available." except PermissionError: - return f"Error: Permission denied executing command '{command}'" - except OSError as e: - return f"Error: OS error executing command '{command}': {str(e)}" - except Exception as e: - return f"Error executing command '{command}': {str(e)}" + return "Error: Permission denied to execute command." + except Exception as exc: + return f"Error executing command: {exc}" diff --git a/src/deep_code_agent/tui/screens/approval_modal.py b/src/deep_code_agent/tui/screens/approval_modal.py index f0294fa..1d2d97c 100644 --- a/src/deep_code_agent/tui/screens/approval_modal.py +++ b/src/deep_code_agent/tui/screens/approval_modal.py @@ -272,7 +272,7 @@ def action_confirm_selection(self) -> None: def action_cancel(self) -> None: """Cancel the modal.""" - self.dismiss() + self._cancel() def _approve(self) -> None: """Approve the action.""" @@ -294,4 +294,6 @@ def _reject(self) -> None: def _cancel(self) -> None: """Cancel the modal.""" + decision = {"type": "reject", "message": "Action cancelled by user"} + self.callback(decision) self.dismiss() diff --git a/tests/test_terminal.py b/tests/test_terminal.py index 4e39f40..bc16e79 100644 --- a/tests/test_terminal.py +++ b/tests/test_terminal.py @@ -26,6 +26,10 @@ def test_blocks_dangerous_commands(self): result = terminal.invoke({"command": "RM -RF /"}) assert result == "Error: Command contains potentially dangerous operation: rm -rf /" + def test_rejects_shell_control_operators(self): + result = terminal.invoke({"command": "echo hi && whoami"}) + assert result == "Error: Command contains disallowed shell control operators." + @patch("deep_code_agent.tools.terminal.subprocess.run") def test_returns_stdout_stderr_and_exit_status(self, mock_run): mock_run.return_value = MagicMock(stdout="hello\n", stderr="warn\n", returncode=1) @@ -38,6 +42,15 @@ def test_returns_stdout_stderr_and_exit_status(self, mock_run): assert "non-zero exit code indicates potential error" in result mock_run.assert_called_once() + @patch.dict("deep_code_agent.tools.terminal.os.environ", {"DEEP_CODE_AGENT_TERMINAL_CWD": "/tmp/agent-root"}) + @patch("deep_code_agent.tools.terminal.subprocess.run") + def test_uses_agent_scoped_workdir_when_configured(self, mock_run): + mock_run.return_value = MagicMock(stdout="ok\n", stderr="", returncode=0) + + terminal.invoke({"command": "pwd"}) + + assert mock_run.call_args.kwargs["cwd"] == "/tmp/agent-root" + @patch("deep_code_agent.tools.terminal.subprocess.run") def test_timeout_expired_is_reported(self, mock_run): mock_run.side_effect = subprocess.TimeoutExpired(cmd="sleep 5", timeout=1) @@ -52,4 +65,12 @@ def test_permission_error_is_reported(self, mock_run): result = terminal.invoke({"command": "restricted"}) - assert result == "Error: Permission denied executing command 'restricted'" + assert result == "Error: Permission denied to execute command." + + @patch("deep_code_agent.tools.terminal.subprocess.run") + def test_unexpected_error_is_reported(self, mock_run): + mock_run.side_effect = RuntimeError("boom") + + result = terminal.invoke({"command": "echo hi"}) + + assert result == "Error executing command: boom" diff --git a/tests/tui/test_approval_modal.py b/tests/tui/test_approval_modal.py index 8bd5a67..2dd3041 100644 --- a/tests/tui/test_approval_modal.py +++ b/tests/tui/test_approval_modal.py @@ -90,3 +90,35 @@ async def run_test(): assert modal.selected_index == 3 asyncio.run(run_test()) + + +def test_approval_modal_cancel_emits_rejection_decision(): + """Cancel should resolve the interrupt instead of leaving it hanging.""" + import asyncio + from deep_code_agent.tui.screens.approval_modal import ApprovalModal + + async def run_test(): + app = App() + async with app.run_test() as pilot: + interrupt_data = { + "action_requests": [ + {"action": {"name": "read_file", "args": {}}} + ] + } + + results = {} + + def callback(decision): + results["decision"] = decision + + modal = ApprovalModal(interrupt_data, callback=callback) + await app.push_screen(modal) + + modal.action_cancel() + + assert results["decision"] == { + "type": "reject", + "message": "Action cancelled by user", + } + + asyncio.run(run_test())