ec33ac1936
Based on claude-code-tools TmuxCLIController, this refactor: - Added DockerTmuxController class for robust tmux session management - Implements send_keys() with configurable delay_enter - Implements capture_pane() for output retrieval - Implements wait_for_prompt() for pattern-based completion detection - Implements wait_for_idle() for content-hash-based idle detection - Implements wait_for_shell_prompt() for shell prompt detection Also includes workflow improvements: - Pre-task git snapshot before agent execution - Post-task commit protocol in agent guidelines Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
384 lines
13 KiB
Python
384 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Plugin Skill Loader - Load and integrate plugin capabilities into Luzia skill system
|
|
|
|
Bridges Claude marketplace plugins with Luzia's skill matching and task dispatch system.
|
|
|
|
Features:
|
|
1. Load plugin capabilities as Luzia skills
|
|
2. Create skill metadata from plugin definitions
|
|
3. Integrate with responsive dispatcher
|
|
4. Cache skills for performance
|
|
5. Track plugin skill usage
|
|
6. Provide plugin-to-skill mapping
|
|
"""
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Dict, List, Optional, Any, Set
|
|
from datetime import datetime
|
|
import logging
|
|
from dataclasses import dataclass
|
|
|
|
from plugin_marketplace import (
|
|
PluginMarketplaceRegistry,
|
|
PluginCapabilityMatcher,
|
|
MarketplacePlugin,
|
|
PluginCapability
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclass
|
|
class PluginSkill:
|
|
"""Skill derived from a plugin capability"""
|
|
skill_id: str
|
|
name: str
|
|
description: str
|
|
plugin_id: str
|
|
plugin_name: str
|
|
capability_name: str
|
|
category: str
|
|
tags: List[str]
|
|
trust_level: str
|
|
keywords: List[str]
|
|
metadata: Dict[str, Any]
|
|
|
|
def to_dict(self):
|
|
return {
|
|
'skill_id': self.skill_id,
|
|
'name': self.name,
|
|
'description': self.description,
|
|
'plugin_id': self.plugin_id,
|
|
'plugin_name': self.plugin_name,
|
|
'capability_name': self.capability_name,
|
|
'category': self.category,
|
|
'tags': self.tags,
|
|
'trust_level': self.trust_level,
|
|
'keywords': self.keywords,
|
|
'metadata': self.metadata
|
|
}
|
|
|
|
|
|
class PluginSkillLoader:
|
|
"""
|
|
Loads plugin capabilities as Luzia skills
|
|
|
|
Converts plugin marketplace definitions into executable skills that can be
|
|
matched to tasks and integrated into the responsive dispatcher.
|
|
"""
|
|
|
|
def __init__(self, registry: Optional[PluginMarketplaceRegistry] = None,
|
|
cache_dir: Optional[Path] = None):
|
|
"""Initialize skill loader
|
|
|
|
Args:
|
|
registry: Plugin marketplace registry (created if not provided)
|
|
cache_dir: Directory for caching skills
|
|
"""
|
|
self.registry = registry or PluginMarketplaceRegistry()
|
|
self.cache_dir = cache_dir or Path("/tmp/.luzia-plugin-skills")
|
|
self.cache_dir.mkdir(parents=True, exist_ok=True)
|
|
self.skills: Dict[str, PluginSkill] = {}
|
|
self.skill_index: Dict[str, Set[str]] = {} # keyword -> skill_ids
|
|
self.category_index: Dict[str, Set[str]] = {} # category -> skill_ids
|
|
self.plugin_skill_map: Dict[str, List[str]] = {} # plugin_id -> skill_ids
|
|
self.matcher = PluginCapabilityMatcher(self.registry)
|
|
self.load_cache()
|
|
|
|
def load_cache(self) -> None:
|
|
"""Load cached skills from disk"""
|
|
cache_file = self.cache_dir / "skills.json"
|
|
if cache_file.exists():
|
|
try:
|
|
data = json.loads(cache_file.read_text())
|
|
self._rebuild_from_dict(data)
|
|
logger.info(f"Loaded {len(self.skills)} plugin skills from cache")
|
|
except Exception as e:
|
|
logger.warning(f"Failed to load skill cache: {e}")
|
|
|
|
def generate_skills_from_plugins(self) -> Dict[str, PluginSkill]:
|
|
"""Generate skills from all registered plugins
|
|
|
|
Returns:
|
|
Dict of generated skills
|
|
"""
|
|
skills = {}
|
|
plugins = self.registry.list_plugins()
|
|
|
|
for plugin in plugins:
|
|
plugin_skills = self._plugin_to_skills(plugin)
|
|
skills.update(plugin_skills)
|
|
self.plugin_skill_map[plugin.id] = list(plugin_skills.keys())
|
|
|
|
self.skills = skills
|
|
self._rebuild_indices()
|
|
self.save_cache()
|
|
logger.info(f"Generated {len(skills)} skills from {len(plugins)} plugins")
|
|
return skills
|
|
|
|
def _plugin_to_skills(self, plugin: MarketplacePlugin) -> Dict[str, PluginSkill]:
|
|
"""Convert plugin capabilities to skills
|
|
|
|
Args:
|
|
plugin: Plugin to convert
|
|
|
|
Returns:
|
|
Dict of skills keyed by skill_id
|
|
"""
|
|
skills = {}
|
|
|
|
for capability in plugin.capabilities:
|
|
skill_id = f"{plugin.id}:{capability.name}"
|
|
skill = PluginSkill(
|
|
skill_id=skill_id,
|
|
name=f"{capability.name} ({plugin.name})",
|
|
description=capability.description,
|
|
plugin_id=plugin.id,
|
|
plugin_name=plugin.name,
|
|
capability_name=capability.name,
|
|
category=capability.category,
|
|
tags=capability.tags,
|
|
trust_level=plugin.trust_level,
|
|
keywords=self._extract_keywords(capability),
|
|
metadata={
|
|
'plugin_url': plugin.url,
|
|
'plugin_vendor': plugin.vendor,
|
|
'plugin_version': plugin.version,
|
|
'requires_auth': getattr(capability, 'requires_auth', False)
|
|
}
|
|
)
|
|
skills[skill_id] = skill
|
|
|
|
return skills
|
|
|
|
def _extract_keywords(self, capability: PluginCapability) -> List[str]:
|
|
"""Extract keywords from capability for matching
|
|
|
|
Args:
|
|
capability: Plugin capability
|
|
|
|
Returns:
|
|
List of keywords
|
|
"""
|
|
keywords = list(capability.tags) + [capability.category]
|
|
|
|
# Add derived keywords from description
|
|
description_lower = capability.description.lower()
|
|
keyword_patterns = {
|
|
'security': ['secure', 'vulnerability', 'threat', 'exploit'],
|
|
'performance': ['speed', 'optimization', 'benchmark', 'latency'],
|
|
'analysis': ['analyze', 'inspect', 'examine', 'review'],
|
|
'code': ['code', 'coding', 'programming', 'developer'],
|
|
'integration': ['integrate', 'api', 'connect', 'interface']
|
|
}
|
|
|
|
for keyword, patterns in keyword_patterns.items():
|
|
for pattern in patterns:
|
|
if pattern in description_lower:
|
|
if keyword not in keywords:
|
|
keywords.append(keyword)
|
|
break
|
|
|
|
return keywords
|
|
|
|
def _rebuild_indices(self) -> None:
|
|
"""Rebuild keyword and category indices"""
|
|
self.skill_index = {}
|
|
self.category_index = {}
|
|
|
|
for skill_id, skill in self.skills.items():
|
|
# Index by keywords
|
|
for keyword in skill.keywords:
|
|
if keyword not in self.skill_index:
|
|
self.skill_index[keyword] = set()
|
|
self.skill_index[keyword].add(skill_id)
|
|
|
|
# Index by category
|
|
if skill.category not in self.category_index:
|
|
self.category_index[skill.category] = set()
|
|
self.category_index[skill.category].add(skill_id)
|
|
|
|
def find_skills_for_task(self, task_description: str,
|
|
min_relevance: float = 0.5) -> List[Dict[str, Any]]:
|
|
"""Find relevant plugin skills for a task
|
|
|
|
Args:
|
|
task_description: Description of the task
|
|
min_relevance: Minimum relevance score (0-1)
|
|
|
|
Returns:
|
|
List of matched skills with relevance info
|
|
"""
|
|
if not self.skills:
|
|
self.generate_skills_from_plugins()
|
|
|
|
matched_plugins = self.matcher.match_plugins(task_description, min_relevance)
|
|
matched_skills = []
|
|
|
|
for plugin_match in matched_plugins:
|
|
plugin_id = plugin_match['id']
|
|
if plugin_id in self.plugin_skill_map:
|
|
for skill_id in self.plugin_skill_map[plugin_id]:
|
|
if skill_id in self.skills:
|
|
skill = self.skills[skill_id]
|
|
matched_skills.append({
|
|
'skill_id': skill.skill_id,
|
|
'name': skill.name,
|
|
'description': skill.description,
|
|
'category': skill.category,
|
|
'plugin_id': skill.plugin_id,
|
|
'plugin_name': skill.plugin_name,
|
|
'relevance_score': plugin_match['relevance_score'],
|
|
'tags': skill.tags,
|
|
'trust_level': skill.trust_level
|
|
})
|
|
|
|
# Sort by relevance
|
|
matched_skills.sort(key=lambda x: x['relevance_score'], reverse=True)
|
|
return matched_skills
|
|
|
|
def get_skill(self, skill_id: str) -> Optional[PluginSkill]:
|
|
"""Get a skill by ID
|
|
|
|
Args:
|
|
skill_id: ID of the skill
|
|
|
|
Returns:
|
|
Skill or None if not found
|
|
"""
|
|
return self.skills.get(skill_id)
|
|
|
|
def list_skills(self, category: Optional[str] = None,
|
|
plugin_id: Optional[str] = None) -> List[PluginSkill]:
|
|
"""List available skills with optional filtering
|
|
|
|
Args:
|
|
category: Optional category filter
|
|
plugin_id: Optional plugin filter
|
|
|
|
Returns:
|
|
List of skills
|
|
"""
|
|
skills = list(self.skills.values())
|
|
|
|
if category:
|
|
skills = [s for s in skills if s.category == category]
|
|
|
|
if plugin_id:
|
|
skills = [s for s in skills if s.plugin_id == plugin_id]
|
|
|
|
return skills
|
|
|
|
def save_cache(self) -> None:
|
|
"""Save skills to cache"""
|
|
cache_file = self.cache_dir / "skills.json"
|
|
data = {
|
|
'timestamp': datetime.now().isoformat(),
|
|
'skill_count': len(self.skills),
|
|
'skills': {
|
|
skill_id: skill.to_dict()
|
|
for skill_id, skill in self.skills.items()
|
|
},
|
|
'plugin_skill_map': {
|
|
pid: list(sids)
|
|
for pid, sids in self.plugin_skill_map.items()
|
|
}
|
|
}
|
|
cache_file.write_text(json.dumps(data, indent=2))
|
|
logger.info(f"Saved {len(self.skills)} plugin skills to cache")
|
|
|
|
def _rebuild_from_dict(self, data: Dict[str, Any]) -> None:
|
|
"""Rebuild skills from cached data"""
|
|
for skill_id, skill_data in data.get('skills', {}).items():
|
|
skill = PluginSkill(
|
|
skill_id=skill_data['skill_id'],
|
|
name=skill_data['name'],
|
|
description=skill_data['description'],
|
|
plugin_id=skill_data['plugin_id'],
|
|
plugin_name=skill_data['plugin_name'],
|
|
capability_name=skill_data['capability_name'],
|
|
category=skill_data['category'],
|
|
tags=skill_data['tags'],
|
|
trust_level=skill_data['trust_level'],
|
|
keywords=skill_data['keywords'],
|
|
metadata=skill_data.get('metadata', {})
|
|
)
|
|
self.skills[skill_id] = skill
|
|
|
|
self.plugin_skill_map = {
|
|
pid: list(sids)
|
|
for pid, sids in data.get('plugin_skill_map', {}).items()
|
|
}
|
|
self._rebuild_indices()
|
|
|
|
def export_for_dispatcher(self) -> Dict[str, Any]:
|
|
"""Export skills in format suitable for responsive dispatcher
|
|
|
|
Returns:
|
|
Dict with dispatcher-compatible skill definitions
|
|
"""
|
|
return {
|
|
'source': 'plugin-marketplace',
|
|
'timestamp': datetime.now().isoformat(),
|
|
'skill_count': len(self.skills),
|
|
'skills': {
|
|
skill_id: {
|
|
'name': skill.name,
|
|
'description': skill.description,
|
|
'category': skill.category,
|
|
'keywords': skill.keywords,
|
|
'tags': skill.tags,
|
|
'plugin_id': skill.plugin_id,
|
|
'trust_level': skill.trust_level,
|
|
'metadata': skill.metadata
|
|
}
|
|
for skill_id, skill in self.skills.items()
|
|
},
|
|
'categories': list(self.category_index.keys()),
|
|
'plugin_count': len(self.plugin_skill_map)
|
|
}
|
|
|
|
def export_for_knowledge_graph(self) -> Dict[str, Any]:
|
|
"""Export skills for knowledge graph ingestion
|
|
|
|
Returns:
|
|
Dict suitable for knowledge graph storage
|
|
"""
|
|
skills_by_category = {}
|
|
for skill in self.skills.values():
|
|
if skill.category not in skills_by_category:
|
|
skills_by_category[skill.category] = []
|
|
skills_by_category[skill.category].append(skill.to_dict())
|
|
|
|
return {
|
|
'source': 'plugin-marketplace-skills',
|
|
'timestamp': datetime.now().isoformat(),
|
|
'total_skills': len(self.skills),
|
|
'skills_by_category': skills_by_category,
|
|
'plugins_used': len(self.plugin_skill_map),
|
|
'trust_distribution': self._get_trust_distribution()
|
|
}
|
|
|
|
def _get_trust_distribution(self) -> Dict[str, int]:
|
|
"""Get distribution of trust levels in skills"""
|
|
distribution: Dict[str, int] = {}
|
|
for skill in self.skills.values():
|
|
trust = skill.trust_level
|
|
distribution[trust] = distribution.get(trust, 0) + 1
|
|
return distribution
|
|
|
|
|
|
# Convenience functions
|
|
def get_plugin_skill_loader(registry: Optional[PluginMarketplaceRegistry] = None,
|
|
cache_dir: Optional[Path] = None) -> PluginSkillLoader:
|
|
"""Get or create plugin skill loader"""
|
|
return PluginSkillLoader(registry, cache_dir)
|
|
|
|
|
|
def generate_all_skills() -> Dict[str, PluginSkill]:
|
|
"""Generate all plugin skills"""
|
|
loader = get_plugin_skill_loader()
|
|
return loader.generate_skills_from_plugins()
|