BAAH：统一自动修复损坏安装并恢复安装事务

app/task/BAAH/manager.py

@@ -23,14 +23,22 @@
 按账号逐个拉起 BAAH 自动代理任务，结束时统一组装任务报告并推送。
 """
 
+import asyncio
+import shutil
 import uuid
 from contextlib import ExitStack, suppress
 from datetime import datetime
 
 import app.core.notify.report as _node_report
 from app.config import config_manager
 from app.core import Config, EmulatorManager, Notify, Publisher
-from app.core.script_update import ScriptUpdateRequest, run_script_update
+from app.core.script_update import (
+    ScriptUpdateRequest,
+    auto_recover_script_install,
+    run_managed_script_install,
+    run_script_update_with_safe_fallback,
+)
+from app.core.script_update_recovery import DownloadRetryFailed
 from app.core.ws import protocol
 from app.models.config import BAAHConfig
 from app.models.emulator import DeviceBase, DeviceProvider
@@ -45,6 +53,15 @@
 
 from .AutoProxy import AutoProxyTask
 from .tools import program_update
+from .tools.managed_recovery import (
+    BAAHInstallAdapter,
+    acknowledge_stale_transaction,
+    inspect_stale_transaction,
+    snapshot_original_receipts,
+    cleanup_original_receipts,
+    receipt_backup_path,
+)
+from app.services.script_install_backup import ManagedInstallBackup
 
 logger = get_logger("BAAH 调度器")
 
@@ -99,7 +116,8 @@
             return "未填写 BAAH 主程序路径, 请检查托管配置中的主程序路径设置！"
         if not path.is_file():
             directory = script_install.workspace(path.parent)
-            if not ((directory / "transaction.json").exists()):
+            receipt = script_install.record_path(path.parent, "installed.json")
+            if not (directory / "transaction.json").is_file() and not receipt.is_file():
                 return "BAAH 主程序不存在，且没有可信的中断安装记录，请检查安装路径"
 
         return "Pass"
@@ -132,50 +150,143 @@
 
     async def _prepare_program_update(self) -> None:
         root = self.script_config.Script.BAAHPath.parent
         self.check_result = "BAAH 更新准备未完成"
-        # 按真实安装路径排队，等待前一 BAAH 运行的全部账号及恢复动作结束；
-        # 跨 MAS 进程仍由文件锁保障，不改变不同安装目录的并行能力。
         lock = update_files.directory_lock(root.resolve())
         await lock.acquire()
         self._directory_lock = lock
         self._install_access.enter_context(script_install.installation_access(root))
+        request = ScriptUpdateRequest(str(self.script_info.script_id), "BAAH")
+        backup = ManagedInstallBackup(
+            root, script_id=request.script_id, script_type="BAAH"
+        )
+        # 原生事务未完成时不能通过清除 journal 绕过正在使用的备份。
+        try:
+            await script_install.run_worker(
+                script_install.recover_install,
+                root,
+                lambda: program_update.assert_idle(root),
+            )
+        except (script_install.InstallUnsafeError, OSError, ValueError):
+            marker = script_install.workspace(root) / "transaction.json"
+            if marker.is_symlink() or not marker.is_file():
+                raise
+
+            async def retry_native_rollback():
+                await script_install.run_worker(
+                    script_install.recover_install,
+                    root,
+                    lambda: program_update.assert_idle(root),
+                )
+
+            outcome = await auto_recover_script_install(
+                request,
+                install_root=root,
+                inspect=lambda: inspect_stale_transaction(root),
+                acknowledge=lambda: acknowledge_stale_transaction(root),
+                reinstall=retry_native_rollback,
+            )
+            await self.script_info.set_log(
+                f"BAAH 原生安装事务由公共入口恢复：{outcome}"
+            )
+        # 检查旧安装不写目录，可以保留 BAAH 内置的 adb server。
+        await asyncio.to_thread(
+            program_update.assert_idle, root, allow_background_adb=True
+        )
+
+        if backup.read() is not None:
+            adapter = BAAHInstallAdapter(root, self.script_config)
+            result = await run_managed_script_install(
+                request,
+                install_root=root,
+                inspect=adapter.inspect,
+                acknowledge=adapter.acknowledge,
+                install=adapter.install,
+                ensure_idle=adapter.ensure_idle,
+                after_rollback=adapter.after_rollback,
+            )
+            await script_install.run_worker(cleanup_original_receipts, root)
+            await self.script_info.set_log(f"BAAH 上次安装恢复：{result}")
+            self.check_result = "Pass"
+            return
+
+        try:
+            await program_update.confirm_current_install(root)
+            if receipt_backup_path(root).exists():
+                await script_install.run_worker(cleanup_original_receipts, root)
+        except (script_install.InstallUnsafeError, OSError, ValueError) as error:
+            # 损坏本体只能在已核验的 MAS 安装身份下自动完整重装。
+            receipt = script_install.record_path(root, "installed.json")
+            if receipt.is_symlink() or not receipt.is_file():
+                raise script_install.InstallUnsafeError(
+                    "BAAH 原安装不完整且缺少可信回执，不能自动修复"
+                ) from error
+            snapshot = receipt.read_bytes()
+            program_update.verify_recovery_receipt(root, snapshot)
+            adapter = BAAHInstallAdapter(
+                root, self.script_config, recovery_receipt=snapshot
+            )
+            result = await auto_recover_script_install(
+                request,
+                install_root=root,
+                inspect=adapter.inspect,
+                acknowledge=adapter.acknowledge,
+                reinstall=adapter.install,
+            )
+            await self.script_info.set_log(f"BAAH 自动修复：{result}")
+            self.check_result = "Pass"
+            return
+
         if self.script_config.Update.TakeoverEnabled:
 
             async def progress(downloaded: int, total: int, speed: float) -> None:
                 await self.script_info.set_log(
-                    f"BAAH 更新下载：{downloaded / 1024**2:.1f} / {total / 1024**2:.1f} MB"
+                    f"BAAH 下载：{downloaded / 1024**2:.1f} / {total / 1024**2:.1f} MB"
                 )
 
-            try:
-                await self.script_info.set_log(
-                    await run_script_update(
-                        ScriptUpdateRequest(self.script_info.script_id, "BAAH"),
-                        program_update.update_before_run,
-                        root,
-                        self.script_config,
-                        progress=progress,
-                    )
-                )
-            except (script_install.InstallUnsafeError, FluxdownCancelUnconfirmed):
-                raise
-            except Exception as error:
-                await program_update.confirm_current_install(root)
-                message = f"BAAH 更新准备失败，沿用当前安装继续任务：{error}"
-                await self.script_info.set_log(message)
-                logger.warning(message)
-                with suppress(Exception):
-                    await Publisher.send(
-                        id=self.task_info.task_id,
-                        type=protocol.TASK_NOTICE,
-                        data=WSTaskNoticeData(level="warning", message=message),
+            async def prepare():
+                try:
+                    return await program_update.prepare_update(
+                        root, self.script_config, progress=progress
                     )
-        else:
-            # 即使关闭接管，上轮未确认安装也必须先恢复。
-            await script_install.run_worker(
-                script_install.recover_install,
+                except (script_install.InstallUnsafeError, FluxdownCancelUnconfirmed):
+                    raise
+                except Exception as error:
+                    raise DownloadRetryFailed(
+                        source="BAAH GitHub/MirrorChyan",
+                        reason=f"BAAH 更新准备失败（{type(error).__name__}）",
+                    ) from error
+
+            candidate = await run_script_update_with_safe_fallback(
+                request,
+                prepare,
                 root,
-                lambda: program_update.assert_idle(root),
+                BAAHInstallAdapter(root, self.script_config).inspect,
             )
+            if isinstance(candidate, program_update.PreparedUpdate):
+                adapter = BAAHInstallAdapter(
+                    root, self.script_config, prepared=candidate
+                )
+                await script_install.run_worker(snapshot_original_receipts, root)
+                try:
+                    result = await run_managed_script_install(
+                        request,
+                        install_root=root,
+                        inspect=adapter.inspect,
+                        acknowledge=adapter.acknowledge,
+                        install=adapter.install,
+                        ensure_idle=adapter.ensure_idle,
+                        after_rollback=adapter.after_rollback,
+                    )
+                finally:
+                    await adapter.ensure_idle()
+                    if candidate.archive.parent.exists():
+                        shutil.rmtree(candidate.archive.parent)
+                    if backup.read() is None:
+                        await script_install.run_worker(cleanup_original_receipts, root)
+            else:
+                await program_update.confirm_current_install(root)
+                result = candidate or "BAAH 当前无需更新"
+            await self.script_info.set_log(str(result))
         self.check_result = "Pass"
 
     async def main_task(self):