1
class BackupActor:
2
    async def handle(self, data: dict) -> None:
3
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
4
        backup_name = f"mini-agent-backup-{timestamp}.tar.gz"
5

6
        # 1. 打包
7
        await asyncio.create_subprocess_exec(
8
            "tar", "-czf", f"/tmp/{backup_name}",
9
            "-C", str(self._base_dir), "data"
10
        )
11

12
        # 2. rsync 到远程
13
        await asyncio.create_subprocess_exec(
14
            "rsync", "-avz", f"/tmp/{backup_name}",
15
            f"{self._rsync_host}:{self._rsync_path}/"
16
        )
17

18
        # 3. 清理本地临时文件和远程旧备份
19
        # ...

1
class HealthCheckActor:
2
    async def handle(self, data: dict) -> None:
3
        issues = []
4

5
        # 磁盘
6
        usage = shutil.disk_usage(self._base_dir)
7
        if usage.percent > self._disk_threshold:
8
            issues.append(f"磁盘使用率 {usage.percent}%")
9

10
        # 内存（可选）
11
        if psutil:
12
            mem = psutil.virtual_memory()
13
            if mem.percent > self._memory_threshold:
14
                issues.append(f"内存使用率 {mem.percent}%")
15

16
        # 数据库
17
        try:
18
            self._todo_manager.list_pending()
19
        except Exception as e:
20
            issues.append(f"数据库异常: {e}")
21

22
        # Pi RPC
23
        try:
24
            subprocess.run([self._pi_path, "--version"],
25
                         capture_output=True, timeout=5, check=True)
26
        except Exception:
27
            issues.append("pi 不可用")
28

29
        if issues:
30
            await self._notifier.alert("系统健康检查异常",
31
                                      "\n".join(f"- {i}" for i in issues))

1
class PiRpcPool:
2
    def __init__(self, size: int = 3):
3
        self._pool: asyncio.Queue[PiRpcWorker] = asyncio.Queue(maxsize=size)
4

5
    async def start(self):
6
        for _ in range(self._size):
7
            worker = PiRpcWorker(self._pi_path)
8
            await worker.start()
9
            await self._pool.put(worker)
10

11
    async def execute(self, prompt: str) -> PiRpcResult:
12
        worker = await self._pool.get()
13
        try:
14
            return await worker.prompt(prompt)
15
        finally:
16
            if worker.is_alive():
17
                await self._pool.put(worker)
18
            else:
19
                # 重启挂掉的 worker
20
                new_worker = PiRpcWorker(self._pi_path)
21
                await new_worker.start()
22
                await self._pool.put(new_worker)