scripts-analyze-task-complexity

#!/usr/bin/env python3 """

title: "Analyze Task Complexity" component_type: script version: "1.0.0" audience: contributor status: stable summary: "Task Complexity Analyzer ========================" keywords: ['analysis', 'analyze', 'api', 'backend', 'complexity'] tokens: ~500 created: 2025-12-22 updated: 2025-12-22 script_name: "analyze-task-complexity.py" language: python executable: true usage: "python3 scripts/analyze-task-complexity.py [options]" python_version: "3.10+" dependencies: [] modifies_files: false network_access: false requires_auth: false

Task Complexity Analyzer

Analyzes incoming requests to determine complexity, domains, and required capabilities. Used by orchestrators to route tasks to appropriate agents and estimate resources.

Features:

• Multi-domain detection (13 task domains)
• Complexity level classification (simple → research)
• Resource estimation (tokens, tool calls)
• Parallel execution detection
• State management requirements
• Confidence scoring

Exit Codes: 0 - Success 1 - Input error

Usage: python3 scripts/analyze-task-complexity.py "Build a REST API with auth" python3 scripts/analyze-task-complexity.py --test echo '{"task": "description"}' | python3 scripts/analyze-task-complexity.py --stdin """

import json import re import sys from dataclasses import dataclass, asdict, field from typing import List, Set, Dict, Tuple, Optional from enum import Enum from datetime import datetime, timezone

class ComplexityLevel(Enum): """Task complexity levels with resource implications.""" SIMPLE = "simple" # Single agent, <5 tool calls MODERATE = "moderate" # 2-3 agents, 5-15 tool calls COMPLEX = "complex" # 4-6 agents, 15-30 tool calls RESEARCH = "research" # 5-10 agents, 30+ tool calls

class TaskDomain(Enum): """Task domains for agent routing.""" RESEARCH = "research" ARCHITECTURE = "architecture" IMPLEMENTATION = "implementation" TESTING = "testing" DOCUMENTATION = "documentation" SECURITY = "security" PERFORMANCE = "performance" THEIA_EXTENSION = "theia_extension" MULTI_AGENT = "multi_agent" EVENT_DRIVEN = "event_driven" DATABASE = "database" API_DESIGN = "api_design" STRATEGY = "strategy"

@dataclass class TaskAnalysis: """Results of task complexity analysis.""" complexity: ComplexityLevel domains: List[TaskDomain] estimated_tokens: int estimated_tool_calls: int parallel_capable: bool requires_state_management: bool trigger_keywords: List[str] confidence: float recommended_agents: List[str] = field(default_factory=list) analysis_timestamp: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())

def to_dict(self) -> Dict:
    """Convert to dictionary with proper enum handling."""
    return {
        "complexity": self.complexity.value,
        "domains": [d.value for d in self.domains],
        "estimated_tokens": self.estimated_tokens,
        "estimated_tool_calls": self.estimated_tool_calls,
        "parallel_capable": self.parallel_capable,
        "requires_state_management": self.requires_state_management,
        "trigger_keywords": self.trigger_keywords,
        "confidence": self.confidence,
        "recommended_agents": self.recommended_agents,
        "analysis_timestamp": self.analysis_timestamp
    }

class TaskAnalyzer: """Analyzes tasks to determine orchestration requirements."""

# Keyword mappings to domains
DOMAIN_KEYWORDS = {
    TaskDomain.RESEARCH: {
        'research', 'analyze', 'compare', 'evaluate', 'investigate',
        'study', 'explore', 'survey', 'review', 'assess', 'find',
        'discover', 'examine', 'benchmark'
    },
    TaskDomain.ARCHITECTURE: {
        'design', 'architect', 'system', 'structure', 'pattern',
        'framework', 'c4', 'adr', 'diagram', 'model', 'blueprint',
        'microservice', 'monolith', 'distributed'
    },
    TaskDomain.IMPLEMENTATION: {
        'implement', 'build', 'create', 'develop', 'code', 'write',
        'generate', 'construct', 'program', 'script', 'refactor',
        'fix', 'update', 'modify', 'add', 'feature'
    },
    TaskDomain.TESTING: {
        'test', 'verify', 'validate', 'qa', 'quality', 'coverage',
        'unit', 'integration', 'e2e', 'assertion', 'mock', 'spec',
        'tdd', 'bdd', 'regression'
    },
    TaskDomain.DOCUMENTATION: {
        'document', 'explain', 'describe', 'comment', 'readme',
        'guide', 'tutorial', 'reference', 'spec', 'manual', 'docs',
        'api-docs', 'openapi', 'swagger'
    },
    TaskDomain.SECURITY: {
        'security', 'audit', 'vulnerability', 'authentication',
        'authorization', 'encrypt', 'secure', 'penetration', 'threat',
        'owasp', 'xss', 'sql-injection', 'csrf', 'rbac', 'jwt'
    },
    TaskDomain.PERFORMANCE: {
        'optimize', 'performance', 'speed', 'efficient', 'scale',
        'benchmark', 'profile', 'throughput', 'latency', 'memory',
        'cache', 'async', 'parallel', 'concurrent'
    },
    TaskDomain.THEIA_EXTENSION: {
        'theia', 'extension', 'vscode', 'ide', 'editor', 'plugin',
        'monaco', 'inversify', 'di', 'contribution', 'widget',
        'language-server', 'lsp'
    },
    TaskDomain.MULTI_AGENT: {
        'agent', 'orchestrate', 'coordinate', 'delegate', 'subagent',
        'autonomous', 'multi-agent', 'workflow', 'pipeline', 'crew',
        'swarm', 'supervisor'
    },
    TaskDomain.EVENT_DRIVEN: {
        'event', 'pub-sub', 'message', 'queue', 'stream', 'kafka',
        'async', 'reactive', 'subscription', 'listener', 'emit',
        'webhook', 'notification'
    },
    TaskDomain.DATABASE: {
        'database', 'sql', 'nosql', 'query', 'schema', 'migration',
        'orm', 'foundationdb', 'd1', 'postgresql', 'table', 'index',
        'transaction', 'acid', 'mongodb', 'redis'
    },
    TaskDomain.API_DESIGN: {
        'api', 'endpoint', 'rest', 'graphql', 'rpc', 'openapi',
        'swagger', 'route', 'controller', 'service', 'interface',
        'grpc', 'protobuf', 'websocket'
    },
    TaskDomain.STRATEGY: {
        'strategy', 'market', 'competitor', 'business', 'brief',
        'consulting', 'analysis', 'opportunity', 'swot', 'tam',
        'positioning', 'gtm', 'roadmap'
    }
}

# Complexity indicators
COMPLEXITY_INDICATORS = {
    ComplexityLevel.RESEARCH: {
        'comprehensive', 'thorough', 'deep dive', 'full analysis',
        'investigate', 'all aspects', 'complete', 'exhaustive',
        'research everything', 'market study'
    },
    ComplexityLevel.COMPLEX: {
        'end-to-end', 'full stack', 'entire', 'complete system',
        'production-ready', 'enterprise', 'scalable', 'robust',
        'multi-component', 'full implementation'
    },
    ComplexityLevel.MODERATE: {
        'multiple', 'several', 'integrate', 'connect', 'combine',
        'few', 'couple', 'with tests', 'add feature'
    },
    ComplexityLevel.SIMPLE: {
        'simple', 'quick', 'basic', 'minimal', 'straightforward',
        'just', 'only', 'single', 'small', 'tiny', 'one'
    }
}

# Agent recommendations by domain
DOMAIN_AGENTS = {
    TaskDomain.RESEARCH: ["market-researcher", "competitive-analyst", "trend-analyst"],
    TaskDomain.ARCHITECTURE: ["backend-architect", "software-design-architect", "senior-architect"],
    TaskDomain.IMPLEMENTATION: ["rust-expert-developer", "backend-development", "frontend-mobile-development"],
    TaskDomain.TESTING: ["testing-specialist", "codi-test-engineer", "rust-qa-specialist"],
    TaskDomain.DOCUMENTATION: ["codi-documentation-writer", "documentation-librarian"],
    TaskDomain.SECURITY: ["security-specialist", "security-auditor", "penetration-testing-agent"],
    TaskDomain.PERFORMANCE: ["application-performance", "performance-profiler", "wasm-optimization-expert"],
    TaskDomain.THEIA_EXTENSION: ["terminal-integration-specialist", "frontend-react-typescript-expert"],
    TaskDomain.MULTI_AGENT: ["orchestrator", "workflow-orchestrator", "use-case-analyzer"],
    TaskDomain.EVENT_DRIVEN: ["websocket-protocol-designer", "backend-architect"],
    TaskDomain.DATABASE: ["database-architect", "foundationdb-expert"],
    TaskDomain.API_DESIGN: ["backend-api-security", "actix-web-specialist"],
    TaskDomain.STRATEGY: ["strategy-brief-generator", "business-intelligence-analyst", "venture-capital-business-analyst"]
}

def analyze(self, task_description: str) -> TaskAnalysis:
    """
    Analyze task to determine complexity and requirements.

    Args:
        task_description: User's task request

    Returns:
        TaskAnalysis with orchestration requirements
    """
    task_lower = task_description.lower()
    words = set(re.findall(r'\b\w+\b', task_lower))

    # Detect domains
    domains = self._detect_domains(words)

    # Detect complexity level
    complexity = self._detect_complexity(task_lower, words, len(domains))

    # Estimate resources
    estimated_tokens, estimated_tool_calls = self._estimate_resources(
        complexity, len(domains), len(task_description)
    )

    # Determine capabilities
    parallel_capable = self._is_parallel_capable(domains)
    requires_state = self._requires_state_management(complexity, domains)

    # Extract trigger keywords
    triggers = self._extract_triggers(words)

    # Calculate confidence
    confidence = self._calculate_confidence(domains, triggers)

    # Recommend agents
    recommended_agents = self._recommend_agents(domains, complexity)

    return TaskAnalysis(
        complexity=complexity,
        domains=domains,
        estimated_tokens=estimated_tokens,
        estimated_tool_calls=estimated_tool_calls,
        parallel_capable=parallel_capable,
        requires_state_management=requires_state,
        trigger_keywords=triggers,
        confidence=confidence,
        recommended_agents=recommended_agents
    )

def _detect_domains(self, words: Set[str]) -> List[TaskDomain]:
    """Detect applicable domains from keywords."""
    detected_domains = []
    domain_scores = {}

    for domain, keywords in self.DOMAIN_KEYWORDS.items():
        overlap = words & keywords
        if overlap:
            domain_scores[domain] = len(overlap)

    # Sort by score and return top domains
    sorted_domains = sorted(domain_scores.items(), key=lambda x: x[1], reverse=True)
    detected_domains = [d[0] for d in sorted_domains]

    # Default to implementation if no domains detected
    if not detected_domains:
        detected_domains.append(TaskDomain.IMPLEMENTATION)

    return detected_domains

def _detect_complexity(
    self,
    task_lower: str,
    words: Set[str],
    domain_count: int
) -> ComplexityLevel:
    """Determine task complexity level."""
    # Check for research indicators
    research_keywords = self.COMPLEXITY_INDICATORS[ComplexityLevel.RESEARCH]
    if words & research_keywords or 'research' in task_lower:
        return ComplexityLevel.RESEARCH

    # Check for complex indicators
    complex_keywords = self.COMPLEXITY_INDICATORS[ComplexityLevel.COMPLEX]
    if words & complex_keywords or domain_count >= 4:
        return ComplexityLevel.COMPLEX

    # Check for moderate indicators
    moderate_keywords = self.COMPLEXITY_INDICATORS[ComplexityLevel.MODERATE]
    if words & moderate_keywords or domain_count >= 2:
        return ComplexityLevel.MODERATE

    # Check for simple indicators
    simple_keywords = self.COMPLEXITY_INDICATORS[ComplexityLevel.SIMPLE]
    if words & simple_keywords:
        return ComplexityLevel.SIMPLE

    # Default based on domain count
    if domain_count >= 3:
        return ComplexityLevel.COMPLEX
    elif domain_count >= 2:
        return ComplexityLevel.MODERATE
    return ComplexityLevel.SIMPLE

def _estimate_resources(
    self,
    complexity: ComplexityLevel,
    domain_count: int,
    description_length: int
) -> Tuple[int, int]:
    """Estimate token and tool call requirements."""
    # Base estimates by complexity
    base_estimates = {
        ComplexityLevel.SIMPLE: (5_000, 3),
        ComplexityLevel.MODERATE: (15_000, 10),
        ComplexityLevel.COMPLEX: (50_000, 25),
        ComplexityLevel.RESEARCH: (100_000, 40)
    }

    tokens, tools = base_estimates[complexity]

    # Adjust for domain count
    domain_multiplier = 1 + (domain_count - 1) * 0.2
    tokens = int(tokens * domain_multiplier)
    tools = int(tools * domain_multiplier)

    # Adjust for description complexity
    if description_length > 500:
        tokens = int(tokens * 1.3)
    elif description_length > 200:
        tokens = int(tokens * 1.1)

    return tokens, tools

def _is_parallel_capable(self, domains: List[TaskDomain]) -> bool:
    """Determine if task can be parallelized."""
    # Research and analysis can often be parallelized
    parallel_domains = {
        TaskDomain.RESEARCH,
        TaskDomain.SECURITY,
        TaskDomain.PERFORMANCE,
        TaskDomain.TESTING,
        TaskDomain.DOCUMENTATION,
        TaskDomain.STRATEGY
    }

    return any(d in parallel_domains for d in domains)

def _requires_state_management(
    self,
    complexity: ComplexityLevel,
    domains: List[TaskDomain]
) -> bool:
    """Determine if stateful coordination is needed."""
    # Complex tasks and certain domains need state management
    stateful_domains = {
        TaskDomain.MULTI_AGENT,
        TaskDomain.EVENT_DRIVEN,
        TaskDomain.DATABASE,
        TaskDomain.STRATEGY
    }

    return (
        complexity in [ComplexityLevel.COMPLEX, ComplexityLevel.RESEARCH] or
        any(d in stateful_domains for d in domains)
    )

def _extract_triggers(self, words: Set[str]) -> List[str]:
    """Extract key trigger words from task."""
    all_triggers = set()
    for keywords in self.DOMAIN_KEYWORDS.values():
        all_triggers.update(keywords)

    return sorted(words & all_triggers)

def _calculate_confidence(
    self,
    domains: List[TaskDomain],
    triggers: List[str]
) -> float:
    """Calculate confidence in analysis."""
    # Base confidence
    confidence = 0.5

    # Increase for each domain detected
    confidence += min(len(domains) * 0.1, 0.3)

    # Increase for trigger keywords
    confidence += min(len(triggers) * 0.05, 0.2)

    return min(confidence, 1.0)

def _recommend_agents(
    self,
    domains: List[TaskDomain],
    complexity: ComplexityLevel
) -> List[str]:
    """Recommend agents for the task."""
    recommended = []
    seen = set()

    for domain in domains:
        agents = self.DOMAIN_AGENTS.get(domain, [])
        for agent in agents:
            if agent not in seen:
                recommended.append(agent)
                seen.add(agent)

    # Limit recommendations based on complexity
    limits = {
        ComplexityLevel.SIMPLE: 2,
        ComplexityLevel.MODERATE: 4,
        ComplexityLevel.COMPLEX: 6,
        ComplexityLevel.RESEARCH: 10
    }

    return recommended[:limits.get(complexity, 4)]

def run_tests(): """Run analyzer tests.""" print("Running Task Complexity Analyzer Tests...") print("=" * 50)

tests_passed = 0
tests_failed = 0
analyzer = TaskAnalyzer()

# Test 1: Simple task detection
print("\n[Test 1] Simple task detection...")
result = analyzer.analyze("Fix a simple bug in one file")
if result.complexity == ComplexityLevel.SIMPLE:
    print(f"  PASS: Detected as {result.complexity.value}")
    tests_passed += 1
else:
    print(f"  FAIL: Expected simple, got {result.complexity.value}")
    tests_failed += 1

# Test 2: Complex task detection
print("\n[Test 2] Complex task detection...")
result = analyzer.analyze("Build a complete end-to-end production-ready authentication system")
if result.complexity in [ComplexityLevel.COMPLEX, ComplexityLevel.RESEARCH]:
    print(f"  PASS: Detected as {result.complexity.value}")
    tests_passed += 1
else:
    print(f"  FAIL: Expected complex/research, got {result.complexity.value}")
    tests_failed += 1

# Test 3: Research task detection
print("\n[Test 3] Research task detection...")
result = analyzer.analyze("Conduct comprehensive market research on AI development tools")
if result.complexity == ComplexityLevel.RESEARCH:
    print(f"  PASS: Detected as {result.complexity.value}")
    tests_passed += 1
else:
    print(f"  FAIL: Expected research, got {result.complexity.value}")
    tests_failed += 1

# Test 4: Domain detection - Security
print("\n[Test 4] Domain detection - Security...")
result = analyzer.analyze("Audit the API for security vulnerabilities and authentication issues")
if TaskDomain.SECURITY in result.domains:
    print(f"  PASS: Security domain detected")
    tests_passed += 1
else:
    print(f"  FAIL: Security domain not detected, got {[d.value for d in result.domains]}")
    tests_failed += 1

# Test 5: Domain detection - Strategy
print("\n[Test 5] Domain detection - Strategy...")
result = analyzer.analyze("Create a strategy brief for the AI consulting market")
if TaskDomain.STRATEGY in result.domains:
    print(f"  PASS: Strategy domain detected")
    tests_passed += 1
else:
    print(f"  FAIL: Strategy domain not detected, got {[d.value for d in result.domains]}")
    tests_failed += 1

# Test 6: Multi-domain detection
print("\n[Test 6] Multi-domain detection...")
result = analyzer.analyze("Design and implement a secure REST API with tests and documentation")
if len(result.domains) >= 3:
    print(f"  PASS: {len(result.domains)} domains detected: {[d.value for d in result.domains[:3]]}")
    tests_passed += 1
else:
    print(f"  FAIL: Expected 3+ domains, got {len(result.domains)}")
    tests_failed += 1

# Test 7: Parallel capability
print("\n[Test 7] Parallel capability detection...")
result = analyzer.analyze("Research market trends and analyze competitor performance")
if result.parallel_capable:
    print(f"  PASS: Parallel capable detected")
    tests_passed += 1
else:
    print(f"  FAIL: Should be parallel capable for research")
    tests_failed += 1

# Test 8: State management detection
print("\n[Test 8] State management requirement...")
result = analyzer.analyze("Orchestrate a multi-agent workflow for database migration")
if result.requires_state_management:
    print(f"  PASS: State management required")
    tests_passed += 1
else:
    print(f"  FAIL: Should require state management")
    tests_failed += 1

# Test 9: Agent recommendations
print("\n[Test 9] Agent recommendations...")
result = analyzer.analyze("Build a strategy brief for the tech market")
if len(result.recommended_agents) > 0 and "strategy-brief-generator" in result.recommended_agents:
    print(f"  PASS: Recommended agents: {result.recommended_agents[:3]}")
    tests_passed += 1
else:
    print(f"  FAIL: Expected strategy-brief-generator in recommendations")
    tests_failed += 1

# Test 10: Resource estimation
print("\n[Test 10] Resource estimation...")
result = analyzer.analyze("Comprehensive research on AI market")
if result.estimated_tokens >= 50000 and result.estimated_tool_calls >= 20:
    print(f"  PASS: Tokens: {result.estimated_tokens:,}, Tools: {result.estimated_tool_calls}")
    tests_passed += 1
else:
    print(f"  FAIL: Resource estimates too low")
    tests_failed += 1

# Test 11: Confidence scoring
print("\n[Test 11] Confidence scoring...")
result = analyzer.analyze("security audit and performance testing")
if 0.5 <= result.confidence <= 1.0:
    print(f"  PASS: Confidence: {result.confidence:.2f}")
    tests_passed += 1
else:
    print(f"  FAIL: Invalid confidence: {result.confidence}")
    tests_failed += 1

# Test 12: Serialization
print("\n[Test 12] JSON serialization...")
result = analyzer.analyze("Build a REST API")
try:
    json_str = json.dumps(result.to_dict(), indent=2)
    parsed = json.loads(json_str)
    if "complexity" in parsed and "domains" in parsed:
        print(f"  PASS: JSON serialization works")
        tests_passed += 1
    else:
        print(f"  FAIL: Missing keys in serialized JSON")
        tests_failed += 1
except Exception as e:
    print(f"  FAIL: Serialization error: {e}")
    tests_failed += 1

print("\n" + "=" * 50)
print(f"Tests: {tests_passed + tests_failed} | Passed: {tests_passed} | Failed: {tests_failed}")

return tests_failed == 0

def main(): """CLI interface for task analysis.""" import argparse

parser = argparse.ArgumentParser(description="Task Complexity Analyzer")
parser.add_argument("task", nargs="*", help="Task description to analyze")
parser.add_argument("--test", action="store_true", help="Run tests")
parser.add_argument("--stdin", action="store_true", help="Read JSON from stdin")
parser.add_argument("--pretty", action="store_true", help="Pretty print output")

args = parser.parse_args()

if args.test:
    success = run_tests()
    sys.exit(0 if success else 1)

if args.stdin:
    try:
        input_data = json.load(sys.stdin)
        task_description = input_data.get("task", "")
    except json.JSONDecodeError as e:
        print(f"Error parsing JSON input: {e}", file=sys.stderr)
        sys.exit(1)
elif args.task:
    task_description = " ".join(args.task)
else:
    parser.print_help()
    sys.exit(1)

analyzer = TaskAnalyzer()
analysis = analyzer.analyze(task_description)

if args.pretty:
    print(json.dumps(analysis.to_dict(), indent=2))
else:
    print(json.dumps(analysis.to_dict()))

if name == "main": main()