79 lines
2.8 KiB
Python
79 lines
2.8 KiB
Python
"""Atomic state persistence for declarative plugins."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from .models import PluginSkillBinding, PluginState
|
|
|
|
|
|
class PluginStateStore:
|
|
def __init__(self, workspace: str | Path) -> None:
|
|
self.workspace = Path(workspace)
|
|
self.root = self.workspace / ".beaver" / "plugins"
|
|
self.path = self.root / "state.json"
|
|
|
|
def list_plugins(self) -> list[PluginState]:
|
|
return [
|
|
PluginState.from_dict(plugin_id, payload if isinstance(payload, dict) else {})
|
|
for plugin_id, payload in sorted(self._read_state().get("plugins", {}).items())
|
|
]
|
|
|
|
def get_plugin(self, plugin_id: str) -> PluginState | None:
|
|
payload = self._read_state().get("plugins", {}).get(plugin_id)
|
|
if not isinstance(payload, dict):
|
|
return None
|
|
return PluginState.from_dict(plugin_id, payload)
|
|
|
|
def set_enabled(self, plugin_id: str, enabled: bool) -> PluginState:
|
|
state = self.get_plugin(plugin_id) or PluginState(plugin_id=plugin_id)
|
|
state.enabled = enabled
|
|
if enabled and state.status == "discovered":
|
|
state.status = "enabled"
|
|
self.upsert_plugin(state)
|
|
return state
|
|
|
|
def upsert_plugin(self, plugin_state: PluginState) -> None:
|
|
state = self._read_state()
|
|
plugins = state.setdefault("plugins", {})
|
|
if not isinstance(plugins, dict):
|
|
plugins = {}
|
|
state["plugins"] = plugins
|
|
plugins[plugin_state.plugin_id] = plugin_state.to_dict()
|
|
self._write_state(state)
|
|
|
|
def update_skill_binding(
|
|
self,
|
|
plugin_id: str,
|
|
skill_name: str,
|
|
binding: PluginSkillBinding,
|
|
) -> PluginState:
|
|
state = self.get_plugin(plugin_id) or PluginState(plugin_id=plugin_id)
|
|
state.skills[skill_name] = binding
|
|
self.upsert_plugin(state)
|
|
return state
|
|
|
|
def _read_state(self) -> dict[str, Any]:
|
|
if not self.path.exists():
|
|
return {"plugins": {}}
|
|
payload = json.loads(self.path.read_text(encoding="utf-8"))
|
|
if not isinstance(payload, dict):
|
|
return {"plugins": {}}
|
|
plugins = payload.get("plugins")
|
|
if not isinstance(plugins, dict):
|
|
payload["plugins"] = {}
|
|
return payload
|
|
|
|
def _write_state(self, state: dict[str, Any]) -> None:
|
|
self.root.mkdir(parents=True, exist_ok=True)
|
|
tmp_path = self.path.with_name("state.json.tmp")
|
|
with tmp_path.open("w", encoding="utf-8") as handle:
|
|
json.dump(state, handle, ensure_ascii=False, sort_keys=True, indent=2)
|
|
handle.write("\n")
|
|
handle.flush()
|
|
os.fsync(handle.fileno())
|
|
os.replace(tmp_path, self.path)
|