1
from __future__ import annotations
2
3
import re, json
4
-from dataclasses import dataclass
4
from pathlib import Path
5
from typing import Any, Dict, List, Optional, TYPE_CHECKING
6
7
from python.helpers import files, print_style
8
+from pydantic import BaseModel
9
10
if TYPE_CHECKING:
11
from agent import Agent
17
)
18
19
META_FILE_NAME = "plugin.json"
20
-SETTINGS_FILE_NAME = "settings.json"
20
+CONFIG_FILE_NAME = "config.json"
21
+DISABLED_FILE_NAME = ".disabled"
22
+ENABLED_FILE_NAME = ".enabled"
23
22
-@dataclass(slots=True)
23
-class PluginListItem:
24
+
25
+class PluginMetadata(BaseModel):
26
+ description: str = ""
27
+
28
+
29
+class PluginListItem(BaseModel):
30
name: str
25
- path: Path
31
+ path: str
32
+ description: str = ""
33
+ has_main_screen: bool = False
34
+ has_config_screen: bool = False
35
36
37
def get_plugin_roots() -> List[str]:
38
"""Plugin root directories, ordered by priority (user first)."""
30
- # Project-specific plugins (commented out for now, will add project/agent plugins later)
31
- # projects = files.find_existing_paths_by_pattern("usr/projects/*/.a0proj/plugins")
39
return [
33
- # *projects,
40
files.get_abs_path(files.USER_DIR, files.PLUGINS_DIR),
41
files.get_abs_path(files.PLUGINS_DIR),
42
]
43
44
39
-def list_plugins() -> List[PluginListItem]:
40
- """Discover plugins by directory convention. First root wins on ID conflict."""
41
- by_id: Dict[str, PluginListItem] = {}
45
+def get_plugins_list():
46
+ result: list[str] = []
47
+ seen_names: set[str] = set()
48
for root in get_plugin_roots():
43
- root_path = Path(root)
44
- if not root_path.exists():
45
- continue
46
- for d in sorted(root_path.iterdir(), key=lambda p: p.name):
47
- if not d.is_dir() or d.name.startswith("."):
49
+ for dir in Path(root).iterdir():
50
+ if not dir.is_dir() or dir.name.startswith("."):
51
+ continue
52
+ if dir.name in seen_names:
53
continue
49
- if d.name not in by_id:
50
- by_id[d.name] = PluginListItem(name=d.name, path=d)
51
- return list(by_id.values())
54
+ if files.exists(str(dir), META_FILE_NAME):
55
+ seen_names.add(dir.name)
56
+ result.append(dir.name)
57
+ result.sort(key=lambda p: Path(p).name)
58
+ return result
59
+
60
+
61
+def get_enhanced_plugins_list(
62
+ custom: bool = True, builtin: bool = True
63
+) -> List[PluginListItem]:
64
+ """Discover plugins by directory convention. First root wins on ID conflict."""
65
+ results = []
66
+
67
+ def load_plugins(root_path: str):
68
+ for d in sorted(Path(root_path).iterdir(), key=lambda p: p.name):
69
+ try:
70
+ if not d.is_dir() or d.name.startswith("."):
71
+ continue
72
+ meta = PluginMetadata.model_validate(
73
+ files.read_file_json(str(d / META_FILE_NAME))
74
+ )
75
+ has_main_screen = files.exists(str(d / "webui" / "main.html"))
76
+ has_config_screen = files.exists(str(d / "webui" / "config.html"))
77
+ results.append(
78
+ PluginListItem(
79
+ name=d.name,
80
+ path=str(d),
81
+ description=meta.description,
82
+ has_main_screen=has_main_screen,
83
+ has_config_screen=has_config_screen,
84
+ )
85
+ )
86
+ except:
87
+ pass
88
+
89
+ if custom:
90
+ load_plugins(files.get_abs_path(files.USER_DIR, files.PLUGINS_DIR))
91
+ if builtin:
92
+ load_plugins(files.get_abs_path(files.PLUGINS_DIR))
93
+ return results
94
95
96
def find_plugin_dir(plugin_name: str):
55
- """Find a single plugin by ID."""
97
if not plugin_name:
98
return None
99
100
# check if the plugin is in the user directory
60
- user_plugin_path = files.get_abs_path(files.USER_DIR, files.PLUGINS_DIR, plugin_name, META_FILE_NAME)
101
+ user_plugin_path = files.get_abs_path(
102
+ files.USER_DIR, files.PLUGINS_DIR, plugin_name, META_FILE_NAME
103
+ )
104
if files.exists(user_plugin_path):
105
return files.get_abs_path(files.USER_DIR, files.PLUGINS_DIR, plugin_name)
63
-
106
+
107
# check if the plugin is in the default directory
65
- default_plugin_path = files.get_abs_path(files.PLUGINS_DIR, plugin_name, META_FILE_NAME)
108
+ default_plugin_path = files.get_abs_path(
109
+ files.PLUGINS_DIR, plugin_name, META_FILE_NAME
110
+ )
111
if files.exists(default_plugin_path):
112
return files.get_abs_path(files.PLUGINS_DIR, plugin_name)
68
-
113
+
114
return None
115
116
117
def get_plugin_paths(*subpaths: str) -> List[str]:
73
- """
74
- Resolve existing directories under each plugin matching subpaths.
75
-
76
- Example:
77
- get_plugin_paths("extensions", "python", "monologue_end")
78
- -> ["/abs/plugins/memory/extensions/python/monologue_end", ...]
79
- """
80
- sub = "/".join(subpaths) if subpaths else ""
118
+ sub = "*/" + "/".join(subpaths) if subpaths else "*"
119
paths: List[str] = []
82
- for plugin in list_plugins():
83
- candidate = str(plugin.path / sub) if sub else str(plugin.path)
84
- if Path(candidate).is_dir() and candidate not in paths:
85
- paths.append(candidate)
120
+ for root in get_plugin_roots():
121
+ paths.extend(
122
+ files.find_existing_paths_by_pattern(files.get_abs_path(root, sub))
123
+ )
124
return paths
125
126
+def get_enabled_plugin_paths(agent:Agent|None, *subpaths: str) -> List[str]:
127
+ enabled = get_enabled_plugins(agent)
128
+ paths: list[str] = []
129
89
-def get_webui_extensions(extension_point:str, filters:List[str]|None=None) -> List[Dict[str, Any]]:
90
- entries: List[Dict[str, Any]] = []
91
- effective_filters = filters or ["*"]
92
- for plugin in list_plugins():
93
- frontend_dir = plugin.path / "extensions" / "webui" / extension_point
94
- if not frontend_dir.is_dir():
130
+ for plugin in enabled:
131
+ base_dir = find_plugin_dir(plugin)
132
+ if not base_dir:
133
continue
96
- matched_files: List[Path] = []
97
- seen: set[str] = set()
98
- for pattern in effective_filters:
99
- for p in frontend_dir.rglob(pattern):
100
- if not p.is_file():
101
- continue
102
- p_str = str(p)
103
- if p_str in seen:
104
- continue
105
- seen.add(p_str)
106
- matched_files.append(p)
134
108
- for ext_file in sorted(matched_files, key=lambda p: p.name):
109
- try:
110
- rel_path = files.deabsolute_path(str(ext_file))
111
- entry: Dict[str, Any] = {
112
- "plugin_name": plugin.name,
113
- "path": rel_path,
114
- }
115
- entries.append(entry)
116
- except Exception:
117
- print_style.PrintStyle.error(f"Failed to load frontend extension file {ext_file}")
135
+ if not subpaths:
136
+ if files.exists(base_dir):
137
+ paths.append(base_dir)
138
+ continue
139
+
140
+ path = files.get_abs_path(base_dir, *subpaths)
141
+ if files.exists(path):
142
+ paths.append(path)
143
+
144
+ return paths
145
+
146
+
147
+def get_enabled_plugins(agent: Agent | None):
148
+ plugins = get_plugins_list()
149
+ active = []
150
+
151
+ if agent:
152
+ from python.helpers import subagents
153
+
154
+ for plugin in plugins:
155
+ # plugins are toggled via .enabled / .disabled files
156
+ # every plugin is on by default, unless disabled in usr dir
157
+ enabled = True
158
+
159
+ if agent:
160
+ agent_paths = subagents.get_paths(
161
+ agent,
162
+ files.PLUGINS_DIR,
163
+ plugin,
164
+ must_exist_completely=True,
165
+ include_default=False,
166
+ include_user=True,
167
+ include_plugins=False,
168
+ include_project=True
169
+ )
170
+
171
+ # go through agent paths in reverse order and determine the state
172
+ for agent_path in reversed(agent_paths):
173
+ if enabled:
174
+ enabled = not files.exists(files.get_abs_path(agent_path, DISABLED_FILE_NAME))
175
+ else:
176
+ enabled = files.exists(files.get_abs_path(agent_path, ENABLED_FILE_NAME))
177
+
178
+
179
+ if enabled:
180
+ active.append(plugin)
181
+
182
+ return active
183
+
184
+
185
+
186
+def get_webui_extensions(extension_point: str, filters: List[str] | None = None):
187
+ entries: List[str] = []
188
+ effective_filters = filters or ["*"]
189
+
190
+ for filter in effective_filters:
191
+ extensions = get_plugin_paths("extensions", "webui", extension_point, filter)
192
+ for extension in extensions:
193
+ rel_path = files.deabsolute_path(extension)
194
+ entries.append(rel_path)
195
+
196
return entries
197
120
-def get_plugin_settings(plugin_name:str, agent:Agent|None):
121
- file_path = find_plugin_file(plugin_name, SETTINGS_FILE_NAME, agent=agent)
198
+
199
+def get_plugin_config(plugin_name: str, agent: Agent | None):
200
+ file_path = find_plugin_asset(plugin_name, CONFIG_FILE_NAME, agent=agent)
201
if file_path:
202
return json.loads(files.read_file(file_path))
203
return None
204
126
-def save_plugin_settings(plugin_name:str, project_name:str, agent_profile:str, settings:dict):
127
- file_path = determine_plugin_save_file_path(plugin_name, project_name, agent_profile, SETTINGS_FILE_NAME)
205
+
206
+def save_plugin_config(
207
+ plugin_name: str, project_name: str, agent_profile: str, settings: dict
208
+):
209
+ file_path = determine_plugin_asset_path(
210
+ plugin_name, project_name, agent_profile, CONFIG_FILE_NAME
211
+ )
212
if file_path:
213
files.write_file(file_path, json.dumps(settings))
214
131
-def find_plugin_file(plugin_name:str, *subpaths:str, agent:Agent|None=None):
132
- profile_name = agent.config.profile if agent and agent.config.profile else ""
215
+
216
+def find_plugin_asset(plugin_name: str, *subpaths: str, agent: Agent | None = None):
217
project_name = ""
218
219
if agent:
220
+ profile_name = agent.config.profile if agent and agent.config.profile else ""
221
+
222
from python.helpers import projects
223
+
224
project_name = projects.get_context_project_name(agent.context) or ""
225
226
if project_name and profile_name:
227
# project/.a0proj/agents/<profile>/plugins/<plugin_name>/...
141
- project_agent_file = projects.get_project_meta_folder(
142
- project_name, files.AGENTS_DIR, profile_name, files.PLUGINS_DIR, plugin_name, *subpaths
228
+ project_agent_file = projects.get_project_meta(
229
+ project_name,
230
+ files.AGENTS_DIR,
231
+ profile_name,
232
+ files.PLUGINS_DIR,
233
+ plugin_name,
234
+ *subpaths,
235
)
236
if files.exists(project_agent_file):
237
return project_agent_file
238
239
if project_name:
240
# project/.a0proj/plugins/<plugin_name>/...
149
- project_file = projects.get_project_meta_folder(project_name, files.PLUGINS_DIR, plugin_name, *subpaths)
241
+ project_file = projects.get_project_meta(
242
+ project_name, files.PLUGINS_DIR, plugin_name, *subpaths
243
+ )
244
if files.exists(project_file):
245
return project_file
246
153
- if profile_name:
154
- from python.helpers import subagents
155
- # usr/agents/<profile>/plugins/<plugin_name>/...
156
- path = files.get_abs_path(subagents.USER_AGENTS_DIR, profile_name, files.PLUGINS_DIR, plugin_name, *subpaths)
157
- if files.exists(path):
158
- return path
247
+ if profile_name:
248
+ from python.helpers import subagents
249
160
- # agents/<profile>/plugins/<plugin_name>/...
161
- path = files.get_abs_path(subagents.DEFAULT_AGENTS_DIR, profile_name, files.PLUGINS_DIR, plugin_name, *subpaths)
162
- if files.exists(path):
163
- return path
250
+ # usr/agents/<profile>/plugins/<plugin_name>/...
251
+ path = files.get_abs_path(
252
+ subagents.USER_AGENTS_DIR,
253
+ profile_name,
254
+ files.PLUGINS_DIR,
255
+ plugin_name,
256
+ *subpaths,
257
+ )
258
+ if files.exists(path):
259
+ return path
260
+
261
+ # agents/<profile>/plugins/<plugin_name>/...
262
+ path = files.get_abs_path(
263
+ subagents.DEFAULT_AGENTS_DIR,
264
+ profile_name,
265
+ files.PLUGINS_DIR,
266
+ plugin_name,
267
+ *subpaths,
268
+ )
269
+ if files.exists(path):
270
+ return path
271
272
# usr/plugins/<plugin_name>/...
273
path = files.get_abs_path(files.USER_DIR, files.PLUGINS_DIR, plugin_name, *subpaths)
280
return path
281
282
return None
176
-
177
-def determine_plugin_save_file_path(plugin_name:str, project_name:str, agent_profile:str, *subpaths:str):
283
+
284
+
285
+def determine_plugin_asset_path(
286
+ plugin_name: str, project_name: str, agent_profile: str, *subpaths: str
287
+):
288
base_path = files.get_abs_path(files.USER_DIR)
179
-
289
+
290
if project_name:
291
from python.helpers import projects
182
- base_path = projects.get_project_meta_folder(project_name)
292
+
293
+ base_path = projects.get_project_meta(project_name)
294
295
if agent_profile:
296
base_path = files.get_abs_path(base_path, files.AGENTS_DIR, agent_profile)
297
187
- return files.get_abs_path(base_path, files.PLUGINS_DIR, plugin_name, *subpaths)
\ No newline at end of file
298
+ return files.get_abs_path(base_path, files.PLUGINS_DIR, plugin_name, *subpaths)