src/logging/diagnostic-memory.ts

import { totalmem } from "node:os";
// Diagnostic memory helpers capture process memory facts for support diagnostics.
import { getHeapStatistics } from "node:v8";
import {
  emitInternalDiagnosticEvent as emitDiagnosticEvent,
  type DiagnosticMemoryPressureEvent,
  type DiagnosticMemoryUsage,
} from "../infra/diagnostic-events.js";
import { createSubsystemLogger } from "./subsystem.js";

// Diagnostic memory sampler with threshold/growth pressure detection and repeat suppression.
const MB = 1024 * 1024;
const GB = 1024 * MB;
const DEFAULT_RSS_WARNING_BYTES = 1536 * MB;
const DEFAULT_RSS_CRITICAL_BYTES = 3072 * MB;
const DEFAULT_HEAP_WARNING_BYTES = 1024 * MB;
const DEFAULT_HEAP_CRITICAL_BYTES = 2048 * MB;
const DEFAULT_HEAP_WARNING_RATIO = 0.5;
const DEFAULT_HEAP_CRITICAL_RATIO = 0.75;
const BUN_HEAP_WARNING_MAX_BYTES = 4 * GB;
const BUN_HEAP_CRITICAL_MAX_BYTES = 6 * GB;
const DEFAULT_RSS_GROWTH_WARNING_BYTES = 512 * MB;
const DEFAULT_RSS_GROWTH_CRITICAL_BYTES = 1024 * MB;
const DEFAULT_GROWTH_WINDOW_MS = 10 * 60 * 1000;
const DEFAULT_PRESSURE_REPEAT_MS = 5 * 60 * 1000;
const BYTE_UNITS = ["B", "KiB", "MiB", "GiB", "TiB"] as const;

const DEFAULT_HEAP_SIZE_LIMIT_BYTES = getHeapStatistics().heap_size_limit;
const DEFAULT_PROCESS_MEMORY_LIMIT_BYTES = process.constrainedMemory();
const DEFAULT_PHYSICAL_MEMORY_BYTES = totalmem();
const DEFAULT_IS_BUN_RUNTIME = typeof process.versions.bun === "string";

const log = createSubsystemLogger("gateway").child("diagnostics/memory");

type DiagnosticMemoryThresholds = {
  rssWarningBytes?: number;
  rssCriticalBytes?: number;
  heapUsedWarningBytes?: number;
  heapUsedCriticalBytes?: number;
  rssGrowthWarningBytes?: number;
  rssGrowthCriticalBytes?: number;
  growthWindowMs?: number;
  pressureRepeatMs?: number;
};

type DiagnosticMemorySample = {
  ts: number;
  memory: DiagnosticMemoryUsage;
};

type DiagnosticMemoryState = {
  lastSample: DiagnosticMemorySample | null;
  lastPressureAtByKey: Map<string, number>;
};

function isPositiveMemoryLimit(value: number | undefined): value is number {
  return typeof value === "number" && Number.isFinite(value) && value > 0;
}

function resolveProcessMemoryLimitBytes(
  processMemoryLimitBytes: number | undefined,
  physicalMemoryBytes: number | undefined,
  isBunRuntime: boolean,
): number | undefined {
  if (!isPositiveMemoryLimit(processMemoryLimitBytes)) {
    // Node can report no constraint even when an explicit heap exceeds physical RAM.
    // Keep Bun's existing no-constraint RSS policy independent of its compatibility heap.
    return !isBunRuntime && isPositiveMemoryLimit(physicalMemoryBytes)
      ? physicalMemoryBytes
      : undefined;
  }
  return isPositiveMemoryLimit(physicalMemoryBytes)
    ? Math.min(processMemoryLimitBytes, physicalMemoryBytes)
    : processMemoryLimitBytes;
}

const state: DiagnosticMemoryState = {
  lastSample: null,
  lastPressureAtByKey: new Map(),
};

// Convert Node's runtime shape into the diagnostic event contract.
function normalizeMemoryUsage(memory: NodeJS.MemoryUsage): DiagnosticMemoryUsage {
  return {
    rssBytes: memory.rss,
    heapTotalBytes: memory.heapTotal,
    heapUsedBytes: memory.heapUsed,
    externalBytes: memory.external,
    arrayBuffersBytes: memory.arrayBuffers,
  };
}

function resolveThresholds(
  thresholds?: DiagnosticMemoryThresholds,
  heapSizeLimitBytes?: number,
  processMemoryLimitBytes?: number,
  physicalMemoryBytes?: number,
  isBunRuntime = false,
): Required<DiagnosticMemoryThresholds> {
  const hasHeapLimit = isPositiveMemoryLimit(heapSizeLimitBytes);
  // Node pressure follows the measured V8 limit, including explicit larger heaps.
  // Bun's node:v8 compatibility metadata retains its existing caps.
  const heapWarningBytes = hasHeapLimit
    ? Math.min(
        Math.floor(heapSizeLimitBytes * DEFAULT_HEAP_WARNING_RATIO),
        isBunRuntime ? BUN_HEAP_WARNING_MAX_BYTES : Infinity,
      )
    : DEFAULT_HEAP_WARNING_BYTES;
  const heapCriticalBytes = hasHeapLimit
    ? Math.min(
        Math.floor(heapSizeLimitBytes * DEFAULT_HEAP_CRITICAL_RATIO),
        isBunRuntime ? BUN_HEAP_CRITICAL_MAX_BYTES : Infinity,
      )
    : DEFAULT_HEAP_CRITICAL_BYTES;
  const usableProcessMemoryLimitBytes = resolveProcessMemoryLimitBytes(
    processMemoryLimitBytes,
    physicalMemoryBytes,
    isBunRuntime,
  );
  const hasProcessMemoryLimit = usableProcessMemoryLimitBytes !== undefined;
  // Bun's node:v8 heap limit is compatibility metadata, not an RSS process budget.
  const useBunRssCaps = isBunRuntime && hasProcessMemoryLimit;
  const useHeapForRss = !isBunRuntime && hasHeapLimit;
  const rssWarningBase = useBunRssCaps
    ? BUN_HEAP_WARNING_MAX_BYTES
    : useHeapForRss
      ? Math.max(DEFAULT_RSS_WARNING_BYTES, heapWarningBytes)
      : DEFAULT_RSS_WARNING_BYTES;
  const rssCriticalBase = useBunRssCaps
    ? BUN_HEAP_CRITICAL_MAX_BYTES
    : useHeapForRss
      ? Math.max(DEFAULT_RSS_CRITICAL_BYTES, heapCriticalBytes)
      : DEFAULT_RSS_CRITICAL_BYTES;
  const processWarningBytes = hasProcessMemoryLimit
    ? Math.floor(usableProcessMemoryLimitBytes * DEFAULT_HEAP_WARNING_RATIO)
    : rssWarningBase;
  const processCriticalBytes = hasProcessMemoryLimit
    ? Math.floor(usableProcessMemoryLimitBytes * DEFAULT_HEAP_CRITICAL_RATIO)
    : rssCriticalBase;
  return {
    rssWarningBytes: thresholds?.rssWarningBytes ?? Math.min(rssWarningBase, processWarningBytes),
    rssCriticalBytes:
      thresholds?.rssCriticalBytes ?? Math.min(rssCriticalBase, processCriticalBytes),
    heapUsedWarningBytes: thresholds?.heapUsedWarningBytes ?? heapWarningBytes,
    heapUsedCriticalBytes: thresholds?.heapUsedCriticalBytes ?? heapCriticalBytes,
    rssGrowthWarningBytes: thresholds?.rssGrowthWarningBytes ?? DEFAULT_RSS_GROWTH_WARNING_BYTES,
    rssGrowthCriticalBytes: thresholds?.rssGrowthCriticalBytes ?? DEFAULT_RSS_GROWTH_CRITICAL_BYTES,
    growthWindowMs: thresholds?.growthWindowMs ?? DEFAULT_GROWTH_WINDOW_MS,
    pressureRepeatMs: thresholds?.pressureRepeatMs ?? DEFAULT_PRESSURE_REPEAT_MS,
  };
}

function pickThresholdPressure(params: {
  memory: DiagnosticMemoryUsage;
  thresholds: Required<DiagnosticMemoryThresholds>;
}): Omit<DiagnosticMemoryPressureEvent, "seq" | "ts" | "type"> | null {
  const { memory, thresholds } = params;
  if (memory.rssBytes >= thresholds.rssCriticalBytes) {
    return {
      level: "critical",
      reason: "rss_threshold",
      memory,
      thresholdBytes: thresholds.rssCriticalBytes,
    };
  }
  if (memory.heapUsedBytes >= thresholds.heapUsedCriticalBytes) {
    return {
      level: "critical",
      reason: "heap_threshold",
      memory,
      thresholdBytes: thresholds.heapUsedCriticalBytes,
    };
  }
  if (memory.rssBytes >= thresholds.rssWarningBytes) {
    return {
      level: "warning",
      reason: "rss_threshold",
      memory,
      thresholdBytes: thresholds.rssWarningBytes,
    };
  }
  if (memory.heapUsedBytes >= thresholds.heapUsedWarningBytes) {
    return {
      level: "warning",
      reason: "heap_threshold",
      memory,
      thresholdBytes: thresholds.heapUsedWarningBytes,
    };
  }
  return null;
}

function pickGrowthPressure(params: {
  previous: DiagnosticMemorySample | null;
  current: DiagnosticMemorySample;
  thresholds: Required<DiagnosticMemoryThresholds>;
}): Omit<DiagnosticMemoryPressureEvent, "seq" | "ts" | "type"> | null {
  const { previous, current, thresholds } = params;
  if (!previous) {
    return null;
  }
  const windowMs = current.ts - previous.ts;
  if (windowMs <= 0 || windowMs > thresholds.growthWindowMs) {
    return null;
  }
  const rssGrowthBytes = current.memory.rssBytes - previous.memory.rssBytes;
  if (rssGrowthBytes >= thresholds.rssGrowthCriticalBytes) {
    return {
      level: "critical",
      reason: "rss_growth",
      memory: current.memory,
      thresholdBytes: thresholds.rssGrowthCriticalBytes,
      rssGrowthBytes,
      windowMs,
    };
  }
  if (rssGrowthBytes >= thresholds.rssGrowthWarningBytes) {
    return {
      level: "warning",
      reason: "rss_growth",
      memory: current.memory,
      thresholdBytes: thresholds.rssGrowthWarningBytes,
      rssGrowthBytes,
      windowMs,
    };
  }
  return null;
}

function shouldEmitPressure(
  pressure: Omit<DiagnosticMemoryPressureEvent, "seq" | "ts" | "type">,
  now: number,
  repeatMs: number,
): boolean {
  const key = `${pressure.level}:${pressure.reason}`;
  const lastAt = state.lastPressureAtByKey.get(key);
  // Pressure events can repeat during sustained memory spikes; throttle per level/reason pair.
  if (lastAt !== undefined && now - lastAt < repeatMs) {
    return false;
  }
  state.lastPressureAtByKey.set(key, now);
  return true;
}

function formatOptionalPressureMetric(label: string, value: number | undefined): string {
  return typeof value === "number" && Number.isFinite(value) ? ` ${label}=${value}` : "";
}

function formatScaledNumber(value: number): string {
  const fixed = value >= 10 ? value.toFixed(1) : value.toFixed(2);
  return fixed.replace(/\.0+$/u, "").replace(/(\.\d*[1-9])0$/u, "$1");
}

function formatReadableBytes(value: number | undefined): string | undefined {
  if (typeof value !== "number" || !Number.isFinite(value) || value < 0) {
    return undefined;
  }
  let scaled = value;
  let unitIndex = 0;
  while (scaled >= 1024 && unitIndex < BYTE_UNITS.length - 1) {
    scaled /= 1024;
    unitIndex++;
  }
  return unitIndex === 0
    ? `${Math.round(scaled)} ${BYTE_UNITS[unitIndex]}`
    : `${formatScaledNumber(scaled)} ${BYTE_UNITS[unitIndex]}`;
}

function formatPressureRatio(params: {
  pressure: Omit<DiagnosticMemoryPressureEvent, "seq" | "ts" | "type">;
  thresholdBytes: number;
}): string | undefined {
  const { pressure, thresholdBytes } = params;
  if (!Number.isFinite(thresholdBytes) || thresholdBytes <= 0) {
    return undefined;
  }
  const value =
    pressure.reason === "heap_threshold"
      ? pressure.memory.heapUsedBytes
      : pressure.reason === "rss_growth"
        ? pressure.rssGrowthBytes
        : pressure.memory.rssBytes;
  if (typeof value !== "number" || !Number.isFinite(value)) {
    return undefined;
  }
  const ratio = (value / thresholdBytes) * 100;
  return `${formatScaledNumber(ratio)}%`;
}

function formatPressureSummary(
  pressure: Omit<DiagnosticMemoryPressureEvent, "seq" | "ts" | "type">,
): string {
  const parts = [
    `rss=${formatReadableBytes(pressure.memory.rssBytes)}`,
    `heap=${formatReadableBytes(pressure.memory.heapUsedBytes)}`,
    pressure.thresholdBytes !== undefined
      ? `threshold=${formatReadableBytes(pressure.thresholdBytes)}`
      : "",
    pressure.thresholdBytes !== undefined
      ? `thresholdRatio=${formatPressureRatio({
          pressure,
          thresholdBytes: pressure.thresholdBytes,
        })}`
      : "",
    pressure.rssGrowthBytes !== undefined
      ? `rssGrowth=${formatReadableBytes(pressure.rssGrowthBytes)}`
      : "",
  ];
  return parts.filter((part): part is string => Boolean(part)).join(" ");
}

function formatPressureNextStep(
  pressure: Omit<DiagnosticMemoryPressureEvent, "seq" | "ts" | "type">,
): string {
  return pressure.level === "critical"
    ? "nextStep=inspect latest stability bundle or run openclaw gateway diagnostics export; restart gateway if process is unstable"
    : "nextStep=run openclaw gateway status --deep and openclaw gateway diagnostics export; restart gateway if pressure persists";
}

function logMemoryPressure(
  pressure: Omit<DiagnosticMemoryPressureEvent, "seq" | "ts" | "type">,
): void {
  const message =
    `memory pressure: level=${pressure.level} reason=${pressure.reason}` +
    ` ${formatPressureSummary(pressure)}` +
    ` rssBytes=${pressure.memory.rssBytes}` +
    ` heapUsedBytes=${pressure.memory.heapUsedBytes}` +
    formatOptionalPressureMetric("thresholdBytes", pressure.thresholdBytes) +
    formatOptionalPressureMetric("rssGrowthBytes", pressure.rssGrowthBytes) +
    formatOptionalPressureMetric("windowMs", pressure.windowMs) +
    (pressure.level === "critical" ? " memoryPressureSnapshot=disabled" : "") +
    ` ${formatPressureNextStep(pressure)}`;
  log.warn(message);
}

export function emitDiagnosticMemorySample(options?: {
  now?: number;
  memoryUsage?: NodeJS.MemoryUsage;
  heapSizeLimitBytes?: number;
  processMemoryLimitBytes?: number;
  physicalMemoryBytes?: number;
  isBunRuntime?: boolean;
  uptimeMs?: number;
  thresholds?: DiagnosticMemoryThresholds;
  emitSample?: boolean;
}): DiagnosticMemoryUsage {
  const now = options?.now ?? Date.now();
  const memory = normalizeMemoryUsage(options?.memoryUsage ?? process.memoryUsage());
  const current = { ts: now, memory };
  const thresholds = resolveThresholds(
    options?.thresholds,
    options?.heapSizeLimitBytes ?? DEFAULT_HEAP_SIZE_LIMIT_BYTES,
    options?.processMemoryLimitBytes ?? DEFAULT_PROCESS_MEMORY_LIMIT_BYTES,
    options?.physicalMemoryBytes ?? DEFAULT_PHYSICAL_MEMORY_BYTES,
    options?.isBunRuntime ?? DEFAULT_IS_BUN_RUNTIME,
  );
  const shouldEmitSample = options?.emitSample !== false;

  if (shouldEmitSample) {
    emitDiagnosticEvent({
      type: "diagnostic.memory.sample",
      memory,
      uptimeMs: options?.uptimeMs ?? Math.round(process.uptime() * 1000),
    });
  }

  const pressure =
    pickThresholdPressure({ memory, thresholds }) ??
    pickGrowthPressure({ previous: state.lastSample, current, thresholds });
  state.lastSample = current;
  if (pressure && shouldEmitPressure(pressure, now, thresholds.pressureRepeatMs)) {
    emitDiagnosticEvent({
      type: "diagnostic.memory.pressure",
      ...pressure,
    });
    logMemoryPressure(pressure);
    if (pressure.level === "critical") {
      log.warn("critical memory pressure snapshot disabled");
    }
  }
  return memory;
}

/** Clears process-local memory diagnostic state for isolated tests. */
export function resetDiagnosticMemoryForTest(): void {
  state.lastSample = null;
  state.lastPressureAtByKey.clear();
}