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 AuditLoggerPlugin, 

14 BasePlugin, 

15 HealthCheckPlugin, 

16 PluginFileWatcher, 

17 PluginManifest, 

18 PluginRegistry, 

19 PluginStatus, 

20 create_registry_with_builtins, 

21) 

22 

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

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

25 

26 

27class PluginManager: 

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

29 

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

31 """ 

32 

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

34 warnings.warn( 

35 "agentos.plugin_manager.PluginManager is deprecated. " 

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

37 DeprecationWarning, 

38 stacklevel=2, 

39 ) 

40 self._registry = PluginRegistry() 

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

42 

43 def discover(self): 

44 discovered = [] 

45 for d in self._plugin_dirs: 

46 names = self._registry.discover(d) 

47 discovered.extend( 

48 PluginInfo( 

49 name=name, 

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

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

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

53 ) 

54 for name in names 

55 ) 

56 return discovered 

57 

58 def register(self, plugin): 

59 if isinstance(plugin, BasePlugin): 

60 return self._registry.register(plugin) 

61 elif isinstance(plugin, PluginInfo): 

62 manifest = PluginManifest( 

63 name=plugin.name, 

64 version=plugin.version, 

65 description=plugin.description, 

66 entry_point=plugin.entry_point, 

67 ) 

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

69 return True 

70 

71 def load(self, name: str): 

72 return self._registry.load(name) 

73 

74 def unload(self, name: str): 

75 plugin = self._registry.get_plugin(name) 

76 if plugin: 

77 plugin.unload() 

78 

79 def add_hook(self, hook_name, callback): 

80 pass # PluginRegistry uses middleware model instead 

81 

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

83 pass # PluginRegistry uses middleware model instead 

84 

85 @property 

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

87 return [n for n, p in self._registry._plugins.items() if p.status == PluginStatus.ACTIVE] 

88 

89 

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

91 

92from dataclasses import dataclass # noqa: E402 

93 

94 

95@dataclass 

96class PluginInfo: 

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

98 

99 name: str 

100 version: str 

101 description: str 

102 entry_point: str 

103 author: str = "" 

104 path: str = "" 

105 

106 

107# ── 公开 API ─────────────────────────────── 

108 

109__all__ = [ 

110 # 新 API（推荐） 

111 "PluginRegistry", 

112 "PluginManifest", 

113 "PluginStatus", 

114 "BasePlugin", 

115 "PluginFileWatcher", 

116 "AuditLoggerPlugin", 

117 "HealthCheckPlugin", 

118 "create_registry_with_builtins", 

119 # 兼容旧 API 

120 "PluginManager", 

121 "PluginInfo", 

122]