#!/usr/bin/env python3 """ Component Activator - Runtime Component Management System
Loads components from JSON registries and makes them callable at runtime. Bridges the gap between component files (documentation) and runtime execution.
Part of Component Activation Infrastructure (Phase 2) Created: 2025-11-29 """
import json import logging from dataclasses import dataclass from pathlib import Path from typing import Dict, List, Optional, Any
logger = logging.getLogger(name)
@dataclass class AgentConfig: """Configuration for an agent""" name: str path: str description: str tags: List[str] tools: Optional[List[str]] = None category: Optional[str] = None use_cases: Optional[List[str]] = None version: str = "1.0.0" status: str = "operational"
@dataclass class SkillConfig: """Configuration for a skill""" name: str path: str description: str tags: List[str] version: str = "1.0.0" status: str = "operational"
@dataclass class CommandConfig: """Configuration for a command""" name: str path: str description: str tags: List[str] version: str = "1.0.0" status: str = "operational"
@dataclass class ScriptConfig: """Configuration for a script""" name: str path: str description: str tags: List[str] category: Optional[str] = None version: str = "1.0.0" status: str = "operational"
@dataclass class HookConfig: """Configuration for a git hook""" name: str path: str description: str tags: List[str] category: Optional[str] = None # pre-commit, pre-push, etc. version: str = "1.0.0" status: str = "operational"
@dataclass class PromptConfig: """Configuration for a prompt template""" name: str path: str description: str tags: List[str] version: str = "1.0.0" status: str = "operational"
class ComponentActivator: """ Component Activation System
Loads components from JSON registries and makes them available for runtime use.
Provides discovery, lookup, and activation capabilities for all component types.
"""
def __init__(self, framework_root: Optional[Path] = None):
"""Initialize component activator"""
self.framework_root = Path(framework_root) if framework_root else Path.cwd()
# Component registries
self.agents: Dict[str, AgentConfig] = {}
self.skills: Dict[str, SkillConfig] = {}
self.commands: Dict[str, CommandConfig] = {}
self.scripts: Dict[str, ScriptConfig] = {}
self.hooks: Dict[str, HookConfig] = {}
self.prompts: Dict[str, PromptConfig] = {}
# Statistics
self.total_loaded = 0
self.failed_loads = 0
def load_registry(self, registry_path: Path) -> int:
"""
Load components from a JSON registry file
Args:
registry_path: Path to registry JSON file
Returns:
Number of components loaded
"""
if not registry_path.exists():
logger.error(f"Registry file not found: {registry_path}")
return 0
try:
with open(registry_path, 'r') as f:
registry_data = json.load(f)
components = registry_data.get('components', {})
loaded_count = 0
def flatten_categories(data):
"""Flatten nested category/list structure to flat list.
Registry formats supported:
- {"total": N, "categories": {"cat1": [...], "cat2": [...]}}
- {"total": N, "list": [...]}
- Direct list: [...]
Returns: flat list of all items
"""
if isinstance(data, list):
return data # Already flat
if isinstance(data, dict):
# Try "list" format first (used by skills, scripts)
if 'list' in data:
return data['list'] if isinstance(data['list'], list) else []
# Try "categories" format (used by agents, commands)
categories = data.get('categories', {})
if categories:
flat = []
for cat_items in categories.values():
if isinstance(cat_items, list):
flat.extend(cat_items)
return flat
return []
# Load each component type (plural keys match registry format)
if 'agents' in components:
loaded_count += self._load_agents(flatten_categories(components['agents']))
if 'skills' in components:
loaded_count += self._load_skills(flatten_categories(components['skills']))
if 'commands' in components:
loaded_count += self._load_commands(flatten_categories(components['commands']))
if 'scripts' in components:
loaded_count += self._load_scripts(flatten_categories(components['scripts']))
if 'hooks' in components:
loaded_count += self._load_hooks(flatten_categories(components['hooks']))
if 'prompts' in components:
loaded_count += self._load_prompts(flatten_categories(components['prompts']))
self.total_loaded += loaded_count
logger.info(f"Loaded {loaded_count} components from {registry_path.name}")
return loaded_count
except Exception as e:
logger.error(f"Error loading registry {registry_path}: {e}")
return 0
def load_all_registries(self) -> int:
"""
Load all standard CODITECT registries
Returns:
Total number of components loaded
"""
registry_files = [
self.framework_root / "config" / "framework-registry.json",
self.framework_root / "skills" / "REGISTRY.json",
]
total = 0
for registry_path in registry_files:
if registry_path.exists():
total += self.load_registry(registry_path)
else:
logger.warning(f"Registry not found: {registry_path}")
logger.info(f"Total components loaded: {total}")
return total
def _load_agents(self, agents_data: List[Dict]) -> int:
"""Load agents from registry data"""
count = 0
for agent_dict in agents_data:
try:
# Derive path from id if not present
agent_id = agent_dict.get('id', agent_dict.get('name', '').lower().replace(' ', '-'))
path = agent_dict.get('path', f"agents/{agent_id}.md")
agent = AgentConfig(
name=agent_dict.get('name', agent_id),
path=path,
description=agent_dict.get('description', ''),
tags=agent_dict.get('tags', []),
tools=agent_dict.get('tools'),
category=agent_dict.get('category'),
use_cases=agent_dict.get('use_cases'),
version=agent_dict.get('version', '1.0.0'),
status=agent_dict.get('status', 'operational')
)
# Index by both title-case name AND kebab-case id for flexible lookup
self.agents[agent.name] = agent
if agent_id != agent.name:
self.agents[agent_id] = agent # Also index by id
count += 1
logger.debug(f"Loaded agent: {agent.name} (id: {agent_id})")
except Exception as e:
logger.error(f"Error loading agent {agent_dict.get('name', 'unknown')}: {e}")
self.failed_loads += 1
return count
def _load_skills(self, skills_data: List[Dict]) -> int:
"""Load skills from registry data"""
count = 0
for skill_dict in skills_data:
try:
# Derive path from id if not present
skill_id = skill_dict.get('id', skill_dict.get('name', '').lower().replace(' ', '-'))
path = skill_dict.get('path', f"skills/{skill_id}/SKILL.md")
skill = SkillConfig(
name=skill_dict.get('name', skill_id),
path=path,
description=skill_dict.get('description', ''),
tags=skill_dict.get('tags', []),
version=skill_dict.get('version', '1.0.0'),
status=skill_dict.get('status', 'operational')
)
# Index by both name and id for flexible lookup
self.skills[skill.name] = skill
if skill_id != skill.name:
self.skills[skill_id] = skill
count += 1
logger.debug(f"Loaded skill: {skill.name}")
except Exception as e:
logger.error(f"Error loading skill {skill_dict.get('name', 'unknown')}: {e}")
self.failed_loads += 1
return count
def _load_commands(self, commands_data: List[Dict]) -> int:
"""Load commands from registry data"""
count = 0
for command_dict in commands_data:
try:
# Derive path from id if not present
command_id = command_dict.get('id', command_dict.get('name', '').lower().replace(' ', '-').replace('/', ''))
path = command_dict.get('path', f"commands/{command_id}.md")
command = CommandConfig(
name=command_dict.get('name', command_id),
path=path,
description=command_dict.get('description', ''),
tags=command_dict.get('tags', []),
version=command_dict.get('version', '1.0.0'),
status=command_dict.get('status', 'operational')
)
# Index by both name and id for flexible lookup
self.commands[command.name] = command
if command_id != command.name:
self.commands[command_id] = command
count += 1
logger.debug(f"Loaded command: {command.name}")
except Exception as e:
logger.error(f"Error loading command {command_dict.get('name', 'unknown')}: {e}")
self.failed_loads += 1
return count
def _load_scripts(self, scripts_data: List[Dict]) -> int:
"""Load scripts from registry data"""
count = 0
for script_dict in scripts_data:
try:
# Scripts usually have path, but derive from id/name as fallback
script_id = script_dict.get('id', script_dict.get('name', '').lower().replace(' ', '-'))
name = script_dict.get('name', script_id)
# Scripts can be .py or .sh
path = script_dict.get('path', f"scripts/{script_id}.py")
script = ScriptConfig(
name=name,
path=path,
description=script_dict.get('description', ''),
tags=script_dict.get('tags', []),
category=script_dict.get('category'),
version=script_dict.get('version', '1.0.0'),
status=script_dict.get('status', 'operational')
)
# Index by both name and id for flexible lookup
self.scripts[script.name] = script
if script_id != script.name:
self.scripts[script_id] = script
count += 1
logger.debug(f"Loaded script: {script.name}")
except Exception as e:
logger.error(f"Error loading script {script_dict.get('name', 'unknown')}: {e}")
self.failed_loads += 1
return count
def _load_hooks(self, hooks_data: List[Dict]) -> int:
"""Load hooks from registry data"""
count = 0
for hook_dict in hooks_data:
try:
# Derive path from id if not present
hook_id = hook_dict.get('id', hook_dict.get('name', '').lower().replace(' ', '-'))
path = hook_dict.get('path', f"hooks/{hook_id}.md")
hook = HookConfig(
name=hook_dict.get('name', hook_id),
path=path,
description=hook_dict.get('description', ''),
tags=hook_dict.get('tags', []),
category=hook_dict.get('category'),
version=hook_dict.get('version', '1.0.0'),
status=hook_dict.get('status', 'operational')
)
# Index by both name and id for flexible lookup
self.hooks[hook.name] = hook
if hook_id != hook.name:
self.hooks[hook_id] = hook
count += 1
logger.debug(f"Loaded hook: {hook.name}")
except Exception as e:
logger.error(f"Error loading hook {hook_dict.get('name', 'unknown')}: {e}")
self.failed_loads += 1
return count
def _load_prompts(self, prompts_data: List[Dict]) -> int:
"""Load prompts from registry data"""
count = 0
for prompt_dict in prompts_data:
try:
# Derive path from id if not present
prompt_id = prompt_dict.get('id', prompt_dict.get('name', '').lower().replace(' ', '-'))
path = prompt_dict.get('path', f"prompts/{prompt_id}.md")
prompt = PromptConfig(
name=prompt_dict.get('name', prompt_id),
path=path,
description=prompt_dict.get('description', ''),
tags=prompt_dict.get('tags', []),
version=prompt_dict.get('version', '1.0.0'),
status=prompt_dict.get('status', 'operational')
)
# Index by both name and id for flexible lookup
self.prompts[prompt.name] = prompt
if prompt_id != prompt.name:
self.prompts[prompt_id] = prompt
count += 1
logger.debug(f"Loaded prompt: {prompt.name}")
except Exception as e:
logger.error(f"Error loading prompt {prompt_dict.get('name', 'unknown')}: {e}")
self.failed_loads += 1
return count
# Component Discovery Methods
def get_agent(self, name: str) -> Optional[AgentConfig]:
"""Get agent configuration by name"""
return self.agents.get(name)
def get_skill(self, name: str) -> Optional[SkillConfig]:
"""Get skill configuration by name"""
return self.skills.get(name)
def get_command(self, name: str) -> Optional[CommandConfig]:
"""Get command configuration by name"""
return self.commands.get(name)
def get_script(self, name: str) -> Optional[ScriptConfig]:
"""Get script configuration by name"""
return self.scripts.get(name)
def get_hook(self, name: str) -> Optional[HookConfig]:
"""Get hook configuration by name"""
return self.hooks.get(name)
def get_prompt(self, name: str) -> Optional[PromptConfig]:
"""Get prompt configuration by name"""
return self.prompts.get(name)
def get_component(self, component_type: str, name: str) -> Optional[Any]:
"""
Get component of any type by name
Args:
component_type: Type of component (agent, skill, command, script, hook, prompt)
name: Component name
Returns:
Component configuration or None
"""
type_map = {
'agent': self.agents,
'skill': self.skills,
'command': self.commands,
'script': self.scripts,
'hook': self.hooks,
'prompt': self.prompts
}
registry = type_map.get(component_type)
if registry is None:
logger.error(f"Unknown component type: {component_type}")
return None
return registry.get(name)
# List Methods
def list_agents(self, category: Optional[str] = None, tag: Optional[str] = None) -> List[AgentConfig]:
"""List all agents, optionally filtered by category or tag"""
agents = list(self.agents.values())
if category:
agents = [a for a in agents if a.category == category]
if tag:
agents = [a for a in agents if tag in a.tags]
return agents
def list_skills(self, tag: Optional[str] = None) -> List[SkillConfig]:
"""List all skills, optionally filtered by tag"""
skills = list(self.skills.values())
if tag:
skills = [s for s in skills if tag in s.tags]
return skills
def list_commands(self, tag: Optional[str] = None) -> List[CommandConfig]:
"""List all commands, optionally filtered by tag"""
commands = list(self.commands.values())
if tag:
commands = [c for c in commands if tag in c.tags]
return commands
def list_scripts(self, category: Optional[str] = None) -> List[ScriptConfig]:
"""List all scripts, optionally filtered by category"""
scripts = list(self.scripts.values())
if category:
scripts = [s for s in scripts if s.category == category]
return scripts
def list_hooks(self, category: Optional[str] = None) -> List[HookConfig]:
"""List all hooks, optionally filtered by category (hook phase)"""
hooks = list(self.hooks.values())
if category:
hooks = [h for h in hooks if h.category == category]
return hooks
def list_prompts(self) -> List[PromptConfig]:
"""List all prompts"""
return list(self.prompts.values())
def list_components(self, component_type: str) -> List[Any]:
"""
List all components of a specific type
Args:
component_type: Type of component (agent, skill, command, script, hook, prompt)
Returns:
List of component configurations
"""
type_map = {
'agent': self.list_agents,
'skill': self.list_skills,
'command': self.list_commands,
'script': self.list_scripts,
'hook': self.list_hooks,
'prompt': self.list_prompts
}
list_func = type_map.get(component_type)
if list_func is None:
logger.error(f"Unknown component type: {component_type}")
return []
return list_func()
# Search Methods
def search_agents(self, query: str) -> List[AgentConfig]:
"""Search agents by name or description"""
query_lower = query.lower()
results = []
for agent in self.agents.values():
if (query_lower in agent.name.lower() or
query_lower in agent.description.lower() or
any(query_lower in tag.lower() for tag in agent.tags)):
results.append(agent)
return results
def search_components(self, query: str, component_type: Optional[str] = None) -> Dict[str, List[Any]]:
"""
Search all components or specific type
Args:
query: Search query string
component_type: Optional component type to restrict search
Returns:
Dictionary mapping component types to matching components
"""
query_lower = query.lower()
results: Dict[str, List[Any]] = {}
def matches_query(component: Any) -> bool:
"""Check if component matches query"""
return (query_lower in component.name.lower() or
query_lower in component.description.lower() or
any(query_lower in tag.lower() for tag in component.tags))
if component_type is None or component_type == 'agent':
results['agent'] = [a for a in self.agents.values() if matches_query(a)]
if component_type is None or component_type == 'skill':
results['skill'] = [s for s in self.skills.values() if matches_query(s)]
if component_type is None or component_type == 'command':
results['command'] = [c for c in self.commands.values() if matches_query(c)]
if component_type is None or component_type == 'script':
results['script'] = [s for s in self.scripts.values() if matches_query(s)]
if component_type is None or component_type == 'hook':
results['hook'] = [h for h in self.hooks.values() if matches_query(h)]
if component_type is None or component_type == 'prompt':
results['prompt'] = [p for p in self.prompts.values() if matches_query(p)]
return results
# Activation Methods
def activate_all(self) -> int:
"""
Activate all loaded components
This method prepares all components for runtime use.
For now, it's a placeholder - actual activation happens via integration
with Orchestrator, AgentDispatcher, etc.
Returns:
Number of components activated
"""
total = (len(self.agents) + len(self.skills) + len(self.commands) +
len(self.scripts) + len(self.hooks) + len(self.prompts))
logger.info(f"Activated {total} components:")
logger.info(f" - {len(self.agents)} agents")
logger.info(f" - {len(self.skills)} skills")
logger.info(f" - {len(self.commands)} commands")
logger.info(f" - {len(self.scripts)} scripts")
logger.info(f" - {len(self.hooks)} hooks")
logger.info(f" - {len(self.prompts)} prompts")
return total
def get_statistics(self) -> Dict[str, Any]:
"""Get component statistics"""
return {
'total_loaded': self.total_loaded,
'failed_loads': self.failed_loads,
'agents': len(self.agents),
'skills': len(self.skills),
'commands': len(self.commands),
'scripts': len(self.scripts),
'hooks': len(self.hooks),
'prompts': len(self.prompts),
'total_active': (len(self.agents) + len(self.skills) + len(self.commands) +
len(self.scripts) + len(self.hooks) + len(self.prompts))
}
def print_summary(self) -> None:
"""Print component activation summary"""
stats = self.get_statistics()
print("\n=== Component Activator Summary ===")
print(f"Total Components Loaded: {stats['total_loaded']}")
print(f"Failed Loads: {stats['failed_loads']}")
print(f"\nBreakdown:")
print(f" Agents: {stats['agents']}")
print(f" Skills: {stats['skills']}")
print(f" Commands: {stats['commands']}")
print(f" Scripts: {stats['scripts']}")
print(f" Hooks: {stats['hooks']}")
print(f" Prompts: {stats['prompts']}")
print(f"\nTotal Active: {stats['total_active']}")
Convenience function for quick activation
def activate_framework(framework_root: Optional[Path] = None) -> ComponentActivator: """ Quick activation of CODITECT framework
Args:
framework_root: Root directory of framework (default: current directory)
Returns:
ComponentActivator instance with all components loaded
"""
activator = ComponentActivator(framework_root)
activator.load_all_registries()
activator.activate_all()
return activator
if name == "main": # Test activation logging.basicConfig(level=logging.INFO)
activator = activate_framework()
activator.print_summary()
# Test search
print("\n=== Search Test: 'git' ===")
results = activator.search_components('git')
for comp_type, components in results.items():
if components:
print(f"\n{comp_type.upper()}:")
for comp in components[:3]: # Show first 3
print(f" - {comp.name}: {comp.description[:80]}...")