diff --git a/deploy.py b/deploy.py
index 20c582a..9445553 100644
--- a/deploy.py
+++ b/deploy.py
@@ -18,6 +18,7 @@
 from tkinter import filedialog, messagebox
 
 from deployer.config import DeployerConfig
+from deployer.install_timing import InstallTiming
 from deployer.logger import DeployerLogger
 from deployer.webview_bridge import InstallationCancelled, run_web_installer
 from deployer.windows_setup import (
@@ -888,6 +889,7 @@ def _append_log_line(self, line: str):
 
     def _install_thread(self):
         log = self.logger
+        timing = InstallTiming(log)
         ws = WindowsSetup(self.config, log)
 
         steps = [
@@ -923,31 +925,41 @@ def _install_thread(self):
                 message = f"Installation cancelled.{suffix}"
                 self.after(0, lambda text=message: self._finish_fail(text))
                 self._running = False
+                timing.finish("cancelled")
                 return
 
             self._set_progress(pct, label)
+            started_at = timing.start_step()
             try:
                 result = fn()
                 if result is not None and not result:
+                    timing.record_step(label, started_at, "failed")
                     rollback_ok = ws.rollback_openclaw_upgrade()
                     suffix = "" if rollback_ok else " Automatic rollback also failed."
                     self._finish_fail(label.rstrip(".") + f" failed.{suffix}")
                     self._running = False
+                    timing.finish("failed")
                     return
             except InstallationCancelled:
+                timing.record_step(label, started_at, "cancelled")
                 self._running = False
                 self.after(0, self._return_to_welcome)
+                timing.finish("cancelled")
                 return
             except Exception as e:
+                timing.record_step(label, started_at, "failed")
                 log.error(f"{label} exception: {e}")
                 rollback_ok = ws.rollback_openclaw_upgrade()
                 suffix = "" if rollback_ok else " Automatic rollback also failed."
                 detail = str(e).strip() or label.rstrip(".") + " failed."
                 self._finish_fail(f"{detail}{suffix}")
                 self._running = False
+                timing.finish("failed")
                 return
+            timing.record_step(label, started_at, "success")
 
         self._running = False
+        timing.finish("success")
         self._finish_ok()
 
     def _prepare_upgrade(self, ws: WindowsSetup) -> bool:
diff --git a/deployer/install_timing.py b/deployer/install_timing.py
new file mode 100644
index 0000000..309075d
--- /dev/null
+++ b/deployer/install_timing.py
@@ -0,0 +1,47 @@
+"""Installation timing metrics shared by both installer frontends."""
+
+from collections.abc import Callable
+from dataclasses import dataclass
+from time import perf_counter
+
+
+@dataclass(frozen=True)
+class StepTiming:
+    label: str
+    status: str
+    duration_ms: int
+
+
+class InstallTiming:
+    """Record readable per-step timings and a compact installation summary."""
+
+    def __init__(self, logger, clock: Callable[[], float] | None = None):
+        self._logger = logger
+        self._clock = clock or perf_counter
+        self._started_at = self._clock()
+        self._steps: list[StepTiming] = []
+        self._finished = False
+
+    def start_step(self) -> float:
+        return self._clock()
+
+    def record_step(self, label: str, started_at: float, status: str) -> None:
+        clean_label = label.rstrip(".")
+        duration_ms = max(0, round((self._clock() - started_at) * 1000))
+        self._steps.append(StepTiming(label=clean_label, status=status, duration_ms=duration_ms))
+        self._logger.info(f"Install timing: {clean_label} [{status}] {duration_ms / 1000:.2f}s")
+
+    def finish(self, status: str) -> None:
+        if self._finished:
+            return
+        self._finished = True
+
+        total_ms = max(0, round((self._clock() - self._started_at) * 1000))
+        self._logger.info(
+            f"Install timing summary: status={status} "
+            f"total={total_ms / 1000:.2f}s steps={len(self._steps)}"
+        )
+        if self._steps:
+            slowest = sorted(self._steps, key=lambda step: step.duration_ms, reverse=True)[:5]
+            details = "; ".join(f"{step.label}={step.duration_ms / 1000:.2f}s" for step in slowest)
+            self._logger.info(f"Slowest install steps: {details}")
diff --git a/deployer/openclaw_upgrade.py b/deployer/openclaw_upgrade.py
index dcf1897..e9ad827 100644
--- a/deployer/openclaw_upgrade.py
+++ b/deployer/openclaw_upgrade.py
@@ -28,6 +28,28 @@ class UpgradePhase(StrEnum):
     ROLLBACK_FAILED = "rollback-failed"
 
 
+class UpgradeBackupMode(StrEnum):
+    FULL = "full"
+    MANAGED_STATE = "managed-state"
+
+
+_MANAGED_STATE_PATHS = (
+    Path(".env"),
+    Path("openclaw.json"),
+    Path("skill_catalog.json"),
+    Path("managed_skill_catalog.json"),
+    Path("skills"),
+    Path("skills_snapshot.json"),
+    Path("skills_snapshot.sig"),
+    Path("skills_signing_key.pub"),
+    Path("skills_signing_key.pem"),
+    Path("MicroClawInstaller.exe"),
+    Path("_internal"),
+    Path("microclaw.ico"),
+    Path("microclaw-uninstall.ico"),
+)
+
+
 ACTIVE_PHASES = {
     UpgradePhase.BACKING_UP,
     UpgradePhase.INSTALLING,
@@ -75,6 +97,8 @@ class UpgradeManifest:
     created_at: str
     updated_at: str
     validation_results: dict[str, bool] = field(default_factory=dict)
+    backup_mode: UpgradeBackupMode = UpgradeBackupMode.FULL
+    config_existed: bool = False
 
 
 def _now() -> str:
@@ -557,6 +581,7 @@ def create(
         state_dir: Path,
         target_version: str,
         installation: OpenClawInstallation,
+        backup_mode: UpgradeBackupMode = UpgradeBackupMode.FULL,
     ) -> OpenClawUpgradeTransaction:
         root = microclaw_root.resolve(strict=False)
         transaction_id = datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ") + "-" + uuid.uuid4().hex[:8]
@@ -584,6 +609,8 @@ def create(
             phase=UpgradePhase.BACKING_UP,
             created_at=timestamp,
             updated_at=timestamp,
+            backup_mode=backup_mode,
+            config_existed=(state_dir / "openclaw.json").exists(),
         )
         try:
             transaction = cls(
@@ -605,6 +632,8 @@ def _read_manifest(cls, manifest_path: Path) -> UpgradeManifest:
             if not isinstance(data, dict):
                 raise ValueError("manifest root must be an object")
             data["phase"] = UpgradePhase(data["phase"])
+            if "backup_mode" in data:
+                data["backup_mode"] = UpgradeBackupMode(data["backup_mode"])
             manifest = UpgradeManifest(**data)
         except (KeyError, TypeError, json.JSONDecodeError) as error:
             raise ValueError("invalid OpenClaw upgrade manifest") from error
@@ -663,6 +692,11 @@ def close(self) -> bool:
     def _validate_manifest(self) -> None:
         if self.manifest.schema_version != 1:
             raise ValueError("unsupported OpenClaw upgrade manifest schema")
+        if self.manifest.backup_mode not in {
+            UpgradeBackupMode.FULL,
+            UpgradeBackupMode.MANAGED_STATE,
+        }:
+            raise ValueError("unsupported OpenClaw upgrade backup mode")
         transaction_id = self.manifest.transaction_id
         if not isinstance(transaction_id, str) or not _TRANSACTION_ID_PATTERN.fullmatch(
             transaction_id
@@ -769,8 +803,55 @@ def report(done: int, total: int) -> None:
 
         return report
 
+    def _publish_backup(self, staging: Path) -> None:
+        metadata_files = {"transaction.json", "backup-files.json"}
+        io_staging = _native_io_path(staging)
+        inventory = {}
+        for path in io_staging.rglob("*"):
+            relative = path.relative_to(io_staging).as_posix()
+            if path.is_file() and relative not in metadata_files:
+                inventory[relative] = path.stat().st_size
+        _atomic_json_write(staging / "backup-files.json", inventory)
+        _atomic_json_write(staging / "transaction.json", self._payload())
+        _fsync_directory(staging)
+        _durable_rename(staging, self.backup_dir)
+        self.set_phase(UpgradePhase.INSTALLING)
+
+    def _backup_managed_state(self) -> None:
+        state_dir = Path(self.manifest.state_dir)
+        if self.manifest.state_existed and not state_dir.exists():
+            raise FileNotFoundError(f"OpenClaw state disappeared before backup: {state_dir}")
+        config_path = state_dir / "openclaw.json"
+        if self.manifest.config_existed and not config_path.exists():
+            raise FileNotFoundError(f"OpenClaw config disappeared before backup: {config_path}")
+        if not self.manifest.config_existed and config_path.exists():
+            raise RuntimeError(f"OpenClaw config appeared before backup: {config_path}")
+
+        _durable_mkdir(self.backup_root)
+        staging = self.backup_dir.with_name(
+            f".{self.manifest.transaction_id}.{uuid.uuid4().hex}.staging"
+        )
+        _durable_mkdir(staging / "state")
+        for relative in _MANAGED_STATE_PATHS:
+            source = state_dir / relative
+            destination = staging / "state" / relative
+            if source.is_dir() and not source.is_symlink():
+                shutil.copytree(_native_io_path(source), _native_io_path(destination))
+            elif source.exists() or source.is_symlink():
+                _durable_mkdir(destination.parent)
+                shutil.copy2(_native_io_path(source), _native_io_path(destination))
+        _fsync_payload_tree(
+            staging / "state",
+            self._payload_progress("Backing up MicroClaw-managed state"),
+        )
+        self._publish_backup(staging)
+
     def backup(self) -> None:
         self.set_phase(UpgradePhase.BACKING_UP)
+        if self.manifest.backup_mode == UpgradeBackupMode.MANAGED_STATE:
+            self._backup_managed_state()
+            return
+
         package_dir = Path(self.manifest.package_dir)
         if self.manifest.package_existed and not package_dir.exists():
             raise FileNotFoundError(f"OpenClaw package disappeared before backup: {package_dir}")
@@ -816,18 +897,7 @@ def backup(self) -> None:
         ):
             _fsync_payload_tree(payload_root, self._payload_progress("Backing up OpenClaw files"))
 
-        metadata_files = {"transaction.json", "backup-files.json"}
-        io_staging = _native_io_path(staging)
-        inventory = {}
-        for path in io_staging.rglob("*"):
-            relative = path.relative_to(io_staging).as_posix()
-            if path.is_file() and relative not in metadata_files:
-                inventory[relative] = path.stat().st_size
-        _atomic_json_write(staging / "backup-files.json", inventory)
-        _atomic_json_write(staging / "transaction.json", self._payload())
-        _fsync_directory(staging)
-        _durable_rename(staging, self.backup_dir)
-        self.set_phase(UpgradePhase.INSTALLING)
+        self._publish_backup(staging)
 
     def mark_verifying(self) -> None:
         self.set_phase(UpgradePhase.VERIFYING)
@@ -927,6 +997,26 @@ def _restore_state(self, failed_dir: Path) -> None:
             self.manifest.state_existed,
         )
 
+    def _restore_managed_state(self, failed_dir: Path) -> None:
+        state_dir = Path(self.manifest.state_dir)
+        for relative in _MANAGED_STATE_PATHS:
+            live = state_dir / relative
+            backup = self.backup_dir / "state" / relative
+            failed = failed_dir / "state" / relative
+            if backup.is_dir() and not backup.is_symlink():
+                self._restore_tree(backup, live, failed, True)
+                continue
+
+            staging = None
+            if backup.exists() or backup.is_symlink():
+                _durable_mkdir(live.parent)
+                staging = live.with_name(f".{live.name}.{uuid.uuid4().hex}.restore")
+                shutil.copy2(_native_io_path(backup), _native_io_path(staging))
+                _fsync_file(staging)
+            self._move_to_failed(live, failed)
+            if staging is not None:
+                _durable_replace(staging, live)
+
     def rollback(self) -> None:
         original_phase = self.manifest.phase
         if original_phase == UpgradePhase.ROLLED_BACK:
@@ -946,9 +1036,12 @@ def rollback(self) -> None:
             if original_phase != UpgradePhase.BACKING_UP:
                 failed_dir = self.backup_dir / "failed"
                 _durable_mkdir(failed_dir)
-                self._restore_package(failed_dir)
-                self._restore_shims()
-                self._restore_state(failed_dir)
+                if self.manifest.backup_mode == UpgradeBackupMode.MANAGED_STATE:
+                    self._restore_managed_state(failed_dir)
+                else:
+                    self._restore_package(failed_dir)
+                    self._restore_shims()
+                    self._restore_state(failed_dir)
         except Exception as error:
             try:
                 self.mark_rollback_failed()
diff --git a/deployer/webview_bridge.py b/deployer/webview_bridge.py
index ae5cf8d..a5f55a7 100644
--- a/deployer/webview_bridge.py
+++ b/deployer/webview_bridge.py
@@ -12,6 +12,7 @@
 from tkinter import filedialog
 
 from deployer.config import DeployerConfig
+from deployer.install_timing import InstallTiming
 from deployer.logger import DeployerLogger
 from deployer.windows_setup import (
     DEFAULT_DESKTOP_DIR,
@@ -362,6 +363,7 @@ def _set_progress_detail(self, detail):
             self._state["progress_detail"] = detail or ""
 
     def _install_thread(self):
+        timing = InstallTiming(self._logger)
         ws = WindowsSetup(self._config, self._logger)
         ws.progress_callback = self._set_progress_detail
 
@@ -402,16 +404,24 @@ def _install_thread(self):
                 rollback_ok = ws.rollback_openclaw_upgrade()
                 suffix = "" if rollback_ok else " Automatic rollback also failed."
                 self._finish_fail(f"Installation cancelled.{suffix}")
+                timing.finish("cancelled")
                 return
             self._set_progress(pct, label)
-            if not self._run_step_with_retry(pct, label, fn, retries):
-                if self.get_state()["status"] == "cancelled":
+            started_at = timing.start_step()
+            step_ok = self._run_step_with_retry(pct, label, fn, retries)
+            step_status = "success" if step_ok else self.get_state()["status"]
+            timing.record_step(label, started_at, step_status)
+            if not step_ok:
+                if step_status == "cancelled":
+                    timing.finish("cancelled")
                     return
                 if not ws.rollback_openclaw_upgrade():
                     with self._state_lock:
                         self._state["error"] += " Automatic rollback also failed."
+                timing.finish("failed")
                 return
 
+        timing.finish("success")
         self._finish_ok()
 
     def _run_step_with_retry(self, pct, label, fn, retries):
@@ -489,6 +499,8 @@ def _confirm_close_running_apps(self, active: ActiveInstallation) -> bool:
         process = f" (PID {', '.join(map(str, active.pids))})" if active.pids else ""
         strings = _STRINGS[self._lang]
         message = strings["runningAppsMessage"].format(process=process)
+        _ACTIVE_WINDOW.restore()
+        _ACTIVE_WINDOW.show()
         return bool(_ACTIVE_WINDOW.create_confirmation_dialog(strings["runningAppsTitle"], message))
 
     def _ensure_node(self, ws):
diff --git a/deployer/windows_setup.py b/deployer/windows_setup.py
index 4e6db10..22f9761 100644
--- a/deployer/windows_setup.py
+++ b/deployer/windows_setup.py
@@ -27,6 +27,7 @@
     RECOVERABLE_PHASES,
     OpenClawInstallation,
     OpenClawUpgradeTransaction,
+    UpgradeBackupMode,
     UpgradeInProgressError,
     UpgradePhase,
     process_is_alive,
@@ -140,6 +141,7 @@
 
 # Hide console windows spawned by subprocess on Windows
 _CREATE_NO_WINDOW = 0x08000000
+_CREATE_SUSPENDED = 0x00000004
 
 # Per-call timeout (seconds) for `openclaw gateway call ...` RPC probes run
 # during post-install validation. Each probe cold-starts the OpenClaw CLI, and
@@ -149,6 +151,168 @@
 _OPENCLAW_RPC_TIMEOUT = 120
 
 
+class _WindowsKillOnCloseJob:
+    """Own a Windows Job Object that terminates its process tree when closed."""
+
+    def __init__(self, handle: object | None):
+        self._handle = handle
+
+    @classmethod
+    def attach(cls, process: subprocess.Popen) -> "_WindowsKillOnCloseJob":
+        if os.name != "nt":
+            return cls(None)
+
+        import ctypes
+        from ctypes import wintypes
+
+        class _IoCounters(ctypes.Structure):
+            _fields_ = [
+                ("ReadOperationCount", ctypes.c_uint64),
+                ("WriteOperationCount", ctypes.c_uint64),
+                ("OtherOperationCount", ctypes.c_uint64),
+                ("ReadTransferCount", ctypes.c_uint64),
+                ("WriteTransferCount", ctypes.c_uint64),
+                ("OtherTransferCount", ctypes.c_uint64),
+            ]
+
+        class _BasicLimitInformation(ctypes.Structure):
+            _fields_ = [
+                ("PerProcessUserTimeLimit", ctypes.c_int64),
+                ("PerJobUserTimeLimit", ctypes.c_int64),
+                ("LimitFlags", wintypes.DWORD),
+                ("MinimumWorkingSetSize", ctypes.c_size_t),
+                ("MaximumWorkingSetSize", ctypes.c_size_t),
+                ("ActiveProcessLimit", wintypes.DWORD),
+                ("Affinity", ctypes.c_size_t),
+                ("PriorityClass", wintypes.DWORD),
+                ("SchedulingClass", wintypes.DWORD),
+            ]
+
+        class _ExtendedLimitInformation(ctypes.Structure):
+            _fields_ = [
+                ("BasicLimitInformation", _BasicLimitInformation),
+                ("IoInfo", _IoCounters),
+                ("ProcessMemoryLimit", ctypes.c_size_t),
+                ("JobMemoryLimit", ctypes.c_size_t),
+                ("PeakProcessMemoryUsed", ctypes.c_size_t),
+                ("PeakJobMemoryUsed", ctypes.c_size_t),
+            ]
+
+        kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)
+        kernel32.CreateJobObjectW.argtypes = [ctypes.c_void_p, wintypes.LPCWSTR]
+        kernel32.CreateJobObjectW.restype = wintypes.HANDLE
+        kernel32.SetInformationJobObject.argtypes = [
+            wintypes.HANDLE,
+            ctypes.c_int,
+            ctypes.c_void_p,
+            wintypes.DWORD,
+        ]
+        kernel32.SetInformationJobObject.restype = wintypes.BOOL
+        kernel32.AssignProcessToJobObject.argtypes = [wintypes.HANDLE, wintypes.HANDLE]
+        kernel32.AssignProcessToJobObject.restype = wintypes.BOOL
+        kernel32.CloseHandle.argtypes = [wintypes.HANDLE]
+        kernel32.CloseHandle.restype = wintypes.BOOL
+
+        handle = kernel32.CreateJobObjectW(None, None)
+        if not handle:
+            error = ctypes.get_last_error()
+            raise OSError(error, ctypes.FormatError(error))
+
+        information = _ExtendedLimitInformation()
+        information.BasicLimitInformation.LimitFlags = 0x00002000
+        if not kernel32.SetInformationJobObject(
+            handle,
+            9,
+            ctypes.byref(information),
+            ctypes.sizeof(information),
+        ) or not kernel32.AssignProcessToJobObject(handle, wintypes.HANDLE(int(process._handle))):
+            error = ctypes.get_last_error()
+            kernel32.CloseHandle(handle)
+            raise OSError(error, ctypes.FormatError(error))
+        return cls(handle)
+
+    def close(self) -> None:
+        handle = self._handle
+        if handle is None:
+            return
+        import ctypes
+        from ctypes import wintypes
+
+        kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)
+        kernel32.CloseHandle.argtypes = [wintypes.HANDLE]
+        kernel32.CloseHandle.restype = wintypes.BOOL
+        kernel32.CloseHandle(handle)
+        self._handle = None
+
+    @staticmethod
+    def resume(process: subprocess.Popen) -> None:
+        if os.name != "nt":
+            return
+
+        import ctypes
+        from ctypes import wintypes
+
+        class _ThreadEntry32(ctypes.Structure):
+            _fields_ = [
+                ("dwSize", wintypes.DWORD),
+                ("cntUsage", wintypes.DWORD),
+                ("th32ThreadID", wintypes.DWORD),
+                ("th32OwnerProcessID", wintypes.DWORD),
+                ("tpBasePri", wintypes.LONG),
+                ("tpDeltaPri", wintypes.LONG),
+                ("dwFlags", wintypes.DWORD),
+            ]
+
+        kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)
+        kernel32.CreateToolhelp32Snapshot.argtypes = [wintypes.DWORD, wintypes.DWORD]
+        kernel32.CreateToolhelp32Snapshot.restype = wintypes.HANDLE
+        kernel32.Thread32First.argtypes = [
+            wintypes.HANDLE,
+            ctypes.POINTER(_ThreadEntry32),
+        ]
+        kernel32.Thread32First.restype = wintypes.BOOL
+        kernel32.Thread32Next.argtypes = [
+            wintypes.HANDLE,
+            ctypes.POINTER(_ThreadEntry32),
+        ]
+        kernel32.Thread32Next.restype = wintypes.BOOL
+        kernel32.OpenThread.argtypes = [wintypes.DWORD, wintypes.BOOL, wintypes.DWORD]
+        kernel32.OpenThread.restype = wintypes.HANDLE
+        kernel32.ResumeThread.argtypes = [wintypes.HANDLE]
+        kernel32.ResumeThread.restype = wintypes.DWORD
+        kernel32.CloseHandle.argtypes = [wintypes.HANDLE]
+        kernel32.CloseHandle.restype = wintypes.BOOL
+
+        snapshot = kernel32.CreateToolhelp32Snapshot(0x00000004, 0)
+        if snapshot == ctypes.c_void_p(-1).value:
+            error = ctypes.get_last_error()
+            raise OSError(error, ctypes.FormatError(error))
+
+        resumed = 0
+        entry = _ThreadEntry32()
+        entry.dwSize = ctypes.sizeof(entry)
+        try:
+            has_entry = bool(kernel32.Thread32First(snapshot, ctypes.byref(entry)))
+            while has_entry:
+                if entry.th32OwnerProcessID == process.pid:
+                    thread = kernel32.OpenThread(0x0002, False, entry.th32ThreadID)
+                    if not thread:
+                        error = ctypes.get_last_error()
+                        raise OSError(error, ctypes.FormatError(error))
+                    try:
+                        if kernel32.ResumeThread(thread) == 0xFFFFFFFF:
+                            error = ctypes.get_last_error()
+                            raise OSError(error, ctypes.FormatError(error))
+                        resumed += 1
+                    finally:
+                        kernel32.CloseHandle(thread)
+                has_entry = bool(kernel32.Thread32Next(snapshot, ctypes.byref(entry)))
+        finally:
+            kernel32.CloseHandle(snapshot)
+        if resumed == 0:
+            raise ProcessLookupError(f"no suspended thread found for process {process.pid}")
+
+
 @dataclass(frozen=True)
 class OpenClawInstallAttempt:
     returncode: int
@@ -169,6 +333,28 @@ class ActiveInstallation:
     gateway: ActiveGateway | None
 
 
+@dataclass(frozen=True)
+class WeixinPluginPolicy:
+    plugins_enabled_present: bool
+    plugins_enabled: bool | None
+    entry_present: bool
+    entry: dict[str, object] | None
+    allow_present: bool
+    allow: list[str] | None
+    deny_present: bool
+    deny: list[str] | None
+
+    @property
+    def expects_enabled(self) -> bool:
+        if self.plugins_enabled is False:
+            return False
+        if self.entry is not None and self.entry.get("enabled") is False:
+            return False
+        if self.deny is not None and "openclaw-weixin" in self.deny:
+            return False
+        return not self.allow or "openclaw-weixin" in self.allow
+
+
 @dataclass(frozen=True)
 class _ProcessInfo:
     parent_pid: int
@@ -199,6 +385,11 @@ def __init__(self, config, logger: DeployerLogger):
         self._git_bin: str | None = None  # path to git bin directory
         self._rollback_actions: list[tuple[str, Callable]] = []
         self._openclaw_transaction: OpenClawUpgradeTransaction | None = None
+        self._openclaw_upgrade_required = True
+        self._weixin_plugin_mutation_required = False
+        self._weixin_policy_snapshot: WeixinPluginPolicy | None = None
+        self._weixin_policy_restore_pending = False
+        self._weixin_registration_verified = False
         # Optional UI hook forwarded to upgrade transactions so long backup /
         # restore file operations can report progress instead of looking frozen.
         self.progress_callback: Callable[[str], None] | None = None
@@ -1651,6 +1842,11 @@ def recover_interrupted_openclaw_upgrade(self) -> bool:
 
     def prepare_openclaw_upgrade(self) -> bool:
         """Block active gateways and snapshot the current package and state."""
+        self._openclaw_upgrade_required = True
+        self._weixin_plugin_mutation_required = False
+        self._weixin_policy_snapshot = None
+        self._weixin_policy_restore_pending = False
+        self._weixin_registration_verified = False
         if not self._gateway_is_stopped_for_upgrade():
             return False
         if not self.recover_interrupted_openclaw_upgrade():
@@ -1658,6 +1854,11 @@ def prepare_openclaw_upgrade(self) -> bool:
         installation = self._detect_openclaw_installation()
         if not self._gateway_is_stopped_for_upgrade():
             return False
+        same_version = installation is not None and installation.version == OPENCLAW_TARGET_VERSION
+        if same_version:
+            self.install_prefix = installation.prefix
+            self._openclaw_upgrade_required = False
+            self._weixin_plugin_mutation_required = self._same_version_weixin_requires_full_backup()
         prefix = (
             installation.prefix if installation is not None else self._choose_npm_install_prefix()
         )
@@ -1672,11 +1873,17 @@ def prepare_openclaw_upgrade(self) -> bool:
         )
         self.install_prefix = prefix
         try:
+            backup_mode = (
+                UpgradeBackupMode.FULL
+                if self._openclaw_upgrade_required or self._weixin_plugin_mutation_required
+                else UpgradeBackupMode.MANAGED_STATE
+            )
             transaction = OpenClawUpgradeTransaction.create(
                 microclaw_root=DEFAULT_DESKTOP_DIR,
                 state_dir=Path.home() / ".openclaw",
                 target_version=OPENCLAW_TARGET_VERSION,
                 installation=source,
+                backup_mode=backup_mode,
             )
             self._openclaw_transaction = transaction
             transaction.progress_callback = self.progress_callback
@@ -1684,6 +1891,16 @@ def prepare_openclaw_upgrade(self) -> bool:
                 transaction.close()
                 self._openclaw_transaction = None
                 return False
+            if backup_mode == UpgradeBackupMode.MANAGED_STATE:
+                self.log.info(
+                    f"OpenClaw {OPENCLAW_TARGET_VERSION} is already installed; "
+                    "creating a lightweight managed-state rollback point."
+                )
+            elif same_version:
+                self.log.info(
+                    "The bundled WeChat plugin requires reconciliation; "
+                    "creating a full rollback point."
+                )
             transaction.backup()
             return True
         except Exception as error:
@@ -2094,9 +2311,82 @@ def _contains_enabled_weixin_plugin(value: object) -> bool:
             return any(WindowsSetup._contains_enabled_weixin_plugin(child) for child in value)
         return False
 
+    @staticmethod
+    def _weixin_plugin_from_inspection(payload: object) -> dict[str, object] | None:
+        if not isinstance(payload, dict):
+            return None
+        plugin = payload.get("plugin")
+        if isinstance(plugin, dict) and plugin.get("id") == "openclaw-weixin":
+            return plugin
+        return None
+
+    @staticmethod
+    def _bundled_weixin_version(plugin_dir: Path) -> str | None:
+        try:
+            package = json.loads((plugin_dir / "package.json").read_text(encoding="utf-8"))
+        except (OSError, json.JSONDecodeError):
+            return None
+        version = package.get("version") if isinstance(package, dict) else None
+        return version if isinstance(version, str) and version else None
+
+    def _weixin_registration_matches(
+        self,
+        payload: object,
+        installed_dir: Path,
+        plugin_dir: Path,
+    ) -> bool:
+        plugin = self._weixin_plugin_from_inspection(payload)
+        if plugin is None or not isinstance(payload, dict):
+            return False
+        install = payload.get("install")
+        version = self._bundled_weixin_version(plugin_dir)
+        if not isinstance(install, dict) or version is None:
+            return False
+        try:
+            plugin_root = Path(str(plugin.get("rootDir", ""))).resolve(strict=False)
+            install_path = Path(str(install.get("installPath", ""))).resolve(strict=False)
+            expected_path = installed_dir.resolve(strict=False)
+        except (OSError, ValueError):
+            return False
+        return (
+            plugin_root == expected_path
+            and install_path == expected_path
+            and plugin.get("version") == version
+            and install.get("version") == version
+        )
+
+    def _inspect_weixin_plugin(self) -> object:
+        return self._run_openclaw_json(["plugins", "inspect", "openclaw-weixin", "--json"])
+
     def _validate_weixin_plugin(self) -> bool:
-        payload = self._run_openclaw_json(["plugins", "list", "--json"])
-        return self._contains_enabled_weixin_plugin(payload)
+        plugin_dir = self._find_bundled_weixin_plugin()
+        if plugin_dir is None:
+            return False
+        state_dir = Path.home() / ".openclaw"
+        payload = self._inspect_weixin_plugin()
+        if not self._weixin_registration_matches(
+            payload,
+            state_dir / "extensions" / "openclaw-weixin",
+            plugin_dir,
+        ):
+            return False
+        policy = getattr(self, "_weixin_policy_snapshot", None)
+        if policy is None:
+            policy = self._read_weixin_plugin_policy(state_dir)
+        if policy is None:
+            return False
+        if policy.expects_enabled:
+            return self._contains_enabled_weixin_plugin(payload)
+
+        plugin = self._weixin_plugin_from_inspection(payload)
+        if plugin is None:
+            return False
+        status = str(plugin.get("status", "")).lower()
+        return (
+            plugin.get("enabled") is False
+            or plugin.get("activated") is False
+            or status == "disabled"
+        )
 
     def _validate_appcontainer_smoke(self) -> bool:
         if not self.appcontainer_enabled:
@@ -2128,6 +2418,41 @@ def _validate_appcontainer_smoke(self) -> bool:
         return result.returncode == 0 and "microclaw-sandbox-ok" in result.stdout
 
     def verify_openclaw_upgrade(self) -> bool:
+        if not self._openclaw_upgrade_required:
+            transaction = self._openclaw_transaction
+            if transaction is not None:
+                transaction.mark_verifying()
+
+            def record_fast_path(name: str, passed: bool) -> None:
+                if transaction is not None:
+                    transaction.record_validation(name, passed)
+
+            version_ok = self._validate_installed_version()
+            record_fast_path("version", version_ok)
+            if not version_ok:
+                self.log.error("OpenClaw validation failed: version")
+                return False
+
+            process = None
+            checks = [("health", self._validate_gateway_health)]
+            if getattr(self, "_weixin_plugin_mutation_required", False):
+                checks.append(("weixin-plugin", self._validate_weixin_plugin))
+            checks.append(("appcontainer", self._validate_appcontainer_smoke))
+            try:
+                process = self._start_validation_gateway()
+                for name, check in checks:
+                    passed = bool(check())
+                    record_fast_path(name, passed)
+                    if not passed:
+                        raise RuntimeError(f"OpenClaw validation failed: {name}")
+                self.log.info("OpenClaw version unchanged; skipping deep RPC upgrade validation.")
+                return True
+            except Exception as error:
+                self.log.error(str(error))
+                return False
+            finally:
+                self._stop_validation_gateway(process)
+
         transaction = self._openclaw_transaction
         if transaction is not None:
             transaction.mark_verifying()
@@ -2754,26 +3079,6 @@ def write_config(self) -> bool:
         else:
             self.log.info("  Skill whitelist: disabled (skills.enable=false in deployer config)")
 
-        # ── Clean stale plugin references ──
-        # Previous installs (or the desktop app) may have added plugins.allow
-        # and plugins.entries for openclaw-weixin.  If the plugin files were
-        # removed during uninstall, these stale references cause
-        # `openclaw plugins install` to fail config validation.
-        # Strip them now; _enable_weixin_in_config() will re-add after install.
-        plugins_cfg = existing.get("plugins", {})
-        allow_list = plugins_cfg.get("allow", [])
-        if isinstance(allow_list, list) and "openclaw-weixin" in allow_list:
-            allow_list.remove("openclaw-weixin")
-            plugins_cfg["allow"] = allow_list
-            self.log.info("  Removed stale plugins.allow entry for openclaw-weixin")
-        p_entries = plugins_cfg.get("entries", {})
-        if "openclaw-weixin" in p_entries:
-            del p_entries["openclaw-weixin"]
-            plugins_cfg["entries"] = p_entries
-            self.log.info("  Removed stale plugins.entries for openclaw-weixin")
-        if plugins_cfg:
-            existing["plugins"] = plugins_cfg
-
         try:
             config_path.write_text(json.dumps(existing, indent=2), encoding="utf-8")
             self.log.success(f"Config written to {config_path}")
@@ -3047,73 +3352,280 @@ def _find_bundled_weixin_plugin(self) -> Path | None:
                 return p
         return None
 
-    def _enable_weixin_in_config(self):
-        """Enable openclaw-weixin plugin in openclaw.json after installation.
+    @staticmethod
+    def _plugin_payload_files(root: Path) -> set[Path]:
+        files: set[Path] = set()
+        for directory, dirnames, filenames in os.walk(root, followlinks=False):
+            relative_directory = Path(directory).relative_to(root)
+            if relative_directory == Path("node_modules"):
+                # OpenClaw may add this peer-dependency link after installation.
+                dirnames[:] = [name for name in dirnames if name != "openclaw"]
+            for filename in filenames:
+                files.add(relative_directory / filename)
+        return files
 
-        Also fixes the ``sourcePath`` recorded by ``openclaw plugins install``:
-        the original value points to a PyInstaller temp directory that is
-        deleted after the installer exits.  We rewrite it to match the
-        permanent ``installPath`` so the gateway can resolve the plugin on
-        any machine.
-        """
-        import json
+    @staticmethod
+    def _files_match(source: Path, installed: Path) -> bool:
+        if source.stat().st_size != installed.stat().st_size:
+            return False
+        source_hash = hashlib.sha256()
+        installed_hash = hashlib.sha256()
+        with source.open("rb") as source_file, installed.open("rb") as installed_file:
+            for chunk in iter(lambda: source_file.read(256 * 1024), b""):
+                source_hash.update(chunk)
+            for chunk in iter(lambda: installed_file.read(256 * 1024), b""):
+                installed_hash.update(chunk)
+        return source_hash.digest() == installed_hash.digest()
+
+    def _weixin_payload_matches(self, bundled: Path, installed: Path) -> bool:
+        if not installed.is_dir():
+            return False
+        try:
+            bundled_files = self._plugin_payload_files(bundled)
+            installed_files = self._plugin_payload_files(installed)
+            if not bundled_files or bundled_files != installed_files:
+                return False
+            return all(
+                self._files_match(bundled / relative, installed / relative)
+                for relative in bundled_files
+            )
+        except OSError as error:
+            self.log.info(f"  插件完整性检查失败，将重新安装: {error}")
+            return False
 
-        config_path = Path.home() / ".openclaw" / "openclaw.json"
+    @staticmethod
+    def _read_weixin_plugin_policy(state_dir: Path) -> WeixinPluginPolicy | None:
+        config_path = state_dir / "openclaw.json"
         try:
-            config = (
-                json.loads(config_path.read_text(encoding="utf-8")) if config_path.exists() else {}
+            config = json.loads(config_path.read_text(encoding="utf-8"))
+        except FileNotFoundError:
+            return WeixinPluginPolicy(
+                False,
+                None,
+                False,
+                None,
+                False,
+                None,
+                False,
+                None,
             )
-            plugins = config.setdefault("plugins", {})
-            entries = plugins.setdefault("entries", {})
-            entries.setdefault("openclaw-weixin", {})["enabled"] = True
-
-            # Fix sourcePath → installPath so it survives across machines
-            installs = plugins.get("installs", {})
-            wx_install = installs.get("openclaw-weixin")
-            if isinstance(wx_install, dict):
-                install_path = wx_install.get("installPath", "")
-                if install_path:
-                    wx_install["source"] = "path"
-                    wx_install["sourcePath"] = install_path
-                    self.log.info(f"  已将 sourcePath 修正为 {install_path}")
-
-            config_path.write_text(json.dumps(config, indent=2), encoding="utf-8")
-            self.log.info("  已在配置中启用插件")
-        except Exception as e:
-            self.log.warn(f"  无法在配置中启用插件: {e}")
+        except (OSError, json.JSONDecodeError):
+            return None
+        if not isinstance(config, dict):
+            return None
+        plugins = config.get("plugins")
+        if plugins is None:
+            return WeixinPluginPolicy(
+                False,
+                None,
+                False,
+                None,
+                False,
+                None,
+                False,
+                None,
+            )
+        if not isinstance(plugins, dict):
+            return None
+        plugins_enabled_present = "enabled" in plugins
+        plugins_enabled = plugins.get("enabled")
+        if plugins_enabled_present and not isinstance(plugins_enabled, bool):
+            return None
+        entries = plugins.get("entries")
+        if entries is not None and not isinstance(entries, dict):
+            return None
+        entry_present = isinstance(entries, dict) and "openclaw-weixin" in entries
+        entry = entries.get("openclaw-weixin") if isinstance(entries, dict) else None
+        if entry_present and not isinstance(entry, dict):
+            return None
 
-    def install_weixin_plugin(self) -> bool:
-        """Install the openclaw-weixin plugin via 'openclaw plugins install'."""
-        self.log.step("正在安装插件…")
+        values: dict[str, list[str] | None] = {}
+        presence: dict[str, bool] = {}
+        for key in ("allow", "deny"):
+            presence[key] = key in plugins
+            value = plugins.get(key)
+            if presence[key] and (
+                not isinstance(value, list) or not all(isinstance(item, str) for item in value)
+            ):
+                return None
+            values[key] = list(value) if isinstance(value, list) else None
+        return WeixinPluginPolicy(
+            plugins_enabled_present=plugins_enabled_present,
+            plugins_enabled=plugins_enabled if isinstance(plugins_enabled, bool) else None,
+            entry_present=entry_present,
+            entry=dict(entry) if isinstance(entry, dict) else None,
+            allow_present=presence["allow"],
+            allow=values["allow"],
+            deny_present=presence["deny"],
+            deny=values["deny"],
+        )
 
+    def _restore_weixin_plugin_policy(
+        self,
+        openclaw_cmd: list[str],
+        env: dict[str, str],
+        policy: WeixinPluginPolicy,
+    ) -> bool:
+        replace_paths: list[str] = []
+        plugins_patch: dict[str, object] = {}
+        if policy.entry_present:
+            plugins_patch["entries"] = {"openclaw-weixin": policy.entry}
+            replace_paths.extend(["--replace-path", "plugins.entries.openclaw-weixin"])
+        else:
+            plugins_patch["entries"] = {"openclaw-weixin": None}
+        plugins_patch["enabled"] = (
+            policy.plugins_enabled if policy.plugins_enabled_present else None
+        )
+        for key, present, value in (
+            ("allow", policy.allow_present, policy.allow),
+            ("deny", policy.deny_present, policy.deny),
+        ):
+            plugins_patch[key] = value if present else None
+            if present:
+                replace_paths.extend(["--replace-path", f"plugins.{key}"])
+        patch = {"plugins": plugins_patch}
+
+        result = self._run(
+            openclaw_cmd + ["config", "patch", "--stdin", *replace_paths],
+            input=json.dumps(patch),
+            capture_output=True,
+            text=True,
+            encoding="utf-8",
+            errors="replace",
+            timeout=60,
+            env=env,
+        )
+        if result.returncode == 0:
+            return True
+        self.log.error(
+            f"插件已更新，但无法恢复用户的插件启用策略: {(result.stderr or result.stdout).strip()}"
+        )
+        return False
+
+    def _same_version_weixin_requires_full_backup(self) -> bool:
         plugin_dir = self._find_bundled_weixin_plugin()
-        if not plugin_dir:
-            self.log.error("未找到插件安装包")
+        if plugin_dir is None:
             return False
+        state_dir = Path.home() / ".openclaw"
+        existing = state_dir / "extensions" / "openclaw-weixin"
+        policy = self._read_weixin_plugin_policy(state_dir)
+        self._weixin_policy_snapshot = policy
+        if policy is None or not self._weixin_payload_matches(plugin_dir, existing):
+            return True
+        try:
+            payload = self._inspect_weixin_plugin()
+        except Exception as error:
+            self.log.info(f"  无法确认微信插件官方注册状态，将使用完整备份: {error}")
+            return True
+        self._weixin_registration_verified = self._weixin_registration_matches(
+            payload, existing, plugin_dir
+        )
+        return not self._weixin_registration_verified
 
+    def _weixin_cli_context(self, state_dir: Path) -> tuple[list[str], dict[str, str]] | None:
         openclaw_cmd = self._find_openclaw_cmd()
         if not openclaw_cmd:
             self.log.error("未找到 openclaw 命令，无法安装插件")
-            return False
-
+            return None
         env = self._get_env()
-        # Enable V8 compile cache so the openclaw CLI cold-start is fast
-        state_dir = Path.home() / ".openclaw"
         cache_dir = state_dir / "compile-cache"
         cache_dir.mkdir(parents=True, exist_ok=True)
         env["NODE_COMPILE_CACHE"] = str(cache_dir)
         env["OPENCLAW_STATE_DIR"] = str(state_dir)
+        return openclaw_cmd, env
 
-        # Remove stale extension directory so 'openclaw plugins install' won't
-        # fail with "plugin already exists".
+    @staticmethod
+    def _create_process_lifetime_job(process: subprocess.Popen) -> _WindowsKillOnCloseJob:
+        return _WindowsKillOnCloseJob.attach(process)
+
+    def _terminate_process_tree(
+        self,
+        process: subprocess.Popen,
+        job: _WindowsKillOnCloseJob | None,
+    ) -> None:
+        if job is not None:
+            job.close()
+        if process.poll() is None:
+            if os.name == "nt":
+                self._run(
+                    ["taskkill", "/PID", str(process.pid), "/T", "/F"],
+                    capture_output=True,
+                    text=True,
+                    timeout=15,
+                )
+            else:
+                process.kill()
+        try:
+            process.wait(timeout=15)
+        except subprocess.TimeoutExpired:
+            process.kill()
+            process.wait(timeout=5)
+
+    def install_weixin_plugin(self) -> bool:
+        """Reconcile the bundled openclaw-weixin plugin through OpenClaw."""
+        self.log.step("正在安装插件…")
+
+        plugin_dir = self._find_bundled_weixin_plugin()
+        if not plugin_dir:
+            self.log.error("未找到插件安装包")
+            return False
+
+        state_dir = Path.home() / ".openclaw"
         existing = state_dir / "extensions" / "openclaw-weixin"
-        if existing.exists():
-            self.log.info("  删除旧版插件目录…")
-            shutil.rmtree(existing, ignore_errors=True)
+        cli_context = self._weixin_cli_context(state_dir)
+        if cli_context is None:
+            return False
+        openclaw_cmd, env = cli_context
+        prior_policy = getattr(self, "_weixin_policy_snapshot", None)
+        if getattr(self, "_weixin_policy_restore_pending", False):
+            if prior_policy is None:
+                self.log.error("插件策略恢复状态丢失，无法安全重试")
+                return False
+            if not self._restore_weixin_plugin_policy(openclaw_cmd, env, prior_policy):
+                return False
+            self._weixin_policy_restore_pending = False
+            self._weixin_registration_verified = True
+            self.log.success("用户插件策略恢复成功")
+            return True
+
+        payload_matches = self._weixin_payload_matches(plugin_dir, existing)
+        if prior_policy is None:
+            prior_policy = self._read_weixin_plugin_policy(state_dir)
+            self._weixin_policy_snapshot = prior_policy
+        registration_matches = False
+        if payload_matches:
+            try:
+                registration_matches = self._weixin_registration_matches(
+                    self._inspect_weixin_plugin(),
+                    existing,
+                    plugin_dir,
+                )
+            except Exception as error:
+                self.log.info(f"  微信插件官方注册检查失败，将重新安装: {error}")
+        if payload_matches and registration_matches and prior_policy is not None:
+            self._weixin_registration_verified = True
+            self.log.info("  微信插件已是内置版本且官方注册完整，跳过重复安装")
+            return True
+        if payload_matches and prior_policy is not None and prior_policy.entry_present:
+            self.log.info("  微信插件官方注册记录缺失，将由 OpenClaw 修复")
+        elif payload_matches:
+            self.log.info("  微信插件配置记录缺失，将由 OpenClaw 修复")
+
+        if prior_policy is None:
+            self.log.error("无法读取现有插件策略，拒绝执行可能覆盖用户配置的插件安装")
+            return False
+        transaction = getattr(self, "_openclaw_transaction", None)
+        if transaction is not None and transaction.manifest.backup_mode != UpgradeBackupMode.FULL:
+            self.log.error(
+                "微信插件状态在安装期间发生变化；轻量事务不会执行强制覆盖，请重新运行安装器"
+            )
+            return False
 
+        proc: subprocess.Popen | None = None
+        job: _WindowsKillOnCloseJob | None = None
         try:
             proc = subprocess.Popen(
-                openclaw_cmd + ["plugins", "install", str(plugin_dir)],
+                openclaw_cmd + ["plugins", "install", "--force", str(plugin_dir)],
                 stdout=subprocess.PIPE,
                 stderr=subprocess.STDOUT,
                 text=True,
@@ -3121,28 +3633,39 @@ def install_weixin_plugin(self) -> bool:
                 errors="replace",
                 bufsize=1,
                 env=env,
-                creationflags=_CREATE_NO_WINDOW,
+                creationflags=_CREATE_NO_WINDOW | _CREATE_SUSPENDED,
             )
-            for line in proc.stdout:
+            job = self._create_process_lifetime_job(proc)
+            job.resume(proc)
+            output, _ = proc.communicate(timeout=120)
+            for line in output.splitlines():
                 stripped = line.rstrip()
                 if stripped:
                     self.log.info(f"  plugin: {stripped}")
-            proc.wait(timeout=120)
 
             if proc.returncode == 0:
+                self._weixin_policy_restore_pending = True
+                if not self._restore_weixin_plugin_policy(openclaw_cmd, env, prior_policy):
+                    return False
+                self._weixin_policy_restore_pending = False
+                self._weixin_registration_verified = True
                 self.log.success("插件安装成功")
-                # Enable the plugin in openclaw.json
-                self._enable_weixin_in_config()
                 return True
-            else:
-                self.log.error(f"插件安装失败 (exit {proc.returncode})")
-                return False
+            self.log.error(f"插件安装失败 (exit {proc.returncode})")
+            return False
         except subprocess.TimeoutExpired:
+            if proc is not None:
+                self._terminate_process_tree(proc, job)
             self.log.error("插件安装超时")
             return False
         except Exception as e:
+            if proc is not None and proc.poll() is None:
+                self._terminate_process_tree(proc, job)
             self.log.error(f"插件安装失败: {e}")
             return False
+        finally:
+            if job is not None:
+                job.close()
 
     def create_desktop_shortcut(self) -> bool:
         """Create a desktop shortcut for the MicroClawDesktop client."""
diff --git a/tests/test_install_timing.py b/tests/test_install_timing.py
new file mode 100644
index 0000000..443c51a
--- /dev/null
+++ b/tests/test_install_timing.py
@@ -0,0 +1,54 @@
+import unittest
+
+from deployer.install_timing import InstallTiming
+
+
+class _Log:
+    def __init__(self):
+        self.messages = []
+
+    def info(self, message):
+        self.messages.append(message)
+
+
+class _Clock:
+    def __init__(self, *values):
+        self._values = iter(values)
+
+    def __call__(self):
+        return next(self._values)
+
+
+class InstallTimingTests(unittest.TestCase):
+    def test_logs_step_total_and_slowest_summary(self):
+        log = _Log()
+        timing = InstallTiming(log, clock=_Clock(10.0, 11.0, 13.5, 16.0))
+
+        started_at = timing.start_step()
+        timing.record_step("Installing Git...", started_at, "success")
+        timing.finish("success")
+
+        self.assertEqual(
+            log.messages,
+            [
+                "Install timing: Installing Git [success] 2.50s",
+                "Install timing summary: status=success total=6.00s steps=1",
+                "Slowest install steps: Installing Git=2.50s",
+            ],
+        )
+
+    def test_finish_only_logs_once(self):
+        log = _Log()
+        timing = InstallTiming(log, clock=_Clock(1.0, 2.0))
+
+        timing.finish("failed")
+        timing.finish("success")
+
+        self.assertEqual(
+            log.messages,
+            ["Install timing summary: status=failed total=1.00s steps=0"],
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tests/test_openclaw_upgrade.py b/tests/test_openclaw_upgrade.py
index 016fe35..7a40063 100644
--- a/tests/test_openclaw_upgrade.py
+++ b/tests/test_openclaw_upgrade.py
@@ -18,6 +18,7 @@
     ACTIVE_PHASES,
     OpenClawInstallation,
     OpenClawUpgradeTransaction,
+    UpgradeBackupMode,
     UpgradePhase,
     process_is_alive,
     process_started_at,
@@ -422,13 +423,17 @@ def _installation(self, *, shim_paths: tuple[Path, ...] | None = None) -> OpenCl
         )
 
     def _create(
-        self, *, installation: OpenClawInstallation | None = None
+        self,
+        *,
+        installation: OpenClawInstallation | None = None,
+        backup_mode: UpgradeBackupMode = UpgradeBackupMode.FULL,
     ) -> OpenClawUpgradeTransaction:
         transaction = OpenClawUpgradeTransaction.create(
             microclaw_root=self.microclaw,
             state_dir=self.state,
             target_version="2026.7.1-1",
             installation=installation or self._installation(),
+            backup_mode=backup_mode,
         )
         self.transactions.append(transaction)
         return transaction
@@ -562,6 +567,8 @@ def test_create_writes_schema_v1_manifest_atomically(self) -> None:
                 "created_at",
                 "updated_at",
                 "validation_results",
+                "backup_mode",
+                "config_existed",
             },
         )
         self.assertEqual(data["schema_version"], 1)
@@ -569,6 +576,8 @@ def test_create_writes_schema_v1_manifest_atomically(self) -> None:
         self.assertEqual(data["target_version"], "2026.7.1-1")
         self.assertEqual(data["phase"], "backing-up")
         self.assertEqual(data["validation_results"], {})
+        self.assertEqual(data["backup_mode"], "full")
+        self.assertTrue(data["config_existed"])
         self.assertEqual(
             set(lock_data),
             {
@@ -798,6 +807,51 @@ def test_backup_preserves_durable_state_writes_inventory_then_installs(self) ->
         self.assertIn("package/old.txt", inventory)
         self.assertEqual(tx.manifest.phase, UpgradePhase.INSTALLING)
 
+    def test_managed_state_backup_restores_all_installer_owned_state(self) -> None:
+        (self.state / ".env").write_text("OLD=1", encoding="utf-8")
+        (self.state / "skills").mkdir()
+        (self.state / "skills" / "managed.txt").write_text("old-skill", encoding="utf-8")
+        (self.state / "skill_catalog.json").write_text("old-catalog", encoding="utf-8")
+        (self.state / "MicroClawInstaller.exe").write_text("old-installer", encoding="utf-8")
+        (self.state / "_internal").mkdir()
+        (self.state / "_internal" / "runtime.dll").write_text("old-runtime", encoding="utf-8")
+        tx = self._create(backup_mode=UpgradeBackupMode.MANAGED_STATE)
+        tx.backup()
+
+        self.assertEqual(tx.manifest.backup_mode, UpgradeBackupMode.MANAGED_STATE)
+        self.assertTrue((tx.backup_dir / "state" / "openclaw.json").exists())
+        self.assertTrue((tx.backup_dir / "state" / ".env").exists())
+        self.assertTrue((tx.backup_dir / "state" / "skills" / "managed.txt").exists())
+        self.assertTrue((tx.backup_dir / "state" / "_internal" / "runtime.dll").exists())
+        self.assertFalse((tx.backup_dir / "package").exists())
+        self.assertFalse((tx.backup_dir / "state" / "cache").exists())
+
+        (self.state / "openclaw.json").write_text('{"changed":true}', encoding="utf-8")
+        (self.state / ".env").write_text("NEW=1", encoding="utf-8")
+        (self.state / "skills" / "managed.txt").write_text("new-skill", encoding="utf-8")
+        (self.state / "managed_skill_catalog.json").write_text("new", encoding="utf-8")
+        (self.state / "MicroClawInstaller.exe").write_text("new-installer", encoding="utf-8")
+        (self.state / "_internal" / "runtime.dll").write_text("new-runtime", encoding="utf-8")
+        (self.package / "old.txt").write_text("changed-package", encoding="utf-8")
+        tx.rollback()
+
+        self.assertEqual((self.state / "openclaw.json").read_text(), '{"gateway":{}}')
+        self.assertEqual((self.state / ".env").read_text(), "OLD=1")
+        self.assertEqual(
+            (self.state / "skills" / "managed.txt").read_text(),
+            "old-skill",
+        )
+        self.assertFalse((self.state / "managed_skill_catalog.json").exists())
+        self.assertEqual(
+            (self.state / "MicroClawInstaller.exe").read_text(),
+            "old-installer",
+        )
+        self.assertEqual(
+            (self.state / "_internal" / "runtime.dll").read_text(),
+            "old-runtime",
+        )
+        self.assertEqual((self.package / "old.txt").read_text(), "changed-package")
+
     @unittest.skipUnless(os.name == "nt", "Windows extended path behavior")
     def test_backup_and_rollback_support_package_paths_beyond_max_path(self) -> None:
         relative = (
diff --git a/tests/test_webview_bridge.py b/tests/test_webview_bridge.py
index a4f2cea..a43ca3a 100644
--- a/tests/test_webview_bridge.py
+++ b/tests/test_webview_bridge.py
@@ -44,6 +44,18 @@ def test_prepare_upgrade_cancels_without_stopping_gateway(self):
         setup.stop_active_installation_for_upgrade.assert_not_called()
         setup.prepare_openclaw_upgrade.assert_not_called()
 
+    def test_running_app_confirmation_brings_installer_to_front(self):
+        active = ActiveInstallation(pids=(1234,), gateway=None)
+        window = unittest.mock.Mock()
+        window.create_confirmation_dialog.return_value = True
+
+        with unittest.mock.patch("deployer.webview_bridge._ACTIVE_WINDOW", window):
+            self.assertTrue(self.bridge._confirm_close_running_apps(active))
+
+        window.restore.assert_called_once_with()
+        window.show.assert_called_once_with()
+        window.create_confirmation_dialog.assert_called_once()
+
     def test_cancelled_interactive_step_does_not_retry_or_fail(self):
         with self.bridge._state_lock:
             self.bridge._state["running"] = True
diff --git a/tests/test_windows_setup_upgrade.py b/tests/test_windows_setup_upgrade.py
index fc06f41..6159bb8 100644
--- a/tests/test_windows_setup_upgrade.py
+++ b/tests/test_windows_setup_upgrade.py
@@ -1,5 +1,6 @@
 import json
 import os
+import subprocess
 import tempfile
 import unittest
 import unittest.mock
@@ -8,7 +9,7 @@
 from pathlib import Path
 from types import SimpleNamespace
 
-from deployer.openclaw_upgrade import UpgradePhase
+from deployer.openclaw_upgrade import UpgradeBackupMode, UpgradePhase
 from deployer.openclaw_version import OPENCLAW_TARGET_VERSION
 from deployer.windows_setup import (
     _OPENCLAW_RPC_TIMEOUT,
@@ -19,8 +20,10 @@
     ActiveGateway,
     ActiveInstallation,
     NodeInstallBlocked,
+    WeixinPluginPolicy,
     WindowsSetup,
     _ProcessInfo,
+    _WindowsKillOnCloseJob,
 )
 
 
@@ -77,6 +80,13 @@ def setUp(self):
         self.ws._git_bin = None
         self.ws._rollback_actions = []
         self.ws._openclaw_transaction = None
+        self.ws._openclaw_upgrade_required = True
+        self.ws._weixin_plugin_mutation_required = False
+        self.ws._weixin_policy_snapshot = None
+        self.ws._weixin_policy_restore_pending = False
+        self.ws._weixin_registration_verified = False
+        self.process_job = unittest.mock.Mock()
+        self.ws._create_process_lifetime_job = unittest.mock.Mock(return_value=self.process_job)
         self.ws.progress_callback = None
         self.ws.appcontainer_enabled = True
         self.ws.weixin_plugin_enabled = True
@@ -95,6 +105,53 @@ def _write_package(self, prefix: Path, version: str) -> Path:
         (package / "package.json").write_text(json.dumps({"version": version}), encoding="utf-8")
         return package
 
+    def _write_weixin_payload(self, root: Path, marker: str = "same") -> None:
+        files = {
+            "package.json": json.dumps(
+                {"name": "@tencent-weixin/openclaw-weixin", "version": "2.4.6"}
+            ),
+            "openclaw.plugin.json": json.dumps({"id": "openclaw-weixin", "version": "2.4.6"}),
+            "dist/index.js": marker,
+            "node_modules/zod/package.json": json.dumps({"version": "4.4.3"}),
+            "node_modules/qrcode-terminal/package.json": json.dumps({"version": "0.12.0"}),
+        }
+        for relative, content in files.items():
+            path = root / relative
+            path.parent.mkdir(parents=True, exist_ok=True)
+            path.write_text(content, encoding="utf-8")
+
+    def _weixin_inspection(
+        self,
+        installed: Path,
+        *,
+        enabled: bool = True,
+        status: str = "loaded",
+        activated: bool = True,
+        tracked: bool = True,
+    ) -> dict[str, object]:
+        payload: dict[str, object] = {
+            "plugin": {
+                "id": "openclaw-weixin",
+                "version": "2.4.6",
+                "rootDir": str(installed),
+                "enabled": enabled,
+                "status": status,
+                "activated": activated,
+            }
+        }
+        if tracked:
+            payload["install"] = {
+                "installPath": str(installed),
+                "version": "2.4.6",
+            }
+        return payload
+
+    @staticmethod
+    def _plugin_process(output: str = "") -> unittest.mock.Mock:
+        process = unittest.mock.Mock(returncode=0)
+        process.communicate.return_value = (output, None)
+        return process
+
     def test_target_version_is_current(self):
         prefix = self.home / ".openclaw-node"
         self._write_package(prefix, OPENCLAW_TARGET_VERSION)
@@ -105,6 +162,20 @@ def test_target_version_is_current(self):
         self.assertTrue(self.ws.check_openclaw_windows())
         self.assertEqual(self.ws.install_prefix, prefix)
 
+    @unittest.skipUnless(os.name == "nt", "Windows Job Object behavior")
+    def test_kill_on_close_job_terminates_assigned_process(self):
+        process = subprocess.Popen(
+            ["powershell", "-NoProfile", "-Command", "Start-Sleep -Seconds 30"],
+            creationflags=0x08000000 | 0x00000004,
+        )
+        job = _WindowsKillOnCloseJob.attach(process)
+
+        job.resume(process)
+        job.close()
+        process.wait(timeout=5)
+
+        self.assertIsNotNone(process.returncode)
+
     def test_other_version_requires_upgrade(self):
         self._write_package(self.home / ".openclaw-node", "2026.3.12")
 
@@ -289,22 +360,48 @@ def test_unknown_port_owner_is_never_terminated(self):
 
         self.ws._run.assert_not_called()
 
-    def test_prepare_still_snapshots_a_stopped_exact_target_installation(self):
+    def test_prepare_uses_managed_state_transaction_for_unchanged_installation(self):
         prefix = self.home / ".openclaw-node"
         self._write_package(prefix, OPENCLAW_TARGET_VERSION)
         self.ws._is_tcp_port_open = unittest.mock.Mock(return_value=False)
         self.ws._find_active_gateway_lock = unittest.mock.Mock(return_value=None)
+        self.ws._same_version_weixin_requires_full_backup = unittest.mock.Mock(return_value=False)
         transaction = unittest.mock.Mock()
 
         with unittest.mock.patch(
             "deployer.windows_setup.OpenClawUpgradeTransaction.create",
             return_value=transaction,
-        ):
+        ) as create:
             self.assertTrue(self.ws.prepare_openclaw_upgrade())
 
+        create.assert_called_once()
+        self.assertEqual(
+            create.call_args.kwargs["backup_mode"],
+            UpgradeBackupMode.MANAGED_STATE,
+        )
         transaction.backup.assert_called_once()
+        self.assertEqual(self.ws.install_prefix, prefix)
+        self.assertFalse(self.ws._openclaw_upgrade_required)
         self.assertIs(self.ws._openclaw_transaction, transaction)
 
+    def test_prepare_uses_full_transaction_when_same_version_plugin_needs_repair(self):
+        prefix = self.home / ".openclaw-node"
+        self._write_package(prefix, OPENCLAW_TARGET_VERSION)
+        self.ws._is_tcp_port_open = unittest.mock.Mock(return_value=False)
+        self.ws._find_active_gateway_lock = unittest.mock.Mock(return_value=None)
+        self.ws._same_version_weixin_requires_full_backup = unittest.mock.Mock(return_value=True)
+        transaction = unittest.mock.Mock()
+
+        with unittest.mock.patch(
+            "deployer.windows_setup.OpenClawUpgradeTransaction.create",
+            return_value=transaction,
+        ) as create:
+            self.assertTrue(self.ws.prepare_openclaw_upgrade())
+
+        self.assertEqual(create.call_args.kwargs["backup_mode"], UpgradeBackupMode.FULL)
+        self.assertTrue(self.ws._weixin_plugin_mutation_required)
+        transaction.backup.assert_called_once()
+
     def test_prepare_backs_up_existing_installation_in_same_prefix(self):
         prefix = self.home / ".openclaw-node"
         package = self._write_package(prefix, "2026.3.12")
@@ -711,6 +808,61 @@ def test_validation_records_every_required_check_and_stops_gateway(self):
         transaction.mark_verifying.assert_called_once()
         self.ws._stop_validation_gateway.assert_called_once_with(process)
 
+    def test_same_version_fast_path_runs_health_without_deep_rpc_checks(self):
+        self.ws._openclaw_upgrade_required = False
+        transaction = unittest.mock.Mock()
+        self.ws._openclaw_transaction = transaction
+        process = unittest.mock.Mock()
+        self.ws._validate_installed_version = unittest.mock.Mock(return_value=True)
+        self.ws._start_validation_gateway = unittest.mock.Mock(return_value=process)
+        self.ws._stop_validation_gateway = unittest.mock.Mock()
+        self.ws._validate_gateway_health = unittest.mock.Mock(return_value=True)
+        self.ws._validate_appcontainer_smoke = unittest.mock.Mock(return_value=True)
+        self.ws._validate_gateway_rpc = unittest.mock.Mock()
+
+        self.assertTrue(self.ws.verify_openclaw_upgrade())
+
+        self.ws._validate_installed_version.assert_called_once()
+        self.ws._start_validation_gateway.assert_called_once()
+        self.ws._validate_gateway_health.assert_called_once()
+        self.ws._validate_appcontainer_smoke.assert_called_once()
+        self.ws._validate_gateway_rpc.assert_not_called()
+        self.ws._stop_validation_gateway.assert_called_once_with(process)
+        transaction.mark_verifying.assert_called_once()
+        self.assertEqual(
+            [call.args for call in transaction.record_validation.call_args_list],
+            [
+                ("version", True),
+                ("health", True),
+                ("appcontainer", True),
+            ],
+        )
+
+    def test_same_version_plugin_repair_validates_plugin_before_commit(self):
+        self.ws._openclaw_upgrade_required = False
+        self.ws._weixin_plugin_mutation_required = True
+        transaction = unittest.mock.Mock()
+        self.ws._openclaw_transaction = transaction
+        process = unittest.mock.Mock()
+        self.ws._validate_installed_version = unittest.mock.Mock(return_value=True)
+        self.ws._start_validation_gateway = unittest.mock.Mock(return_value=process)
+        self.ws._stop_validation_gateway = unittest.mock.Mock()
+        self.ws._validate_gateway_health = unittest.mock.Mock(return_value=True)
+        self.ws._validate_weixin_plugin = unittest.mock.Mock(return_value=True)
+        self.ws._validate_appcontainer_smoke = unittest.mock.Mock(return_value=True)
+
+        self.assertTrue(self.ws.verify_openclaw_upgrade())
+
+        self.assertEqual(
+            [call.args for call in transaction.record_validation.call_args_list],
+            [
+                ("version", True),
+                ("health", True),
+                ("weixin-plugin", True),
+                ("appcontainer", True),
+            ],
+        )
+
     def test_failed_validation_returns_false_without_rolling_back_inside_validator(self):
         transaction = unittest.mock.Mock()
         process = unittest.mock.Mock()
@@ -792,6 +944,501 @@ def test_failed_rollback_health_check_discards_transaction(self):
         transaction.discard.assert_called_once()
         transaction.complete_rollback.assert_not_called()
 
+    def test_weixin_install_skips_matching_bundled_payload(self):
+        bundled = self.root / "bundled-weixin"
+        installed = self.home / ".openclaw" / "extensions" / "openclaw-weixin"
+        self._write_weixin_payload(bundled)
+        self._write_weixin_payload(installed)
+        peer = installed / "node_modules" / "openclaw" / "package.json"
+        peer.parent.mkdir(parents=True)
+        peer.write_text("{}", encoding="utf-8")
+        config = installed.parents[1] / "openclaw.json"
+        config.write_text(
+            json.dumps(
+                {
+                    "plugins": {
+                        "entries": {
+                            "openclaw-weixin": {
+                                "enabled": False,
+                            }
+                        }
+                    }
+                }
+            ),
+            encoding="utf-8",
+        )
+        self.ws._find_bundled_weixin_plugin = unittest.mock.Mock(return_value=bundled)
+        self.ws._find_openclaw_cmd = unittest.mock.Mock(return_value=["openclaw.cmd"])
+        self.ws._get_env = unittest.mock.Mock(return_value={})
+        self.ws._inspect_weixin_plugin = unittest.mock.Mock(
+            return_value=self._weixin_inspection(installed, enabled=False, status="disabled")
+        )
+
+        with unittest.mock.patch("deployer.windows_setup.subprocess.Popen") as popen:
+            self.assertTrue(self.ws.install_weixin_plugin())
+
+        self.ws._inspect_weixin_plugin.assert_called_once()
+        popen.assert_not_called()
+
+    def test_weixin_missing_entry_stays_on_fast_path_when_official_registration_matches(self):
+        bundled = self.root / "bundled-weixin"
+        installed = self.home / ".openclaw" / "extensions" / "openclaw-weixin"
+        self._write_weixin_payload(bundled)
+        self._write_weixin_payload(installed)
+        config_path = installed.parents[1] / "openclaw.json"
+        config_path.write_text(
+            json.dumps(
+                {
+                    "plugins": {
+                        "allow": [],
+                        "deny": [],
+                    }
+                }
+            ),
+            encoding="utf-8",
+        )
+        self.ws._find_bundled_weixin_plugin = unittest.mock.Mock(return_value=bundled)
+        self.ws._inspect_weixin_plugin = unittest.mock.Mock(
+            return_value=self._weixin_inspection(installed)
+        )
+
+        self.assertFalse(self.ws._same_version_weixin_requires_full_backup())
+
+        self.assertFalse(self.ws._weixin_policy_snapshot.entry_present)
+        self.assertTrue(self.ws._weixin_registration_verified)
+
+    def test_weixin_matching_payload_repairs_missing_config_record(self):
+        bundled = self.root / "bundled-weixin"
+        installed = self.home / ".openclaw" / "extensions" / "openclaw-weixin"
+        self._write_weixin_payload(bundled)
+        self._write_weixin_payload(installed)
+        self.ws._find_bundled_weixin_plugin = unittest.mock.Mock(return_value=bundled)
+        self.ws._find_openclaw_cmd = unittest.mock.Mock(return_value=["openclaw.cmd"])
+        self.ws._get_env = unittest.mock.Mock(return_value={})
+        self.ws._inspect_weixin_plugin = unittest.mock.Mock(return_value={})
+        self.ws._run = unittest.mock.Mock(
+            return_value=SimpleNamespace(returncode=0, stdout="", stderr="")
+        )
+        process = self._plugin_process()
+
+        with unittest.mock.patch(
+            "deployer.windows_setup.subprocess.Popen", return_value=process
+        ) as popen:
+            self.assertTrue(self.ws.install_weixin_plugin())
+
+        self.assertEqual(
+            popen.call_args.args[0],
+            ["openclaw.cmd", "plugins", "install", "--force", str(bundled)],
+        )
+
+    def test_weixin_install_uses_force_without_deleting_existing_plugin(self):
+        bundled = self.root / "bundled-weixin"
+        installed = self.home / ".openclaw" / "extensions" / "openclaw-weixin"
+        self._write_weixin_payload(bundled, marker="new")
+        self._write_weixin_payload(installed, marker="old")
+        old_marker = installed / "old-only.txt"
+        old_marker.write_text("preserve until OpenClaw replaces it", encoding="utf-8")
+        self.ws._find_bundled_weixin_plugin = unittest.mock.Mock(return_value=bundled)
+        self.ws._find_openclaw_cmd = unittest.mock.Mock(return_value=["openclaw.cmd"])
+        self.ws._get_env = unittest.mock.Mock(return_value={})
+        self.ws._run = unittest.mock.Mock(
+            return_value=SimpleNamespace(returncode=0, stdout="", stderr="")
+        )
+        process = self._plugin_process()
+
+        with unittest.mock.patch(
+            "deployer.windows_setup.subprocess.Popen", return_value=process
+        ) as popen:
+            self.assertTrue(self.ws.install_weixin_plugin())
+
+        command = popen.call_args.args[0]
+        self.assertEqual(
+            command,
+            ["openclaw.cmd", "plugins", "install", "--force", str(bundled)],
+        )
+        self.assertEqual(popen.call_args.kwargs["creationflags"], 0x08000000 | 0x00000004)
+        self.assertTrue(old_marker.exists())
+        process.communicate.assert_called_once_with(timeout=120)
+        self.ws._create_process_lifetime_job.assert_called_once_with(process)
+        self.process_job.resume.assert_called_once_with(process)
+        self.process_job.close.assert_called()
+        self.ws._run.assert_called_once()
+
+    def test_weixin_install_timeout_terminates_job_before_returning(self):
+        bundled = self.root / "bundled-weixin"
+        installed = self.home / ".openclaw" / "extensions" / "openclaw-weixin"
+        self._write_weixin_payload(bundled, marker="new")
+        self._write_weixin_payload(installed, marker="old")
+        config_path = installed.parents[1] / "openclaw.json"
+        config_path.write_text(
+            json.dumps(
+                {
+                    "plugins": {
+                        "entries": {"openclaw-weixin": {"enabled": True}},
+                    }
+                }
+            ),
+            encoding="utf-8",
+        )
+        self.ws._find_bundled_weixin_plugin = unittest.mock.Mock(return_value=bundled)
+        self.ws._find_openclaw_cmd = unittest.mock.Mock(return_value=["openclaw.cmd"])
+        self.ws._get_env = unittest.mock.Mock(return_value={})
+        process = unittest.mock.Mock(returncode=None, pid=4321)
+        process.communicate.side_effect = subprocess.TimeoutExpired("openclaw", 120)
+        process.poll.return_value = 1
+
+        with unittest.mock.patch("deployer.windows_setup.subprocess.Popen", return_value=process):
+            self.assertFalse(self.ws.install_weixin_plugin())
+
+        self.process_job.close.assert_called()
+        process.wait.assert_called_once_with(timeout=15)
+
+    def test_weixin_force_install_restores_user_plugin_policy(self):
+        bundled = self.root / "bundled-weixin"
+        installed = self.home / ".openclaw" / "extensions" / "openclaw-weixin"
+        self._write_weixin_payload(bundled, marker="new")
+        self._write_weixin_payload(installed, marker="old")
+        config_path = installed.parents[1] / "openclaw.json"
+        config_path.write_text(
+            json.dumps(
+                {
+                    "plugins": {
+                        "allow": ["other-plugin"],
+                        "deny": ["openclaw-weixin"],
+                        "entries": {
+                            "openclaw-weixin": {
+                                "enabled": False,
+                            }
+                        },
+                    }
+                }
+            ),
+            encoding="utf-8",
+        )
+        self.ws._find_bundled_weixin_plugin = unittest.mock.Mock(return_value=bundled)
+        self.ws._find_openclaw_cmd = unittest.mock.Mock(return_value=["openclaw.cmd"])
+        self.ws._get_env = unittest.mock.Mock(return_value={"TEST": "1"})
+        process = self._plugin_process()
+        self.ws._run = unittest.mock.Mock(
+            return_value=SimpleNamespace(returncode=0, stdout="", stderr="")
+        )
+
+        with unittest.mock.patch("deployer.windows_setup.subprocess.Popen", return_value=process):
+            self.assertTrue(self.ws.install_weixin_plugin())
+
+        restore = self.ws._run.call_args
+        self.assertEqual(
+            restore.args[0],
+            [
+                "openclaw.cmd",
+                "config",
+                "patch",
+                "--stdin",
+                "--replace-path",
+                "plugins.entries.openclaw-weixin",
+                "--replace-path",
+                "plugins.allow",
+                "--replace-path",
+                "plugins.deny",
+            ],
+        )
+        self.assertEqual(
+            json.loads(restore.kwargs["input"]),
+            {
+                "plugins": {
+                    "entries": {"openclaw-weixin": {"enabled": False}},
+                    "enabled": None,
+                    "allow": ["other-plugin"],
+                    "deny": ["openclaw-weixin"],
+                }
+            },
+        )
+        self.assertEqual(restore.kwargs["env"]["TEST"], "1")
+        self.assertEqual(
+            restore.kwargs["env"]["OPENCLAW_STATE_DIR"],
+            str(self.home / ".openclaw"),
+        )
+
+    def test_weixin_force_install_preserves_global_policy_without_existing_entry(self):
+        bundled = self.root / "bundled-weixin"
+        installed = self.home / ".openclaw" / "extensions" / "openclaw-weixin"
+        self._write_weixin_payload(bundled, marker="new")
+        self._write_weixin_payload(installed, marker="old")
+        config_path = installed.parents[1] / "openclaw.json"
+        config_path.write_text(
+            json.dumps(
+                {
+                    "plugins": {
+                        "allow": ["other-plugin"],
+                        "deny": ["openclaw-weixin"],
+                    }
+                }
+            ),
+            encoding="utf-8",
+        )
+        self.ws._find_bundled_weixin_plugin = unittest.mock.Mock(return_value=bundled)
+        self.ws._find_openclaw_cmd = unittest.mock.Mock(return_value=["openclaw.cmd"])
+        self.ws._get_env = unittest.mock.Mock(return_value={})
+        self.ws._run = unittest.mock.Mock(
+            return_value=SimpleNamespace(returncode=0, stdout="", stderr="")
+        )
+        process = self._plugin_process()
+
+        with unittest.mock.patch("deployer.windows_setup.subprocess.Popen", return_value=process):
+            self.assertTrue(self.ws.install_weixin_plugin())
+
+        patch = json.loads(self.ws._run.call_args.kwargs["input"])
+        self.assertEqual(
+            patch,
+            {
+                "plugins": {
+                    "entries": {"openclaw-weixin": None},
+                    "enabled": None,
+                    "allow": ["other-plugin"],
+                    "deny": ["openclaw-weixin"],
+                }
+            },
+        )
+        self.assertFalse(self.ws._weixin_policy_snapshot.expects_enabled)
+
+    def test_weixin_policy_treats_empty_allow_as_open_and_global_disable_as_closed(self):
+        state_dir = self.home / ".openclaw"
+        state_dir.mkdir()
+        config_path = state_dir / "openclaw.json"
+        config_path.write_text(
+            json.dumps(
+                {
+                    "plugins": {
+                        "allow": [],
+                        "entries": {"openclaw-weixin": {"enabled": True}},
+                    }
+                }
+            ),
+            encoding="utf-8",
+        )
+
+        policy = self.ws._read_weixin_plugin_policy(state_dir)
+
+        self.assertIsNotNone(policy)
+        self.assertTrue(policy.expects_enabled)
+
+        config_path.write_text(
+            json.dumps(
+                {
+                    "plugins": {
+                        "enabled": False,
+                        "allow": [],
+                        "entries": {"openclaw-weixin": {"enabled": True}},
+                    }
+                }
+            ),
+            encoding="utf-8",
+        )
+
+        policy = self.ws._read_weixin_plugin_policy(state_dir)
+
+        self.assertIsNotNone(policy)
+        self.assertFalse(policy.expects_enabled)
+
+    def test_weixin_policy_restore_retry_does_not_reinstall_or_reread_policy(self):
+        bundled = self.root / "bundled-weixin"
+        installed = self.home / ".openclaw" / "extensions" / "openclaw-weixin"
+        self._write_weixin_payload(bundled, marker="new")
+        self._write_weixin_payload(installed, marker="old")
+        config_path = installed.parents[1] / "openclaw.json"
+        config_path.write_text(
+            json.dumps(
+                {
+                    "plugins": {
+                        "allow": ["other-plugin"],
+                        "deny": ["openclaw-weixin"],
+                        "entries": {"openclaw-weixin": {"enabled": False}},
+                    }
+                }
+            ),
+            encoding="utf-8",
+        )
+        self.ws._find_bundled_weixin_plugin = unittest.mock.Mock(return_value=bundled)
+        self.ws._find_openclaw_cmd = unittest.mock.Mock(return_value=["openclaw.cmd"])
+        self.ws._get_env = unittest.mock.Mock(return_value={})
+        self.ws._run = unittest.mock.Mock(
+            side_effect=[
+                SimpleNamespace(returncode=1, stdout="", stderr="transient"),
+                SimpleNamespace(returncode=0, stdout="", stderr=""),
+            ]
+        )
+        process = self._plugin_process()
+
+        with unittest.mock.patch(
+            "deployer.windows_setup.subprocess.Popen", return_value=process
+        ) as popen:
+            self.assertFalse(self.ws.install_weixin_plugin())
+            self.assertTrue(self.ws._weixin_policy_restore_pending)
+            config_path.write_text(
+                json.dumps(
+                    {
+                        "plugins": {
+                            "allow": ["openclaw-weixin"],
+                            "entries": {"openclaw-weixin": {"enabled": True}},
+                        }
+                    }
+                ),
+                encoding="utf-8",
+            )
+            self.assertTrue(self.ws.install_weixin_plugin())
+
+        popen.assert_called_once()
+        self.assertEqual(self.ws._run.call_count, 2)
+        first_patch = json.loads(self.ws._run.call_args_list[0].kwargs["input"])
+        second_patch = json.loads(self.ws._run.call_args_list[1].kwargs["input"])
+        self.assertEqual(first_patch, second_patch)
+        self.assertEqual(first_patch["plugins"]["allow"], ["other-plugin"])
+        self.assertEqual(first_patch["plugins"]["deny"], ["openclaw-weixin"])
+        self.assertFalse(self.ws._weixin_policy_restore_pending)
+
+    def test_weixin_matching_payload_with_missing_official_record_is_reinstalled(self):
+        bundled = self.root / "bundled-weixin"
+        installed = self.home / ".openclaw" / "extensions" / "openclaw-weixin"
+        self._write_weixin_payload(bundled)
+        self._write_weixin_payload(installed)
+        config_path = installed.parents[1] / "openclaw.json"
+        config_path.write_text(
+            json.dumps(
+                {
+                    "plugins": {
+                        "entries": {"openclaw-weixin": {"enabled": True}},
+                    }
+                }
+            ),
+            encoding="utf-8",
+        )
+        self.ws._find_bundled_weixin_plugin = unittest.mock.Mock(return_value=bundled)
+        self.ws._find_openclaw_cmd = unittest.mock.Mock(return_value=["openclaw.cmd"])
+        self.ws._get_env = unittest.mock.Mock(return_value={})
+        self.ws._inspect_weixin_plugin = unittest.mock.Mock(
+            return_value=self._weixin_inspection(installed, tracked=False)
+        )
+        self.ws._run = unittest.mock.Mock(
+            return_value=SimpleNamespace(returncode=0, stdout="", stderr="")
+        )
+        process = self._plugin_process()
+
+        with unittest.mock.patch(
+            "deployer.windows_setup.subprocess.Popen", return_value=process
+        ) as popen:
+            self.assertTrue(self.ws.install_weixin_plugin())
+
+        popen.assert_called_once()
+
+    def test_weixin_force_install_refuses_lightweight_transaction_after_state_changes(self):
+        bundled = self.root / "bundled-weixin"
+        installed = self.home / ".openclaw" / "extensions" / "openclaw-weixin"
+        self._write_weixin_payload(bundled, marker="new")
+        self._write_weixin_payload(installed, marker="old")
+        config_path = installed.parents[1] / "openclaw.json"
+        config_path.write_text(
+            json.dumps(
+                {
+                    "plugins": {
+                        "entries": {"openclaw-weixin": {"enabled": True}},
+                    }
+                }
+            ),
+            encoding="utf-8",
+        )
+        transaction = unittest.mock.Mock()
+        transaction.manifest.backup_mode = UpgradeBackupMode.MANAGED_STATE
+        self.ws._openclaw_transaction = transaction
+        self.ws._find_bundled_weixin_plugin = unittest.mock.Mock(return_value=bundled)
+        self.ws._find_openclaw_cmd = unittest.mock.Mock(return_value=["openclaw.cmd"])
+        self.ws._get_env = unittest.mock.Mock(return_value={})
+
+        with unittest.mock.patch("deployer.windows_setup.subprocess.Popen") as popen:
+            self.assertFalse(self.ws.install_weixin_plugin())
+
+        popen.assert_not_called()
+
+    def test_weixin_install_rechecks_cached_registration_before_fast_path(self):
+        bundled = self.root / "bundled-weixin"
+        installed = self.home / ".openclaw" / "extensions" / "openclaw-weixin"
+        self._write_weixin_payload(bundled)
+        self._write_weixin_payload(installed)
+        config_path = installed.parents[1] / "openclaw.json"
+        config_path.write_text(
+            json.dumps(
+                {
+                    "plugins": {
+                        "entries": {"openclaw-weixin": {"enabled": True}},
+                    }
+                }
+            ),
+            encoding="utf-8",
+        )
+        transaction = unittest.mock.Mock()
+        transaction.manifest.backup_mode = UpgradeBackupMode.MANAGED_STATE
+        self.ws._openclaw_transaction = transaction
+        self.ws._weixin_registration_verified = True
+        self.ws._find_bundled_weixin_plugin = unittest.mock.Mock(return_value=bundled)
+        self.ws._find_openclaw_cmd = unittest.mock.Mock(return_value=["openclaw.cmd"])
+        self.ws._get_env = unittest.mock.Mock(return_value={})
+        self.ws._inspect_weixin_plugin = unittest.mock.Mock(
+            return_value=self._weixin_inspection(installed, tracked=False)
+        )
+
+        with unittest.mock.patch("deployer.windows_setup.subprocess.Popen") as popen:
+            self.assertFalse(self.ws.install_weixin_plugin())
+
+        self.ws._inspect_weixin_plugin.assert_called_once()
+        popen.assert_not_called()
+
+    def test_weixin_validation_accepts_user_disabled_plugin(self):
+        bundled = self.root / "bundled-weixin"
+        installed = self.home / ".openclaw" / "extensions" / "openclaw-weixin"
+        self._write_weixin_payload(bundled)
+        self.ws._find_bundled_weixin_plugin = unittest.mock.Mock(return_value=bundled)
+        self.ws._inspect_weixin_plugin = unittest.mock.Mock(
+            return_value=self._weixin_inspection(
+                installed,
+                enabled=False,
+                status="disabled",
+                activated=False,
+            )
+        )
+        self.ws._weixin_policy_snapshot = WeixinPluginPolicy(
+            plugins_enabled_present=False,
+            plugins_enabled=None,
+            entry_present=True,
+            entry={"enabled": False},
+            allow_present=True,
+            allow=["other-plugin"],
+            deny_present=True,
+            deny=["openclaw-weixin"],
+        )
+
+        self.assertTrue(self.ws._validate_weixin_plugin())
+
+    def test_write_config_preserves_existing_weixin_plugin_settings(self):
+        config_path = self.home / ".openclaw" / "openclaw.json"
+        config_path.parent.mkdir(parents=True)
+        plugins = {
+            "allow": ["openclaw-weixin"],
+            "entries": {
+                "openclaw-weixin": {
+                    "enabled": False,
+                    "config": {"userSetting": "preserved"},
+                }
+            },
+        }
+        config_path.write_text(json.dumps({"plugins": plugins}), encoding="utf-8")
+        self.ws._deploy_managed_skills = unittest.mock.Mock()
+        self.ws._install_officecli = unittest.mock.Mock()
+        self.ws._generate_skill_snapshot = unittest.mock.Mock()
+
+        self.assertTrue(self.ws.write_config())
+
+        written = json.loads(config_path.read_text(encoding="utf-8"))
+        self.assertEqual(written["plugins"], plugins)
+
     def test_desktop_update_preserves_upgrade_transaction_directories(self):
         install_dir = self.root / ".microclaw"
         backup_marker = install_dir / "backups" / "openclaw" / "tx" / "marker.txt"