Industrial Waste Heat Recovery

SKILL.md — Industrial Waste-Heat Recovery Skill Registry

Overview

industrial-waste-heat-recovery is a production-grade harness skill with a flexible, modular agent architecture. It moves beyond rigid hierarchies to implement a dynamic skill-registry pattern with chain-of-thought routing, specialized sub-agents, and comprehensive lifecycle management.

Architecture Philosophy

From Static to Dynamic

Traditional (v1.0): Fixed main → sub-skill chain

main.md → sub-1 → sub-2 → sub-3 → sub-4 → sub-5 → output

Flexible Registry (v2.0): Dynamic router → specialized agents

Input → Intent Router → Skill Registry → Agent Pool → Synthesizer → Quality Gate → Output

Key Architectural Improvements

1. Skill Registry Pattern — Dynamic skill resolution, registration, validation
2. Chain-of-Thought Router — Intelligent agent selection based on query analysis
3. Specialized Sub-Agents — Reusable, composable domain agents
4. Hooks System — Lifecycle management, state synchronization, event emission
5. Tool Definitions — Schema-based tool execution with validation
6. Graceful Degradation — Multi-level fallback with explicit limitation flags

---

Skill Registry System

Registration Schema

Each skill/agent registers with:

interface SkillRegistration {
  id: string;                    // Unique identifier
  name: string;                  // Display name
  version: string;               // Semver
  description: string;           // When to use
  intent_patterns: string[];     // Trigger patterns
  input_schema: JSONSchema;      // Expected input structure
  output_schema: JSONSchema;     // Output structure
  tools: ToolDefinition[];       // Available tools
  quality_gates: QualityGate[];  // Validation gates
  dependencies: string[];        // Required skills/agents
  tags: string[];                // Categorization tags
  category: SkillCategory;       // Primary domain
}

Skill Categories

enum SkillCategory {
  INTAKE = "intake",             // Requirements gathering
  EVIDENCE = "evidence",         // Data collection
  ANALYSIS = "analysis",         // Core domain analysis
  KNOWLEDGE = "knowledge",       // Academic/professional research
  SYNTHESIS = "synthesis",       // Result aggregation
  VALIDATION = "validation",     // Quality assurance
  UTILITY = "utility"            // Helper functions
}

Registry Operations

Register Skill

function registerSkill(skill: SkillRegistration): boolean {
  // Validate input/output schemas
  // Check dependency satisfaction
  // Add to registry index
  // Emit registration event
}

Resolve Skills

function resolveSkills(query: string, context: Context): SkillResolution[] {
  // Analyze query intent
  // Match against intent_patterns
  // Check dependencies
  // Return ordered skill list
}

Execute Skill

async function executeSkill(
  skillId: string,
  input: ValidatedInput,
  context: ExecutionContext
): Promise {
  // Validate input against schema
  // Acquire required tools
  // Execute with hooks
  // Validate output
  // Emit completion event
}

---

Chain-of-Thought Router

Intent Analysis Pipeline

Input Query
    ↓
[Parse] Tokenize + Entity Extraction
    ↓
[Classify] Domain Intent Recognition
    ↓
[Route] Skill Selection & Composition
    ↓
[Execute] Agent Orchestration
    ↓
[Synthesize] Result Aggregation

Router Logic

class IntentRouter:
    def analyze(self, query: str, context: Context) -> RoutePlan:
        # Step 1: Extract domain entities
        entities = self.extract_entities(query)
        
        # Step 2: Classify intent
        intent = self.classify_intent(query, entities)
        
        # Step 3: Select relevant skills
        skills = self.registry.resolve(intent, entities)
        
        # Step 4: Compose execution plan
        plan = self.compose_plan(skills, intent)
        
        # Step 5: Validate dependencies
        self.validate_dependencies(plan)
        
        return plan

Intent Patterns

| Domain | Trigger Patterns | Primary Skills | |--------|-----------------|----------------| | Requirements | "analyze", "assess", "evaluate", input variables | intake, evidence | | System Design | "design", "select", "size", heat exchangers/ORC | analysis, integration | | Optimization | "optimize", "improve", "maximize", pinch | analysis, synthesis | | Economic | "cost", "payback", "ROI", investment | analysis, validation | | Research | "research", "latest", "state-of-art", academic | knowledge, evidence |

---

Specialized Sub-Agents

Agent Pool

interface AgentPool {
  intake: RequirementsIntakeAgent;
  evidence: EvidenceCollectorAgent;
  characterization: SourceCharacterizationAgent;
  selection: TechnologySelectionAgent;
  integration: PinchIntegrationAgent;
  economic: EconomicAssessmentAgent;
  knowledge: KnowledgeQueryAgent;
  synthesis: ResultSynthesisAgent;
  validation: QualityGateAgent;
}

Agent Definition Schema

interface AgentDefinition {
  id: string;
  name: string;
  role: string;
  category: AgentCategory;
  capabilities: Capability[];
  input_schema: JSONSchema;
  output_schema: JSONSchema;
  tools: ToolReference[];
  execution_strategy: ExecutionStrategy;
  quality_criteria: QualityCriterion[];
  fallback_strategy: FallbackStrategy;
}

Agent Types

1. RequirementsIntakeAgent

Purpose: Clarify analysis scope, constraints, inputs, language Input: Raw user query Output: Structured requirements Tools: Conversation, entity extraction Quality: All required fields confirmed

2. EvidenceCollectorAgent

Purpose: Fetch authoritative real-time and reference data Input: Requirements object Output: Evidence bundle with sources Tools: WebSearch, WebFetch, Read Quality: ≥3 sources, ≥1 authoritative

3. SourceCharacterizationAgent

Purpose: Analyze waste-heat sources (temperature, flow, quality) Input: Process parameters Output: Source characterization Tools: Thermodynamics calculations, reference data Quality: All sources characterized

4. TechnologySelectionAgent

Purpose: Select recovery technology (HE, ORC, thermoelectric) Input: Source characterization Output: Technology recommendation Tools: Decision matrices, performance data Quality: Selection with justification

5. PinchIntegrationAgent

Purpose: Perform pinch analysis for heat integration Input: Stream data, temperatures Output: Pinch analysis, composite curves Tools: Pinch algorithms, temperature intervals Quality: Pinch point identified, targets calculated

6. EconomicAssessmentAgent

Purpose: Assess economics (payback, ROI, emissions) Input: System design, energy recovery Output: Economic analysis with scenarios Tools: Financial models, emission factors Quality: Multi-scenario analysis

7. KnowledgeQueryAgent

Purpose: Query knowledge base for academic evidence Input: Analysis keywords Output: Academic citations with tiers Tools: Read (SECOND-KNOWLEDGE-BRAIN.md), WebSearch Quality: ≥1 Tier 1-2 source

8. ResultSynthesisAgent

Purpose: Combine all analysis into coherent report Input: All agent outputs Output: Structured report Tools: Synthesis templates Quality: All sections present, internally consistent

9. QualityGateAgent

Purpose: Validate output against quality gates Input: Draft report Output: Validated report or failure with fixes Tools: Validation rules Quality: All gates passed

---

Hooks System

Hook Types

Lifecycle Hooks

interface LifecycleHooks {
  beforeRegistration?: (skill: SkillRegistration) => void;
  afterRegistration?: (skill: SkillRegistration) => void;
  beforeExecution?: (skillId: string, input: any) => void;
  afterExecution?: (skillId: string, output: any) => void;
  onError?: (error: Error, context: any) => void;
}

State Synchronization Hooks

interface StateHooks {
  beforeStateChange?: (oldState: State, newState: State) => boolean;
  afterStateChange?: (state: State) => void;
  onStateValidationError?: (error: ValidationError) => void;
}

Event Emission Hooks

interface EventHooks {
  onEvent?: (event: DomainEvent) => void;
  onEventBatch?: (events: DomainEvent[]) => void;
  onErrorEvent?: (event: ErrorEvent) => void;
}

Hook Execution Order

1. beforeRegistration
2. afterRegistration
   ↓
3. beforeExecution (for each skill)
4. [Skill Execution]
5. afterExecution
   ↓
6. beforeStateChange
7. afterStateChange
   ↓
8. onEvent / onEventBatch

Hook Implementation

class HookManager:
    def __init__(self):
        self.lifecycle_hooks = []
        self.state_hooks = []
        self.event_hooks = []
    
    def register_lifecycle_hook(self, hook: LifecycleHook):
        self.lifecycle_hooks.append(hook)
    
    def execute_before_execution(self, skill_id: str, input: any):
        for hook in self.lifecycle_hooks:
            if hasattr(hook, 'before_execution'):
                hook.before_execution(skill_id, input)
    
    def execute_after_execution(self, skill_id: str, output: any):
        for hook in self.lifecycle_hooks:
            if hasattr(hook, 'after_execution'):
                hook.after_execution(skill_id, output)

---

Tool Definitions

Tool Schema

interface ToolDefinition {
  id: string;
  name: string;
  description: string;
  category: ToolCategory;
  input_schema: JSONSchema;
  output_schema: JSONSchema;
  execution_handler: ToolHandler;
  validation_rules: ValidationRule[];
  error_handling: ErrorHandlingStrategy;
  rate_limits?: RateLimit;
  cache_policy?: CachePolicy;
}

Tool Categories

enum ToolCategory {
  DATA_FETCH = "data_fetch",           // WebSearch, WebFetch
  FILE_IO = "file_io",                 // Read, Write
  COMPUTATION = "computation",         // Calculations
  KNOWLEDGE = "knowledge",             // Knowledge base queries
  VALIDATION = "validation",           // Schema validation
  SYNTHESIS = "synthesis"             // Result composition
}

Tool Registry

class ToolRegistry:
    def __init__(self):
        self.tools = {}
    
    def register_tool(self, tool: ToolDefinition):
        self.tools[tool.id] = tool
    
    def get_tool(self, tool_id: str) -> ToolDefinition:
        return self.tools.get(tool_id)
    
    def execute_tool(self, tool_id: str, input: any) -> any:
        tool = self.get_tool(tool_id)
        if not tool:
            raise ToolNotFoundError(tool_id)
        
        # Validate input
        self.validate_input(tool, input)
        
        # Execute
        try:
            output = tool.execution_handler(input)
            self.validate_output(tool, output)
            return output
        except Exception as e:
            return self.handle_error(tool, e, input)

---

Quality Gates

Universal Gates (U1-U6)

| Gate | Criterion | Auto-Fix | Enforcement | |------|-----------|----------|-------------| | U1 | ≥3 sources, ≥1 authoritative | Fetch from KB/evidence | Append sources | | U2 | Disclosure before recommendation | Prepend disclosure | Block until present | | U3 | Evidence hierarchy stated (Tier 1-4) | Annotate tiers | Tag each source | | U4 | Language matches preference | Translate | Detect and translate | | U5 | Output uses template | Reformat | Check sections | | U6 | Claims traceable to sources | Flag unsupported | Mark with source |

Domain Gates (G1-G4)

| Gate | Criterion | Auto-Fix | Enforcement | |------|-----------|----------|-------------| | G1 | Sources characterized (T, flow, quality) | Characterize | Required | | G2 | Recovery tech selected with justification | Select & justify | Required | | G3 | Pinch analysis performed | Perform pinch | Required | | G4 | Economics assessed (multi-scenario) | Run scenarios | Required |

Gate Enforcement

class QualityGateAgent:
    def __init__(self, gates: List[QualityGate]):
        self.gates = gates
        self.max_retries = 2
    
    def enforce(self, output: any) -> ValidationResult:
        for gate in self.gates:
            for attempt in range(self.max_retries + 1):
                result = gate.check(output)
                if result.passed:
                    break
                if attempt  int:
        primary_available = availability['primary_sources']
        kb_available = availability['knowledge_base']
        input_complete = availability['input_data']
        
        if primary_available:
            return 0
        elif kb_available:
            return 2
        elif input_complete:
            return 3
        else:
            return 4
    
    def emit_notice(self, level: int) -> str:
        if level == 0:
            return ""
        return f"""
---
⚠️ LIMITATION NOTICE
This output was generated with reduced data availability (Level {level}).
Cross-check with current data before acting on it.
Substituted/missing sources are flagged inline.
---
"""

---

Configuration Management

Configuration Schema

See /config/harness/config.yaml for complete configuration.

Environment Variables

# Knowledge Base
KNOWLEDGE_BASE_PATH=SECOND-KNOWLEDGE-BRAIN.md
KNOWLEDGE_UPDATE_SCHEDULE=weekly

# API Limits
WEBSEARCH_RATE_LIMIT=10/min
WEBFETCH_RATE_LIMIT=5/min

# Quality
MAX_RETRIES=2
QUALITY_GATE_STRICT=true

# Logging
LOG_LEVEL=INFO
LOG_FORMAT=structured
LOG_PATH=logs/harness.log

# Performance
MAX_CONTEXT_TOKENS=100000
TOKEN_OPTIMIZATION=true

---

Performance Optimization

Context Window Management

class ContextManager:
    def __init__(self, max_tokens: int = 100000):
        self.max_tokens = max_tokens
        self.usage = {}
    
    def estimate_tokens(self, text: str) -> int:
        # Approximate token count
        return len(text.split()) * 1.3
    
    def prune_content(self, content: str, target: int) -> str:
        # Intelligent pruning preserving structure
        if self.estimate_tokens(content)  ErrorResult:
        category = self.classify(error)
        config = self.retry_config.get(category, {"max_retries": 0})
        
        if config.get("max_retries", 0) > context.retry_count:
            return ErrorResult(retry=True, backoff=config.get("backoff"))
        elif config.get("fallback"):
            return ErrorResult(fallback=True, fallback_value=self.get_fallback(category))
        else:
            return ErrorResult(fail=True, error_message=self.format_error(error))

---

Extensibility

Adding New Skills

1. Define skill registration schema
2. Implement skill logic
3. Register with SkillRegistry
4. Add intent patterns to router
5. Document quality gates

Adding New Tools

1. Define tool schema
2. Implement execution handler
3. Register with ToolRegistry
4. Configure rate limits and caching

Adding New Hooks

1. Define hook interface
2. Implement hook logic
3. Register with HookManager
4. Configure execution order

---

Version History

• v2.0.0 — Production-grade flexible architecture with skill registry, hooks, modular agents
• v1.0.0 — Initial hierarchical skill implementation

---

References

• /config/harness/ — Harness configuration
• /config/agents/ — Agent definitions
• /config/skills/ — Skill registrations
• /tools/registry/ — Registry implementations
• /tools/hooks/ — Hook implementations
• /references/ — Domain references
• /scripts/ — Automation scripts

Source & license

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

• Author: dungnotnull
• Source: dungnotnull/industrial-waste-heat-recovery-agent-skill
• License: MIT

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