281 lines
9.8 KiB
Python
281 lines
9.8 KiB
Python
"""Manifest-based Addon loader for Odysseus.
|
|
|
|
Addons are intentionally declarative in this first layer. They can describe
|
|
capabilities and contribute prompt/UI metadata, but Odysseus does not execute
|
|
arbitrary addon code automatically.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
import shutil
|
|
import uuid
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
from src.constants import ADDONS_DIR, ADDONS_FILE
|
|
|
|
ADDON_SCHEMA_VERSION = 1
|
|
_ID_RE = re.compile(r"^[a-z0-9][a-z0-9_.-]{1,63}$")
|
|
_ALLOWED_CONTRIBUTIONS = {
|
|
"agent_prompt",
|
|
"tool_hint",
|
|
"settings_panel",
|
|
"ui_panel",
|
|
"api_proxy",
|
|
"skill_template",
|
|
}
|
|
|
|
|
|
def _atomic_write_json(path: str, data: Any, *, indent: Optional[int] = None) -> None:
|
|
os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
|
|
tmp = f"{path}.tmp.{os.getpid()}"
|
|
with open(tmp, "w", encoding="utf-8") as f:
|
|
json.dump(data, f, indent=indent, ensure_ascii=False)
|
|
f.flush()
|
|
os.fsync(f.fileno())
|
|
os.replace(tmp, path)
|
|
|
|
|
|
def _safe_addon_id(value: Any) -> str:
|
|
raw = str(value or "").strip().lower()
|
|
if "/" in raw or "\\" in raw or ".." in raw:
|
|
raise ValueError("Addon id must not contain path separators")
|
|
addon_id = raw
|
|
addon_id = re.sub(r"[^a-z0-9_.-]+", "-", addon_id).strip(".-_")
|
|
if not addon_id:
|
|
addon_id = f"addon-{uuid.uuid4().hex[:8]}"
|
|
if not _ID_RE.match(addon_id):
|
|
raise ValueError("Addon id must be 2-64 chars and use a-z, 0-9, dot, dash, or underscore")
|
|
return addon_id
|
|
|
|
|
|
def _addon_dir(addon_id: str) -> Path:
|
|
safe_id = _safe_addon_id(addon_id)
|
|
base = Path(ADDONS_DIR).resolve()
|
|
path = (base / safe_id).resolve()
|
|
if base not in path.parents and path != base:
|
|
raise ValueError("Invalid addon path")
|
|
return path
|
|
|
|
|
|
def _default_manifest(addon_id: str) -> Dict[str, Any]:
|
|
return {
|
|
"schema_version": ADDON_SCHEMA_VERSION,
|
|
"id": addon_id,
|
|
"name": addon_id.replace("-", " ").replace("_", " ").title(),
|
|
"version": "0.1.0",
|
|
"description": "",
|
|
"author": "",
|
|
"enabled": False,
|
|
"entrypoints": {},
|
|
"contributions": [],
|
|
"permissions": [],
|
|
}
|
|
|
|
|
|
def _normalize_contribution(item: Any) -> Optional[Dict[str, Any]]:
|
|
if not isinstance(item, dict):
|
|
return None
|
|
kind = str(item.get("type") or item.get("kind") or "").strip()
|
|
if kind not in _ALLOWED_CONTRIBUTIONS:
|
|
return None
|
|
normalized = dict(item)
|
|
normalized["type"] = kind
|
|
if "id" in normalized:
|
|
normalized["id"] = str(normalized["id"])[:80]
|
|
return normalized
|
|
|
|
|
|
def normalize_manifest(manifest: Dict[str, Any], *, existing_enabled: Optional[bool] = None) -> Dict[str, Any]:
|
|
if not isinstance(manifest, dict):
|
|
raise ValueError("Addon manifest must be a JSON object")
|
|
if not manifest.get("id") and not manifest.get("name"):
|
|
raise ValueError("Addon manifest requires id or name")
|
|
addon_id = _safe_addon_id(manifest.get("id") or manifest.get("name"))
|
|
normalized = _default_manifest(addon_id)
|
|
normalized.update({
|
|
"schema_version": int(manifest.get("schema_version") or ADDON_SCHEMA_VERSION),
|
|
"id": addon_id,
|
|
"name": str(manifest.get("name") or normalized["name"])[:120],
|
|
"version": str(manifest.get("version") or "0.1.0")[:40],
|
|
"description": str(manifest.get("description") or "")[:1000],
|
|
"author": str(manifest.get("author") or "")[:120],
|
|
"enabled": bool(existing_enabled if existing_enabled is not None else manifest.get("enabled", False)),
|
|
"entrypoints": manifest.get("entrypoints") if isinstance(manifest.get("entrypoints"), dict) else {},
|
|
"permissions": manifest.get("permissions") if isinstance(manifest.get("permissions"), list) else [],
|
|
})
|
|
contributions = manifest.get("contributions") if isinstance(manifest.get("contributions"), list) else []
|
|
normalized["contributions"] = [
|
|
c for c in (_normalize_contribution(item) for item in contributions) if c is not None
|
|
]
|
|
return normalized
|
|
|
|
|
|
def _load_registry() -> Dict[str, Dict[str, Any]]:
|
|
if not os.path.exists(ADDONS_FILE):
|
|
return {}
|
|
try:
|
|
with open(ADDONS_FILE, "r", encoding="utf-8") as f:
|
|
raw = json.load(f)
|
|
except (OSError, json.JSONDecodeError):
|
|
return {}
|
|
items = raw.get("addons", raw) if isinstance(raw, dict) else raw
|
|
if not isinstance(items, list):
|
|
return {}
|
|
registry: Dict[str, Dict[str, Any]] = {}
|
|
for item in items:
|
|
try:
|
|
addon = normalize_manifest(item)
|
|
registry[addon["id"]] = addon
|
|
except Exception:
|
|
continue
|
|
return registry
|
|
|
|
|
|
def _save_registry(registry: Dict[str, Dict[str, Any]]) -> None:
|
|
_atomic_write_json(ADDONS_FILE, {"addons": list(registry.values())}, indent=2)
|
|
|
|
|
|
def scan_addons() -> List[Dict[str, Any]]:
|
|
"""Load addon manifests from disk and merge stored enabled state."""
|
|
os.makedirs(ADDONS_DIR, exist_ok=True)
|
|
registry = _load_registry()
|
|
changed = False
|
|
for child in sorted(Path(ADDONS_DIR).iterdir()):
|
|
if not child.is_dir():
|
|
continue
|
|
manifest_path = child / "addon.json"
|
|
if not manifest_path.exists():
|
|
continue
|
|
try:
|
|
with open(manifest_path, "r", encoding="utf-8") as f:
|
|
manifest = json.load(f)
|
|
addon_id = _safe_addon_id(manifest.get("id") or child.name)
|
|
enabled = registry.get(addon_id, {}).get("enabled")
|
|
addon = normalize_manifest({**manifest, "id": addon_id}, existing_enabled=enabled)
|
|
addon["path"] = str(manifest_path)
|
|
registry[addon_id] = addon
|
|
changed = True
|
|
except Exception as exc:
|
|
registry[child.name] = {
|
|
**_default_manifest(child.name),
|
|
"name": child.name,
|
|
"enabled": False,
|
|
"error": str(exc),
|
|
"path": str(manifest_path),
|
|
}
|
|
changed = True
|
|
if changed:
|
|
_save_registry(registry)
|
|
return list(registry.values())
|
|
|
|
|
|
def list_addons() -> List[Dict[str, Any]]:
|
|
return sorted(scan_addons(), key=lambda item: item.get("name", "").lower())
|
|
|
|
|
|
def get_addon(addon_id: str) -> Optional[Dict[str, Any]]:
|
|
safe_id = _safe_addon_id(addon_id)
|
|
for item in list_addons():
|
|
if item.get("id") == safe_id:
|
|
return item
|
|
return None
|
|
|
|
|
|
def save_addon_manifest(manifest: Dict[str, Any]) -> Dict[str, Any]:
|
|
registry = _load_registry()
|
|
if not manifest.get("id") and not manifest.get("name"):
|
|
raise ValueError("Addon manifest requires id or name")
|
|
existing = registry.get(_safe_addon_id(manifest.get("id") or manifest.get("name")))
|
|
addon = normalize_manifest(manifest, existing_enabled=existing.get("enabled") if existing else None)
|
|
addon_dir = _addon_dir(addon["id"])
|
|
addon_dir.mkdir(parents=True, exist_ok=True)
|
|
manifest_path = addon_dir / "addon.json"
|
|
_atomic_write_json(str(manifest_path), addon, indent=2)
|
|
addon["path"] = str(manifest_path)
|
|
registry[addon["id"]] = addon
|
|
_save_registry(registry)
|
|
return addon
|
|
|
|
|
|
def set_addon_enabled(addon_id: str, enabled: bool) -> Dict[str, Any]:
|
|
registry = _load_registry()
|
|
addon = get_addon(addon_id)
|
|
if not addon:
|
|
raise KeyError(addon_id)
|
|
addon["enabled"] = bool(enabled)
|
|
registry[addon["id"]] = addon
|
|
_save_registry(registry)
|
|
manifest_path = _addon_dir(addon["id"]) / "addon.json"
|
|
if manifest_path.exists():
|
|
persisted = dict(addon)
|
|
persisted.pop("path", None)
|
|
_atomic_write_json(str(manifest_path), persisted, indent=2)
|
|
return addon
|
|
|
|
|
|
def delete_addon(addon_id: str) -> bool:
|
|
safe_id = _safe_addon_id(addon_id)
|
|
registry = _load_registry()
|
|
existed = safe_id in registry
|
|
registry.pop(safe_id, None)
|
|
_save_registry(registry)
|
|
addon_dir = _addon_dir(safe_id)
|
|
if addon_dir.exists():
|
|
shutil.rmtree(addon_dir)
|
|
existed = True
|
|
return existed
|
|
|
|
|
|
def addon_template(addon_id: str = "my-addon") -> Dict[str, Any]:
|
|
safe_id = _safe_addon_id(addon_id)
|
|
manifest = _default_manifest(safe_id)
|
|
manifest.update({
|
|
"name": safe_id.replace("-", " ").title(),
|
|
"description": "Describe what this addon changes or adds.",
|
|
"contributions": [
|
|
{
|
|
"type": "agent_prompt",
|
|
"id": f"{safe_id}.guidance",
|
|
"text": "Short, specific guidance the agent may use when this addon is enabled.",
|
|
},
|
|
{
|
|
"type": "settings_panel",
|
|
"id": f"{safe_id}.settings",
|
|
"title": "Addon Settings",
|
|
"description": "Describe configurable options for this addon.",
|
|
},
|
|
],
|
|
})
|
|
return manifest
|
|
|
|
|
|
def get_enabled_addon_contributions() -> List[Dict[str, Any]]:
|
|
contributions: List[Dict[str, Any]] = []
|
|
for addon in list_addons():
|
|
if not addon.get("enabled") or addon.get("error"):
|
|
continue
|
|
for contribution in addon.get("contributions", []):
|
|
item = dict(contribution)
|
|
item["addon_id"] = addon["id"]
|
|
item["addon_name"] = addon.get("name") or addon["id"]
|
|
contributions.append(item)
|
|
return contributions
|
|
|
|
|
|
def get_addons_prompt() -> str:
|
|
lines: List[str] = []
|
|
for item in get_enabled_addon_contributions():
|
|
if item.get("type") not in {"agent_prompt", "tool_hint", "skill_template"}:
|
|
continue
|
|
text = str(item.get("text") or item.get("description") or "").strip()
|
|
if not text:
|
|
continue
|
|
lines.append(f"- {item.get('addon_name')} ({item.get('type')}): {text[:1200]}")
|
|
if not lines:
|
|
return ""
|
|
return "Enabled Odysseus addons may contribute these untrusted instructions or hints:\n" + "\n".join(lines)
|