fix(fsproxy): wait on Windows pipes without select() (#6459)

app/adapters/system/fsproxy.py

@@ -25,7 +25,7 @@ import subprocess
 import sys
 import threading
 from pathlib import Path
-from typing import Any, Callable, Dict, List, Optional
+from typing import Any, Callable, Dict, List, Optional, Union
 
 from app.runtime.settings import RuntimeSettingsCompat
 
@@ -364,18 +364,64 @@ class FileSystemProxy:
         :return: 响应行
         """
         timeout = self._timeout if timeout is None else timeout
+        # Windows 的 select() 只能等 socket。subprocess.PIPE 是匿名管道，
+        # selectors.DefaultSelector.select() 会抛 WinError 10038（WSAENOTSOCK），
+        # 整理链每次 stat 都会直接失败。POSIX 继续用 selector，Windows 改走线程等待。
+        if sys.platform == "win32":
+            return self._read_line_windows(timeout)
         if not self._selector.select(timeout=timeout):
-            logger.error(f"文件系统操作 {timeout} 秒无响应，判定挂载挂死，正在回收代理进程")
-            self._shutdown()
-            raise FileSystemTimeout(
-                errno_module.ETIMEDOUT,
-                f"文件系统操作超过 {timeout} 秒无响应，挂载可能已无响应"
-            )
+            self._raise_timeout(timeout)
         line = self._process.stdout.readline()
         if not line:
             raise BrokenPipeError("文件系统代理进程已退出")
         return line
 
+    def _read_line_windows(self, timeout: float) -> bytes:
+        """
+        在辅助线程上读一行，用 join(timeout) 实现可放弃等待。
+
+        不能用 select()/WaitForSingleObject：匿名管道在 Windows 上不是
+        可选择的套接字，也不是可靠的「有数据」同步对象。
+        :param timeout: 本次读取的超时秒数
+        :return: 响应行
+        """
+        if self._process is None or self._process.stdout is None:
+            raise BrokenPipeError("文件系统代理进程已退出")
+        stdout = self._process.stdout
+        holder: List[Union[bytes, BaseException]] = []
+
+        def _read() -> None:
+            try:
+                holder.append(stdout.readline())
+            except Exception as err:  # noqa: BLE001
+                holder.append(err)
+
+        reader = threading.Thread(target=_read, name="fsproxy-stdout", daemon=True)
+        reader.start()
+        reader.join(timeout)
+        if reader.is_alive():
+            self._raise_timeout(timeout)
+        if not holder:
+            raise BrokenPipeError("文件系统代理进程已退出")
+        line = holder[0]
+        if isinstance(line, BaseException):
+            raise line
+        if not line:
+            raise BrokenPipeError("文件系统代理进程已退出")
+        return line
+
+    def _raise_timeout(self, timeout: float) -> None:
+        """
+        判定挂载无响应：杀掉代理并抛出可被整理链处理的超时。
+        :param timeout: 已等待的秒数
+        """
+        logger.error(f"文件系统操作 {timeout} 秒无响应，判定挂载挂死，正在回收代理进程")
+        self._shutdown()
+        raise FileSystemTimeout(
+            errno_module.ETIMEDOUT,
+            f"文件系统操作超过 {timeout} 秒无响应，挂载可能已无响应"
+        )
+
     def _ensure_worker(self):
         """
         确保代理进程可用，不可用时重新启动。
@@ -383,15 +429,22 @@ class FileSystemProxy:
         if self._process is not None and self._process.poll() is None:
             return
         self._shutdown()
+        popen_kwargs: Dict[str, Any] = {
+            "stdin": subprocess.PIPE,
+            "stdout": subprocess.PIPE,
+            "stderr": subprocess.DEVNULL,
+            "bufsize": 0,
+        }
+        if sys.platform == "win32":
+            # NSSM 服务里再拉 python.exe 时避免弹出控制台窗口
+            popen_kwargs["creationflags"] = getattr(subprocess, "CREATE_NO_WINDOW", 0)
         self._process = subprocess.Popen(
             [sys.executable, str(_WORKER_PATH)],
-            stdin=subprocess.PIPE,
-            stdout=subprocess.PIPE,
-            stderr=subprocess.DEVNULL,
-            bufsize=0,
+            **popen_kwargs,
         )
-        self._selector = selectors.DefaultSelector()
-        self._selector.register(self._process.stdout, selectors.EVENT_READ)
+        if sys.platform != "win32":
+            self._selector = selectors.DefaultSelector()
+            self._selector.register(self._process.stdout, selectors.EVENT_READ)
         logger.debug(f"文件系统代理进程已启动: pid={self._process.pid}")
 
     def _shutdown(self):

tests/test_fs_proxy.py

@@ -141,6 +141,47 @@ def test_timeout_raises_and_kills_worker(tmp_path, monkeypatch):
         proxy.close()
 
 
+def test_windows_pipe_wait_stat_works(tmp_path, monkeypatch):
+    """
+    Windows 上 select() 不能等匿名管道（WinError 10038），必须走线程等待。
+    强制走这条路径时，stat 语义仍要与直接调用一致。
+    """
+    monkeypatch.setattr(sys, "platform", "win32")
+    proxy = FileSystemProxy(timeout=30)
+    try:
+        media = tmp_path / "win.mkv"
+        media.write_bytes(b"abcd")
+        result = proxy.stat(media)
+        assert result["size"] == 4
+        assert result["is_file"] is True
+        assert proxy._selector is None
+    finally:
+        proxy.close()
+
+
+def test_windows_pipe_wait_timeout_kills_worker(tmp_path, monkeypatch):
+    """
+    线程等待路径同样必须能在超时后放弃并回收冻住的 worker。
+    """
+    import app.adapters.system.fsproxy as fsproxy_module
+
+    monkeypatch.setattr(sys, "platform", "win32")
+    stuck_worker = tmp_path / "stuck_worker.py"
+    stuck_worker.write_text("import time\nwhile True:\n    time.sleep(60)\n", encoding="utf-8")
+    monkeypatch.setattr(fsproxy_module, "_WORKER_PATH", stuck_worker)
+
+    proxy = FileSystemProxy(timeout=0.5)
+    try:
+        started = time.monotonic()
+        with pytest.raises(FileSystemTimeout) as excinfo:
+            proxy.stat(tmp_path / "whatever.mkv")
+        assert time.monotonic() - started < 10
+        assert excinfo.value.errno == errno.ETIMEDOUT
+        assert proxy._process is None
+    finally:
+        proxy.close()
+
+
 def test_timeout_is_an_oserror():
     """
     超时必须是 OSError 子类：整理链与监控 watcher 对 OSError 已有完整的退避