openclaw/src/plugin-sdk/pair-loop-guard-runtime.ts

/** Resolved pair-loop guard settings in milliseconds for runtime checks. */
export type PairLoopGuardSettings = {
  /** Whether protection is active after config and channel capability gates. */
  enabled: boolean;
  /** Number of pair events allowed before cooldown starts. */
  maxEventsPerWindow: number;
  /** Rolling event window size in milliseconds. */
  windowMs: number;
  /** Suppression duration in milliseconds once the threshold is exceeded. */
  cooldownMs: number;
};

/** User-facing pair-loop guard config accepted by channel plugins. */
export type PairLoopGuardConfig = {
  /** Enables or disables loop protection for the channel/account scope. */
  enabled?: boolean;
  /** Number of pair events allowed before cooldown starts. */
  maxEventsPerWindow?: number;
  /** Rolling event window size in seconds for config files. */
  windowSeconds?: number;
  /** Suppression duration in seconds for config files. */
  cooldownSeconds?: number;
};

const PAIR_LOOP_GUARD_CONFIG_KEYS = [
  "enabled",
  "maxEventsPerWindow",
  "windowSeconds",
  "cooldownSeconds",
] as const satisfies ReadonlyArray<keyof PairLoopGuardConfig>;

/** Result of recording one pair interaction against the loop guard. */
export type PairLoopGuardResult =
  | { suppressed: false }
  | { suppressed: true; cooldownUntilMs: number };

/** Snapshot entry for observability and tests. */
export type PairLoopGuardSnapshotEntry = {
  /** Internal pair key containing scope, conversation, and unordered participant ids. */
  key: string;
  /** Number of retained events in the current window. */
  recentCount: number;
  /** Epoch milliseconds when cooldown ends, or zero when inactive. */
  cooldownUntilMs: number;
};

type PairLoopGuardEntry = {
  recentMs: number[];
  windowMs: number;
  cooldownStartedAtMs: number;
  cooldownUntilMs: number;
};

/** In-memory guard for suppressing repeated bidirectional bot pair loops. */
export type PairLoopGuard = {
  /** Records one sender/receiver interaction and reports whether it enters or is inside cooldown. */
  recordAndCheck: (params: {
    /** Channel/account/provider scope that owns this conversation. */
    scopeId: string;
    /** Conversation/thread identifier where the bidirectional exchange happened. */
    conversationId: string;
    /** Sender id for this event; paired with receiverId without direction. */
    senderId: string;
    /** Receiver id for this event; paired with senderId without direction. */
    receiverId: string;
    /** Resolved guard thresholds for the current channel/account. */
    settings: PairLoopGuardSettings;
    /** Optional test/runtime clock override in epoch milliseconds. */
    nowMs?: number;
  }) => PairLoopGuardResult;
  /** Clears all tracked pair state and scheduled pruning state. */
  clear: () => void;
  /** Returns tracked pair counters for diagnostics and tests without exposing mutable state. */
  snapshot: () => PairLoopGuardSnapshotEntry[];
};

const DEFAULT_PRUNE_INTERVAL_MS = 60_000;
const KEY_SEPARATOR = "\u0001";

/** Default plugin-facing loop guard config before per-channel overrides. */
export const DEFAULT_PAIR_LOOP_GUARD_CONFIG: Required<PairLoopGuardConfig> = {
  enabled: true,
  maxEventsPerWindow: 20,
  windowSeconds: 60,
  cooldownSeconds: 60,
};

/** Default runtime loop guard settings derived from the default config. */
export const DEFAULT_PAIR_LOOP_GUARD_SETTINGS: PairLoopGuardSettings = {
  enabled: DEFAULT_PAIR_LOOP_GUARD_CONFIG.enabled,
  maxEventsPerWindow: DEFAULT_PAIR_LOOP_GUARD_CONFIG.maxEventsPerWindow,
  windowMs: DEFAULT_PAIR_LOOP_GUARD_CONFIG.windowSeconds * 1000,
  cooldownMs: DEFAULT_PAIR_LOOP_GUARD_CONFIG.cooldownSeconds * 1000,
};

/** Merges pair-loop configs from broad defaults to narrow overrides, ignoring undefined values. */
export function mergePairLoopGuardConfig(
  ...configs: Array<PairLoopGuardConfig | undefined>
): PairLoopGuardConfig | undefined {
  const merged: PairLoopGuardConfig = {};
  let hasValue = false;
  for (const config of configs) {
    if (!config) {
      continue;
    }
    for (const key of PAIR_LOOP_GUARD_CONFIG_KEYS) {
      if (config[key] !== undefined) {
        switch (key) {
          case "enabled":
            merged.enabled = config.enabled;
            break;
          case "maxEventsPerWindow":
            merged.maxEventsPerWindow = config.maxEventsPerWindow;
            break;
          case "windowSeconds":
            merged.windowSeconds = config.windowSeconds;
            break;
          case "cooldownSeconds":
            merged.cooldownSeconds = config.cooldownSeconds;
            break;
        }
        hasValue = true;
      }
    }
  }
  return hasValue ? merged : undefined;
}

function positiveInteger(value: unknown): number | undefined {
  return typeof value === "number" && Number.isFinite(value) && value > 0
    ? Math.floor(value)
    : undefined;
}

/** Resolves runtime loop guard settings from config/defaults and the channel default-enabled gate. */
export function resolvePairLoopGuardSettings(params: {
  config?: PairLoopGuardConfig;
  defaultsConfig?: PairLoopGuardConfig;
  defaultEnabled: boolean;
}): PairLoopGuardSettings {
  const configuredEnabled =
    typeof params.config?.enabled === "boolean"
      ? params.config.enabled
      : typeof params.defaultsConfig?.enabled === "boolean"
        ? params.defaultsConfig.enabled
        : DEFAULT_PAIR_LOOP_GUARD_CONFIG.enabled;
  const maxEventsPerWindow =
    positiveInteger(params.config?.maxEventsPerWindow) ??
    positiveInteger(params.defaultsConfig?.maxEventsPerWindow) ??
    DEFAULT_PAIR_LOOP_GUARD_CONFIG.maxEventsPerWindow;
  const windowSeconds =
    positiveInteger(params.config?.windowSeconds) ??
    positiveInteger(params.defaultsConfig?.windowSeconds) ??
    DEFAULT_PAIR_LOOP_GUARD_CONFIG.windowSeconds;
  const cooldownSeconds =
    positiveInteger(params.config?.cooldownSeconds) ??
    positiveInteger(params.defaultsConfig?.cooldownSeconds) ??
    DEFAULT_PAIR_LOOP_GUARD_CONFIG.cooldownSeconds;

  return {
    // Channel-level capability gates can disable protection even when config/defaults enable it.
    enabled: params.defaultEnabled && configuredEnabled,
    maxEventsPerWindow,
    windowMs: windowSeconds * 1000,
    cooldownMs: cooldownSeconds * 1000,
  };
}

function buildPairKey(params: {
  scopeId: string;
  conversationId: string;
  senderId: string;
  receiverId: string;
}): string {
  // Sort sender/receiver so A->B and B->A count as the same bot loop pair.
  const lhs = params.senderId < params.receiverId ? params.senderId : params.receiverId;
  const rhs = params.senderId < params.receiverId ? params.receiverId : params.senderId;
  return [params.scopeId, params.conversationId, lhs, rhs].join(KEY_SEPARATOR);
}

function pruneRecentTimestamps(entry: PairLoopGuardEntry, nowMs: number, windowMs: number): void {
  const cutoff = nowMs - windowMs;
  entry.recentMs = entry.recentMs.filter((timestampMs) => timestampMs > cutoff);
}

function countCurrentWindowEvents(entry: PairLoopGuardEntry, nowMs: number): number {
  return entry.recentMs.filter((timestampMs) => timestampMs <= nowMs).length;
}

/** Creates an in-memory pair-loop guard with bounded periodic pruning. */
export function createPairLoopGuard(params?: { pruneIntervalMs?: number }): PairLoopGuard {
  const tracked = new Map<string, PairLoopGuardEntry>();
  const pruneIntervalMs = params?.pruneIntervalMs ?? DEFAULT_PRUNE_INTERVAL_MS;
  let nextPruneAtMs = 0;

  function pruneInactiveTrackedPairs(nowMs: number): void {
    if (pruneIntervalMs <= 0 || nowMs < nextPruneAtMs) {
      return;
    }
    nextPruneAtMs = nowMs + pruneIntervalMs;
    for (const [key, entry] of tracked) {
      pruneRecentTimestamps(entry, nowMs, entry.windowMs);
      if (entry.recentMs.length === 0 && entry.cooldownUntilMs <= nowMs) {
        tracked.delete(key);
      }
    }
  }

  function recordAndCheck(paramsLocal: {
    scopeId: string;
    conversationId: string;
    senderId: string;
    receiverId: string;
    settings: PairLoopGuardSettings;
    nowMs?: number;
  }): PairLoopGuardResult {
    if (!paramsLocal.settings.enabled) {
      return { suppressed: false };
    }
    if (
      !paramsLocal.scopeId ||
      !paramsLocal.conversationId ||
      !paramsLocal.senderId ||
      !paramsLocal.receiverId
    ) {
      return { suppressed: false };
    }
    if (paramsLocal.senderId === paramsLocal.receiverId) {
      return { suppressed: false };
    }

    const maxEventsPerWindow = Math.floor(paramsLocal.settings.maxEventsPerWindow);
    const windowMs = Math.floor(paramsLocal.settings.windowMs);
    const cooldownMs = Math.floor(paramsLocal.settings.cooldownMs);
    if (maxEventsPerWindow <= 0 || windowMs <= 0 || cooldownMs <= 0) {
      return { suppressed: false };
    }

    const nowMs = paramsLocal.nowMs ?? Date.now();
    pruneInactiveTrackedPairs(nowMs);

    const key = buildPairKey(paramsLocal);
    let entry = tracked.get(key);
    if (!entry) {
      entry = { recentMs: [], windowMs, cooldownStartedAtMs: 0, cooldownUntilMs: 0 };
      tracked.set(key, entry);
    }
    if (entry.cooldownStartedAtMs <= nowMs && entry.cooldownUntilMs > nowMs) {
      return { suppressed: true, cooldownUntilMs: entry.cooldownUntilMs };
    }

    entry.windowMs = windowMs;
    pruneRecentTimestamps(entry, nowMs, windowMs);
    entry.recentMs.push(nowMs);
    if (countCurrentWindowEvents(entry, nowMs) > maxEventsPerWindow) {
      entry.cooldownStartedAtMs = nowMs;
      entry.cooldownUntilMs = nowMs + cooldownMs;
      // Keep only future records during cooldown; past events should not extend suppression.
      entry.recentMs = entry.recentMs.filter((timestampMs) => timestampMs > nowMs);
      return { suppressed: true, cooldownUntilMs: entry.cooldownUntilMs };
    }

    return { suppressed: false };
  }

  return {
    recordAndCheck,
    clear: () => {
      tracked.clear();
      nextPruneAtMs = 0;
    },
    snapshot: () =>
      Array.from(tracked.entries()).map(([key, entry]) => ({
        key,
        recentCount: entry.recentMs.length,
        cooldownUntilMs: entry.cooldownUntilMs,
      })),
  };
}