From 060eb01d7d251560c60c6ba75401cc5a04dae541 Mon Sep 17 00:00:00 2001
From: Br1an67 <932039080@qq.com>
Date: Mon, 2 Mar 2026 01:04:07 +0800
Subject: [PATCH] fix: skip user parameter in open_process on Windows

The `user` parameter for running subprocesses as a different user is
not supported on Windows and causes `NotImplementedError` in asyncio's
subprocess transport.

Conditionally include the `user` parameter only on non-Windows
platforms so the SDK works on Windows.
---
 .../_internal/transport/subprocess_cli.py     | 21 ++++----
 tests/test_transport.py                       | 48 +++++++++++++++++++
 2 files changed, 60 insertions(+), 9 deletions(-)

diff --git a/src/claude_agent_sdk/_internal/transport/subprocess_cli.py b/src/claude_agent_sdk/_internal/transport/subprocess_cli.py
index 1f0aac58..25c967ed 100644
--- a/src/claude_agent_sdk/_internal/transport/subprocess_cli.py
+++ b/src/claude_agent_sdk/_internal/transport/subprocess_cli.py
@@ -366,15 +366,18 @@ async def connect(self) -> None:
             # For backward compat: use debug_stderr file object if no callback and debug is on
             stderr_dest = PIPE if should_pipe_stderr else None
 
-            self._process = await anyio.open_process(
-                cmd,
-                stdin=PIPE,
-                stdout=PIPE,
-                stderr=stderr_dest,
-                cwd=self._cwd,
-                env=process_env,
-                user=self._options.user,
-            )
+            # user parameter is not supported on Windows
+            process_kwargs: dict[str, object] = {
+                "stdin": PIPE,
+                "stdout": PIPE,
+                "stderr": stderr_dest,
+                "cwd": self._cwd,
+                "env": process_env,
+            }
+            if platform.system() != "Windows" and self._options.user is not None:
+                process_kwargs["user"] = self._options.user
+
+            self._process = await anyio.open_process(cmd, **process_kwargs)
 
             if self._process.stdout:
                 self._stdout_stream = TextReceiveStream(self._process.stdout)
diff --git a/tests/test_transport.py b/tests/test_transport.py
index 65e6ada7..dc7aa1e1 100644
--- a/tests/test_transport.py
+++ b/tests/test_transport.py
@@ -503,6 +503,54 @@ async def _test():
 
         anyio.run(_test)
 
+    def test_connect_user_param_skipped_on_windows(self):
+        """Test that user parameter is not passed to open_process on Windows."""
+
+        async def _test():
+            options = make_options(user="claude")
+
+            with (
+                patch(
+                    "anyio.open_process", new_callable=AsyncMock
+                ) as mock_open_process,
+                patch(
+                    "claude_agent_sdk._internal.transport.subprocess_cli.platform"
+                ) as mock_platform,
+            ):
+                mock_platform.system.return_value = "Windows"
+
+                # Mock version check process
+                mock_version_process = MagicMock()
+                mock_version_process.stdout = MagicMock()
+                mock_version_process.stdout.receive = AsyncMock(
+                    return_value=b"2.0.0 (Claude Code)"
+                )
+                mock_version_process.terminate = MagicMock()
+                mock_version_process.wait = AsyncMock()
+
+                # Mock main process
+                mock_process = MagicMock()
+                mock_process.stdout = MagicMock()
+                mock_stdin = MagicMock()
+                mock_stdin.aclose = AsyncMock()
+                mock_process.stdin = mock_stdin
+                mock_process.returncode = None
+
+                mock_open_process.side_effect = [mock_version_process, mock_process]
+
+                transport = SubprocessCLITransport(
+                    prompt="test",
+                    options=options,
+                )
+
+                await transport.connect()
+
+                # Check the second call (main process) does NOT include user
+                second_call_kwargs = mock_open_process.call_args_list[1].kwargs
+                assert "user" not in second_call_kwargs
+
+        anyio.run(_test)
+
     def test_build_command_with_sandbox_only(self):
         """Test building CLI command with sandbox settings (no existing settings)."""
         import json