File size: 3,656 Bytes
e6b2bd6 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 | #!/usr/bin/env python3
"""PluginLoader — Discovers and loads RulePlugin subclasses from a directory.
Supports hot-reload: call load() again after SIGHUP to re-import all plugins.
"""
import importlib
import importlib.util
import os
import sys
import time
# Import from sibling module
try:
from .rule_plugin import RulePlugin
except ImportError:
from rule_plugin import RulePlugin
def _log(msg: str) -> None:
"""Minimal logger — same pattern as sidecar.py."""
try:
with open("/tmp/claude-sidecar.log", "a") as f:
f.write(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] [plugin_loader] {msg}\n")
except Exception:
pass
class PluginLoader:
"""Loads RulePlugin subclasses from a directory of .py files."""
def __init__(self, rules_dir: str):
self.rules_dir = rules_dir
self.plugins: list[RulePlugin] = []
self._last_injection: str = "" # delta-dedup cache
def load(self) -> list[RulePlugin]:
"""(Re)load all plugins from rules_dir. Returns list of instances."""
self.plugins = []
if not os.path.isdir(self.rules_dir):
_log(f"Rules directory not found: {self.rules_dir}")
return self.plugins
for filename in sorted(os.listdir(self.rules_dir)):
if not filename.endswith(".py") or filename.startswith("_"):
continue
filepath = os.path.join(self.rules_dir, filename)
module_name = f"sidecar_rules_{filename[:-3]}"
try:
# Force re-import for hot-reload
if module_name in sys.modules:
del sys.modules[module_name]
spec = importlib.util.spec_from_file_location(module_name, filepath)
if spec is None or spec.loader is None:
continue
module = importlib.util.module_from_spec(spec)
sys.modules[module_name] = module
spec.loader.exec_module(module)
# Find all RulePlugin subclasses in the module
for attr_name in dir(module):
attr = getattr(module, attr_name)
if (
isinstance(attr, type)
and issubclass(attr, RulePlugin)
and attr is not RulePlugin
):
instance = attr()
self.plugins.append(instance)
_log(f"Loaded plugin: {instance.name} from {filename}")
except Exception as e:
_log(f"Error loading plugin {filename}: {e}")
_log(f"Total plugins loaded: {len(self.plugins)}")
return self.plugins
def match_rules(self, prompt: str, session_state: dict) -> str:
"""Run all plugins, return combined injection text for matching rules."""
injections = []
for plugin in self.plugins:
try:
if plugin.match(prompt, session_state):
text = plugin.inject()
if text:
injections.append(text)
_log(f"Rule fired: {plugin.name}")
except Exception as e:
_log(f"Error in plugin {plugin.name}.match/inject: {e}")
result = "\n\n".join(injections) if injections else ""
# Delta-dedup: suppress if identical to last injection
if result and result == self._last_injection:
_log(f"plugin-dedup: suppressed identical injection ({len(result)} chars)")
return ""
self._last_injection = result
return result
|