Aim Training Sensitivity Converter Skill Registry

Skill Registry System Documentation

Overview

The Skill Registry System provides the foundational architecture for managing, discovering, executing, and monitoring all skills within the Aim Training & Sensitivity Converter harness. This system implements a modular, extensible pattern that supports dynamic skill loading, dependency resolution, and comprehensive validation.

Architecture

Core Components

┌─────────────────────────────────────────────────────────────────┐
│                    Skill Registry System                        │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  ┌─────────────────┐      ┌─────────────────┐                  │
│  │ Skill Registry  │◄─────│  Skill Loader   │                  │
│  │   (Singleton)   │      │   (Dynamic)     │                  │
│  └────────┬────────┘      └─────────────────┘                  │
│           │                                                       │
│           ├──► Skill Discovery                                  │
│           ├──► Skill Registration                               │
│           ├──► Dependency Resolution                            │
│           └──► Skill Execution                                 │
│                                                                 │
│  ┌─────────────────┐      ┌─────────────────┐                  │
│  │ Hooks Manager   │◄─────│ Event Emitter   │                  │
│  │   (Lifecycle)   │      │   (Events)       │                  │
│  └─────────────────┘      └─────────────────┘                  │
│                                                                 │
│  ┌─────────────────┐      ┌─────────────────┐                  │
│  │ Quality Gate    │◄─────│  Validator      │                  │
│  │   (Enforcer)    │      │   (Schema)       │                  │
│  └─────────────────┘      └─────────────────┘                  │
└─────────────────────────────────────────────────────────────────┘

Skill Lifecycle States

UNLOADED → DISCOVERED → REGISTERED → VALIDATED → ACTIVE → EXECUTING
   ▲                                               │
   └─────────────────── RETIRED ◄────────────────┘

Skill Registration Protocol

Registration Schema

Each skill must conform to the following registration schema:

{
  "skill_id": "string (unique identifier)",
  "name": "string (human-readable name)",
  "version": "string (semantic version)",
  "description": "string (when to trigger, what it does)",
  "category": "enum [main, sub, utility, tool]",
  "priority": "number (0-1000, execution order)",
  "dependencies": ["array of skill_ids"],
  "input_schema": {
    "type": "object",
    "properties": {},
    "required": []
  },
  "output_schema": {
    "type": "object",
    "properties": {},
    "required": []
  },
  "quality_gates": ["array of gate IDs"],
  "metadata": {
    "author": "string",
    "created_at": "ISO datetime",
    "updated_at": "ISO datetime",
    "tags": ["array of strings"],
    "compatibility": {
      "min_version": "string",
      "max_version": "string"
    }
  }
}

Registration Process

1. Discovery Phase

• Scan skills directory for .md files with YAML frontmatter
• Parse frontmatter for metadata
• Validate basic schema compliance

1. Validation Phase

• Verify skill_id uniqueness
• Check dependency satisfaction
• Validate input/output schemas
• Confirm quality gate definitions

1. Registration Phase

• Add to registry index
• Create skill instance
• Register hooks
• Initialize monitoring

Skill Definition Format

Skills are defined in markdown files with YAML frontmatter:

---
name: skill-name
description: When to trigger and what the skill does
version: "1.0.0"
category: sub
priority: 100
dependencies: []
quality_gates: ["G1", "G2"]
---

## Role & Persona
[Role definition]

## Workflow (Harness Flow)
[Execution workflow]

## Tools
[Required tools]

## Output Format
[Output template]

## Quality Gates
[Quality gate definitions]

Skill Resolution & Execution

Dependency Resolution

The registry uses topological sorting to resolve skill dependencies:

def resolve_execution_order(skills: List[Skill]) -> List[Skill]:
    """
    Resolve execution order using topological sort.

    Args:
        skills: List of skills to execute

    Returns:
        Ordered list of skills with dependencies satisfied

    Raises:
        CircularDependencyError: If circular dependencies detected
    """
    graph = build_dependency_graph(skills)
    return topological_sort(graph)

Execution Modes

The registry supports multiple execution modes:

1. Sequential Mode: Execute skills one at a time in dependency order
2. Parallel Mode: Execute independent skills concurrently
3. Streaming Mode: Execute skills as soon as dependencies are satisfied
4. Interactive Mode: Pause between skills for user confirmation

Execution Flow

┌─────────────────┐
│  Skill Trigger  │
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│ Pre-Flight Hook │
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│ Resolve Deps    │
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│ Execute Skills  │◄─────┐
└────────┬────────┘      │
         │               │
         ▼               │
┌─────────────────┐      │
│ Quality Gates   │──────┘
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│ Post-Exec Hook  │
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│    Output       │
└─────────────────┘

Skill I/O Schemas

Input Schema Standard

All skills must declare their input schema using JSON Schema format:

{
  "type": "object",
  "properties": {
    "user_input": {
      "type": "string",
      "description": "Raw user input message"
    },
    "context": {
      "type": "object",
      "properties": {
        "language": {"type": "string", "enum": ["en", "vi"]},
        "session_id": {"type": "string"},
        "timestamp": {"type": "string", "format": "date-time"}
      }
    },
    "parameters": {
      "type": "object",
      "description": "Skill-specific parameters"
    }
  },
  "required": ["user_input"]
}

Output Schema Standard

All skills must produce output conforming to:

{
  "type": "object",
  "properties": {
    "status": {
      "type": "string",
      "enum": ["success", "partial", "failed", "degraded"]
    },
    "result": {
      "type": "object",
      "description": "Primary skill output data"
    },
    "metadata": {
      "type": "object",
      "properties": {
        "execution_time_ms": {"type": "number"},
        "token_count": {"type": "number"},
        "quality_gates_passed": {"type": "array", "items": {"type": "string"}},
        "warnings": {"type": "array", "items": {"type": "string"}}
      }
    },
    "next_actions": {
      "type": "array",
      "items": {
        "type": "object",
        "properties": {
          "skill_id": {"type": "string"},
          "reason": {"type": "string"}
        }
      }
    }
  },
  "required": ["status", "result"]
}

Quality Gates Integration

Gate Registration

Quality gates are registered separately and linked to skills:

class QualityGate:
    def __init__(self, gate_id: str, description: str,
                 check_fn: Callable, auto_fix_fn: Optional[Callable] = None):
        self.gate_id = gate_id
        self.description = description
        self.check_fn = check_fn
        self.auto_fix_fn = auto_fix_fn

Gate Enforcement

The registry enforces quality gates in two phases:

1. Pre-Execution: Validate input against gate requirements
2. Post-Execution: Validate output against gate standards

def enforce_quality_gates(skill: Skill, input_data: Dict,
                         output_data: Dict) -> GateResult:
    """
    Enforce quality gates for a skill execution.

    Args:
        skill: The skill being executed
        input_data: Input data provided to skill
        output_data: Output data produced by skill

    Returns:
        GateResult with pass/fail status and auto-fix suggestions
    """
    for gate_id in skill.quality_gates:
        gate = get_gate(gate_id)
        if not gate.check_fn(output_data):
            if gate.auto_fix_fn:
                output_data = gate.auto_fix_fn(output_data)
            else:
                return GateResult(failed=True, gate_id=gate_id)
    return GateResult(failed=False)

Hooks Integration

Hook Points

The registry defines hook points throughout the skill lifecycle:

| Hook Point | Description | Context | |------------|-------------|---------| | PRE_SKILL_LOAD | Before skill is loaded | {skill_id, skill_metadata} | | POST_SKILL_LOAD | After skill is loaded | {skill_id, skill_instance} | | PRE_SKILL_EXEC | Before skill execution | {skill_id, input_data} | | POST_SKILL_EXEC | After skill execution | {skill_id, output_data, metrics} | | ON_SKILL_ERROR | On skill execution error | {skill_id, error, stack_trace} | | ON_QUALITY_GATE_FAIL | On quality gate failure | {skill_id, gate_id, output_data} |

Hook Registration

def register_hook(hook_point: str, handler: Callable, priority: int = 100):
    """
    Register a hook handler for a specific hook point.

    Args:
        hook_point: The hook point identifier
        handler: The callable to execute
        priority: Handler priority (lower = higher priority)
    """
    hooks_manager.register_hook(hook_point, handler, priority)

Skill Discovery Protocol

Directory Scanning

The registry discovers skills by scanning the skills/ directory:

def discover_skills(directory: Path) -> List[SkillMetadata]:
    """
    Discover all skills in the given directory.

    Args:
        directory: Path to skills directory

    Returns:
        List of skill metadata objects
    """
    skills = []
    for skill_file in directory.glob("**/*.md"):
        metadata = parse_skill_frontmatter(skill_file)
        if metadata:
            skills.append(metadata)
    return skills

Metadata Parsing

Skill metadata is extracted from YAML frontmatter:

def parse_skill_frontmatter(file_path: Path) -> Optional[SkillMetadata]:
    """
    Parse skill frontmatter from markdown file.

    Args:
        file_path: Path to skill markdown file

    Returns:
        SkillMetadata object or None if parsing fails
    """
    with open(file_path, 'r', encoding='utf-8') as f:
        content = f.read()
        if content.startswith('---'):
            frontmatter_end = content.find('---', 3)
            yaml_content = content[3:frontmatter_end]
            return SkillMetadata.from_yaml(yaml_content)
    return None

Skill Execution Monitoring

Execution Metrics

The registry tracks the following metrics for each skill execution:

• Execution Time: Time taken to complete skill execution
• Token Usage: Number of tokens consumed
• Quality Gate Passes: Number of quality gates passed
• Error Count: Number of errors encountered
• Retry Count: Number of retry attempts

Performance Monitoring

def track_execution(skill_id: str, metrics: ExecutionMetrics):
    """
    Track execution metrics for a skill.

    Args:
        skill_id: The skill identifier
        metrics: The execution metrics to track
    """
    if skill_id not in execution_history:
        execution_history[skill_id] = []
    execution_history[skill_id].append({
        "timestamp": datetime.now(),
        "metrics": metrics
    })

    # Check for performance anomalies
    if metrics.execution_time_ms > get_baseline(skill_id) * 2:
        logging.warning(f"Skill {skill_id} execution time anomaly detected")

Error Handling & Recovery

Error Classification

The registry classifies errors into categories:

1. Load Errors: Errors during skill loading
2. Validation Errors: Errors during schema validation
3. Execution Errors: Errors during skill execution
4. Quality Gate Errors: Errors during quality gate enforcement
5. System Errors: Errors in the registry system itself

Recovery Strategies

| Error Type | Recovery Strategy | Max Retries | |------------|-------------------|-------------| | Load Error | Skip skill, mark as unavailable | 0 | | Validation Error | Return validation errors to user | 0 | | Execution Error | Retry with exponential backoff | 3 | | Quality Gate Error | Apply auto-fix, then retry | 2 | | System Error | Degrade to safe mode | 1 |

Graceful Degradation

The registry implements graceful degradation when errors occur:

def degrade_gracefully(error: Exception, context: SkillContext) -> SkillResult:
    """
    Degrade gracefully when an error occurs.

    Args:
        error: The exception that occurred
        context: The skill execution context

    Returns:
        SkillResult with degraded but safe output
    """
    logging.error(f"Skill {context.skill_id} failed, degrading: {error}")

    return SkillResult(
        status="degraded",
        result=get_safe_baseline(context.skill_id),
        metadata={
            "error": str(error),
            "degradation_level": determine_degradation_level(error)
        }
    )

Registry API

Core Functions

# Skill Registration
register_skill(skill: Skill) -> None
unregister_skill(skill_id: str) -> bool
get_skill(skill_id: str) -> Optional[Skill]
list_skills() -> List[Skill]

# Skill Execution
execute_skill(skill_id: str, input_data: Dict) -> SkillResult
execute_skills(skill_ids: List[str], input_data: Dict) -> List[SkillResult]
execute_parallel(skill_ids: List[str], input_data: Dict) -> List[SkillResult]

# Dependency Management
resolve_dependencies(skill_id: str) -> List[Skill]
check_circular_dependencies() -> bool
get_execution_graph() -> DependencyGraph

# Quality Gates
register_quality_gate(gate: QualityGate) -> None
enforce_gates(skill_id: str, output_data: Dict) -> GateResult
get_gate_status(skill_id: str) -> List[GateStatus]

# Monitoring
get_execution_history(skill_id: str) -> List[ExecutionRecord]
get_performance_metrics(skill_id: str) -> PerformanceMetrics
get_registry_stats() -> RegistryStats

Extension Points

Custom Skill Loaders

Implement custom skill loaders by extending SkillLoader:

class CustomSkillLoader(SkillLoader):
    def load(self, source: str) -> Skill:
        # Custom loading logic
        pass

    def validate(self, skill: Skill) -> bool:
        # Custom validation logic
        pass

Custom Validators

Implement custom validators by extending SkillValidator:

class CustomValidator(SkillValidator):
    def validate_input(self, skill: Skill, input_data: Dict) -> ValidationResult:
        # Custom input validation
        pass

    def validate_output(self, skill: Skill, output_data: Dict) -> ValidationResult:
        # Custom output validation
        pass

Best Practices

1. Skill Design

• Keep skills focused on a single responsibility
• Use clear, descriptive skill names
• Document all input/output parameters
• Include comprehensive error handling

1. Dependency Management

• Minimize dependencies between skills
• Avoid circular dependencies
• Use explicit dependency declarations

1. Quality Gates

• Define clear pass/fail criteria
• Provide auto-fix functions when possible
• Document gate requirements and limitations

1. Performance

• Monitor execution metrics regularly
• Optimize slow-performing skills
• Use parallel execution for independent skills

1. Testing

• Test skills in isolation
• Test skill integration points
• Validate against quality gates

Registry Configuration

Environment Variables

# Skill Discov

…

## Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

- **Author:** [dungnotnull](https://github.com/dungnotnull)
- **Source:** [dungnotnull/aim-training-sensitivity-converter-agent-skill](https://github.com/dungnotnull/aim-training-sensitivity-converter-agent-skill)
- **License:** MIT

Install and usage instructions live in the source repository linked above.