[TEST] bugfix/ladybug — support knowledgeIds scoping and optimize query execution

packages/ladybug/README.md

@@ -20,7 +20,7 @@ The package owns:
 - Optimized File-node Bulk Upsert (`bulkUpsertFiles`) — maps files to Parquet rows, writes them to temporary files on disk, and executes single-transaction `DELETE` and SQL `COPY FROM` commands.
 - File-node Snapshotting (`snapshotFilesToVersion`) — copies live files to snapshots before updates.
 - Concept-graph writes (`upsertConcept`, `attachFileToConcept`, `upsertTestsEdge`, `upsertContract`, `attachFileToContract`, `upsertGuidepost`, `attachGuidepost`) — `:Concept` / `:Contract` / `:Guidepost` nodes plus their file edges, matching the neo4j provider's merge policy so the ConceptGraphStrategy persists identically on either backend.
-- Stubbed read-side search (`src/search.ts`) — `IGraphSearchRepository` methods are declared but throw `"Ladybug search not implemented yet"`. The interface keeps types honest under a Ladybug-only deployment; real implementations (LadybugDB-native column lookups) land in a follow-up PR. Until then, MCP smart_search / keyword_lookup / list_knowledge / retrieve_file(metadata) will fail loudly rather than return wrong results when the active graph provider is Ladybug.
+- Read-side search (`src/search/`) — LadybugDB-native implementations of the `IGraphSearchRepository` surface: `runSmartSearchChannel` (fused per-channel scoring over `File` text, paths, keywords, classes/functions, and internal/external imports), `keywordLookup`, `listKnowledgeBases`, `fetchFileMetadata`, and `fetchRepoNames`. These back MCP smart_search / keyword_lookup / list_knowledge / retrieve_file(metadata) when the active graph provider is Ladybug. They are not named package exports — they are wired into the registered provider's `search` namespace in `src/provider.ts`.
 
 ## Public exports
 
@@ -31,7 +31,9 @@ function pingLadybug(): Promise<PingResult>;
 
 function upsertKnowledgeNode(doc: KnowledgeDoc): Promise<void>;
 function setKnowledgeStateInGraph(knowledgeId: string, state: KnowledgeState): Promise<void>;
+function setKnowledgeBranchInGraph(knowledgeId: string, branch: string): Promise<void>;
 function deleteKnowledgeGraph(knowledgeId: string): Promise<void>;
+function vacuumOrphanEntities(): Promise<void>;
 function upsertFileNode(input: UpsertFileNodeInput): Promise<void>;
 function bulkUpsertFiles(knowledgeId: string, fileStream: AsyncIterable<UpsertFileNodeInput>): Promise<void>;
 function deleteFileNodes(knowledgeId: string, paths: string[]): Promise<void>;

packages/ladybug/src/client.ts

@@ -1,6 +1,7 @@
 import { Database, Connection, PreparedStatement, type LbugValue } from "@ladybugdb/core";
 import { getConfigValue } from "@bb/config";
 import { Config } from "@bb/types";
+import { ensureSchema } from "./schema.ts";
 
 export interface PingResult {
   ok: boolean;
@@ -46,183 +47,35 @@ async function doConnect(): Promise<void> {
   }
 }
 
-async function ensureSchema(c: Connection): Promise<void> {
-  const nodeTables = [
-    `CREATE NODE TABLE Knowledge (
-      knowledgeId STRING PRIMARY KEY,
-      createdAt STRING,
-      sourceKind STRING,
-      sourceUrl STRING,
-      branch STRING,
-      repoName STRING,
-      state STRING,
-      updatedAt STRING
-    )`,
-    `CREATE NODE TABLE Repo (
-      id STRING PRIMARY KEY,
-      orgId STRING,
-      knowledgeId STRING,
-      repoId STRING,
-      repoUrl STRING,
-      branch STRING,
-      purpose STRING,
-      summary STRING,
-      architecture STRING,
-      dataFlow STRING,
-      majorSubsystems STRING[],
-      keyPatterns STRING[],
-      updatedAt STRING
-    )`,
-    `CREATE NODE TABLE Folder (
-      id STRING PRIMARY KEY,
-      orgId STRING,
-      knowledgeId STRING,
-      repoId STRING,
-      folderPath STRING,
-      purpose STRING,
-      summary STRING,
-      dependencyGraph STRING,
-      updatedAt STRING
-    )`,
-    `CREATE NODE TABLE File (
-      id STRING PRIMARY KEY,
-      orgId STRING,
-      knowledgeId STRING,
-      repoId STRING,
-      relativePath STRING,
-      language STRING,
-      sha STRING,
-      sizeBytes INT64,
-      purpose STRING,
-      summary STRING,
-      businessContext STRING,
-      dataFlowDirection STRING,
-      ontologyConcepts STRING[],
-      businessEntities STRING[],
-      systemCapabilities STRING[],
-      sideEffects STRING[],
-      configDependencies STRING[],
-      integrationSurface STRING[],
-      contractsProvided STRING[],
-      contractsConsumed STRING[],
-      sectionNames STRING[],
-      sectionDescriptions STRING[],
-      isBigFile BOOLEAN,
-      totalChunks INT64,
-      totalTokenCount INT64,
-      updatedAt STRING
-    )`,
-    `CREATE NODE TABLE FileVersion (
-      id STRING PRIMARY KEY,
-      knowledgeId STRING,
-      relativePath STRING,
-      commitHash STRING,
-      language STRING,
-      sha STRING,
-      sizeBytes INT64,
-      purpose STRING,
-      summary STRING,
-      businessContext STRING,
-      dataFlowDirection STRING,
-      ontologyConcepts STRING[],
-      businessEntities STRING[],
-      systemCapabilities STRING[],
-      sideEffects STRING[],
-      configDependencies STRING[],
-      integrationSurface STRING[],
-      contractsProvided STRING[],
-      contractsConsumed STRING[],
-      sectionNames STRING[],
-      sectionDescriptions STRING[],
-      snapshotAt STRING
-    )`,
-    `CREATE NODE TABLE Keyword (
-      name STRING PRIMARY KEY
-    )`,
-    `CREATE NODE TABLE Class (
-      signature STRING PRIMARY KEY
-    )`,
-    `CREATE NODE TABLE Function (
-      signature STRING PRIMARY KEY
-    )`,
-    `CREATE NODE TABLE Module (
-      name STRING PRIMARY KEY
-    )`,
-    `CREATE NODE TABLE Concept (
-      id STRING PRIMARY KEY,
-      orgId STRING,
-      knowledgeId STRING,
-      slug STRING,
-      kind STRING,
-      name STRING,
-      rationale STRING,
-      enrichmentRunId STRING,
-      createdAt STRING,
-      updatedAt STRING
-    )`,
-    `CREATE NODE TABLE Contract (
-      id STRING PRIMARY KEY,
-      orgId STRING,
-      knowledgeId STRING,
-      slug STRING,
-      kind STRING,
-      name STRING,
-      enrichmentRunId STRING,
-      createdAt STRING,
-      updatedAt STRING
-    )`,
-    `CREATE NODE TABLE Guidepost (
-      id STRING PRIMARY KEY,
-      orgId STRING,
-      knowledgeId STRING,
-      slug STRING,
-      kind STRING,
-      note STRING,
-      area STRING,
-      enrichmentRunId STRING,
-      createdAt STRING,
-      updatedAt STRING
-    )`,
-  ];
-
-  const relTables = [
-    `CREATE REL TABLE HAS_REPO (FROM Knowledge TO Repo)`,
-    `CREATE REL TABLE HAS_FILE (FROM Knowledge TO File)`,
-    `CREATE REL TABLE CONTAINS (FROM Repo TO Folder, FROM Folder TO Folder, FROM Folder TO File)`,
-    `CREATE REL TABLE HAS_KEYWORD (FROM File TO Keyword, FROM Folder TO Keyword, FROM Repo TO Keyword)`,
-    `CREATE REL TABLE HAS_CLASS (FROM File TO Class)`,
-    `CREATE REL TABLE HAS_FUNCTION (FROM File TO Function)`,
-    `CREATE REL TABLE HAS_IMPORT_INTERNAL (FROM File TO Module)`,
-    `CREATE REL TABLE HAS_IMPORT_EXTERNAL (FROM File TO Module)`,
-    `CREATE REL TABLE HAS_VERSION (FROM File TO FileVersion)`,
-    `CREATE REL TABLE HAS_CONCEPT (FROM File TO Concept, enrichmentRunId STRING, createdAt STRING, updatedAt STRING)`,
-    `CREATE REL TABLE PLAYS_ROLE (FROM File TO Concept, enrichmentRunId STRING, createdAt STRING, updatedAt STRING)`,
-    `CREATE REL TABLE BELONGS_TO_DOMAIN (FROM File TO Concept, enrichmentRunId STRING, createdAt STRING, updatedAt STRING)`,
-    `CREATE REL TABLE TESTS (FROM File TO File, enrichmentRunId STRING, createdAt STRING, updatedAt STRING)`,
-    `CREATE REL TABLE DEFINES (FROM File TO Contract, enrichmentRunId STRING, createdAt STRING, updatedAt STRING)`,
-    `CREATE REL TABLE CONSUMES (FROM File TO Contract, enrichmentRunId STRING, createdAt STRING, updatedAt STRING)`,
-    `CREATE REL TABLE ABOUT (FROM Guidepost TO File, FROM Guidepost TO Concept, FROM Guidepost TO Contract, enrichmentRunId STRING, createdAt STRING, updatedAt STRING)`,
-  ];
-
-  for (const q of [...nodeTables, ...relTables]) {
-    try {
-      await c.query(q);
-    } catch (e: unknown) {
-      const msg = e instanceof Error ? e.message : String(e);
-      if (
-        !msg.includes("already exists") &&
-        !msg.includes("table already exists") &&
-        !msg.includes("Binder exception")
-      ) {
-        throw e;
-      }
-    }
-  }
-}
-
 export async function closeLadybug(): Promise<void> {
+  // Drop cached prepared statements first: each is bound to the connection we
+  // are about to close, so reusing one after a reconnect would execute against
+  // a destroyed handle. (`__resetForTests` already does this; the production
+  // close path must too.)
+  preparedCache.clear();
+
+  const openConn = conn;
+  const openDb = db;
   conn = null;
   db = null;
+
+  // Release the native handles so the on-disk file lock is freed and pending
+  // writes are flushed. Best-effort: a failed connection close must not block
+  // the database close.
+  try {
+    if (openConn !== null) {
+      await openConn.close();
+    }
+  } catch {
+    // ignore
+  }
+  try {
+    if (openDb !== null) {
+      await openDb.close();
+    }
+  } catch {
+    // ignore
+  }
 }
 
 export async function pingLadybug(): Promise<PingResult> {
@@ -260,31 +113,54 @@ export function _getConnection(): Connection {
 //   return rows as T[];
 // }
 
-// Add a global cache map at the top of client.ts
+// Cache compiled plans so repeated, parameter-stable queries (the MERGE /
+// DELETE / search builders) only pay preparation cost once. Keyed by query
+// text — see `_runCypherOnce` for queries whose text is unique per call.
 const preparedCache = new Map<string, PreparedStatement>();
 
+async function executePrepared<T>(
+  c: Connection,
+  prepared: PreparedStatement,
+  params: Record<string, LbugValue>,
+): Promise<T[]> {
+  const result = await c.execute(prepared, params);
+  const singleResult = Array.isArray(result) ? result[0] : result;
+  if (!singleResult) {
+    throw new Error("No query result returned from LadybugDB");
+  }
+  const rows = await singleResult.getAll();
+  return rows as T[];
+}
+
 export async function _runCypher<T = unknown>(query: string, params: Record<string, LbugValue> = {}): Promise<T[]> {
   const c = _getConnection();
 
-  // 1. Check if the query has already been compiled and compiled plan is cached
+  // Reuse the compiled plan if we have seen this exact query before.
   let prepared = preparedCache.get(query);
-
   if (!prepared) {
     prepared = await c.prepare(query);
     if (!prepared.isSuccess()) {
       throw new Error(`Failed to prepare query: ${prepared.getErrorMessage()}`);
     }
-    // 2. Store it for future iterations in the ingest loop
     preparedCache.set(query, prepared);
   }
 
-  const result = await c.execute(prepared, params);
-  const singleResult = Array.isArray(result) ? result[0] : result;
-  if (!singleResult) {
-    throw new Error("No query result returned from LadybugDB");
+  return executePrepared<T>(c, prepared, params);
+}
+
+// One-shot execution for queries whose text is unique per call — notably the
+// `COPY <table> FROM '<temp path>'` commands in `bulkUpsertFiles`, whose
+// embedded parquet path differs every time. Caching those would grow
+// `preparedCache` without bound (one dead native PreparedStatement per call),
+// defeating the cache and leaking memory. Prepare, run, and let the statement
+// be collected.
+export async function _runCypherOnce<T = unknown>(query: string, params: Record<string, LbugValue> = {}): Promise<T[]> {
+  const c = _getConnection();
+  const prepared = await c.prepare(query);
+  if (!prepared.isSuccess()) {
+    throw new Error(`Failed to prepare query: ${prepared.getErrorMessage()}`);
   }
-  const rows = await singleResult.getAll();
-  return rows as T[];
+  return executePrepared<T>(c, prepared, params);
 }
 
 // Clear the cache if tests reset

packages/ladybug/src/files.ts

@@ -1,4 +1,4 @@
-import { _runCypher } from "./client.ts";
+import { _runCypher, _runCypherOnce } from "./client.ts";
 import { ParquetSchema, ParquetWriter } from "parquetjs";
 import { join } from "node:path";
 import { unlinkSync } from "node:fs";
@@ -248,30 +248,32 @@ export async function bulkUpsertFiles(
       );
     }
 
-    // Execute COPY FROM commands exactly once
+    // Execute COPY FROM commands exactly once. These embed a unique temp-file
+    // path per call, so run them uncached (see `_runCypherOnce`) to avoid
+    // leaking a prepared statement per ingestion.
     if (fileCount > 0) {
-      await _runCypher(`COPY File FROM '${fileWriterInfo.path}'`);
+      await _runCypherOnce(`COPY File FROM '${fileWriterInfo.path}'`);
     }
     if (hasFileCount > 0) {
-      await _runCypher(`COPY HAS_FILE FROM '${hasFileRelWriterInfo.path}'`);
+      await _runCypherOnce(`COPY HAS_FILE FROM '${hasFileRelWriterInfo.path}'`);
     }
     if (containsCount > 0) {
-      await _runCypher(`COPY CONTAINS FROM '${containsRelWriterInfo.path}' (FROM='Folder', TO='File')`);
+      await _runCypherOnce(`COPY CONTAINS FROM '${containsRelWriterInfo.path}' (FROM='Folder', TO='File')`);
     }
     if (keywordCount > 0) {
-      await _runCypher(`COPY HAS_KEYWORD FROM '${hasKeywordRelWriterInfo.path}' (FROM='File', TO='Keyword')`);
+      await _runCypherOnce(`COPY HAS_KEYWORD FROM '${hasKeywordRelWriterInfo.path}' (FROM='File', TO='Keyword')`);
     }
     if (classCount > 0) {
-      await _runCypher(`COPY HAS_CLASS FROM '${hasClassRelWriterInfo.path}'`);
+      await _runCypherOnce(`COPY HAS_CLASS FROM '${hasClassRelWriterInfo.path}'`);
     }
     if (functionCount > 0) {
-      await _runCypher(`COPY HAS_FUNCTION FROM '${hasFunctionRelWriterInfo.path}'`);
+      await _runCypherOnce(`COPY HAS_FUNCTION FROM '${hasFunctionRelWriterInfo.path}'`);
     }
     if (importIntCount > 0) {
-      await _runCypher(`COPY HAS_IMPORT_INTERNAL FROM '${hasImportInternalRelWriterInfo.path}'`);
+      await _runCypherOnce(`COPY HAS_IMPORT_INTERNAL FROM '${hasImportInternalRelWriterInfo.path}'`);
     }
     if (importExtCount > 0) {
-      await _runCypher(`COPY HAS_IMPORT_EXTERNAL FROM '${hasImportExternalRelWriterInfo.path}'`);
+      await _runCypherOnce(`COPY HAS_IMPORT_EXTERNAL FROM '${hasImportExternalRelWriterInfo.path}'`);
     }
   } finally {
     for (const p of tempPaths) {