Multi-Agent Orchestration: The Real Production Guide

// src/orchestrator/task-decomposer.ts
import Anthropic from '@anthropic-ai/sdk'

const anthropic = new Anthropic()

interface Task {
  id: string
  type: 'code_generation' | 'code_review' | 'testing' | 'documentation' | 'deployment'
  description: string
  dependencies: string[] // IDs of tasks that must complete first
  assignedAgent: string
  status: 'pending' | 'in_progress' | 'completed' | 'failed'
  result?: unknown
  error?: string
}

export async function decomposeRequest(request: string, context: {
  projectType: string
  existingFeatures: string[]
  techStack: string[]
}): Promise<Task[]> {
  const response = await anthropic.messages.create({
    model: 'claude-sonnet-4-5',
    system: `You are a technical project manager who breaks down feature requests into executable subtasks.
    
Output a JSON array of tasks. Each task has:
- id: unique string
- type: one of [code_generation, code_review, testing, documentation, deployment]
- description: specific, executable description (not vague)
- dependencies: array of task ids that must complete before this task
- assignedAgent: which specialist agent handles this

Rules:
- Be specific. "Generate the user profile schema" not "work on database"
- Make dependencies explicit. A review task always depends on its generation task
- Testing always depends on code generation
- Deployment always depends on testing`,
    messages: [
      {
        role: 'user',
        content: `Decompose this feature request into tasks:
${request}

Project context:
- Type: ${context.projectType}
- Tech stack: ${context.techStack.join(', ')}
- Existing features: ${context.existingFeatures.join(', ')}`,
      },
    ],
  })

  const content = response.content[0]
  if (content.type !== 'text') throw new Error('Expected text response')

  return JSON.parse(content.text) as Task[]
}

// src/orchestrator/agent-router.ts
const AGENT_REGISTRY = {
  code_generation: {
    model: 'claude-opus-4-5',
    tools: ['read_file', 'write_file', 'search_codebase', 'run_linter'],
    systemPrompt: 'You are a senior software engineer. Write clean, tested, production-ready code.',
  },
  code_review: {
    model: 'claude-opus-4-5',
    tools: ['read_file', 'search_codebase', 'check_types'],
    systemPrompt: 'You are a code reviewer. Be critical. Find bugs, security issues, and design problems. Do not be lenient because the code compiles.',
  },
  testing: {
    model: 'claude-sonnet-4-5',
    tools: ['read_file', 'write_file', 'run_tests', 'check_coverage'],
    systemPrompt: 'You are a QA engineer. Write comprehensive tests. Aim for 95% coverage. Test happy paths and failure cases equally.',
  },
  documentation: {
    model: 'claude-haiku-3-5',  // Cheaper for doc generation
    tools: ['read_file', 'write_file'],
    systemPrompt: 'You write clear, concise technical documentation for developers.',
  },
  deployment: {
    model: 'claude-sonnet-4-5',
    tools: ['run_build', 'run_tests', 'deploy', 'check_health', 'rollback'],
    systemPrompt: 'You are a DevOps engineer. Deploy carefully. Verify before and after. Roll back at the first sign of problems.',
  },
}

export function routeTask(task: Task) {
  const agent = AGENT_REGISTRY[task.type]
  if (!agent) throw new Error(`No agent registered for task type: ${task.type}`)
  return agent
}

// src/orchestrator/state-manager.ts
import { Redis } from '@upstash/redis'

const redis = Redis.fromEnv()

interface WorkflowState {
  workflowId: string
  request: string
  tasks: Task[]
  startedAt: Date
  completedAt?: Date
  status: 'running' | 'completed' | 'failed' | 'partial'
}

export class WorkflowStateManager {
  async createWorkflow(workflowId: string, request: string, tasks: Task[]): Promise<void> {
    const state: WorkflowState = {
      workflowId,
      request,
      tasks,
      startedAt: new Date(),
      status: 'running',
    }
    // Persist with 24-hour TTL
    await redis.set(`workflow:${workflowId}`, JSON.stringify(state), { ex: 86400 })
  }

  async updateTaskStatus(
    workflowId: string,
    taskId: string,
    status: Task['status'],
    result?: unknown,
    error?: string
  ): Promise<void> {
    const state = await this.getWorkflow(workflowId)
    if (!state) throw new Error(`Workflow ${workflowId} not found`)

    const task = state.tasks.find(t => t.id === taskId)
    if (!task) throw new Error(`Task ${taskId} not found in workflow ${workflowId}`)

    task.status = status
    if (result !== undefined) task.result = result
    if (error !== undefined) task.error = error

    await redis.set(`workflow:${workflowId}`, JSON.stringify(state), { ex: 86400 })
  }

  async getReadyTasks(workflowId: string): Promise<Task[]> {
    const state = await this.getWorkflow(workflowId)
    if (!state) return []

    const completedTaskIds = new Set(
      state.tasks.filter(t => t.status === 'completed').map(t => t.id)
    )

    return state.tasks.filter(task =>
      task.status === 'pending' &&
      task.dependencies.every(depId => completedTaskIds.has(depId))
    )
  }

  private async getWorkflow(workflowId: string): Promise<WorkflowState | null> {
    const data = await redis.get<string>(`workflow:${workflowId}`)
    return data ? JSON.parse(data) : null
  }
}

// Every agent validates its input before execution
async function validateAgentInput(task: Task, context: Record<string, unknown>): Promise<void> {
  const required = getRequiredContextFields(task.type)

  for (const field of required) {
    if (context[field] === undefined || context[field] === null) {
      throw new Error(
        `Task ${task.id} (${task.type}) missing required context: ${field}. ` +
        `This dependency task may have failed.`
      )
    }
  }

  // Validate output quality from dependency tasks
  if (task.type === 'code_review' && context.generatedCode) {
    const code = context.generatedCode as string
    if (code.length < 10) {
      throw new Error(`Code generation produced suspiciously short output: ${code}`)
    }
  }
}

// Quality gate before passing output to downstream agents
async function validateAgentOutput(task: Task, output: unknown): Promise<boolean> {
  switch (task.type) {
    case 'code_generation':
      // Code must be non-empty and not contain obvious errors
      const code = output as string
      if (!code || code.length < 20) return false
      if (code.includes('TODO: implement') && !code.includes('// TODO')) return false
      return true

    case 'testing':
      // Tests must exist and pass
      const testResult = output as { passed: number; failed: number; coverage: number }
      if (testResult.failed > 0) return false
      if (testResult.coverage < 80) return false
      return true

    case 'code_review':
      // Review must not contain CRITICAL issues
      const review = output as { severity: 'CRITICAL' | 'HIGH' | 'MEDIUM' | 'LOW'; issues: string[] }
      return review.severity !== 'CRITICAL'

    default:
      return true
  }
}

// Track agent health and remove unreliable agents
class AgentHealthMonitor {
  private failureCount = new Map<string, number>()
  private disabledAgents = new Set<string>()
  private readonly FAILURE_THRESHOLD = 3

  recordFailure(agentType: string): void {
    const count = (this.failureCount.get(agentType) ?? 0) + 1
    this.failureCount.set(agentType, count)

    if (count >= this.FAILURE_THRESHOLD) {
      this.disabledAgents.add(agentType)
      console.error(`Agent ${agentType} disabled after ${count} consecutive failures`)
      this.scheduleReenablement(agentType, 5 * 60 * 1000) // Try again in 5 minutes
    }
  }

  recordSuccess(agentType: string): void {
    this.failureCount.set(agentType, 0)
    this.disabledAgents.delete(agentType)
  }

  isAvailable(agentType: string): boolean {
    return !this.disabledAgents.has(agentType)
  }

  private scheduleReenablement(agentType: string, delayMs: number): void {
    setTimeout(() => {
      this.disabledAgents.delete(agentType)
      this.failureCount.set(agentType, 0)
      console.log(`Agent ${agentType} re-enabled for testing`)
    }, delayMs)
  }
}

// src/orchestrator/resource-manager.ts
import pLimit from 'p-limit'

export class ResourceManager {
  // Limit concurrent LLM API calls
  private apiCallLimiter = pLimit(10)

  // Limit concurrent tool executions (they use more memory)
  private toolExecutionLimiter = pLimit(5)

  // Token budget per workflow (prevents runaway costs)
  private tokenBudgets = new Map<string, number>()
  private tokenUsage = new Map<string, number>()

  async executeWithRateLimit<T>(
    fn: () => Promise<T>,
    type: 'api_call' | 'tool_execution'
  ): Promise<T> {
    const limiter = type === 'api_call'
      ? this.apiCallLimiter
      : this.toolExecutionLimiter

    return limiter(() => fn())
  }

  setTokenBudget(workflowId: string, budget: number): void {
    this.tokenBudgets.set(workflowId, budget)
    this.tokenUsage.set(workflowId, 0)
  }

  recordTokenUsage(workflowId: string, tokens: number): void {
    const current = this.tokenUsage.get(workflowId) ?? 0
    this.tokenUsage.set(workflowId, current + tokens)
  }

  isOverBudget(workflowId: string): boolean {
    const budget = this.tokenBudgets.get(workflowId) ?? Infinity
    const used = this.tokenUsage.get(workflowId) ?? 0
    return used > budget
  }
}

// Every agent invocation produces a structured trace
interface AgentTrace {
  workflowId: string
  taskId: string
  agentType: string
  startTime: Date
  endTime: Date
  durationMs: number
  inputSummary: string
  outputSummary: string
  tokenUsage: { input: number; output: number }
  success: boolean
  error?: string
}

async function traceAgentExecution(
  workflowId: string,
  task: Task,
  fn: () => Promise<unknown>
): Promise<unknown> {
  const startTime = new Date()
  let result: unknown
  let error: Error | undefined

  try {
    result = await fn()
    return result
  } catch (e) {
    error = e instanceof Error ? e : new Error(String(e))
    throw error
  } finally {
    const trace: AgentTrace = {
      workflowId,
      taskId: task.id,
      agentType: task.type,
      startTime,
      endTime: new Date(),
      durationMs: Date.now() - startTime.getTime(),
      inputSummary: summarizeForLog(task.description),
      outputSummary: result ? summarizeForLog(result) : 'no output',
      tokenUsage: { input: 0, output: 0 }, // populated from LLM response
      success: !error,
      error: error?.message,
    }

    await persistTrace(trace)
  }
}