Coverage for agentos/plugins/lifecycle.py: 0%
165 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-08 12:20 +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, PluginStatus, RegisteredPlugin
23# ── 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 """插件健康状态摘要。"""
71 plugin_name: str
72 status: str
73 uptime_seconds: float
74 error_count: int
75 last_error: str
78class HealthStatus:
79 """插件健康状态。"""
81 status: str = "healthy" # healthy | degraded | unhealthy
82 details: dict = field(default_factory=dict)
83 uptime_seconds: float = 0.0
84 error_count: int = 0
86 @property
87 def is_healthy(self) -> bool:
88 return self.status == "healthy"
90 def to_dict(self) -> dict:
91 return {
92 "status": self.status,
93 "details": self.details,
94 "uptime_seconds": self.uptime_seconds,
95 "error_count": self.error_count,
96 }
99from dataclasses import dataclass, field # noqa: E402
102@dataclass
103class LifecycleReport:
104 """插件生命周期报告。"""
106 plugin_name: str
107 status: PluginStatus
108 load_time_ms: float = 0.0
109 init_time_ms: float = 0.0
110 uptime_seconds: float = 0.0
111 health: HealthStatus | None = None
112 error: str | None = None
115# ── Lifecycle Manager ────────────────────────────
118class LifecycleManager:
119 """
120 生命周期管理器 — 协调所有插件的init/start/stop。
121 支持: 批量初始化、健康检查轮询、优雅降级。
122 """
124 def __init__(self, registry: PluginRegistry):
125 self.registry = registry
126 self._reports: dict[str, LifecycleReport] = {}
127 self._health_check_task: asyncio.Task | None = None
129 async def init_all(self, configs: dict[str, dict] | None = None):
130 """初始化所有已注册的LOADED状态插件。"""
131 configs = configs or {}
132 plugins = self.registry.by_status(PluginStatus.LOADED)
134 for rp in plugins:
135 if rp.instance and isinstance(rp.instance, LifecyclePlugin):
136 start = time.time()
137 try:
138 cfg = configs.get(rp.manifest.name, {})
139 await rp.instance.on_init(cfg)
140 rp.status = PluginStatus.INITIALIZED
141 init_ms = (time.time() - start) * 1000
142 self._reports[rp.manifest.name] = LifecycleReport(
143 plugin_name=rp.manifest.name,
144 status=PluginStatus.INITIALIZED,
145 load_time_ms=rp.load_time_ms,
146 init_time_ms=init_ms,
147 )
148 except Exception as e:
149 rp.status = PluginStatus.ERROR
150 rp.error = str(e)
151 await self._handle_error(rp, e)
153 async def start_all(self):
154 """启动所有INITIALIZED状态的插件。"""
155 plugins = self.registry.by_status(PluginStatus.INITIALIZED)
157 for rp in plugins:
158 await self.start_one(rp.manifest.name)
160 async def start_one(self, name: str) -> LifecycleReport:
161 rp = self.registry.get(name)
162 if not rp:
163 raise KeyError(f"Plugin '{name}' not found")
164 if rp.status not in (PluginStatus.INITIALIZED, PluginStatus.STOPPED):
165 raise RuntimeError(f"Cannot start plugin '{name}' in status {rp.status}")
167 try:
168 if rp.instance and isinstance(rp.instance, LifecyclePlugin):
169 await rp.instance.on_start()
170 rp.status = PluginStatus.ACTIVE
171 rp.error = None
172 report = self._reports.get(
173 name, LifecycleReport(plugin_name=name, status=PluginStatus.ACTIVE)
174 )
175 report.status = PluginStatus.ACTIVE
176 report.uptime_seconds = (
177 rp.instance.uptime_seconds
178 if rp.instance and isinstance(rp.instance, LifecyclePlugin)
179 else 0
180 )
181 self._reports[name] = report
182 return report
183 except Exception as e:
184 rp.status = PluginStatus.ERROR
185 rp.error = str(e)
186 return LifecycleReport(plugin_name=name, status=PluginStatus.ERROR, error=str(e))
188 async def stop_all(self, graceful: bool = True):
189 """停止所有ACTIVE插件。"""
190 plugins = self.registry.by_status(PluginStatus.ACTIVE)
192 for rp in plugins:
193 await self.stop_one(rp.manifest.name, graceful)
195 async def stop_one(self, name: str, graceful: bool = True):
196 rp = self.registry.get(name)
197 if not rp:
198 return
200 rp.status = PluginStatus.STOPPING
201 try:
202 if rp.instance and isinstance(rp.instance, LifecyclePlugin):
203 await rp.instance.on_stop()
204 except Exception as e:
205 rp.error = str(e)
206 finally:
207 rp.status = PluginStatus.STOPPED
208 if name in self._reports:
209 self._reports[name].status = PluginStatus.STOPPED
211 async def health_check_all(self) -> dict[str, HealthStatus]:
212 """对所有ACTIVE插件执行健康检查。"""
213 results = {}
214 plugins = self.registry.by_status(PluginStatus.ACTIVE)
216 for rp in plugins:
217 if rp.instance and isinstance(rp.instance, LifecyclePlugin):
218 try:
219 health = await rp.instance.health_check()
220 except Exception as e:
221 health = HealthStatus(
222 status="unhealthy",
223 details={"error": str(e)},
224 error_count=rp.instance._error_count if rp.instance else 0,
225 )
226 health.uptime_seconds = rp.instance.uptime_seconds if rp.instance else 0
227 results[rp.manifest.name] = health
229 return results
231 def start_health_polling(self, interval_seconds: float = 30.0):
232 """启动后台健康检查轮询。"""
234 async def _poll():
235 while True:
236 try:
237 await self.health_check_all()
238 except Exception:
239 pass
240 await asyncio.sleep(interval_seconds)
242 self._health_check_task = asyncio.ensure_future(_poll())
244 def stop_health_polling(self):
245 if self._health_check_task:
246 self._health_check_task.cancel()
247 self._health_check_task = None
249 def report(self) -> list[LifecycleReport]:
250 """获取所有插件的生命周期报告。"""
251 reports = list(self._reports.values())
252 # Add any not tracked in _reports
253 tracked = {r.plugin_name for r in reports}
254 for rp in self.registry.list_all():
255 if rp.manifest.name not in tracked:
256 reports.append(
257 LifecycleReport(
258 plugin_name=rp.manifest.name,
259 status=rp.status,
260 load_time_ms=rp.load_time_ms,
261 error=rp.error,
262 )
263 )
264 return reports
266 def summary(self) -> str:
267 reports = self.report()
268 lines = [f"共 {len(reports)} 个插件"]
269 for r in reports:
270 lines.append(
271 f" [{r.status.value}] {r.plugin_name} (load:{r.load_time_ms:.0f}ms, init:{r.init_time_ms:.0f}ms)"
272 )
273 if r.error:
274 lines.append(f" error: {r.error}")
275 return "\n".join(lines)
277 async def _handle_error(self, rp: RegisteredPlugin, error: Exception):
278 if rp.instance and isinstance(rp.instance, LifecyclePlugin):
279 try:
280 handled = await rp.instance.on_error(error)
281 if handled:
282 rp.error = None
283 except Exception:
284 pass