"""
Pattern 30: Coordination Pattern

Agents coordinate activities, resources, and timing to achieve shared goals.
Manages dependencies, synchronization, and conflict resolution.
"""

from typing import Dict, Any, List, Optional, Set
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from typing_extensions import TypedDict


class TaskStatus(Enum):
    """Status of coordinated task."""
    PENDING = "pending"
    READY = "ready"
    IN_PROGRESS = "in_progress"
    BLOCKED = "blocked"
    COMPLETED = "completed"
    FAILED = "failed"


class ResourceStatus(Enum):
    """Status of shared resource."""
    AVAILABLE = "available"
    ALLOCATED = "allocated"
    IN_USE = "in_use"
    MAINTENANCE = "maintenance"


@dataclass
class CoordinatedTask:
    """Task requiring coordination."""
    task_id: str
    name: str
    assigned_to: Optional[str] = None
    dependencies: Set[str] = field(default_factory=set)
    required_resources: Set[str] = field(default_factory=set)
    status: TaskStatus = TaskStatus.PENDING
    started_at: Optional[datetime] = None
    completed_at: Optional[datetime] = None
    estimated_duration: timedelta = field(default_factory=lambda: timedelta(hours=1))


@dataclass
class SharedResource:
    """Shared resource requiring coordination."""
    resource_id: str
    name: str
    capacity: int
    current_users: Set[str] = field(default_factory=set)
    status: ResourceStatus = ResourceStatus.AVAILABLE
    
    def is_available(self) -> bool:
        """Check if resource is available."""
        return self.status == ResourceStatus.AVAILABLE and len(self.current_users)  str:
        """Request to execute a task."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name}, evaluating if you can execute a task."),
            ("human", f"Task: {task.name}\n"
                     f"Dependencies: {len(task.dependencies)} tasks\n"
                     f"Required resources: {task.required_resources}\n"
                     f"Estimated duration: {task.estimated_duration}\n\n"
                     f"Can you execute this task? Respond 'yes' or 'no' with brief reason.")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({})
        
        return str(response.content)[:150]
    
    def start_task(self, task: CoordinatedTask):
        """Start executing task."""
        self.assigned_tasks.add(task.task_id)
        task.status = TaskStatus.IN_PROGRESS
        task.started_at = datetime.now()
        print(f"   ▶️ {self.name} started: {task.name}")
    
    def complete_task(self, task: CoordinatedTask):
        """Complete task execution."""
        task.status = TaskStatus.COMPLETED
        task.completed_at = datetime.now()
        self.assigned_tasks.discard(task.task_id)
        print(f"   ✅ {self.name} completed: {task.name}")


class CoordinationPattern:
    """
    Coordination Pattern - Multi-agent activity coordination.
    
    Agents coordinate tasks, resources, and timing through a central
    coordinator that manages dependencies and resolves conflicts.
    """
    
    def __init__(self):
        self.agents: Dict[str, CoordinatedAgent] = {}
        self.tasks: Dict[str, CoordinatedTask] = {}
        self.resources: Dict[str, SharedResource] = {}
        self.workflow: Optional[StateGraph] = None
        self._build_workflow()
    
    def _build_workflow(self):
        """Build LangGraph coordination workflow."""
        workflow = StateGraph(CoordinationState)
        
        def check_dependencies(state: CoordinationState) -> CoordinationState:
            """Check task dependencies."""
            state["current_action"] = "checking_dependencies"
            return state
        
        def allocate_resources(state: CoordinationState) -> CoordinationState:
            """Allocate resources to tasks."""
            state["current_action"] = "allocating_resources"
            return state
        
        def execute_tasks(state: CoordinationState) -> CoordinationState:
            """Execute ready tasks."""
            state["current_action"] = "executing_tasks"
            return state
        
        # Add nodes
        workflow.add_node("check_deps", check_dependencies)
        workflow.add_node("allocate", allocate_resources)
        workflow.add_node("execute", execute_tasks)
        
        # Add edges
        workflow.set_entry_point("check_deps")
        workflow.add_edge("check_deps", "allocate")
        workflow.add_edge("allocate", "execute")
        workflow.add_edge("execute", END)
        
        self.workflow = workflow.compile()
    
    def register_agent(self, agent: CoordinatedAgent):
        """Register agent for coordination."""
        self.agents[agent.name] = agent
    
    def add_task(self, task: CoordinatedTask):
        """Add task to coordination."""
        self.tasks[task.task_id] = task
    
    def add_resource(self, resource: SharedResource):
        """Add shared resource."""
        self.resources[resource.resource_id] = resource
    
    def coordinate_execution(self):
        """Coordinate task execution."""
        print("\n   🔄 Coordinating task execution...")
        
        # Check dependencies and make tasks ready
        for task_id, task in self.tasks.items():
            if task.status == TaskStatus.PENDING:
                deps_met = all(
                    self.tasks.get(dep_id, CoordinatedTask("", "")).status == TaskStatus.COMPLETED
                    for dep_id in task.dependencies
                )
                
                if deps_met:
                    task.status = TaskStatus.READY
                    print(f"      ✓ {task.name} is ready (dependencies met)")
        
        # Allocate resources
        for task_id, task in self.tasks.items():
            if task.status == TaskStatus.READY:
                resources_available = all(
                    self.resources.get(res_id, SharedResource("", "", 0)).is_available()
                    for res_id in task.required_resources
                )
                
                if resources_available:
                    # Allocate resources
                    for res_id in task.required_resources:
                        resource = self.resources[res_id]
                        resource.status = ResourceStatus.ALLOCATED
                    
                    # Assign to agent
                    assigned = self._assign_task(task)
                    if assigned:
                        print(f"      ✓ {task.name} assigned to {assigned}")
                else:
                    task.status = TaskStatus.BLOCKED
                    print(f"      ⚠️ {task.name} blocked (resources unavailable)")
    
    def _assign_task(self, task: CoordinatedTask) -> Optional[str]:
        """Assign task to available agent."""
        for agent_name, agent in self.agents.items():
            if len(agent.assigned_tasks)  Dict[str, Any]:
        """Execute coordination pattern."""
        completed = sum(1 for t in self.tasks.values() 
                       if t.status == TaskStatus.COMPLETED)
        in_progress = sum(1 for t in self.tasks.values() 
                         if t.status == TaskStatus.IN_PROGRESS)
        blocked = sum(1 for t in self.tasks.values() 
                     if t.status == TaskStatus.BLOCKED)
        
        return {
            "pattern": "Coordination",
            "scenario": scenario,
            "total_agents": len(self.agents),
            "total_tasks": len(self.tasks),
            "completed_tasks": completed,
            "in_progress": in_progress,
            "blocked_tasks": blocked,
            "total_resources": len(self.resources)
        }


# Example usage
if __name__ == "__main__":
    print("=== Coordination Pattern Demo ===\n")
    
    # Create coordination pattern
    print("1. Creating coordination system...")
    pattern = CoordinationPattern()
    
    # Register agents
    agents = [
        CoordinatedAgent("Alice"),
        CoordinatedAgent("Bob"),
        CoordinatedAgent("Carol")
    ]
    
    for agent in agents:
        pattern.register_agent(agent)
    
    # Add shared resources
    print("\n2. Adding shared resources...")
    pattern.add_resource(SharedResource("db1", "Database", capacity=2))
    pattern.add_resource(SharedResource("api1", "API Server", capacity=3))
    pattern.add_resource(SharedResource("gpu1", "GPU Cluster", capacity=1))
    
    # Add tasks with dependencies
    print("\n3. Adding tasks with dependencies...")
    
    task1 = CoordinatedTask("t1", "Data Collection", 
                           required_resources={"db1"})
    task2 = CoordinatedTask("t2", "Data Processing",
                           dependencies={"t1"},
                           required_resources={"gpu1"})
    task3 = CoordinatedTask("t3", "Model Training",
                           dependencies={"t2"},
                           required_resources={"gpu1"})
    task4 = CoordinatedTask("t4", "API Development",
                           required_resources={"api1"})
    task5 = CoordinatedTask("t5", "Testing",
                           dependencies={"t3", "t4"},
                           required_resources={"api1", "db1"})
    
    for task in [task1, task2, task3, task4, task5]:
        pattern.add_task(task)
    
    # Round 1: Coordinate execution
    print("\n4. Coordination Round 1:")
    pattern.coordinate_execution()
    
    # Simulate completion of task1
    print("\n5. Simulating task completions...")
    pattern.complete_task_coordination("t1")
    pattern.complete_task_coordination("t4")
    
    # Round 2: More tasks should be ready
    print("\n6. Coordination Round 2:")
    pattern.coordinate_execution()
    
    # Complete more tasks
    pattern.complete_task_coordination("t2")
    
    # Round 3: Final tasks
    print("\n7. Coordination Round 3:")
    pattern.coordinate_execution()
    
    # Statistics
    print("\n8. Coordination Statistics:")
    summary = pattern.run("Software Development Project")
    print(f"   Total Tasks: {summary['total_tasks']}")
    print(f"   Completed: {summary['completed_tasks']}")
    print(f"   In Progress: {summary['in_progress']}")
    print(f"   Blocked: {summary['blocked_tasks']}")
    print(f"   Resources: {summary['total_resources']}")
    
    # Task timeline
    print("\n9. Task Status:")
    for task_id, task in pattern.tasks.items():
        status_icon = {"completed": "✅", "in_progress": "▶️", 
                      "blocked": "⚠️", "pending": "⏸️", "ready": "🟢"}.get(task.status.value, "❓")
        deps = f" (deps: {len(task.dependencies)})" if task.dependencies else ""
        assigned = f" -> {task.assigned_to}" if task.assigned_to else ""
        print(f"   {status_icon} {task.name}: {task.status.value}{deps}{assigned}")
    
    # Resource utilization
    print("\n10. Resource Utilization:")
    for res_id, resource in pattern.resources.items():
        usage = f"{len(resource.current_users)}/{resource.capacity}"
        print(f"   {resource.name}: {usage} users, status: {resource.status.value}")
    
    print("\n✓ Coordination pattern enabled multi-agent collaboration!")

    

Agentic Architecture Pattern: Coordination

Pattern 30: 30 Coordination – Line-by-Line Explanation

Overview

Pattern 30: Coordination Pattern

Agents coordinate activities, resources, and timing to achieve shared goals. Manages dependencies, synchronization, and conflict resolution.

Code Structure Analysis

File Information

• Pattern Number: 30
• Pattern Name: 30 Coordination
• File Path: patterns\collaboration\30_coordination.py
• Total Lines: 357

Detailed Code Explanation

Module Docstring

"""
Pattern 30: Coordination Pattern

Agents coordinate activities, resources, and timing to achieve shared goals.
Manages dependencies, synchronization, and conflict resolution.
"""

Explanation: This module docstring describes the purpose and key features of the 30 Coordination pattern.

Imports Section

from typing import Dict, Any, List, Optional, Set
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from typing_extensions import TypedDict  # First 20 imports shown

Explanation: The module imports necessary dependencies: – from typing import Dict, Any, List, Optional, Set: Type hints for better code documentation – from dataclasses import dataclass, field: Data class decorators for clean data structures – from datetime import datetime, timedelta: Date and time utilities – from enum import Enum: Enumeration support for fixed sets of values – from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration – from langchain_openai import ChatOpenAI: LangChain framework integration – from langgraph.graph import StateGraph, END: LangGraph for workflow orchestration – from typing_extensions import TypedDict: Type hints for better code documentation

Enumerations

TaskStatus (Line 18)

class TaskStatus(Enum):
    PENDING
    READY
    IN_PROGRESS
    BLOCKED
    COMPLETED
    FAILED

Purpose: Status of coordinated task.

Values: – PENDING: Initial state, not yet processed – READY: Represents ready – IN_PROGRESS: Represents in progress – BLOCKED: Represents blocked – COMPLETED: Successfully finished – FAILED: Execution failed

ResourceStatus (Line 28)

class ResourceStatus(Enum):
    AVAILABLE
    ALLOCATED
    IN_USE
    MAINTENANCE

Purpose: Status of shared resource.

Values: – AVAILABLE: Represents available – ALLOCATED: Represents allocated – IN_USE: Represents in use – MAINTENANCE: Represents maintenance

Classes

TaskStatus (Line 18)

Description: Status of coordinated task.

ResourceStatus (Line 28)

Description: Status of shared resource.

CoordinatedTask (Line 37)

Description: Task requiring coordination.

SharedResource (Line 51)

Description: Shared resource requiring coordination.

Methods:

is_available (Line 59)

def is_available(self):
    """Check if resource is available."""

Purpose: Check if resource is available.

Parameters: – self: Instance reference

CoordinationState (Line 64)

Description: State for coordination workflow.

CoordinatedAgent (Line 72)

Description: Agent participating in coordination.

Methods:

__init__ (Line 75)

def __init__(self, name, model_name):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – name: Identifier or label – model_name: LLM model to use (e.g., “gpt-4”)

request_task (Line 81)

def request_task(self, task):
    """Request to execute a task."""

Purpose: Request to execute a task.

Parameters: – self: Instance reference – task: Parameter for task

start_task (Line 97)

def start_task(self, task):
    """Start executing task."""

Purpose: Start executing task.

Parameters: – self: Instance reference – task: Parameter for task

complete_task (Line 104)

def complete_task(self, task):
    """Complete task execution."""

Purpose: Complete task execution.

Parameters: – self: Instance reference – task: Parameter for task

CoordinationPattern (Line 112)

Description: Coordination Pattern – Multi-agent activity coordination.

Agents coordinate tasks, resources, and timing through a central coordinator that manages dependencies and resolves conflicts.

Methods:

__init__ (Line 120)

def __init__(self):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference

_build_workflow (Line 127)

def _build_workflow(self):
    """Build LangGraph coordination workflow."""

Purpose: Build LangGraph coordination workflow.

Parameters: – self: Instance reference

register_agent (Line 159)

def register_agent(self, agent):
    """Register agent for coordination."""

Purpose: Register agent for coordination.

Parameters: – self: Instance reference – agent: Parameter for agent

add_task (Line 163)

def add_task(self, task):
    """Add task to coordination."""

Purpose: Add task to coordination.

Parameters: – self: Instance reference – task: Parameter for task

add_resource (Line 167)

def add_resource(self, resource):
    """Add shared resource."""

Purpose: Add shared resource.

Parameters: – self: Instance reference – resource: Parameter for resource

coordinate_execution (Line 171)

def coordinate_execution(self):
    """Coordinate task execution."""

Purpose: Coordinate task execution.

Parameters: – self: Instance reference

_assign_task (Line 209)

def _assign_task(self, task):
    """Assign task to available agent."""

Purpose: Assign task to available agent.

Parameters: – self: Instance reference – task: Parameter for task

complete_task_coordination (Line 228)

def complete_task_coordination(self, task_id):
    """Complete task and release resources."""

Purpose: Complete task and release resources.

Parameters: – self: Instance reference – task_id: Parameter for task_id

run (Line 247)

def run(self, scenario):
    """Execute coordination pattern."""

Purpose: Execute coordination pattern.

Parameters: – self: Instance reference – scenario: Parameter for scenario

Design Patterns Used

• Strategy Pattern: Different execution strategies for various scenarios
• State Pattern: Managing different states in the workflow
• Chain of Responsibility: Sequential processing through workflow nodes
• Observer Pattern: Tracking and monitoring pattern execution
• Factory Pattern: Creating instances with proper configuration

Usage Example

# Initialize the pattern
pattern = TaskStatus(name="example", model_name="gpt-4")

# Configure and execute
result = pattern.execute(data={"key": "value"})

# Check results
print(result)

Key Concepts

• LLM Integration: Uses LangChain/OpenAI for intelligent decision-making
• Workflow Orchestration: LangGraph manages execution flow through nodes
• State Management: TypedDict defines workflow state structure
• Error Handling: Comprehensive error handling and fallback mechanisms
• Extensibility: Designed for easy customization and extension

Extension Points

This pattern can be extended by: 1. Adding custom workflow nodes 2. Implementing additional validation logic 3. Integrating with external services 4. Customizing LLM prompts and responses 5. Adding monitoring and logging capabilities