Skip to main content

#!/usr/bin/env python3 """ Ensure Component Registered - Complete Component Registration System

Ensures all components are:

  1. Indexed in SQLite (platform.db - ADR-118 Tier 1)
  2. Registered in framework-registry.json
  3. Activated in component-activation-status.json

Part of CODITECT Component Activation Infrastructure Created: 2025-12-29 """

import json import re import sqlite3 from datetime import datetime, timezone from pathlib import Path from typing import Dict, List, Optional, Set, Tuple import argparse import sys

Paths

FRAMEWORK_ROOT = Path(file).parent.parent.parent REGISTRY_PATH = FRAMEWORK_ROOT / "config" / "framework-registry.json" ACTIVATION_PATH = FRAMEWORK_ROOT / ".coditect" / "component-activation-status.json" ACTIVATION_FALLBACK = FRAMEWORK_ROOT / "config" / "component-activation-status.json"

ADR-114 & ADR-118: Use centralized path discovery

sys.path.insert(0, str(Path(file).parent)) try: from paths import get_platform_db_path, PLATFORM_DB DB_PATH = PLATFORM_DB # Component data goes to platform.db (Tier 1) except ImportError: # Fallback for backward compatibility _user_data = Path.home() / "PROJECTS" / ".coditect-data" / "context-storage" if _user_data.exists(): PLATFORM_DB = _user_data / "platform.db" else: PLATFORM_DB = Path.home() / ".coditect" / "context-storage" / "platform.db" DB_PATH = PLATFORM_DB # Backward compatibility alias

Component type configurations

COMPONENT_CONFIGS = { "agent": { "dir": "agents", "patterns": [".md"], "recursive": False, "exclude_files": ["README.md", "INDEX.md"], "exclude_dirs": [], "registry_key": "agents", "path_template": "agents/{id}.md", }, "command": { "dir": "commands", "patterns": [".md"], "recursive": False, "exclude_files": ["README.md", "INDEX.md", "GUIDE.md", "COMMAND-GUIDE.md"], "exclude_dirs": [], "registry_key": "commands", "path_template": "commands/{id}.md", }, "skill": { "dir": "skills", "patterns": ["/SKILL.md"], "recursive": False, "exclude_files": [], "exclude_dirs": [], "registry_key": "skills", "path_template": "skills/{id}/SKILL.md", }, "script": { "dir": "scripts", "patterns": ["**/.py", "**/.sh"], "recursive": True, "exclude_files": ["init.py"], "exclude_dirs": ["pycache", "tests", ".git", "venv", ".venv"], "registry_key": "scripts", "path_template": "scripts/{id}.py", }, "hook": { "dir": "hooks", "patterns": [".md", ".sh", ".py"], "recursive": False, "exclude_files": ["README.md", "INDEX.md", "HOOKS-INDEX.md", "PHASE2-3-ADVANCED-HOOKS.md", "init.py"], "exclude_dirs": ["pycache"], "registry_key": "hooks", "path_template": "hooks/{id}.md", }, }

def discover_filesystem_components() -> Dict[str, Set[str]]: """Discover all components on filesystem including nested directories.""" components = {}

for comp_type, config in COMPONENT_CONFIGS.items():
    comp_dir = FRAMEWORK_ROOT / config["dir"]
    if not comp_dir.exists():
        components[comp_type] = set()
        continue

    files = set()
    exclude_files = set(config.get("exclude_files", []))
    exclude_dirs = set(config.get("exclude_dirs", []))

    for pattern in config["patterns"]:
        for f in comp_dir.glob(pattern):
            # Skip excluded directories
            if any(p in exclude_dirs for p in f.parts):
                continue
            # Skip excluded files
            if f.name in exclude_files:
                continue
            # Skip test files for scripts
            if comp_type == "script":
                if f.stem.startswith("test_") or "/tests/" in str(f):
                    continue

            # Extract ID from filename
            if comp_type == "skill":
                comp_id = f.parent.name
            else:
                comp_id = f.stem
            files.add(comp_id)

    components[comp_type] = files

return components

def get_indexed_components() -> Dict[str, Set[str]]: """Get components indexed in SQLite.""" if not DB_PATH.exists(): return {t: set() for t in COMPONENT_CONFIGS}

conn = sqlite3.connect(DB_PATH)
cursor = conn.cursor()

components = {t: set() for t in COMPONENT_CONFIGS}

try:
    cursor.execute("SELECT type, id FROM components")
    for row in cursor.fetchall():
        comp_type = row[0]
        comp_id = row[1]
        if comp_type in components:
            components[comp_type].add(comp_id)
except sqlite3.OperationalError:
    pass  # Table doesn't exist

conn.close()
return components

def get_registered_components() -> Dict[str, Set[str]]: """Get components registered in framework-registry.json.""" if not REGISTRY_PATH.exists(): return {t: set() for t in COMPONENT_CONFIGS}

with open(REGISTRY_PATH, 'r') as f:
    registry = json.load(f)

components = {t: set() for t in COMPONENT_CONFIGS}
reg_components = registry.get("components", {})

for comp_type, config in COMPONENT_CONFIGS.items():
    key = config["registry_key"]
    if key not in reg_components:
        continue

    data = reg_components[key]

    # Handle both list and categories formats
    items = []
    if isinstance(data, dict):
        if "list" in data:
            items = data["list"]
        elif "categories" in data:
            for cat_items in data["categories"].values():
                if isinstance(cat_items, list):
                    items.extend(cat_items)
    elif isinstance(data, list):
        items = data

    for item in items:
        if isinstance(item, dict):
            comp_id = item.get("id", item.get("name", "").lower().replace(" ", "-"))
            components[comp_type].add(comp_id)

return components

def get_activated_components() -> Dict[str, Set[str]]: """Get components activated in component-activation-status.json.""" activation_path = ACTIVATION_PATH if ACTIVATION_PATH.exists() else ACTIVATION_FALLBACK

if not activation_path.exists():
    return {t: set() for t in COMPONENT_CONFIGS}

with open(activation_path, 'r') as f:
    activation = json.load(f)

components = {t: set() for t in COMPONENT_CONFIGS}

for comp in activation.get("components", []):
    comp_type = comp.get("type")
    comp_name = comp.get("name", "").lower().replace(" ", "-")
    comp_path = comp.get("path", "")

    if comp_type in components and comp.get("activated", False):
        # Add normalized name
        components[comp_type].add(comp_name)

        # Also extract and add the file stem (id) from path for matching
        # e.g., "agents/git-workflow-orchestrator.md" -> "git-workflow-orchestrator"
        if comp_path:
            from pathlib import Path
            path_stem = Path(comp_path).stem
            # For skills, the id is the parent directory name
            if comp_type == "skill" and path_stem == "SKILL":
                path_stem = Path(comp_path).parent.name
            components[comp_type].add(path_stem)

return components

def extract_component_metadata(comp_type: str, comp_id: str) -> Dict: """Extract metadata from component file.""" config = COMPONENT_CONFIGS[comp_type]

# Find the file
if comp_type == "skill":
    file_path = FRAMEWORK_ROOT / config["dir"] / comp_id / "SKILL.md"
elif comp_type == "script":
    file_path = FRAMEWORK_ROOT / config["dir"] / f"{comp_id}.py"
    if not file_path.exists():
        file_path = FRAMEWORK_ROOT / config["dir"] / "core" / f"{comp_id}.py"
else:
    file_path = FRAMEWORK_ROOT / config["dir"] / f"{comp_id}.md"

if not file_path.exists():
    return {"id": comp_id, "name": comp_id.replace("-", " ").title(), "description": ""}

content = file_path.read_text(errors='ignore')

# Extract name from first heading
name_match = re.search(r'^#\s+(.+)$', content, re.MULTILINE)
name = name_match.group(1).strip() if name_match else comp_id.replace("-", " ").title()

# Extract description from first paragraph after heading
desc_match = re.search(r'^#.+\n\n(.+?)(?:\n\n|\n#)', content, re.MULTILINE | re.DOTALL)
description = desc_match.group(1).strip()[:200] if desc_match else ""

# Clean up description
description = re.sub(r'\*\*|\*|`', '', description)
description = re.sub(r'\[([^\]]+)\]\([^)]+\)', r'\1', description)

return {
    "id": comp_id,
    "name": name,
    "description": description,
}

def add_to_registry(comp_type: str, comp_id: str, metadata: Dict) -> bool: """Add component to framework-registry.json.""" if not REGISTRY_PATH.exists(): return False

with open(REGISTRY_PATH, 'r') as f:
    registry = json.load(f)

config = COMPONENT_CONFIGS[comp_type]
key = config["registry_key"]

if "components" not in registry:
    registry["components"] = {}

if key not in registry["components"]:
    registry["components"][key] = {"total": 0, "list": []}

data = registry["components"][key]

# Determine where to add (list or categories)
new_item = {
    "id": comp_id,
    "name": metadata.get("name", comp_id),
    "description": metadata.get("description", ""),
    "tags": [],
}

if "list" in data:
    # Check if already exists
    existing_ids = {item.get("id") for item in data["list"] if isinstance(item, dict)}
    if comp_id not in existing_ids:
        data["list"].append(new_item)
        data["total"] = len(data["list"])
elif "categories" in data:
    # Add to "general" category
    if "general" not in data["categories"]:
        data["categories"]["general"] = []
    existing_ids = {item.get("id") for item in data["categories"]["general"] if isinstance(item, dict)}
    if comp_id not in existing_ids:
        data["categories"]["general"].append(new_item)
        # Update total
        total = sum(len(cat) for cat in data["categories"].values() if isinstance(cat, list))
        data["total"] = total

# Update timestamp
registry["last_updated"] = datetime.now(timezone.utc).isoformat()

with open(REGISTRY_PATH, 'w') as f:
    json.dump(registry, f, indent=2)

return True

def add_to_activation(comp_type: str, comp_id: str, metadata: Dict) -> bool: """Add component to component-activation-status.json.""" activation_path = ACTIVATION_PATH if ACTIVATION_PATH.exists() else ACTIVATION_FALLBACK

if not activation_path.exists():
    return False

with open(activation_path, 'r') as f:
    activation = json.load(f)

config = COMPONENT_CONFIGS[comp_type]

# Check if already exists
existing = {(c.get("type"), c.get("name").lower().replace(" ", "-"))
            for c in activation.get("components", [])
            if c.get("name")}

if (comp_type, comp_id) in existing:
    return False

new_component = {
    "type": comp_type,
    "name": metadata.get("name", comp_id),
    "path": config["path_template"].format(id=comp_id),
    "activated": True,
    "version": "1.0.0",
    "status": "operational",
    "reason": "Auto-registered by pre-commit hook",
    "activated_at": datetime.now(timezone.utc).isoformat(),
}

if "components" not in activation:
    activation["components"] = []

activation["components"].append(new_component)

# Update summary
if "activation_summary" in activation:
    activation["activation_summary"]["total_components"] = len(activation["components"])
    activation["activation_summary"]["activated"] = sum(
        1 for c in activation["components"] if c.get("activated")
    )
    activation["activation_summary"]["last_updated"] = datetime.now(timezone.utc).isoformat()

with open(activation_path, 'w') as f:
    json.dump(activation, f, indent=2)

return True

def ensure_all_registered(quiet: bool = False, dry_run: bool = False) -> Dict: """Ensure all filesystem components are indexed, registered, and activated."""

def log(msg):
    if not quiet:
        print(msg)

fs_components = discover_filesystem_components()
indexed = get_indexed_components()
registered = get_registered_components()
activated = get_activated_components()

stats = {
    "checked": 0,
    "already_complete": 0,
    "newly_registered": 0,
    "newly_activated": 0,
    "errors": 0,
    "by_type": {},
}

for comp_type, fs_ids in fs_components.items():
    type_stats = {"registered": 0, "activated": 0}

    for comp_id in fs_ids:
        stats["checked"] += 1

        is_registered = comp_id in registered.get(comp_type, set())
        is_activated = comp_id in activated.get(comp_type, set())

        if is_registered and is_activated:
            stats["already_complete"] += 1
            continue

        # Extract metadata for new components
        metadata = extract_component_metadata(comp_type, comp_id)

        if not is_registered:
            if dry_run:
                log(f"  Would register: {comp_type}/{comp_id}")
            else:
                try:
                    add_to_registry(comp_type, comp_id, metadata)
                    stats["newly_registered"] += 1
                    type_stats["registered"] += 1
                    log(f"  Registered: {comp_type}/{comp_id}")
                except Exception as e:
                    log(f"  Error registering {comp_type}/{comp_id}: {e}")
                    stats["errors"] += 1

        if not is_activated:
            if dry_run:
                log(f"  Would activate: {comp_type}/{comp_id}")
            else:
                try:
                    add_to_activation(comp_type, comp_id, metadata)
                    stats["newly_activated"] += 1
                    type_stats["activated"] += 1
                    log(f"  Activated: {comp_type}/{comp_id}")
                except Exception as e:
                    log(f"  Error activating {comp_type}/{comp_id}: {e}")
                    stats["errors"] += 1

    stats["by_type"][comp_type] = type_stats

return stats

def main(): parser = argparse.ArgumentParser( description="Ensure all components are indexed, registered, and activated" ) parser.add_argument( "--quiet", "-q", action="store_true", help="Suppress output" ) parser.add_argument( "--dry-run", "-n", action="store_true", help="Show what would be done without making changes" ) parser.add_argument( "--check-only", "-c", action="store_true", help="Only check and report, don't make changes" )

args = parser.parse_args()

if not args.quiet:
    print("=" * 60)
    print("CODITECT Component Registration Check")
    print("=" * 60)

    if args.dry_run:
        print("MODE: Dry run (no changes)")
    elif args.check_only:
        print("MODE: Check only")
    else:
        print("MODE: Full registration")
    print()

stats = ensure_all_registered(
    quiet=args.quiet,
    dry_run=args.dry_run or args.check_only
)

if not args.quiet:
    print()
    print("=" * 60)
    print("Summary")
    print("=" * 60)
    print(f"Components checked: {stats['checked']}")
    print(f"Already complete: {stats['already_complete']}")
    print(f"Newly registered: {stats['newly_registered']}")
    print(f"Newly activated: {stats['newly_activated']}")
    print(f"Errors: {stats['errors']}")

# Exit with error if there are unregistered components in check-only mode
if args.check_only:
    unregistered = stats['checked'] - stats['already_complete']
    if unregistered > 0:
        if not args.quiet:
            print(f"\n⚠️  {unregistered} components need registration")
        sys.exit(1)

sys.exit(0)

if name == "main": main()