Coverage for agentos/plugins/lifecycle.py: 0%
165 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-06 10:59 +0800
1"""
2AgentOS v0.70 — 插件生命周期管理器。
3基因来源: Kubernetes Pod Lifecycle + Spring Boot Actuator
5生命周期钩子:
6- on_load() → 插件加载完成
7- on_init(config) → 初始化配置
8- on_start() → 开始工作
9- on_stop() → 优雅关闭
10- on_error(e) → 异常处理
11- health_check() → 健康检查
12"""
14from __future__ import annotations
16import asyncio
17import time
18from abc import ABC
19from dataclasses import dataclass, field
21from agentos.plugins.registry import PluginRegistry, RegisteredPlugin, PluginStatus
24# ── Abstract Plugin Base ─────────────────────────
26class LifecyclePlugin(ABC):
27 """插件基类 — 实现标准生命周期钩子。"""
29 def __init__(self, config: dict | None = None):
30 self.config = config or {}
31 self._started_at: float = 0.0
32 self._error_count: int = 0
34 @property
35 def uptime_seconds(self) -> float:
36 if self._started_at == 0:
37 return 0.0
38 return time.time() - self._started_at
40 async def on_load(self):
41 """插件加载后调用。"""
43 async def on_init(self, config: dict):
44 """初始化配置后调用。"""
45 self.config = config
47 async def on_start(self):
48 """开始工作时调用。"""
49 self._started_at = time.time()
51 async def on_stop(self):
52 """优雅关闭时调用。"""
53 self._started_at = 0.0
55 async def on_error(self, error: Exception) -> bool:
56 """
57 异常处理钩子。返回True表示已处理（可恢复），False表示致命错误。
58 """
59 self._error_count += 1
60 return False
62 async def health_check(self) -> HealthStatus:
63 """健康检查 — 子类可覆盖。"""
64 return HealthStatus.HEALTHY
67@dataclass
68class PluginHealth:
69 """插件健康状态摘要。"""
70 plugin_name: str
71 status: str
72 uptime_seconds: float
73 error_count: int
74 last_error: str
77class HealthStatus:
78 """插件健康状态。"""
79 status: str = "healthy" # healthy | degraded | unhealthy
80 details: dict = field(default_factory=dict)
81 uptime_seconds: float = 0.0
82 error_count: int = 0
84 @property
85 def is_healthy(self) -> bool:
86 return self.status == "healthy"
88 def to_dict(self) -> dict:
89 return {
90 "status": self.status,
91 "details": self.details,
92 "uptime_seconds": self.uptime_seconds,
93 "error_count": self.error_count,
94 }
97from dataclasses import dataclass, field
100@dataclass
101class LifecycleReport:
102 """插件生命周期报告。"""
104 plugin_name: str
105 status: PluginStatus
106 load_time_ms: float = 0.0
107 init_time_ms: float = 0.0
108 uptime_seconds: float = 0.0
109 health: HealthStatus | None = None
110 error: str | None = None
113# ── Lifecycle Manager ────────────────────────────
115class LifecycleManager:
116 """
117 生命周期管理器 — 协调所有插件的init/start/stop。
118 支持: 批量初始化、健康检查轮询、优雅降级。
119 """
121 def __init__(self, registry: PluginRegistry):
122 self.registry = registry
123 self._reports: dict[str, LifecycleReport] = {}
124 self._health_check_task: asyncio.Task | None = None
126 async def init_all(self, configs: dict[str, dict] | None = None):
127 """初始化所有已注册的LOADED状态插件。"""
128 configs = configs or {}
129 plugins = self.registry.by_status(PluginStatus.LOADED)
131 for rp in plugins:
132 if rp.instance and isinstance(rp.instance, LifecyclePlugin):
133 start = time.time()
134 try:
135 cfg = configs.get(rp.manifest.name, {})
136 await rp.instance.on_init(cfg)
137 rp.status = PluginStatus.INITIALIZED
138 init_ms = (time.time() - start) * 1000
139 self._reports[rp.manifest.name] = LifecycleReport(
140 plugin_name=rp.manifest.name,
141 status=PluginStatus.INITIALIZED,
142 load_time_ms=rp.load_time_ms,
143 init_time_ms=init_ms,
144 )
145 except Exception as e:
146 rp.status = PluginStatus.ERROR
147 rp.error = str(e)
148 await self._handle_error(rp, e)
150 async def start_all(self):
151 """启动所有INITIALIZED状态的插件。"""
152 plugins = self.registry.by_status(PluginStatus.INITIALIZED)
154 for rp in plugins:
155 await self.start_one(rp.manifest.name)
157 async def start_one(self, name: str) -> LifecycleReport:
158 rp = self.registry.get(name)
159 if not rp:
160 raise KeyError(f"Plugin '{name}' not found")
161 if rp.status not in (PluginStatus.INITIALIZED, PluginStatus.STOPPED):
162 raise RuntimeError(f"Cannot start plugin '{name}' in status {rp.status}")
164 try:
165 if rp.instance and isinstance(rp.instance, LifecyclePlugin):
166 await rp.instance.on_start()
167 rp.status = PluginStatus.ACTIVE
168 rp.error = None
169 report = self._reports.get(name, LifecycleReport(plugin_name=name, status=PluginStatus.ACTIVE))
170 report.status = PluginStatus.ACTIVE
171 report.uptime_seconds = rp.instance.uptime_seconds if rp.instance and isinstance(rp.instance, LifecyclePlugin) else 0
172 self._reports[name] = report
173 return report
174 except Exception as e:
175 rp.status = PluginStatus.ERROR
176 rp.error = str(e)
177 return LifecycleReport(plugin_name=name, status=PluginStatus.ERROR, error=str(e))
179 async def stop_all(self, graceful: bool = True):
180 """停止所有ACTIVE插件。"""
181 plugins = self.registry.by_status(PluginStatus.ACTIVE)
183 for rp in plugins:
184 await self.stop_one(rp.manifest.name, graceful)
186 async def stop_one(self, name: str, graceful: bool = True):
187 rp = self.registry.get(name)
188 if not rp:
189 return
191 rp.status = PluginStatus.STOPPING
192 try:
193 if rp.instance and isinstance(rp.instance, LifecyclePlugin):
194 await rp.instance.on_stop()
195 except Exception as e:
196 rp.error = str(e)
197 finally:
198 rp.status = PluginStatus.STOPPED
199 if name in self._reports:
200 self._reports[name].status = PluginStatus.STOPPED
202 async def health_check_all(self) -> dict[str, HealthStatus]:
203 """对所有ACTIVE插件执行健康检查。"""
204 results = {}
205 plugins = self.registry.by_status(PluginStatus.ACTIVE)
207 for rp in plugins:
208 if rp.instance and isinstance(rp.instance, LifecyclePlugin):
209 try:
210 health = await rp.instance.health_check()
211 except Exception as e:
212 health = HealthStatus(
213 status="unhealthy",
214 details={"error": str(e)},
215 error_count=rp.instance._error_count if rp.instance else 0,
216 )
217 health.uptime_seconds = rp.instance.uptime_seconds if rp.instance else 0
218 results[rp.manifest.name] = health
220 return results
222 def start_health_polling(self, interval_seconds: float = 30.0):
223 """启动后台健康检查轮询。"""
224 async def _poll():
225 while True:
226 try:
227 await self.health_check_all()
228 except Exception:
229 pass
230 await asyncio.sleep(interval_seconds)
232 self._health_check_task = asyncio.ensure_future(_poll())
234 def stop_health_polling(self):
235 if self._health_check_task:
236 self._health_check_task.cancel()
237 self._health_check_task = None
239 def report(self) -> list[LifecycleReport]:
240 """获取所有插件的生命周期报告。"""
241 reports = list(self._reports.values())
242 # Add any not tracked in _reports
243 tracked = {r.plugin_name for r in reports}
244 for rp in self.registry.list_all():
245 if rp.manifest.name not in tracked:
246 reports.append(LifecycleReport(
247 plugin_name=rp.manifest.name,
248 status=rp.status,
249 load_time_ms=rp.load_time_ms,
250 error=rp.error,
251 ))
252 return reports
254 def summary(self) -> str:
255 reports = self.report()
256 lines = [f"共 {len(reports)} 个插件"]
257 for r in reports:
258 lines.append(f" [{r.status.value}] {r.plugin_name} (load:{r.load_time_ms:.0f}ms, init:{r.init_time_ms:.0f}ms)")
259 if r.error:
260 lines.append(f" error: {r.error}")
261 return "\n".join(lines)
263 async def _handle_error(self, rp: RegisteredPlugin, error: Exception):
264 if rp.instance and isinstance(rp.instance, LifecyclePlugin):
265 try:
266 handled = await rp.instance.on_error(error)
267 if handled:
268 rp.error = None
269 except Exception:
270 pass