feat(agent): improve Windows shell compatibility

app/agent/llm/capability.py

@@ -12,11 +12,11 @@ from pathlib import Path
 from typing import Any, Dict, Optional
 from uuid import uuid4
 
-from app.application.notification import get_notification_configs
-from app.runtime.settings import get_runtime_setting
-
-from app.runtime.log import logger
 from app.adapters.network.http import RequestUtils
+from app.agent.shell import agent_text_subprocess_kwargs
+from app.application.notification import get_notification_configs
+from app.runtime.log import logger
+from app.runtime.settings import get_runtime_setting
 
 
 class AgentCapabilityProvider(ABC):
@@ -286,7 +286,13 @@ class OpenAIChatAudioProvider(AudioCapabilityProvider):
                 "wav",
                 str(output_path),
             ]
-            result = subprocess.run(cmd, capture_output=True, text=True, check=False)
+            result = subprocess.run(
+                cmd,
+                capture_output=True,
+                text=True,
+                check=False,
+                **agent_text_subprocess_kwargs(),
+            )
             if result.returncode != 0 or not output_path.exists():
                 logger.warning(
                     "%s STT 音频转 WAV 失败: returncode=%s, stderr=%s",
@@ -354,7 +360,13 @@ class OpenAIChatAudioProvider(AudioCapabilityProvider):
             "libopus",
             str(output_path),
         ]
-        result = subprocess.run(cmd, capture_output=True, text=True, check=False)
+        result = subprocess.run(
+            cmd,
+            capture_output=True,
+            text=True,
+            check=False,
+            **agent_text_subprocess_kwargs(),
+        )
         if result.returncode != 0 or not output_path.exists():
             logger.warning(
                 "%s TTS 音频转 Opus 失败，将使用 WAV 原文件: returncode=%s, stderr=%s",

app/agent/mcp.py

@@ -5,13 +5,14 @@ from __future__ import annotations
 import asyncio
 import hashlib
 import json
-import os
 import re
 import uuid
 from dataclasses import dataclass
 from typing import Any, Optional
 from urllib.parse import urljoin
 
+from app.adapters.network.http import AsyncRequestUtils
+from app.agent.shell import build_agent_subprocess_env
 from app.application.configuration import get_configured_system_config
 from app.runtime.log import logger
 from app.schemas.agent import (
@@ -20,7 +21,6 @@ from app.schemas.agent import (
     AgentMcpServerToolInfo,
 )
 from app.schemas.types import SystemConfigKey
-from app.adapters.network.http import AsyncRequestUtils
 
 MCP_PROTOCOL_VERSION = "2025-11-25"
 MCP_CLIENT_NAME = "MoviePilot Agent"
@@ -202,8 +202,7 @@ class _StdioMcpSession:
         """启动 stdio MCP 子进程。"""
         if not self.server.command:
             raise RuntimeError("stdio MCP 服务器缺少启动命令")
-        env = os.environ.copy()
-        env.update(self.server.env or {})
+        env = build_agent_subprocess_env(self.server.env)
         self.process = await asyncio.create_subprocess_exec(
             self.server.command,
             *(self.server.args or []),
@@ -241,7 +240,10 @@ class _StdioMcpSession:
                 line = await self.process.stderr.readline()
                 if not line:
                     break
-                logger.debug(f"MCP stdio[{self.server.name}] stderr: {line.decode(errors='replace').strip()}")
+                logger.debug(
+                    f"MCP stdio[{self.server.name}] stderr: "
+                    f"{line.decode('utf-8', errors='replace').strip()}"
+                )
         except asyncio.CancelledError:
             return
 

app/agent/prompt/System Core Prompt.txt

@@ -71,6 +71,7 @@ You act as a proactive agent. Your goal is to fully resolve the user's media-rel
 - Read the relevant file before changing it, then pick the editing tool by scope. Use `apply_patch` when one logical change spans multiple files, adds new files, or deletes files: submit a single patch wrapped in `*** Begin Patch` / `*** End Patch` with `*** Add File:`, `*** Update File:`, and `*** Delete File:` sections; every context and removed line must match the current content exactly, and the whole patch is validated before any file is written. Use `edit_file` for a single localized exact replacement within one already-read file; make `old_text` unique with enough surrounding context, and use `replace_all=true` only when every match must change. Use `write_file` for one standalone new file; set `overwrite=true` only for an intentional full rewrite, and use `read_file(include_metadata=true)` plus `expected_sha256` when preserving the previously read version matters.
 - When implementation depends on a Python or Node.js API, first identify the installed or locked dependency version from environment metadata, requirements, package manifests, lockfiles, local source, and type declarations. Use `rg` against the relevant package directory, `.venv`, or `node_modules` instead of scanning the entire project without bounds. If local evidence is insufficient, use `search_web` and then `browse_webpage` to read the matching version of the official documentation. Do not guess signatures from memory, mix examples from incompatible versions, or install a package only to inspect its API.
 - Use structured file tools for source edits because they enforce file access boundaries and conflict checks. Never use shell redirection, inline scripts, or another tool to bypass a file-tool permission denial.
+- On Windows, follow the command environment injected in `<system_info>` for the entire task. If Git Bash is selected, use Git Bash/POSIX syntax and prefer `git` for repository work. Otherwise, when PowerShell 7 is selected, use `pwsh` syntax consistently. Never mix cmd.exe, Windows PowerShell, PowerShell 7, and POSIX syntax in one workflow, and treat command and file text as UTF-8.
 - Use `execute_command` for administrator-only multi-file diagnostics, tests, Git, service operations, SSH, or an exact command the user requested. Use `action=run` for short bounded commands. Use `action=start` for long-running or interactive commands, including SSH; then continue with `read`, `wait`, `write`, or `kill` using the returned `session_id`. Do not start a background session for a short command that can finish within `action=run`.
 </tool_strategy>
 

app/agent/prompt/__init__.py

@@ -11,6 +11,7 @@ import yaml
 
 from app.adapters.system.host import SystemUtils
 from app.agent.llm.capability import AgentCapabilityManager
+from app.agent.shell import resolve_agent_shell
 from app.runtime.log import logger
 from app.runtime.settings import get_runtime_setting
 from app.schemas.notification import (
@@ -265,6 +266,9 @@ class PromptManager:
             f"- 运行环境: {SystemUtils.platform} {'docker' if SystemUtils.is_docker() else ''}",
             "- 详细运行状态和数据库通过 `query_doctor_report` 或 `execute_command` 查询；配置值先加载对应 Skill，再通过 `moviepilot_api` 的配置 operation 查询。",
         ]
+        shell = resolve_agent_shell()
+        if shell:
+            info_lines.append(shell.prompt_guidance())
         path_lines = self._get_runtime_path_lines()
         if path_lines:
             info_lines.extend(

app/agent/shell.py

@@ -0,0 +1,179 @@
+"""Agent 命令解释器选择与子进程编码策略。"""
+
+from __future__ import annotations
+
+import os
+import shutil
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, Callable, Mapping, Optional
+
+WINDOWS_UTF8_ENVIRONMENT = {
+    "LANG": "C.UTF-8",
+    "LC_ALL": "C.UTF-8",
+    "PYTHONIOENCODING": "utf-8",
+    "PYTHONUTF8": "1",
+}
+
+_POWERSHELL_UTF8_PREFIX = (
+    "$utf8 = [System.Text.UTF8Encoding]::new($false); "
+    "[Console]::InputEncoding = $utf8; "
+    "[Console]::OutputEncoding = $utf8; "
+    "$OutputEncoding = $utf8; "
+    "$PSDefaultParameterValues['*:Encoding'] = 'utf8'; "
+    "chcp.com 65001 | Out-Null;"
+)
+
+CommandFinder = Callable[[str], Optional[str]]
+
+
+@dataclass(frozen=True)
+class AgentShell:
+    """描述 Windows Agent 实际使用的命令解释器及模型指导。"""
+
+    kind: str
+    executable: str
+    arguments: tuple[str, ...]
+    command_prefix: str = ""
+    git_available: bool = False
+
+    def build_argv(self, command: str) -> list[str]:
+        """生成无需再次经过系统默认 shell 的命令参数。"""
+        command_text = f"{self.command_prefix}\n{command}" if self.command_prefix else command
+        return [self.executable, *self.arguments, command_text]
+
+    def prompt_guidance(self) -> str:
+        """返回不包含可执行文件绝对路径的 Windows 命令指导。"""
+        if self.kind == "git-bash":
+            return (
+                "- Windows Agent 命令环境: Git Bash（UTF-8）。所有 `execute_command` "
+                "命令统一使用 Git Bash/POSIX 语法；涉及仓库、源码状态或版本控制时优先使用 "
+                "`git`，不要混用 cmd.exe 或 PowerShell 语法。"
+            )
+        if self.kind == "pwsh":
+            git_guidance = "；仓库和版本控制操作仍优先使用 `git`" if self.git_available else ""
+            return (
+                "- Windows Agent 命令环境: PowerShell 7 (`pwsh`, UTF-8)。所有 "
+                f"`execute_command` 命令统一使用 PowerShell 语法{git_guidance}，不要混用 "
+                "cmd.exe、Windows PowerShell 或 POSIX shell 语法。"
+            )
+        git_guidance = "；仓库和版本控制操作优先使用 `git`" if self.git_available else ""
+        return (
+            "- Windows Agent 命令环境: cmd.exe UTF-8 兼容回退"
+            f"{git_guidance}。仅在 Git Bash 和 PowerShell 7 不可用时使用 cmd 语法。"
+        )
+
+
+def _find_git_bash(git_path: str, command_finder: CommandFinder) -> Optional[str]:
+    """从 Git for Windows 常见布局中定位同一安装目录下的 bash.exe。"""
+    git_executable = Path(git_path)
+    git_root = git_executable.parent.parent
+    candidates = (
+        git_executable.parent / "bash.exe",
+        git_root / "bin" / "bash.exe",
+        git_root / "usr" / "bin" / "bash.exe",
+    )
+    for candidate in candidates:
+        if candidate.is_file():
+            return str(candidate)
+
+    bash_path = command_finder("bash") or command_finder("bash.exe")
+    if not bash_path:
+        return None
+    bash_executable = Path(bash_path)
+    if bash_executable.parent == git_executable.parent:
+        return str(bash_executable)
+    if bash_executable.parent.parent == git_root:
+        return str(bash_executable)
+    return None
+
+
+def resolve_agent_shell(
+    *,
+    platform_name: Optional[str] = None,
+    command_finder: Optional[CommandFinder] = None,
+    environment: Optional[Mapping[str, str]] = None,
+) -> Optional[AgentShell]:
+    """
+    解析 Windows Agent 命令解释器，非 Windows 返回 None 以保留原有 shell 行为。
+
+    :param platform_name: 用于测试覆盖的 os.name，默认读取当前平台
+    :param command_finder: 命令定位函数，默认使用 shutil.which
+    :param environment: 用于读取 COMSPEC 的环境变量
+    :return: Windows 命令解释器策略，非 Windows 返回 None
+    """
+    if (platform_name or os.name) != "nt":
+        return None
+
+    finder = command_finder or shutil.which
+    current_environment = environment if environment is not None else os.environ
+    git_path = finder("git") or finder("git.exe")
+    if git_path:
+        bash_path = _find_git_bash(git_path, finder)
+        if bash_path:
+            return AgentShell(
+                kind="git-bash",
+                executable=bash_path,
+                arguments=("--noprofile", "--norc", "-lc"),
+                git_available=True,
+            )
+
+    pwsh_path = finder("pwsh") or finder("pwsh.exe")
+    if pwsh_path:
+        return AgentShell(
+            kind="pwsh",
+            executable=pwsh_path,
+            arguments=("-NoLogo", "-NoProfile", "-Command"),
+            command_prefix=_POWERSHELL_UTF8_PREFIX,
+            git_available=bool(git_path),
+        )
+
+    cmd_path = (
+        current_environment.get("COMSPEC")
+        or finder("cmd")
+        or finder("cmd.exe")
+        or "cmd.exe"
+    )
+    return AgentShell(
+        kind="cmd",
+        executable=cmd_path,
+        arguments=("/d", "/s", "/c"),
+        command_prefix="chcp 65001>nul &",
+        git_available=bool(git_path),
+    )
+
+
+def build_agent_subprocess_env(
+    overrides: Optional[Mapping[str, Any]] = None,
+    *,
+    platform_name: Optional[str] = None,
+) -> dict[str, str]:
+    """
+    构造 Agent 子进程环境，并在 Windows 上强制覆盖为 UTF-8 编解码。
+
+    :param overrides: 调用方附加的环境变量
+    :param platform_name: 用于测试覆盖的 os.name，默认读取当前平台
+    :return: 可直接传给 subprocess 的字符串环境变量
+    """
+    environment = os.environ.copy()
+    for key, value in (overrides or {}).items():
+        if value is not None:
+            environment[str(key)] = str(value)
+    if (platform_name or os.name) == "nt":
+        # Python 会把该变量的任意非空值视为启用，必须移除而不是写入 "0"。
+        environment.pop("PYTHONLEGACYWINDOWSSTDIO", None)
+        environment.update(WINDOWS_UTF8_ENVIRONMENT)
+    return environment
+
+
+def agent_text_subprocess_kwargs(
+    *, platform_name: Optional[str] = None
+) -> dict[str, Any]:
+    """为 Windows Agent 文本子进程补充 UTF-8 解码参数，POSIX 返回空参数。"""
+    if (platform_name or os.name) != "nt":
+        return {}
+    return {
+        "encoding": "utf-8",
+        "errors": "replace",
+        "env": build_agent_subprocess_env(platform_name="nt"),
+    }

app/agent/tools/impl/_terminal_session.py

@@ -13,6 +13,7 @@ from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Any, Optional
 
+from app.agent.shell import build_agent_subprocess_env, resolve_agent_shell
 from app.agent.tools.impl._command_safety import validate_command_safety
 from app.runtime.log import logger
 from app.runtime.settings import get_runtime_setting
@@ -163,15 +164,8 @@ class _TerminalSessionManager:
 
     @staticmethod
     def _build_env(env: Optional[dict[str, Any]]) -> dict[str, str]:
-        """合并环境变量，并把值稳定转换为字符串。"""
-        merged_env = os.environ.copy()
-        if not env:
-            return merged_env
-        for key, value in env.items():
-            if value is None:
-                continue
-            merged_env[str(key)] = str(value)
-        return merged_env
+        """合并环境变量，Windows 强制覆盖为 UTF-8 编解码。"""
+        return build_agent_subprocess_env(env)
 
     @staticmethod
     def _validate_command(command: str, *, confirmed: bool = False) -> None:
@@ -307,12 +301,21 @@ class _TerminalSessionManager:
         self, command: str, cwd: str, env: dict[str, str]
     ) -> _TerminalSession:
         """通过普通 stdin/stdout/stderr 管道启动命令会话。"""
-        process = await asyncio.create_subprocess_shell(
-            command,
-            cwd=cwd,
-            env=env,
-            **self._pipe_subprocess_kwargs(),
-        )
+        shell = resolve_agent_shell()
+        if shell:
+            process = await asyncio.create_subprocess_exec(
+                *shell.build_argv(command),
+                cwd=cwd,
+                env=env,
+                **self._pipe_subprocess_kwargs(),
+            )
+        else:
+            process = await asyncio.create_subprocess_shell(
+                command,
+                cwd=cwd,
+                env=env,
+                **self._pipe_subprocess_kwargs(),
+            )
         session = _TerminalSession(
             session_id=f"term_{uuid.uuid4().hex[:12]}",
             command=command,

app/agent/tools/impl/execute_command.py

@@ -14,18 +14,18 @@ from typing import Any, Literal, Optional, TextIO, Type
 
 from pydantic import BaseModel, Field
 
-from app.agent.tools.impl._command_safety import validate_command_safety
+from app.agent.shell import build_agent_subprocess_env, resolve_agent_shell
 from app.agent.tools.base import MoviePilotTool
-from app.agent.tools.tags import ToolTag
+from app.agent.tools.impl._command_safety import validate_command_safety
 from app.agent.tools.impl._terminal_session import (
     TERMINAL_DEFAULT_READ_BYTES,
     TERMINAL_MAX_READ_BYTES,
     TERMINAL_WAIT_DEFAULT_MS,
     get_terminal_session_manager,
 )
+from app.agent.tools.tags import ToolTag
 from app.runtime.log import logger
 
-
 DEFAULT_TIMEOUT_SECONDS = 60
 MAX_TIMEOUT_SECONDS = 300
 MAX_OUTPUT_PREVIEW_BYTES = 32 * 1024
@@ -466,11 +466,20 @@ class ExecuteCommandTool(MoviePilotTool):
         normalized_timeout, timeout_note = self._normalize_timeout(timeout)
 
         async with _command_semaphore:
-            process = await asyncio.create_subprocess_shell(
-                command,
-                cwd=cwd,
-                **self._subprocess_kwargs(),
-            )
+            shell = resolve_agent_shell()
+            if shell:
+                process = await asyncio.create_subprocess_exec(
+                    *shell.build_argv(command),
+                    cwd=cwd,
+                    env=build_agent_subprocess_env(),
+                    **self._subprocess_kwargs(),
+                )
+            else:
+                process = await asyncio.create_subprocess_shell(
+                    command,
+                    cwd=cwd,
+                    **self._subprocess_kwargs(),
+                )
             output = _CommandOutput(preview_limit_bytes=MAX_OUTPUT_PREVIEW_BYTES)
             wait_task = asyncio.create_task(process.wait())
             reader_tasks = [

app/agent/tools/impl/service.py

@@ -12,6 +12,7 @@ from typing import Any, ClassVar, Dict, Optional, Type
 
 from pydantic import BaseModel, ConfigDict, Field
 
+from app.agent.shell import agent_text_subprocess_kwargs
 from app.agent.tools.base import MoviePilotTool
 from app.agent.tools.tags import ToolTag
 from app.runtime.settings import get_runtime_setting
@@ -325,6 +326,7 @@ def _run_service_script(
         capture_output=True,
         text=True,
         check=False,
+        **agent_text_subprocess_kwargs(),
     )
     return _parse_script_payload(completed.stdout)
 
@@ -505,6 +507,7 @@ def _run_database_script(arguments: Dict[str, Any], *, root_path: Path) -> dict[
         capture_output=True,
         text=True,
         check=False,
+        **agent_text_subprocess_kwargs(),
     )
     if completed.returncode != 0:
         error_text = completed.stderr.strip() or completed.stdout.strip() or "数据库脚本执行失败"

docs/rules/05-architecture.md

@@ -86,6 +86,7 @@ to make the directory tree look symmetrical.
 | `app/agent/lifecycle.py` | Runtime generation admission, initialization, idle-session collection and bounded manager shutdown |
 | `app/agent/tasks.py` | Background prompts, durable scheduled-task execution and heartbeat wakeups |
 | `app/agent/orchestrator.py` | One `MoviePilotAgent` execution instance: prompt/tool/middleware assembly, model invocation, streaming and per-agent state |
+| `app/agent/shell.py` | Agent command-shell selection and subprocess text-encoding policy; Windows prefers Git Bash, then PowerShell 7, while POSIX keeps native shell/PTY behavior |
 | `app/agent/policy/api.py` | Fixed `moviepilot_api` operation registry, HTTP route templates and per-operation authorization/effect policy; no arbitrary URL or method input |
 | `app/agent/policy/mcp.py` | Generated external MCP input-contract builder for the fixed API registry; owns exact English oneOf parameter projection, not runtime authorization |
 | `app/agent/tools/impl/service.py` | Admin-only external MCP wrappers for downloader, media-server and database Skill scripts; synchronous scripts run only through the Agent blocking executor |
@@ -977,6 +978,7 @@ driven workflow registration.
 | `app/agent/lifecycle.py` | Agent manager admission, startup, idle collection and bounded shutdown owner |
 | `app/agent/tasks.py` | Background prompt, scheduled task and heartbeat execution owner |
 | `app/agent/orchestrator.py` | Per-session `MoviePilotAgent` execution and LLM/tool/middleware orchestration only |
+| `app/agent/shell.py` | Agent command-shell selection and subprocess UTF-8 policy, with Windows-only Git Bash/PowerShell 7 routing |
 | `app/agent/loader.py` | Agent-specific capability discovery and canonical entrypoint/service materialization; reuses the generic Capability Runtime while keeping Agent ownership under `app/agent/` |
 | `app/agent/__init__.py` | Implementation-free package root; exact historical Agent symbols are supplied by the Compat overlay only, while host callers import `orchestrator.py` or the relevant owner directly |
 | `app/agent/llm/__init__.py` | Implementation-free package root; only the verified historical `LLMHelper` symbol is supplied by exact Compat routing |

tests/test_agent_shell.py

@@ -0,0 +1,361 @@
+import asyncio
+import json
+import subprocess
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from app.agent.llm.capability import OpenAIChatAudioProvider
+from app.agent.mcp import _StdioMcpSession
+from app.agent.prompt import PromptManager
+from app.agent.shell import (
+    AgentShell,
+    agent_text_subprocess_kwargs,
+    build_agent_subprocess_env,
+    resolve_agent_shell,
+)
+from app.agent.tools.impl._terminal_session import _TerminalSessionManager
+from app.agent.tools.impl.execute_command import ExecuteCommandTool
+from app.agent.tools.impl.service import _run_service_script
+from app.schemas.agent import AgentMcpServerConfig
+
+
+def _finder(paths: dict[str, str]):
+    """构造只返回测试声明命令路径的 which 替身。"""
+    return lambda command: paths.get(command)
+
+
+def _fake_process(stdout: bytes = b"", stderr: bytes = b""):
+    """构造满足 Agent 异步子进程读取合同的最小进程替身。"""
+    stdout_reader = asyncio.StreamReader()
+    stdout_reader.feed_data(stdout)
+    stdout_reader.feed_eof()
+    stderr_reader = asyncio.StreamReader()
+    stderr_reader.feed_data(stderr)
+    stderr_reader.feed_eof()
+
+    process = SimpleNamespace(
+        pid=12345,
+        returncode=0,
+        stdin=None,
+        stdout=stdout_reader,
+        stderr=stderr_reader,
+        terminate=lambda: None,
+        kill=lambda: None,
+    )
+
+    async def wait() -> int:
+        """立即返回成功退出码。"""
+        return 0
+
+    process.wait = wait
+    return process
+
+
+def test_windows_shell_prefers_git_bash(tmp_path: Path) -> None:
+    """Windows 检测到 Git for Windows 时应优先使用同目录 Git Bash。"""
+    git_root = tmp_path / "Git"
+    git_path = git_root / "cmd" / "git.exe"
+    bash_path = git_root / "bin" / "bash.exe"
+    git_path.parent.mkdir(parents=True)
+    bash_path.parent.mkdir(parents=True)
+    git_path.touch()
+    bash_path.touch()
+
+    shell = resolve_agent_shell(
+        platform_name="nt",
+        command_finder=_finder({"git": str(git_path), "pwsh": "ignored"}),
+        environment={},
+    )
+
+    assert shell is not None
+    assert shell.kind == "git-bash"
+    assert shell.build_argv("git status") == [
+        str(bash_path),
+        "--noprofile",
+        "--norc",
+        "-lc",
+        "git status",
+    ]
+    assert "Git Bash" in shell.prompt_guidance()
+    assert str(bash_path) not in shell.prompt_guidance()
+
+
+def test_windows_shell_uses_pwsh_when_git_is_unavailable() -> None:
+    """Windows 没有 Git 时应统一选择 PowerShell 7，而不是旧 powershell.exe。"""
+    shell = resolve_agent_shell(
+        platform_name="nt",
+        command_finder=_finder({"pwsh": "C:/PowerShell/pwsh.exe"}),
+        environment={},
+    )
+
+    assert shell is not None
+    assert shell.kind == "pwsh"
+    argv = shell.build_argv("Get-ChildItem")
+    assert argv[:4] == [
+        "C:/PowerShell/pwsh.exe",
+        "-NoLogo",
+        "-NoProfile",
+        "-Command",
+    ]
+    assert "[Console]::OutputEncoding" in argv[-1]
+    assert "chcp.com 65001" in argv[-1]
+    assert argv[-1].endswith("Get-ChildItem")
+    assert "PowerShell 7" in shell.prompt_guidance()
+
+
+def test_windows_shell_keeps_git_priority_when_only_pwsh_can_host_commands() -> None:
+    """Git 存在但无法定位 Git Bash 时，pwsh 仍应提示仓库操作优先使用 git。"""
+    shell = resolve_agent_shell(
+        platform_name="nt",
+        command_finder=_finder(
+            {"git": "C:/Portable/git.exe", "pwsh": "C:/PowerShell/pwsh.exe"}
+        ),
+        environment={},
+    )
+
+    assert shell is not None
+    assert shell.kind == "pwsh"
+    assert shell.git_available is True
+    assert "优先使用 `git`" in shell.prompt_guidance()
+
+
+def test_windows_shell_uses_utf8_cmd_fallback_without_git_or_pwsh() -> None:
+    """Git Bash 与 pwsh 都不可用时应保留显式 UTF-8 的 cmd 兼容回退。"""
+    shell = resolve_agent_shell(
+        platform_name="nt",
+        command_finder=_finder({}),
+        environment={"COMSPEC": "C:/Windows/System32/cmd.exe"},
+    )
+
+    assert shell is not None
+    assert shell.kind == "cmd"
+    assert shell.build_argv("dir")[-1] == "chcp 65001>nul &\ndir"
+
+
+def test_posix_shell_policy_preserves_native_shell_and_environment(monkeypatch) -> None:
+    """Linux/macOS 不应被 Windows shell 选择或 UTF-8 环境覆盖改变。"""
+    monkeypatch.setenv("PYTHONIOENCODING", "locale-default")
+
+    assert resolve_agent_shell(platform_name="posix") is None
+    environment = build_agent_subprocess_env(
+        {"CUSTOM": "value"}, platform_name="posix"
+    )
+    assert environment["PYTHONIOENCODING"] == "locale-default"
+    assert environment["CUSTOM"] == "value"
+    assert agent_text_subprocess_kwargs(platform_name="posix") == {}
+
+
+def test_windows_subprocess_environment_forces_utf8_after_overrides() -> None:
+    """Windows 调用方不能通过自定义环境重新引入本地默认编码。"""
+    environment = build_agent_subprocess_env(
+        {
+            "PYTHONIOENCODING": "cp936",
+            "PYTHONLEGACYWINDOWSSTDIO": "1",
+            "PYTHONUTF8": "0",
+            "LANG": "zh_CN.GBK",
+        },
+        platform_name="nt",
+    )
+
+    assert environment["PYTHONIOENCODING"] == "utf-8"
+    assert environment["PYTHONUTF8"] == "1"
+    assert environment["LANG"] == "C.UTF-8"
+    assert environment["LC_ALL"] == "C.UTF-8"
+    assert "PYTHONLEGACYWINDOWSSTDIO" not in environment
+
+
+def test_prompt_injects_selected_windows_shell_without_executable_path() -> None:
+    """模型提示词应知道实际 Windows shell，但不能暴露安装绝对路径。"""
+    shell = AgentShell(
+        kind="pwsh",
+        executable="C:/secret/location/pwsh.exe",
+        arguments=("-Command",),
+    )
+    with patch("app.agent.prompt.resolve_agent_shell", return_value=shell):
+        moviepilot_info = PromptManager()._get_moviepilot_info()
+
+    assert "PowerShell 7" in moviepilot_info
+    assert "C:/secret/location/pwsh.exe" not in moviepilot_info
+
+
+@pytest.mark.anyio
+async def test_execute_command_uses_selected_windows_shell_and_utf8_environment() -> None:
+    """一次性命令在 Windows 策略下应绕过隐式 cmd 并传入 UTF-8 环境。"""
+    shell = AgentShell(
+        kind="pwsh",
+        executable="pwsh.exe",
+        arguments=("-Command",),
+        command_prefix="utf8-prefix;",
+    )
+    process = _fake_process(stdout="中文输出".encode("utf-8"))
+    create_exec = AsyncMock(return_value=process)
+    tool = ExecuteCommandTool(session_id="session", user_id="user")
+
+    with (
+        patch(
+            "app.agent.tools.impl.execute_command.resolve_agent_shell",
+            return_value=shell,
+        ),
+        patch(
+            "app.agent.tools.impl.execute_command.build_agent_subprocess_env",
+            return_value={"PYTHONUTF8": "1"},
+        ),
+        patch(
+            "app.agent.tools.impl.execute_command.asyncio.create_subprocess_exec",
+            create_exec,
+        ),
+        patch(
+            "app.agent.tools.impl.execute_command.asyncio.create_subprocess_shell"
+        ) as create_shell,
+    ):
+        result = await tool.run(action="run", command="Get-Location", timeout=1)
+
+    assert "中文输出" in result
+    assert create_exec.await_args.args == (
+        "pwsh.exe",
+        "-Command",
+        "utf8-prefix;\nGet-Location",
+    )
+    assert create_exec.await_args.kwargs["env"] == {"PYTHONUTF8": "1"}
+    create_shell.assert_not_called()
+
+
+@pytest.mark.anyio
+async def test_terminal_session_uses_selected_windows_shell() -> None:
+    """后台会话与一次性命令必须共享同一 Windows shell 策略。"""
+    shell = AgentShell(
+        kind="git-bash",
+        executable="bash.exe",
+        arguments=("-lc",),
+    )
+    process = _fake_process()
+    create_exec = AsyncMock(return_value=process)
+    manager = _TerminalSessionManager()
+
+    with (
+        patch(
+            "app.agent.tools.impl._terminal_session.resolve_agent_shell",
+            return_value=shell,
+        ),
+        patch(
+            "app.agent.tools.impl._terminal_session.asyncio.create_subprocess_exec",
+            create_exec,
+        ),
+        patch(
+            "app.agent.tools.impl._terminal_session.asyncio.create_subprocess_shell"
+        ) as create_shell,
+    ):
+        session = await manager._start_pipe_session(
+            "git status", "C:/MoviePilot", {"PYTHONUTF8": "1"}
+        )
+        assert session.wait_task is not None
+        await session.wait_task
+
+    assert create_exec.await_args.args == ("bash.exe", "-lc", "git status")
+    assert create_exec.await_args.kwargs["env"] == {"PYTHONUTF8": "1"}
+    create_shell.assert_not_called()
+
+
+@pytest.mark.anyio
+async def test_stdio_mcp_inherits_agent_utf8_environment() -> None:
+    """stdio MCP 子进程也必须继承 Windows Agent 的 UTF-8 环境。"""
+    process = _fake_process()
+    create_exec = AsyncMock(return_value=process)
+    server = AgentMcpServerConfig(
+        id="fake",
+        name="Fake MCP",
+        transport="stdio",
+        command="server.exe",
+        env={"PYTHONIOENCODING": "cp936"},
+    )
+    session = _StdioMcpSession(server)
+
+    with (
+        patch(
+            "app.agent.mcp.build_agent_subprocess_env",
+            return_value={"PYTHONIOENCODING": "utf-8"},
+        ) as build_env,
+        patch("app.agent.mcp.asyncio.create_subprocess_exec", create_exec),
+    ):
+        await session.__aenter__()
+        await session.__aexit__(None, None, None)
+
+    build_env.assert_called_once_with(server.env)
+    assert create_exec.await_args.kwargs["env"] == {"PYTHONIOENCODING": "utf-8"}
+
+
+def test_service_script_uses_windows_utf8_text_subprocess_kwargs(tmp_path: Path) -> None:
+    """固定 Skill Python 子进程应显式使用 UTF-8 文本解码和环境。"""
+    completed = subprocess.CompletedProcess(
+        args=[],
+        returncode=0,
+        stdout=json.dumps({"success": True}, ensure_ascii=False),
+        stderr="",
+    )
+    with (
+        patch(
+            "app.agent.tools.impl.service.get_runtime_setting",
+            return_value=tmp_path,
+        ),
+        patch(
+            "app.agent.tools.impl.service.agent_text_subprocess_kwargs",
+            return_value={
+                "encoding": "utf-8",
+                "errors": "replace",
+                "env": {"PYTHONUTF8": "1"},
+            },
+        ),
+        patch(
+            "app.agent.tools.impl.service.subprocess.run",
+            return_value=completed,
+        ) as runner,
+    ):
+        payload = _run_service_script(
+            relative_script="skills/demo.py",
+            selector_flag=None,
+            selector_value=None,
+            action="list",
+            arguments={},
+        )
+
+    assert payload["success"] is True
+    assert runner.call_args.kwargs["encoding"] == "utf-8"
+    assert runner.call_args.kwargs["errors"] == "replace"
+    assert runner.call_args.kwargs["env"] == {"PYTHONUTF8": "1"}
+
+
+def test_audio_ffmpeg_uses_windows_utf8_text_subprocess_kwargs(tmp_path: Path) -> None:
+    """Agent 音频 ffmpeg 文本输出也应走统一的 Windows UTF-8 解码参数。"""
+    input_path = tmp_path / "input.wav"
+    input_path.write_bytes(b"wav")
+    expected_output = input_path.with_suffix(".opus")
+
+    def run_ffmpeg(_command, **_kwargs):
+        """模拟 ffmpeg 成功生成输出文件。"""
+        expected_output.write_bytes(b"opus")
+        return subprocess.CompletedProcess([], 0, "", "")
+
+    provider = OpenAIChatAudioProvider()
+    with (
+        patch(
+            "app.agent.llm.capability.agent_text_subprocess_kwargs",
+            return_value={
+                "encoding": "utf-8",
+                "errors": "replace",
+                "env": {"PYTHONUTF8": "1"},
+            },
+        ),
+        patch(
+            "app.agent.llm.capability.subprocess.run",
+            side_effect=run_ffmpeg,
+        ) as runner,
+    ):
+        output_path = provider._convert_wav_to_opus(input_path)
+
+    assert output_path == expected_output
+    assert runner.call_args.kwargs["encoding"] == "utf-8"
+    assert runner.call_args.kwargs["errors"] == "replace"
+    assert runner.call_args.kwargs["env"] == {"PYTHONUTF8": "1"}