Meta-Delegation Concepts¶

class PlatformCLI(Protocol):
    """Abstract interface for platform CLIs."""

    def spawn_subprocess(
        self,
        goal: str,
        persona: str,
        working_dir: str,
        environment: Dict[str, str]
    ) -> subprocess.Popen:
        """Spawn subprocess with platform-specific command."""
        ...

    def format_prompt(self, goal: str, persona: str, context: str) -> str:
        """Format prompt for platform's expected input format."""
        ...

    def parse_output(self, output: str) -> Dict[str, Any]:
        """Parse platform-specific output format."""
        ...

@dataclass
class PersonaStrategy:
    """Defines persona-specific behavior."""
    name: str
    communication_style: str
    thoroughness_level: str
    evidence_collection_priority: List[str]
    prompt_template: str

GUIDE = PersonaStrategy(
    name="guide",
    communication_style="socratic",
    thoroughness_level="balanced",
    evidence_collection_priority=[
        "documentation",
        "code_file",
        "test_file"
    ],
    prompt_template="""
    You are a teaching guide. Your goal is to help the user learn by:
    1. Breaking down complex concepts
    2. Providing clear explanations
    3. Including educational examples
    4. Encouraging understanding

    Task: {goal}
    Success Criteria: {success_criteria}

    Focus on creating materials that teach, not just deliver.
    """
)

QA_ENGINEER = PersonaStrategy(
    name="qa_engineer",
    communication_style="precise",
    thoroughness_level="exhaustive",
    evidence_collection_priority=[
        "test_file",
        "test_results",
        "validation_report",
        "code_file"
    ],
    prompt_template="""
    You are a QA engineer. Your goal is to ensure quality by:
    1. Identifying edge cases and error conditions
    2. Writing comprehensive test suites
    3. Validating against all success criteria
    4. Documenting findings thoroughly

    Task: {goal}
    Success Criteria: {success_criteria}

    Test everything rigorously and document all findings.
    """
)

ARCHITECT = PersonaStrategy(
    name="architect",
    communication_style="strategic",
    thoroughness_level="holistic",
    evidence_collection_priority=[
        "architecture_doc",
        "api_spec",
        "diagram",
        "code_file"
    ],
    prompt_template="""
    You are a system architect. Your goal is to design systems by:
    1. Understanding requirements and constraints
    2. Creating high-level architecture
    3. Defining clear interfaces and contracts
    4. Considering scalability and maintainability

    Task: {goal}
    Success Criteria: {success_criteria}

    Focus on design decisions, structure, and system boundaries.
    """
)

JUNIOR_DEV = PersonaStrategy(
    name="junior_dev",
    communication_style="task_focused",
    thoroughness_level="adequate",
    evidence_collection_priority=[
        "code_file",
        "test_file",
        "documentation"
    ],
    prompt_template="""
    You are a junior developer. Your goal is to implement tasks by:
    1. Following specifications closely
    2. Writing clean, working code
    3. Including basic tests
    4. Documenting what you built

    Task: {goal}
    Success Criteria: {success_criteria}

    Focus on getting it working correctly and cleanly.
    """
)

class ScenarioCategory(Enum):
    HAPPY_PATH = "happy_path"
    ERROR_HANDLING = "error_handling"
    BOUNDARY_CONDITIONS = "boundary_conditions"
    SECURITY = "security"
    PERFORMANCE = "performance"
    INTEGRATION = "integration"

Goal: "Create a user registration API"

Generated Scenarios:
1. Happy Path: Valid user registration with all required fields
   → Expects: 201 Created, user ID returned

2. Error Handling: Registration with duplicate email
   → Expects: 409 Conflict, clear error message

3. Boundary: Username at maximum length (255 chars)
   → Expects: 201 Created or 400 with validation message

4. Security: SQL injection attempt in email field
   → Expects: Input sanitized, 400 Bad Request

5. Performance: 1000 concurrent registrations
   → Expects: All succeed, response time < 2s per request

def evaluate_success(
    success_criteria: str,
    evidence: List[EvidenceItem],
    execution_log: str
) -> Tuple[int, str]:
    """
    Evaluate success using multi-factor analysis.

    Returns: (score: 0-100, notes: str)
    """
    score = 0
    notes = []

    # 1. Parse success criteria into requirements
    requirements = parse_criteria(success_criteria)

    # 2. Check evidence for each requirement
    for req in requirements:
        req_score = evaluate_requirement(req, evidence)
        score += req_score
        if req_score < req.weight:
            notes.append(f"Incomplete: {req.description}")

    # 3. Bonus for quality indicators
    if has_tests_passing(evidence, execution_log):
        score += 10
        notes.append("Bonus: All tests passing")

    if has_documentation(evidence):
        score += 5
        notes.append("Bonus: Documentation included")

    # 4. Normalize to 0-100
    final_score = min(100, score)

    return (final_score, "\n".join(notes))

EVIDENCE_PATTERNS = {
    "code_file": ["*.py", "*.js", "*.ts", "*.go", "*.rs", "*.java"],
    "test_file": ["test_*.py", "*_test.py", "*.test.js", "test/*.py"],
    "documentation": ["README.md", "*.md", "docs/*.md"],
    "architecture_doc": ["architecture.md", "design.md", "ARCHITECTURE.md"],
    "api_spec": ["*.yaml", "*.json", "openapi.yaml", "swagger.yaml"],
    "test_results": ["test_output.txt", "pytest.log", "test_results.xml"],
    "execution_log": ["subprocess.log", "execution.log"],
    "validation_report": ["validation_report.md", "success_report.md"],
    "diagram": ["*.mmd", "*.svg", "*.png"],
    "configuration": ["*.yaml", "*.json", "*.toml", "*.ini"]
}

try:
    # Run delegation
    result = orchestrate_delegation(...)
except DelegationTimeout:
    # Timeout exceeded
    # Return partial results if available
except DelegationError:
    # Subprocess crashed
    # Collect diagnostic information
except Exception:
    # Unexpected error
    # Cleanup and re-raise
finally:
    # Always cleanup subprocess
    cleanup_subprocess()

import asyncio
from amplihack.meta_delegation import run_meta_delegation

async def run_concurrent_delegations():
    """Run multiple delegations in parallel."""
    tasks = [
        run_meta_delegation(goal="Task 1", success_criteria="..."),
        run_meta_delegation(goal="Task 2", success_criteria="..."),
        run_meta_delegation(goal="Task 3", success_criteria="...")
    ]

    results = await asyncio.gather(*tasks)
    return results

from amplihack.meta_delegation import PlatformCLI, register_platform

class MyPlatformCLI(PlatformCLI):
    """Custom platform implementation."""

    def spawn_subprocess(self, goal, persona, working_dir, environment):
        # Launch your platform's CLI
        return subprocess.Popen([
            "my-platform-cli",
            "--goal", goal,
            "--persona", persona
        ], cwd=working_dir, env=environment)

    def format_prompt(self, goal, persona, context):
        # Format for your platform's input
        return f"[{persona}] {goal}\n\nContext:\n{context}"

    def parse_output(self, output):
        # Parse your platform's output format
        return {"stdout": output}

# Register platform
register_platform("my-platform", MyPlatformCLI())

# Use it
result = run_meta_delegation(
    goal="...",
    success_criteria="...",
    platform="my-platform"
)

from amplihack.meta_delegation import PersonaStrategy, register_persona

RESEARCHER = PersonaStrategy(
    name="researcher",
    communication_style="analytical",
    thoroughness_level="deep",
    evidence_collection_priority=[
        "documentation",
        "analysis_report",
        "data_file"
    ],
    prompt_template="""
    You are a researcher. Your goal is to investigate by:
    1. Gathering information systematically
    2. Analyzing data thoroughly
    3. Documenting findings clearly
    4. Drawing evidence-based conclusions

    Task: {goal}
    Success Criteria: {success_criteria}

    Research deeply and document your methodology.
    """
)

# Register persona
register_persona("researcher", RESEARCHER)

# Use it
result = run_meta_delegation(
    goal="...",
    success_criteria="...",
    persona_type="researcher"
)