Coverage for agentos/plugin_manager.py: 52%

1""" 

2AgentOS 插件系统兼容层 — 统一入口，委托到 agentos.plugins 模块。 

3 

4v1.14.3 版本存在两套并行插件体系（顶层 plugin_manager.py + plugins/ 目录）， 

5v1.14.6 将 plugin_manager.py 改为 plugins/ 的兼容性包装，消除碎片化。 

6""" 

7 

8from __future__ import annotations 

9 

10import warnings 

11 

12from agentos.plugins import ( 

13 PluginRegistry, 

14 PluginManifest, 

15 PluginStatus, 

16 BasePlugin, 

17 PluginFileWatcher, 

18 AuditLoggerPlugin, 

19 HealthCheckPlugin, 

20 create_registry_with_builtins, 

21) 

22 

23# ── 兼容性导出 ──────────────────────────── 

24# 保留旧 PluginManager 接口但内部委托到 PluginRegistry 

25 

26class PluginManager: 

27 """兼容性包装：旧 PluginManager API 委托到 PluginRegistry。 

28 

29 Deprecated: 请直接使用 agentos.plugins.PluginRegistry。 

30 """ 

31 

32 def __init__(self, plugin_dirs: list[str] | None = None): 

33 warnings.warn( 

34 "agentos.plugin_manager.PluginManager is deprecated. " 

35 "Use agentos.plugins.PluginRegistry directly.", 

36 DeprecationWarning, 

37 stacklevel=2, 

38 ) 

39 self._registry = PluginRegistry() 

40 self._plugin_dirs = plugin_dirs or ["./plugins"] 

41 

42 def discover(self): 

43 discovered = [] 

44 for d in self._plugin_dirs: 

45 names = self._registry.discover(d) 

46 discovered.extend( 

47 PluginInfo( 

48 name=name, 

49 version=self._registry._manifests[name].version, 

50 description=self._registry._manifests[name].description, 

51 entry_point=self._registry._manifests[name].entry_point, 

52 ) 

53 for name in names 

54 ) 

55 return discovered 

56 

57 def register(self, plugin): 

58 if isinstance(plugin, BasePlugin): 

59 return self._registry.register(plugin) 

60 elif isinstance(plugin, PluginInfo): 

61 manifest = PluginManifest( 

62 name=plugin.name, 

63 version=plugin.version, 

64 description=plugin.description, 

65 entry_point=plugin.entry_point, 

66 ) 

67 self._registry._manifests[plugin.name] = manifest 

68 return True 

69 

70 def load(self, name: str): 

71 return self._registry.load(name) 

72 

73 def unload(self, name: str): 

74 plugin = self._registry.get_plugin(name) 

75 if plugin: 

76 plugin.unload() 

77 

78 def add_hook(self, hook_name, callback): 

79 pass # PluginRegistry uses middleware model instead 

80 

81 def call_hook(self, hook_name, *args, **kwargs): 

82 pass # PluginRegistry uses middleware model instead 

83 

84 @property 

85 def loaded_plugins(self) -> list[str]: 

86 return [n for n, p in self._registry._plugins.items() 

87 if p.status == PluginStatus.ACTIVE] 

88 

89 

90# ── 兼容类型 ─────────────────────────────── 

91 

92from dataclasses import dataclass 

93 

94@dataclass 

95class PluginInfo: 

96 """旧版 PluginInfo 兼容类型。""" 

97 name: str 

98 version: str 

99 description: str 

100 entry_point: str 

101 author: str = "" 

102 path: str = "" 

103 

104 

105# ── 公开 API ─────────────────────────────── 

106 

107__all__ = [ 

108 # 新 API（推荐） 

109 "PluginRegistry", 

110 "PluginManifest", 

111 "PluginStatus", 

112 "BasePlugin", 

113 "PluginFileWatcher", 

114 "AuditLoggerPlugin", 

115 "HealthCheckPlugin", 

116 "create_registry_with_builtins", 

117 # 兼容旧 API 

118 "PluginManager", 

119 "PluginInfo", 

120]