企业级多智能体通信“背压流控与死信队列”（Multi-Agent Backpressure & Dead-Letter Queue Governance）架构实战：破解异构响应延迟倾斜、消息缓冲区雪崩与确定性重放补救

/**
 * 企业级多智能体背压流控与死信队列（DLQ）治理引擎
 * 具备基于信用值的流控、GPU负载自适应调节、毒丸消息隔离与确定性重放能力
 */

import { EventEmitter } from 'node:events';
import { createHash } from 'node:crypto';

// 消息优先级枚举定义
export enum MessagePriority {
  P0_CRITICAL = 0, // 紧急核心任务，走专用保留通道
  P1_NORMAL = 1,   // 常规业务流程
  P2_LOW = 2       // 弱一致性离线分析，可主动降级
}

// 死信故障分诊三元分类
export enum DLQCategory {
  TRANSIENT_RETRYABLE = 'TRANSIENT_RETRYABLE', // 瞬态网络或过载抖动，可退避重试
  POISON_PAYLOAD = 'POISON_PAYLOAD',           // 毒丸消息（Prompt注入/语法死循环），严禁重试
  SCHEMA_MISMATCH = 'SCHEMA_MISMATCH',         // 格式未对齐，可由修复引擎自愈
  IRREVERSIBLE_PARTIAL = 'IRREVERSIBLE_PARTIAL' // 部分外部写副作用已生效，需人工审查
}

// 协同消息体结构定义
export interface AgentMessage<T = unknown> {
  readonly messageId: string;         // 消息唯一标识
  readonly correlationId: string;     // 分布式全链路追踪 ID
  readonly fromAgent: string;         // 发送方智能体标识
  readonly toAgent: string;           // 接收方智能体标识
  readonly payload: T;                // 业务载荷
  readonly priority: MessagePriority; // 消息优先级
  readonly idempotencyKey: string;    // 业务幂等键
  readonly timestamp: number;         // 发送时间戳
  retryCount: number;                 // 重试次数
}

// 死信快照结构定义
export interface DeadLetterRecord<T = unknown> {
  readonly recordId: string;
  readonly originalMessage: AgentMessage<T>;
  readonly errorStack: string;
  readonly category: DLQCategory;
  readonly contextFingerprint: string;
  readonly quarantinedAt: number;
  readonly resolved: boolean;
}

// 消费者能力与资源状态接口
export interface ConsumerCapacity {
  readonly agentId: string;
  readonly maxConcurrency: number;     // 最大允许并发 GPU 推理槽位
  readonly activeTasks: number;        // 当前活跃任务数
  readonly gpuMemoryUsageRatio: number; // GPU 显存占用率 (0.0 ~ 1.0)
}

/**
 * 响应式背压信道管理器（Credit-Based Flow Controller）
 */
export class ReactiveBackpressureChannel extends EventEmitter {
  private credits: number = 0;
  private readonly buffer: AgentMessage[] = [];
  private readonly maxBufferSize: number = 100;

  constructor(
    public readonly producerId: string,
    public readonly consumerId: string
  ) {
    super();
  }

  // 下游向本信道授权新增信用额度
  public grantCredits(granted: number): void {
    if (granted <= 0) return;
    this.credits += granted;
    this.drainBuffer(); // 触发缓冲区排队消息发送
  }

  // 生产者尝试发送消息
  public send(msg: AgentMessage): boolean {
    // P0 紧急消息不消耗常规 Credit，直接放行
    if (msg.priority === MessagePriority.P0_CRITICAL) {
      this.emit('dispatch', msg);
      return true;
    }

    if (this.credits > 0) {
      this.credits--;
      this.emit('dispatch', msg);
      return true;
    }

    // 触发背压：信用耗尽，进入有界等待队列
    if (this.buffer.length >= this.maxBufferSize) {
      // 队列溢出，执行未决消息降级策略
      if (msg.priority === MessagePriority.P2_LOW) {
        this.emit('dropped_degrade', msg, 'BUFFER_FULL_P2_DROPPED');
        return false;
      }
      throw new Error(`[背压溢出] 目标智能体 ${this.consumerId} 队列已满，拒绝接入`);
    }

    this.buffer.push(msg);
    this.emit('backpressure_engaged', {
      consumerId: this.consumerId,
      bufferedCount: this.buffer.length
    });
    return true;
  }

  // 排空有界缓冲区
  private drainBuffer(): void {
    while (this.credits > 0 && this.buffer.length > 0) {
      const msg = this.buffer.shift()!;
      this.credits--;
      this.emit('dispatch', msg);
    }
  }

  public getAvailableCredits(): number {
    return this.credits;
  }

  public getBufferedCount(): number {
    return this.buffer.length;
  }
}

/**
 * 企业级智能体死信中枢（Agent DLQ Triage Center）
 */
export class AgentDeadLetterCenter extends EventEmitter {
  private readonly dlqStore: Map<string, DeadLetterRecord> = new Map();
  private readonly processedIdempotencyKeys: Set<string> = new Set();

  // 计算全链路上下文指纹哈希
  private calculateFingerprint(msg: AgentMessage, errorReason: string): string {
    const raw = `${msg.idempotencyKey}:${msg.fromAgent}:${msg.toAgent}:${JSON.stringify(msg.payload)}:${errorReason}`;
    return createHash('sha256').update(raw).digest('hex');
  }

  // 记录并隔离死信
  public quarantine(msg: AgentMessage, error: Error, category: DLQCategory): DeadLetterRecord {
    const fingerprint = this.calculateFingerprint(msg, error.message);
    const record: DeadLetterRecord = {
      recordId: `dlq_${Date.now()}_${msg.messageId.slice(0, 8)}`,
      originalMessage: { ...msg },
      errorStack: error.stack || error.message,
      category,
      contextFingerprint: fingerprint,
      quarantinedAt: Date.now(),
      resolved: false
    };

    this.dlqStore.set(record.recordId, record);
    this.emit('quarantined', record);
    return record;
  }

  // 确定性安全重放
  public async replay(
    recordId: string,
    targetDispatcher: (msg: AgentMessage) => Promise<boolean>
  ): Promise<{ success: boolean; reason: string }> {
    const record = this.dlqStore.get(recordId);
    if (!record) {
      return { success: false, reason: '死信记录不存在' };
    }

    if (record.resolved) {
      return { success: false, reason: '该死信已经成功自愈并解决' };
    }

    // 严禁重放明确判定的毒丸消息
    if (record.category === DLQCategory.POISON_PAYLOAD) {
      return { success: false, reason: '毒丸消息被硬性隔离，禁止重放，需人工修正代码或提示词' };
    }

    // 幂等前置守卫检查
    if (this.processedIdempotencyKeys.has(record.originalMessage.idempotencyKey)) {
      record.resolved = true;
      return { success: true, reason: '检测到幂等键已由外部旁路提交，直接标记已解决' };
    }

    // 递增重试计数并执行重发
    const retryMsg: AgentMessage = {
      ...record.originalMessage,
      retryCount: record.originalMessage.retryCount + 1,
      timestamp: Date.now()
    };

    try {
      const dispatched = await targetDispatcher(retryMsg);
      if (dispatched) {
        record.resolved = true;
        this.processedIdempotencyKeys.add(retryMsg.idempotencyKey);
        this.emit('replayed_success', record);
        return { success: true, reason: '重放成功' };
      }
      return { success: false, reason: '重放被下游背压拦截' };
    } catch (err) {
      return { success: false, reason: `重放执行异常: ${(err as Error).message}` };
    }
  }

  public getPendingRecords(): DeadLetterRecord[] {
    return Array.from(this.dlqStore.values()).filter(r => !r.resolved);
  }
}