/**
 * LanceDB Storage Layer with Multi-Scope Support
 */

import type * as LanceDB from "@lancedb/lancedb";
import { createHash, randomUUID } from "node:crypto";
import { existsSync, accessSync, constants, mkdirSync, realpathSync, lstatSync } from "node:fs";
import { dirname, join } from "node:path";
import { logWarn } from "./stderr-log.js";
import { readConsistencyInterval as envReadConsistencyInterval } from "./env-config.js";
import type { DurableMemoryCategory } from "./memory-schema.js";
import { matchesScopeFilter, type ScopeMatchMode } from "./scope-policy.js";
import { detectEmotionIfEnabled } from "./emotion-detector.js";
import { withWriteLock } from "./distill-lock.js";
import { incrementWriteCount } from "./activity-counter.js";
import type { MemoryStorePort, MemoryStoreStats, MemoryStoreUpdate } from "./memory-store-port.js";
import { clearVersionGroupMetadata, planSingletonVersionGroupRepairs } from "./version-manager.js";

// ============================================================================
// Types
// ============================================================================

/**
 * Memory categories (v1.1 six-category system, inspired by OpenViking).
 *
 * User Memory (4):
 *   profile      — 用户身份/背景（静态，合并优先）
 *   preferences  — 偏好/倾向（合并优先）
 *   entities     — 持续存在的名词：项目/工具/人物（合并优先）
 *   events       — 发生过的事（追加，不合并）
 *
 * Agent Memory (2):
 *   cases        — 问题→解决方案对（追加，不合并）
 *   patterns     — 可复用的流程/模式（合并优先）
 *
 * Legacy categories kept for backward compatibility.
 */
export type MemoryCategory =
  | DurableMemoryCategory
  | "preference" | "fact" | "decision" | "entity" | "other";

export interface MemoryEntry {
  id: string;
  text: string;
  vector: number[];
  category: MemoryCategory;
  scope: string;
  importance: number;
  timestamp: number;
  metadata?: string; // JSON string for extensible metadata — includes l0_abstract/l1_overview/l2_content/tier
  language?: string;   // ISO 639-1: "zh"|"ja"|"ko"|"en"
  fts_text?: string;   // Pre-tokenized text for FTS indexing
}

export interface MemorySearchResult {
  entry: MemoryEntry;
  score: number;
}

export interface StoreConfig {
  dbPath: string;
  vectorDim: number;
  /**
   * LanceDB readConsistencyInterval (seconds): 0 = strong consistency (external
   * writes visible on every read), positive = bounded staleness window.
   * Omitted → RECALLNEST_READ_CONSISTENCY_INTERVAL decides (default 0; "off"
   * restores the legacy unchecked-handle behavior).
   */
  readConsistencyInterval?: number;
}

export type LegacyScopeIssueKind = "missing" | "empty" | "global";

export interface LegacyScopeAuditSample {
  id: string;
  kind: LegacyScopeIssueKind;
  scope: string | null;
  category: MemoryEntry["category"];
  timestamp: number;
  text: string;
}

export interface LegacyScopeAudit {
  totalCount: number;
  counts: Record<LegacyScopeIssueKind, number>;
  samples: LegacyScopeAuditSample[];
}

// ============================================================================
// Deterministic ID
// ============================================================================

const RECALLNEST_NS = "recallnest:v1";

/**
 * Generate a deterministic UUID-formatted ID from scope + text.
 * Same inputs always produce the same ID — prevents duplicate entries
 * with different random UUIDs.
 *
 * Format: 8-4-4-4-12 hex (same shape as crypto.randomUUID).
 */
export function deterministicId(scope: string, text: string): string {
  const hash = createHash("sha256")
    .update(`${RECALLNEST_NS}\0${scope}\0${text}`)
    .digest("hex");
  // Format as UUID: 8-4-4-4-12
  return `${hash.slice(0, 8)}-${hash.slice(8, 12)}-${hash.slice(12, 16)}-${hash.slice(16, 20)}-${hash.slice(20, 32)}`;
}

// ============================================================================
// LanceDB Dynamic Import
// ============================================================================

let lancedbImportPromise: Promise<typeof import("@lancedb/lancedb")> | null = null;

export const loadLanceDB = async (): Promise<typeof import("@lancedb/lancedb")> => {
  if (!lancedbImportPromise) {
    lancedbImportPromise = import("@lancedb/lancedb");
  }
  try {
    return await lancedbImportPromise;
  } catch (err) {
    throw new Error(`memory-lancedb-pro: failed to load LanceDB. ${String(err)}`, { cause: err });
  }
};

// ============================================================================
// Utility Functions
// ============================================================================

function clampInt(value: number, min: number, max: number): number {
  if (!Number.isFinite(value)) return min;
  return Math.min(max, Math.max(min, Math.floor(value)));
}

function escapeSqlLiteral(value: string): string {
  return value.replace(/'/g, "''");
}

export function classifyLegacyScope(scope: unknown): LegacyScopeIssueKind | undefined {
  if (scope == null) {
    return "missing";
  }
  if (typeof scope !== "string") {
    return "missing";
  }
  const normalized = scope.trim();
  if (normalized.length === 0) {
    return "empty";
  }
  if (normalized === "global") {
    return "global";
  }
  return undefined;
}

// ============================================================================
// Storage Path Validation
// ============================================================================

/**
 * Validate and prepare the storage directory before LanceDB connection.
 * Resolves symlinks, creates missing directories, and checks write permissions.
 * Returns the resolved absolute path on success, or throws a descriptive error.
 */
export function validateStoragePath(dbPath: string): string {
  let resolvedPath = dbPath;

  // Resolve symlinks if the path already exists
  try {
    if (existsSync(dbPath)) {
      const stats = lstatSync(dbPath);
      if (stats.isSymbolicLink()) {
        try {
          resolvedPath = realpathSync(dbPath);
        } catch (err: any) {
          throw new Error(
            `dbPath "${dbPath}" is a symlink whose target does not exist.\n` +
            `  Fix: Create the target directory, or update the symlink to point to a valid path.\n` +
            `  Details: ${err.code || ""} ${err.message}`
          );
        }
      }
    }
  } catch (err: any) {
    // Re-throw our own descriptive errors
    if (err.message.includes("symlink")) throw err;
    // Other lstat failures — continue with original path
  }

  // Create directory if it doesn't exist
  if (!existsSync(resolvedPath)) {
    try {
      mkdirSync(resolvedPath, { recursive: true });
    } catch (err: any) {
      throw new Error(
        `Failed to create dbPath directory "${resolvedPath}".\n` +
        `  Fix: Ensure the parent directory "${dirname(resolvedPath)}" exists and is writable,\n` +
        `       or create it manually: mkdir -p "${resolvedPath}"\n` +
        `  Details: ${err.code || ""} ${err.message}`
      );
    }
  }

  // Check write permissions
  try {
    accessSync(resolvedPath, constants.W_OK);
  } catch (err: any) {
    throw new Error(
      `dbPath directory "${resolvedPath}" is not writable.\n` +
      `  Fix: Check permissions with: ls -la "${dirname(resolvedPath)}"\n` +
      `       Or grant write access: chmod u+w "${resolvedPath}"\n` +
      `  Details: ${err.code || ""} ${err.message}`
    );
  }

  return resolvedPath;
}

// ============================================================================
// Memory Store
// ============================================================================

const TABLE_NAME = "memories";

/**
 * 把 scopeFilter 编译成 SQL 条件。**store 里所有 scope 过滤的唯一出处**
 * （2026-08-16 收敛，此前同一条规则在 6 个方法里各抄了一遍）。
 *
 * `mode` 见 scope-policy.ts 的 ScopeMatchMode：默认 `family` = 历史行为（有冒号精确 /
 * 无冒号前缀），`exact` = 调用方明说给的是具体 scope 名、一律精确。
 * 应用层双检 `matchesScopeFilter(row, filter, mode)` 必须传同一个 mode，否则 SQL 收紧了
 * 应用层又放行（或反之），两层判据打架。
 */
function scopeWhereClause(scopeFilter: string[], mode: ScopeMatchMode = "family"): string {
  return scopeFilter
    .map(scope => {
      const safe = escapeSqlLiteral(scope);
      if (mode === "exact") return `scope = '${safe}'`;
      return scope.includes(":")
        ? `scope = '${safe}'`
        : `scope LIKE '${safe}%'`;
    })
    .join(" OR ");
}

export class MemoryStore implements MemoryStorePort {
  private db: LanceDB.Connection | null = null;
  private table: LanceDB.Table | null = null;
  private initPromise: Promise<void> | null = null;
  private ftsIndexCreated = false;
  /** Per-id serial queues for patchMetadata; entries self-remove on settle. */
  private readonly metadataPatchQueues = new Map<string, Promise<void>>();

  /** Set when the init-time schema migration fails; surfaced in store error messages. */
  private schemaMigrationError: string | null = null;

  constructor(private readonly config: StoreConfig) {}

  get dbPath(): string {
    return this.config.dbPath;
  }

  private async ensureInitialized(): Promise<void> {
    if (this.table) {
      return;
    }
    if (this.initPromise) {
      return this.initPromise;
    }

    this.initPromise = this.doInitialize().catch((err) => {
      this.initPromise = null;
      throw err;
    });
    return this.initPromise;
  }

  private async doInitialize(): Promise<void> {
    const lancedb = await loadLanceDB();

    let db: LanceDB.Connection;
    try {
      // Without a consistency interval a long-lived handle pins its manifest
      // version — resident MCP/API/UI servers would never see CLI or
      // sibling-process writes. Explicit config wins so tests can pin behavior.
      db = await lancedb.connect(this.config.dbPath, {
        readConsistencyInterval:
          this.config.readConsistencyInterval ?? envReadConsistencyInterval(),
      });
    } catch (err: any) {
      const code = err.code || "";
      const message = err.message || String(err);
      throw new Error(
        `Failed to open LanceDB at "${this.config.dbPath}": ${code} ${message}\n` +
        `  Fix: Verify the path exists and is writable. Check parent directory permissions.`
      );
    }

    let table: LanceDB.Table;

    // Idempotent table init: try openTable first, create only if missing,
    // and handle the race where tableNames() misses an existing table but
    // createTable then sees it (LanceDB eventual consistency).
    try {
      table = await db.openTable(TABLE_NAME);

      // Backward compatibility: add missing columns to existing tables.
      // Read the schema definition, not sample rows — a row-sampling check can
      // never fire on a 0-row table, which deadlocked every write with
      // "Found field not in schema" on empty pre-language/fts_text tables.
      try {
        const schema = await table.schema();
        const fields = new Set(schema.fields.map((f) => f.name));
        const missing: Array<{ name: string; valueSql: string }> = [];
        if (!fields.has("language")) missing.push({ name: "language", valueSql: "'en'" });
        if (!fields.has("fts_text")) missing.push({ name: "fts_text", valueSql: "text" });
        if (missing.length > 0) {
          logWarn(`Adding missing columns for backward compatibility: ${missing.map((c) => c.name).join(", ")}`);
          await table.addColumns(missing);
        }
        this.schemaMigrationError = null;
      } catch (err) {
        this.schemaMigrationError = err instanceof Error ? err.message : String(err);
        logWarn("Could not check/migrate table schema — subsequent writes may fail:", err);
      }
    } catch (_openErr) {
      // Table doesn't exist yet — create it
      const schemaEntry: MemoryEntry = {
        id: "__schema__",
        text: "",
        vector: Array.from({ length: this.config.vectorDim }).fill(0) as number[],
        category: "other",
        scope: "__schema__",
        importance: 0,
        timestamp: 0,
        metadata: "{}",
        language: "en",
        fts_text: "__schema__",
      };

      try {
        table = await db.createTable(TABLE_NAME, [schemaEntry]);
        await table.delete('id = "__schema__"');
      } catch (createErr) {
        // Race: another caller (or eventual consistency) created the table
        // between our failed openTable and this createTable — just open it.
        if (String(createErr).includes("already exists")) {
          table = await db.openTable(TABLE_NAME);
        } else {
          throw createErr;
        }
      }
    }

    // Validate vector dimensions
    // Note: LanceDB returns Arrow Vector objects, not plain JS arrays.
    // Array.isArray() returns false for Arrow Vectors, so use .length instead.
    const sample = await table.query().limit(1).toArray();
    if (sample.length > 0 && sample[0]?.vector?.length) {
      const existingDim = sample[0].vector.length;
      if (existingDim !== this.config.vectorDim) {
        throw new Error(
          `Vector dimension mismatch: table=${existingDim}, config=${this.config.vectorDim}. Create a new table/dbPath or set matching embedding.dimensions.`
        );
      }
    }

    // Create FTS index for BM25 search (graceful fallback if unavailable)
    try {
      await this.createFtsIndex(table);
      this.ftsIndexCreated = true;
    } catch (err) {
      logWarn("Failed to create FTS index, falling back to vector-only search:", err);
      this.ftsIndexCreated = false;
    }

    this.db = db;
    this.table = table;
  }

  private async createFtsIndex(table: LanceDB.Table): Promise<void> {
    try {
      // Check if FTS index already exists
      const indices = await table.listIndices();
      const hasFtsIndex = indices?.some((idx: any) =>
        idx.indexType === "FTS" || idx.columns?.includes("fts_text")
      );

      if (!hasFtsIndex) {
        // LanceDB @lancedb/lancedb >=0.26: use Index.fts() config
        const lancedb = await loadLanceDB();
        await table.createIndex("fts_text", {
          config: (lancedb as any).Index.fts(),
        });
      }
    } catch (err) {
      throw new Error(`FTS index creation failed: ${err instanceof Error ? err.message : String(err)}`);
    }
  }

  async store(entry: Omit<MemoryEntry, "id" | "timestamp"> & { id?: string }): Promise<MemoryEntry> {
    await this.ensureInitialized();

    const fullEntry: MemoryEntry = {
      ...entry,
      id: entry.id || deterministicId(entry.scope, entry.text),
      timestamp: Date.now(),
      metadata: entry.metadata || "{}",
      language: entry.language || "en",
      fts_text: entry.fts_text || entry.text,
    };

    const emotionResult = detectEmotionIfEnabled(fullEntry.text);
    if (emotionResult) {
      const existingMeta = JSON.parse(fullEntry.metadata || "{}");
      existingMeta.emotion = emotionResult;
      fullEntry.metadata = JSON.stringify(existingMeta);
    }

    try {
      // P0-2: go through upsert (delete+add under the global store-write lock) rather
      // than a bare table.add — with the deterministic id above, storing the same
      // (scope,text) twice now yields one row, not a dup-id pair (fixes the append bug).
      return await this.upsert(fullEntry);
    } catch (err: any) {
      const code = err.code || "";
      const message = err.message || String(err);
      throw new Error(
        `Failed to store memory in "${this.config.dbPath}": ${code} ${message}` +
        (this.schemaMigrationError ? `\n  Schema migration failed at init: ${this.schemaMigrationError}` : "")
      );
    }
  }

  /**
   * Batch store multiple entries at once — much faster than individual store() calls.
   * LanceDB handles bulk inserts efficiently with a single index update.
   */
  async storeBatch(entries: (Omit<MemoryEntry, "id" | "timestamp"> & { id?: string })[]): Promise<number> {
    if (entries.length === 0) return 0;
    await this.ensureInitialized();

    const fullEntries: MemoryEntry[] = entries.map(entry => ({
      ...entry,
      // P0-2: honor a caller-supplied id (consolidation insight/pattern idempotency),
      // else the deterministic content id — unified with store() (was: always recompute).
      id: entry.id ?? deterministicId(entry.scope, entry.text),
      timestamp: Date.now(),
      metadata: entry.metadata || "{}",
      language: entry.language || "en",
      fts_text: entry.fts_text || entry.text,
    }));

    const enrichedEntries = fullEntries.map(e => {
      const emotionResult = detectEmotionIfEnabled(e.text);
      if (emotionResult) {
        const meta = JSON.parse(e.metadata || "{}");
        meta.emotion = emotionResult;
        return { ...e, metadata: JSON.stringify(meta) };
      }
      return e;
    });

    // P2: collapse in-batch duplicate ids (latest-wins) BEFORE writing. Two entries
    // with the same (scope,text) → same deterministic id; adding both would create a
    // fresh dup-id pair in one shot (delete-set was deduped but the add was not).
    const dedupedById = new Map<string, MemoryEntry>();
    for (const e of enrichedEntries) dedupedById.set(e.id, e); // later occurrence wins
    const toWrite = [...dedupedById.values()];

    try {
      await withWriteLock("store-write", async () => {
        // P0-1/P0-2/P1: atomic idempotent batch write via mergeInsert (no delete+add
        // window a crash could tear). toWrite is already in-batch deduped by id above,
        // so the merge key matches ≤1 existing row (given no pre-existing dup-id rows).
        await this.table!.mergeInsert("id").whenMatchedUpdateAll().whenNotMatchedInsertAll().execute(toWrite);
      }, { expireMs: 30_000 });
    } catch (err: any) {
      const code = err.code || "";
      const message = err.message || String(err);
      throw new Error(
        `Failed to batch store ${entries.length} memories in "${this.config.dbPath}": ${code} ${message}` +
        (this.schemaMigrationError ? `\n  Schema migration failed at init: ${this.schemaMigrationError}` : "")
      );
    }
    // Per-scope write counts drive dream activation. Best-effort — never block writes.
    try {
      const statsPath = this.activityStatsPath();
      const countByScope = new Map<string, number>();
      for (const e of toWrite) countByScope.set(e.scope, (countByScope.get(e.scope) ?? 0) + 1);
      for (const [scope, n] of countByScope) await incrementWriteCount(scope, n, { statsPath });
    } catch { /* activity tracking is best-effort */ }
    return toWrite.length;
  }

  /**
   * P0-1/P0-2/P1: Idempotent, crash-atomic write of a fully-formed entry, serialized
   * across processes.
   *
   * Uses mergeInsert("id") (same primitive as update()) so the write is atomic — no
   * delete+add window that a crash between the two steps could tear, losing the row
   * (P1). The global `store-write` lock still serializes concurrent merges across the
   * 11 mcp-server processes so two whenNotMatched inserts can't both fire for the same
   * id (the P0-1 race). "Same id overwrites" gives content-addressed idempotency to
   * every store()/storeBatch() caller whose id is deterministicId(scope,text).
   * Precondition: no pre-existing dup-id rows in the table (run the dedup migration
   * scripts/cleanup-duplicate-ids.ts first) — else the merge key match is ambiguous.
   */
  /** Path to this store's activity-counter stats file, beside the LanceDB dir. */
  private activityStatsPath(): string {
    return join(dirname(this.config.dbPath), "activity-stats.json");
  }

  async upsert(entry: MemoryEntry): Promise<MemoryEntry> {
    await this.ensureInitialized();
    await withWriteLock("store-write", async () => {
      await this.table!.mergeInsert("id").whenMatchedUpdateAll().whenNotMatchedInsertAll().execute([entry]);
    }, { expireMs: 30_000 });
    // Per-scope write count drives dream activation. Best-effort — never block a write.
    try {
      await incrementWriteCount(entry.scope, 1, { statsPath: this.activityStatsPath() });
    } catch { /* activity tracking is best-effort */ }
    return entry;
  }

  /**
   * Import a pre-built entry while preserving its id/timestamp.
   * Used for re-embedding / migration / A/B testing across embedding models.
   * Intentionally separate from `store()` to keep normal writes simple.
   */
  async importEntry(entry: MemoryEntry): Promise<MemoryEntry> {
    await this.ensureInitialized();

    if (!entry.id || typeof entry.id !== "string") {
      throw new Error("importEntry requires a stable id");
    }

    const vector = entry.vector || [];
    if (!Array.isArray(vector) || vector.length !== this.config.vectorDim) {
      throw new Error(
        `Vector dimension mismatch: expected ${this.config.vectorDim}, got ${Array.isArray(vector) ? vector.length : 'non-array'}`
      );
    }

    const full: MemoryEntry = {
      ...entry,
      scope: entry.scope,
      importance: Number.isFinite(entry.importance) ? entry.importance : 0.5,
      timestamp: Number.isFinite(entry.timestamp) ? entry.timestamp : Date.now(),
      metadata: entry.metadata || "{}",
      language: entry.language || "en",
      fts_text: entry.fts_text || entry.text,
    };

    // HP-emo: Backfill emotion if absent during import
    try {
      const metaParsed = JSON.parse(full.metadata || "{}");
      if (!metaParsed.emotion) {
        const emotionResult = detectEmotionIfEnabled(full.text);
        if (emotionResult) {
          metaParsed.emotion = emotionResult;
          full.metadata = JSON.stringify(metaParsed);
        }
      }
    } catch { /* malformed metadata — skip backfill */ }

    await this.table!.add([full]);
    return full;
  }

  async hasId(id: string): Promise<boolean> {
    await this.ensureInitialized();
    const safeId = escapeSqlLiteral(id);
    const res = await this.table!.query().select(["id"]).where(`id = '${safeId}'`).limit(1).toArray();
    return res.length > 0;
  }

  async vectorSearch(
    vector: number[],
    limit = 5,
    minScore = 0.3,
    scopeFilter?: string[],
    scopeMatch: ScopeMatchMode = "family",
  ): Promise<MemorySearchResult[]> {
    await this.ensureInitialized();

    const safeLimit = clampInt(limit, 1, 20);
    const fetchLimit = Math.min(safeLimit * 10, 200); // Over-fetch for scope filtering

    let query = this.table!.vectorSearch(vector).distanceType('cosine').limit(fetchLimit);

    // Apply scope filter if provided. Default mode "family" = 历史行为（含冒号精确 /
    // 无冒号前缀）；exact 由调用方显式声明。**必须下推到 SQL，不能事后在应用层滤**——
    // top-k 在收窄前就被兄弟 scope 占满，滤完只剩残缺候选（2026-08-16 互审 C1）。
    if (scopeFilter && scopeFilter.length > 0) {
      query = query.where(`(${scopeWhereClause(scopeFilter, scopeMatch)})`);
    }

    const results = await query.toArray();
    const mapped: MemorySearchResult[] = [];

    for (const row of results) {
      const distance = Number(row._distance ?? 0);
      const score = 1 / (1 + distance);

      if (score < minScore) continue;

      const rowScope = (row.scope as string | undefined) ?? "";

      // Double-check scope filter in application layer (same mode as the SQL above)
      if (!matchesScopeFilter(rowScope, scopeFilter, scopeMatch)) {
        continue;
      }

      mapped.push({
        entry: {
          id: row.id as string,
          text: row.text as string,
          vector: row.vector as number[],
          category: row.category as MemoryEntry["category"],
          scope: rowScope,
          importance: Number(row.importance),
          timestamp: Number(row.timestamp),
          metadata: (row.metadata as string) || "{}",
        },
        score,
      });

      if (mapped.length >= safeLimit) break;
    }

    return mapped;
  }

  async bm25Search(query: string, limit = 5, scopeFilter?: string[]): Promise<MemorySearchResult[]> {
    await this.ensureInitialized();

    if (!this.ftsIndexCreated) {
      return []; // Fallback to vector-only if FTS unavailable
    }

    const safeLimit = clampInt(limit, 1, 20);

    try {
      // Use FTS query type explicitly
      let searchQuery = this.table!.search(query, "fts").limit(safeLimit);

      // Apply scope filter if provided (family mode, same as vectorSearch's default)
      if (scopeFilter && scopeFilter.length > 0) {
        searchQuery = searchQuery.where(`(${scopeWhereClause(scopeFilter)})`);
      }

      const results = await searchQuery.toArray();
      const mapped: MemorySearchResult[] = [];

      for (const row of results) {
        const rowScope = (row.scope as string | undefined) ?? "";

        // Double-check scope filter in application layer (prefix-aware)
        if (!matchesScopeFilter(rowScope, scopeFilter)) {
          continue;
        }

        // LanceDB FTS _score is raw BM25 (unbounded). Normalize with sigmoid.
        // LanceDB may return BigInt for numeric columns; coerce safely.
        const rawScore = (row._score != null) ? Number(row._score) : 0;
        const normalizedScore = rawScore > 0 ? 1 / (1 + Math.exp(-rawScore / 5)) : 0.5;

        mapped.push({
          entry: {
            id: row.id as string,
            text: row.text as string,
            vector: row.vector as number[],
            category: row.category as MemoryEntry["category"],
            scope: rowScope,
            importance: Number(row.importance),
            timestamp: Number(row.timestamp),
            metadata: (row.metadata as string) || "{}",
          },
          score: normalizedScore,
        });
      }

      return mapped;
    } catch (err) {
      logWarn("BM25 search failed, falling back to empty results:", err);
      return [];
    }
  }

  async delete(id: string, scopeFilter?: string[]): Promise<boolean> {
    await this.ensureInitialized();

    // Support both full UUID and short prefix (8+ hex chars)
    const uuidRegex = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i;
    const prefixRegex = /^[0-9a-f]{8,}$/i;
    const isFullId = uuidRegex.test(id);
    const isPrefix = !isFullId && prefixRegex.test(id);

    if (!isFullId && !isPrefix) {
      throw new Error(`Invalid memory ID format: ${id}`);
    }

    let candidates: any[];
    if (isFullId) {
      candidates = await this.table!.query().where(`id = '${id}'`).limit(1).toArray();
    } else {
      // v2.5.2 (2026-05-27): Use SQL LIKE for prefix lookup instead of app-layer scan
      // with limit(1000) — the old `.select().limit(1000).filter()` pattern silently
      // missed entries beyond row 1000 in 90K+ libraries (Codex trio review finding,
      // ref: ~/Desktop/codex-v2.5.1-fix-review-20260527.md, store.delete limit 漏洞).
      // Now mirrors getById prefix behavior: LIKE with limit(2) detects ambiguity.
      const safeId = escapeSqlLiteral(id);
      candidates = await this.table!.query()
        .select(["id", "scope"])
        .where(`id LIKE '${safeId}%'`)
        .limit(2)
        .toArray();
      if (candidates.length > 1) {
        throw new Error(`Ambiguous prefix "${id}" matches ${candidates.length}+ memories. Use a longer prefix or full ID.`);
      }
    }
    if (candidates.length === 0) {
      return false;
    }

    const resolvedId = candidates[0].id as string;
    const rowScope = (candidates[0].scope as string | undefined) ?? "";

    // Check scope permissions
    if (!matchesScopeFilter(rowScope, scopeFilter)) {
      throw new Error(`Memory ${resolvedId} is outside accessible scopes`);
    }

    await this.table!.delete(`id = '${resolvedId}'`);
    return true;
  }

  async list(
    scopeFilter?: string[],
    category?: string,
    limit = 20,
    offset = 0,
    scopeMatch: ScopeMatchMode = "family",
  ): Promise<MemoryEntry[]> {
    await this.ensureInitialized();

    let query = this.table!.query();

    // Build where conditions
    const conditions: string[] = [];

    if (scopeFilter && scopeFilter.length > 0) {
      // 收窄必须在 SQL 里：limit 是在这之后下推的，事后过滤等于先截断再筛，
      // 目标 scope 实际拿到多少行不可控（2026-08-16 互审 C1/K3）。
      conditions.push(`((${scopeWhereClause(scopeFilter, scopeMatch)}))`);
    }

    if (category) {
      conditions.push(`category = '${escapeSqlLiteral(category)}'`);
    }

    if (conditions.length > 0) {
      query = query.where(conditions.join(" AND "));
    }

    // Fetch all matching rows (no pre-limit) so app-layer sort is correct across full dataset
    const results = await query
      .select(["id", "text", "category", "scope", "importance", "timestamp", "metadata"])
      .toArray();

    return results
      .map((row): MemoryEntry => ({
        id: row.id as string,
        text: row.text as string,
        vector: [], // Don't include vectors in list results for performance
        category: row.category as MemoryEntry["category"],
        scope: (row.scope as string | undefined) ?? "",
        importance: Number(row.importance),
        timestamp: Number(row.timestamp),
        metadata: (row.metadata as string) || "{}",
      }))
      .sort((a, b) => (b.timestamp || 0) - (a.timestamp || 0))
      .slice(offset, offset + limit);
  }

  /**
   * True DB-level pagination — where/limit/offset are pushed down to LanceDB,
   * so unlike list() this never materializes the full match set in JS. Use for
   * full-corpus maintenance sweeps (auto-gc, promotion scan, lint/checkup).
   *
   * Caveats:
   * - No orderBy support in the LanceDB query API: rows come back in fragment
   *   scan order, NOT timestamp order. Stable across pages for a read-only
   *   sweep, but concurrent writes rewrite fragments and may shift rows
   *   between pages (a missed row is picked up by the next sweep). Do not use
   *   for user-facing ordered listings — that's what list() is for.
   * - includeVector loads the embedding column; only request it when the
   *   consumer actually needs vectors (they dominate row size).
   */
  async listPage(opts: {
    scopeFilter?: string[];
    category?: string;
    limit?: number;
    offset?: number;
    includeVector?: boolean;
  } = {}): Promise<MemoryEntry[]> {
    await this.ensureInitialized();
    const { scopeFilter, category, limit = 1000, offset = 0, includeVector = false } = opts;

    let query = this.table!.query();

    const conditions: string[] = [];
    if (scopeFilter && scopeFilter.length > 0) {
      conditions.push(`((${scopeWhereClause(scopeFilter)}))`);
    }
    if (category) {
      conditions.push(`category = '${escapeSqlLiteral(category)}'`);
    }
    if (conditions.length > 0) {
      query = query.where(conditions.join(" AND "));
    }

    const columns = ["id", "text", "category", "scope", "importance", "timestamp", "metadata"];
    if (includeVector) columns.push("vector");

    const results = await query
      .select(columns)
      .limit(limit)
      .offset(offset)
      .toArray();

    return results.map((row): MemoryEntry => ({
      id: row.id as string,
      text: row.text as string,
      vector: includeVector ? Array.from(row.vector as Iterable<number>) : [],
      category: row.category as MemoryEntry["category"],
      scope: (row.scope as string | undefined) ?? "",
      importance: Number(row.importance),
      timestamp: Number(row.timestamp),
      metadata: (row.metadata as string) || "{}",
    }));
  }

  async get(id: string, scopeFilter?: string[]): Promise<MemoryEntry | null> {
    await this.ensureInitialized();

    const uuidRegex = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i;
    const prefixRegex = /^[0-9a-f]{8,}$/i;
    const isFullId = uuidRegex.test(id);
    const isPrefix = !isFullId && prefixRegex.test(id);

    if (!isFullId && !isPrefix) {
      throw new Error(`Invalid memory ID format: ${id}`);
    }

    let rows: any[];
    if (isFullId) {
      const safeId = escapeSqlLiteral(id);
      rows = await this.table!
        .query()
        .select(["id", "text", "vector", "category", "scope", "importance", "timestamp", "metadata"])
        .where(`id = '${safeId}'`)
        .limit(1)
        .toArray();
    } else {
      const all = await this.table!
        .query()
        .select(["id", "text", "vector", "category", "scope", "importance", "timestamp", "metadata"])
        .toArray();
      rows = all.filter((r: any) => (r.id as string).startsWith(id));
      if (rows.length > 1) {
        throw new Error(`Ambiguous prefix "${id}" matches ${rows.length} memories. Use a longer prefix or full ID.`);
      }
    }

    if (rows.length === 0) return null;

    const row = rows[0];
    const rowScope = (row.scope as string | undefined) ?? "";
    if (!matchesScopeFilter(rowScope, scopeFilter)) {
      throw new Error(`Memory ${id} is outside accessible scopes`);
    }

    return {
      id: row.id as string,
      text: row.text as string,
      vector: Array.from(row.vector as Iterable<number>),
      category: row.category as MemoryEntry["category"],
      scope: rowScope,
      importance: Number(row.importance),
      timestamp: Number(row.timestamp),
      metadata: (row.metadata as string) || "{}",
    };
  }

  /**
   * Get a single entry by ID (full UUID) or 8+ hex prefix.
   * Lightweight read used by AccessTracker for metadata updates and by skill-engine
   * for outcome recording.
   *
   * v2.5.1 (2026-05-27) — 加 prefix lookup support：和 update() / delete() 行为对齐，
   * 让 agent 拿到 short id（如 store_skill 返回的 8 位前缀、人类粘贴的截断 id）也能
   * resolve 到完整 entry。歧义 prefix（匹配 >1 条）返回 null 让 caller 处理。
   * 现有 caller 全部传完整 UUID，向后兼容。
   */
  async getById(id: string): Promise<MemoryEntry | null> {
    await this.ensureInitialized();

    // Support both full UUID and 8+ hex prefix (same pattern as update() / delete()).
    const uuidRegex = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i;
    const prefixRegex = /^[0-9a-f]{8,}$/i;
    const isFullId = uuidRegex.test(id);
    const isPrefix = !isFullId && prefixRegex.test(id);
    if (!isFullId && !isPrefix) return null;

    const safeId = escapeSqlLiteral(id);
    const whereClause = isFullId ? `id = '${safeId}'` : `id LIKE '${safeId}%'`;

    const rows = await this.table!.query()
      .select(["id", "text", "vector", "category", "scope", "importance", "timestamp", "metadata"])
      .where(whereClause)
      .limit(2) // detect ambiguous prefix (matches >1)
      .toArray();
    if (rows.length === 0) return null;
    if (isPrefix && rows.length > 1) return null; // ambiguous prefix
    const row = rows[0];
    return {
      id: row.id as string,
      text: row.text as string,
      vector: Array.from(row.vector as Iterable<number>),
      category: row.category as MemoryEntry["category"],
      scope: (row.scope as string | undefined) ?? "",
      importance: Number(row.importance),
      timestamp: Number(row.timestamp),
      metadata: (row.metadata as string) || "{}",
    };
  }

  /**
   * Batch-fetch vectors for a list of entry IDs.
   * Used by graph-export for cross-scope semantic bridge computation.
   * Single LanceDB query — much faster than N individual getById calls.
   */
  async getVectors(ids: string[]): Promise<Map<string, number[]>> {
    await this.ensureInitialized();
    const result = new Map<string, number[]>();
    if (ids.length === 0) return result;

    // 分批:避免 id 很多时 OR 链过长（如 data-checkup/memory-lint 一次补几千条向量）。
    const BATCH = 500;
    for (let i = 0; i < ids.length; i += BATCH) {
      const batch = ids.slice(i, i + BATCH);
      const conditions = batch.map(id => `id = '${escapeSqlLiteral(id)}'`).join(" OR ");
      const rows = await this.table!.query()
        .select(["id", "vector"])
        .where(conditions)
        .limit(batch.length)
        .toArray();
      for (const row of rows) {
        const vec = Array.from(row.vector as Iterable<number>);
        if (vec.length > 0) result.set(row.id as string, vec);
      }
    }
    return result;
  }

  async stats(scopeFilter?: string[], scopeMatch: ScopeMatchMode = "family"): Promise<MemoryStoreStats> {
    await this.ensureInitialized();

    let query = this.table!.query();

    if (scopeFilter && scopeFilter.length > 0) {
      query = query.where(`((${scopeWhereClause(scopeFilter, scopeMatch)}))`);
    }

    const results = await query.select(["scope", "category"]).toArray();

    const scopeCounts: Record<string, number> = {};
    const categoryCounts: Record<string, number> = {};

    for (const row of results) {
      const scope = (row.scope as string | undefined) ?? "";
      const category = row.category as string;

      scopeCounts[scope] = (scopeCounts[scope] || 0) + 1;
      categoryCounts[category] = (categoryCounts[category] || 0) + 1;
    }

    return {
      totalCount: results.length,
      scopeCounts,
      categoryCounts,
    };
  }

  async auditLegacyScopes(limit = 20): Promise<LegacyScopeAudit> {
    await this.ensureInitialized();

    const safeLimit = clampInt(limit, 1, 200);
    const rows = await this.table!
      .query()
      .select(["id", "text", "category", "scope", "timestamp"])
      .toArray();

    const counts: Record<LegacyScopeIssueKind, number> = {
      missing: 0,
      empty: 0,
      global: 0,
    };

    const samples = rows
      .map((row) => {
        const scopeValue = (row.scope as string | null | undefined) ?? null;
        const kind = classifyLegacyScope(scopeValue);
        if (!kind) return null;
        counts[kind] += 1;
        return {
          id: row.id as string,
          kind,
          scope: scopeValue,
          category: row.category as MemoryEntry["category"],
          timestamp: Number(row.timestamp),
          text: row.text as string,
        } satisfies LegacyScopeAuditSample;
      })
      .filter((row): row is LegacyScopeAuditSample => row !== null)
      .sort((a, b) => b.timestamp - a.timestamp)
      .slice(0, safeLimit);

    return {
      totalCount: counts.missing + counts.empty + counts.global,
      counts,
      samples,
    };
  }

  async update(
    id: string,
    updates: MemoryStoreUpdate,
    scopeFilter?: string[]
  ): Promise<MemoryEntry | null> {
    await this.ensureInitialized();

    // Support both full UUID and short prefix (8+ hex chars), same as delete()
    const uuidRegex = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i;
    const prefixRegex = /^[0-9a-f]{8,}$/i;
    const isFullId = uuidRegex.test(id);
    const isPrefix = !isFullId && prefixRegex.test(id);

    if (!isFullId && !isPrefix) {
      throw new Error(`Invalid memory ID format: ${id}`);
    }

    let rows: any[];
    if (isFullId) {
      const safeId = escapeSqlLiteral(id);
      rows = await this.table!.query().where(`id = '${safeId}'`).limit(1).toArray();
    } else {
      // Prefix match
      const all = await this.table!.query().select(["id", "text", "vector", "category", "scope", "importance", "timestamp", "metadata", "language", "fts_text"]).toArray();
      rows = all.filter((r: any) => (r.id as string).startsWith(id));
      if (rows.length > 1) {
        throw new Error(`Ambiguous prefix "${id}" matches ${rows.length} memories. Use a longer prefix or full ID.`);
      }
    }

    if (rows.length === 0) return null;

    const row = rows[0];
    const rowScope = (row.scope as string | undefined) ?? "";

    // Check scope permissions
    if (!matchesScopeFilter(rowScope, scopeFilter)) {
      throw new Error(`Memory ${id} is outside accessible scopes`);
    }

    // Build updated entry, preserving original timestamp and language fields
    const updated: MemoryEntry = {
      id: row.id as string,
      text: updates.text ?? (row.text as string),
      // row.vector 可能为 null（历史病行，2026-07-23 dream/auto-gc 生产实锤：
      // cc:82a919ea 连崩三轮 Array.from(null)）。null → [] 让 metadata-only update
      // 不再被病行绊倒；空向量本就不可检索，语义不变。
      vector: updates.vector ?? (row.vector ? Array.from(row.vector as Iterable<number>) : []),
      category: updates.category ?? (row.category as MemoryEntry["category"]),
      scope: rowScope,
      importance: updates.importance ?? Number(row.importance),
      timestamp: updates.timestamp ?? Number(row.timestamp),
      metadata: updates.metadata ?? ((row.metadata as string) || "{}"),
      language: updates.language ?? ((row.language as string) || "en"),
      fts_text: updates.fts_text ?? ((row.fts_text as string) || (updates.text ?? (row.text as string))),
    };

    // HP-emo: Re-detect emotion when text changes
    if (updates.text) {
      const emotionResult = detectEmotionIfEnabled(updates.text);
      if (emotionResult) {
        const meta = JSON.parse(updated.metadata || "{}");
        meta.emotion = emotionResult;
        updated.metadata = JSON.stringify(meta);
      }
    }

    // Atomic upsert via mergeInsert — replaces the previous delete+add two-step,
    // which could lose the row entirely on a crash between steps and exposed a
    // window where concurrent reads saw the id momentarily vanish.
    //
    // 2026-08-14: 包进 store-write 锁与 upsert/storeBatch 对齐 —— 此前 update 是唯一
    // 裸奔的写路径（dream 的 3a/3b/auto-gc 全走它），与其他进程的 mergeInsert 并发时
    // 靠 LanceDB 乐观重试硬扛，是 commit conflict 的隐性源之一。
    await withWriteLock("store-write", async () => {
      await this.table!
        .mergeInsert("id")
        .whenMatchedUpdateAll()
        .whenNotMatchedInsertAll()
        .execute([updated]);
    }, { expireMs: 30_000 });

    return updated;
  }

  /**
   * Serialized read-modify-write for the metadata column.
   *
   * Every direct `getById → mutate metadata → update` round-trip can lose
   * concurrent increments (last write wins). This method funnels metadata
   * patches for the same id through a per-id promise queue so each patch
   * reads the result of the previous one.
   *
   * patchFn receives the parsed metadata object plus the freshly-read entry
   * (for callers that need importance/timestamp context) and returns the new
   * metadata object. A throwing patchFn abandons that patch — the error
   * propagates to the caller, later queued patches still run.
   *
   * Scope of guarantee: in-process only. Cross-process writers (multiple
   * mcp-server instances) and the remaining direct update() callers
   * (consolidation/capture/forget/prospective/conflict engines, auto-gc)
   * are NOT serialized by this queue. Long-term direction: make this the
   * only legal metadata write path.
   */
  async patchMetadata(
    id: string,
    patchFn: (meta: Record<string, unknown>, entry: MemoryEntry) => Record<string, unknown>,
    scopeFilter?: string[],
  ): Promise<MemoryEntry | null> {
    await this.ensureInitialized();

    // Queue key must be the resolved full UUID — otherwise a prefix caller and
    // a full-id caller targeting the same row would race on separate queues.
    const uuidRegex = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i;
    let fullId: string;
    if (uuidRegex.test(id)) {
      fullId = id;
    } else {
      const resolved = await this.getById(id);
      if (!resolved) return null;
      fullId = resolved.id;
    }

    const run = async (): Promise<MemoryEntry | null> => {
      const entry = await this.getById(fullId);
      if (!entry) return null;
      let meta: Record<string, unknown>;
      try {
        const parsed: unknown = JSON.parse(entry.metadata || "{}");
        meta = parsed !== null && typeof parsed === "object" && !Array.isArray(parsed)
          ? (parsed as Record<string, unknown>)
          : {};
      } catch {
        meta = {};
      }
      const patched = patchFn(meta, entry);
      return this.update(fullId, { metadata: JSON.stringify(patched) }, scopeFilter);
    };

    const prev = this.metadataPatchQueues.get(fullId) ?? Promise.resolve();
    // Isolate prior failures so one rejected patch can't wedge the queue.
    const result = prev.then(run, run);
    // The stored tail settles regardless of outcome; the caller keeps the real
    // rejection via `result`. Only the current tail removes itself.
    const tail = result.then(
      () => undefined,
      () => undefined,
    ).then(() => {
      if (this.metadataPatchQueues.get(fullId) === tail) {
        this.metadataPatchQueues.delete(fullId);
      }
    });
    this.metadataPatchQueues.set(fullId, tail);
    return result;
  }

  /** Number of ids with in-flight metadata patches (test/diagnostic hook). */
  get pendingMetadataPatchCount(): number {
    return this.metadataPatchQueues.size;
  }

  /**
   * Batched metadata read-modify-write: one store-write lock acquisition, one
   * mergeInsert commit per batch of up to 200 rows — versus one commit per row
   * through update()/patchMetadata().
   *
   * 为什么存在（2026-08-14 dream 写爆库根治）：LanceDB 每次写 commit 生成一个新
   * manifest 版本文件，manifest 大小 ∝ 全表 fragment 数（实测 348-436KB/个）。dream
   * 的 3a/3b/auto-gc 全部逐条 update，一轮 --auto 写出 ~6600 个 commit ≈ 2GB+ 的
   * _versions 增量。本方法把"一批相关行的 metadata 变更"折成单次 commit。
   *
   * 语义要点：
   * - patchFn 在 **锁内、以库里最新行为底座** 应用 —— 不接受调用方预先物化的完整
   *   metadata 串。这是互审 R1 的 P0 修正：3b 持有的是 3a 之前 gather 的旧条目，
   *   预物化串批量写会把 3a 刚提交的 version_group/sourceMemories 再抹一遍
   *   （与它要修的 Bug-1 同病，只是从循环内搬到阶段间）。
   * - patches 按数组序在同一临界区内依次应用：同批的 patchFn 之间可通过调用方
   *   闭包传递决策（3a 靠它让 canonical 与 members 共享同一个 groupId）。
   * - 单条 patchFn 抛错或 id 不存在/越权：跳过该条继续（stderr 记一行），不炸整批 ——
   *   与 dream"单 scope 失败不阻断其他"的分级一致。返回成功写入数。
   * - 与 patchMetadata 的 per-id 队列并存：本方法自身持 store-write 锁做批内原子，
   *   跨进程窗口由锁消除；进程内与单条 patchMetadata 的交错等同于现状并发水平。
   */
  async patchMetadataBatch(
    patches: Array<{
      id: string;
      patchFn: (meta: Record<string, unknown>, entry: MemoryEntry) => Record<string, unknown>;
    }>,
    scopeFilter?: string[],
  ): Promise<number> {
    await this.ensureInitialized();
    if (patches.length === 0) return 0;

    let written = 0;
    // 分批：避免 OR 链过长（抄 getVectors 的范式，批更保守——每批一个 commit 本身就是收益）
    const BATCH = 200;
    for (let i = 0; i < patches.length; i += BATCH) {
      const batch = patches.slice(i, i + BATCH);
      written += await withWriteLock("store-write", async () => {
        // 锁内读最新行
        const conditions = batch.map(p => `id = '${escapeSqlLiteral(p.id)}'`).join(" OR ");
        const rows = await this.table!.query().where(conditions).limit(batch.length).toArray();
        const rowById = new Map<string, Record<string, unknown>>();
        for (const row of rows) rowById.set(row.id as string, row as Record<string, unknown>);

        const toWrite: MemoryEntry[] = [];
        for (const { id, patchFn } of batch) {
          const row = rowById.get(id);
          if (!row) continue; // 行不存在：跳过（dream 处理途中被并发删除属正常）
          const rowScope = (row.scope as string | undefined) ?? "";
          if (!matchesScopeFilter(rowScope, scopeFilter)) continue;
          let meta: Record<string, unknown>;
          try {
            const parsed: unknown = JSON.parse((row.metadata as string) || "{}");
            meta = parsed !== null && typeof parsed === "object" && !Array.isArray(parsed)
              ? (parsed as Record<string, unknown>)
              : {};
          } catch {
            meta = {};
          }
          const entry: MemoryEntry = {
            id: row.id as string,
            text: row.text as string,
            // null 向量防护同 update()：历史病行 → []，metadata-only 批量写不被绊倒
            vector: row.vector ? Array.from(row.vector as Iterable<number>) : [],
            category: row.category as MemoryEntry["category"],
            scope: rowScope,
            importance: Number(row.importance),
            timestamp: Number(row.timestamp),
            metadata: (row.metadata as string) || "{}",
            language: (row.language as string) || "en",
            fts_text: (row.fts_text as string) || (row.text as string),
          };
          try {
            const patched = patchFn(meta, entry);
            toWrite.push({ ...entry, metadata: JSON.stringify(patched) });
          } catch (err) {
            console.error(`[patchMetadataBatch] patchFn failed for ${id}, skipping: ${err instanceof Error ? err.message : String(err)}`);
          }
        }
        if (toWrite.length === 0) return 0;
        await this.table!.mergeInsert("id").whenMatchedUpdateAll().whenNotMatchedInsertAll().execute(toWrite);
        return toWrite.length;
      }, { expireMs: 30_000 });
    }
    return written;
  }

  /**
   * Dissolve version groups that have only one surviving member.
   *
   * The membership scan and metadata rewrite share the global store-write lock.
   * Keeping both inside one critical section matters: planning during auto-GC and
   * patching later could otherwise dismantle a group that gained a new member in
   * between. Hard deletes that bypass the store may still create a new singleton
   * after this pass; the next scheduled pass repairs it.
   */
  async repairSingletonVersionGroups(): Promise<number> {
    await this.ensureInitialized();

    return withWriteLock("store-write", async () => {
      const membershipRows = await this.table!.query()
        .select(["id", "metadata"])
        .toArray();
      const repairs = planSingletonVersionGroupRepairs(
        membershipRows.map(row => ({
          id: row.id as string,
          metadata: (row.metadata as string) || "{}",
        })),
      );
      if (repairs.length === 0) return 0;

      let repaired = 0;
      const BATCH = 200;
      for (let i = 0; i < repairs.length; i += BATCH) {
        const batch = repairs.slice(i, i + BATCH);
        const expectedById = new Map(batch.map(item => [item.id, item.groupId]));
        const conditions = batch
          .map(item => `id = '${escapeSqlLiteral(item.id)}'`)
          .join(" OR ");
        const rows = await this.table!.query()
          .where(conditions)
          .limit(batch.length)
          .toArray();

        const toWrite: MemoryEntry[] = [];
        for (const row of rows) {
          const id = row.id as string;
          const expectedGroupId = expectedById.get(id);
          if (!expectedGroupId) continue;
          let meta: Record<string, unknown>;
          try {
            const parsed: unknown = JSON.parse((row.metadata as string) || "{}");
            meta = parsed !== null && typeof parsed === "object" && !Array.isArray(parsed)
              ? (parsed as Record<string, unknown>)
              : {};
          } catch {
            continue;
          }
          // The group id is rechecked against the locked, current row.
          if (meta.version_group !== expectedGroupId) continue;
          clearVersionGroupMetadata(meta, expectedGroupId);
          toWrite.push({
            id,
            text: row.text as string,
            vector: row.vector ? Array.from(row.vector as Iterable<number>) : [],
            category: row.category as MemoryEntry["category"],
            scope: (row.scope as string | undefined) ?? "",
            importance: Number(row.importance),
            timestamp: Number(row.timestamp),
            metadata: JSON.stringify(meta),
            language: (row.language as string) || "en",
            fts_text: (row.fts_text as string) || (row.text as string),
          });
        }
        if (toWrite.length === 0) continue;
        // Matched rows only: a concurrent hard delete must never be resurrected.
        const result = await this.table!.mergeInsert("id").whenMatchedUpdateAll().execute(toWrite);
        repaired += result.numUpdatedRows;
      }
      return repaired;
    }, { expireMs: 30 * 60_000 });
  }

  async bulkDelete(scopeFilter: string[], beforeTimestamp?: number): Promise<number> {
    await this.ensureInitialized();

    const conditions: string[] = [];

    if (scopeFilter.length > 0) {
      // family 模式（历史行为）。⚠️ 这是**破坏性**操作走前缀：`bulkDelete(["memory"])`
      // 会连 `memory:pivot` 一起删。当前无生产 caller；要加 caller 请先显式传 exact
      // （同 forgetByScope 的处理，2026-08-16）。
      conditions.push(`(${scopeWhereClause(scopeFilter)})`);
    }

    if (beforeTimestamp) {
      conditions.push(`timestamp < ${beforeTimestamp}`);
    }

    if (conditions.length === 0) {
      throw new Error("Bulk delete requires at least scope or timestamp filter for safety");
    }

    const whereClause = conditions.join(" AND ");

    // Count first
    const countResults = await this.table!.query().where(whereClause).toArray();
    const deleteCount = countResults.length;

    // Then delete
    if (deleteCount > 0) {
      await this.table!.delete(whereClause);
    }

    return deleteCount;
  }

  /**
   * Force lazy initialization to complete.
   *
   * `hasFtsSupport` is meaningless before init: `ftsIndexCreated` starts false and
   * only flips once `doInitialize()` has attempted to build the FTS index. Any caller
   * that *branches* on `hasFtsSupport` must await this first, or it will read the
   * pre-init false and take the wrong branch. Cheap to call repeatedly —
   * `ensureInitialized()` returns immediately once the table exists.
   */
  async ready(): Promise<void> {
    await this.ensureInitialized();
  }

  get hasFtsSupport(): boolean {
    return this.ftsIndexCreated;
  }
}