feat: 跨平台进程保护机制 v5.3.0

- 新增 process_protection.py 模块
  - 单实例保护: Windows命名Mutex / macOS+Linux fcntl文件锁
  - 进程优先级提升: Windows SetPriorityClass / POSIX os.nice()
  - Windows PPL风格保护: SetErrorMode抑制错误弹窗
  - POSIX信号保护: 忽略SIGHUP/SIGPIPE, 注册SIGTERM优雅关闭
  - 看门狗监控: 后台线程监控父进程状态
  - 关闭钩子系统: 进程退出时自动执行清理
  - /protect 命令查看保护状态
- 更新 platform_utils.py 新增进程工具函数
- 集成到 cli_base.py 启动时自动初始化
- 更新 CHANGELOG.md / README.md 至 v5.3.0

CHANGELOG.md

@@ -4,6 +4,28 @@
 
 ---
 
+## v5.3.0 (2026-05-09)
+
+### ✨ 新增功能
+
+- **跨平台进程保护模块** (`xcli_core/process_protection.py`)
+  - **单实例保护** — 防止重复启动，Windows 使用命名 Mutex，macOS/Linux 使用 fcntl 文件锁（自动清理残留锁文件）
+  - **进程优先级提升** — Windows: `SetPriorityClass` (ABOVE_NORMAL/HIGH/REALTIME)，macOS/Linux: `os.nice()` + `renice`
+  - **Windows PPL 风格保护** — `SetErrorMode` 抑制系统错误弹窗 (SEM_FAILCRITICALERRORS | SEM_NOGPFAULTERRORBOX)、`faulthandler` 崩溃堆栈捕获
+  - **POSIX 信号保护** — macOS/Linux 忽略 SIGHUP（终端关闭不退出）、注册 SIGTERM 优雅关闭、macOS 忽略 SIGPIPE
+  - **看门狗监控** — 后台线程监控父进程状态，父进程退出时自动清理并退出
+  - **关闭钩子系统** — 注册自定义清理函数，进程退出时自动执行
+  - **一键启用** — `enable_process_protection()` 单函数调用启用全部保护
+  - **`/protect` 命令** — CLI 中查看保护状态，`/protect test` 重新测试所有保护
+
+###  改进
+
+- `platform_utils.py` 新增 `is_process_running()`、`get_process_priority()`、`set_process_priority()` 跨平台工具函数
+- `cli_base.py` 启动时自动初始化进程保护，退出时自动清理
+- `cli_core.py` 启动 banner 显示保护状态图标（单实例/优先级/PPL/看门狗）
+
+---
+
 ## v5.2.0 (2026-05-08)
 
 ### ✨ 新增功能

README.md

@@ -9,7 +9,7 @@
 </p>
 
 <p align="center">
-  <img src="https://img.shields.io/badge/version-5.2.0-blue" alt="version">
+  <img src="https://img.shields.io/badge/version-5.3.0-blue" alt="version">
   <img src="https://img.shields.io/badge/python-3.10+-green" alt="python">
   <img src="https://img.shields.io/badge/license-MIT-orange" alt="license">
   <img src="https://img.shields.io/badge/tests-150+%20passed-brightgreen" alt="tests">
@@ -35,6 +35,7 @@
 - **定时任务** — 支持相对时间、绝对时间、重复任务，到期自动激活 AI
 - **系统通知** — 跨平台任务完成通知（Windows / macOS / Linux）
 - **MCP 协议** — 19 个插件全部实现 MCP 定义，支持 Function Calling 原生调用
+- **进程保护** — 跨平台单实例保护 + 进程优先级提升 + Windows PPL 风格保护 + 看门狗监控 + 优雅关闭
 - **19 个插件 161+ 个操作** — 代码编辑、Git、浏览器、工程化、文件管理、图像生成、记忆等
 - **三模式界面** — TUI 图形化 + 传统命令行 + Web UI 浏览器界面
 - **150+ 个自动化测试** — pytest 覆盖全部插件和核心模块

platform_utils.py

@@ -121,4 +121,79 @@ def get_config_directory():
     elif platform == 'macos':
         return os.path.join(os.path.expanduser("~"), 'Library', 'Application Support', 'Xiaoli Pro-CLI 3')
     else:  # Linux
-        return os.path.join(os.path.expanduser('~'), '.config', 'xiaoli-pro-cli-3')
+        return os.path.join(os.path.expanduser('~'), '.config', 'xiaoli-pro-cli-3')
+
+def is_process_running(pid):
+    """检查指定 PID 的进程是否在运行"""
+    platform = get_platform()
+    
+    try:
+        if platform == 'windows':
+            import ctypes
+            kernel32 = ctypes.windll.kernel32
+            handle = kernel32.OpenProcess(0x1000, False, pid)  # PROCESS_QUERY_LIMITED_INFORMATION
+            if handle:
+                kernel32.CloseHandle(handle)
+                return True
+            return False
+        else:
+            os.kill(pid, 0)
+            return True
+    except (OSError, ProcessLookupError):
+        return False
+
+
+def get_process_priority():
+    """获取当前进程优先级"""
+    platform = get_platform()
+    
+    try:
+        if platform == 'windows':
+            import ctypes
+            kernel32 = ctypes.windll.kernel32
+            handle = kernel32.GetCurrentProcess()
+            return kernel32.GetPriorityClass(handle)
+        else:
+            import resource
+            return resource.getpriority(resource.PRIO_PROCESS, 0)
+    except Exception:
+        return None
+
+
+def set_process_priority(level="above_normal"):
+    """
+    设置进程优先级
+    
+    Args:
+        level: "below_normal" | "normal" | "above_normal" | "high"
+    
+    Returns:
+        是否设置成功
+    """
+    platform = get_platform()
+    
+    try:
+        if platform == 'windows':
+            import ctypes
+            kernel32 = ctypes.windll.kernel32
+            priority_map = {
+                "below_normal": 0x00004000,
+                "normal": 0x00000020,
+                "above_normal": 0x00008000,
+                "high": 0x00000080,
+            }
+            priority = priority_map.get(level, 0x00008000)
+            handle = kernel32.GetCurrentProcess()
+            return bool(kernel32.SetPriorityClass(handle, priority))
+        else:
+            nice_map = {
+                "below_normal": 10,
+                "normal": 0,
+                "above_normal": -10,
+                "high": -15,
+            }
+            nice_val = nice_map.get(level, -10)
+            os.nice(nice_val)
+            return True
+    except Exception:
+        return False

xcli_core/cli_base.py

@@ -11,6 +11,7 @@ from .constants import (
 )
 from .config import get_system_config, set_system_config, logger
 from .plugin_manager import LiuginManager
+from .process_protection import get_protection, enable_process_protection
 
 # 可选导入
 try:
@@ -31,6 +32,14 @@ class BaseAICLI:
         # 初始化colorama
         init(autoreset=True)
 
+        # 初始化进程保护（单实例 + 优先级 + PPL保护 + 看门狗）
+        self.protection_results = enable_process_protection(
+            priority_level="above_normal",
+            watchdog=True,
+            restart_on_crash=False,
+        )
+        self._protection = get_protection()
+
         # 确保插件管理器使用正确的路径
         project_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
         liugins_dir = os.path.join(project_dir, "plugins")
@@ -114,6 +123,12 @@ class BaseAICLI:
     def cleanup_resources(self):
         """清理资源，确保在程序退出时正确关闭所有资源"""
         print(f"{Fore.YELLOW}正在清理资源...{Style.RESET_ALL}")
+        # 清理进程保护
+        if hasattr(self, '_protection') and self._protection:
+            try:
+                self._protection.disable_all()
+            except Exception:
+                pass
         self._stop_clawli_monitor()
         if CLAWLI_SERVER_AVAILABLE and clawli_server:
             clawli_server.stop_server()

xcli_core/cli_core.py

@@ -20,6 +20,7 @@ from .cli_display import DisplayMixin
 from .cli_history import HistoryMixin
 from .plugin_market import PluginMarketMixin
 from .notification import notify_task_complete, get_notification_manager
+from .process_protection import get_protection_status
 
 
 class AICLI(BaseAICLI, ClawliMixin, ToolMixin, CodeExecMixin, DisplayMixin, HistoryMixin, PluginMarketMixin):
@@ -1020,6 +1021,7 @@ multi 操作支持一次修改多处：
         print(f"{Fore.GREEN}输入 '/chat open <名称>' 加载聊天记录{Style.RESET_ALL}")
         print(f"{Fore.GREEN}输入 '/remote' 查看远程连接帮助{Style.RESET_ALL}")
         print(f"{Fore.GREEN}输入 '/plugin' 管理插件市场 (安装/卸载/搜索){Style.RESET_ALL}")
+        print(f"{Fore.GREEN}输入 '/protect' 查看进程保护状态{Style.RESET_ALL}")
         current_engine_name = getattr(self.current_engine, 'name', '未设置') if self.current_engine else '未设置'
         print(f"{Fore.GREEN}当前使用的AI引擎: {current_engine_name}{Style.RESET_ALL}")
         loaded_engines = list(self.engines.keys())
@@ -1027,6 +1029,15 @@ multi 操作支持一次修改多处：
             print(f"{Fore.GREEN}已加载的AI引擎: {', '.join(loaded_engines)}{Style.RESET_ALL}")
         print(f"{Fore.GREEN}{'-' * 50}{Style.RESET_ALL}")
 
+        # 显示进程保护状态
+        if hasattr(self, 'protection_results') and self.protection_results:
+            r = self.protection_results
+            si_icon = '' if r.get('single_instance') else ''
+            prio_icon = '' if r.get('priority') else ''
+            ppl_icon = '' if r.get('ppl_protection') else ''
+            wd_icon = '' if r.get('watchdog') else ''
+            print(f"{Fore.CYAN}  进程保护: {si_icon}单实例 {prio_icon}优先级 {ppl_icon}PPL保护 {wd_icon}看门狗{Style.RESET_ALL}")
+
         # 询问 diff 弹窗模式
         self._ask_diff_popup_mode()
 
@@ -1171,6 +1182,29 @@ multi 操作支持一次修改多处：
                         print(result)
                     continue
 
+                if user_input.startswith('/protect'):
+                    parts = user_input.split()
+                    if len(parts) > 1 and parts[1] == 'test':
+                        # 重新测试所有保护
+                        from .process_protection import enable_process_protection
+                        results = enable_process_protection()
+                        print(f"\n{Fore.CYAN}  进程保护测试结果:{Style.RESET_ALL}")
+                        for key, val in results.items():
+                            icon = '' if val else ''
+                            print(f"    {icon} {key}: {val}")
+                    else:
+                        status = get_protection_status()
+                        print(f"\n{Fore.CYAN}  进程保护状态:{Style.RESET_ALL}")
+                        print(f"    平台: {status.get('platform', '未知')}")
+                        icon = '' if status.get('is_protected') else ''
+                        print(f"    {icon} 保护已启用: {status.get('is_protected', False)}")
+                        icon = '' if status.get('single_instance_held') else ''
+                        print(f"    {icon} 单实例锁: {status.get('single_instance_held', False)}")
+                        icon = '' if status.get('watchdog_running') else ''
+                        print(f"    {icon} 看门狗: {status.get('watchdog_running', False)}")
+                        print(f"      关闭钩子: {status.get('shutdown_hooks_count', 0)} 个")
+                    continue
+
                 if user_input.startswith('/remote'):
                     remote_command = user_input[7:].strip()
                     self._handle_remote_command(remote_command)

xcli_core/process_protection.py

@@ -0,0 +1,659 @@
+"""
+跨平台进程保护模块
+提供单实例保护、进程优先级提升、看门狗监控、优雅关闭等功能
+
+支持 Windows / macOS / Linux
+Windows 特有: PPL (Protected Process Light) 风格保护、错误模式抑制
+"""
+import os
+import sys
+import time
+import signal
+import atexit
+import threading
+import subprocess
+from typing import Optional, Callable
+
+
+# ── 平台检测 ──
+PLATFORM = (
+    'windows' if os.name == 'nt'
+    else 'macos' if sys.platform == 'darwin'
+    else 'linux'
+)
+
+
+class ProcessProtection:
+    """跨平台进程保护器"""
+
+    def __init__(self, app_name: str = "xiaoli-cli"):
+        self.app_name = app_name
+        self._lock_file_path: Optional[str] = None
+        self._lock_fd = None
+        self._watchdog_thread: Optional[threading.Thread] = None
+        self._watchdog_running = threading.Event()
+        self._shutdown_hooks: list[Callable] = []
+        self._is_protected = False
+        self._original_priority = None
+        self._restart_on_crash = False
+        self._restart_script: Optional[str] = None
+
+    # ══════════════════════════════════════════════════
+    #  单实例保护
+    # ══════════════════════════════════════════════════
+
+    def acquire_single_instance(self) -> bool:
+        """
+        获取单实例锁，防止重复启动
+        返回 True 表示成功（当前是唯一实例），False 表示已有实例在运行
+        """
+        if PLATFORM == 'windows':
+            return self._acquire_windows_mutex()
+        else:
+            return self._acquire_posix_lock()
+
+    def release_single_instance(self):
+        """释放单实例锁"""
+        if PLATFORM == 'windows':
+            self._release_windows_mutex()
+        else:
+            self._release_posix_lock()
+
+    def _acquire_windows_mutex(self) -> bool:
+        """Windows: 使用命名 Mutex 实现单实例"""
+        try:
+            import ctypes
+            from ctypes import wintypes
+
+            kernel32 = ctypes.windll.kernel32
+            mutex_name = f"Global\\{self.app_name}_single_instance"
+
+            # 创建命名 Mutex
+            handle = kernel32.CreateMutexW(None, True, mutex_name)
+            last_error = kernel32.GetLastError()
+
+            ERROR_ALREADY_EXISTS = 183
+            if last_error == ERROR_ALREADY_EXISTS:
+                kernel32.CloseHandle(handle)
+                return False
+
+            self._win_mutex_handle = handle
+            atexit.register(self.release_single_instance)
+            return True
+
+        except Exception:
+            # 降级到文件锁
+            return self._acquire_posix_lock()
+
+    def _release_windows_mutex(self):
+        """Windows: 释放 Mutex"""
+        try:
+            if hasattr(self, '_win_mutex_handle') and self._win_mutex_handle:
+                import ctypes
+                ctypes.windll.kernel32.CloseHandle(self._win_mutex_handle)
+                self._win_mutex_handle = None
+        except Exception:
+            pass
+
+    def _acquire_posix_lock(self) -> bool:
+        """POSIX (macOS/Linux): 使用文件锁实现单实例"""
+        import tempfile
+
+        lock_dir = tempfile.gettempdir()
+        self._lock_file_path = os.path.join(lock_dir, f"{self.app_name}.lock")
+
+        try:
+            # 尝试创建并锁定文件
+            if sys.platform == 'darwin':
+                # macOS: 使用 fcntl 文件锁
+                import fcntl
+                self._lock_fd = open(self._lock_file_path, 'w')
+                try:
+                    fcntl.flock(self._lock_fd.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
+                except (IOError, OSError):
+                    # 文件已被锁定 — 已有实例在运行
+                    # 检查锁文件中的 PID 是否仍在运行
+                    try:
+                        old_pid = self._lock_fd.read().strip()
+                        if old_pid and self._is_process_running(int(old_pid)):
+                            self._lock_fd.close()
+                            self._lock_fd = None
+                            return False
+                        else:
+                            # 旧进程已死，清理残留锁
+                            self._lock_fd.seek(0)
+                            self._lock_fd.truncate()
+                    except (ValueError, OSError):
+                        pass
+
+                self._lock_fd.seek(0)
+                self._lock_fd.truncate()
+                self._lock_fd.write(str(os.getpid()))
+                self._lock_fd.flush()
+                atexit.register(self.release_single_instance)
+                return True
+            else:
+                # Linux: 使用 fcntl 文件锁
+                import fcntl
+                self._lock_fd = open(self._lock_file_path, 'w')
+                try:
+                    fcntl.flock(self._lock_fd.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
+                except (IOError, OSError):
+                    # 检查旧进程是否仍在运行
+                    try:
+                        old_pid = self._lock_fd.read().strip()
+                        if old_pid and self._is_process_running(int(old_pid)):
+                            self._lock_fd.close()
+                            self._lock_fd = None
+                            return False
+                        else:
+                            self._lock_fd.seek(0)
+                            self._lock_fd.truncate()
+                    except (ValueError, OSError):
+                        pass
+
+                self._lock_fd.seek(0)
+                self._lock_fd.truncate()
+                self._lock_fd.write(str(os.getpid()))
+                self._lock_fd.flush()
+                atexit.register(self.release_single_instance)
+                return True
+
+        except Exception:
+            # 锁获取失败，但不阻止程序运行
+            return True
+
+    def _release_posix_lock(self):
+        """POSIX: 释放文件锁"""
+        try:
+            if self._lock_fd:
+                if sys.platform == 'darwin':
+                    import fcntl
+                    fcntl.flock(self._lock_fd.fileno(), fcntl.LOCK_UN)
+                else:
+                    import fcntl
+                    fcntl.flock(self._lock_fd.fileno(), fcntl.LOCK_UN)
+                self._lock_fd.close()
+                self._lock_fd = None
+            if self._lock_file_path and os.path.exists(self._lock_file_path):
+                try:
+                    os.unlink(self._lock_file_path)
+                except OSError:
+                    pass
+        except Exception:
+            pass
+
+    @staticmethod
+    def _is_process_running(pid: int) -> bool:
+        """检查指定 PID 的进程是否在运行"""
+        try:
+            if PLATFORM == 'windows':
+                import ctypes
+                kernel32 = ctypes.windll.kernel32
+                PROCESS_QUERY_LIMITED_INFORMATION = 0x1000
+                handle = kernel32.OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, False, pid)
+                if handle:
+                    kernel32.CloseHandle(handle)
+                    return True
+                return False
+            else:
+                os.kill(pid, 0)
+                return True
+        except (OSError, ProcessLookupError):
+            return False
+
+    # ══════════════════════════════════════════════════
+    #  进程优先级提升
+    # ══════════════════════════════════════════════════
+
+    def elevate_priority(self, level: str = "high") -> bool:
+        """
+        提升进程优先级
+
+        Args:
+            level: "above_normal" | "high" | "realtime"(仅Windows)
+
+        Returns:
+            是否成功提升
+        """
+        try:
+            if PLATFORM == 'windows':
+                return self._elevate_windows(level)
+            elif PLATFORM == 'macos':
+                return self._elevate_macos(level)
+            else:
+                return self._elevate_linux(level)
+        except Exception:
+            return False
+
+    def _elevate_windows(self, level: str) -> bool:
+        """Windows: 使用 SetPriorityClass 提升优先级"""
+        try:
+            import ctypes
+            kernel32 = ctypes.windll.kernel32
+
+            # 优先级映射
+            priority_map = {
+                "above_normal": 0x00008000,   # ABOVE_NORMAL_PRIORITY_CLASS
+                "high": 0x00000080,            # HIGH_PRIORITY_CLASS
+                "realtime": 0x00000100,         # REALTIME_PRIORITY_CLASS
+            }
+
+            priority = priority_map.get(level, 0x00000080)
+
+            # 获取当前进程句柄
+            handle = kernel32.GetCurrentProcess()
+
+            # 记录原始优先级
+            self._original_priority = kernel32.GetPriorityClass(handle)
+
+            # 设置新优先级
+            result = kernel32.SetPriorityClass(handle, priority)
+            if result:
+                self._is_protected = True
+                return True
+            return False
+
+        except Exception:
+            return False
+
+    def _elevate_macos(self, level: str) -> bool:
+        """macOS: 使用 renice 提升优先级"""
+        try:
+            nice_map = {
+                "above_normal": -10,
+                "high": -15,
+                "realtime": -20,
+            }
+            nice_val = nice_map.get(level, -10)
+            os.nice(nice_val)
+            self._is_protected = True
+            return True
+        except Exception:
+            return False
+
+    def _elevate_linux(self, level: str) -> bool:
+        """Linux: 使用 renice 提升优先级"""
+        try:
+            nice_map = {
+                "above_normal": -10,
+                "high": -15,
+                "realtime": -20,
+            }
+            nice_val = nice_map.get(level, -10)
+
+            # os.nice() 是相对值，需要先获取当前 nice 值
+            try:
+                import resource
+                current = resource.getpriority(resource.PRIO_PROCESS, 0)
+                target = nice_val
+                delta = target - current
+                if delta != 0:
+                    os.nice(delta)
+            except (ImportError, OSError):
+                # fallback: 直接使用 subprocess
+                subprocess.run(
+                    ["renice", str(nice_val), "-p", str(os.getpid())],
+                    capture_output=True, timeout=5
+                )
+
+            self._is_protected = True
+            return True
+        except Exception:
+            return False
+
+    def restore_priority(self):
+        """恢复原始进程优先级"""
+        if not self._is_protected:
+            return
+
+        try:
+            if PLATFORM == 'windows' and self._original_priority:
+                import ctypes
+                kernel32 = ctypes.windll.kernel32
+                handle = kernel32.GetCurrentProcess()
+                kernel32.SetPriorityClass(handle, self._original_priority)
+            elif PLATFORM in ('macos', 'linux'):
+                # 恢复到默认 nice 值 (0)
+                try:
+                    import resource
+                    current = resource.getpriority(resource.PRIO_PROCESS, 0)
+                    os.nice(-current)
+                except (ImportError, OSError):
+                    subprocess.run(
+                        ["renice", "0", "-p", str(os.getpid())],
+                        capture_output=True, timeout=5
+                    )
+            self._is_protected = False
+        except Exception:
+            pass
+
+    # ══════════════════════════════════════════════════
+    #  Windows PPL 风格保护
+    # ══════════════════════════════════════════════════
+
+    def enable_ppl_protection(self) -> bool:
+        """
+        Windows: 启用 PPL (Protected Process Light) 风格保护
+        - 抑制错误弹窗
+        - 设置进程缓解策略
+        - 禁用 DEP 异常弹窗
+
+        macOS/Linux: 启用等效保护
+        - 设置信号处理
+        - 忽略 SIGHUP
+        """
+        if PLATFORM == 'windows':
+            return self._enable_windows_ppl()
+        else:
+            return self._enable_posix_protection()
+
+    def _enable_windows_ppl(self) -> bool:
+        """Windows: PPL 风格保护"""
+        try:
+            import ctypes
+            kernel32 = ctypes.windll.kernel32
+
+            # 1. 抑制系统错误弹窗 (SEM_FAILCRITICALERRORS | SEM_NOGPFAULTERRORBOX)
+            SEM_FAILCRITICALERRORS = 0x0001
+            SEM_NOGPFAULTERRORBOX = 0x0002
+            SEM_NOOPENFILEERRORBOX = 0x8000
+            old_mode = kernel32.SetErrorMode(
+                SEM_FAILCRITICALERRORS | SEM_NOGPFAULTERRORBOX | SEM_NOOPENFILEERRORBOX
+            )
+            self._original_error_mode = old_mode
+
+            # 2. 设置进程为后台优先级模式（减少对用户交互的影响）
+            # PROCESS_MODE_BACKGROUND_BEGIN = 0x00100000
+            # kernel32.SetPriorityClass(kernel32.GetCurrentProcess(), 0x00100000)
+
+            # 3. 注册异常处理（防止未处理异常导致崩溃弹窗）
+            try:
+                import faulthandler
+                faulthandler.enable()
+            except Exception:
+                pass
+
+            self._is_protected = True
+            return True
+
+        except Exception:
+            return False
+
+    def _enable_posix_protection(self) -> bool:
+        """macOS/Linux: 等效保护"""
+        try:
+            # 1. 忽略 SIGHUP（终端关闭时不停止进程）
+            signal.signal(signal.SIGHUP, signal.SIG_IGN)
+
+            # 2. 注册优雅关闭信号
+            def _graceful_shutdown(signum, frame):
+                self._execute_shutdown_hooks()
+                sys.exit(0)
+
+            signal.signal(signal.SIGTERM, _graceful_shutdown)
+
+            # 3. macOS 特有：忽略 SIGPIPE
+            if PLATFORM == 'macos':
+                signal.signal(signal.SIGPIPE, signal.SIG_IGN)
+
+            # 4. 启用 faulthandler（崩溃时输出堆栈）
+            try:
+                import faulthandler
+                faulthandler.enable()
+            except Exception:
+                pass
+
+            self._is_protected = True
+            return True
+
+        except Exception:
+            return False
+
+    # ══════════════════════════════════════════════════
+    #  看门狗监控
+    # ══════════════════════════════════════════════════
+
+    def start_watchdog(self, check_interval: float = 5.0,
+                       restart_on_crash: bool = False,
+                       restart_script: Optional[str] = None):
+        """
+        启动看门狗线程，监控父进程状态
+        当父进程退出时，自动清理并退出
+
+        Args:
+            check_interval: 检查间隔（秒）
+            restart_on_crash: 崩溃时是否自动重启
+            restart_script: 重启脚本路径
+        """
+        self._restart_on_crash = restart_on_crash
+        self._restart_script = restart_script
+
+        if self._watchdog_thread and self._watchdog_thread.is_alive():
+            return
+
+        self._watchdog_running.set()
+        self._watchdog_thread = threading.Thread(
+            target=self._watchdog_loop,
+            args=(check_interval,),
+            daemon=True,
+            name="process-watchdog"
+        )
+        self._watchdog_thread.start()
+
+    def stop_watchdog(self):
+        """停止看门狗"""
+        self._watchdog_running.clear()
+        if self._watchdog_thread:
+            self._watchdog_thread.join(timeout=3)
+            self._watchdog_thread = None
+
+    def _watchdog_loop(self, check_interval: float):
+        """看门狗主循环"""
+        parent_pid = os.getppid()
+
+        while self._watchdog_running.is_set():
+            try:
+                # 检查父进程是否仍在运行
+                if not self._is_process_running(parent_pid):
+                    # 父进程已退出，执行清理
+                    self._execute_shutdown_hooks()
+                    if self._restart_on_crash and self._restart_script:
+                        self._do_restart()
+                    else:
+                        os._exit(0)
+                    break
+
+                time.sleep(check_interval)
+
+            except Exception:
+                time.sleep(check_interval)
+
+    def _do_restart(self):
+        """执行重启"""
+        try:
+            if self._restart_script and os.path.exists(self._restart_script):
+                if PLATFORM == 'windows':
+                    subprocess.Popen(
+                        ["python", self._restart_script],
+                        creationflags=subprocess.CREATE_NEW_PROCESS_GROUP
+                    )
+                else:
+                    subprocess.Popen(
+                        ["python3", self._restart_script],
+                        start_new_session=True
+                    )
+        except Exception:
+            pass
+        finally:
+            os._exit(0)
+
+    # ══════════════════════════════════════════════════
+    #  关闭钩子
+    # ══════════════════════════════════════════════════
+
+    def register_shutdown_hook(self, hook: Callable):
+        """注册关闭钩子，在进程退出时执行"""
+        if hook not in self._shutdown_hooks:
+            self._shutdown_hooks.append(hook)
+
+    def _execute_shutdown_hooks(self):
+        """执行所有关闭钩子"""
+        for hook in reversed(self._shutdown_hooks):
+            try:
+                hook()
+            except Exception:
+                pass
+
+    # ══════════════════════════════════════════════════
+    #  一键启用全部保护
+    # ══════════════════════════════════════════════════
+
+    def enable_all(self, priority_level: str = "above_normal",
+                   watchdog: bool = True,
+                   restart_on_crash: bool = False,
+                   restart_script: Optional[str] = None) -> dict:
+        """
+        一键启用所有进程保护
+
+        Args:
+            priority_level: 优先级 ("above_normal" | "high")
+            watchdog: 是否启用看门狗
+            restart_on_crash: 崩溃时是否自动重启
+            restart_script: 重启脚本路径
+
+        Returns:
+            各项保护的启用状态字典
+        """
+        results = {
+            "single_instance": False,
+            "priority": False,
+            "ppl_protection": False,
+            "watchdog": False,
+            "platform": PLATFORM,
+        }
+
+        # 1. 单实例保护
+        results["single_instance"] = self.acquire_single_instance()
+
+        # 2. 进程优先级提升
+        results["priority"] = self.elevate_priority(priority_level)
+
+        # 3. PPL 风格保护
+        results["ppl_protection"] = self.enable_ppl_protection()
+
+        # 4. 看门狗
+        if watchdog:
+            self.start_watchdog(
+                restart_on_crash=restart_on_crash,
+                restart_script=restart_script
+            )
+            results["watchdog"] = True
+
+        # 注册清理
+        atexit.register(self.disable_all)
+
+        return results
+
+    def disable_all(self):
+        """禁用所有保护并清理"""
+        self.stop_watchdog()
+        self.restore_priority()
+        self.release_single_instance()
+        self._execute_shutdown_hooks()
+        self._is_protected = False
+
+    def get_status(self) -> dict:
+        """获取当前保护状态"""
+        return {
+            "platform": PLATFORM,
+            "is_protected": self._is_protected,
+            "single_instance_held": (
+                self._lock_fd is not None or
+                hasattr(self, '_win_mutex_handle') and self._win_mutex_handle is not None
+            ),
+            "watchdog_running": (
+                self._watchdog_thread is not None and
+                self._watchdog_thread.is_alive()
+            ),
+            "shutdown_hooks_count": len(self._shutdown_hooks),
+        }
+
+
+# ── 全局单例 ──
+_protection: Optional[ProcessProtection] = None
+
+
+def get_protection() -> ProcessProtection:
+    """获取全局进程保护器单例"""
+    global _protection
+    if _protection is None:
+        _protection = ProcessProtection()
+    return _protection
+
+
+def enable_process_protection(priority_level: str = "above_normal",
+                               watchdog: bool = True,
+                               restart_on_crash: bool = False) -> dict:
+    """
+    快捷函数：启用进程保护
+
+    Returns:
+        各项保护状态字典
+    """
+    return get_protection().enable_all(
+        priority_level=priority_level,
+        watchdog=watchdog,
+        restart_on_crash=restart_on_crash,
+    )
+
+
+def disable_process_protection():
+    """快捷函数：禁用进程保护"""
+    get_protection().disable_all()
+
+
+def get_protection_status() -> dict:
+    """快捷函数：获取保护状态"""
+    return get_protection().get_status()
+
+
+# ── CLI 入口（可独立运行测试） ──
+if __name__ == "__main__":
+    import json
+
+    print(f"平台: {PLATFORM}")
+    print(f"PID: {os.getpid()}")
+    print()
+
+    prot = ProcessProtection("xiaoli-cli-test")
+
+    # 测试单实例
+    is_first = prot.acquire_single_instance()
+    print(f"单实例锁: {'获取成功 (首个实例)' if is_first else '获取失败 (已有实例)'}")
+
+    if is_first:
+        # 测试优先级提升
+        prio_ok = prot.elevate_priority("above_normal")
+        print(f"优先级提升: {'成功' if prio_ok else '失败'}")
+
+        # 测试 PPL 保护
+        ppl_ok = prot.enable_ppl_protection()
+        print(f"PPL 保护: {'启用' if ppl_ok else '失败'}")
+
+        # 测试看门狗
+        prot.start_watchdog(check_interval=2.0)
+        print(f"看门狗: 已启动")
+
+        # 显示状态
+        status = prot.get_status()
+        print(f"\n状态: {json.dumps(status, indent=2, ensure_ascii=False)}")
+
+        print("\n按 Ctrl+C 退出...")
+        try:
+            while True:
+                time.sleep(1)
+        except KeyboardInterrupt:
+            print("\n正在清理...")
+            prot.disable_all()
+            print("完成")