memind/memory-v2-runtime.mjs

import path from 'node:path';
import { applyGooseCanaryMemoryPolicy } from './goose-canary-memory-policy.mjs';
import { pathToFileURL } from 'node:url';
import { createLangGraphHttpClient, createLangGraphMemoryBackend } from './memory-v2-langgraph.mjs';
import { createLegacyMemoryBackend, createMemoryV2, resolveMemoryV2Policy } from './memory-v2.mjs';
import { createLettaHttpClient, createLettaMemoryBackend } from './memory-v2-letta.mjs';
import { createMemoryV2PluginBackends } from './memory-v2-plugin-backends.mjs';
import { createPersonalMemoryShadowPipeline } from './memory-v2-personal-shadow.mjs';
import {
  createPersonalMemoryCandidateStore,
  ensurePersonalMemoryCandidateSchema,
} from './memory-v2-personal-store.mjs';
import {
  autoReviewCandidateIfPending,
  runCandidateAutoReviewBatch,
} from './memory-v2-candidate-auto-review.mjs';
import {
  MEMORY_V2_PRODUCT_EVENT_TYPES,
  ensureMemoryV2ProductEventsSchema,
  recordMemoryV2ProductEvent,
} from './memory-v2-product-events.mjs';
import { createMem0HttpClient, createMem0MemoryBackend } from './memory-v2-mem0.mjs';
import { createNeo4jHttpClient, createNeo4jMemoryBackend } from './memory-v2-neo4j.mjs';
import { createPgvectorMemoryBackend } from './memory-v2-pgvector.mjs';
import {
  backfillLegacyMemoriesToPgvector,
  syncLegacyUserMemoriesToPgvector,
} from './memory-v2-pgvector-backfill.mjs';
import { createQdrantHttpClient, createQdrantMemoryBackend } from './memory-v2-qdrant.mjs';
import { createRedisStreamsClient, createRedisStreamsMemoryBackend } from './memory-v2-redis-streams.mjs';
import { createWeaviateHttpClient, createWeaviateMemoryBackend } from './memory-v2-weaviate.mjs';
import {
  createMemoryV2LifecycleService,
  resolveMemoryV2LifecycleWorkerScopes,
} from './memory-v2-lifecycle.mjs';

const DEFAULT_PGVECTOR_URL_ENV = 'MEMORY_PGVECTOR_DATABASE_URL';

function readFlag(env, key, fallback = false) {
  const raw = env?.[key];
  if (raw == null || raw === '') return fallback;
  const normalized = String(raw).trim().toLowerCase();
  if (['1', 'true', 'yes', 'on'].includes(normalized)) return true;
  if (['0', 'false', 'no', 'off'].includes(normalized)) return false;
  return fallback;
}

function resolveModuleSpecifier(specifier) {
  const value = String(specifier ?? '').trim();
  if (!value) return null;
  if (value.startsWith('.') || value.startsWith('/')) {
    return pathToFileURL(path.resolve(value)).href;
  }
  return value;
}

async function loadEmbeddingModule(moduleSpecifier, importModule) {
  const resolved = resolveModuleSpecifier(moduleSpecifier);
  if (!resolved) return null;
  const imported = await importModule(resolved);
  const fn = imported?.embedQuery ?? imported?.default;
  if (typeof fn !== 'function') {
    throw new Error(
      `Memory V2 pgvector embedding module must export embedQuery or default: ${moduleSpecifier}`,
    );
  }
  return fn;
}

async function createPgPool(connectionString, { importPg, max }) {
  const imported = await importPg('pg');
  const PgPool = imported?.Pool ?? imported?.default?.Pool;
  if (typeof PgPool !== 'function') {
    throw new Error('pg module does not export Pool');
  }
  return new PgPool({ connectionString, max });
}

export async function createMemoryV2Runtime({
  legacyMemoryService = null,
  mysqlPool = null,
  env = process.env,
  logger = console,
  importPg = (specifier) => import(specifier),
  importRedis = (specifier) => import(specifier),
  importModule = (specifier) => import(specifier),
  fetchImpl = globalThis.fetch,
  lettaClient = null,
  weaviateClient = null,
  neo4jClient = null,
  redisStreamsClient = null,
  langGraphClient = null,
} = {}) {
  const vectorEnabled = readFlag(env, 'MEMORY_VECTOR_ENABLED', false);
  const pgvectorEnabled = readFlag(env, 'MEMORY_PGVECTOR_ENABLED', vectorEnabled);
  const connectionString = String(env?.[DEFAULT_PGVECTOR_URL_ENV] ?? '').trim();
  const embeddingModule = String(env?.MEMORY_PGVECTOR_EMBEDDING_MODULE ?? '').trim();
  const backends = [createLegacyMemoryBackend(legacyMemoryService)];
  const closers = [];

  let pgvectorBackend = null;
  let pgPoolRef = null;
  let pgEmbedQuery = null;
  if (pgvectorEnabled && connectionString) {
    try {
      const embedQuery = await loadEmbeddingModule(embeddingModule, importModule);
      if (!embedQuery) {
        pgvectorBackend = createPgvectorMemoryBackend({
          enabled: false,
          tableName: env?.MEMORY_PGVECTOR_TABLE,
          unavailableReason: 'embedding_module_not_configured',
        });
      } else {
        const pool = await createPgPool(connectionString, {
          importPg,
          max: Math.max(1, Number(env?.MEMORY_PGVECTOR_POOL_MAX ?? 5) || 5),
        });
        closers.push(() => pool.end?.());
        pgPoolRef = pool;
        pgEmbedQuery = embedQuery;
        pgvectorBackend = createPgvectorMemoryBackend({
          pool,
          enabled: pgvectorEnabled,
          tableName: env?.MEMORY_PGVECTOR_TABLE,
          embedQuery,
          defaultLimit: Number(env?.MEMORY_PGVECTOR_LIMIT ?? 8) || 8,
        });
      }
    } catch (err) {
      logger?.warn?.(
        `[memory-v2] pgvector backend unavailable: ${err instanceof Error ? err.message : err}`,
      );
      pgvectorBackend = createPgvectorMemoryBackend({
        enabled: false,
        unavailableReason: 'init_failed',
      });
    }
  }

  if (pgvectorBackend) backends.push(pgvectorBackend);
  let qdrantBackend = null;
  const qdrantEnabled = readFlag(env, 'MEMORY_QDRANT_ENABLED', false);
  const qdrantUrl = String(env?.MEMORY_QDRANT_URL ?? '').trim();
  const qdrantCollection = String(env?.MEMORY_QDRANT_COLLECTION ?? '').trim();
  const qdrantEmbeddingModule = String(env?.MEMORY_QDRANT_EMBEDDING_MODULE ?? '').trim();
  if (qdrantEnabled || qdrantUrl || qdrantCollection) {
    try {
      const qdrantEmbedQuery = await loadEmbeddingModule(qdrantEmbeddingModule, importModule);
      const qdrantClient = qdrantEnabled && qdrantUrl && qdrantEmbedQuery
        ? createQdrantHttpClient({
            url: qdrantUrl,
            apiKey: env?.MEMORY_QDRANT_API_KEY,
            fetchImpl,
            timeoutMs: Number(env?.MEMORY_QDRANT_TIMEOUT_MS ?? 3000) || 3000,
          })
        : null;
      qdrantBackend = createQdrantMemoryBackend({
        enabled: qdrantEnabled,
        url: qdrantUrl,
        collection: qdrantCollection || undefined,
        client: qdrantClient,
        embedQuery: qdrantEmbedQuery,
        defaultLimit: Number(env?.MEMORY_QDRANT_LIMIT ?? 8) || 8,
      });
    } catch (err) {
      logger?.warn?.(
        `[memory-v2] qdrant backend unavailable: ${err instanceof Error ? err.message : err}`,
      );
      qdrantBackend = createQdrantMemoryBackend({
        enabled: false,
        unavailableReason: 'init_failed',
      });
    }
  }
  if (qdrantBackend) backends.push(qdrantBackend);
  let weaviateBackend = null;
  const weaviateEnabled = readFlag(env, 'MEMORY_WEAVIATE_ENABLED', false);
  const weaviateUrl = String(env?.MEMORY_WEAVIATE_URL ?? '').trim();
  const weaviateCollection = String(env?.MEMORY_WEAVIATE_COLLECTION ?? '').trim();
  const weaviateEmbeddingModule = String(env?.MEMORY_WEAVIATE_EMBEDDING_MODULE ?? '').trim();
  if (weaviateEnabled || weaviateUrl || weaviateCollection) {
    try {
      const weaviateEmbedQuery = await loadEmbeddingModule(weaviateEmbeddingModule, importModule);
      const resolvedWeaviateClient = weaviateClient ?? (
        weaviateEnabled && weaviateUrl && weaviateEmbedQuery
          ? createWeaviateHttpClient({
              url: weaviateUrl,
              apiKey: env?.MEMORY_WEAVIATE_API_KEY,
              fetchImpl,
              timeoutMs: Number(env?.MEMORY_WEAVIATE_TIMEOUT_MS ?? 3000) || 3000,
            })
          : null
      );
      weaviateBackend = createWeaviateMemoryBackend({
        enabled: weaviateEnabled,
        url: weaviateUrl,
        collection: weaviateCollection || undefined,
        client: resolvedWeaviateClient,
        embedQuery: weaviateEmbedQuery,
        defaultLimit: Number(env?.MEMORY_WEAVIATE_LIMIT ?? 8) || 8,
      });
    } catch (err) {
      logger?.warn?.(
        `[memory-v2] weaviate backend unavailable: ${err instanceof Error ? err.message : err}`,
      );
      weaviateBackend = createWeaviateMemoryBackend({
        enabled: false,
        unavailableReason: 'init_failed',
      });
    }
  }
  if (weaviateBackend) backends.push(weaviateBackend);
  let mem0Backend = null;
  const mem0Enabled = readFlag(env, 'MEMORY_MEM0_ENABLED', false);
  const mem0ApiKey = String(env?.MEMORY_MEM0_API_KEY ?? '').trim();
  const mem0ProjectId = String(env?.MEMORY_MEM0_PROJECT_ID ?? '').trim();
  const mem0BaseUrl = String(env?.MEMORY_MEM0_BASE_URL ?? 'https://api.mem0.ai').trim();
  const mem0ModelProviderKeyId = String(env?.MEMORY_MEM0_MODEL_PROVIDER_KEY_ID ?? '').trim();
  const mem0Model = String(env?.MEMORY_MEM0_MODEL ?? '').trim();
  const mem0ModelApiType = String(env?.MEMORY_MEM0_MODEL_API ?? '').trim();
  if (mem0Enabled || mem0ApiKey || mem0ProjectId) {
    try {
      const mem0Client = mem0Enabled && mem0ApiKey
        ? createMem0HttpClient({
            apiKey: mem0ApiKey,
            projectId: mem0ProjectId || undefined,
            baseUrl: mem0BaseUrl,
            writePath: env?.MEMORY_MEM0_WRITE_PATH || undefined,
            compactPath: env?.MEMORY_MEM0_COMPACT_PATH || undefined,
            modelProviderKeyId: mem0ModelProviderKeyId || undefined,
            model: mem0Model || undefined,
            modelApiType: mem0ModelApiType || undefined,
            fetchImpl,
            timeoutMs: Number(env?.MEMORY_MEM0_TIMEOUT_MS ?? 3000) || 3000,
          })
        : null;
      mem0Backend = createMem0MemoryBackend({
        enabled: mem0Enabled,
        apiKey: mem0ApiKey,
        projectId: mem0ProjectId || undefined,
        client: mem0Client,
      });
    } catch (err) {
      logger?.warn?.(
        `[memory-v2] mem0 backend unavailable: ${err instanceof Error ? err.message : err}`,
      );
      mem0Backend = createMem0MemoryBackend({
        enabled: false,
        unavailableReason: 'init_failed',
      });
    }
  }
  if (mem0Backend) backends.push(mem0Backend);
  let lettaBackend = null;
  const lettaEnabled = readFlag(env, 'MEMORY_LETTA_ENABLED', false);
  const lettaApiKey = String(env?.MEMORY_LETTA_API_KEY ?? '').trim();
  const lettaProjectId = String(env?.MEMORY_LETTA_PROJECT_ID ?? '').trim();
  const lettaAgentId = String(env?.MEMORY_LETTA_AGENT_ID ?? '').trim();
  const lettaBaseUrl = String(env?.MEMORY_LETTA_BASE_URL ?? 'https://api.letta.com').trim();
  const lettaModelProviderKeyId = String(env?.MEMORY_LETTA_MODEL_PROVIDER_KEY_ID ?? '').trim();
  const lettaModel = String(env?.MEMORY_LETTA_MODEL ?? '').trim();
  const lettaModelApiType = String(env?.MEMORY_LETTA_MODEL_API ?? '').trim();
  if (lettaEnabled || lettaApiKey || lettaProjectId || lettaAgentId) {
    try {
      const resolvedLettaClient = lettaClient ?? (
        lettaEnabled && lettaApiKey && lettaAgentId
          ? createLettaHttpClient({
              apiKey: lettaApiKey,
              baseUrl: lettaBaseUrl,
              agentId: lettaAgentId,
              projectId: lettaProjectId || undefined,
              resolvePath: env?.MEMORY_LETTA_RESOLVE_PATH || undefined,
              writePath: env?.MEMORY_LETTA_WRITE_PATH || undefined,
              compactPath: env?.MEMORY_LETTA_COMPACT_PATH || undefined,
              modelProviderKeyId: lettaModelProviderKeyId || undefined,
              model: lettaModel || undefined,
              modelApiType: lettaModelApiType || undefined,
              fetchImpl,
              timeoutMs: Number(env?.MEMORY_LETTA_TIMEOUT_MS ?? 3000) || 3000,
            })
          : null
      );
      lettaBackend = createLettaMemoryBackend({
        enabled: lettaEnabled,
        apiKey: lettaApiKey,
        projectId: lettaProjectId || undefined,
        agentId: lettaAgentId || undefined,
        client: resolvedLettaClient,
      });
    } catch (err) {
      logger?.warn?.(
        `[memory-v2] letta backend unavailable: ${err instanceof Error ? err.message : err}`,
      );
      lettaBackend = createLettaMemoryBackend({
        enabled: false,
        unavailableReason: 'init_failed',
      });
    }
  }
  if (lettaBackend) backends.push(lettaBackend);
  let neo4jBackend = null;
  const neo4jEnabled = readFlag(env, 'MEMORY_NEO4J_ENABLED', false);
  const neo4jUri = String(env?.MEMORY_NEO4J_URI ?? '').trim();
  const neo4jHttpUrl = String(env?.MEMORY_NEO4J_HTTP_URL ?? '').trim();
  const neo4jUsername = String(env?.MEMORY_NEO4J_USER ?? '').trim();
  const neo4jPassword = String(env?.MEMORY_NEO4J_PASSWORD ?? '').trim();
  if (neo4jEnabled || neo4jUri || neo4jHttpUrl || neo4jUsername || neo4jPassword) {
    try {
      const resolvedNeo4jClient = neo4jClient ?? (
        neo4jEnabled && neo4jHttpUrl && neo4jUsername && neo4jPassword
          ? createNeo4jHttpClient({
              httpUrl: neo4jHttpUrl,
              username: neo4jUsername,
              password: neo4jPassword,
              database: env?.MEMORY_NEO4J_DATABASE || 'neo4j',
              fetchImpl,
              timeoutMs: Number(env?.MEMORY_NEO4J_TIMEOUT_MS ?? 3000) || 3000,
            })
          : null
      );
      neo4jBackend = createNeo4jMemoryBackend({
        enabled: neo4jEnabled,
        uri: neo4jUri,
        httpUrl: neo4jHttpUrl,
        username: neo4jUsername,
        password: neo4jPassword,
        client: resolvedNeo4jClient,
      });
    } catch (err) {
      logger?.warn?.(
        `[memory-v2] neo4j backend unavailable: ${err instanceof Error ? err.message : err}`,
      );
      neo4jBackend = createNeo4jMemoryBackend({
        enabled: false,
        unavailableReason: 'init_failed',
      });
    }
  }
  if (neo4jBackend) backends.push(neo4jBackend);
  let redisStreamsBackend = null;
  const redisStreamsEnabled = readFlag(env, 'MEMORY_REDIS_STREAMS_ENABLED', false);
  const redisStreamsUrl = String(env?.MEMORY_REDIS_STREAMS_URL ?? '').trim();
  const redisStreamsStream = String(env?.MEMORY_REDIS_STREAMS_STREAM ?? '').trim();
  if (redisStreamsEnabled || redisStreamsUrl || redisStreamsStream) {
    try {
      const resolvedRedisStreamsClient = redisStreamsClient ?? (
        redisStreamsEnabled && redisStreamsUrl
          ? await createRedisStreamsClient({
              url: redisStreamsUrl,
              stream: redisStreamsStream || undefined,
              importRedis,
            })
          : null
      );
      if (resolvedRedisStreamsClient?.close) closers.push(() => resolvedRedisStreamsClient.close());
      redisStreamsBackend = createRedisStreamsMemoryBackend({
        enabled: redisStreamsEnabled,
        url: redisStreamsUrl,
        stream: redisStreamsStream || undefined,
        client: resolvedRedisStreamsClient,
      });
    } catch (err) {
      logger?.warn?.(
        `[memory-v2] redis-streams backend unavailable: ${err instanceof Error ? err.message : err}`,
      );
      redisStreamsBackend = createRedisStreamsMemoryBackend({
        enabled: false,
        unavailableReason: 'init_failed',
      });
    }
  }
  if (redisStreamsBackend) backends.push(redisStreamsBackend);
  let langGraphBackend = null;
  const langGraphEnabled = readFlag(env, 'MEMORY_LANGGRAPH_ENABLED', false);
  const langGraphPolicyId = String(env?.MEMORY_LANGGRAPH_POLICY_ID ?? '').trim();
  const langGraphUrl = String(env?.MEMORY_LANGGRAPH_URL ?? '').trim();
  const langGraphModelProviderKeyId = String(env?.MEMORY_LANGGRAPH_MODEL_PROVIDER_KEY_ID ?? '').trim();
  const langGraphModel = String(env?.MEMORY_LANGGRAPH_MODEL ?? '').trim();
  const langGraphModelApiType = String(env?.MEMORY_LANGGRAPH_MODEL_API ?? '').trim();
  if (langGraphEnabled || langGraphPolicyId || langGraphUrl) {
    try {
      const resolvedLangGraphClient = langGraphClient ?? (
        langGraphEnabled && langGraphUrl
          ? createLangGraphHttpClient({
              baseUrl: langGraphUrl,
              apiKey: env?.MEMORY_LANGGRAPH_API_KEY,
              resolvePath: env?.MEMORY_LANGGRAPH_RESOLVE_PATH || undefined,
              modelProviderKeyId: langGraphModelProviderKeyId || undefined,
              model: langGraphModel || undefined,
              modelApiType: langGraphModelApiType || undefined,
              fetchImpl,
              timeoutMs: Number(env?.MEMORY_LANGGRAPH_TIMEOUT_MS ?? 3000) || 3000,
            })
          : null
      );
      langGraphBackend = createLangGraphMemoryBackend({
        enabled: langGraphEnabled,
        policyId: langGraphPolicyId || undefined,
        client: resolvedLangGraphClient,
      });
    } catch (err) {
      logger?.warn?.(
        `[memory-v2] langgraph backend unavailable: ${err instanceof Error ? err.message : err}`,
      );
      langGraphBackend = createLangGraphMemoryBackend({
        enabled: false,
        unavailableReason: 'init_failed',
      });
    }
  }
  if (langGraphBackend) backends.push(langGraphBackend);
  backends.push(...createMemoryV2PluginBackends({
    exclude: [
      ...(pgvectorBackend ? ['pgvector'] : []),
      ...(qdrantBackend ? ['qdrant'] : []),
      ...(weaviateBackend ? ['weaviate'] : []),
      ...(mem0Backend ? ['mem0'] : []),
      ...(lettaBackend ? ['letta'] : []),
      ...(neo4jBackend ? ['neo4j'] : []),
      ...(redisStreamsBackend ? ['redis-streams'] : []),
      ...(langGraphBackend ? ['langgraph'] : []),
    ],
  }));

  let personalCandidateStore = null;
  if (readFlag(env, 'MEMORY_CANDIDATE_PERSISTENCE_ENABLED', false) && mysqlPool?.query) {
    try {
      await ensurePersonalMemoryCandidateSchema(mysqlPool);
      await ensureMemoryV2ProductEventsSchema(mysqlPool);
      personalCandidateStore = createPersonalMemoryCandidateStore(mysqlPool);
      const originalSaveCandidate = personalCandidateStore.saveCandidate.bind(personalCandidateStore);
      personalCandidateStore.saveCandidate = async (candidate, options = {}) => {
        const persisted = await originalSaveCandidate(candidate, options);
        if (persisted?.inserted) {
          void recordMemoryV2ProductEvent(mysqlPool, {
            eventType: MEMORY_V2_PRODUCT_EVENT_TYPES.CANDIDATE_SAVED,
            userId: candidate.userId,
            sessionId: candidate.sessionId,
            candidateId: candidate.id,
            data: {
              policyReason: candidate.policyReason,
              memoryType: candidate.memoryType,
              status: persisted.status,
              autoAccept: Boolean(options.autoAccept),
            },
          }).catch(() => {});
          if (persisted.status === 'candidate') {
            try {
              const reviewed = await autoReviewCandidateIfPending(mysqlPool, {
                ...candidate,
                status: 'candidate',
              }, { env });
              if (reviewed.reviewed && reviewed.status) {
                persisted.status = reviewed.status;
              }
            } catch (err) {
              logger?.warn?.(
                `[memory-v2] candidate auto-review skipped: ${err instanceof Error ? err.message : err}`,
              );
            }
          }
        }
        return persisted;
      };
    } catch (err) {
      logger?.warn?.(
        `[memory-v2] candidate persistence unavailable: ${err instanceof Error ? err.message : err}`,
      );
    }
  }
  const personalShadowPipeline = createPersonalMemoryShadowPipeline({
    env,
    store: personalCandidateStore,
  });
  const memory = createMemoryV2({
    legacyMemoryService,
    backends,
    env,
    logger,
    personalShadowPipeline,
  });
  const lifecycle = createMemoryV2LifecycleService({ pool: mysqlPool, env, logger });
  let lifecycleTimer = null;
  const lifecycleWorkerEnabled = readFlag(env, 'MEMORY_LIFECYCLE_WORKER_ENABLED', false);
  const lifecycleWorkerScopes = resolveMemoryV2LifecycleWorkerScopes(lifecycle.policy);
  if (lifecycleWorkerEnabled && mysqlPool?.query && lifecycleWorkerScopes.length > 0) {
    const intervalMs = Math.max(60_000, Number(lifecycle.policy.compactIntervalHours ?? 24) * 3_600_000);
    const runLifecycle = async () => {
      try {
        for (const userId of lifecycleWorkerScopes) {
          const input = userId == null ? {} : { userId };
          await runCandidateAutoReviewBatch(mysqlPool, {
            env,
            userId,
            limit: 200,
          });
          await lifecycle.expire(input);
          await lifecycle.compact(input);
          await lifecycle.promote(input);
          await lifecycle.reflect(input);
        }
      } catch (err) {
        logger?.warn?.(`[memory-v2] lifecycle worker skipped: ${err instanceof Error ? err.message : err}`);
      }
    };
    lifecycleTimer = setInterval(runLifecycle, intervalMs);
    lifecycleTimer.unref?.();
  }
  const originalGetStatus = memory.getStatus.bind(memory);
  memory.getStatus = () => ({
    ...originalGetStatus(),
    lifecycle: lifecycle.getStatus(),
  });
  memory.lifecycle = lifecycle;

  const pgBackfillEnabled = readFlag(env, 'MEMORY_PGVECTOR_BACKFILL_ENABLED', pgvectorEnabled);
  let pgBackfillBusy = false;
  let pgBackfillCursor = { updatedAt: 0, id: '' };

  async function syncPgvectorFromLegacy({ userId = null, sessionId = null } = {}) {
    if (!pgBackfillEnabled || !pgPoolRef || !pgEmbedQuery || !mysqlPool?.query) return;
    if (userId) {
      try {
        await syncLegacyUserMemoriesToPgvector({
          mysqlPool,
          pgPool: pgPoolRef,
          embedMemory: (item) => pgEmbedQuery(item.text, item),
          tableName: env?.MEMORY_PGVECTOR_TABLE,
          userId,
          sessionId,
          limit: Math.max(1, Number(env?.MEMORY_PGVECTOR_SYNC_USER_LIMIT ?? 50) || 50),
        });
      } catch (err) {
        logger?.warn?.(
          `[memory-v2] pgvector user sync skipped: ${err instanceof Error ? err.message : err}`,
        );
      }
      return;
    }
    if (pgBackfillBusy) return;
    pgBackfillBusy = true;
    try {
      const maxBatches = Math.max(1, Number(env?.MEMORY_PGVECTOR_BACKFILL_MAX_BATCHES ?? 3) || 3);
      const batchLimit = Math.max(1, Number(env?.MEMORY_PGVECTOR_BACKFILL_LIMIT ?? 50) || 50);
      for (let batch = 0; batch < maxBatches; batch += 1) {
        const result = await backfillLegacyMemoriesToPgvector({
          mysqlPool,
          pgPool: pgPoolRef,
          embedMemory: (item) => pgEmbedQuery(item.text, item),
          tableName: env?.MEMORY_PGVECTOR_TABLE,
          cursor: pgBackfillCursor,
          limit: batchLimit,
          dryRun: false,
        });
        pgBackfillCursor = result.nextCursor;
        if (!result.hasMore || result.scanned === 0) break;
      }
    } catch (err) {
      logger?.warn?.(
        `[memory-v2] pgvector backfill skipped: ${err instanceof Error ? err.message : err}`,
      );
    } finally {
      pgBackfillBusy = false;
    }
  }

  const originalLifecyclePromote = lifecycle.promote.bind(lifecycle);
  lifecycle.promote = async (input = {}) => {
    const result = await originalLifecyclePromote(input);
    if (Number(result?.promoted ?? 0) > 0) {
      void recordMemoryV2ProductEvent(mysqlPool, {
        eventType: MEMORY_V2_PRODUCT_EVENT_TYPES.PROMOTED,
        userId: input?.userId ?? null,
        data: {
          promoted: result.promoted,
          promotedUserIds: result.promotedUserIds ?? [],
        },
      }).catch(() => {});
      const promotedUserIds = Array.isArray(result?.promotedUserIds)
        ? result.promotedUserIds
        : input?.userId ? [input.userId] : [];
      for (const userId of promotedUserIds) {
        await syncPgvectorFromLegacy({ userId });
      }
    }
    return result;
  };

  function wrapMemorySideEffect(methodName) {
    const original = memory[methodName].bind(memory);
    memory[methodName] = async (input = {}) => {
      const result = await original(input);
      if (input?.userId || (result?.memories ?? 0) > 0 || (result?.analyzed ?? 0) > 0) {
        await syncPgvectorFromLegacy({
          userId: input?.userId ?? null,
          sessionId: input?.sessionId ?? null,
        });
      }
      return result;
    };
  }

  if (pgBackfillEnabled && pgPoolRef && pgEmbedQuery && mysqlPool) {
    wrapMemorySideEffect('write');
    wrapMemorySideEffect('compact');
  }

  memory.close = async () => {
    if (lifecycleTimer) clearInterval(lifecycleTimer);
    const results = await Promise.allSettled(closers.map((close) => close()));
    for (const result of results) {
      if (result.status === 'rejected') {
        logger?.warn?.(
          `[memory-v2] close skipped: ${result.reason instanceof Error ? result.reason.message : result.reason}`,
        );
      }
    }
  };

  return memory;
}

export async function createManagedMemoryV2Runtime({
  legacyMemoryService = null,
  mysqlPool = null,
  configService = null,
  env = process.env,
  logger = console,
  importPg = (specifier) => import(specifier),
  importRedis = (specifier) => import(specifier),
  importModule = (specifier) => import(specifier),
  fetchImpl = globalThis.fetch,
  lettaClient = null,
  weaviateClient = null,
  neo4jClient = null,
  redisStreamsClient = null,
  langGraphClient = null,
} = {}) {
  let activeRuntime = null;
  let activeFingerprint = null;
  let activeMeta = { source: 'env', updatedAt: null, updatedBy: null, configError: null };

  async function loadRuntimeState() {
    if (!configService?.getRuntimeState) {
      return {
        source: 'env',
        updatedAt: null,
        updatedBy: null,
        fingerprint: 'env-only',
        effectiveEnv: { ...env },
        configError: null,
      };
    }
    try {
      const state = await configService.getRuntimeState();
      return {
        source: state?.source ?? 'admin-db',
        updatedAt: state?.updatedAt ?? null,
        updatedBy: state?.updatedBy ?? null,
        fingerprint: state?.fingerprint ?? `admin-db:${Date.now()}`,
        effectiveEnv: { ...env, ...(state?.overrides ?? {}) },
        configError: null,
      };
    } catch (err) {
      logger?.warn?.(
        `[memory-v2] admin config unavailable, using process env: ${err instanceof Error ? err.message : err}`,
      );
      return {
        source: 'env-fallback',
        updatedAt: null,
        updatedBy: null,
        fingerprint: 'env-fallback',
        effectiveEnv: { ...env },
        configError: err instanceof Error ? err.message : String(err),
      };
    }
  }

  async function ensureRuntime() {
    const state = await loadRuntimeState();
    state.effectiveEnv = applyGooseCanaryMemoryPolicy(state.effectiveEnv, env);
    const canaryMode = String(env.TKMIND_GOOSE_CANARY ?? 'off').trim().toLowerCase();
    if (canaryMode === 'all' || canaryMode === 'users') {
      state.source = `goose-canary:${state.source}`;
    }
    state.fingerprint = `${state.fingerprint}:goose-canary=${canaryMode}`;
    if (activeRuntime && activeFingerprint === state.fingerprint) {
      activeMeta = state;
      return activeRuntime;
    }
    const nextRuntime = await createMemoryV2Runtime({
      legacyMemoryService,
      mysqlPool,
      env: state.effectiveEnv,
      logger,
      importPg,
      importRedis,
      importModule,
      fetchImpl,
      lettaClient,
      weaviateClient,
      neo4jClient,
      redisStreamsClient,
      langGraphClient,
    });
    const previous = activeRuntime;
    activeRuntime = nextRuntime;
    activeFingerprint = state.fingerprint;
    activeMeta = state;
    if (previous?.close) {
      await previous.close().catch((err) => {
        logger?.warn?.(
          `[memory-v2] previous runtime close skipped: ${err instanceof Error ? err.message : err}`,
        );
      });
    }
    return activeRuntime;
  }

  const managed = {
    get policy() {
      return activeRuntime?.policy ?? resolveMemoryV2Policy({ env });
    },

    async getStatus() {
      const runtime = await ensureRuntime();
      const status = await Promise.resolve(runtime.getStatus());
      return {
        ...status,
        configSource: activeMeta.source,
        configUpdatedAt: activeMeta.updatedAt,
        configUpdatedBy: activeMeta.updatedBy,
        configError: activeMeta.configError,
      };
    },

    async resolve(input = {}) {
      const runtime = await ensureRuntime();
      return runtime.resolve(input);
    },

    async write(input = {}) {
      const runtime = await ensureRuntime();
      return runtime.write(input);
    },

    async compact(input = {}) {
      const runtime = await ensureRuntime();
      return runtime.compact(input);
    },

    async observePersonalMemory(input = {}) {
      const runtime = await ensureRuntime();
      return runtime.observePersonalMemory(input);
    },

    async listMemories(input = {}) {
      const runtime = await ensureRuntime();
      return runtime.lifecycle?.listMemories?.(input) ?? [];
    },

    async forgetMemory(input = {}) {
      const runtime = await ensureRuntime();
      return runtime.lifecycle?.forgetMemory?.(input) ?? { ok: false, skipped: true, reason: 'unavailable' };
    },

    async runLifecycle(input = {}) {
      const runtime = await ensureRuntime();
      const lifecycle = runtime.lifecycle;
      if (!lifecycle) return { ok: false, reason: 'unavailable' };
      return {
        expire: await lifecycle.expire(input),
        compact: await lifecycle.compact(input),
        promote: await lifecycle.promote(input),
        reflect: await lifecycle.reflect(input),
      };
    },

    async close() {
      const runtime = activeRuntime;
      activeRuntime = null;
      activeFingerprint = null;
      if (runtime?.close) {
        await runtime.close();
      }
    },
  };

  return managed;
}