Verify the FTS language on the serving path before the first query

Migration-time verification only protects hosts that run the migration
runner at boot; one that migrated once and later changed FTS_LANGUAGE
would silently degrade recall with nothing rechecking. The knowledge
plane now awaits the same read-only catalog check, memoized, before
its first search or capture — the fail-fast validate step (Hibernate
validate, Rails check_all_pending!) adapted to a synchronous mount.
Success is remembered forever; a failure clears the memo so a
transient connection error at first use does not poison the plane
until restart. The verifier is exported for hosts that prefer running
it from a real readiness probe.

Author: TheGreatAxios

IMPLEMENTATION.md

@@ -131,10 +131,13 @@ An ordered slice of a version's text, keyed by `(version_id, ordinal)`
 that powers the lexical search channel — this is the only place FTS is
 computed; no separate FTS table exists. Its language comes from
 `FTS_LANGUAGE` at migration time; the query side binds the same configured
-language as a `regconfig` parameter, and `runKnowledgeMigrations` verifies
-the column's actual language against the configuration (read back from the
-catalog) so a mismatch fails at startup instead of silently degrading
-recall. Chunks are **never** reused across
+language as a `regconfig` parameter. The invariant is verified twice, both
+read-only against the catalog: `runKnowledgeMigrations` checks after
+applying (the deploy step), and the knowledge plane runs the same check
+once, memoized, before its first query (the serving path) — so a mismatch
+or unmigrated schema fails loudly on first use regardless of who ran the
+migrations. Hosts with a readiness probe can call the exported
+`verifyFtsLanguage` there instead. Chunks are **never** reused across
 versions — every new version gets a fresh full insert of its own chunks.
 
 ### `knowledge_entity` / `knowledge_edge`

src/core/fts-language.test.ts

@@ -2,6 +2,7 @@ import { describe, expect, it } from "bun:test";
 import { readFile } from "node:fs/promises";
 import { join } from "node:path";
 import {
+  createFtsVerification,
   DEFAULT_FTS_LANGUAGE,
   FTS_LANGUAGE_TOKEN,
   parseFtsLanguage,
@@ -91,3 +92,108 @@ describe("verifyFtsLanguage", () => {
     ).rejects.toThrow("schema not migrated");
   });
 });
+
+describe("createFtsVerification", () => {
+  const ENGLISH_EXPR = "to_tsvector('english'::regconfig, text)";
+
+  it("verifies once and memoizes success", async () => {
+    let calls = 0;
+    const ensure = createFtsVerification(
+      {
+        query: (sqlText: string) => {
+          calls += 1;
+          return Promise.resolve(
+            sqlText.includes("pg_ts_config") ? [{ ok: 1 }] : [{ expr: ENGLISH_EXPR }],
+          );
+        },
+      },
+      "english",
+    );
+    await ensure();
+    await ensure();
+    // one pg_ts_config lookup + one catalog read, never repeated
+    expect(calls).toBe(2);
+  });
+
+  it("clears the memo on failure so a transient error does not poison the plane", async () => {
+    let attempt = 0;
+    const ensure = createFtsVerification(
+      {
+        query: (sqlText: string) => {
+          if (attempt === 0 && sqlText.includes("pg_ts_config")) {
+            attempt += 1;
+            return Promise.reject(new Error("connection refused"));
+          }
+          return Promise.resolve(
+            sqlText.includes("pg_ts_config") ? [{ ok: 1 }] : [{ expr: ENGLISH_EXPR }],
+          );
+        },
+      },
+      "english",
+    );
+    await expect(ensure()).rejects.toThrow("connection refused");
+    await ensure();
+  });
+
+  it("keeps throwing on a real mismatch", async () => {
+    const ensure = createFtsVerification(
+      {
+        query: (sqlText: string) =>
+          Promise.resolve(
+            sqlText.includes("pg_ts_config") ? [{ ok: 1 }] : [{ expr: ENGLISH_EXPR }],
+          ),
+      },
+      "german",
+    );
+    await expect(ensure()).rejects.toThrow('built with "english"');
+    await expect(ensure()).rejects.toThrow('built with "english"');
+  });
+});
+
+describe("createFtsVerification concurrency", () => {
+  const ENGLISH_EXPR = "to_tsvector('english'::regconfig, text)";
+
+  it("concurrent first calls share one in-flight verification", async () => {
+    let calls = 0;
+    let release!: () => void;
+    const gate = new Promise<void>((r) => (release = r));
+    const ensure = createFtsVerification(
+      {
+        query: async (sqlText: string) => {
+          calls += 1;
+          await gate;
+          return sqlText.includes("pg_ts_config") ? [{ ok: 1 }] : [{ expr: ENGLISH_EXPR }];
+        },
+      },
+      "english",
+    );
+    const a = ensure();
+    const b = ensure();
+    release();
+    await Promise.all([a, b]);
+    expect(calls).toBe(2);
+  });
+
+  it("concurrent callers all see a shared failure, then a retry succeeds", async () => {
+    let attempt = 0;
+    const ensure = createFtsVerification(
+      {
+        query: (sqlText: string) => {
+          if (attempt === 0 && sqlText.includes("pg_ts_config")) {
+            attempt += 1;
+            return Promise.reject(new Error("connection refused"));
+          }
+          return Promise.resolve(
+            sqlText.includes("pg_ts_config") ? [{ ok: 1 }] : [{ expr: ENGLISH_EXPR }],
+          );
+        },
+      },
+      "english",
+    );
+    const a = ensure();
+    const b = ensure();
+    await expect(a).rejects.toThrow("connection refused");
+    await expect(b).rejects.toThrow("connection refused");
+    await ensure();
+  });
+});

src/core/fts-language.ts

@@ -30,7 +30,25 @@ export function parseFtsLanguage(raw: string | undefined): string {
   return raw;
 }
 
-interface FtsVerifySqlClient {
+/**
+ * Memoized serving-path wrapper around verifyFtsLanguage: the first call
+ * verifies, success is remembered forever, and a failure clears the memo so
+ * a transient DB error at first use does not poison the plane until
+ * restart (a real mismatch keeps failing on every retry anyway).
+ */
+export function createFtsVerification(
+  client: FtsVerifySqlClient,
+  ftsLanguage: string,
+): () => Promise<void> {
+  let verified: Promise<void> | undefined;
+  return () =>
+    (verified ??= verifyFtsLanguage(client, ftsLanguage).catch((err) => {
+      verified = undefined;
+      throw err;
+    }));
+}
+
+export interface FtsVerifySqlClient {
   query: (sql: string, params: readonly unknown[]) => Promise<Array<Record<string, unknown>>>;
 }
 

src/index.ts

@@ -27,6 +27,11 @@ export type { KnowledgePlane } from "./knowledge.ts";
 export { CaptureLog, type CaptureEvent } from "./capture-log.ts";
 // Migrations
 export { runKnowledgeMigrations } from "./migrations.ts";
+export {
+  type FtsVerifySqlClient,
+  parseFtsLanguage,
+  verifyFtsLanguage,
+} from "./core/fts-language.ts";
 // Granular mount (compose your own)
 export { mountKnowledgeRoutes, type GrantConfig } from "./routes/mount.ts";
 

src/knowledge.ts

@@ -5,6 +5,8 @@
 import type { EngineConfig } from "./config.ts";
 import { log } from "./log.ts";
 import { createDb, type Db, type RawSql } from "./db/client.ts";
+import { createFtsVerification } from "./core/fts-language.ts";
+import { createEmbedRegistrySqlClient } from "./core/embed-sql.ts";
 import type { VisibilitySpec } from "./core/schemas/document.ts";
 import { captureDocument } from "./services/capture.ts";
 import {
@@ -57,9 +59,21 @@ export function createKnowledgePlane(config: KnowledgeConfig): KnowledgePlane {
   const { db, sql }: { db: Db; sql: RawSql } = createDb(engineConfig);
   const deps = { db, sql, config: engineConfig };
 
+  // Serving-path schema validation, industry-standard fail-fast shape
+  // (Hibernate validate / Rails check_all_pending!): the mount is
+  // synchronous, so "before accepting traffic" becomes a memoized check
+  // awaited by the first query. Read-only; migration stays a deploy step.
+  // Hosts with a real readiness probe can call verifyFtsLanguage there
+  // instead — this memo then resolves against an already-verified schema.
+  const ensureVerified = createFtsVerification(
+    createEmbedRegistrySqlClient(sql),
+    engineConfig.ftsLanguage,
+  );
+
   return {
     async search(params) {
       try {
+        await ensureVerified();
         const result = await hybridSearch(deps, {
           tenantId: params.tenantId,
           principalId: params.principalId,
@@ -114,6 +128,7 @@ export function createKnowledgePlane(config: KnowledgeConfig): KnowledgePlane {
     },
 
     async capture(params) {
+      await ensureVerified();
       const adapter = params.adapter ?? "mcp";
       const externalRef =
         params.externalRef ??

src/migrations.ts

@@ -11,6 +11,7 @@ import {
   parseFtsLanguage,
   verifyFtsLanguage,
 } from "./core/fts-language.ts";
+import { createEmbedRegistrySqlClient } from "./core/embed-sql.ts";
 
 const MIGRATIONS_DIR = join(import.meta.dir, "..", "migrations");
 
@@ -58,15 +59,7 @@ export async function runKnowledgeMigrations(
     // generated column was actually built with; a previously-migrated
     // database under a different language must fail loudly here, not
     // degrade recall silently at query time.
-    await verifyFtsLanguage(
-      {
-        query: async (text, params) =>
-          (await sql.unsafe(text, [...params] as never[])) as unknown as Array<
-            Record<string, unknown>
-          >,
-      },
-      ftsLanguage,
-    );
+    await verifyFtsLanguage(createEmbedRegistrySqlClient(sql), ftsLanguage);
   } finally {
     await sql.end();
   }