feat(agent): unified cross-platform shell execution contract (cmd/bash)

Windows runs 'cmd.exe /d /s /c <command>' as a string command line; Linux uses ['/bin/bash', '-lc', command]. Removes implicit shell=True behavior and aligns the agent system prompt per platform.

SYSTEM_PROMPT.md

@@ -13,37 +13,19 @@
 
 ## 1. 运行环境
 
-- 操作系统：Windows。文件路径形如 `C:\Users\14890\Desktop\haocode`。
 - 当前工作目录：**haocode 项目根目录**。所有相对路径都相对它解析；
   每条 bash 命令都以它作为工作目录启动。
-- Python：conda 环境 `haocode`（Python 3.10，PyQt6、openai 已装），直接用 `python`。
+- Python：直接用 `python`（3.10 环境，PyQt6、openai 已装）。
 - 前端是本地网页（`ui/web/`），改动前端文件后需重启应用才生效。
 
-### 1.1 shell 真相（重要：直接决定命令能不能跑对）
-
-命令经 **cmd.exe** 执行（不是 git-bash）。但 `C:\Program Files\Git\usr\bin` 在 PATH 上，
-所以 `ls` `grep` `cat` `head` `tail` `wc` `rm` `sed` `awk` 都能直接用，
-管道 `|`、重定向 `>` `2>&1`、`&&` 也都可用。
-
-⚠️ 下列写法会**静默出错**或报错，务必按右列的写法：
-
-| ❌ 不要写 | ✅ 改成 | 原因 |
-|---|---|---|
-| `echo a; echo b` | `echo a && echo b`（或分两行写） | cmd 不认 `;`，会把 `; echo b` 当参数原样输出 |
-| `echo $HOME` | `echo %USERPROFILE%` | cmd 用 `%VAR%`；`$VAR` 不会被展开 |
-| `for i in 1 2 3; do ...; done` | `bash -c "for i in 1 2 3; do ...; done"` | bash 语法必须显式调用 bash |
-| 单独一条 `cd core` | `cd core && <命令>` | **每次调用都是新进程，cd 不会跨调用保留** |
-| `echo 'x'` | `echo x` | cmd 内建命令不剥单引号（`grep 'x'` 等 msys 程序会正常剥） |
-
-多条命令用**换行**分隔最稳（实测可用）。需要 `$(...)`、单引号、`[ ]` 测试等真实
-bash 语义时，一律写成 `bash -c "..."`。
+{{SHELL_PLATFORM_SECTION}}
 
 ## 2. 可用工具
 
 | 工具 | 用途 | 关键约束 |
 |---|---|---|
 | `read` | 读取**文本**文件（带行号） | 单次 ≤2000 行 / 50KB；大文件用 `offset`/`limit` 分页；**不要用于图片或二进制** |
-| `bash` | 执行 shell 命令 | 经 cmd.exe；默认 120 秒超时（上限 600）；输出 50KB 截断；**同批有它则整批串行** |
+| `bash` | 执行 shell 命令 | 默认 120 秒超时（上限 600）；输出 50KB 截断；**同批有它则整批串行** |
 | `write` | 新建或**完整覆盖**文件 | 自动创建父目录；原子写入；只用于新建或整体重写 |
 | `edit` | 精确文本替换 | `oldText` 必须与**原文件**逐字符一致且唯一；各条区间不得重叠；**整批全有或全无** |
 
@@ -70,13 +52,13 @@ bash 语义时，一律写成 `bash -c "..."`。
 
 | 参数 | 类型 | 必填 | 默认 | 说明 |
 |---|---|---|---|---|
-| `command` | string | ✅ | — | 要执行的命令（cmd.exe 语法，可用 Git 的 unix 工具） |
+| `command` | string | ✅ | — | 要执行的命令（平台 shell 语法见第 1.1 节） |
 | `timeout` | number | | 120 | 超时秒数，**上限 600**（传更大按 600） |
 
 - 返回值：`$ 命令` + stdout + `[stderr]` + `[exit N] (耗时)`；**退出码非 0 时结果视为失败**。
 - 超时：到期会**杀掉整棵进程树**并返回 `命令超时（>Ns）已终止`。
   长任务（全量测试、构建、下载）请显式传 `timeout`；短查询不必传。
-- 输出超过 50KB 会被截断 → 用 `-n` / `head` / `findstr` 或更精确的命令收窄输出后再逐步放宽。
+- 输出超过 50KB 会被截断 → 用 `-n` / `head` 或更精确的命令收窄输出后再逐步放宽。
 - 需要等待的场景直接跑命令并设好 `timeout`，不要用反复 `sleep` 试探。
 
 ### 3.3 write
@@ -128,7 +110,7 @@ bash 语义时，一律写成 `bash -c "..."`。
 
 | 工具返回 | 含义与你的动作 |
 |---|---|
-| `文件不存在: <绝对路径>` | 路径写错了。用 `bash` 的 `dir` / `ls` 确认真实路径，**不要猜** |
+| `文件不存在: <绝对路径>` | 路径写错了。用 `bash` 的 `ls` 确认真实路径，**不要猜** |
 | `path 不能为空` / `command 不能为空` | 参数缺失，补齐后重试 |
 | `[起始行 offset=N 超出文件范围，该文件共 M 行]` | 用 M 以内的 offset 重读 |
 | `[文件为空（0 行）]` | 文件确实为空 → 用 `write` |
@@ -147,7 +129,7 @@ bash 语义时，一律写成 `bash -c "..."`。
 1. **先看清再动手**：改代码前先 `read` / `bash` 确认现状；不凭空猜路径、函数名、行号。
 2. **小步快跑**：一次做一个明确改动；改完立刻用 `bash` 验证（编译、测试、脚本）。
 3. **验证要真实**：说「已完成」之前必须有工具输出作证据（命令结果 / 测试结果）。
-4. **推荐流程**：定位（`grep` / `dir`）→ 精读（`read`）→ 改动（`edit` / `write`）→ 验证（`bash`）。
+4. **推荐流程**：定位（`grep` / `ls`）→ 精读（`read`）→ 改动（`edit` / `write`）→ 验证（`bash`）。
 5. **范围克制**：只做用户要求的事；不顺手重构、不批量格式化、不改无关文件。
 6. **不谎报**：没跑过的命令不说「已运行」；没读到的内容不说「文件里是…」；失败就照实说失败。
 7. **输出克制**：结论先行、简洁；长内容用列表/表格；不复述用户原话；涉及文件时写清路径。
@@ -160,7 +142,7 @@ bash 语义时，一律写成 `bash -c "..."`。
   （除非用户在本轮明确要求并给出路径）。
 - **禁止**读取或输出 `data/config.json` 中的 API 密钥等敏感内容
   （可以确认文件存在，但不要展示内容）。
-- **禁止**向 conda 环境 `haocode` 安装或卸载包；**禁止**修改系统目录、注册表、环境变量。
+- **禁止**在当前 Python 环境安装或卸载包；**禁止**修改系统目录、注册表、环境变量。
 - 网络请求只允许用户已配置的 API 端点；不要主动上传数据或抓取外部内容。
 - 涉及用户数据（`data/*.db`）默认只读；除用户明确要求，不要写入或迁移数据。
 

core/agent/tools.py

@@ -32,6 +32,7 @@ from typing import Any, Callable, Dict, List, Optional
 
 from .types import (AgentMessage, AgentTool, AgentToolResult, AbortSignal,
                     ToolCall, new_id)
+from ..platform_shell import kill_process_tree, popen_flags, shell_command  # P1-02
 
 try:
     from core.debug_log import debug_log as _dbg_log  # 🆕 计时观察日志（线程安全/静默）
@@ -448,8 +449,8 @@ def tool_bash(tool_call_id: str, args: Dict[str, Any], signal: AbortSignal,
         return AgentToolResult.text("command 不能为空", is_error=True)
     timeout = min(float(args.get("timeout", 120)), 600)
     cwd = ctx.get("cwd") or os.getcwd()
-    # 对照 pi bash 工具：命令始终经 shell 解释（支持管道/别名/内置命令）
-    use_shell = ctx.get("shell", True)
+    # P1-02：显式 shell 契约（Windows cmd.exe /d /c；Linux /bin/bash -lc），
+    # 不再依赖 shell=True 的平台默认值；ctx 的 "shell" 键已废弃（bash 工具即 shell 工具）
     on_timer = ctx.get("on_timer")
     t0 = time.time()
     # 🆕 计时观察①：计时器启动时刻 + 模型实际传的 timeout 值
@@ -466,19 +467,9 @@ def tool_bash(tool_call_id: str, args: Dict[str, Any], signal: AbortSignal,
             pass
 
     def _kill_tree(proc):
-        """杀整个进程树：Windows 上默认 kill 只杀 cmd 壳，孤儿子进程继续
-        持管道 → 假超时（设定10s 实际20s）。taskkill /T 整树杀。"""
-        try:
-            if os.name == "nt":
-                subprocess.run(["taskkill", "/F", "/T", "/PID", str(proc.pid)],
-                               capture_output=True, timeout=10)
-            else:
-                proc.kill()
-        except Exception:
-            try:
-                proc.kill()
-            except Exception:
-                pass
+        """杀整个进程树：Windows taskkill /F /T；Linux 独立进程组
+        SIGTERM → 宽限 → SIGKILL（见 core.platform_shell.kill_process_tree）。"""
+        kill_process_tree(proc)
 
     # 🐛 修复：text=True 不带 encoding 时按系统码页（中文 Windows=GBK）解码，
     # 子进程输出 UTF-8（python/git/中文 echo）→ _readerthread UnicodeDecodeError。
@@ -486,10 +477,11 @@ def tool_bash(tool_call_id: str, args: Dict[str, Any], signal: AbortSignal,
     _env = dict(os.environ, PYTHONIOENCODING="utf-8")
     try:
         proc = subprocess.Popen(
-            command, shell=use_shell,
+            shell_command(command),  # P1-02：显式 shell 命令
             stdout=subprocess.PIPE, stderr=subprocess.PIPE,
             text=True, encoding="utf-8", errors="replace",
             cwd=cwd, env=_env,
+            **popen_flags(),  # P1-02：Linux 独立进程组（start_new_session）
         )
     except Exception as e:
         return AgentToolResult.text(f"执行失败: {e}", is_error=True)

core/llm_engine.py

@@ -37,8 +37,10 @@ from core.agent import (Agent, AgentConfig, AgentEvent, AgentMessage, AgentRunne
                         default_tools, from_openai_messages,
                         openai_stream)
 from core.agent.stream_fn import _pick_reasoning, _pick_usage
+# P0-01：配置读取统一入口（环境变量 HAOCODE_CONFIG_FILE 优先，缺省回落项目内 data/config.json）
+from core.config_paths import load_config
+from core import platform_shell  # P1-02：提示词平台段（通用正文 + 短 shell/path 段）
 
-CONFIG_PATH = os.path.join(os.path.dirname(__file__), "..", "data", "config.json")
 PROJECT_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
 # 🌟 Agent 系统提示词文件（项目根目录，core 前面那个 .md）
 SYSTEM_PROMPT_FILE = os.path.join(PROJECT_ROOT, "SYSTEM_PROMPT.md")
@@ -49,25 +51,25 @@ _FALLBACK_SYSTEM_PROMPT = (
 
 
 def load_system_prompt() -> str:
-    """读取 SYSTEM_PROMPT.md；缺失时用兜底短提示词。"""
+    """读取 SYSTEM_PROMPT.md；缺失时用兜底短提示词。
+
+    P1-02：通用正文 + 运行时插入的短平台 shell/path 段（占位符替换）；
+    每次请求重读本函数即重读文件，Windows 段不会出现在 Linux 请求中，反之亦然。
+    """
+    text = None
     try:
         with open(SYSTEM_PROMPT_FILE, "r", encoding="utf-8") as f:
             text = f.read().strip()
-        if text:
-            return text
     except FileNotFoundError:
         pass
-    return _FALLBACK_SYSTEM_PROMPT
+    if not text:
+        text = _FALLBACK_SYSTEM_PROMPT
+    return platform_shell.apply_platform_section(text)
 
 
 def _load_config() -> dict:
-    if os.path.exists(CONFIG_PATH):
-        try:
-            with open(CONFIG_PATH, "r", encoding="utf-8") as f:
-                return json.load(f)
-        except Exception as e:
-            print(f"[llm_engine] 读取配置失败: {e}")
-    return {}
+    """P0-01：统一走 core.config_paths.load_config（环境变量优先、容错、可见警告）。"""
+    return load_config()
 
 
 def _provider_info(config: dict, provider_name: str) -> dict:

core/platform_shell.py

@@ -0,0 +1,141 @@
+# -*- coding: utf-8 -*-
+"""P1-02：窄平台进程适配 —— 显式 shell 契约 + 进程树终止 + 提示词平台段。
+
+契约（不依赖 shell=True 的平台默认值）：
+- Windows：字符串命令行 ``cmd.exe /d /s /c "<command>"``（/s 保留内部引号；
+  不能传 argv 列表——list2cmdline 的引号转义 cmd 不认）
+- Linux：  ``/bin/bash -lc <command>``（登录 shell，可读 profile 别名/函数）
+
+进程树终止：
+- Windows：taskkill /F /T（既有等价机制，整树强杀）
+- Linux：  start_new_session 独立 POSIX 进程组；SIGTERM 整组 → 宽限期 → SIGKILL 整组
+  （安全不变量：仅当 getpgid == 子进程 pid，即确认独立组后才 killpg，
+  否则退化为单进程 kill，绝不误杀调用方所在组）
+
+系统提示词：通用正文 + 运行时插入的短平台 shell/path 段（占位符
+``{{SHELL_PLATFORM_SECTION}}``）。不维护两份完整提示词。
+"""
+import os
+import signal
+import subprocess
+import time
+
+#: 系统提示词中的平台段占位符
+SHELL_SECTION_PLACEHOLDER = "{{SHELL_PLATFORM_SECTION}}"
+
+_KILL_GRACE_S = 3.0      # Linux：SIGTERM → SIGKILL 宽限
+_TASKKILL_TIMEOUT_S = 10  # Windows：taskkill 自身超时
+
+
+def is_windows() -> bool:
+    return os.name == "nt"
+
+
+def shell_command(command: str):
+    """显式 shell 命令（bash 工具的 shell 契约，替代 shell=True 平台默认值）。
+
+    - Windows：返回**字符串命令行** ``cmd.exe /d /s /c "<command>"``。
+      必须走字符串而非 argv 列表：列表形态会被 CPython ``list2cmdline``
+      给内部引号加反斜杠转义，cmd.exe 不认 ``\"``；字符串形态下命令行
+      原样交给 CreateProcessW，cmd 自行按 /s 规则解析 /c 参数
+      （外层引号剥离、内部引号保留，实测引号路径/&&/管道/%VAR% 均正确）。
+    - Linux：返回 argv 列表 ``[\ "/bin/bash", "-lc", command]``（POSIX 下
+      列表参数无引号转义问题，-l 登录 shell 可读 profile 别名/函数）。
+    """
+    if is_windows():
+        return 'cmd.exe /d /s /c "' + command + '"'
+    return ["/bin/bash", "-lc", command]
+
+
+def popen_flags() -> dict:
+    """进程树管理所需的 Popen 附加参数：Linux 独立进程组。"""
+    if is_windows():
+        return {}
+    return {"start_new_session": True}
+
+
+def kill_process_tree(proc, grace_s: float = _KILL_GRACE_S) -> None:
+    """终止 proc 的整棵子进程树；最坏情况也保证返回（不抛异常）。
+
+    - Windows：taskkill /F /T 整树强杀（既有机制）。
+    - Linux：独立进程组 SIGTERM → 宽限 grace_s 秒 → SIGKILL 整组。
+      若无法确认独立组（进程已退 / 标志未生效）→ 退化单进程 kill。
+    """
+    if proc is None:
+        return
+
+    if is_windows():
+        try:
+            subprocess.run(["taskkill", "/F", "/T", "/PID", str(proc.pid)],
+                           capture_output=True, timeout=_TASKKILL_TIMEOUT_S)
+        except Exception:
+            try:
+                proc.kill()
+            except Exception:
+                pass
+        return
+
+    # ---- POSIX ----
+    pgid = None
+    try:
+        pgid = os.getpgid(proc.pid)
+    except Exception:
+        pass
+    if pgid is None or pgid != proc.pid:
+        # 进程已退出，或没有独立组 → 不能 killpg（可能误伤调用方组）
+        try:
+            proc.kill()
+        except Exception:
+            pass
+        return
+
+    def _pg(sig: int) -> None:
+        try:
+            os.killpg(pgid, sig)
+        except Exception:
+            pass
+
+    _pg(signal.SIGTERM)
+    deadline = time.time() + grace_s
+    while time.time() < deadline:
+        try:
+            if proc.poll() is not None:
+                return
+        except Exception:
+            return
+        time.sleep(0.05)
+    _pg(signal.SIGKILL)
+
+
+# ----------------------------------------------------------------------
+# 系统提示词平台段（短段，运行时插入；通用正文保持平台无关）
+# ----------------------------------------------------------------------
+
+_WIN_SECTION = """### 1.1 平台 shell 与路径（Windows）
+
+- 操作系统：Windows。路径形如 `C:\\Users\\14890\\Desktop\\haocode`；不确定真实路径时先 `dir` / `ls`。
+- bash 命令经 **cmd.exe** 执行（不是 git-bash）；`C:\\Program Files\\Git\\usr\\bin` 在 PATH 上，
+  `ls` `grep` `cat` `head` `tail` `wc` `rm` `sed` `awk` 可直接用，`|` `>` `2>&1` `&&` 可用。
+- ⚠️ cmd 陷阱：`;` 不是分隔符（用 `&&` 或换行）；环境变量用 `%VAR%`（`$VAR` 不展开）；
+  内建命令不剥单引号（`echo 'x'` 原样带引号）；每次调用都是新进程，`cd` 不跨调用保留
+  （写 `cd core && <命令>`）；需要 `for` / `$(...)` / `[ ]` 测试等真实 bash 语义时
+  一律 `bash -c "..."`。"""
+
+_LINUX_SECTION = """### 1.1 平台 shell 与路径（Linux）
+
+- 操作系统：Linux。路径为 POSIX 风格（`/home/user/...`、`/tmp/...`）。
+- bash 命令经 **/bin/bash -lc** 执行（登录 shell）：`;` `&&` `|` `$(...)`、单引号、
+  环境变量 `$VAR`、profile 里的别名与函数都可用。
+- 每次调用都是新进程：`cd` 不跨调用保留（写 `cd core && <命令>` 或直接用绝对路径）。"""
+
+
+def shell_prompt_section() -> str:
+    """当前平台的短 shell/path 段。"""
+    return _WIN_SECTION if is_windows() else _LINUX_SECTION
+
+
+def apply_platform_section(text: str) -> str:
+    """把占位符替换为当前平台段；无占位符（如兜底提示词）时原样返回。"""
+    if SHELL_SECTION_PLACEHOLDER in text:
+        return text.replace(SHELL_SECTION_PLACEHOLDER, shell_prompt_section())
+    return text

tests/test_cross_platform_shell.py

@@ -0,0 +1,253 @@
+# -*- coding: utf-8 -*-
+"""
+P1-02 跨平台 shell 契约单测（纯函数级 + 真实进程级，无 UI 依赖）
+
+覆盖：
+  A. 平台参数：Windows `cmd.exe /d /c` argv / Linux `/bin/bash -lc` argv；
+     Linux Popen 独立进程组（start_new_session），Windows 无该参数
+  B. 提示词平台段：通用正文完全相同、只插入对应平台段、互不串段；
+     load_system_prompt() 与文件内容一致（当前平台）
+  C. 进程树（真实进程，父+孙）：
+     - 当前平台：超时 → 父与孙都不存在
+     - 当前平台：主动中止 → 父与孙都不存在
+     - Linux 专属：独立进程组整组终止（Windows 上跳过）
+  D. kill_process_tree 安全边界：已退出的进程 / 无独立组 → 不抛异常、不误杀
+
+运行: PYTHONIOENCODING=utf-8 python tests/test_cross_platform_shell.py
+"""
+import os
+import subprocess
+import sys
+import tempfile
+import time
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from core import platform_shell  # noqa: E402
+from core.agent.tools import tool_bash  # noqa: E402
+from core.agent.types import AbortSignal  # noqa: E402
+
+RESULTS = []
+
+
+def check(name, cond, extra=""):
+    RESULTS.append((name, bool(cond)))
+    print(f"{'PASS' if cond else 'FAIL'}  {name}  {extra if not cond else ''}", flush=True)
+
+
+def text_of(res):
+    c = res.content
+    if isinstance(c, str):
+        return c
+    return "".join(x.get("text", "") for x in c if isinstance(x, dict))
+
+
+_TMP = tempfile.mkdtemp(prefix="haocode_crossshell_")
+PY = sys.executable
+IS_WIN = os.name == "nt"
+
+# ======================================================================
+# A. 平台参数（mock 平台，纯逻辑）
+# ======================================================================
+_orig_is_win = platform_shell.is_windows
+
+
+def _mock_win(v):
+    platform_shell.is_windows = lambda: v
+
+
+try:
+    _mock_win(True)
+    check("A1 Windows 命令 = cmd.exe /d /s /c \"<cmd>\"",
+          platform_shell.shell_command("echo hi")
+          == 'cmd.exe /d /s /c "echo hi"',
+          repr(platform_shell.shell_command("echo hi")))
+    check("A2 Windows popen_flags 无进程组参数",
+          platform_shell.popen_flags() == {}, repr(platform_shell.popen_flags()))
+
+    _mock_win(False)
+    check("A3 Linux argv = /bin/bash -lc <cmd>",
+          platform_shell.shell_command("echo hi") == ["/bin/bash", "-lc", "echo hi"],
+          repr(platform_shell.shell_command("echo hi")))
+    check("A4 Linux popen_flags 独立进程组",
+          platform_shell.popen_flags() == {"start_new_session": True},
+          repr(platform_shell.popen_flags()))
+finally:
+    platform_shell.is_windows = _orig_is_win
+
+# ======================================================================
+# B. 提示词平台段
+# ======================================================================
+_PROMPT_FILE = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
+                            "SYSTEM_PROMPT.md")
+with open(_PROMPT_FILE, "r", encoding="utf-8") as f:
+    _PROMPT_TEXT = f.read().strip()
+
+check("B1 通用正文含占位符",
+      platform_shell.SHELL_SECTION_PLACEHOLDER in _PROMPT_TEXT)
+
+_GENERIC_NO_LEAK = ("cmd.exe" not in _PROMPT_TEXT
+                    and "/bin/bash" not in _PROMPT_TEXT)
+check("B2 通用正文无平台泄漏（无 cmd.exe / 无 /bin/bash）", _GENERIC_NO_LEAK)
+
+try:
+    _mock_win(True)
+    full_win = platform_shell.apply_platform_section(_PROMPT_TEXT)
+    sec_win = platform_shell.shell_prompt_section()
+    _mock_win(False)
+    full_lin = platform_shell.apply_platform_section(_PROMPT_TEXT)
+    sec_lin = platform_shell.shell_prompt_section()
+finally:
+    platform_shell.is_windows = _orig_is_win
+
+check("B3 Windows 段含 cmd.exe、无 /bin/bash",
+      "cmd.exe" in sec_win and "/bin/bash" not in sec_win)
+check("B4 Linux 段含 /bin/bash -lc、无 cmd.exe",
+      "/bin/bash -lc" in sec_lin and "cmd.exe" not in sec_lin)
+check("B5 只有对应平台段被插入",
+      full_win == _PROMPT_TEXT.replace(platform_shell.SHELL_SECTION_PLACEHOLDER, sec_win)
+      and full_lin == _PROMPT_TEXT.replace(platform_shell.SHELL_SECTION_PLACEHOLDER, sec_lin))
+check("B6 通用正文两平台完全相同",
+      full_win.replace(sec_win, "§SEC§") == full_lin.replace(sec_lin, "§SEC§"))
+check("B7 无占位符文本原样返回",
+      platform_shell.apply_platform_section("无占位符的兜底文本") == "无占位符的兜底文本")
+
+# load_system_prompt()（当前平台，真实文件）
+from core import llm_engine  # noqa: E402  (PyQt6/openai 已装，无需 QApplication)
+check("B8 load_system_prompt = 文件内容 + 当前平台段",
+      llm_engine.load_system_prompt()
+      == platform_shell.apply_platform_section(_PROMPT_TEXT))
+
+# ======================================================================
+# C. 进程树（真实进程：父 + 孙，心跳文件证明生死）
+# ======================================================================
+_CHILD = os.path.join(_TMP, "child.py")
+with open(_CHILD, "w", encoding="utf-8") as f:
+    f.write(
+        "import sys, time\n"
+        "hb = open(sys.argv[1], 'a', encoding='utf-8')\n"
+        "t0 = time.time()\n"
+        "while time.time() - t0 < 30:\n"
+        "    hb.write(f'{time.time():.3f}\\n')\n"
+        "    hb.flush()\n"
+        "    time.sleep(0.2)\n"
+        "open(sys.argv[2], 'w').write('done')\n"
+    )
+
+
+def _spawn_parent(hb_path, done_path, out_path):
+    """写一个父进程脚本（启动孙进程后挂 30s），返回 bash 命令字符串。"""
+    parent = os.path.join(_TMP, f"parent_{os.path.basename(hb_path)}.py")
+    with open(parent, "w", encoding="utf-8") as f:
+        f.write(
+            "import subprocess, sys, time\n"
+            "p = subprocess.Popen([sys.executable, "
+            + repr(_CHILD) + ", sys.argv[1], sys.argv[2]], "
+            "stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)\n"
+            "time.sleep(30)\n"
+        )
+    cmd = f'"{PY}" -u "{parent}" "{hb_path}" "{done_path}"'
+    return cmd
+
+
+def _hb_last(hb_path):
+    try:
+        with open(hb_path, "r", encoding="utf-8") as f:
+            lines = [l.strip() for l in f if l.strip()]
+        return float(lines[-1]) if lines else None
+    except FileNotFoundError:
+        return None
+
+
+def _tree_dead(hb_path, done_path, quiesce=1.2):
+    """等待静默后判断：孙进程不再有心跳、且未跑完 → 进程树已死。"""
+    time.sleep(quiesce)
+    last = _hb_last(hb_path)
+    return last is not None and (time.time() - last) > 0.6, os.path.exists(done_path)
+
+
+def run_tree_case(tag, timeout=None, abort_after=None):
+    hb = os.path.join(_TMP, f"hb_{tag}.txt")
+    done = os.path.join(_TMP, f"done_{tag}.txt")
+    out = os.path.join(_TMP, f"out_{tag}.log")
+    cmd = _spawn_parent(hb, done, out)
+
+    updates, timers = [], []
+    sig = AbortSignal()
+    if abort_after is not None:
+        import threading
+        def _aborter():
+            time.sleep(abort_after)
+            sig.abort("test")
+        threading.Thread(target=_aborter, daemon=True).start()
+
+    t0 = time.time()
+    res = tool_bash(f"call_{tag}", {"command": cmd, "timeout": timeout or 60}, sig,
+                    lambda t: updates.append(t),
+                    {"cwd": _TMP, "on_timer": lambda e, tt: timers.append((e, tt))})
+    dur = time.time() - t0
+    return res, dur, hb, done, timers
+
+
+# ---- C1: 超时 → 父与孙都不存在 ----
+res, dur, hb, done, timers = run_tree_case("timeout", timeout=3)
+txt = text_of(res)
+dead, finished = _tree_dead(hb, done)
+check("C1.1 超时返回错误结果", res.is_error and "超时" in txt, f"{txt!r}")
+check("C1.2 超时快速返回（<15s，含 taskkill 缓冲）", dur < 15, f"dur={dur:.1f}s")
+check("C1.3 超时后父+孙都不存在（心跳停止且未跑完）", dead and not finished,
+      f"hb_last={_hb_last(hb)} finished={finished}")
+check("C1.4 读秒滴答仍工作", len(timers) >= 2, f"timers={timers}")
+
+# ---- C2: 主动中止 → 父与孙都不存在 ----
+res, dur, hb, done, _timers2 = run_tree_case("abort", timeout=60, abort_after=1.5)
+txt = text_of(res)
+dead, finished = _tree_dead(hb, done)
+check("C2.1 中止返回错误结果", res.is_error and "中止" in txt, f"{txt!r}")
+check("C2.2 中止后父+孙都不存在（心跳停止且未跑完）", dead and not finished,
+      f"hb_last={_hb_last(hb)} finished={finished}")
+
+# ---- C3: Linux 独立进程组整组终止（Windows 跳过） ----
+if not IS_WIN:
+    proc = subprocess.Popen(
+        platform_shell.shell_command("sleep 30"),
+        stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
+        **platform_shell.popen_flags())
+    time.sleep(0.3)
+    try:
+        pgid = os.getpgid(proc.pid)
+        check("C3.1 start_new_session 生效（pgid == 子 pid）", pgid == proc.pid,
+              f"pgid={pgid} pid={proc.pid}")
+        platform_shell.kill_process_tree(proc, grace_s=1.0)
+        time.sleep(0.5)
+        check("C3.2 整组终止后子进程不存在",
+              proc.poll() is not None, f"poll={proc.poll()}")
+    except Exception as e:
+        try:
+            platform_shell.kill_process_tree(proc, grace_s=1.0)
+        except Exception:
+            pass
+        check("C3 Linux 进程组用例", False, repr(e))
+else:
+    print("SKIP  C3 Linux 进程组用例（当前为 Windows，Windows 用例见 C1/C2）", flush=True)
+
+# ======================================================================
+# D. kill_process_tree 安全边界
+# ======================================================================
+_p = subprocess.Popen([PY, "-c", "pass"])
+_p.wait(timeout=5)
+try:
+    platform_shell.kill_process_tree(_p, grace_s=0.5)
+    check("D1 已退出进程：不抛异常", True)
+except Exception as e:
+    check("D1 已退出进程：不抛异常", False, repr(e))
+
+try:
+    platform_shell.kill_process_tree(None)
+    check("D2 None 输入：不抛异常", True)
+except Exception as e:
+    check("D2 None 输入：不抛异常", False, repr(e))
+
+# ======================================================================
+print(f"\n===== {sum(1 for _, ok in RESULTS if ok)}/{len(RESULTS)} passed =====", flush=True)
+sys.exit(0 if all(ok for _, ok in RESULTS) else 1)