diff --git a/src/claude_code_sdk/_internal/transport/subprocess_cli.py b/src/claude_code_sdk/_internal/transport/subprocess_cli.py index 288b2e7..d84626a 100644 --- a/src/claude_code_sdk/_internal/transport/subprocess_cli.py +++ b/src/claude_code_sdk/_internal/transport/subprocess_cli.py @@ -36,6 +36,7 @@ class SubprocessCLITransport(Transport): self._process: Process | None = None self._stdout_stream: TextReceiveStream | None = None self._stderr_stream: TextReceiveStream | None = None + self._received_any_output = False def _find_cli(self) -> str: """Find Claude Code CLI binary.""" @@ -75,6 +76,10 @@ class SubprocessCLITransport(Transport): def _build_command(self) -> list[str]: """Build CLI command with arguments.""" cmd = [self._cli_path, "--output-format", "stream-json", "--verbose"] + + # Add debug flag when using bypassPermissions to capture permission warnings + if self._options.permission_mode == "bypassPermissions": + cmd.append("--debug") if self._options.system_prompt: cmd.extend(["--system-prompt", self._options.system_prompt]) @@ -129,7 +134,7 @@ class SubprocessCLITransport(Transport): stdout=PIPE, stderr=PIPE, cwd=self._cwd, - env={**os.environ, "CLAUDE_CODE_ENTRYPOINT": "sdk-py"}, + env={**os.environ, "CLAUDE_CODE_ENTRYPOINT": "sdk-cli"}, ) if self._process.stdout: @@ -214,6 +219,7 @@ class SubprocessCLITransport(Transport): try: data = json.loads(json_buffer) json_buffer = "" + self._received_any_output = True try: yield data except GeneratorExit: @@ -225,14 +231,68 @@ class SubprocessCLITransport(Transport): pass await self._process.wait() - if self._process.returncode is not None and self._process.returncode != 0: - stderr_output = "\n".join(stderr_lines) - if stderr_output and "error" in stderr_output.lower(): + stderr_output = "\n".join(stderr_lines) + + # First check for specific error messages in stderr + if self._options.permission_mode == "bypassPermissions": + # Check for known bypassPermissions issues + if "cannot be used with root/sudo privileges" in stderr_output: raise ProcessError( - "CLI process failed", - exit_code=self._process.returncode, + "bypassPermissions mode cannot be used when running as root/sudo. " + "This is a security restriction in Claude Code. To fix this:\n" + "1. Run as a non-root user, or\n" + "2. Use permission_mode='acceptEdits' instead", + exit_code=self._process.returncode or 0, stderr=stderr_output, ) + + if "bypassPermissions mode is disabled" in stderr_output: + raise ProcessError( + "bypassPermissions mode is disabled by Claude Code settings. " + "The CLI will fall back to default permissions which requires user input, " + "but the SDK runs without stdin. To fix this, either:\n" + "1. Enable bypassPermissions in your Claude Code settings, or\n" + "2. Use permission_mode='acceptEdits' instead", + exit_code=self._process.returncode or 0, + stderr=stderr_output, + ) + + # Then check for the general "no output" case + if not self._received_any_output: + # This can happen for various reasons depending on permission mode + permission_mode = self._options.permission_mode or "default" + + if permission_mode == "bypassPermissions": + # Special handling for bypassPermissions silent failures + raise ProcessError( + "Claude Code CLI terminated without producing any output when using " + "bypassPermissions mode. This can happen when:\n" + "1. Running as root/sudo (security restriction)\n" + "2. bypassPermissions is disabled in settings\n" + "3. Other security restrictions are in place\n\n" + "Try using permission_mode='acceptEdits' instead, which is " + "the recommended mode for SDK usage.", + exit_code=self._process.returncode or 0, + stderr=stderr_output if stderr_output else None, + ) + elif permission_mode == "default": + raise ProcessError( + "Claude Code CLI appears to have terminated without output. " + "This often happens when the CLI is waiting for interactive " + "permission prompts that cannot be answered in SDK mode. " + "Try using permission_mode='acceptEdits' or 'bypassPermissions'.", + exit_code=self._process.returncode or 0, + stderr=stderr_output if stderr_output else None, + ) + + # Finally, handle any other non-zero exit codes + if self._process.returncode is not None and self._process.returncode != 0: + # Always raise an error for non-zero exit codes + raise ProcessError( + f"CLI process failed with exit code {self._process.returncode}", + exit_code=self._process.returncode, + stderr=stderr_output if stderr_output else None, + ) def is_connected(self) -> bool: """Check if subprocess is running.""" diff --git a/tests/test_transport.py b/tests/test_transport.py index 65702bc..0cdfe1c 100644 --- a/tests/test_transport.py +++ b/tests/test_transport.py @@ -132,3 +132,179 @@ class TestSubprocessCLITransport: # So we just verify the transport can be created and basic structure is correct assert transport._prompt == "test" assert transport._cli_path == "/usr/bin/claude" + + def test_bypass_permissions_disabled_error(self): + """Test that bypassPermissions being disabled raises proper error.""" + from claude_code_sdk._errors import ProcessError + from anyio.streams.text import TextReceiveStream + from anyio import ClosedResourceError + + async def _test(): + with patch("anyio.open_process") as mock_exec: + # Create a mock process that simulates the CLI behavior + mock_process = MagicMock() + mock_process.returncode = 0 # CLI exits successfully + mock_process.wait = AsyncMock() + + # Create mock stdout/stderr streams + mock_stdout_stream = AsyncMock() + mock_stderr_stream = AsyncMock() + + # Simulate empty stdout (no JSON messages) followed by closed stream + async def stdout_iter(self): + raise ClosedResourceError() + yield # This won't be reached + + # Simulate stderr with the warning message + async def stderr_iter(self): + yield "[ERROR] bypassPermissions mode is disabled by settings" + raise ClosedResourceError() + + type(mock_stdout_stream).__aiter__ = stdout_iter + type(mock_stderr_stream).__aiter__ = stderr_iter + + mock_process.stdout = MagicMock() + mock_process.stderr = MagicMock() + mock_exec.return_value = mock_process + + # Patch TextReceiveStream to return our mocks + with patch("claude_code_sdk._internal.transport.subprocess_cli.TextReceiveStream") as mock_text_stream: + mock_text_stream.side_effect = [mock_stdout_stream, mock_stderr_stream] + + transport = SubprocessCLITransport( + prompt="test", + options=ClaudeCodeOptions(permission_mode="bypassPermissions"), + cli_path="/usr/bin/claude", + ) + + await transport.connect() + + with pytest.raises(ProcessError) as exc_info: + # Consume all messages to trigger the error check + async for _ in transport.receive_messages(): + pass + + assert "bypassPermissions mode is disabled" in str(exc_info.value) + assert "requires user input" in str(exc_info.value) + assert "acceptEdits" in str(exc_info.value) + + anyio.run(_test) + + def test_bypass_permissions_adds_debug_flag(self): + """Test that bypassPermissions mode adds --debug flag to command.""" + transport = SubprocessCLITransport( + prompt="test", + options=ClaudeCodeOptions(permission_mode="bypassPermissions"), + cli_path="/usr/bin/claude", + ) + + cmd = transport._build_command() + assert "--debug" in cmd + + def test_bypass_permissions_root_user_error(self): + """Test that running as root with bypassPermissions raises proper error.""" + from claude_code_sdk._errors import ProcessError + from anyio import ClosedResourceError + + async def _test(): + with patch("anyio.open_process") as mock_exec: + # Create a mock process that exits with code 1 (root user rejection) + mock_process = MagicMock() + mock_process.returncode = 1 + mock_process.wait = AsyncMock() + + # Create mock stdout/stderr streams + mock_stdout_stream = AsyncMock() + mock_stderr_stream = AsyncMock() + + # Simulate empty stdout + async def stdout_iter(self): + raise ClosedResourceError() + yield # Won't be reached + + # Simulate stderr with root user error + async def stderr_iter(self): + yield "--dangerously-skip-permissions cannot be used with root/sudo privileges for security reasons" + raise ClosedResourceError() + + type(mock_stdout_stream).__aiter__ = stdout_iter + type(mock_stderr_stream).__aiter__ = stderr_iter + + mock_process.stdout = MagicMock() + mock_process.stderr = MagicMock() + mock_exec.return_value = mock_process + + # Patch TextReceiveStream to return our mocks + with patch("claude_code_sdk._internal.transport.subprocess_cli.TextReceiveStream") as mock_text_stream: + mock_text_stream.side_effect = [mock_stdout_stream, mock_stderr_stream] + + transport = SubprocessCLITransport( + prompt="test", + options=ClaudeCodeOptions(permission_mode="bypassPermissions"), + cli_path="/usr/bin/claude", + ) + + await transport.connect() + + with pytest.raises(ProcessError) as exc_info: + async for _ in transport.receive_messages(): + pass + + assert "cannot be used when running as root/sudo" in str(exc_info.value) + assert "Run as a non-root user" in str(exc_info.value) + + anyio.run(_test) + + def test_bypass_permissions_no_output(self): + """Test bypassPermissions mode that exits without any output.""" + from claude_code_sdk._errors import ProcessError + from anyio import ClosedResourceError + + async def _test(): + with patch("anyio.open_process") as mock_exec: + # Create a mock process that exits cleanly but produces no output + mock_process = MagicMock() + mock_process.returncode = 0 # Clean exit + mock_process.wait = AsyncMock() + + # Create mock stdout/stderr streams that immediately close + mock_stdout_stream = AsyncMock() + mock_stderr_stream = AsyncMock() + + # Both streams immediately close without yielding anything + async def empty_iter(self): + raise ClosedResourceError() + yield # Won't be reached + + type(mock_stdout_stream).__aiter__ = empty_iter + type(mock_stderr_stream).__aiter__ = empty_iter + + mock_process.stdout = MagicMock() + mock_process.stderr = MagicMock() + mock_exec.return_value = mock_process + + # Patch TextReceiveStream to return our mocks + with patch("claude_code_sdk._internal.transport.subprocess_cli.TextReceiveStream") as mock_text_stream: + mock_text_stream.side_effect = [mock_stdout_stream, mock_stderr_stream] + + transport = SubprocessCLITransport( + prompt="test", + options=ClaudeCodeOptions(permission_mode="bypassPermissions"), + cli_path="/usr/bin/claude", + ) + + await transport.connect() + + with pytest.raises(ProcessError) as exc_info: + # This simulates the "no response" issue - the iterator completes + # without yielding any messages + async for _ in transport.receive_messages(): + pass + + error_msg = str(exc_info.value) + assert "terminated without producing any output" in error_msg + assert "bypassPermissions mode" in error_msg + assert "Running as root/sudo" in error_msg + assert "acceptEdits" in error_msg + + anyio.run(_test)