fix(sdk-python): resolve Windows command shims on the effective PATH (#13862)

* fix(sdk-python): resolve Windows command shims on the effective PATH

* style(sdk-python): format Windows shim regression assertion

* fix(sdk-python): secure Windows command lookup and cleanup

* style(sdk-python): align test import grouping with CI

---------

Co-authored-by: 易良 <[email protected]>

docs/developers/sdk-python.md

@@ -28,6 +28,14 @@ pip install --pre qwen-code-sdk
 ```
 
 If `qwen` is not on `PATH`, pass `path_to_qwen_executable` explicitly.
+On Windows, the SDK discovers npm's `qwen.cmd` shim only through the effective
+child `PATH`, ignoring extensionless shell shims and non-launchable `PATHEXT`
+associations. A command in the current directory is considered only when that
+directory is explicitly included in `PATH`.
+You can override the lookup path through the `env` option; Windows environment
+variable names are case-insensitive, so `PATH`, `Path`, and `path` are equivalent.
+Closing a running Windows shim stops its CLI process tree through the system
+`taskkill` utility; process exit and stderr cleanup waits have bounded timeouts.
 
 Before writing SDK code, make sure the CLI works in the same shell:
 

packages/sdk-python/README.md

@@ -23,6 +23,17 @@ pip install --pre qwen-code-sdk
 You can also point the SDK at an explicit CLI binary or script with
 `path_to_qwen_executable`.
 
+On Windows, command names are resolved only from the effective child `PATH`,
+including case-insensitive overrides in `QueryOptions.env`. Supported command
+extensions are `.com`, `.exe`, `.bat`, and `.cmd`; extensionless npm shell shims
+and unrelated `PATHEXT` associations are ignored. The current directory is searched
+only when explicitly included in `PATH`. Explicit Python or JavaScript file paths
+continue to use their corresponding runtime.
+
+Closing a running Windows command shim uses the system `taskkill /T /F` command to
+stop both the shim and its CLI descendants before the parent process exits. Exit
+and stderr cleanup waits are bounded so an inherited pipe cannot stall shutdown.
+
 Before using the SDK, verify that the CLI works in the same environment:
 
 ```bash

packages/sdk-python/src/qwen_code_sdk/transport.py

@@ -3,7 +3,9 @@
 from __future__ import annotations
 
 import asyncio
+import errno
 import json
+import ntpath
 import os
 import subprocess
 import sys
@@ -19,6 +21,32 @@ from .types import QueryOptions
 _PROCESS_STREAM_LIMIT = 4 * 1024 * 1024
 
 
+def _resolve_windows_command(command: str, env: dict[str, str]) -> str:
+    if ntpath.dirname(command):
+        return command
+
+    supported = {".COM", ".EXE", ".BAT", ".CMD"}
+    suffix = ntpath.splitext(command)[1].upper()
+    extensions = (
+        [""]
+        if suffix in supported
+        else [
+            extension
+            for extension in env.get("PATHEXT", ".COM;.EXE;.BAT;.CMD").split(";")
+            if extension.upper() in supported
+        ]
+    )
+    for directory in env.get("PATH", "").split(";"):
+        directory = directory.strip('"')
+        if not directory:
+            continue
+        for extension in extensions:
+            candidate = os.path.abspath(os.path.join(directory, command + extension))
+            if os.path.isfile(candidate):
+                return candidate
+    raise FileNotFoundError(errno.ENOENT, os.strerror(errno.ENOENT), command)
+
+
 @dataclass(frozen=True)
 class SpawnInfo:
     command: str
@@ -70,6 +98,14 @@ class ProcessTransport:
             return
 
         spawn_info = prepare_spawn_info(self._options.path_to_qwen_executable)
+        env = {**os.environ, **(self._options.env or {})}
+        command = spawn_info.command
+        if sys.platform == "win32":
+            env = {key.upper(): value for key, value in os.environ.items()}
+            env.update(
+                {key.upper(): value for key, value in (self._options.env or {}).items()}
+            )
+            command = _resolve_windows_command(command, env)
         args = [*spawn_info.args, *build_cli_arguments(self._options)]
         stderr_target = (
             asyncio.subprocess.PIPE
@@ -78,10 +114,10 @@ class ProcessTransport:
         )
 
         self._process = await asyncio.create_subprocess_exec(
-            spawn_info.command,
+            command,
             *args,
             cwd=self._options.cwd,
-            env={**os.environ, **(self._options.env or {})},
+            env=env,
             stdin=asyncio.subprocess.PIPE,
             stdout=asyncio.subprocess.PIPE,
             stderr=stderr_target,
@@ -170,8 +206,12 @@ class ProcessTransport:
             )
 
         if self._stderr_task is not None:
-            await self._stderr_task
+            stderr_task = self._stderr_task
             self._stderr_task = None
+            try:
+                await asyncio.wait_for(stderr_task, timeout=5.0)
+            except asyncio.TimeoutError:
+                pass
 
     async def close(self) -> None:
         if self._closed:
@@ -186,15 +226,43 @@ class ProcessTransport:
             self._process.stdin.close()
             self._input_closed = True
 
-        if self._process.returncode is None:
-            self._process.terminate()
-            try:
-                await asyncio.wait_for(self._process.wait(), timeout=5.0)
-            except asyncio.TimeoutError:
-                self._process.kill()
-                await self._process.wait()
+        try:
+            if self._process.returncode is None:
+                tree_killed = False
+                if sys.platform == "win32":
+                    # Kill the shim and its CLI before the parent PID disappears.
+                    taskkill = Path(os.environ.get("SystemRoot", r"C:\Windows"))
+                    taskkill = taskkill / "System32" / "taskkill.exe"
+                    try:
+                        result = await asyncio.to_thread(
+                            subprocess.run,
+                            [str(taskkill), "/PID", str(self._process.pid), "/T", "/F"],
+                            stdout=subprocess.DEVNULL,
+                            stderr=subprocess.DEVNULL,
+                            timeout=5.0,
+                            creationflags=subprocess.CREATE_NO_WINDOW,
+                        )
+                        tree_killed = result.returncode == 0
+                    except (OSError, subprocess.TimeoutExpired):
+                        pass
+                if not tree_killed and self._process.returncode is None:
+                    self._process.terminate()
+                try:
+                    await asyncio.wait_for(self._process.wait(), timeout=5.0)
+                except asyncio.TimeoutError:
+                    if self._process.returncode is None:
+                        self._process.kill()
+                    await asyncio.wait_for(self._process.wait(), timeout=5.0)
 
-        await self._finalize_exit()
+            await self._finalize_exit()
+        finally:
+            if self._stderr_task is not None:
+                self._stderr_task.cancel()
+                try:
+                    await self._stderr_task
+                except asyncio.CancelledError:
+                    pass
+                self._stderr_task = None
 
 
 def build_cli_arguments(options: QueryOptions) -> list[str]:

packages/sdk-python/tests/unit/test_transport.py

@@ -1,13 +1,19 @@
 from __future__ import annotations
 
 import asyncio
+import os
 import subprocess
 import sys
 from pathlib import Path
 from typing import Any
 
 import pytest
-from qwen_code_sdk.transport import build_cli_arguments, prepare_spawn_info
+from qwen_code_sdk.transport import (
+    ProcessTransport,
+    _resolve_windows_command,
+    build_cli_arguments,
+    prepare_spawn_info,
+)
 from qwen_code_sdk.types import QueryOptions, TimeoutOptions
 
 VALID_UUID = "123e4567-e89b-12d3-a456-426614174000"
@@ -21,6 +27,256 @@ class DummyProcess:
         self.returncode = 0
 
 
+@pytest.mark.parametrize("command", ["qwen", "qwen.CMD"])
+def test_windows_lookup_ignores_implicit_current_directory(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, command: str
+) -> None:
+    trusted = tmp_path / "trusted"
+    trusted.mkdir()
+    shim = trusted / "qwen.CMD"
+    shim.write_text("@echo trusted\n", encoding="utf-8")
+    (tmp_path / "qwen.CMD").write_text("@echo decoy\n", encoding="utf-8")
+    monkeypatch.chdir(tmp_path)
+
+    assert _resolve_windows_command(command, {"PATH": str(trusted)}) == str(shim)
+    with pytest.raises(FileNotFoundError):
+        _resolve_windows_command(command, {"PATH": str(tmp_path / "missing")})
+
+
+def test_windows_lookup_skips_shell_twins_and_unsupported_pathext(
+    tmp_path: Path,
+) -> None:
+    first = tmp_path / "first"
+    second = tmp_path / "second"
+    first.mkdir()
+    second.mkdir()
+    (first / "qwen").write_text("#!/bin/sh\n", encoding="utf-8")
+    (first / "qwen.JS").write_text("console.log('decoy')\n", encoding="utf-8")
+    shim = second / "qwen.CMD"
+    shim.write_text("@echo trusted\n", encoding="utf-8")
+
+    assert _resolve_windows_command(
+        "qwen", {"PATH": f"{first};{second}", "PATHEXT": ".JS;.CMD;.EXE"}
+    ) == str(shim)
+
+
+def test_windows_lookup_supports_dotted_command_names(tmp_path: Path) -> None:
+    shim = tmp_path / "qwen.custom.CMD"
+    shim.write_text("@echo trusted\n", encoding="utf-8")
+
+    assert _resolve_windows_command("qwen.custom", {"PATH": str(tmp_path)}) == str(shim)
+
+
+@pytest.mark.parametrize("path", [None, "", ";", '""'])
+def test_windows_lookup_does_not_fall_back_to_current_directory(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, path: str | None
+) -> None:
+    monkeypatch.chdir(tmp_path)
+    (tmp_path / "qwen.CMD").write_text("@echo decoy\n", encoding="utf-8")
+
+    with pytest.raises(FileNotFoundError):
+        _resolve_windows_command("qwen", {} if path is None else {"PATH": path})
+
+
+@pytest.mark.asyncio
+async def test_windows_child_environment_is_case_insensitive(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    import qwen_code_sdk.transport as transport_module
+
+    shim = tmp_path / "qwen.CMD"
+    shim.write_text("@echo trusted\n", encoding="utf-8")
+    captured: dict[str, Any] = {}
+
+    async def fake_spawn(*args: Any, **kwargs: Any) -> DummyProcess:
+        captured.update(command=args[0], env=kwargs["env"])
+        return DummyProcess()
+
+    monkeypatch.setattr(transport_module.sys, "platform", "win32")
+    monkeypatch.setattr(asyncio, "create_subprocess_exec", fake_spawn)
+    monkeypatch.setenv("PATH", "inherited")
+    monkeypatch.setenv("QWEN_TEST_VALUE", "parent")
+    options = QueryOptions(
+        env={"Path": "wrong", "path": str(tmp_path), "qwen_test_value": "child"}
+    )
+    await ProcessTransport(options).start()
+
+    assert captured["command"] == str(shim)
+    assert captured["env"]["PATH"] == str(tmp_path)
+    assert captured["env"]["QWEN_TEST_VALUE"] == "child"
+    assert all(key == key.upper() for key in captured["env"])
+
+
+@pytest.mark.asyncio
+async def test_child_environment_inherits_parent_and_applies_overrides(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    script = tmp_path / "environment_peer.py"
+    script.write_text(
+        "import json, os\n"
+        "keys = ['QWEN_TEST_PARENT', 'QWEN_TEST_CHILD', 'QWEN_TEST_OVERRIDE', 'PATH']\n"
+        "print(json.dumps({key: os.environ.get(key) for key in keys}))\n",
+        encoding="utf-8",
+    )
+    monkeypatch.setenv("QWEN_TEST_PARENT", "inherited")
+    monkeypatch.setenv("QWEN_TEST_OVERRIDE", "parent")
+    transport = ProcessTransport(
+        QueryOptions(
+            path_to_qwen_executable=str(script),
+            env={
+                "QWEN_TEST_CHILD": "child",
+                "QWEN_TEST_OVERRIDE": "override",
+                "PATH": "controlled-child-path",
+            },
+        )
+    )
+    try:
+        await transport.start()
+        messages = await asyncio.wait_for(_collect_transport_messages(transport), 10)
+    finally:
+        await transport.close()
+
+    assert messages == [
+        {
+            "QWEN_TEST_PARENT": "inherited",
+            "QWEN_TEST_CHILD": "child",
+            "QWEN_TEST_OVERRIDE": "override",
+            "PATH": "controlled-child-path",
+        }
+    ]
+    assert os.environ["QWEN_TEST_OVERRIDE"] == "parent"
+
+
+@pytest.mark.skipif(sys.platform != "win32", reason="Windows process tree cleanup")
+@pytest.mark.parametrize("debug", [False, True])
+@pytest.mark.asyncio
+async def test_windows_close_stops_shim_cli_and_grandchild(
+    tmp_path: Path,
+    debug: bool,
+) -> None:
+    import ctypes
+
+    script = tmp_path / "cli.py"
+    script.write_text(
+        "import json, os, subprocess, sys, time\n"
+        "child = subprocess.Popen([sys.executable, '-c', "
+        "'import time; time.sleep(60)'])\n"
+        "print(json.dumps({'pids': [os.getpid(), child.pid]}), flush=True)\n"
+        "time.sleep(60)\n",
+        encoding="utf-8",
+    )
+    shim = tmp_path / "qwen.cmd"
+    shim.write_text(f'@echo off\n"{sys.executable}" "{script}" %*\n', encoding="utf-8")
+    transport = ProcessTransport(
+        QueryOptions(path_to_qwen_executable=str(shim), debug=debug)
+    )
+    kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)
+    kernel32.OpenProcess.restype = ctypes.c_void_p
+    kernel32.WaitForSingleObject.argtypes = [ctypes.c_void_p, ctypes.c_ulong]
+    kernel32.CloseHandle.argtypes = [ctypes.c_void_p]
+    handles = []
+    pids = []
+    messages = transport.read_messages()
+    try:
+        await transport.start()
+        message = await asyncio.wait_for(anext(messages), timeout=10)
+        pids = message["pids"]
+        handles = [kernel32.OpenProcess(0x00100000, False, pid) for pid in pids]
+        assert all(handles)
+        await asyncio.wait_for(transport.close(), timeout=15)
+        assert all(kernel32.WaitForSingleObject(handle, 0) == 0 for handle in handles)
+    finally:
+        await messages.aclose()
+        await transport.close()
+        for pid, handle in zip(pids, handles, strict=True):
+            if handle and kernel32.WaitForSingleObject(handle, 0) == 258:
+                await asyncio.to_thread(
+                    subprocess.run,
+                    [
+                        str(
+                            Path(os.environ.get("SystemRoot", r"C:\Windows"))
+                            / "System32"
+                            / "taskkill.exe"
+                        ),
+                        "/PID",
+                        str(pid),
+                        "/T",
+                        "/F",
+                    ],
+                    stdout=subprocess.DEVNULL,
+                    stderr=subprocess.DEVNULL,
+                    check=False,
+                    timeout=5,
+                )
+            if handle:
+                kernel32.CloseHandle(handle)
+
+
+@pytest.mark.asyncio
+async def test_stderr_join_is_bounded_and_cancels_reader(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    wait_for = asyncio.wait_for
+
+    async def short_wait(awaitable: Any, timeout: float | None) -> Any:
+        assert timeout == 5.0
+        return await wait_for(awaitable, timeout=0.01)
+
+    monkeypatch.setattr(asyncio, "wait_for", short_wait)
+    transport = ProcessTransport(QueryOptions())
+    transport._process = DummyProcess()
+    reader = asyncio.create_task(asyncio.sleep(60))
+    transport._stderr_task = reader
+
+    await transport.close()
+
+    assert reader.cancelled()
+    assert transport._stderr_task is None
+
+
+@pytest.mark.asyncio
+async def test_close_bounds_wait_after_kill(monkeypatch: pytest.MonkeyPatch) -> None:
+    wait_for = asyncio.wait_for
+
+    async def short_wait(awaitable: Any, timeout: float | None) -> Any:
+        assert timeout == 5.0
+        return await wait_for(awaitable, timeout=0.01)
+
+    class HungProcess(DummyProcess):
+        returncode: Any
+
+        def __init__(self) -> None:
+            super().__init__()
+            self.returncode = None
+            self.killed = False
+
+        def terminate(self) -> None:
+            pass
+
+        def kill(self) -> None:
+            self.killed = True
+
+        async def wait(self) -> int:
+            await asyncio.sleep(60)
+            return 0
+
+    monkeypatch.setattr(sys, "platform", "linux")
+    monkeypatch.setattr(asyncio, "wait_for", short_wait)
+    transport = ProcessTransport(QueryOptions())
+    process = HungProcess()
+    transport._process = process
+    reader = asyncio.create_task(asyncio.sleep(60))
+    transport._stderr_task = reader
+
+    with pytest.raises(asyncio.TimeoutError):
+        await transport.close()
+
+    assert process.killed
+    assert reader.cancelled()
+
+
 def test_build_cli_arguments_maps_supported_options() -> None:
     args = build_cli_arguments(
         QueryOptions(
@@ -155,6 +411,78 @@ def test_prepare_spawn_info_uses_node_for_javascript_files(tmp_path: Path) -> No
     assert spawn_info.args == [str(script_path.resolve())]
 
 
+@pytest.mark.skipif(sys.platform != "win32", reason="Windows command shim discovery")
+@pytest.mark.parametrize("executable", [None, "qwen", "qwen.cmd", "absolute"])
+@pytest.mark.parametrize(
+    "path_source", ["inherited", "PATH", "Path", "path", "PATH+Path", "Path+PATH"]
+)
+@pytest.mark.asyncio
+async def test_transport_starts_windows_command_shim_on_path(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    executable: str | None,
+    path_source: str,
+) -> None:
+    bin_dir = tmp_path / "npm prefix with spaces"
+    bin_dir.mkdir()
+    script_path = bin_dir / "fake_cli.py"
+    script_path.write_text(
+        "import json, os, sys\n"
+        "print(json.dumps({'args': sys.argv[1:], 'path': os.environ['PATH']}))\n",
+        encoding="utf-8",
+    )
+    shim_path = bin_dir / "qwen.cmd"
+    shim_path.write_text(
+        f'@echo off\n"{sys.executable}" "{script_path}" %*\n',
+        encoding="utf-8",
+    )
+    cli_path = str(bin_dir)
+    if executable == "absolute":
+        executable = str(shim_path)
+    env = None
+    if path_source == "inherited":
+        monkeypatch.setenv("PATH", cli_path)
+    else:
+        monkeypatch.setenv("PATH", str(tmp_path / "empty"))
+        if "+" in path_source:
+            first_key, last_key = path_source.split("+")
+            env = {first_key: str(tmp_path / "empty"), last_key: cli_path}
+        else:
+            env = {path_source: cli_path}
+
+    transport = ProcessTransport(
+        QueryOptions(path_to_qwen_executable=executable, env=env, model="model name")
+    )
+    try:
+        await transport.start()
+        messages = await asyncio.wait_for(
+            _collect_transport_messages(transport), timeout=10
+        )
+    finally:
+        await transport.close()
+
+    assert messages == [
+        {
+            "args": build_cli_arguments(QueryOptions(model="model name")),
+            "path": cli_path,
+        }
+    ]
+
+
+async def _collect_transport_messages(transport: ProcessTransport) -> list[Any]:
+    return [message async for message in transport.read_messages()]
+
+
+@pytest.mark.asyncio
+async def test_transport_preserves_missing_executable_error(tmp_path: Path) -> None:
+    transport = ProcessTransport(
+        QueryOptions(path_to_qwen_executable=str(tmp_path / "missing-qwen"))
+    )
+
+    with pytest.raises(FileNotFoundError):
+        await transport.start()
+
+
 @pytest.mark.asyncio
 async def test_transport_reads_stream_json_lines_larger_than_64_kib(
     tmp_path: Path,
@@ -234,7 +562,7 @@ async def test_transport_discards_stderr_when_debug_is_disabled(
         fromlist=["ProcessTransport"],
     )
     transport = transport_module.ProcessTransport(
-        QueryOptions(timeout=TimeoutOptions())
+        QueryOptions(path_to_qwen_executable=sys.executable, timeout=TimeoutOptions())
     )
 
     await transport.start()