---
title: "Agent Orchestration"
description: "Agent orchestration patterns for agentic loops, multi-agent coordination, alternative frameworks, and multi-scenario workflows. Use when building autonomous agent loops, coordinating multiple agents, evaluating CrewAI/AutoGen/Swarm, or orchestrating complex multi-step scenarios."
canonical: "https://orchestkit.yonyon.ai/docs/reference/skills/agent-orchestration"
---

# Agent Orchestration

Agent orchestration patterns for agentic loops, multi-agent coordination, alternative frameworks, and multi-scenario workflows. Use when building autonomous agent loops, coordinating multiple agents, evaluating CrewAI/AutoGen/Swarm, or orchestrating complex multi-step scenarios.

<span className="badge badge-gray">Reference</span> <span className="badge badge-orange">high</span>

> **Not directly invocable** — no slash command and no model auto-selection. An agent loads it explicitly via `Read()`.

<ContextualSkillSidebar slug="agent-orchestration" />

> **Agent Orchestration** Agent orchestration patterns for agentic loops, multi-agent coordination, alternative frameworks, and multi-scenario workflows. Use when building autonomous agent loops, coordinating multiple agents, evaluating CrewAI/AutoGen/Swarm, or orchestrating complex multi-step scenarios.


# Agent Orchestration

Comprehensive patterns for building and coordinating AI agents -- from single-agent reasoning loops to multi-agent systems and framework selection. Coordination and multi-scenario categories have individual rule files in `rules/` loaded on-demand; loop and framework tutorials live upstream (see [Upstream coverage](#upstream-coverage-do-not-restate)), with house defaults in `references/ork-delta.md`.

> **CC native `/workflows` (2.1.154):** Claude Code now ships *dynamic workflows* — ask Claude to create a workflow and it orchestrates tens-to-hundreds of agents in the background; view runs with `/workflows`. This is **complementary** to the patterns here: use CC `/workflows` for large-scale, fire-and-forget **background** fan-out (you check back later); use the bounded **foreground** Agent Teams / Task-tool patterns below when ≤8 agents must coordinate within a single skill invocation via shared memory (handoff files, mesh messaging). Different scale, not a replacement.
>
> **Ask only when genuinely blocked (CC 2.1.154):** CC now reserves the multiple-choice question prompt for decisions it genuinely cannot make itself, rather than asking when it already has enough context to proceed. When orchestrating agents, don't gate progress on an `AskUserQuestion` the lead can resolve from available context — reserve prompts for true branch points (irreversible actions, missing requirements). This complements ork's voice-friendly decision guidance.

## Quick Reference

| Category | Rules | Impact | When to Use |
|----------|-------|--------|-------------|
| [Agent Loops](#agent-loops) | upstream | HIGH | ReAct reasoning, plan-and-execute, self-correction |
| [Multi-Agent Coordination](#multi-agent-coordination) | 2 | CRITICAL | Supervisor routing, agent debate, result synthesis |
| [Alternative Frameworks](#alternative-frameworks) | upstream | HIGH | CrewAI crews, AutoGen teams, framework comparison |
| [Multi-Scenario](#multi-scenario) | 2 | MEDIUM | Parallel scenario orchestration, difficulty routing |

**Total: 4 rules across 4 categories.** Loop and framework tutorials moved to first-party sources; the rescued house defaults live in `references/ork-delta.md`.

## Quick Start

```python
# 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"
```

```python
# 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)
```

## Agent Loops

Patterns for autonomous LLM reasoning: ReAct (Reasoning + Acting), Plan-and-Execute with replanning, self-correction loops, and sliding-window memory management.

**Key decisions:** Max steps 5-15, temperature 0.3-0.7, memory window 10-20 messages.

## Multi-Agent Coordination

Fan-out/fan-in parallelism, supervisor routing with dependency ordering, conflict resolution (confidence-based or LLM arbitration), result synthesis, and CC Agent Teams (mesh topology for peer messaging in CC 2.1.33+).

**Key decisions:** 3-8 specialists, parallelize independent agents, use Task tool (star) for simple work, Agent Teams (mesh) for cross-cutting concerns.

## Alternative Frameworks

CrewAI hierarchical crews with Flows (1.8+), OpenAI Agents SDK handoffs and guardrails (0.12+), Microsoft Agent Framework (AutoGen + SK merger), GPT-5.2-Codex for long-horizon coding, and AG2 for open-source flexibility.

**Key decisions:** Match framework to team expertise + use case. LangGraph for state machines, CrewAI for role-based teams, OpenAI SDK for handoff workflows, MS Agent for enterprise compliance.

## Multi-Scenario

Orchestrate a single skill across 3 parallel scenarios (simple/medium/complex) with progressive difficulty scaling (1x/3x/8x), milestone synchronization, and cross-scenario result aggregation.

**Key decisions:** Free-running with checkpoints, always 3 scenarios, 1x/3x/8x exponential scaling, 30s/90s/300s time budgets.

## Upstream coverage (do not restate)

Local tutorials for these topics were retired; consult the first-party source and keep only house deltas in `references/ork-delta.md`.

| Topic | First-party source |
|-------|--------------------|
| ReAct / plan-and-execute / self-correction loop implementations | OpenAI function calling guide (https://platform.openai.com/docs/guides/function-calling); LangGraph tutorials (context7: /langchain-ai/langgraph) |
| Fan-out coordination, result-synthesis boilerplate, and the generic multi-agent design checklist | Python asyncio docs (https://docs.python.org/3/library/asyncio-task.html); Anthropic "Building effective agents" (https://www.anthropic.com/research/building-effective-agents); `ork:langgraph` supervisor patterns |
| CrewAI (crews, Flows, MCP tools, guardrails) | CrewAI docs (https://docs.crewai.com); context7: /crewaiinc/crewai |
| OpenAI Agents SDK (handoffs, sessions, guardrails, MCP) | https://openai.github.io/openai-agents-python/ ; context7: /openai/openai-agents-python |
| Microsoft Agent Framework / AutoGen (teams, termination, A2A) | https://learn.microsoft.com/en-us/agent-framework/ ; context7: /microsoft/autogen |
| GPT-5.2-Codex capabilities, pricing, IDE integrations | OpenAI model docs (https://platform.openai.com/docs/models) |
| Multi-scenario state machine, architecture and skill-agnostic template deep-dives | Superseded in-skill by `rules/scenario-orchestrator.md` and `rules/scenario-routing.md` |

## References

- `references/ork-delta.md` - House defaults and dated decisions rescued from retired tutorials
- `references/framework-comparison.md` - Condensed framework decision matrix and use-case table
- `references/langgraph-implementation.md` - LangGraph 1.2+ implementation of the multi-scenario orchestrator
- `references/claude-code-instance-management.md` - Running 3 parallel Claude Code instances for scenario demos

## Key Decisions

| Decision | Recommendation |
|----------|----------------|
| Single vs multi-agent | Single for focused tasks, multi for decomposable work |
| Max loop steps | 5-15 (prevent infinite loops) |
| Agent count | 3-8 specialists per workflow |
| Framework | Match to team expertise + use case |
| Topology | Task tool (star) for simple; Agent Teams (mesh) for complex |
| Scenario count | Always 3: simple, medium, complex |

## Common Mistakes

- No step limit in agent loops (infinite loops)
- No memory management (context overflow)
- No error isolation in multi-agent (one failure crashes all)
  - Note (CC 2.1.161): parallel *tool calls* now fail independently — a failed Bash no longer cancels siblings in the same batch. This caveat still applies at the agent-orchestration level, not to tool batches; `claude agents` rows now show `done/total` for fanned-out work.
  - Note (CC 2.1.157): `claude agents` honors the `agent` field in `settings.json` for dispatched sessions; `--agent &lt;name&gt;` overrides it — pin the agent type explicitly when dispatching.
- Missing synthesis step (raw agent outputs not useful)
- Mixing frameworks in one project (complexity explosion)
- Using Agent Teams for simple sequential work (use Task tool)
- Sequential instead of parallel scenarios (defeats purpose)

## Related Skills

- `ork:langgraph` - LangGraph workflow patterns (supervisor, routing, state)
- `function-calling` - Tool definitions and execution
- `ork:task-dependency-patterns` - Task management with Agent Teams workflow

## Capability Details

### react-loop
**Keywords:** react, reason, act, observe, loop, agent
**Solves:**
- Implement ReAct pattern
- Create reasoning loops
- Build iterative agents

### plan-execute
**Keywords:** plan, execute, replan, multi-step, autonomous
**Solves:**
- Create plan then execute steps
- Implement replanning on failure
- Build goal-oriented agents

### supervisor-coordination
**Keywords:** supervisor, route, coordinate, fan-out, fan-in, parallel
**Solves:**
- Route tasks to specialized agents
- Run agents in parallel
- Aggregate multi-agent results

### agent-debate
**Keywords:** debate, conflict, resolution, arbitration, consensus
**Solves:**
- Resolve agent disagreements
- Implement LLM arbitration
- Handle conflicting outputs

### result-synthesis
**Keywords:** synthesize, combine, aggregate, merge, summary
**Solves:**
- Combine outputs from multiple agents
- Create executive summaries
- Score confidence across findings

### crewai-patterns
**Keywords:** crewai, crew, hierarchical, delegation, role-based, flows
**Solves:**
- Build role-based agent teams
- Implement hierarchical coordination
- Use Flows for event-driven orchestration

### autogen-patterns
**Keywords:** autogen, microsoft, agent framework, teams, enterprise, a2a
**Solves:**
- Build enterprise agent systems
- Use AutoGen/SK merged framework
- Implement A2A protocol

### framework-selection
**Keywords:** choose, compare, framework, decision, which, crewai, autogen, openai
**Solves:**
- Select appropriate framework
- Compare framework capabilities
- Match framework to requirements

### scenario-orchestrator
**Keywords:** scenario, parallel, fan-out, difficulty, progressive, demo
**Solves:**
- Run skill across multiple difficulty levels
- Implement parallel scenario execution
- Aggregate cross-scenario results

### scenario-routing
**Keywords:** route, synchronize, milestone, checkpoint, scaling
**Solves:**
- Route tasks by difficulty level
- Synchronize at milestones
- Scale inputs progressively


---

## Rules (4)

### Resolve agent disagreements through confidence scores, LLM arbitration, or majority voting — HIGH


# Agent Debate & Conflict Resolution

Patterns for handling disagreements between agents and establishing communication channels.

## Conflict Resolution

```python
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)
```

## Structured Conflict Detection

```python
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}
```

## Agent Communication Bus

```python
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
```

## Resolution Strategies

| Strategy | When to Use | Trade-off |
|----------|-------------|-----------|
| Confidence-based | Agents provide confidence scores | Fast but requires calibrated scores |
| LLM arbitration | Complex disagreements | Higher quality but adds LLM cost |
| Majority voting | 3+ agents on same question | Simple but requires odd count |
| Weighted consensus | Agents have different expertise | Best for specialized teams |
| Human-in-the-loop | High-stakes decisions | Most reliable but slowest |

## Common Mistakes

- No timeout per agent (one slow agent blocks all)
- No error isolation (one failure crashes workflow)
- Over-coordination (too much overhead)
- Using Agent Teams for simple sequential work (use Task tool)
- Broadcasting when a direct message suffices (wastes tokens)

**Incorrect — no conflict resolution strategy:**
```python
async def multi_agent_analysis(task: str):
    results = await asyncio.gather(agent1(task), agent2(task))
    return results  # Return conflicting results without resolution
```

**Correct — LLM arbitration resolves conflicts:**
```python
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]
```


### Coordinate specialist agents through a central supervisor with parallel execution and timeouts — CRITICAL


# Supervisor Pattern

Central coordinator that routes tasks to specialist agents, supports parallel and sequential execution, and aggregates results.

## Fan-Out/Fan-In

```python
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)
```

## Supervisor with Routing

```python
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)
```

## Supervisor-Worker with Timeout

```python
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"}
```

## CC Agent Teams (CC 2.1.33+)

CC 2.1.33 introduces native Agent Teams with peer-to-peer messaging and mesh topology.

### Star vs Mesh Topology

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

### Dual-Mode Decision Tree

```
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 (GA since CC 2.1.33)
```

### Team Formation

```
# 1. Create team with shared task list
# CC 2.1.178+: one implicit team per session — no TeamCreate.
# Spawn teammates directly via Agent(name=...). Requires
# CLAUDE_CODE_EXPERIMENTAL_AGENT_TEAMS=1 (set in ork.settings.json).

# 2. Create tasks in shared list
TaskCreate(subject="Design API schema", description="...")
TaskCreate(subject="Build React components", description="...")
TaskUpdate(taskId="2", addBlockedBy=["1"])  # link deps after the task exists

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

### Peer Messaging

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

# No broadcast primitive -- send per-teammate or post to the shared task list
SendMessage(to="frontend-dev",
  message="Auth header format changed to Bearer",
  summary="Breaking auth change")
```

### Cost Comparison

| Scenario | Task Tool | Agent Teams | Ratio |
|----------|-----------|-------------|-------|
| 3-agent review | ~150K tokens | ~400K tokens | 2.7x |
| 8-agent feature | ~500K tokens | ~1.2M tokens | 2.4x |
| 6-agent research | ~300K tokens | ~800K tokens | 2.7x |

## Key Decisions

| Decision | Recommendation |
|----------|----------------|
| Agent count | 3-8 specialists |
| Parallelism | Parallelize independent agents |
| Worker timeout | 30s default |
| Communication | Shared state, message bus, or SendMessage (CC 2.1.33+) |
| Topology | Task tool (star) for simple; Agent Teams (mesh) for complex |

**Incorrect — sequential execution of independent agents:**
```python
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]
```

**Correct — parallel fan-out for independent agents:**
```python
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
```


### Test skills across three parallel scenarios with progressive difficulty and synchronized execution — MEDIUM


# Multi-Scenario Orchestrator

Run a single skill across 3 parallel scenarios (simple/medium/complex) with synchronized execution and progressive difficulty.

## Core Pattern

```
+---------------------------------------------------------------------+
|                   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                         |
+---------------------------------------------------------------------+
```

## When to Use

| Scenario | Example |
|----------|---------|
| **Skill demos** | Show `/ork:implement` on simple, medium, complex tasks |
| **Progressive testing** | Validate skill scales with complexity |
| **Comparative analysis** | How does approach differ by difficulty? |
| **Training/tutorials** | Show skill progression from easy to hard |

## LangGraph Implementation

```python
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)
```

## Skill-Agnostic Template

```python
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)
```

## Difficulty Scaling

| Level | Complexity | Input Size | Time Budget | Quality |
|-------|------------|------------|-------------|---------|
| Simple | 1x | Small (10-100) | 30s | Basic |
| Medium | 3x | Medium (30-300) | 90s | Good |
| Complex | 8x | Large (80-800) | 300s | Excellent |

## Output Example

```json
{
  "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"
  }
}
```

## Common Mistakes

- Sequential instead of parallel (defeats purpose)
- No synchronization (results appear disjointed)
- Unclear difficulty scaling (differ in scale, not approach)
- Missing aggregation (individual results lack comparative insights)

**Incorrect — running scenarios sequentially:**
```python
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]
```

**Correct — parallel execution of all scenarios:**
```python
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)
```


### Synchronize milestones, scale difficulty, and recover from failures across multi-scenario orchestration — MEDIUM


# Scenario Routing & Synchronization

Milestone synchronization modes, difficulty scaling strategies, checkpointing, and failure recovery for multi-scenario orchestration.

## Synchronization Modes

| Mode | Description | Use When |
|------|-------------|----------|
| **Free-running** | All run independently | Demo videos, production |
| **Milestone-sync** | Wait at 30%, 70%, 100% | Comparative analysis |
| **Lock-step** | All proceed together | Training, tutorials |

### Milestone Synchronization

```python
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
```

## Input Scaling Strategies

### Linear Scaling (I/O-bound skills)

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

### Adaptive Scaling (per-skill tuning)

```python
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
    }
}
```

### Complexity Detection

```python
# 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
```

## Failure Recovery

### One Scenario Fails (Independent)

```python
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
```

### Timeout Handling

```python
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,
        }
```

### All Scenarios Fail (Systematic)

```python
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)
```

## Checkpointing

### Scenario-Level Checkpoints

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

### Full-State Snapshots

```python
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)
```

## Quality Metrics Framework

### Functional Metrics (per-skill)

```python
{
    "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},
    }
}
```

### Comparative Metrics

```python
{
    "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"
    }
}
```

## Multi-Host Execution

For greater parallelism, run scenarios on different machines sharing the same database:

```bash
# 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
```

## Key Decisions

| Decision | Recommendation |
|----------|----------------|
| Synchronization mode | Free-running with checkpoints |
| Scenario count | Always 3: simple, medium, complex |
| Input scaling | 1x, 3x, 8x (exponential) |
| Time budgets | 30s, 90s, 300s |
| Checkpoint frequency | Every milestone + completion |

**Incorrect — no timeout on skill invocation:**
```python
async def run_scenario(scenario_id: str, skill_name: str):
    result = await invoke_skill(skill_name, get_input(scenario_id))  # Hangs forever
    return result
```

**Correct — timeout prevents infinite hangs:**
```python
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}
```



---

## References (4)

### Claude Code Instance Management

# Claude Code Instance Management: Multi-Scenario Demos

**Structure 3 parallel Claude Code terminal instances for simultaneous scenario execution with shared state synchronization.**

## Instance Architecture

```
┌─────────────────────────────────────────────────────────────────────┐
│                    COORDINATOR PROCESS (Python)                      │
│                   (Runs orchestrator graph)                          │
├─────────────────────────────────────────────────────────────────────┤
│  ┌─────────────┐      ┌─────────────┐      ┌─────────────┐         │
│  │  Terminal 1 │      │  Terminal 2 │      │  Terminal 3 │         │
│  │  (Simple)   │      │  (Medium)   │      │  (Complex)  │         │
│  │             │      │             │      │             │         │
│  │  Session:   │      │  Session:   │      │  Session:   │         │
│  │  simple-123 │      │  medium-123 │      │  complex-123│         │
│  └─────────────┘      └─────────────┘      └─────────────┘         │
│       │                    │                    │                   │
│  Claude Code instances     Claude Code instances     Claude Code     │
│  (3 parallel processes)    (3 parallel processes)    instance        │
│                                                                     │
│  ┌─────────────────────────────────────────────────────────┐       │
│  │   PostgreSQL Checkpoint Table                            │       │
│  │   (Shared state synchronization across instances)        │       │
│  └─────────────────────────────────────────────────────────┘       │
│                                                                     │
└─────────────────────────────────────────────────────────────────────┘
```

## Setup Instructions

### Step 1: Prepare the Project

Ensure your project has the orchestrator graph and shared utilities:

```bash
# At project root
mkdir -p backend/app/workflows/multi_scenario
cp src/skills/multi-scenario-orchestration/references/langgraph-implementation.py \
   backend/app/workflows/multi_scenario/orchestrator.py

# Create coordinator script
cat > backend/app/workflows/multi_scenario/coordinator.py << 'EOF'
"""
Main coordinator that launches and monitors 3 Claude Code instances.
"""
import asyncio
import subprocess
import os
from pathlib import Path

PROJECT_ROOT = Path(__file__).parent.parent.parent.parent
SCENARIOS = ["simple", "medium", "complex"]
SKILL_NAME = "your-skill-name"  # Change this

async def launch_scenario_instance(scenario_id: str, orchestration_id: str):
    """Launch one Claude Code instance for a scenario."""

    env = os.environ.copy()
    env["SCENARIO_ID"] = scenario_id
    env["ORCHESTRATION_ID"] = orchestration_id
    env["PROJECT_ROOT"] = str(PROJECT_ROOT)

    # Launch Claude Code instance
    process = subprocess.Popen(
        [
            "claude", "code",
            str(PROJECT_ROOT),
            "--session", f"scenario-{scenario_id}-{orchestration_id}",
            "--skill", SKILL_NAME,
        ],
        env=env,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE
    )

    print(f"[COORDINATOR] Launched {scenario_id} instance (PID: {process.pid})")
    return process

async def monitor_instances(processes: dict):
    """Monitor all instances for completion."""

    while any(p.poll() is None for p in processes.values()):
        for scenario_id, process in processes.items():
            if process.poll() is not None:
                print(f"[COORDINATOR] {scenario_id} instance completed")

        await asyncio.sleep(1)

async def main():
    orchestration_id = "demo-001"

    print(f"[COORDINATOR] Starting orchestration {orchestration_id}")
    print(f"[COORDINATOR] Launching 3 parallel instances...")

    # Launch all instances
    processes = {}
    for scenario_id in SCENARIOS:
        process = await launch_scenario_instance(scenario_id, orchestration_id)
        processes[scenario_id] = process

    print(f"[COORDINATOR] All instances launched. Monitoring...")

    # Monitor
    await monitor_instances(processes)

    print(f"[COORDINATOR] All instances completed")

if __name__ == "__main__":
    asyncio.run(main())
EOF
```

### Step 2: Create Scenario Runner Script

Create `/backend/app/workflows/multi_scenario/run_scenario.py`:

```python
"""
Runner for single scenario. Invoked by Claude Code instance.
Sets up environment from SCENARIO_ID and ORCHESTRATION_ID env vars.
"""
import os
import asyncio
from orchestrator import (
    build_scenario_orchestrator,
    ScenarioOrchestratorState,
    ScenarioDefinition,
    ScenarioProgress,
)
from langgraph.checkpoint.postgres import PostgresSaver
from psycopg_pool import ConnectionPool

async def run_scenario():
    # Read from environment
    scenario_id = os.getenv("SCENARIO_ID", "simple")
    orchestration_id = os.getenv("ORCHESTRATION_ID", "demo-001")
    project_root = os.getenv("PROJECT_ROOT", ".")

    print(f"[{scenario_id.upper()}] Starting scenario execution")
    print(f"[{scenario_id.upper()}] Orchestration ID: {orchestration_id}")

    # Setup checkpointer — from_conn_string is a @contextmanager, so for a
    # long-running orchestrator use an explicit pool with the PostgresSaver
    # constructor (a `with` block would close the pool mid-run).
    db_url = os.getenv("DATABASE_URL", "postgresql://localhost/orchestkit")
    pool = ConnectionPool(db_url, max_size=20, kwargs={"autocommit": True, "prepare_threshold": 0})
    checkpointer = PostgresSaver(pool)
    checkpointer.setup()  # first run creates the checkpoint tables

    # Build orchestrator
    app = build_scenario_orchestrator(checkpointer=checkpointer)

    # Prepare scenario definitions
    configs = {
        "simple": {
            "complexity_multiplier": 1.0,
            "input_size": 100,
            "time_budget_seconds": 30,
            "skill_params": {"batch_size": 10, "cache_enabled": True}
        },
        "medium": {
            "complexity_multiplier": 3.0,
            "input_size": 300,
            "time_budget_seconds": 90,
            "skill_params": {"batch_size": 50, "cache_enabled": True}
        },
        "complex": {
            "complexity_multiplier": 8.0,
            "input_size": 800,
            "time_budget_seconds": 300,
            "skill_params": {"batch_size": 100, "cache_enabled": True, "parallel_workers": 4}
        }
    }

    cfg = configs[scenario_id]

    # Build initial state
    initial_state: ScenarioOrchestratorState = {
        "orchestration_id": orchestration_id,
        "start_time_unix": int(time.time()),
        "skill_name": "your-skill-name",
        "skill_version": "1.0.0",

        # Current scenario only
        "scenario_simple": None,
        "scenario_medium": None,
        "scenario_complex": None,
        "progress_simple": None,
        "progress_medium": None,
        "progress_complex": None,
    }

    # Set only the relevant scenario
    initial_state[f"scenario_{scenario_id}"] = ScenarioDefinition(
        name=scenario_id,
        difficulty={"simple": "easy", "medium": "intermediate", "complex": "advanced"}[scenario_id],
        complexity_multiplier=cfg["complexity_multiplier"],
        input_size=cfg["input_size"],
        dataset_characteristics={"distribution": "uniform"},
        time_budget_seconds=cfg["time_budget_seconds"],
        memory_limit_mb={"simple": 256, "medium": 512, "complex": 1024}[scenario_id],
        error_tolerance={"simple": 0.0, "medium": 0.05, "complex": 0.1}[scenario_id],
        skill_params=cfg["skill_params"],
        expected_quality={"simple": "basic", "medium": "good", "complex": "excellent"}[scenario_id],
        quality_metrics=["accuracy", "coverage"]
    )

    initial_state[f"progress_{scenario_id}"] = ScenarioProgress(scenario_id=scenario_id)

    # Run orchestrator
    config = {"configurable": {"thread_id": f"orch-{orchestration_id}"}}

    print(f"[{scenario_id.upper()}] Invoking orchestrator...")

    try:
        # Stream progress
        async for update in app.astream(initial_state, config=config, stream_mode="updates"):
            if f"progress_{scenario_id}" in update:
                progress = update[f"progress_{scenario_id}"]
                print(f"[{scenario_id.upper()}] Progress: {progress.progress_pct:.1f}% "
                      f"({progress.items_processed} items, {progress.elapsed_ms}ms)")

        print(f"[{scenario_id.upper()}] Scenario complete")

    except Exception as e:
        print(f"[{scenario_id.upper()}] Error: {e}")
        raise

if __name__ == "__main__":
    import time
    asyncio.run(run_scenario())
```

## Execution: Three-Terminal Mode

### Terminal 1: Coordinator

```bash
cd /path/to/project
python backend/app/workflows/multi_scenario/coordinator.py
```

**Output:**
```
[COORDINATOR] Starting orchestration demo-001
[COORDINATOR] Launching 3 parallel instances...
[COORDINATOR] Launched simple instance (PID: 1234)
[COORDINATOR] Launched medium instance (PID: 1235)
[COORDINATOR] Launched complex instance (PID: 1236)
[COORDINATOR] All instances launched. Monitoring...
```

### Terminal 2: Simple Scenario

```bash
cd /path/to/project
export SCENARIO_ID=simple
export ORCHESTRATION_ID=demo-001
python backend/app/workflows/multi_scenario/run_scenario.py
```

**Output:**
```
[SIMPLE] Starting scenario execution
[SIMPLE] Orchestration ID: demo-001
[SIMPLE] Invoking orchestrator...
[SIMPLE] Progress: 10.0% (10 items, 100ms)
[SIMPLE] Progress: 20.0% (20 items, 200ms)
...
[SIMPLE] Progress: 100.0% (100 items, 1050ms)
[SIMPLE] Scenario complete
```

### Terminal 3: Medium Scenario

```bash
cd /path/to/project
export SCENARIO_ID=medium
export ORCHESTRATION_ID=demo-001
python backend/app/workflows/multi_scenario/run_scenario.py
```

**Output:**
```
[MEDIUM] Starting scenario execution
[MEDIUM] Orchestration ID: demo-001
[MEDIUM] Invoking orchestrator...
[MEDIUM] Progress: 3.3% (10 items, 100ms)
[MEDIUM] Progress: 6.7% (20 items, 200ms)
...
[MEDIUM] Progress: 100.0% (300 items, 3100ms)
[MEDIUM] Scenario complete
```

### Terminal 4 (Optional): Complex Scenario

If you have 4 terminals, run complex in parallel:

```bash
export SCENARIO_ID=complex
export ORCHESTRATION_ID=demo-001
python backend/app/workflows/multi_scenario/run_scenario.py
```

## Shared State Synchronization

### PostgreSQL Checkpoint Schema

```sql
-- Create checkpoint table (run once)
CREATE TABLE IF NOT EXISTS scenario_orchestration_checkpoints (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    orchestration_id VARCHAR(255) NOT NULL,
    scenario_id VARCHAR(50) NOT NULL,
    milestone_name VARCHAR(100),
    progress_pct FLOAT,
    timestamp_unix BIGINT NOT NULL,
    state_snapshot JSONB,
    metrics JSONB,
    created_at TIMESTAMP DEFAULT NOW(),
    INDEX idx_orchestration_id (orchestration_id),
    INDEX idx_scenario_id (scenario_id),
    INDEX idx_timestamp (timestamp_unix)
);

-- View progress across all scenarios
SELECT
    orchestration_id,
    scenario_id,
    progress_pct,
    milestone_name,
    timestamp_unix,
    (timestamp_unix / 1000.0) as seconds_elapsed
FROM scenario_orchestration_checkpoints
WHERE orchestration_id = 'demo-001'
ORDER BY scenario_id, progress_pct;

-- Example output:
-- orchestration_id | scenario_id | progress_pct | milestone_name | seconds_elapsed
-- demo-001         | simple      | 30           | checkpoint_1   | 1.2
-- demo-001         | simple      | 70           | checkpoint_2   | 2.8
-- demo-001         | simple      | 100          | completion     | 3.1
-- demo-001         | medium      | 30           | checkpoint_1   | 3.5
-- demo-001         | medium      | 70           | checkpoint_2   | 8.2
-- demo-001         | medium      | 100          | completion     | 9.3
-- demo-001         | complex     | 30           | checkpoint_1   | 9.1
-- demo-001         | complex     | 70           | checkpoint_2   | 22.5
-- demo-001         | complex     | 100          | completion     | 25.7
```

### Monitor Progress from Coordinator

```python
"""Monitor script to watch progress across all instances."""
import asyncio
import time
from datetime import datetime
import psycopg2

async def monitor_orchestration(orchestration_id: str, interval: int = 2):
    """Watch progress of all scenarios."""

    conn = psycopg2.connect("dbname=orchestkit user=postgres")
    cursor = conn.cursor()

    print(f"Monitoring orchestration {orchestration_id}...\n")

    while True:
        cursor.execute("""
            SELECT
                scenario_id,
                MAX(progress_pct) as progress,
                MAX(timestamp_unix) as last_update
            FROM scenario_orchestration_checkpoints
            WHERE orchestration_id = %s
            GROUP BY scenario_id
            ORDER BY scenario_id
        """, (orchestration_id,))

        rows = cursor.fetchall()
        if not rows:
            print("No progress yet...")
            await asyncio.sleep(interval)
            continue

        # Clear screen and print progress
        print(f"\r{datetime.now().strftime('%H:%M:%S')}")
        print("-" * 50)

        all_complete = True
        for scenario_id, progress, timestamp in rows:
            bar_length = int(progress / 5)  # 20-char bar
            bar = "█" * bar_length + "░" * (20 - bar_length)

            print(f"{scenario_id:10} │{bar}│ {progress:3.0f}%")

            if progress < 100:
                all_complete = False

        if all_complete:
            print("\n✓ All scenarios complete!")
            break

        await asyncio.sleep(interval)

    conn.close()

if __name__ == "__main__":
    asyncio.run(monitor_orchestration("demo-001"))
```

## Synchronization at Milestones

To enable forced synchronization at milestones (all scenarios pause and wait):

```python
# In run_scenario.py
import asyncpg

async def wait_for_milestone_sync(
    orchestration_id: str,
    scenario_id: str,
    milestone_pct: int,
    timeout_seconds: int = 30
):
    """Wait for all scenarios to reach milestone."""

    # Poll the progress table directly with an asyncpg pool — the langgraph
    # checkpointer is a @contextmanager factory, not a general connection source.
    pool = await asyncpg.create_pool(DATABASE_URL)
    start = time.time()

    while time.time() - start < timeout_seconds:
        # Query checkpoint status
        async with pool.acquire() as conn:
            result = await conn.fetch("""
                SELECT DISTINCT scenario_id, MAX(progress_pct)
                FROM scenario_orchestration_checkpoints
                WHERE orchestration_id = $1
                GROUP BY scenario_id
            """, orchestration_id)

            scenarios_at_milestone = {
                row["scenario_id"]: row["max"] >= milestone_pct
                for row in result
            }

            if all(scenarios_at_milestone.values()):
                print(f"[{scenario_id.upper()}] All scenarios reached {milestone_pct}%")
                return True

        await asyncio.sleep(0.5)

    print(f"[{scenario_id.upper()}] Sync timeout at {milestone_pct}%")
    return False
```

## Advanced: Multi-Host Execution

For even greater parallelism, run scenarios on different machines:

```bash
# Host 1: Coordinator + Simple
python backend/app/workflows/multi_scenario/coordinator.py

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

# Host 3: Complex (different machine, same DB)
export DATABASE_URL="postgresql://user:pass@coordinator-host/orchestkit"
export SCENARIO_ID=complex
export ORCHESTRATION_ID=demo-001
python backend/app/workflows/multi_scenario/run_scenario.py
```

PostgreSQL checkpoints serve as the distributed state store.

## Best Practices

1. **Unique Orchestration IDs**: Use timestamp or UUID for each demo run
2. **Session Isolation**: Each instance gets its own Claude Code session
3. **Checkpointing**: Always enable PostgreSQL persistence
4. **Monitoring**: Watch progress via checkpoint table queries
5. **Timeout Handling**: Allow asynchronous completion, don't force lock-step
6. **Error Recovery**: Failed instances can be restarted without resetting state

## Troubleshooting

**Instances get stuck at milestone:**
→ Increase `timeout_seconds` in `wait_for_milestone_sync()`

**Database connection errors:**
→ Check `DATABASE_URL` environment variable, ensure PostgreSQL is running

**One instance much slower than others:**
→ This is expected! Use Mode A (free-running), not lock-step. Slower instance will eventually complete.

**Memory usage grows over time:**
→ Enable checkpointing to disk, reduce batch sizes for complex scenario


### Framework Comparison

# Framework Comparison

Decision matrix for choosing between multi-agent frameworks.

## Feature Comparison

| Feature | LangGraph | CrewAI | OpenAI SDK | MS Agent |
|---------|-----------|--------|------------|----------|
| State Management | Excellent | Good | Basic | Good |
| Persistence | Built-in | Plugin | Manual | Built-in |
| Streaming | Native | Limited | Native | Native |
| Human-in-Loop | Native | Manual | Manual | Native |
| Memory | Via Store | Built-in | Manual | Manual |
| Observability | Langfuse/LangSmith | Limited | Tracing | Azure Monitor |
| Learning Curve | Steep | Easy | Medium | Medium |
| Production Ready | Yes | Yes | Yes | Q1 2026 |

## Use Case Matrix

| Use Case | Best Framework | Why |
|----------|---------------|-----|
| Complex state machines | LangGraph | Native StateGraph, persistence |
| Role-based teams | CrewAI | Built-in delegation, backstories |
| OpenAI-only projects | OpenAI SDK | Native integration, handoffs |
| Enterprise/compliance | MS Agent | Azure integration, A2A |
| Research/experiments | AG2 | Open-source, flexible |
| Quick prototypes | CrewAI | Minimal boilerplate |
| Long-running workflows | LangGraph | Checkpointing, recovery |
| Customer support bots | OpenAI SDK | Handoffs, guardrails |

## Decision Tree

```
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
  |           |
  +-- Open-source priority?
        |
        +-- Yes --> AG2
        |
        +-- No --> LangGraph (default)
```

## Migration Paths

### From AutoGen to MS Agent Framework

```python
# 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.5")
agent = AssistantAgent(name="assistant", model_client=model_client)
```

### From Custom to LangGraph

```python
# 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")
```

## Cost Considerations

| Framework | Licensing | Infra Cost | LLM Cost |
|-----------|-----------|------------|----------|
| LangGraph | MIT | Self-host / LangGraph Cloud | Any LLM |
| CrewAI | MIT | Self-host | Any LLM |
| OpenAI SDK | MIT | Self-host | OpenAI only |
| MS Agent | MIT | Self-host / Azure | Any LLM |
| AG2 | Apache 2.0 | Self-host | Any LLM |

## Performance Characteristics

| Framework | Cold Start | Latency | Throughput |
|-----------|------------|---------|------------|
| LangGraph | ~100ms | Low | High |
| CrewAI | ~200ms | Medium | Medium |
| OpenAI SDK | ~50ms | Low | High |
| MS Agent | ~150ms | Medium | High |

## Team Expertise Requirements

| Framework | Python | LLM | Infra |
|-----------|--------|-----|-------|
| LangGraph | Expert | Expert | Medium |
| CrewAI | Beginner | Beginner | Low |
| OpenAI SDK | Medium | Medium | Low |
| MS Agent | Medium | Medium | High |

## Recommendation Summary

1. **Default choice**: LangGraph (most capable, production-proven)
2. **Fastest to prototype**: CrewAI (minimal code, intuitive)
3. **OpenAI shops**: OpenAI Agents SDK (native integration)
4. **Enterprise**: Microsoft Agent Framework (compliance, Azure)
5. **Research**: AG2 (open community, experimental features)


### Langgraph Implementation

# LangGraph Implementation: Multi-Scenario Orchestration

Complete Python implementation of the multi-scenario orchestration pattern using LangGraph 1.2+.

## 1. State Definition

```python
from typing import TypedDict, Annotated, Literal
from dataclasses import dataclass, field, asdict
from operator import add
import time
from datetime import datetime

@dataclass
class ScenarioProgress:
    """Track execution state for one scenario."""
    scenario_id: str
    status: Literal["pending", "running", "paused", "complete", "failed"]
    progress_pct: float = 0.0

    # Milestones
    milestones_reached: list[str] = field(default_factory=list)
    current_milestone: str = "start"

    # Timing
    start_time_ms: int = 0
    elapsed_ms: int = 0
    elapsed_checkpoints: dict = field(default_factory=dict)  # {milestone: time_ms}

    # Metrics
    memory_used_mb: int = 0
    items_processed: int = 0
    batch_count: int = 0

    # Results
    partial_results: list[dict] = field(default_factory=list)
    quality_scores: dict = field(default_factory=dict)

    # Errors
    errors: list[dict] = field(default_factory=list)

    def to_dict(self):
        return asdict(self)

@dataclass
class ScenarioDefinition:
    """Configuration for one scenario."""
    name: str  # "simple", "medium", "complex"
    difficulty: Literal["easy", "intermediate", "advanced"]
    complexity_multiplier: float  # 1.0, 3.0, 8.0

    # Inputs
    input_size: int
    dataset_characteristics: dict  # {"distribution": "uniform"}

    # Constraints
    time_budget_seconds: int
    memory_limit_mb: int
    error_tolerance: float  # 0-1

    # Skill params
    skill_params: dict

    # Expectations
    expected_quality: Literal["basic", "good", "excellent"]
    quality_metrics: list[str]

    def to_dict(self):
        return asdict(self)

class ScenarioOrchestratorState(TypedDict, total=False):
    """State for the entire orchestration."""

    # Orchestration metadata
    orchestration_id: str
    start_time_unix: int
    skill_name: str
    skill_version: str

    # Scenario definitions
    scenario_simple: ScenarioDefinition
    scenario_medium: ScenarioDefinition
    scenario_complex: ScenarioDefinition

    # Progress tracking
    progress_simple: ScenarioProgress
    progress_medium: ScenarioProgress
    progress_complex: ScenarioProgress

    # Synchronization
    sync_points: dict  # {milestone: bool}
    last_sync_time: int

    # Aggregated results
    final_results: dict
```

## 2. Node Implementations

### Supervisor Node

```python
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.

    Returns Send commands that trigger parallel execution.
    """
    print(f"[SUPERVISOR] Starting orchestration {state['orchestration_id']}")

    # Initialize progress for each scenario
    for scenario_id in ["simple", "medium", "complex"]:
        progress = ScenarioProgress(
            scenario_id=scenario_id,
            status="pending",
            start_time_ms=int(time.time() * 1000)
        )
        state[f"progress_{scenario_id}"] = progress

    # Return Send commands for parallel execution
    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 (simple, medium, or complex).

    Receives scenario_id from supervisor via Send.
    """
    scenario_id = state.get("scenario_id")
    progress = state[f"progress_{scenario_id}"]
    scenario_def = state[f"scenario_{scenario_id}"]

    print(f"[SCENARIO {scenario_id.upper()}] Starting ({scenario_def.complexity_multiplier}x complexity)")

    progress.status = "running"
    progress.start_time_ms = int(time.time() * 1000)

    try:
        # Execute skill for this scenario
        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
        progress.partial_results.append(result)

        print(f"[SCENARIO {scenario_id.upper()}] Complete in {progress.elapsed_ms}ms")

        return {f"progress_{scenario_id}": progress}

    except Exception as e:
        progress.status = "failed"
        progress.errors.append({
            "timestamp": datetime.now().isoformat(),
            "message": str(e),
            "severity": "error"
        })
        print(f"[SCENARIO {scenario_id.upper()}] Failed: {e}")

        return {f"progress_{scenario_id}": progress}


async def execute_skill_with_milestones(
    skill_name: str,
    scenario_def: ScenarioDefinition,
    progress: ScenarioProgress,
    state: ScenarioOrchestratorState
) -> dict:
    """
    Execute skill, recording milestones and checkpoints.

    This is where you call YOUR SKILL.
    """

    milestones = [0, 30, 50, 70, 90, 100]  # Percentage checkpoints
    results = {"batches": [], "quality": {}}

    input_items = generate_test_data(
        size=scenario_def.input_size,
        characteristics=scenario_def.dataset_characteristics
    )

    batch_size = scenario_def.skill_params.get("batch_size", 10)

    for batch_idx, batch in enumerate(chunks(input_items, batch_size)):
        # Execute skill on this batch
        # Replace this with your actual skill invocation
        batch_result = await invoke_skill(
            skill_name=skill_name,
            input_data=batch,
            params=scenario_def.skill_params
        )

        results["batches"].append(batch_result)
        progress.batch_count += 1
        progress.items_processed += len(batch)

        # Update progress percentage
        progress.progress_pct = (progress.items_processed / scenario_def.input_size) * 100

        # Check if we've reached a milestone
        reached_milestones = [m for m in milestones if m <= progress.progress_pct]
        new_milestones = [m for m in reached_milestones if m not in progress.milestones_reached]

        for milestone in new_milestones:
            progress.milestones_reached.append(milestone)
            elapsed = int(time.time() * 1000) - progress.start_time_ms
            progress.elapsed_checkpoints[f"milestone_{milestone}"] = elapsed

            print(f"  [{progress.scenario_id}] Reached {milestone}% at {elapsed}ms")

            # Optional: Wait for other scenarios at major milestones
            if milestone in [30, 70]:
                await synchronize_at_milestone(milestone, state)

    # Score results
    results["quality"] = calculate_quality_metrics(results["batches"], scenario_def.quality_metrics)
    progress.quality_scores = results["quality"]

    return results
```

### Synchronization Node

```python
async def synchronize_at_milestone(
    milestone_pct: int,
    state: ScenarioOrchestratorState,
    timeout_seconds: int = 30
) -> bool:
    """
    Optional: Wait for other scenarios at major milestones.

    Returns True if all scenarios reached milestone, False if timeout.
    """

    start = time.time()
    milestone_key = f"checkpoint_{milestone_pct}"

    while time.time() - start < timeout_seconds:
        simple_at_milestone = milestone_pct in state["progress_simple"].milestones_reached
        medium_at_milestone = milestone_pct in state["progress_medium"].milestones_reached
        complex_at_milestone = milestone_pct in state["progress_complex"].milestones_reached

        all_reached = simple_at_milestone and medium_at_milestone and complex_at_milestone

        if all_reached:
            state["sync_points"][milestone_key] = True
            print(f"[SYNC] All scenarios reached {milestone_pct}%")
            return True

        # Check if any scenario failed
        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
```

### Aggregator Node

```python
async def scenario_aggregator(state: ScenarioOrchestratorState) -> dict:
    """
    Collect all scenario results and synthesize findings.
    """

    print("[AGGREGATOR] Combining results from all scenarios")

    aggregated = {
        "orchestration_id": state["orchestration_id"],
        "skill": state["skill_name"],
        "timestamp": datetime.now().isoformat(),

        # Raw results
        "results_by_scenario": {
            "simple": state["progress_simple"].partial_results[-1] if state["progress_simple"].partial_results else {},
            "medium": state["progress_medium"].partial_results[-1] if state["progress_medium"].partial_results else {},
            "complex": state["progress_complex"].partial_results[-1] if state["progress_complex"].partial_results else {},
        },

        # Metrics
        "metrics": {},

        # Comparison
        "comparison": {},

        # Recommendations
        "recommendations": []
    }

    # Calculate comparative metrics
    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,
            "errors": len(progress.errors)
        }

    # Compare quality vs. complexity
    simple_quality = state["progress_simple"].quality_scores.get("overall", 0)
    medium_quality = state["progress_medium"].quality_scores.get("overall", 0)
    complex_quality = state["progress_complex"].quality_scores.get("overall", 0)

    aggregated["comparison"]["quality_ranking"] = {
        "best": max(
            ("simple", simple_quality),
            ("medium", medium_quality),
            ("complex", complex_quality),
            key=lambda x: x[1]
        )[0],
        "scores": {
            "simple": simple_quality,
            "medium": medium_quality,
            "complex": complex_quality
        }
    }

    # Time complexity analysis
    simple_time = state["progress_simple"].elapsed_ms
    medium_time = state["progress_medium"].elapsed_ms
    complex_time = state["progress_complex"].elapsed_ms

    simple_size = 100 * 1.0
    medium_size = 100 * 3.0
    complex_size = 100 * 8.0

    aggregated["comparison"]["time_per_item_ms"] = {
        "simple": simple_time / simple_size,
        "medium": medium_time / medium_size,
        "complex": complex_time / complex_size,
    }

    # Identify scaling issues
    if complex_time / complex_size > simple_time / simple_size * 2:
        aggregated["recommendations"].append("Sublinear scaling—excellent performance with increased load")
    elif complex_time / complex_size < simple_time / simple_size * 0.8:
        aggregated["recommendations"].append("Superlinear scaling—overhead increases with load")

    # Success patterns
    success_patterns = []
    for scenario_id in ["simple", "medium", "complex"]:
        if state[f"progress_{scenario_id}"].status == "complete" and state[f"progress_{scenario_id}"].errors == []:
            success_patterns.append(scenario_id)

    aggregated["recommendations"].append(f"Successful in all scenarios: {', '.join(success_patterns)}")

    return {"final_results": aggregated}
```

## 3. Graph Construction

```python
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.types import Command

def build_scenario_orchestrator(
    checkpointer: PostgresSaver | None = None
) -> Any:
    """
    Build the complete orchestration graph.
    """

    graph = StateGraph(ScenarioOrchestratorState)

    # Nodes
    graph.add_node("supervisor", scenario_supervisor)
    graph.add_node("scenario_worker", scenario_worker)
    graph.add_node("aggregator", scenario_aggregator)

    # Edges
    graph.add_edge(START, "supervisor")

    # Fan-out: supervisor sends to 3 parallel workers
    graph.add_conditional_edges(
        "supervisor",
        lambda _: ["scenario_worker", "scenario_worker", "scenario_worker"]
    )

    # Workers converge at aggregator
    graph.add_edge("scenario_worker", "aggregator")
    graph.add_edge("aggregator", END)

    # Compile with checkpointing
    return graph.compile(checkpointer=checkpointer)
```

## 4. Invocation Example

```python
import asyncio
import uuid
from langgraph.checkpoint.postgres import PostgresSaver
from psycopg_pool import ConnectionPool

async def main():
    # Setup checkpointing — from_conn_string is a @contextmanager; use an
    # explicit pool + PostgresSaver constructor for a long-running orchestrator
    pool = ConnectionPool(
        "postgresql://user:password@localhost/orchestkit",
        max_size=20,
        kwargs={"autocommit": True, "prepare_threshold": 0},
    )
    checkpointer = PostgresSaver(pool)
    checkpointer.setup()  # first run creates the checkpoint tables

    # Build orchestrator
    app = build_scenario_orchestrator(checkpointer=checkpointer)

    # Prepare initial state
    initial_state: ScenarioOrchestratorState = {
        "orchestration_id": f"demo-{uuid.uuid4().hex[:8]}",
        "start_time_unix": int(time.time()),
        "skill_name": "your-skill-name",
        "skill_version": "1.0.0",

        # Scenarios
        "scenario_simple": ScenarioDefinition(
            name="simple",
            difficulty="easy",
            complexity_multiplier=1.0,
            input_size=100,
            dataset_characteristics={"distribution": "uniform"},
            time_budget_seconds=30,
            memory_limit_mb=256,
            error_tolerance=0.0,
            skill_params={"batch_size": 10, "cache_enabled": True},
            expected_quality="basic",
            quality_metrics=["accuracy", "coverage"]
        ),
        "scenario_medium": ScenarioDefinition(
            name="medium",
            difficulty="intermediate",
            complexity_multiplier=3.0,
            input_size=300,
            dataset_characteristics={"distribution": "uniform"},
            time_budget_seconds=90,
            memory_limit_mb=512,
            error_tolerance=0.05,
            skill_params={"batch_size": 50, "cache_enabled": True},
            expected_quality="good",
            quality_metrics=["accuracy", "coverage"]
        ),
        "scenario_complex": ScenarioDefinition(
            name="complex",
            difficulty="advanced",
            complexity_multiplier=8.0,
            input_size=800,
            dataset_characteristics={"distribution": "skewed"},
            time_budget_seconds=300,
            memory_limit_mb=1024,
            error_tolerance=0.1,
            skill_params={"batch_size": 100, "cache_enabled": True, "parallel_workers": 4},
            expected_quality="excellent",
            quality_metrics=["accuracy", "coverage", "latency"]
        ),

        # Progress tracking
        "progress_simple": ScenarioProgress(scenario_id="simple"),
        "progress_medium": ScenarioProgress(scenario_id="medium"),
        "progress_complex": ScenarioProgress(scenario_id="complex"),

        # Synchronization
        "sync_points": {},
        "last_sync_time": 0,
    }

    # Run with thread_id for checkpointing
    config = {"configurable": {"thread_id": f"orch-{initial_state['orchestration_id']}"}}

    print("Starting multi-scenario orchestration...")
    result = await app.ainvoke(initial_state, config=config)

    # Print results
    final = result["final_results"]
    print("\n" + "="*60)
    print("ORCHESTRATION RESULTS")
    print("="*60)
    print(f"Orchestration ID: {final['orchestration_id']}")
    print(f"Skill: {final['skill']}")
    print("\nQuality Comparison:")
    for scenario, score in final["comparison"]["quality_ranking"]["scores"].items():
        print(f"  {scenario}: {score:.2f}")
    print("\nTime per Item (ms):")
    for scenario, time in final["comparison"]["time_per_item_ms"].items():
        print(f"  {scenario}: {time:.2f}ms")
    print("\nRecommendations:")
    for rec in final["recommendations"]:
        print(f"  • {rec}")

if __name__ == "__main__":
    asyncio.run(main())
```

## 5. Helper Functions

```python
def chunks(items: list, size: int):
    """Split items into chunks."""
    for i in range(0, len(items), size):
        yield items[i:i + size]

def generate_test_data(size: int, characteristics: dict) -> list:
    """Generate test data based on scenario characteristics."""
    import random

    distribution = characteristics.get("distribution", "uniform")

    if distribution == "uniform":
        return [{"id": i, "value": random.random()} for i in range(size)]
    elif distribution == "skewed":
        # Zipfian distribution
        return [
            {"id": i, "value": random.random() ** 2}
            for i in range(size)
        ]
    else:
        return [{"id": i, "value": random.random()} for i in range(size)]

async def invoke_skill(
    skill_name: str,
    input_data: list,
    params: dict
) -> dict:
    """
    Invoke your skill here.

    Replace with actual skill invocation.
    """
    # Simulate processing
    await asyncio.sleep(0.1)  # 100ms per batch

    return {
        "processed": len(input_data),
        "quality_score": 0.85 + (random.random() * 0.15),
        "timestamp": datetime.now().isoformat()
    }

def calculate_quality_metrics(batches: list, metrics: list[str]) -> dict:
    """Calculate quality metrics across batches."""
    if not batches:
        return {metric: 0.0 for metric in metrics}

    scores = {
        "accuracy": sum(b.get("quality_score", 0) for b in batches) / len(batches),
        "coverage": 1.0,
    }

    return {metric: scores.get(metric, 0.0) for metric in metrics}
```

## 6. Streaming Results (Real-time Progress)

```python
async def stream_orchestration_progress(
    app,
    initial_state: ScenarioOrchestratorState,
    config: dict
):
    """
    Stream progress updates as scenarios execute.
    """

    async for step in app.astream(initial_state, config=config, stream_mode="updates"):
        print(f"\n[UPDATE] {step}")

        # Extract progress from step
        if "progress_simple" in step:
            p = step["progress_simple"]
            print(f"  Simple: {p.progress_pct:.1f}% ({p.items_processed} items)")

        if "progress_medium" in step:
            p = step["progress_medium"]
            print(f"  Medium: {p.progress_pct:.1f}% ({p.items_processed} items)")

        if "progress_complex" in step:
            p = step["progress_complex"]
            print(f"  Complex: {p.progress_pct:.1f}% ({p.items_processed} items)")
```

## Key Features

1. **Fan-Out/Fan-In**: All 3 scenarios execute in parallel
2. **Milestone Tracking**: Progress recorded at key checkpoints
3. **Synchronization**: Optional wait points at 30% and 70%
4. **Error Isolation**: One scenario's failure doesn't block others
5. **Checkpointing**: State saved to PostgreSQL for recovery
6. **Aggregation**: Cross-scenario analysis and recommendations
7. **Streaming**: Real-time progress updates

## Testing

```python
@pytest.mark.asyncio
async def test_multi_scenario_orchestration():
    # Mock checkpointer
    from langgraph.checkpoint.memory import MemorySaver

    app = build_scenario_orchestrator(checkpointer=MemorySaver())

    initial_state = {...}  # Setup
    config = {"configurable": {"thread_id": "test-123"}}

    result = await app.ainvoke(initial_state, config=config)

    assert result["final_results"]["orchestration_id"]
    assert "simple" in result["final_results"]["metrics"]
    assert "medium" in result["final_results"]["metrics"]
    assert "complex" in result["final_results"]["metrics"]
```


### Ork Delta

# Agent Orchestration: ork delta

House decisions rescued from retired reference and rule files during the wrap + delta
conversion (2026-07-31). Maintained for src/skills/agent-orchestration. Vendor tutorials
are not restated here; see "Upstream coverage (do not restate)" in SKILL.md for where each
retired topic now lives.

## Bound every agent loop: 5-15 max steps, 10-20 message memory window
Why: House defaults set in the v2.0.0 consolidation of agent-loops (metadata.json, Feb 2026). The unbounded while-True ReAct loop and the context-overflow crash are the two failure modes these numbers exist to prevent; upstream tutorials omit the guardrail values.
Upstream: OpenAI function calling guide (https://platform.openai.com/docs/guides/function-calling); LangGraph plan-and-execute tutorials (context7: /langchain-ai/langgraph).

## Cap multi-agent fan-out at 8 parallel workers, 30s per-worker timeout, one retry
Why: House defaults from the v2.0.0 consolidation of multi-agent-orchestration (Feb 2026), rescued from the retired coordination-patterns reference. One hung specialist must never stall the whole fan-out, and beyond 8 parallel specialists synthesis quality degraded in ork demo runs.
Upstream: asyncio.gather with return_exceptions (https://docs.python.org/3/library/asyncio-task.html); supervisor routing in ork:langgraph.

## End every multi-agent run with a synthesis step, never return raw gathered output
Why: v2.0.0 house rule (Feb 2026), rescued from the retired multi-synthesis rule: asyncio.gather output is a list of disconnected findings, so every ork multi-agent flow closes with category grouping, an executive summary, and a confidence score.
Upstream: SKILL.md Quick Start (synthesize_findings example); LangGraph multi-agent tutorials (context7: /langchain-ai/langgraph).

## Default new multi-agent work to LangGraph and never mix frameworks in one project
Why: House recommendation from the retired frameworks-comparison rule (v2.0.0, Feb 2026): LangGraph is the only framework ork ships a dedicated skill for (ork:langgraph), and framework mixing is a recorded complexity explosion in SKILL.md Common Mistakes.
Upstream: references/framework-comparison.md (in-skill decision matrix); per-framework docs listed in SKILL.md Upstream coverage.

## Scale multi-scenario demos 1x/3x/8x with 30s/90s/300s budgets, always exactly 3 scenarios
Why: House convention for ork skill demos from the v2.0.0 consolidation of multi-scenario-orchestration (Feb 2026). Exponential rather than linear scaling because most skills carry fixed per-run overhead that linear steps fail to expose.
Upstream: rules/scenario-orchestrator.md and rules/scenario-routing.md (maintained in this skill; no vendor surface covers it).

## Never block forever at a scenario sync point: time out, log, proceed, isolate failures
Why: Design decision rescued from the retired multi-scenario state-machine and architecture references (Feb 2026): milestone sync is for interactive demos, free-running is for production, and a slow or failed scenario must never deadlock its siblings.
Upstream: rules/scenario-routing.md (synchronize_at_milestone with timeout and failure isolation).



---

## Checklists (1)

### Framework Selection

# Framework Selection Checklist

Choose the right multi-agent framework.

## Requirements Analysis

- [ ] Use case clearly defined
- [ ] Complexity level assessed (single vs multi-agent)
- [ ] State management needs identified
- [ ] Human-in-the-loop requirements defined
- [ ] Observability needs documented

## Framework Evaluation

### LangGraph
- [ ] Need complex stateful workflows
- [ ] Require persistence and checkpoints
- [ ] Want streaming support
- [ ] Need human-in-the-loop
- [ ] Already using LangChain ecosystem

### CrewAI
- [ ] Role-based collaboration pattern
- [ ] Hierarchical team structure
- [ ] Agent delegation needed
- [ ] Quick prototyping required
- [ ] Built-in memory preferred

### OpenAI Agents SDK
- [ ] OpenAI-native ecosystem
- [ ] Handoff pattern fits use case
- [ ] Need built-in guardrails
- [ ] Want OpenAI tracing
- [ ] Simpler agent definition preferred

### Microsoft Agent Framework
- [ ] Enterprise compliance requirements
- [ ] Using Azure ecosystem
- [ ] Need A2A protocol support
- [ ] Want AutoGen+SK merger features
- [ ] Long-term Microsoft support preferred

### AG2 (Community AutoGen)
- [ ] Open-source flexibility priority
- [ ] Community-driven development OK
- [ ] AutoGen familiarity exists
- [ ] Custom modifications needed

## Technical Considerations

- [ ] Team expertise with framework
- [ ] Framework maturity level acceptable
- [ ] Community support adequate
- [ ] Documentation quality sufficient
- [ ] Production readiness validated

## Integration Assessment

- [ ] Observability tool compatibility (Langfuse, etc.)
- [ ] LLM provider compatibility
- [ ] Existing codebase integration
- [ ] Testing framework support
- [ ] CI/CD pipeline compatibility

## Risk Mitigation

- [ ] Fallback strategy defined
- [ ] Framework lock-in assessed
- [ ] Migration path understood
- [ ] Version update strategy
- [ ] Community health evaluated

## Decision Documentation

- [ ] Framework choice documented
- [ ] Rationale recorded
- [ ] Alternatives considered listed
- [ ] Trade-offs acknowledged
- [ ] Review date scheduled