fix(agent): harden terminal execution and unblock cancelled approvals

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

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}"

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()

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"

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())