# ReAct agent loop
async def react_loop(question: str, tools: dict, max_steps: int = 10) -> str:
    history = REACT_PROMPT.format(tools=list(tools.keys()), question=question)
    for step in range(max_steps):
        response = await llm.chat([{"role": "user", "content": history}])
        if "Final Answer:" in response.content:
            return response.content.split("Final Answer:")[-1].strip()
        if "Action:" in response.content:
            action = parse_action(response.content)
            result = await tools[action.name](*action.args)
            history += f"\nObservation: {result}\n"
    return "Max steps reached without answer"

# Supervisor with fan-out/fan-in
async def multi_agent_analysis(content: str) -> dict:
    agents = [("security", security_agent), ("perf", perf_agent)]
    tasks = [agent(content) for _, agent in agents]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    return await synthesize_findings(results)

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.conditions import TextMentionTermination
from autogen_ext.models.openai import OpenAIChatCompletionClient

# Create model client
model_client = OpenAIChatCompletionClient(model="gpt-5.2")

# Define agents
planner = AssistantAgent(
    name="planner",
    description="Plans complex tasks and breaks them into steps",
    model_client=model_client,
    system_message="You are a planning expert. Break tasks into actionable steps."
)

executor = AssistantAgent(
    name="executor",
    description="Executes planned tasks",
    model_client=model_client,
    system_message="You execute tasks according to the plan."
)

reviewer = AssistantAgent(
    name="reviewer",
    description="Reviews work and provides feedback",
    model_client=model_client,
    system_message="You review work. Say 'APPROVED' if satisfactory."
)

# Create team with termination condition
termination = TextMentionTermination("APPROVED")
team = RoundRobinGroupChat(
    participants=[planner, executor, reviewer],
    termination_condition=termination
)

# Run team
result = await team.run(task="Create a marketing strategy")

from autogen_agentchat.teams import SelectorGroupChat

# Selector chooses next speaker based on context
team = SelectorGroupChat(
    participants=[analyst, writer, reviewer],
    model_client=model_client,
    termination_condition=termination
)

from autogen_agentchat.conditions import (
    TextMentionTermination,
    MaxMessageTermination,
    TokenUsageTermination,
    TimeoutTermination,
    OrTerminationCondition,
)

termination = OrTerminationCondition(
    TextMentionTermination("DONE"),
    MaxMessageTermination(max_messages=20),
    TimeoutTermination(timeout_seconds=300)
)

from autogen_core.tools import FunctionTool

def search_database(query: str) -> str:
    """Search the database for information."""
    results = db.search(query)
    return json.dumps(results)

search_tool = FunctionTool(search_database, description="Search the database")

researcher = AssistantAgent(
    name="researcher",
    description="Researches information",
    model_client=model_client,
    tools=[search_tool],
    system_message="Use the search tool to find information."
)

# Save state
state = await team.save_state()

# Restore state
await team.load_state(state)

# Resume conversation
result = await team.run(task="Continue from where we left off")

from autogen_agentchat.protocols import A2AProtocol

protocol = A2AProtocol(
    agent=my_agent,
    endpoint="https://api.example.com/agent",
    auth_token=os.environ["A2A_TOKEN"]
)

response = await protocol.send(
    to="external-agent-id",
    message="Process this request"
)

async for message in team.run_stream(task="Analyze this data"):
    print(f"{message.source}: {message.content}")

from agents import Agent, Runner, handoff, RunConfig
from agents.extensions.handoff_prompt import RECOMMENDED_PROMPT_PREFIX

researcher_agent = Agent(
    name="researcher",
    instructions=f"""{RECOMMENDED_PROMPT_PREFIX}
You are a research specialist. Gather information and facts.
When research is complete, hand off to the writer.""",
    model="gpt-5.2"
)

writer_agent = Agent(
    name="writer",
    instructions=f"""{RECOMMENDED_PROMPT_PREFIX}
You are a content writer. Create compelling content from research.""",
    model="gpt-5.2"
)

orchestrator = Agent(
    name="orchestrator",
    instructions=f"""{RECOMMENDED_PROMPT_PREFIX}
You coordinate research and writing tasks.""",
    model="gpt-5.2",
    handoffs=[handoff(agent=researcher_agent), handoff(agent=writer_agent)]
)

async def run_workflow(task: str):
    runner = Runner()
    config = RunConfig(nest_handoff_history=True)
    result = await runner.run(orchestrator, task, run_config=config)
    return result.final_output

# Old AutoGen 0.2
# from autogen import AssistantAgent, UserProxyAgent

# New AutoGen 0.4+
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat

# Key differences:
# - No UserProxyAgent needed for simple tasks
# - Teams replace GroupChat
# - Explicit termination conditions required
# - Model client separate from agent

team = RoundRobinGroupChat(
    participants=[planner, executor, reviewer]
    # No termination condition - infinite loop risk
)
result = await team.run(task="Complete task")

termination = OrTerminationCondition(
    TextMentionTermination("APPROVED"),
    MaxMessageTermination(max_messages=20)
)
team = RoundRobinGroupChat(
    participants=[planner, executor, reviewer],
    termination_condition=termination
)
result = await team.run(task="Complete task")

Start
  |
  +-- Need complex state machines?
  |     +-- Yes --> LangGraph
  |     +-- No
  |           |
  +-- Role-based collaboration?
  |     +-- Yes --> CrewAI
  |     +-- No
  |           |
  +-- OpenAI ecosystem only?
  |     +-- Yes --> OpenAI Agents SDK
  |     +-- No
  |           |
  +-- Enterprise requirements?
  |     +-- Yes --> Microsoft Agent Framework
  |     +-- No
  |           |
  +-- Long-horizon coding tasks?
  |     +-- Yes --> GPT-5.2-Codex
  |     +-- No
  |           |
  +-- Open-source priority?
        +-- Yes --> AG2
        +-- No --> LangGraph (default)

# AutoGen 0.2 (old)
from autogen import AssistantAgent, UserProxyAgent
agent = AssistantAgent(name="assistant", llm_config=config)

# MS Agent Framework (new)
from autogen_agentchat.agents import AssistantAgent
from autogen_ext.models.openai import OpenAIChatCompletionClient
model_client = OpenAIChatCompletionClient(model="gpt-5.2")
agent = AssistantAgent(name="assistant", model_client=model_client)

# Custom orchestration (old)
async def workflow(task):
    step1 = await agent1.run(task)
    step2 = await agent2.run(step1)
    return step2

# LangGraph (new)
from langgraph.graph import StateGraph
workflow = StateGraph(State)
workflow.add_node("agent1", agent1_node)
workflow.add_node("agent2", agent2_node)
workflow.add_edge("agent1", "agent2")

Task Duration > 1 hour?
  +-- Yes --> GPT-5.2-Codex
  +-- No
        +-- Multiple files affected?
        |     +-- Yes --> GPT-5.2-Codex
        |     +-- No
        |           +-- Security review needed?
        |                 +-- Yes --> GPT-5.2-Codex
        |                 +-- No --> GPT-5.2 (standard)

# Overkill: LangGraph for single-step task
workflow = StateGraph(State)
workflow.add_node("summarize", summarize_node)
workflow.add_edge(START, "summarize")
workflow.add_edge("summarize", END)
app = workflow.compile()
result = await app.ainvoke({"text": "..."})

# Simple task = single agent
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-5.2")
result = await llm.ainvoke("Summarize this text: ...")

from crewai import Agent, Crew, Task, Process
from crewai.flow.flow import Flow, listen, start

# Manager coordinates the team
manager = Agent(
    role="Project Manager",
    goal="Coordinate team efforts and ensure project success",
    backstory="Experienced project manager skilled at delegation",
    allow_delegation=True,
    memory=True,
    verbose=True
)

# Specialist agents
researcher = Agent(
    role="Researcher",
    goal="Provide accurate research and analysis",
    backstory="Expert researcher with deep analytical skills",
    allow_delegation=False,
    verbose=True
)

writer = Agent(
    role="Writer",
    goal="Create compelling content",
    backstory="Skilled writer who creates engaging content",
    allow_delegation=False,
    verbose=True
)

# Manager-led task
project_task = Task(
    description="Create a comprehensive market analysis report",
    expected_output="Executive summary, analysis, recommendations",
    agent=manager
)

# Hierarchical crew
crew = Crew(
    agents=[manager, researcher, writer],
    tasks=[project_task],
    process=Process.hierarchical,
    manager_llm="gpt-5.2",
    memory=True,
    verbose=True
)

result = crew.kickoff()

from crewai.flow.flow import Flow, listen, start, router

class ResearchFlow(Flow):
    @start()
    def generate_topic(self):
        return "AI Safety"

    @listen(generate_topic)
    def research_topic(self, topic):
        return f"Research findings on {topic}"

    @router(research_topic)
    def route_result(self, result):
        if "sufficient" in result:
            return "success"
        return "retry"

    @listen("success")
    def handle_success(self):
        return "Workflow completed"

    @listen("retry")
    def handle_retry(self):
        return "Retrying..."

flow = ResearchFlow()
result = flow.kickoff()

from crewai.flow.flow import Flow, listen, start, and_, or_

class ParallelFlow(Flow):
    @start()
    def task_a(self):
        return "Result A"

    @start()
    def task_b(self):
        return "Result B"

    @listen(and_(task_a, task_b))
    def combine_results(self):
        """Triggers when BOTH complete"""
        return "Combined results"

from pydantic import BaseModel

class ReportOutput(BaseModel):
    title: str
    summary: str
    findings: list[str]
    confidence: float

task = Task(
    description="Analyze market trends",
    expected_output="Structured market analysis",
    agent=analyst,
    output_pydantic=ReportOutput
)

result = crew.kickoff()
report = result.pydantic

from crewai.tasks import TaskOutput

def validate_length(result: TaskOutput) -> tuple[bool, any]:
    if len(result.raw.split()) < 100:
        return (False, "Content too brief, expand analysis")
    return (True, result.raw)

task = Task(
    description="Write comprehensive analysis",
    expected_output="Detailed analysis (100+ words)",
    agent=writer,
    guardrail=validate_length,
    guardrail_max_retries=3
)

from crewai import Agent

agent = Agent(
    role="Research Analyst",
    goal="Research and analyze information",
    backstory="Expert analyst",
    mcps=[
        "https://mcp.example.com/mcp?api_key=your_key",
        "crewai-amp:financial-data",
    ]
)

from crewai import Agent, Crew, Task, CrewBase, agent, task, crew

@CrewBase
class ResearchCrew:
    agents_config = 'config/agents.yaml'
    tasks_config = 'config/tasks.yaml'

    @agent
    def researcher(self) -> Agent:
        return Agent(config=self.agents_config['researcher'], tools=[search_tool])

    @task
    def research_task(self) -> Task:
        return Task(config=self.tasks_config['research'])

    @crew
    def crew(self) -> Crew:
        return Crew(agents=self.agents, tasks=self.tasks, process=Process.sequential)

result = ResearchCrew().crew().kickoff(inputs={"topic": "AI Safety"})

crew = Crew(
    agents=[researcher, writer, reviewer],
    tasks=[research_task, write_task, review_task],
    process=Process.sequential  # No delegation, rigid order
)

crew = Crew(
    agents=[manager, researcher, writer, reviewer],
    tasks=[project_task],
    process=Process.hierarchical,  # Manager delegates dynamically
    manager_llm="gpt-5.2"
)

async def plan_and_execute(goal: str) -> str:
    """Create plan first, then execute steps."""
    # 1. Generate plan
    plan = await llm.chat([{
        "role": "user",
        "content": f"Create a step-by-step plan to: {goal}\n\nFormat as numbered list."
    }])

    steps = parse_plan(plan.content)
    results = []

    # 2. Execute each step
    for i, step in enumerate(steps):
        result = await execute_step(step, context=results)
        results.append({"step": step, "result": result})

        # 3. Check if replanning needed
        if should_replan(results):
            return await plan_and_execute(
                f"{goal}\n\nProgress so far: {results}"
            )

    # 4. Synthesize final answer
    return await synthesize(goal, results)

interface AgentStep {
  thought: string;
  action?: string;
  actionInput?: unknown;
  observation?: string;
}

interface AgentResult {
  answer: string;
  steps: AgentStep[];
  totalCost: number;
  iterations: number;
}

export async function reactAgent(
  task: string,
  options: { maxIterations?: number; verbose?: boolean } = {}
): Promise<AgentResult> {
  const { maxIterations = 10, verbose = false } = options;
  const steps: AgentStep[] = [];
  let totalCost = 0;

  const systemPrompt = `You are an autonomous agent that can use tools.

Available tools:
${tools.map(t => `- ${t.name}: ${t.description}`).join('\n')}

Use this format:
Thought: [Your reasoning]
Action: [tool name]
Action Input: {"param": "value"}
Observation: [Tool result]

When done: Answer: [Final answer]`;

  const messages = [
    { role: 'system' as const, content: systemPrompt },
    { role: 'user' as const, content: task }
  ];

  for (let i = 0; i < maxIterations; i++) {
    const response = await openai.chat.completions.create({
      model: 'gpt-5.2', messages, temperature: 0.1
    });

    const content = response.choices[0].message.content!;
    totalCost += (response.usage!.total_tokens / 1000) * 0.01;

    if (content.includes('Answer:')) {
      const answer = content.split('Answer:')[1].trim();
      return { answer, steps, totalCost, iterations: i + 1 };
    }

    const parsed = parseAgentResponse(content);
    const step: AgentStep = { thought: parsed.thought };

    if (parsed.action && parsed.actionInput) {
      step.action = parsed.action;
      step.actionInput = JSON.parse(parsed.actionInput);
      step.observation = await executeTool(parsed.action, step.actionInput);
    }

    steps.push(step);
    messages.push({ role: 'assistant', content });
    if (step.observation) {
      messages.push({ role: 'user', content: `Observation: ${step.observation}` });
    }
  }

  throw new Error(`Agent exceeded max iterations (${maxIterations})`);
}

export async function functionCallingAgent(task: string): Promise<AgentResult> {
  const steps: AgentStep[] = [];
  let totalCost = 0;

  const messages = [
    { role: 'system' as const, content: 'You are a helpful assistant with access to tools.' },
    { role: 'user' as const, content: task }
  ];

  const openaiTools = tools.map(tool => ({
    type: 'function' as const,
    function: { name: tool.name, description: tool.description, parameters: tool.parameters }
  }));

  let iteration = 0;
  while (iteration < 10) {
    const response = await openai.chat.completions.create({
      model: 'gpt-5.2', messages, tools: openaiTools
    });

    const message = response.choices[0].message;
    totalCost += (response.usage!.total_tokens / 1000) * 0.01;

    // No tool calls = final answer
    if (!message.tool_calls) {
      return { answer: message.content!, steps, totalCost, iterations: iteration + 1 };
    }

    messages.push(message as any);
    for (const toolCall of message.tool_calls) {
      const tool = tools.find(t => t.name === toolCall.function.name);
      if (!tool) continue;
      const args = JSON.parse(toolCall.function.arguments);
      const result = await tool.execute(args);
      steps.push({
        thought: `Calling ${toolCall.function.name}`,
        action: toolCall.function.name,
        actionInput: args,
        observation: JSON.stringify(result)
      });
      messages.push({ role: 'tool', tool_call_id: toolCall.id, content: JSON.stringify(result) });
    }
    iteration++;
  }

  throw new Error('Agent exceeded max iterations');
}

async def execute_goal(goal: str):
    steps = ["step1", "step2", "step3"]  # Hardcoded, no reasoning
    for step in steps:
        await execute_step(step)

async def plan_and_execute(goal: str):
    plan = await llm.chat([{"role": "user", "content": f"Create plan for: {goal}"}])
    steps = parse_plan(plan.content)
    for step in steps:
        result = await execute_step(step)
        if should_replan(result):
            return await plan_and_execute(f"{goal}\nProgress: {result}")

REACT_PROMPT = """You are an agent that reasons step by step.

For each step, respond with:
Thought: [your reasoning about what to do next]
Action: [tool_name(arg1, arg2)]
Observation: [you'll see the result here]

When you have the final answer:
Thought: I now have enough information
Final Answer: [your response]

Available tools: {tools}

Question: {question}
"""

async def react_loop(question: str, tools: dict, max_steps: int = 10) -> str:
    """Execute ReAct reasoning loop."""
    history = REACT_PROMPT.format(tools=list(tools.keys()), question=question)

    for step in range(max_steps):
        response = await llm.chat([{"role": "user", "content": history}])
        history += response.content

        # Check for final answer
        if "Final Answer:" in response.content:
            return response.content.split("Final Answer:")[-1].strip()

        # Extract and execute action
        if "Action:" in response.content:
            action = parse_action(response.content)
            result = await tools[action.name](*action.args)
            history += f"\nObservation: {result}\n"

    return "Max steps reached without answer"

async def self_correcting_agent(task: str, max_retries: int = 3) -> str:
    """Agent that validates and corrects its own output."""
    for attempt in range(max_retries):
        response = await llm.chat([{"role": "user", "content": task}])

        # Self-validate
        validation = await llm.chat([{
            "role": "user",
            "content": f"""Validate this response for the task: {task}

Response: {response.content}

Check for:
1. Correctness - Is it factually accurate?
2. Completeness - Does it fully answer the task?
3. Format - Is it properly formatted?

If valid, respond: VALID
If invalid, respond: INVALID: [what's wrong and how to fix]"""
        }])

        if "VALID" in validation.content:
            return response.content

        # Correct based on feedback
        task = f"{task}\n\nPrevious attempt had issues: {validation.content}"

    return response.content  # Return best attempt

class AgentMemory:
    """Sliding window memory for agents."""

    def __init__(self, max_messages: int = 20):
        self.messages = []
        self.max_messages = max_messages
        self.summary = ""

    def add(self, role: str, content: str):
        self.messages.append({"role": role, "content": content})
        if len(self.messages) > self.max_messages:
            self._compress()

    def _compress(self):
        """Summarize oldest messages."""
        old = self.messages[:10]
        self.messages = self.messages[10:]
        summary = summarize(old)
        self.summary = f"{self.summary}\n{summary}"

    def get_context(self) -> list:
        """Get messages with summary prefix."""
        context = []
        if self.summary:
            context.append({
                "role": "system",
                "content": f"Previous context summary: {self.summary}"
            })
        return context + self.messages

async def react_loop(question: str, tools: dict):
    history = f"Question: {question}"
    while True:  # Infinite loop risk
        response = await llm.chat([{"role": "user", "content": history}])
        if "Final Answer:" in response.content:
            return response.content

async def react_loop(question: str, tools: dict, max_steps: int = 10):
    history = f"Question: {question}"
    for step in range(max_steps):  # Bounded loop
        response = await llm.chat([{"role": "user", "content": history}])
        if "Final Answer:" in response.content:
            return response.content.split("Final Answer:")[-1]
    return "Max steps reached without answer"

async def resolve_conflicts(findings: list[dict]) -> list[dict]:
    """When agents disagree, resolve by confidence or LLM."""
    conflicts = detect_conflicts(findings)

    if not conflicts:
        return findings

    for conflict in conflicts:
        # Option 1: Higher confidence wins
        winner = max(conflict.agents, key=lambda a: a.confidence)

        # Option 2: LLM arbitration
        resolution = await llm.chat([{
            "role": "user",
            "content": f"""Two agents disagree:

Agent A ({conflict.agent_a.name}): {conflict.agent_a.finding}
Agent B ({conflict.agent_b.name}): {conflict.agent_b.finding}

Which is more likely correct and why?"""
        }])

        conflict.resolution = parse_resolution(resolution.content)

    return apply_resolutions(findings, conflicts)

async def resolve_agent_conflicts(
    findings: list[dict], llm: Any
) -> dict:
    """Resolve conflicts between agent outputs."""
    conflicts = []
    for i, f1 in enumerate(findings):
        for f2 in findings[i+1:]:
            if f1.get("recommendation") != f2.get("recommendation"):
                conflicts.append((f1, f2))

    if not conflicts:
        return {"status": "no_conflicts", "findings": findings}

    # LLM arbitration
    resolution = await llm.ainvoke(f"""
        Agents disagree. Determine best recommendation:
        Agent 1: {conflicts[0][0]}
        Agent 2: {conflicts[0][1]}
        Provide: winner, reasoning, confidence (0-1)
    """)
    return {"status": "resolved", "resolution": resolution}

class AgentBus:
    """Message passing between agents."""

    def __init__(self):
        self.messages = []
        self.subscribers = {}

    def publish(self, from_agent: str, message: dict):
        """Broadcast message to all agents."""
        msg = {"from": from_agent, "data": message, "ts": time.time()}
        self.messages.append(msg)
        for callback in self.subscribers.values():
            callback(msg)

    def subscribe(self, agent_id: str, callback):
        """Register agent to receive messages."""
        self.subscribers[agent_id] = callback

    def get_history(self, agent_id: str = None) -> list:
        """Get message history, optionally filtered."""
        if agent_id:
            return [m for m in self.messages if m["from"] == agent_id]
        return self.messages

async def multi_agent_analysis(task: str):
    results = await asyncio.gather(agent1(task), agent2(task))
    return results  # Return conflicting results without resolution

async def multi_agent_analysis(task: str):
    results = await asyncio.gather(agent1(task), agent2(task))
    if results[0] != results[1]:  # Conflict detected
        resolution = await llm.chat([{
            "role": "user",
            "content": f"Agent 1: {results[0]}\nAgent 2: {results[1]}\nWhich is correct?"
        }])
        return resolution.content
    return results[0]

async def multi_agent_analysis(content: str) -> dict:
    """Fan-out to specialists, fan-in to synthesize."""
    agents = [
        ("security", security_agent),
        ("performance", performance_agent),
        ("code_quality", quality_agent),
        ("architecture", architecture_agent),
    ]

    # Fan-out: Run all agents in parallel
    tasks = [agent(content) for _, agent in agents]
    results = await asyncio.gather(*tasks, return_exceptions=True)

    # Filter successful results
    findings = [
        {"agent": name, "result": result}
        for (name, _), result in zip(agents, results)
        if not isinstance(result, Exception)
    ]

    # Fan-in: Synthesize findings
    return await synthesize_findings(findings)

class Supervisor:
    """Central coordinator that routes to specialists."""

    def __init__(self, agents: dict):
        self.agents = agents  # {"security": agent, "performance": agent}
        self.completed = []

    async def run(self, task: str) -> dict:
        """Route task through appropriate agents."""
        # 1. Determine which agents to use
        plan = await self.plan_routing(task)

        # 2. Execute in dependency order
        results = {}
        for agent_name in plan.execution_order:
            if plan.can_parallelize(agent_name):
                batch = plan.get_parallel_batch(agent_name)
                batch_results = await asyncio.gather(*[
                    self.agents[name](task, context=results)
                    for name in batch
                ])
                results.update(dict(zip(batch, batch_results)))
            else:
                results[agent_name] = await self.agents[agent_name](
                    task, context=results
                )

        return results

    async def plan_routing(self, task: str) -> RoutingPlan:
        """Use LLM to determine agent routing."""
        response = await llm.chat([{
            "role": "user",
            "content": f"""Task: {task}

Available agents: {list(self.agents.keys())}

Which agents should handle this task?
What order? Can any run in parallel?"""
        }])
        return parse_routing_plan(response.content)

class SupervisorCoordinator:
    """Central supervisor that routes tasks to worker agents."""

    def __init__(self, workers: dict[str, Agent]):
        self.workers = workers
        self.execution_log: list[dict] = []

    async def route_and_execute(
        self, task: str, required_agents: list[str], parallel: bool = True
    ) -> dict[str, Any]:
        context = {"task": task, "results": {}}

        if parallel:
            tasks = [self._run_worker(name, task, context) for name in required_agents]
            results = await asyncio.gather(*tasks, return_exceptions=True)
            return dict(zip(required_agents, results))
        else:
            for name in required_agents:
                context["results"][name] = await self._run_worker(name, task, context)
            return context["results"]

    async def _run_worker(self, name: str, task: str, context: dict) -> dict:
        """Execute single worker with timeout."""
        try:
            result = await asyncio.wait_for(
                self.workers[name].run(task, context), timeout=30.0
            )
            self.execution_log.append({"agent": name, "status": "success", "result": result})
            return result
        except asyncio.TimeoutError:
            return {"error": f"{name} timed out"}

Star (Task tool):              Mesh (Agent Teams):
      Lead                           Lead (delegate)
     /||\                          /  |  \
    / || \                        /   |   \
   A  B  C  D                   A <-> B <-> C
   (no cross-talk)              (peer messaging)

Complexity Assessment:
+-- Score < 3.0  -> Task tool subagents (cheaper, simpler)
+-- Score 3.0-3.5 -> User choice (recommend Teams for cross-cutting)
+-- Score > 3.5  -> Agent Teams (if CLAUDE_CODE_EXPERIMENTAL_AGENT_TEAMS=1)

# 1. Create team with shared task list
TeamCreate(team_name="feature-auth", description="User auth implementation")

# 2. Create tasks in shared list
TaskCreate(subject="Design API schema", description="...")
TaskCreate(subject="Build React components", description="...", addBlockedBy=["1"])

# 3. Spawn teammates
Task(prompt="You are the backend architect...",
     team_name="feature-auth", name="backend-dev",
     subagent_type="backend-system-architect")

# Direct message (default)
SendMessage(type="message", recipient="frontend-dev",
  content="API contract: GET /users/:id -> {id, name, email}",
  summary="API contract ready")

# Broadcast (expensive -- use sparingly)
SendMessage(type="broadcast",
  content="Auth header format changed to Bearer",
  summary="Breaking auth change")

async def analyze(content: str):
    security_result = await security_agent(content)  # Wait
    perf_result = await performance_agent(content)   # Wait
    quality_result = await quality_agent(content)    # Wait
    return [security_result, perf_result, quality_result]

async def analyze(content: str):
    tasks = [
        security_agent(content),
        performance_agent(content),
        quality_agent(content)
    ]
    results = await asyncio.gather(*tasks)  # Run in parallel
    return results

async def synthesize_findings(findings: list[dict]) -> dict:
    """Combine multiple agent outputs into coherent result."""
    # Group by category
    by_category = {}
    for f in findings:
        cat = f.get("category", "general")
        by_category.setdefault(cat, []).append(f)

    # Synthesize each category
    synthesis = await llm.chat([{
        "role": "user",
        "content": f"""Synthesize these agent findings into a coherent summary:

{json.dumps(by_category, indent=2)}

Output format:
- Executive summary (2-3 sentences)
- Key findings by category
- Recommendations
- Confidence score (0-1)"""
    }])

    return parse_synthesis(synthesis.content)

export async function multiAgentCollaboration(
  task: string,
  agents: Agent[]
): Promise<AgentResult> {
  const steps: AgentStep[] = [];
  let totalCost = 0;

  // 1. Coordinator plans the task
  const planResponse = await openai.chat.completions.create({
    model: 'gpt-5.2',
    messages: [
      {
        role: 'system',
        content: `You are a coordinator. Break down tasks and assign to agents:
${agents.map(a => `- ${a.name}: ${a.role}`).join('\n')}

Provide a numbered plan with agent assignments.`
      },
      { role: 'user', content: `Task: ${task}\n\nProvide a step-by-step plan.` }
    ]
  });

  const plan = planResponse.choices[0].message.content!;
  steps.push({ thought: 'Coordinator planning', observation: plan });

  // 2. Execute agent subtasks in parallel
  const agentResults = await Promise.all(
    agents.map(async (agent) => {
      const response = await openai.chat.completions.create({
        model: 'gpt-5.2',
        messages: [
          { role: 'system', content: agent.systemPrompt },
          { role: 'user', content: `Task: ${task}\n\nPlan:\n${plan}\n\nComplete your part.` }
        ]
      });
      return { agent: agent.name, result: response.choices[0].message.content! };
    })
  );

  // 3. Synthesize results
  const synthesisResponse = await openai.chat.completions.create({
    model: 'gpt-5.2',
    messages: [
      { role: 'system', content: 'Synthesize agent results into a coherent final answer.' },
      { role: 'user', content: `Task: ${task}\n\nAgent Results:\n${JSON.stringify(agentResults, null, 2)}` }
    ]
  });

  return {
    answer: synthesisResponse.choices[0].message.content!,
    steps,
    totalCost,
    iterations: agents.length + 2
  };
}

{
  "quality_comparison": {
    "security_agent": {"score": 0.92, "findings": 3},
    "perf_agent": {"score": 0.88, "findings": 5}
  },
  "consensus_items": ["Use parameterized queries", "Add rate limiting"],
  "disagreements": []
}

{
  "success_patterns": ["Caching strategy effective", "Batch size 50+ preferred"],
  "failure_patterns": ["Timeout at >5000 items per batch"]
}

{
  "priority_1": "Fix SQL injection in auth module (security + quality agree)",
  "priority_2": "Add connection pooling (performance agent)",
  "confidence": 0.91
}

async def multi_agent_analysis(task: str):
    results = await asyncio.gather(agent1(task), agent2(task), agent3(task))
    return results  # Returns list of disconnected findings

async def multi_agent_analysis(task: str):
    results = await asyncio.gather(agent1(task), agent2(task), agent3(task))
    synthesis = await llm.chat([{
        "role": "user",
        "content": f"Synthesize these findings: {json.dumps(results)}"
    }])
    return parse_synthesis(synthesis.content)

+---------------------------------------------------------------------+
|                   MULTI-SCENARIO ORCHESTRATOR                        |
+---------------------------------------------------------------------+
|  [Coordinator] --+--> [Scenario 1: Simple]       (Easy)             |
|       ^          |      +--> [Skill Instance 1]                     |
|       |          +--> [Scenario 2: Medium]       (Intermediate)     |
|       |          |      +--> [Skill Instance 2]                     |
|       |          +--> [Scenario 3: Complex]      (Advanced)         |
|       |                 +--> [Skill Instance 3]                     |
|       |                                                             |
|   [State Manager] <---- All instances report progress               |
|   [Aggregator] --> Cross-scenario synthesis                         |
+---------------------------------------------------------------------+

from langgraph.graph import StateGraph, START, END
from langgraph.types import Command, Send

async def scenario_supervisor(state: ScenarioOrchestratorState) -> list[Command]:
    """Route to all 3 scenarios in parallel."""
    for scenario_id in ["simple", "medium", "complex"]:
        state[f"progress_{scenario_id}"] = ScenarioProgress(
            scenario_id=scenario_id, status="pending",
            start_time_ms=int(time.time() * 1000)
        )

    return [
        Send("scenario_worker", {"scenario_id": "simple", **state}),
        Send("scenario_worker", {"scenario_id": "medium", **state}),
        Send("scenario_worker", {"scenario_id": "complex", **state}),
    ]

async def scenario_worker(state: ScenarioOrchestratorState) -> dict:
    """Execute one scenario."""
    scenario_id = state.get("scenario_id")
    progress = state[f"progress_{scenario_id}"]
    scenario_def = state[f"scenario_{scenario_id}"]

    progress.status = "running"
    try:
        result = await execute_skill_with_milestones(
            skill_name=state["skill_name"],
            scenario_def=scenario_def,
            progress=progress, state=state
        )
        progress.status = "complete"
        progress.elapsed_ms = int(time.time() * 1000) - progress.start_time_ms
        return {f"progress_{scenario_id}": progress}
    except Exception as e:
        progress.status = "failed"
        progress.errors.append({"message": str(e)})
        return {f"progress_{scenario_id}": progress}

async def scenario_aggregator(state: ScenarioOrchestratorState) -> dict:
    """Collect all results and synthesize findings."""
    aggregated = {
        "orchestration_id": state["orchestration_id"],
        "metrics": {},
        "comparison": {},
        "recommendations": []
    }

    for scenario_id in ["simple", "medium", "complex"]:
        progress = state[f"progress_{scenario_id}"]
        aggregated["metrics"][scenario_id] = {
            "elapsed_ms": progress.elapsed_ms,
            "items_processed": progress.items_processed,
            "quality_scores": progress.quality_scores,
        }

    return {"final_results": aggregated}

# Build graph
graph = StateGraph(ScenarioOrchestratorState)
graph.add_node("supervisor", scenario_supervisor)
graph.add_node("scenario_worker", scenario_worker)
graph.add_node("aggregator", scenario_aggregator)
graph.add_edge(START, "supervisor")
graph.add_edge("scenario_worker", "aggregator")
graph.add_edge("aggregator", END)
app = graph.compile(checkpointer=checkpointer)

from abc import ABC, abstractmethod

class SkillOrchestrator(ABC):
    """Abstract orchestrator for any user-invocable skill."""

    def __init__(self, skill_name: str, skill_version: str):
        self.skill_name = skill_name
        self.skill_version = skill_version

    @abstractmethod
    async def invoke_skill(self, input_data: list[dict], scenario_params: dict) -> dict:
        """Invoke your skill on input data."""
        pass

    @abstractmethod
    def get_scenario_configs(self) -> dict[str, dict]:
        """Return configs for simple/medium/complex."""
        pass

    @abstractmethod
    def calculate_quality_metrics(self, results: list[dict], metric_names: list[str]) -> dict:
        """Calculate quality metrics from results."""
        pass

    async def orchestrate(self, orchestration_id: str) -> dict:
        """Run all 3 scenarios in parallel and aggregate."""
        results = await asyncio.gather(
            self.run_scenario("simple", orchestration_id),
            self.run_scenario("medium", orchestration_id),
            self.run_scenario("complex", orchestration_id),
            return_exceptions=True
        )
        return self.aggregate_results(results)

{
  "orchestration_id": "demo-001",
  "quality_comparison": {
    "simple": 0.92, "medium": 0.88, "complex": 0.84
  },
  "scaling_analysis": {
    "time_per_item_ms": {
      "simple": 0.012, "medium": 0.012, "complex": 0.032
    },
    "recommendation": "Sublinear scaling up to 3x, superlinear at 8x"
  }
}

async def orchestrate(skill_name: str):
    simple = await run_scenario("simple", skill_name)  # Wait
    medium = await run_scenario("medium", skill_name)  # Wait
    complex = await run_scenario("complex", skill_name) # Wait
    return [simple, medium, complex]

async def orchestrate(skill_name: str):
    results = await asyncio.gather(
        run_scenario("simple", skill_name),
        run_scenario("medium", skill_name),
        run_scenario("complex", skill_name)
    )
    return aggregate_results(results)

async def synchronize_at_milestone(
    milestone_pct: int,
    state: ScenarioOrchestratorState,
    timeout_seconds: int = 30
) -> bool:
    """Wait for all scenarios to reach milestone."""
    start = time.time()

    while time.time() - start < timeout_seconds:
        simple_at = milestone_pct in state["progress_simple"].milestones_reached
        medium_at = milestone_pct in state["progress_medium"].milestones_reached
        complex_at = milestone_pct in state["progress_complex"].milestones_reached

        if simple_at and medium_at and complex_at:
            print(f"[SYNC] All scenarios reached {milestone_pct}%")
            return True

        if any(state[f"progress_{s}"].status == "failed"
               for s in ["simple", "medium", "complex"]):
            print(f"[SYNC] A scenario failed, proceeding without sync")
            return False

        await asyncio.sleep(0.5)

    print(f"[SYNC] Timeout at {milestone_pct}%, proceeding")
    return False

Simple:  100 items
Medium:  300 items (3x)
Complex: 800 items (8x)
Time: O(n) -- expected medium ~3x simple

SKILL_SCALING_PROFILES = {
    "performance-testing": {
        "scaling": "linear",
        "simple": 10, "medium": 30, "complex": 80
    },
    "security-scanning": {
        "scaling": "sublinear",
        "simple": 20, "medium": 100, "complex": 500
    },
    "data-transformation": {
        "scaling": "quadratic",
        "simple": 100, "medium": 200, "complex": 300
    }
}

# Calculate actual time complexity
simple_tpi = simple_time / simple_size    # time per item
medium_tpi = medium_time / medium_size
complex_tpi = complex_time / complex_size
ratio = complex_tpi / simple_tpi  # >2 = superlinear

try:
    result = await invoke_skill(batch)
except Exception as e:
    progress.errors.append({"message": str(e), "batch_index": i})
    # Don't raise -- let other scenarios continue

async def invoke_skill_with_timeout(skill, input_data, timeout_seconds):
    try:
        return await asyncio.wait_for(
            invoke_skill(skill, input_data),
            timeout=timeout_seconds
        )
    except asyncio.TimeoutError:
        return {
            "processed": len(input_data),
            "results": [],
            "error": "timeout",
            "quality_score": 0.0,
        }

async def orchestrator_with_recovery(initial_state):
    result = await app.ainvoke(initial_state)

    all_failed = all(
        state[f"progress_{s}"].status == "failed"
        for s in ["simple", "medium", "complex"]
    )

    if all_failed:
        # 1. Reduce resource contention
        # 2. Retry with smaller batches
        # 3. Or abort with diagnostic info
        return retry_with_reduced_load(initial_state)

INSERT INTO scenario_checkpoints (
    orchestration_id, scenario_id, milestone_pct, elapsed_ms, state_snapshot
) VALUES (
    'demo-001', 'medium', 30, 3200, '{"items": 90, "results": [...]}'
);

async def checkpoint_full_state(state: ScenarioOrchestratorState):
    checkpoint_data = {
        "orchestration_id": state["orchestration_id"],
        "timestamp": datetime.now().isoformat(),
        "progress_simple": state["progress_simple"].to_dict(),
        "progress_medium": state["progress_medium"].to_dict(),
        "progress_complex": state["progress_complex"].to_dict(),
    }
    await db.insert("full_state_checkpoints", checkpoint_data)

{
    "performance-testing": {
        "latency_p95_ms": {"target": "<500ms", "weight": 0.5},
        "error_rate": {"target": "<1%", "weight": 0.5},
    },
    "security-scanning": {
        "vulnerabilities_found": {"target": ">0", "weight": 0.3},
        "coverage_pct": {"target": "100%", "weight": 0.7},
    }
}

{
    "quality_scaling": {
        "formula": "complex_quality / simple_quality",
        "expected": 1.0,
        "acceptable": ">0.8"
    },
    "time_efficiency": {
        "formula": "simple_tpi / complex_tpi",
        "expected": 1.0,
        "acceptable": ">0.5"
    }
}

# Host 1: Coordinator + Simple
python coordinator.py

# Host 2: Medium (different machine, same DB)
export DATABASE_URL="postgresql://user:pass@coordinator-host/orchestkit"
export SCENARIO_ID=medium
python run_scenario.py

# Host 3: Complex
export SCENARIO_ID=complex
python run_scenario.py

async def run_scenario(scenario_id: str, skill_name: str):
    result = await invoke_skill(skill_name, get_input(scenario_id))  # Hangs forever
    return result

async def run_scenario(scenario_id: str, skill_name: str):
    timeout = {"simple": 30, "medium": 90, "complex": 300}[scenario_id]
    try:
        result = await asyncio.wait_for(
            invoke_skill(skill_name, get_input(scenario_id)),
            timeout=timeout
        )
        return result
    except asyncio.TimeoutError:
        return {"error": "timeout", "scenario_id": scenario_id}

Time →
─────────────────────────────────────────────────┐
Simple   ███████████████ Complete at 1.2s        │
         └─────────────────────────────────────┘ │
                                                  │
Medium      ██████████████████████░░░░ In progress at 3.5s
            └──────────────────────────────────┘ │
                                                  │
Complex      ██████████░░░░░░░░░░░░░░░░ In progress at 25.7s
             └─────────────────────────────────────┘ │
─────────────────────────────────────────────────┘

# Each scenario runs its own event loop
# No waiting between scenarios
# Checkpoints are independent

Time →
─────────────────────────────────────────────────┐
Simple   ███ PAUSE ███ PAUSE ███ Complete        │
         └───┬────────┬────────┬────────────────┘│
                │      │      │                  │
Medium      ██ PAUSE ██ PAUSE ██████░░░░ In-prog │
            └────┬────────┬──────────────────────┘│
                 │      │                        │
Complex      █ PAUSE █ PAUSE ██░░░░░░░░░░░░░░░ │
             └──┬────────┬──────────────────────────┘│
─────────────────────────────────────────────────┘

async def synchronize_at_milestone(milestone_pct, timeout_seconds=60):
    while time.time() - start < timeout_seconds:
        if all_scenarios_at_milestone:
            return True
        await asyncio.sleep(0.5)

    # Timeout: proceed anyway (don't block forever)
    return False

Time →
─────────────────────────────────────────────────┐
Step 1:  Simple   ███ | Medium   ███ | Complex █  │
         └────────────────────────────────────────┘│
Step 2:  Simple   ███ | Medium   ███ | Complex █  │
         └────────────────────────────────────────┘│
Step 3:  All complete together                    │
─────────────────────────────────────────────────┘

Simple:  100 items
Medium:  100 + 200 = 300 items (+200%)
Complex: 300 + 500 = 800 items (+267%)

Time complexity: O(n)
Expected medium time ≈ 3x simple
Expected complex time ≈ 8x simple

Simple:  100 items
Medium:  100 × 3 = 300 items (3x)
Complex: 100 × 8 = 800 items (8x)

Time complexity: O(n) or O(n log n)
Expected medium time ≈ 3x simple (if linear)
Expected complex time ≈ 8x simple (if linear)

Simple:  100 items
Medium:  300 items (3x)
Complex: 800 items (8x)

But if algorithm is O(n²):
Expected medium time ≈ 9x simple
Expected complex time ≈ 64x simple

# Calculate actual time complexity
simple_time = 1.2  # seconds
medium_time = 3.5
complex_time = 25.7

simple_size = 100
medium_size = 300
complex_size = 800

# Time per item
simple_tpi = simple_time / simple_size        # 0.012 s/item
medium_tpi = medium_time / medium_size        # 0.012 s/item
complex_tpi = complex_time / complex_size     # 0.032 s/item

# Ratio indicates scaling behavior
ratio = complex_tpi / simple_tpi  # 2.67 → O(n log n) or worse

SKILL_SCALING_PROFILES = {
    "performance-testing": {
        "scaling": "linear",
        "simple": 10,
        "medium": 30,
        "complex": 80
    },
    "security-scanning": {
        "scaling": "sublinear",  # Gets faster with caching
        "simple": 20,
        "medium": 100,
        "complex": 500
    },
    "data-transformation": {
        "scaling": "quadratic",  # O(n²) worst case
        "simple": 100,
        "medium": 200,  # Limit increase
        "complex": 300
    }
}

{
    "performance-testing": {
        "latency_p95_ms": {"target": "<500ms", "weight": 0.5},
        "error_rate": {"target": "<1%", "weight": 0.5},
    },
    "security-scanning": {
        "vulnerabilities_found": {"target": ">0", "weight": 0.3},
        "coverage_pct": {"target": "100%", "weight": 0.7},
    }
}