Agent-to-Agent Communication: Protocols That Work

// src/agents/messages/code-review-message.ts
import { z } from 'zod'

export const CodeReviewIssueSchema = z.object({
  file: z.string(),
  line: z.number().optional(),
  severity: z.enum(['CRITICAL', 'HIGH', 'MEDIUM', 'LOW', 'INFO']),
  category: z.enum(['security', 'bug', 'performance', 'style', 'test_coverage', 'documentation']),
  description: z.string(),
  suggestedFix: z.string().optional(),
  mustFix: z.boolean(), // CRITICAL and HIGH should be mustFix: true
})

export const CodeReviewResultSchema = z.object({
  reviewId: z.string(),
  generationTaskId: z.string(), // Links back to the generation task
  overallVerdict: z.enum(['APPROVE', 'REQUEST_CHANGES', 'REJECT']),
  summary: z.string(),
  issues: z.array(CodeReviewIssueSchema),
  requiredChanges: z.array(z.string()), // Specific, actionable change requests
  optionalImprovements: z.array(z.string()),
  testCoverageAdequate: z.boolean(),
  securityConcerns: z.boolean(),
})

export type CodeReviewIssue = z.infer<typeof CodeReviewIssueSchema>
export type CodeReviewResult = z.infer<typeof CodeReviewResultSchema>

// Validation at both send and receive
export function validateCodeReviewResult(data: unknown): CodeReviewResult {
  const result = CodeReviewResultSchema.safeParse(data)
  if (!result.success) {
    throw new Error(
      `Invalid code review message: ${result.error.flatten().fieldErrors}`
    )
  }
  return result.data
}

// src/agents/message-bus.ts
type MessageHandler<T> = (message: T) => Promise<void>

class AgentMessageBus {
  private handlers = new Map<string, MessageHandler<unknown>[]>()
  private messageQueue: { topic: string; message: unknown }[] = []
  private processing = false

  subscribe<T>(topic: string, handler: MessageHandler<T>): () => void {
    const handlers = this.handlers.get(topic) ?? []
    handlers.push(handler as MessageHandler<unknown>)
    this.handlers.set(topic, handlers)

    // Return unsubscribe function
    return () => {
      const current = this.handlers.get(topic) ?? []
      this.handlers.set(topic, current.filter(h => h !== handler))
    }
  }

  async publish(topic: string, message: unknown): Promise<void> {
    this.messageQueue.push({ topic, message })
    if (!this.processing) {
      await this.processQueue()
    }
  }

  private async processQueue(): Promise<void> {
    this.processing = true

    while (this.messageQueue.length > 0) {
      const { topic, message } = this.messageQueue.shift()!
      const handlers = this.handlers.get(topic) ?? []

      await Promise.all(handlers.map(handler => handler(message)))
    }

    this.processing = false
  }
}

export const messageBus = new AgentMessageBus()

// Topic constants prevent typos
export const TOPICS = {
  CODE_GENERATED: 'agent:code:generated',
  CODE_REVIEWED: 'agent:code:reviewed',
  TESTS_GENERATED: 'agent:tests:generated',
  TESTS_PASSED: 'agent:tests:passed',
  TESTS_FAILED: 'agent:tests:failed',
  DEPLOYMENT_REQUESTED: 'agent:deployment:requested',
  DEPLOYMENT_COMPLETED: 'agent:deployment:completed',
  DEPLOYMENT_FAILED: 'agent:deployment:failed',
} as const

// src/agents/shared-workspace.ts
import { Redis } from '@upstash/redis'

const redis = Redis.fromEnv()

interface VersionedContent {
  content: string
  version: number
  lastModifiedBy: string
  lastModifiedAt: Date
}

export class SharedWorkspace {
  async read(key: string): Promise<VersionedContent | null> {
    const data = await redis.get<string>(key)
    return data ? JSON.parse(data) : null
  }

  // Optimistic write: fails if version doesn't match
  async write(
    key: string,
    content: string,
    expectedVersion: number,
    agentId: string
  ): Promise<{ success: true; version: number } | { success: false; reason: string }> {
    // Lua script for atomic check-and-set
    const script = `
      local current = redis.call('get', KEYS[1])
      if current then
        local data = cjson.decode(current)
        if data.version ~= tonumber(ARGV[1]) then
          return {0, 'version_conflict', data.version}
        end
      else
        if tonumber(ARGV[1]) ~= 0 then
          return {0, 'not_found', 0}
        end
      end
      
      local newVersion = tonumber(ARGV[1]) + 1
      local newData = cjson.encode({
        content = ARGV[2],
        version = newVersion,
        lastModifiedBy = ARGV[3],
        lastModifiedAt = ARGV[4]
      })
      redis.call('set', KEYS[1], newData)
      return {1, 'ok', newVersion}
    `

    const result = await redis.eval(
      script,
      [key],
      [expectedVersion.toString(), content, agentId, new Date().toISOString()]
    ) as [number, string, number]

    if (result[0] === 1) {
      return { success: true, version: result[2] }
    } else {
      return { success: false, reason: result[1] }
    }
  }

  // Retry loop for agents that need to write with contention
  async writeWithRetry(
    key: string,
    transform: (current: VersionedContent | null) => string,
    agentId: string,
    maxAttempts = 5
  ): Promise<{ success: true } | { success: false; reason: string }> {
    for (let attempt = 0; attempt < maxAttempts; attempt++) {
      const current = await this.read(key)
      const expectedVersion = current?.version ?? 0
      const newContent = transform(current)

      const result = await this.write(key, newContent, expectedVersion, agentId)

      if (result.success) return { success: true }

      if (result.reason === 'version_conflict') {
        // Exponential backoff before retry
        await sleep(Math.pow(2, attempt) * 100)
        continue
      }

      return { success: false, reason: result.reason }
    }

    return { success: false, reason: 'max_attempts_exceeded' }
  }
}

// src/agents/event-system.ts
type EventType =
  | 'code.committed'
  | 'tests.failed'
  | 'deployment.succeeded'
  | 'deployment.failed'
  | 'quality.gate.passed'
  | 'quality.gate.failed'

interface AgentEvent {
  eventId: string
  type: EventType
  timestamp: Date
  payload: Record<string, unknown>
  sourceAgent: string
  workflowId: string
}

class EventDrivenOrchestrator {
  private subscriptions = new Map<EventType, ((event: AgentEvent) => Promise<void>)[]>()

  // Agent registration: declare what events trigger what reactions
  registerAgent(config: {
    agentId: string
    reactsTo: EventType[]
    handler: (event: AgentEvent) => Promise<void>
  }): void {
    for (const eventType of config.reactsTo) {
      const handlers = this.subscriptions.get(eventType) ?? []
      handlers.push(config.handler)
      this.subscriptions.set(eventType, handlers)
    }
  }

  async emit(event: AgentEvent): Promise<void> {
    // Persist event for audit trail
    await this.persistEvent(event)

    // Notify subscribers
    const handlers = this.subscriptions.get(event.type) ?? []
    await Promise.all(handlers.map(handler => handler(event)))
  }
}

// Configuration: who reacts to what
const orchestrator = new EventDrivenOrchestrator()

orchestrator.registerAgent({
  agentId: 'test-runner',
  reactsTo: ['code.committed'],
  handler: async (event) => {
    await testRunnerAgent.run({ codeLocation: event.payload.path })
  },
})

orchestrator.registerAgent({
  agentId: 'deployment-agent',
  reactsTo: ['quality.gate.passed'],
  handler: async (event) => {
    await deploymentAgent.deploy({ workflowId: event.workflowId })
  },
})

orchestrator.registerAgent({
  agentId: 'rollback-agent',
  reactsTo: ['tests.failed', 'deployment.failed'],
  handler: async (event) => {
    await rollbackAgent.rollback({ workflowId: event.workflowId })
  },
})

orchestrator.registerAgent({
  agentId: 'notification-agent',
  reactsTo: ['deployment.succeeded', 'deployment.failed', 'tests.failed'],
  handler: async (event) => {
    await notificationAgent.notify(event)
  },
})

// src/agents/negotiation-protocol.ts
type ConflictingClaim = {
  agentId: string
  claim: string
  evidence: string
  proposedAction: string
  priority: number // Lower is higher priority
}

async function resolveConflict(
  claims: ConflictingClaim[],
  context: Record<string, unknown>
): Promise<{ resolution: string; requiresHuman: boolean }> {
  // Sort by priority (security > performance > convenience)
  const sorted = [...claims].sort((a, b) => a.priority - b.priority)
  const highestPriority = sorted[0]

  // Search for compromise position
  const compromisePrompt = `
Multiple agents have conflicting recommendations:

${sorted.map(c => `${c.agentId}: ${c.claim}\nProposed: ${c.proposedAction}`).join('\n\n')}

Context: ${JSON.stringify(context)}

Find a compromise that:
1. Addresses the highest-priority concern (${highestPriority.agentId}: ${highestPriority.claim})
2. Satisfies as many other concerns as possible
3. Is actionable (not "evaluate later")

If no compromise exists, say "ESCALATE: [reason]"
  `

  const response = await anthropic.messages.create({
    model: 'claude-sonnet-4-5',
    messages: [{ role: 'user', content: compromisePrompt }],
  })

  const resolution = response.content[0].type === 'text'
    ? response.content[0].text
    : ''

  if (resolution.startsWith('ESCALATE:')) {
    return {
      resolution: resolution.replace('ESCALATE: ', ''),
      requiresHuman: true,
    }
  }

  return { resolution, requiresHuman: false }
}