┌────────────────────────────────────────────────────────────────┐
│                    Agent Communication Complexity               │
├────────────────────────────────────────────────────────────────┤
│                                                                 │
│  Agents:  5      10      20      50      100     200           │
│  Paths:   10     45      190     1,225   4,950   19,900        │
│                                                                 │
│  └─ Linear growth would be: 5x, 10x, 20x, 50x, 100x, 200x      │
│  └─ Actual is:       2x, 4.5x, 19x,  122.5x, 495x, 1,990x     │
│                                                                 │
│  This is why point-to-point communication fails at scale.      │
└────────────────────────────────────────────────────────────────┘

// Anti-pattern: Synchronous agent spawning
// When 1,000 tasks arrive simultaneously, agents get overwhelmed

async function processTasks(tasks) {
  // ❌ BAD: Creates agents for every task immediately
  const results = await Promise.all(
    tasks.map(task => spawnAgentAndProcess(task))
  );
  return results;
}

// Result: 1,000 simultaneous API calls to OpenAI
// Cost spike: $500 in 30 seconds
// Rate limit exceeded, most requests fail

// Anti-pattern: No circuit breakers or timeouts
// When one agent fails, dependent agents keep trying

async function researchAndWrite(topic) {
  const research = await researchAgent.query(topic); // Fails after 30s
  // ❌ BAD: No timeout, no fallback
  const draft = await writerAgent.create(research); // Times out too
  // ❌ BAD: Error propagates
  const edited = await editorAgent.review(draft); // Also fails
  return edited;
}

// Result: Three failed operations, wasted tokens, angry users

// Anti-pattern: Storing full state for every agent interaction
// Each conversation includes all previous context

const conversation = {
  messages: [
    { role: 'system', content: 'You are a helpful assistant...' }, // 500 tokens
    { role: 'user', content: 'Previous question...' },              // 1,200 tokens
    { role: 'assistant', content: 'Previous answer...' },           // 2,800 tokens
    { role: 'user', content: 'Another question...' },               // 1,500 tokens
    { role: 'assistant', content: 'Another answer...' },          // 2,200 tokens
    // ... continues for 50+ turns
  ]
};

// Result: 10th turn sends 45,000 tokens, costs $1.35
// 50th turn sends 225,000 tokens, costs $6.75
// User stops using the feature

// Anti-pattern: Naive retry without backoff or coordination
// Multiple systems retry simultaneously

async function callWithRetry(fn, maxRetries = 3) {
  for (let i = 0; i < maxRetries; i++) {
    try {
      return await fn();
    } catch (e) {
      // ❌ BAD: No exponential backoff
      // ❌ BAD: All retries happen at same time across agents
      await sleep(1000); // Fixed 1s delay
    }
  }
  throw new Error('Failed after retries');
}

// Result: When API is degraded, retry load multiplies 3x
// 100 agents × 3 retries = 300 requests hitting struggling API

┌─────────────────────────────────────────────────────────────────┐
│              Event-Driven Agent Architecture                     │
├─────────────────────────────────────────────────────────────────┤
│                                                                   │
│   ┌──────────┐    ┌──────────┐    ┌──────────┐    ┌──────────┐   │
│   │  Agent A │    │  Agent B │    │  Agent C │    │  Agent D │   │
│   │ (Research)│    │(Writing) │    │(Editing) │    │(Review)  │   │
│   └────┬─────┘    └────┬─────┘    └────┬─────┘    └────┬─────┘   │
│        │               │               │               │          │
│        │  PUBLISH      │  PUBLISH      │  PUBLISH      │          │
│        ▼               ▼               ▼               ▼          │
│   ┌──────────────────────────────────────────────────────────┐   │
│   │                    Message Bus (Redis/RabbitMQ)          │   │
│   │  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐         │   │
│   │  │research.done│  │ draft.ready │  │ edit.complete│         │   │
│   │  └─────────────┘  └─────────────┘  └─────────────┘         │   │
│   └──────────────────────────────────────────────────────────┘   │
│        │               │               │               │          │
│        │  SUBSCRIBE    │  SUBSCRIBE    │  SUBSCRIBE    │          │
│        ▼               ▼               ▼               ▼          │
│   ┌──────────┐    ┌──────────┐    ┌──────────┐    ┌──────────┐   │
│   │   Listens│    │  Listens │    │  Listens │    │  Listens │   │
│   │ for tasks│    │for draft │    │for review │    │for final │   │
│   └──────────┘    └──────────┘    └──────────┘    └──────────┘   │
│                                                                   │
│   Key Benefit: Agents don't know about each other directly        │
│   Agent B doesn't know WHO created the draft, just THAT one exists│
└─────────────────────────────────────────────────────────────────┘

// Core event types for agent orchestration

interface AgentEvent {
  // Metadata
  eventId: string;           // UUID v4
  eventType: string;         // Fully qualified: agent.task.assigned
  timestamp: ISO8601;        // UTC
  version: string;           // Schema version: "2.1.0"
  
  // Context
  correlationId: string;     // Traces request across agents
  causationId: string;       // Points to triggering event
  sessionId: string;         // Groups related events
  
  // Content
  payload: unknown;          // Event-specific data
  
  // Routing
  source: AgentRef;          // Who emitted the event
  target?: AgentRef;         // Intended recipient (optional)
  priority: 'low' | 'normal' | 'high' | 'critical';
  
  // Lifecycle
  ttl?: number;             // Time-to-live in seconds
  retryCount: number;        // How many times processed
  deadline?: ISO8601;        // When event becomes stale
}

interface AgentRef {
  agentId: string;
  agentType: string;         // 'research', 'writing', 'review'
  version: string;
  instanceId: string;        // For multi-instance agents
}

// Specific event types
interface TaskAssignedEvent extends AgentEvent {
  eventType: 'agent.task.assigned';
  payload: {
    taskId: string;
    taskType: string;
    requirements: TaskRequirements;
    context: SharedContext;
    maxDuration: number;       // Seconds
    costBudget?: number;      // Max tokens/cost
  };
}

interface TaskCompletedEvent extends AgentEvent {
  eventType: 'agent.task.completed';
  payload: {
    taskId: string;
    result: TaskResult;
    metrics: {
      duration: number;       // Seconds
      tokensUsed: number;
      cost: number;
      retries: number;
    };
  };
}

interface TaskFailedEvent extends AgentEvent {
  eventType: 'agent.task.failed';
  payload: {
    taskId: string;
    error: {
      code: string;
      message: string;
      recoverable: boolean;
    };
    attempts: number;
    nextRetry?: ISO8601;
  };
}

# Consistent naming enables routing and filtering

# Format: domain.entity.action.status
# Examples:

events:
  # Task lifecycle
  - agent.task.assigned
  - agent.task.started
  - agent.task.progress      # Intermediate updates
  - agent.task.completed
  - agent.task.failed
  - agent.task.cancelled
  - agent.task.expired       # TTL exceeded
  
  # Agent lifecycle
  - agent.instance.started
  - agent.instance.heartbeat
  - agent.instance.stopped
  - agent.instance.crashed
  - agent.instance.scale.up
  - agent.instance.scale.down
  
  # System events
  - system.circuit.opened
  - system.circuit.closed
  - system.rate.limit.exceeded
  - system.cost.budget.warning
  - system.cost.budget.exceeded

// n8n Code Node: Publish Events to Redis Streams

const Redis = require('ioredis');
const { v4: uuidv4 } = require('uuid');

const redis = new Redis({
  host: $env.REDIS_HOST,
  port: $env.REDIS_PORT,
  password: $env.REDIS_PASSWORD,
  maxRetriesPerRequest: 3
});

async function publishEvent(eventData) {
  const event = {
    eventId: uuidv4(),
    eventType: eventData.type,
    timestamp: new Date().toISOString(),
    version: '2.1.0',
    correlationId: eventData.correlationId || uuidv4(),
    causationId: eventData.causationId,
    sessionId: eventData.sessionId,
    payload: JSON.stringify(eventData.payload),
    source: JSON.stringify(eventData.source),
    target: eventData.target ? JSON.stringify(eventData.target) : null,
    priority: eventData.priority || 'normal',
    retryCount: eventData.retryCount || 0,
    deadline: eventData.deadline
  };

  // Publish to stream with maxlen to prevent unbounded growth
  const streamKey = `events:${eventData.type.split('.')[0]}`;
  await redis.xadd(
    streamKey,
    'MAXLEN', '~', 10000,  // Approximate max length
    '*',                   // Auto-generate ID
    ...Object.entries(event).flat()
  );

  // Also publish to type-specific stream for targeted consumers
  await redis.xadd(
    `events:type:${eventData.type}`,
    'MAXLEN', '~', 5000,
    '*',
    ...Object.entries(event).flat()
  );

  return event;
}

// Example: Publish task assigned event
const event = await publishEvent({
  type: 'agent.task.assigned',
  correlationId: $json.correlationId,
  causationId: $json.triggerEventId,
  sessionId: $json.sessionId,
  payload: {
    taskId: `task-${Date.now()}`,
    taskType: 'content-research',
    requirements: {
      topic: $json.topic,
      depth: 'comprehensive',
      sources: ['industry-reports', 'competitor-analysis'],
      outputFormat: 'structured-json'
    },
    context: {
      industry: $json.industry,
      targetAudience: $json.audience,
      brandVoice: $json.brandVoice
    },
    maxDuration: 300,  // 5 minutes
    costBudget: 0.50   // $0.50 max
  },
  source: {
    agentId: 'orchestrator-1',
    agentType: 'workflow-orchestrator',
    version: '2.0.0',
    instanceId: 'orc-1a'
  },
  priority: 'high'
});

return {
  json: {
    published: true,
    eventId: event.eventId,
    correlationId: event.correlationId
  }
};

// n8n Code Node: Consume Events with Redis Consumer Groups

const Redis = require('ioredis');

const redis = new Redis({
  host: $env.REDIS_HOST,
  port: $env.REDIS_PORT,
  password: $env.REDIS_PASSWORD
});

const CONSUMER_GROUP = 'writing-agents';
const CONSUMER_NAME = `writer-${$env.INSTANCE_ID || '1'}`;
const STREAM_KEY = 'events:type:agent.task.assigned';

async function consumeEvents() {
  try {
    // Ensure consumer group exists
    try {
      await redis.xgroup('CREATE', STREAM_KEY, CONSUMER_GROUP, '$', 'MKSTREAM');
    } catch (e) {
      // Group already exists, ignore error
      if (!e.message.includes('already exists')) throw e;
    }

    // Read events with timeout
    const messages = await redis.xreadgroup(
      'GROUP', CONSUMER_GROUP, CONSUMER_NAME,
      'BLOCK', 5000,  // Wait up to 5 seconds
      'COUNT', 10,    // Max 10 messages per batch
      'STREAMS', STREAM_KEY, '>'
    );

    if (!messages) {
      return { json: { processed: 0, events: [] } };
    }

    const processed = [];

    for (const [, entries] of messages) {
      for (const [id, fields] of entries) {
        const event = Object.fromEntries(
          fields.map((v, i) => i % 2 === 0 ? [v, fields[i + 1]] : null).filter(Boolean)
        );

        try {
          // Parse payload
          const payload = JSON.parse(event.payload);
          
          // Check priority
          if (event.priority === 'critical') {
            // Process immediately, don't batch
            await processCriticalTask(payload);
          } else {
            // Queue for batch processing
            await queueTask(payload);
          }

          // Acknowledge success
          await redis.xack(STREAM_KEY, CONSUMER_GROUP, id);
          
          processed.push({
            eventId: event.eventId,
            status: 'success',
            taskId: payload.taskId
          });

        } catch (error) {
          // Log failure but don't ack - will be retried
          processed.push({
            eventId: event.eventId,
            status: 'failed',
            error: error.message,
            willRetry: event.retryCount < 3
          });

          // If max retries exceeded, send to dead letter queue
          if (event.retryCount >= 3) {
            await redis.xadd(
              'events:dead-letter',
              '*',
              'originalEvent', JSON.stringify(event),
              'error', error.message,
              'failedAt', new Date().toISOString()
            );
            await redis.xack(STREAM_KEY, CONSUMER_GROUP, id);
          }
        }
      }
    }

    return {
      json: {
        processed: processed.length,
        events: processed
      }
    };

  } finally {
    await redis.quit();
  }
}

async function processCriticalTask(payload) {
  // Critical path processing
  // Skip queue, execute immediately
  const result = await executeAgentTask(payload);
  return result;
}

async function queueTask(payload) {
  // Add to internal queue for batch processing
  await redis.lpush('queue:pending-tasks', JSON.stringify(payload));
}

return await consumeEvents();

// n8n Workflow: Multi-Priority Queue System

const Redis = require('ioredis');
const redis = new Redis($env.REDIS_URL);

const PRIORITIES = ['critical', 'high', 'normal', 'low'];
const QUEUE_TTLS = {
  critical: 60,    // 1 minute
  high: 300,       // 5 minutes
  normal: 3600,    // 1 hour
  low: 86400       // 24 hours
};

class PriorityQueueManager {
  constructor(redis) {
    this.redis = redis;
    this.queues = {
      critical: 'agent:queue:critical',
      high: 'agent:queue:high',
      normal: 'agent:queue:normal',
      low: 'agent:queue:low'
    };
  }

  async enqueue(task, priority = 'normal') {
    const queue = this.queues[priority] || this.queues.normal;
    const ttl = QUEUE_TTLS[priority];

    const enrichedTask = {
      ...task,
      priority,
      enqueuedAt: Date.now(),
      expiresAt: Date.now() + (ttl * 1000),
      attempts: 0
    };

    // Use LPUSH for stack-like behavior (LIFO for same priority)
    // Or RPUSH for FIFO - depends on use case
    await this.redis.lpush(queue, JSON.stringify(enrichedTask));
    
    // Track queue depth metrics
    await this.redis.hincrby('metrics:queue-depth', priority, 1);

    return {
      taskId: enrichedTask.id,
      priority,
      queueDepth: await this.redis.llen(queue)
    };
  }

  async dequeue() {
    // Try each priority in order
    for (const priority of PRIORITIES) {
      const queue = this.queues[priority];
      const task = await this.redis.rpop(queue);
      
      if (task) {
        await this.redis.hincrby('metrics:queue-depth', priority, -1);
        return {
          task: JSON.parse(task),
          priority,
          waitTime: Date.now() - JSON.parse(task).enqueuedAt
        };
      }
    }
    
    return null; // No tasks available
  }

  async getQueueStats() {
    const stats = {};
    for (const [priority, queue] of Object.entries(this.queues)) {
      const length = await this.redis.llen(queue);
      const oldestTask = await this.redis.lindex(queue, -1);
      const oldestAge = oldestTask 
        ? Date.now() - JSON.parse(oldestTask).enqueuedAt 
        : 0;
      
      stats[priority] = {
        depth: length,
        oldestTaskAgeMs: oldestAge,
        maxWaitTime: QUEUE_TTLS[priority]
      };
    }
    return stats;
  }
}

// Usage in n8n
const queueManager = new PriorityQueueManager(redis);

// Producer: Add task with priority
const enqueueResult = await queueManager.enqueue({
  id: `task-${Date.now()}`,
  type: 'content-generation',
  payload: {
    topic: $json.topic,
    requirements: $json.requirements
  }
}, $json.priority || 'normal');

// Consumer: Get next task
const nextTask = await queueManager.dequeue();

// Monitor queue health
const stats = await queueManager.getQueueStats();

return {
  json: {
    enqueued: enqueueResult,
    nextTask,
    queueStats: stats
  }
};

// n8n Workflow: Delayed Task Scheduler

const Redis = require('ioredis');
const redis = new Redis($env.REDIS_URL);

class DelayedScheduler {
  constructor(redis) {
    this.redis = redis;
    this.delayedSet = 'agent:delayed-tasks';
    this.readyQueue = 'agent:queue:ready';
  }

  async schedule(task, executeAt) {
    const scheduledTask = {
      ...task,
      scheduledAt: Date.now(),
      executeAt: executeAt.getTime(),
      id: `delayed-${Date.now()}-${Math.random().toString(36).substr(2, 9)}`
    };

    // Add to sorted set with execute time as score
    await this.redis.zadd(
      this.delayedSet,
      scheduledTask.executeAt,
      JSON.stringify(scheduledTask)
    );

    return {
      scheduled: true,
      taskId: scheduledTask.id,
      executeAt: scheduledTask.executeAt,
      delayMs: scheduledTask.executeAt - Date.now()
    };
  }

  async pollAndPromote() {
    const now = Date.now();
    
    // Get all tasks ready to execute (score <= now)
    const readyTasks = await this.redis.zrangebyscore(
      this.delayedSet,
      '-inf',
      now,
      'WITHSCORES'
    );

    const promoted = [];

    for (let i = 0; i < readyTasks.length; i += 2) {
      const task = readyTasks[i];
      const score = readyTasks[i + 1];
      
      // Remove from delayed set
      await this.redis.zrem(this.delayedSet, task);
      
      // Add to ready queue
      await this.redis.lpush(this.readyQueue, task);
      
      promoted.push(JSON.parse(task));
    }

    // Get next task time for scheduling
    const nextTask = await this.redis.zrange(this.delayedSet, 0, 0, 'WITHSCORES');
    const nextExecution = nextTask.length > 0 
      ? parseInt(nextTask[1]) 
      : null;

    return {
      promoted: promoted.length,
      tasks: promoted.map(t => t.id),
      nextExecution,
      delayedRemaining: await this.redis.zcard(this.delayedSet)
    };
  }

  async getSchedule(windowHours = 24) {
    const now = Date.now();
    const windowEnd = now + (windowHours * 60 * 60 * 1000);
    
    const tasks = await this.redis.zrangebyscore(
      this.delayedSet,
      now,
      windowEnd,
      'WITHSCORES'
    );

    return tasks.reduce((acc, val, idx, arr) => {
      if (idx % 2 === 0) {
        const task = JSON.parse(val);
        const score = parseInt(arr[idx + 1]);
        acc.push({
          ...task,
          executeAt: new Date(score).toISOString(),
          timeUntil: score - now
        });
      }
      return acc;
    }, []);
  }
}

// Usage
const scheduler = new DelayedScheduler(redis);

// Schedule a task for later
const scheduleResult = await scheduler.schedule({
  type: 'social-post',
  payload: {
    platform: 'linkedin',
    content: $json.postContent,
    hashtags: $json.hashtags
  }
}, new Date($json.scheduledTime));

// Poll for ready tasks (run this every minute via cron)
const promotionResult = await scheduler.pollAndPromote();

// View upcoming schedule
const upcoming = await scheduler.getSchedule(48); // Next 48 hours

return {
  json: {
    scheduled: scheduleResult,
    promoted: promotionResult,
    upcoming: upcoming.slice(0, 10) // Show first 10
  }
};

// n8n Workflow: Saga Pattern Implementation

const Redis = require('ioredis');
const redis = new Redis($env.REDIS_URL);

class SagaOrchestrator {
  constructor(redis) {
    this.redis = redis;
  }

  async startSaga(sagaId, steps) {
    const saga = {
      id: sagaId,
      status: 'running',
      startedAt: Date.now(),
      steps: steps.map((step, idx) => ({
        id: `step-${idx}`,
        name: step.name,
        action: step.action,
        compensation: step.compensation,
        status: 'pending',
        input: null,
        output: null
      })),
      currentStep: 0
    };

    await this.redis.setex(
      `saga:${sagaId}`,
      3600, // 1 hour TTL
      JSON.stringify(saga)
    );

    return this.executeNextStep(sagaId);
  }

  async executeNextStep(sagaId) {
    const saga = JSON.parse(await this.redis.get(`saga:${sagaId}`));
    
    if (saga.currentStep >= saga.steps.length) {
      // Saga complete
      saga.status = 'completed';
      saga.completedAt = Date.now();
      await this.redis.setex(`saga:${sagaId}`, 86400, JSON.stringify(saga));
      return { status: 'completed', saga };
    }

    const step = saga.steps[saga.currentStep];
    step.status = 'executing';
    step.startedAt = Date.now();
    await this.redis.setex(`saga:${sagaId}`, 3600, JSON.stringify(saga));

    try {
      // Execute the step action
      const result = await this.executeAction(step.action, saga);
      
      step.status = 'completed';
      step.output = result;
      step.completedAt = Date.now();
      saga.currentStep++;

      await this.redis.setex(`saga:${sagaId}`, 3600, JSON.stringify(saga));
      
      // Continue to next step
      return this.executeNextStep(sagaId);

    } catch (error) {
      step.status = 'failed';
      step.error = error.message;
      
      // Trigger compensation
      await this.compensate(saga, saga.currentStep);
      
      saga.status = 'compensated';
      saga.failedAt = Date.now();
      await this.redis.setex(`saga:${sagaId}`, 86400, JSON.stringify(saga));
      
      return { status: 'compensated', saga, error: error.message };
    }
  }

  async executeAction(action, saga) {
    // Dispatch to appropriate agent based on action type
    switch (action.type) {
      case 'agent-task':
        return await this.callAgent(action.agentType, action.payload);
      case 'webhook':
        return await fetch(action.url, {
          method: 'POST',
          body: JSON.stringify({ sagaId: saga.id, ...action.payload })
        });
      case 'n8n-subflow':
        return await this.triggerSubflow(action.workflowId, action.payload);
      default:
        throw new Error(`Unknown action type: ${action.type}`);
    }
  }

  async compensate(saga, failedStepIndex) {
    // Run compensation in reverse order
    for (let i = failedStepIndex; i >= 0; i--) {
      const step = saga.steps[i];
      if (step.compensation && step.output) {
        try {
          await this.executeAction(step.compensation, saga);
          step.compensationStatus = 'completed';
        } catch (e) {
          step.compensationStatus = 'failed';
          // Log for manual intervention
          await this.redis.lpush('saga:compensation-failures', JSON.stringify({
            sagaId: saga.id,
            stepId: step.id,
            error: e.message,
            timestamp: new Date().toISOString()
          }));
        }
      }
    }
  }

  async getSagaStatus(sagaId) {
    const saga = await this.redis.get(`saga:${sagaId}`);
    return saga ? JSON.parse(saga) : null;
  }
}

// Usage: Content creation saga
const orchestrator = new SagaOrchestrator(redis);

const sagaSteps = [
  {
    name: 'research',
    action: {
      type: 'agent-task',
      agentType: 'research-agent',
      payload: { topic: $json.topic, depth: 'comprehensive' }
    },
    compensation: {
      type: 'webhook',
      url: 'https://api.company.com/cleanup-research',
      payload: { action: 'cleanup' }
    }
  },
  {
    name: 'draft',
    action: {
      type: 'agent-task',
      agentType: 'writer-agent',
      payload: { /* uses previous step output */ }
    },
    compensation: {
      type: 'webhook',
      url: 'https://api.company.com/cleanup-draft',
      payload: { action: 'cleanup' }
    }
  },
  {
    name: 'publish',
    action: {
      type: 'webhook',
      url: 'https://cms.company.com/api/publish',
      payload: { /* uses draft output */ }
    },
    compensation: {
      type: 'webhook',
      url: 'https://cms.company.com/api/unpublish',
      payload: { action: 'unpublish' }
    }
  }
];

const sagaId = `saga-${Date.now()}`;
const result = await orchestrator.startSaga(sagaId, sagaSteps);

return {
  json: result
};

// n8n Workflow: Circuit Breaker for Agent Calls

const Redis = require('ioredis');
const redis = new Redis($env.REDIS_URL);

class CircuitBreaker {
  constructor(redis, config) {
    this.redis = redis;
    this.config = {
      failureThreshold: config.failureThreshold || 5,
      successThreshold: config.successThreshold || 3,
      timeout: config.timeout || 60000, // 1 minute
      ...config
    };
    this.stateKey = `circuit:${config.name}`;
  }

  async getState() {
    const state = await this.redis.get(this.stateKey);
    if (!state) {
      return {
        status: 'CLOSED',
        failures: 0,
        successes: 0,
        lastFailureTime: null,
        openedAt: null
      };
    }
    return JSON.parse(state);
  }

  async recordSuccess() {
    const state = await this.getState();
    
    if (state.status === 'HALF_OPEN') {
      state.successes++;
      if (state.successes >= this.config.successThreshold) {
        state.status = 'CLOSED';
        state.failures = 0;
        state.successes = 0;
        await this.redis.publish('circuit:events', JSON.stringify({
          circuit: this.config.name,
          event: 'CLOSED',
          timestamp: new Date().toISOString()
        }));
      }
    }
    
    await this.redis.setex(this.stateKey, 3600, JSON.stringify(state));
    return state;
  }

  async recordFailure() {
    const state = await this.getState();
    state.failures++;
    state.lastFailureTime = Date.now();

    if (state.status === 'CLOSED' && state.failures >= this.config.failureThreshold) {
      state.status = 'OPEN';
      state.openedAt = Date.now();
      await this.redis.publish('circuit:events', JSON.stringify({
        circuit: this.config.name,
        event: 'OPENED',
        timestamp: new Date().toISOString()
      }));
    }

    await this.redis.setex(this.stateKey, 3600, JSON.stringify(state));
    return state;
  }

  async canExecute() {
    const state = await this.getState();
    
    if (state.status === 'CLOSED') {
      return { allowed: true, state };
    }

    if (state.status === 'OPEN') {
      const timeOpen = Date.now() - state.openedAt;
      
      if (timeOpen > this.config.timeout) {
        // Transition to HALF_OPEN
        state.status = 'HALF_OPEN';
        state.successes = 0;
        await this.redis.setex(this.stateKey, 3600, JSON.stringify(state));
        await this.redis.publish('circuit:events', JSON.stringify({
          circuit: this.config.name,
          event: 'HALF_OPEN',
          timestamp: new Date().toISOString()
        }));
        return { allowed: true, state, trial: true };
      }
      
      return { 
        allowed: false, 
        state,
        retryAfter: Math.ceil((this.config.timeout - timeOpen) / 1000)
      };
    }

    if (state.status === 'HALF_OPEN') {
      return { allowed: true, state, trial: true };
    }
  }

  async execute(operation) {
    const permission = await this.canExecute();
    
    if (!permission.allowed) {
      throw new Error(`Circuit breaker OPEN for ${this.config.name}. Retry after ${permission.retryAfter}s`);
    }

    try {
      const result = await operation();
      await this.recordSuccess();
      return { result, state: permission.state, trial: permission.trial };
    } catch (error) {
      await this.recordFailure();
      throw error;
    }
  }
}

// Usage for LLM API calls
const openAICircuit = new CircuitBreaker(redis, {
  name: 'openai-api',
  failureThreshold: 5,
  successThreshold: 3,
  timeout: 60000
});

try {
  const result = await openAICircuit.execute(async () => {
    const response = await fetch('https://api.openai.com/v1/chat/completions', {
      method: 'POST',
      headers: {
        'Authorization': `Bearer ${$env.OPENAI_API_KEY}`,
        'Content-Type': 'application/json'
      },
      body: JSON.stringify({
        model: 'gpt-4o',
        messages: $json.messages
      })
    });
    
    if (!response.ok) {
      throw new Error(`OpenAI API error: ${response.status}`);
    }
    
    return await response.json();
  });

  return {
    json: {
      success: true,
      data: result.result,
      circuitState: result.state.status,
      trial: result.trial || false
    }
  };

} catch (error) {
  return {
    json: {
      success: false,
      error: error.message,
      fallback: 'Using cached response or alternative model'
    }
  };
}

// n8n Workflow: Smart Retry with Exponential Backoff and Jitter

class RetryWithBackoff {
  constructor(config = {}) {
    this.maxRetries = config.maxRetries || 5;
    this.baseDelay = config.baseDelay || 1000; // 1 second
    this.maxDelay = config.maxDelay || 60000; // 60 seconds
    this.jitterFactor = config.jitterFactor || 0.3; // 30% randomization
  }

  calculateDelay(attempt) {
    // Exponential: base * 2^attempt
    const exponential = this.baseDelay * Math.pow(2, attempt);
    const capped = Math.min(exponential, this.maxDelay);
    
    // Add jitter: ±jitterFactor
    const jitter = capped * this.jitterFactor * (Math.random() * 2 - 1);
    const finalDelay = capped + jitter;
    
    return Math.max(0, Math.floor(finalDelay));
  }

  async execute(operation, context) {
    const attempts = [];
    
    for (let attempt = 0; attempt < this.maxRetries; attempt++) {
      const startTime = Date.now();
      
      try {
        const result = await operation();
        
        attempts.push({
          attempt: attempt + 1,
          duration: Date.now() - startTime,
          success: true
        });
        
        return {
          success: true,
          result,
          attempts,
          totalDuration: Date.now() - attempts[0].startTime
        };
        
      } catch (error) {
        const duration = Date.now() - startTime;
        
        attempts.push({
          attempt: attempt + 1,
          duration,
          success: false,
          error: error.message
        });

        if (attempt === this.maxRetries - 1) {
          // Final attempt failed
          return {
            success: false,
            error: error.message,
            attempts,
            totalDuration: Date.now() - attempts[0].startTime
          };
        }

        // Calculate and apply backoff
        const delay = this.calculateDelay(attempt);
        attempts[attempt].nextDelay = delay;
        
        await this.sleep(delay);
      }
    }
  }

  sleep(ms) {
    return new Promise(resolve => setTimeout(resolve, ms));
  }
}

// Usage
const retryConfig = {
  maxRetries: 5,
  baseDelay: 2000,    // Start with 2 seconds
  maxDelay: 32000,    // Cap at 32 seconds
  jitterFactor: 0.2   // 20% jitter
};

const retrier = new RetryWithBackoff(retryConfig);

const result = await retrier.execute(async () => {
  const response = await fetch('https://api.anthropic.com/v1/messages', {
    method: 'POST',
    headers: {
      'x-api-key': $env.ANTHROPIC_API_KEY,
      'Content-Type': 'application/json'
    },
    body: JSON.stringify({
      model: 'claude-3-7-sonnet-20250219',
      max_tokens: 4096,
      messages: $json.messages
    })
  });

  if (response.status === 429) {
    const retryAfter = response.headers.get('retry-after');
    throw new Error(`Rate limited. Retry after: ${retryAfter}`);
  }

  if (!response.ok) {
    throw new Error(`API error: ${response.status}`);
  }

  return await response.json();
});

return {
  json: result
};

// n8n Workflow: Dead Letter Queue Pattern

const Redis = require('ioredis');
const redis = new Redis($env.REDIS_URL);

class DeadLetterQueue {
  constructor(redis) {
    this.redis = redis;
    this.dlqKey = 'queue:dead-letter';
    this.maxSize = 10000; // Keep last 10k failed events
  }

  async add(event, error, context) {
    const deadEvent = {
      originalEvent: event,
      error: {
        message: error.message,
        stack: error.stack,
        code: error.code
      },
      context: {
        ...context,
        failedAt: new Date().toISOString(),
        retryCount: event.retryCount || 0
      },
      id: `dlq-${Date.now()}-${Math.random().toString(36).substr(2, 9)}`
    };

    // Add to DLQ with timestamp score for sorting
    await this.redis.zadd(
      this.dlqKey,
      Date.now(),
      JSON.stringify(deadEvent)
    );

    // Trim to max size
    const currentSize = await this.redis.zcard(this.dlqKey);
    if (currentSize > this.maxSize) {
      await this.redis.zremrangebyrank(this.dlqKey, 0, currentSize - this.maxSize - 1);
    }

    // Alert if DLQ is growing fast
    if (currentSize > this.maxSize * 0.8) {
      await this.redis.publish('alerts:dlq', JSON.stringify({
        severity: 'warning',
        message: `DLQ at ${Math.round((currentSize / this.maxSize) * 100)}% capacity`,
        timestamp: new Date().toISOString()
      }));
    }

    return deadEvent.id;
  }

  async list(limit = 50, offset = 0) {
    const events = await this.redis.zrevrange(
      this.dlqKey,
      offset,
      offset + limit - 1
    );
    
    return events.map(e => {
      const parsed = JSON.parse(e);
      return {
        ...parsed,
        age: Date.now() - new Date(parsed.context.failedAt).getTime()
      };
    });
  }

  async retry(eventId) {
    // Find event
    const events = await this.redis.zrange(this.dlqKey, 0, -1);
    const event = events.find(e => JSON.parse(e).id === eventId);
    
    if (!event) {
      throw new Error('Event not found in DLQ');
    }

    const parsed = JSON.parse(event);
    
    // Re-queue original event
    await this.redis.lpush('queue:pending-tasks', JSON.stringify({
      ...parsed.originalEvent,
      retryCount: (parsed.originalEvent.retryCount || 0) + 1,
      retryFromDLQ: true,
      originalFailure: parsed.error
    }));

    // Remove from DLQ
    await this.redis.zrem(this.dlqKey, event);

    return { success: true, message: 'Event re-queued' };
  }

  async getStats() {
    const total = await this.redis.zcard(this.dlqKey);
    
    // Get time distribution
    const now = Date.now();
    const oneHourAgo = now - 3600000;
    const oneDayAgo = now - 86400000;
    
    const lastHour = await this.redis.zcount(this.dlqKey, oneHourAgo, now);
    const lastDay = await this.redis.zcount(this.dlqKey, oneDayAgo, now);
    
    // Group by error type
    const allEvents = await this.redis.zrange(this.dlqKey, 0, -1);
    const errorTypes = allEvents.reduce((acc, e) => {
      const error = JSON.parse(e).error.code || 'unknown';
      acc[error] = (acc[error] || 0) + 1;
      return acc;
    }, {});

    return {
      total,
      lastHour,
      lastDay,
      errorTypes,
      oldestEvent: total > 0 ? await this.redis.zrange(this.dlqKey, 0, 0) : null
    };
  }
}

// Usage in n8n
const dlq = new DeadLetterQueue(redis);

// When processing fails
try {
  await processEvent($json.event);
} catch (error) {
  const dlqId = await dlq.add($json.event, error, {
    workflowId: $env.WORKFLOW_ID,
    nodeId: $env.NODE_ID
  });
  
  return {
    json: {
      processed: false,
      movedToDLQ: true,
      dlqId,
      error: error.message
    }
  };
}

// Monitor DLQ stats
const stats = await dlq.getStats();

// List recent failures for review
const recentFailures = await dlq.list(20);

return {
  json: {
    dlqStats: stats,
    recentFailures
  }
};

// n8n Workflow: Token Optimization Strategies

class TokenOptimizer {
  constructor() {
    this.cache = new Map(); // In production, use Redis
    this.modelCosts = {
      'gpt-4o': { input: 2.50, output: 10.00, per: 1000000 }, // per 1M tokens
      'gpt-4o-mini': { input: 0.15, output: 0.60, per: 1000000 },
      'claude-3-7-sonnet': { input: 3.00, output: 15.00, per: 1000000 },
      'claude-3-5-haiku': { input: 0.80, output: 4.00, per: 1000000 }
    };
  }

  // Strategy 1: Intelligent Model Selection
  selectModel(taskComplexity) {
    const selection = {
      simple: 'gpt-4o-mini',      // Classification, extraction
      medium: 'claude-3-5-haiku', // Analysis, summarization
      complex: 'gpt-4o',          // Reasoning, generation
      critical: 'claude-3-7-sonnet' // High-stakes decisions
    };
    
    return selection[taskComplexity] || selection.medium;
  }

  // Strategy 2: Context Compression
  compressContext(messages, maxTokens = 4000) {
    let totalTokens = this.estimateTokens(messages);
    
    if (totalTokens <= maxTokens) {
      return messages;
    }

    // Priority order: system > last user > last assistant > older messages
    const systemMsgs = messages.filter(m => m.role === 'system');
    const userMsgs = messages.filter(m => m.role === 'user');
    const assistantMsgs = messages.filter(m => m.role === 'assistant');

    const compressed = [...systemMsgs];
    
    // Always keep the most recent exchange
    if (userMsgs.length > 0) {
      compressed.push(userMsgs[userMsgs.length - 1]);
    }
    if (assistantMsgs.length > 0) {
      compressed.push(assistantMsgs[assistantMsgs.length - 1]);
    }

    // Add older messages if space permits
    const remaining = maxTokens - this.estimateTokens(compressed);
    if (remaining > 0) {
      const olderMessages = [...userMsgs.slice(0, -1), ...assistantMsgs.slice(0, -1)]
        .sort((a, b) => messages.indexOf(a) - messages.indexOf(b));
      
      for (const msg of olderMessages) {
        if (this.estimateTokens([...compressed, msg]) <= maxTokens) {
          compressed.push(msg);
        } else {
          // Summarize old messages instead of dropping
          const summary = this.createSummary(compressed);
          return [systemMsgs[0], { role: 'user', content: `Previous context: ${summary}` }, compressed[compressed.length - 1]];
        }
      }
    }

    return compressed;
  }

  // Strategy 3: Response Caching
  async getCachedResponse(promptHash) {
    // In production: Redis or similar
    const cached = this.cache.get(promptHash);
    if (cached && Date.now() - cached.timestamp < 3600000) { // 1 hour TTL
      return cached.response;
    }
    return null;
  }

  async cacheResponse(promptHash, response) {
    this.cache.set(promptHash, {
      response,
      timestamp: Date.now()
    });
  }

  // Strategy 4: Batch Processing
  async batchProcess(items, batchSize = 10) {
    const results = [];
    const batches = [];
    
    for (let i = 0; i < items.length; i += batchSize) {
      batches.push(items.slice(i, i + batchSize));
    }

    for (const batch of batches) {
      // Process batch in single call
      const batchResult = await this.processBatch(batch);
      results.push(...batchResult);
    }

    return results;
  }

  // Strategy 5: Streaming for Long Outputs
  async streamProcess(messages, onChunk) {
    let fullResponse = '';
    let tokenCount = 0;
    
    const response = await fetch('https://api.openai.com/v1/chat/completions', {
      method: 'POST',
      headers: {
        'Authorization': `Bearer ${$env.OPENAI_API_KEY}`,
        'Content-Type': 'application/json'
      },
      body: JSON.stringify({
        model: 'gpt-4o',
        messages,
        stream: true,
        max_tokens: 4000
      })
    });

    const reader = response.body.getReader();
    
    while (true) {
      const { done, value } = await reader.read();
      if (done) break;
      
      const chunk = new TextDecoder().decode(value);
      const lines = chunk.split('\n').filter(line => line.trim());
      
      for (const line of lines) {
        if (line.startsWith('data: ')) {
          const data = line.slice(6);
          if (data === '[DONE]') continue;
          
          try {
            const parsed = JSON.parse(data);
            const content = parsed.choices[0]?.delta?.content || '';
            fullResponse += content;
            tokenCount++;
            
            // Call progress handler
            if (onChunk) {
              await onChunk(content, tokenCount);
            }
            
            // Early termination if we have enough
            if (tokenCount >= 1000 && this.hasCompleteAnswer(fullResponse)) {
              reader.releaseLock();
              return { response: fullResponse, tokens: tokenCount, truncated: true };
            }
          } catch (e) {
            // Ignore parsing errors in stream
          }
        }
      }
    }

    return { response: fullResponse, tokens: tokenCount };
  }

  // Helper methods
  estimateTokens(messages) {
    // Rough estimation: ~4 characters per token
    const text = messages.map(m => m.content).join('');
    return Math.ceil(text.length / 4);
  }

  createSummary(messages) {
    // In production: Use a smaller model for summarization
    return `Conversation with ${messages.length} messages about ${messages[0].content.substring(0, 50)}...`;
  }

  hasCompleteAnswer(text) {
    // Detect if response seems complete
    return text.trim().endsWith('.') || text.trim().endsWith('}');
  }

  calculateCost(model, inputTokens, outputTokens) {
    const costs = this.modelCosts[model];
    if (!costs) return 0;
    
    const inputCost = (inputTokens / costs.per) * costs.input;
    const outputCost = (outputTokens / costs.per) * costs.output;
    
    return inputCost + outputCost;
  }
}

// Usage
const optimizer = new TokenOptimizer();

// Example workflow
const task = $json.task;
const complexity = optimizer.assessComplexity(task); // Your logic here

// Select appropriate model
const model = optimizer.selectModel(complexity);

// Check cache
const cacheKey = optimizer.hashPrompt(task.prompt);
const cached = await optimizer.getCachedResponse(cacheKey);

if (cached) {
  return {
    json: {
      result: cached,
      cost: 0,
      source: 'cache'
    }
  };
}

// Compress context
const compressedMessages = optimizer.compressContext(task.messages, 3000);

// Call API with optimized parameters
const response = await optimizer.streamProcess(compressedMessages, (chunk, count) => {
  // Update progress if needed
  console.log(`Received ${count} tokens`);
});

// Cache result
await optimizer.cacheResponse(cacheKey, response.response);

// Calculate cost
const cost = optimizer.calculateCost(model, optimizer.estimateTokens(compressedMessages), response.tokens);

return {
  json: {
    result: response.response,
    cost,
    tokens: response.tokens,
    truncated: response.truncated || false,
    model
  }
};