fix(embeddings): version-stamp embeddings and batch the sync embed path

db/migrations/20260526120000_event_embeddings_model_stamp.sql

@@ -0,0 +1,111 @@
+-- migrate:up
+
+-- Version-stamp every embedding with the model that produced it. Without this,
+-- swapping EMBEDDINGS_MODEL to a different model of the SAME dimensionality
+-- silently mixes incompatible vector spaces in event_embeddings with no way to
+-- detect or segregate the mismatched rows. The stamp lets future similarity
+-- queries scope to a single model and makes a model swap auditable.
+--
+-- NULL = produced before this column existed (legacy rows, unknown model).
+ALTER TABLE public.event_embeddings ADD COLUMN IF NOT EXISTS embedding_model text;
+
+COMMENT ON COLUMN public.event_embeddings.embedding_model IS 'Model/version stamp of the embedding model that produced this vector (e.g. "Xenova/bge-base-en-v1.5"). NULL = legacy row written before stamping. Vectors from different stamps are NOT comparable even at equal dimensionality.';
+
+-- Expose the stamp through current_event_records so similarity queries on the
+-- view can scope to a single model. Appended at the end so CREATE OR REPLACE
+-- keeps the existing column order; otherwise byte-identical to the baseline.
+CREATE OR REPLACE VIEW public.current_event_records AS
+ SELECT e.id,
+    e.organization_id,
+    e.entity_ids,
+    e.origin_id,
+    e.title,
+    e.payload_type,
+    e.payload_text,
+    e.payload_data,
+    e.payload_template,
+    e.attachments,
+    e.metadata,
+    e.score,
+    emb.embedding,
+    e.author_name,
+    e.source_url,
+    e.occurred_at,
+    e.created_at,
+    e.origin_parent_id,
+    COALESCE(length(e.payload_text), 0) AS content_length,
+    e.search_tsv,
+    e.origin_type,
+    e.connector_key,
+    e.connection_id,
+    e.feed_key,
+    e.feed_id,
+    e.run_id,
+    e.semantic_type,
+    e.client_id,
+    e.created_by,
+    e.interaction_type,
+    e.interaction_status,
+    e.interaction_input_schema,
+    e.interaction_input,
+    e.interaction_output,
+    e.interaction_error,
+    e.supersedes_event_id,
+    emb.embedding_model
+   FROM (public.events e
+     LEFT JOIN public.event_embeddings emb ON ((emb.event_id = e.id)))
+  WHERE (NOT (EXISTS ( SELECT 1
+           FROM public.events newer
+          WHERE (newer.supersedes_event_id = e.id))));
+
+-- migrate:down
+
+-- CREATE OR REPLACE VIEW cannot REMOVE a column from an existing view (Postgres
+-- only allows appending columns at the end). Drop and recreate
+-- current_event_records WITHOUT embedding_model; the recreated view no longer
+-- references the column, so the subsequent DROP COLUMN succeeds.
+DROP VIEW IF EXISTS public.current_event_records;
+CREATE VIEW public.current_event_records AS
+ SELECT e.id,
+    e.organization_id,
+    e.entity_ids,
+    e.origin_id,
+    e.title,
+    e.payload_type,
+    e.payload_text,
+    e.payload_data,
+    e.payload_template,
+    e.attachments,
+    e.metadata,
+    e.score,
+    emb.embedding,
+    e.author_name,
+    e.source_url,
+    e.occurred_at,
+    e.created_at,
+    e.origin_parent_id,
+    COALESCE(length(e.payload_text), 0) AS content_length,
+    e.search_tsv,
+    e.origin_type,
+    e.connector_key,
+    e.connection_id,
+    e.feed_key,
+    e.feed_id,
+    e.run_id,
+    e.semantic_type,
+    e.client_id,
+    e.created_by,
+    e.interaction_type,
+    e.interaction_status,
+    e.interaction_input_schema,
+    e.interaction_input,
+    e.interaction_output,
+    e.interaction_error,
+    e.supersedes_event_id
+   FROM (public.events e
+     LEFT JOIN public.event_embeddings emb ON ((emb.event_id = e.id)))
+  WHERE (NOT (EXISTS ( SELECT 1
+           FROM public.events newer
+          WHERE (newer.supersedes_event_id = e.id))));
+
+ALTER TABLE public.event_embeddings DROP COLUMN IF EXISTS embedding_model;

packages/connector-worker/src/__tests__/embeddings-model-stamp.test.ts

@@ -0,0 +1,39 @@
+/**
+ * Finding #3 reproducer (part a): embeddings must be version-stamped so a
+ * same-dimension model swap can't silently mix incompatible vector spaces.
+ *
+ * `resolveServiceModel` is the guard that `fetchEmbeddingsFromService` applies
+ * to every service response. Asserted directly (pure function) so the check is
+ * immune to bun's process-global `mock.module` of '../embeddings.js' in the
+ * sibling executor tests.
+ *
+ *   - a service `model` that differs from the worker's expectation throws
+ *     (fail loud) — the exact silent-mixing case, even at equal dimensionality;
+ *   - a matching `model` resolves to that stamp;
+ *   - an omitted `model` falls back to the worker's expectation.
+ *
+ * Part (b) — the stamp is mapped onto each streamed event — is asserted in
+ * executor-batch-embed.test.ts; persistence onto event_embeddings is asserted
+ * server-side in insert-event-embedding-model.test.ts.
+ */
+
+import { describe, expect, test } from 'bun:test';
+import { resolveServiceModel } from '../embeddings-model.js';
+
+const EXPECTED = 'Xenova/bge-base-en-v1.5';
+
+describe('embeddings model version stamp guard (Finding #3)', () => {
+  test('rejects a same-dimension model mismatch (fail loud)', () => {
+    expect(() => resolveServiceModel('some-other-model-v2', EXPECTED)).toThrow(
+      /returned model 'some-other-model-v2' but this worker expects/
+    );
+  });
+
+  test('resolves to the service model on a match', () => {
+    expect(resolveServiceModel(EXPECTED, EXPECTED)).toBe(EXPECTED);
+  });
+
+  test('falls back to the expected model when the service omits one', () => {
+    expect(resolveServiceModel(undefined, EXPECTED)).toBe(EXPECTED);
+  });
+});

packages/connector-worker/src/__tests__/executor-batch-embed.test.ts

@@ -0,0 +1,140 @@
+/**
+ * Finding #12 reproducer: the sync embedding path must batch a whole event
+ * chunk into ONE embedding call (one HTTP round-trip / vectorized pass), not
+ * one call per event, while still mapping each vector back to its source event.
+ *
+ * Strategy: mock `executeCompiledConnector` to invoke `onEventChunk` with a
+ * 3-event chunk, mock `batchGenerateEmbeddings` to return distinguishable
+ * vectors, and assert:
+ *   - batchGenerateEmbeddings was called exactly ONCE (not 3x),
+ *   - it received all 3 chunk texts in one call,
+ *   - each streamed ContentItem carries the vector for its own text + the model
+ *     stamp,
+ *   - an event with empty text gets no embedding (per-event association held).
+ */
+
+import { afterEach, beforeEach, describe, expect, mock, test } from 'bun:test';
+
+// biome-ignore lint/suspicious/noExplicitAny: test seam — module mocks need loose types
+type AnyFn = (...args: any[]) => any;
+
+// Return a vector derived from the text so we can assert per-event mapping.
+const batchGenerateEmbeddingsMock = mock<AnyFn>(async (texts: string[]) => ({
+  embeddings: texts.map((t) => [t.length, 0, 0]),
+  model: 'stub-model-v1',
+}));
+
+let capturedHooks: { onEventChunk: (events: unknown[]) => Promise<void> } | undefined;
+
+const executeCompiledConnectorMock = mock<AnyFn>(async (args: { hooks: typeof capturedHooks }) => {
+  capturedHooks = args.hooks;
+  return { mode: 'sync', checkpoint: null };
+});
+
+mock.module('../executor/runtime.js', () => ({
+  executeCompiledConnector: executeCompiledConnectorMock,
+}));
+
+mock.module('../embeddings.js', () => ({
+  batchGenerateEmbeddings: batchGenerateEmbeddingsMock,
+  generateEmbedding: async () => [0, 0, 0],
+}));
+
+mock.module('../compile-connector.js', () => ({
+  compileConnectorFromFile: async () => 'compiled-code',
+  findBundledConnectorFile: () => '/fake/path',
+}));
+
+mock.module('../executor/subprocess.js', () => ({
+  SubprocessExecutor: class {
+    // biome-ignore lint/suspicious/noExplicitAny: test stub
+    constructor(_opts: any) {}
+  },
+}));
+
+import type { ContentItem } from '../daemon/client.js';
+import { executeRun } from '../daemon/executor.js';
+
+function makeStubClient() {
+  const streamed: ContentItem[] = [];
+  const client = {
+    id: 'test-worker',
+    version: 'test',
+    streamed,
+    async heartbeat() {},
+    async stream(batch: { items: ContentItem[] }) {
+      streamed.push(...batch.items);
+    },
+    async complete() {},
+    async completeAction() {},
+    async completeEmbeddings() {},
+    async completeAuth() {},
+    async emitAuthArtifact() {},
+    async pollAuthSignal() {
+      return { signal: null };
+    },
+    async fetchEventsForEmbedding() {
+      return [];
+    },
+  };
+  return client;
+}
+
+describe('sync embedding path batches per chunk (Finding #12)', () => {
+  beforeEach(() => {
+    capturedHooks = undefined;
+    executeCompiledConnectorMock.mockClear();
+    batchGenerateEmbeddingsMock.mockClear();
+  });
+
+  afterEach(() => {
+    batchGenerateEmbeddingsMock.mockClear();
+  });
+
+  test('one chunk of N events triggers exactly one batch call with all texts mapped back', async () => {
+    const client = makeStubClient();
+
+    executeCompiledConnectorMock.mockImplementationOnce(async (args: { hooks: typeof capturedHooks }) => {
+      capturedHooks = args.hooks;
+      // One chunk, three events. The third has empty text (no embeddable
+      // content) — it must still stream through, just without a vector.
+      await capturedHooks!.onEventChunk([
+        { origin_id: 'a', payload_text: 'aa', occurred_at: new Date(), origin_type: 'post' },
+        { origin_id: 'b', payload_text: 'bbbb', occurred_at: new Date(), origin_type: 'post' },
+        { origin_id: 'c', payload_text: '', title: '', occurred_at: new Date(), origin_type: 'post' },
+      ]);
+      return { mode: 'sync', checkpoint: null };
+    });
+
+    const job = {
+      run_id: 500,
+      run_type: 'sync',
+      connector_key: 'fake',
+      feed_key: 'feed',
+      compiled_code: 'compiled-code',
+      // biome-ignore lint/suspicious/noExplicitAny: minimal job shape
+    } as any;
+
+    // batchSize=10 so the chunk does not flush mid-loop; default generateEmbeddings=true.
+    // biome-ignore lint/suspicious/noExplicitAny: minimal env
+    const result = await executeRun(client as any, job, {} as any, { batchSize: 10 });
+    expect(result.error).toBeUndefined();
+
+    // (1) exactly ONE batch call for the whole chunk — not one per event.
+    expect(batchGenerateEmbeddingsMock).toHaveBeenCalledTimes(1);
+
+    // (2) the call received only the embeddable texts ('a'+'b'; 'c' is empty).
+    const callArgs = batchGenerateEmbeddingsMock.mock.calls[0]![0] as string[];
+    expect(callArgs).toEqual(['aa', 'bbbb']);
+
+    // (3) per-event mapping: 'aa' (len 2) and 'bbbb' (len 4) get their own
+    //     vectors + the model stamp; the empty event gets no embedding.
+    const byId = new Map(client.streamed.map((it) => [it.id, it]));
+    expect(byId.get('a')!.embedding).toEqual([2, 0, 0]);
+    expect(byId.get('a')!.embedding_model).toBe('stub-model-v1');
+    expect(byId.get('b')!.embedding).toEqual([4, 0, 0]);
+    expect(byId.get('b')!.embedding_model).toBe('stub-model-v1');
+    expect(byId.get('c')!.embedding).toBeUndefined();
+    expect(byId.get('c')!.embedding_model).toBeUndefined();
+  });
+});

packages/connector-worker/src/__tests__/executor-heartbeat.test.ts

@@ -24,9 +24,10 @@ const executeCompiledConnectorMock = mock<AnyFn>(async () => ({
   output: { ok: true },
 }));
 
-const batchGenerateEmbeddingsMock = mock<AnyFn>(async (texts: string[]) =>
-  texts.map(() => [0.1, 0.2, 0.3])
-);
+const batchGenerateEmbeddingsMock = mock<AnyFn>(async (texts: string[]) => ({
+  embeddings: texts.map(() => [0.1, 0.2, 0.3]),
+  model: 'test-model',
+}));
 
 mock.module('../executor/runtime.js', () => ({
   executeCompiledConnector: executeCompiledConnectorMock,
@@ -163,7 +164,7 @@ describe('executor heartbeats (lobu#860)', () => {
 
     batchGenerateEmbeddingsMock.mockImplementationOnce(async (texts: string[]) => {
       await fireIntervalTicks(2);
-      return texts.map(() => [0.1, 0.2, 0.3]);
+      return { embeddings: texts.map(() => [0.1, 0.2, 0.3]), model: 'test-model' };
     });
 
     const job = {

packages/connector-worker/src/daemon/client.ts

@@ -139,6 +139,8 @@ export interface ContentItem {
   metadata?: Record<string, unknown>;
   origin_parent_id?: string;
   embedding?: number[];
+  /** Model/version stamp that produced `embedding`; persisted so vector spaces never mix. */
+  embedding_model?: string;
   origin_type?: string;
   semantic_type?: string;
 }
@@ -185,7 +187,7 @@ export interface EmbedEvent {
 export interface CompleteEmbeddingsRequest {
   run_id: number;
   worker_id: string;
-  embeddings: Array<{ event_id: number; embedding: number[] }>;
+  embeddings: Array<{ event_id: number; embedding: number[]; embedding_model?: string }>;
   error_message?: string;
 }