Implement batched memory ingestion pipeline with dedupe

src/services/memory/index.js

@@ -1,11 +1,8 @@
 import { createMemoryCache } from './cache.js';
-import { embedText } from './embedder.js';
-import { insertMemory, normalizeEvents } from './ingestor.js';
+import { ingestMemoryWindow } from './ingestor.js';
 import { listPruneCandidates, selectPrunableMemories } from './pruner.js';
 import { filterAndRankMemories, updateMemoryUsage } from './retriever.js';
 import { startScheduler } from './scheduler.js';
-import { createMemoryRecord, validateMemoryItem } from './schema.js';
-import { summarizeEvents } from './summarizer.js';
 
 const memoryStore = new Map();
 const cache = createMemoryCache();
@@ -49,44 +46,7 @@ const memoryRepository = {
  */
 export async function ingestEvents(events, options = {}) {
   const repository = options.repository ?? memoryRepository;
-  const normalizedEvents = normalizeEvents(events);
-  const records = [];
-
-  for (const event of normalizedEvents) {
-    const summary = summarizeEvents([event], options);
-    const embedding = await embedText(`${summary}\n${event.content}`, options);
-    const record = createMemoryRecord({
-      agent_id: event.agent_id,
-      session_id: typeof event.metadata?.session_id === 'string' ? event.metadata.session_id : 'unknown',
-      type: typeof event.metadata?.type === 'string' ? event.metadata.type : 'event',
-      content: event.content,
-      summary,
-      tags: event.tags,
-      importance: Number.isFinite(event.metadata?.importance) ? event.metadata.importance : 0.5,
-      embedding_id: Array.isArray(embedding) && embedding.length > 0 ? crypto.randomUUID() : null,
-      created_at: event.timestamp,
-      last_seen: event.timestamp,
-      source: typeof event.metadata?.source === 'string' ? event.metadata.source : 'ingest',
-      ttl_days: Number.isFinite(event.metadata?.ttl_days) ? event.metadata.ttl_days : null,
-      merged_into: typeof event.metadata?.merged_into === 'string' ? event.metadata.merged_into : null,
-      pinned: Boolean(event.metadata?.pinned),
-    });
-
-    const validation = validateMemoryItem(record);
-    if (!validation.valid) {
-      console.error('memory_validation_error', {
-        stage: 'ingest',
-        reason: 'record_failed_schema_validation',
-        fields: validation.errors,
-        item: record,
-      });
-      throw new TypeError('Memory ingest rejected: invalid durable record format');
-    }
-
-    await insertMemory(repository, record);
-    records.push(record);
-  }
-
+  const records = await ingestMemoryWindow({ events }, { ...options, repository });
   cache.clear();
   return records;
 }

src/services/memory/ingestor.js

@@ -1,4 +1,100 @@
-import { normalizeEvent } from './schema.js';
+import { createHash, randomUUID } from 'node:crypto';
+import { createMemoryRecord, normalizeEvent, validateMemoryItem } from './schema.js';
+import { embedText } from './embedder.js';
+import { summarizeEvents } from './summarizer.js';
+
+const WINDOW_BUCKET_MS = 60_000;
+const MAX_DEDUPE_KEYS = 2000;
+const dedupeWindow = new Map();
+
+function pruneDedupeWindow() {
+  while (dedupeWindow.size > MAX_DEDUPE_KEYS) {
+    const oldestKey = dedupeWindow.keys().next().value;
+    if (!oldestKey) break;
+    dedupeWindow.delete(oldestKey);
+  }
+}
+
+function redactObviousPii(text) {
+  if (typeof text !== 'string' || text.length === 0) return '';
+  return text
+    .replace(/[A-Z0-9._%+-]+@[A-Z0-9.-]+\.[A-Z]{2,}/gi, '[redacted:email]')
+    .replace(/\b(?:\+?\d{1,2}[\s.-]?)?(?:\(?\d{3}\)?[\s.-]?\d{3}[\s.-]?\d{4})\b/g, '[redacted:phone]')
+    .replace(/\b\d{3}-\d{2}-\d{4}\b/g, '[redacted:ssn]');
+}
+
+function estimateTokens(text) {
+  if (!text) return 0;
+  return Math.ceil(text.length / 4);
+}
+
+function chunkEvents(events, options = {}) {
+  const maxChunkChars = Number.isFinite(options.maxChunkChars) && options.maxChunkChars > 0
+    ? Math.floor(options.maxChunkChars)
+    : 1_200;
+  const maxChunkTokens = Number.isFinite(options.maxChunkTokens) && options.maxChunkTokens > 0
+    ? Math.floor(options.maxChunkTokens)
+    : 300;
+
+  const chunks = [];
+  let currentChunk = [];
+  let currentChars = 0;
+  let currentTokens = 0;
+
+  const flush = () => {
+    if (currentChunk.length > 0) {
+      chunks.push(currentChunk);
+      currentChunk = [];
+      currentChars = 0;
+      currentTokens = 0;
+    }
+  };
+
+  for (const event of events) {
+    const eventChars = event.content.length;
+    const eventTokens = estimateTokens(event.content);
+    const overBudget = (currentChars + eventChars > maxChunkChars) || (currentTokens + eventTokens > maxChunkTokens);
+    if (overBudget) {
+      flush();
+    }
+
+    currentChunk.push(event);
+    currentChars += eventChars;
+    currentTokens += eventTokens;
+  }
+
+  flush();
+  return chunks;
+}
+
+function hashIngestionWindow({ agentId, bucket, content }) {
+  return createHash('sha256').update(`${agentId}|${bucket}|${content}`).digest('hex');
+}
+
+async function isDuplicateWindow(repository, dedupeKey) {
+  if (dedupeWindow.has(dedupeKey)) return true;
+  if (typeof repository?.hasIngestionFingerprint === 'function') {
+    const exists = await repository.hasIngestionFingerprint(dedupeKey);
+    if (exists) return true;
+  }
+  return false;
+}
+
+async function rememberWindowFingerprint(repository, dedupeKey, payload) {
+  dedupeWindow.set(dedupeKey, payload);
+  pruneDedupeWindow();
+  if (typeof repository?.storeIngestionFingerprint === 'function') {
+    await repository.storeIngestionFingerprint(dedupeKey, payload);
+  }
+}
+
+function buildTelemetryEmitter(options = {}) {
+  if (typeof options.emitTelemetry === 'function') return options.emitTelemetry;
+  if (typeof options.telemetry?.emit === 'function') {
+    return (event, payload) => options.telemetry.emit(event, payload);
+  }
+  return () => {};
+}
 
 /**
  * @param {import('./schema.js').MemoryEvent[]} events
@@ -8,6 +104,148 @@ export function normalizeEvents(events) {
   return events.map(normalizeEvent);
 }
 
+/**
+ * @param {{
+ *  events?: import('./schema.js').MemoryEvent[],
+ *  runtimeCache?: { getRecentRuntimeEvents?: (scope: Record<string, unknown>, params?: { since?: number, limit?: number }) => Record<string, unknown>[] },
+ *  runtimeScope?: Record<string, unknown>,
+ *  logSource?: { getEvents?: (params: { from: number, to: number, agent_id?: string }) => Promise<Record<string, unknown>[]> | Record<string, unknown>[], fetchEvents?: (params: { from: number, to: number, agent_id?: string }) => Promise<Record<string, unknown>[]> | Record<string, unknown>[] },
+ *  agent_id?: string,
+ *  windowStart?: number,
+ *  windowEnd?: number,
+ * }} input
+ * @param {{
+ *  repository?: { insertMemory: (memory: import('./schema.js').MemoryRecord) => Promise<unknown>, hasIngestionFingerprint?: (dedupeKey: string) => Promise<boolean>, storeIngestionFingerprint?: (dedupeKey: string, payload: Record<string, unknown>) => Promise<void>, linkEmbedding?: (memoryId: string, embedding: { id: string, vector: number[] }) => Promise<unknown> },
+ *  maxSummaryLength?: number,
+ *  maxChunkChars?: number,
+ *  maxChunkTokens?: number,
+ *  embedText?: (value: string) => Promise<number[]> | number[],
+ *  emitTelemetry?: (event: string, payload: Record<string, unknown>) => void,
+ *  telemetry?: { emit: (event: string, payload: Record<string, unknown>) => void },
+ *  now?: number,
+ *  windowBucketMs?: number,
+ * }} options
+ * @returns {Promise<import('./schema.js').MemoryRecord[]>}
+ */
+export async function ingestMemoryWindow(input = {}, options = {}) {
+  const repository = options.repository;
+  if (!repository || typeof repository.insertMemory !== 'function') {
+    throw new TypeError('Memory ingest requires a repository with insertMemory');
+  }
+
+  const emitTelemetry = buildTelemetryEmitter(options);
+  const windowStart = Number.isFinite(input.windowStart) ? Number(input.windowStart) : 0;
+  const windowEnd = Number.isFinite(input.windowEnd) ? Number(input.windowEnd) : (Number.isFinite(options.now) ? Number(options.now) : Date.now());
+
+  const runtimeEvents = typeof input.runtimeCache?.getRecentRuntimeEvents === 'function'
+    ? input.runtimeCache.getRecentRuntimeEvents(input.runtimeScope ?? {}, { since: windowStart, limit: 500 })
+    : [];
+
+  const logReader = input.logSource?.fetchEvents ?? input.logSource?.getEvents;
+  const logEvents = typeof logReader === 'function'
+    ? await logReader({ from: windowStart, to: windowEnd, agent_id: input.agent_id })
+    : [];
+
+  const normalizedEvents = normalizeEvents([
+    ...(Array.isArray(input.events) ? input.events : []),
+    ...(Array.isArray(runtimeEvents) ? runtimeEvents : []),
+    ...(Array.isArray(logEvents) ? logEvents : []),
+  ]);
+
+  const redactedEvents = normalizedEvents.map((event) => ({
+    ...event,
+    content: redactObviousPii(event.content),
+  }));
+
+  const eventsByAgent = new Map();
+  for (const event of redactedEvents) {
+    const events = eventsByAgent.get(event.agent_id) ?? [];
+    events.push(event);
+    eventsByAgent.set(event.agent_id, events);
+  }
+
+  const insertedRecords = [];
+  const bucketMs = Number.isFinite(options.windowBucketMs) && options.windowBucketMs > 0
+    ? Math.floor(options.windowBucketMs)
+    : WINDOW_BUCKET_MS;
+  const bucket = Math.floor(windowEnd / bucketMs);
+
+  for (const [agentId, agentEvents] of eventsByAgent.entries()) {
+    if (!agentEvents.length) continue;
+    const sorted = [...agentEvents].sort((a, b) => a.timestamp - b.timestamp);
+    const dedupeContent = sorted.map((event) => `${event.timestamp}:${event.content}`).join('\n');
+    const dedupeKey = hashIngestionWindow({ agentId, bucket, content: dedupeContent });
+
+    if (await isDuplicateWindow(repository, dedupeKey)) {
+      continue;
+    }
+
+    const chunks = chunkEvents(sorted, options);
+    for (const chunk of chunks) {
+      const summary = summarizeEvents(chunk, options);
+      const contextText = chunk.map((event) => event.content).join('\n');
+      const embedding = await embedText(`${summary}\n${contextText}`, options);
+      const chunkStart = Math.min(...chunk.map((event) => event.timestamp));
+      const chunkEnd = Math.max(...chunk.map((event) => event.timestamp));
+
+      const record = createMemoryRecord({
+        agent_id: agentId,
+        session_id: typeof chunk[0]?.metadata?.session_id === 'string' ? chunk[0].metadata.session_id : 'unknown',
+        type: 'batch_window',
+        content: contextText,
+        summary,
+        tags: [...new Set(chunk.flatMap((event) => event.tags))],
+        importance: Number.isFinite(chunk[0]?.metadata?.importance) ? Number(chunk[0].metadata.importance) : 0.5,
+        embedding_id: Array.isArray(embedding) && embedding.length > 0 ? randomUUID() : null,
+        created_at: chunkStart,
+        last_seen: chunkEnd,
+        source: typeof chunk[0]?.metadata?.source === 'string' ? chunk[0].metadata.source : 'ingest',
+        ttl_days: Number.isFinite(chunk[0]?.metadata?.ttl_days) ? Number(chunk[0].metadata.ttl_days) : null,
+        merged_into: null,
+        pinned: false,
+      });
+
+      const validation = validateMemoryItem(record);
+      if (!validation.valid) {
+        console.error('memory_validation_error', {
+          stage: 'ingest',
+          reason: 'record_failed_schema_validation',
+          fields: validation.errors,
+          item: record,
+        });
+        throw new TypeError('Memory ingest rejected: invalid durable record format');
+      }
+
+      await insertMemory(repository, record);
+      insertedRecords.push(record);
+
+      if (record.embedding_id && typeof repository.linkEmbedding === 'function') {
+        await repository.linkEmbedding(record.id, {
+          id: record.embedding_id,
+          vector: embedding,
+        });
+      }
+
+      emitTelemetry('memory:ingested', {
+        memory_id: record.id,
+        agent_id: record.agent_id,
+        dedupe_key: dedupeKey,
+        chunk_events: chunk.length,
+        window_start: windowStart,
+        window_end: windowEnd,
+      });
+    }
+
+    await rememberWindowFingerprint(repository, dedupeKey, {
+      agent_id: agentId,
+      window_start: windowStart,
+      window_end: windowEnd,
+    });
+  }
+
+  return insertedRecords;
+}
+
 /**
  * @param {{ insertMemory: (memory: import('./schema.js').MemoryRecord) => Promise<unknown> }} repository
  * @param {import('./schema.js').MemoryRecord} memory

test/memory-ingestor.test.mjs

@@ -0,0 +1,90 @@
+import test from 'node:test';
+import assert from 'node:assert/strict';
+
+import { ingestMemoryWindow } from '../src/services/memory/ingestor.js';
+
+test('ingestMemoryWindow gathers sources, redacts pii, chunks, embeds, links, and emits telemetry', async () => {
+  const inserted = [];
+  const linked = [];
+  const telemetry = [];
+
+  const repository = {
+    async insertMemory(memory) {
+      inserted.push(memory);
+      return memory;
+    },
+    async linkEmbedding(memoryId, embedding) {
+      linked.push({ memoryId, embedding });
+    },
+  };
+
+  const runtimeCache = {
+    getRecentRuntimeEvents() {
+      return [
+        { agent_id: 'agent-1', content: 'runtime event [email protected]', timestamp: 1100, tags: ['runtime'], metadata: {} },
+      ];
+    },
+  };
+
+  const logSource = {
+    async getEvents() {
+      return [
+        { agent_id: 'agent-1', content: 'call me at 555-555-1212', timestamp: 1200, tags: ['log'], metadata: {} },
+      ];
+    },
+  };
+
+  const records = await ingestMemoryWindow(
+    {
+      events: [
+        { agent_id: 'agent-1', content: 'hello', timestamp: 1000, tags: ['direct'], metadata: { session_id: 's1' } },
+      ],
+      runtimeCache,
+      logSource,
+      windowStart: 900,
+      windowEnd: 2000,
+    },
+    {
+      repository,
+      maxChunkChars: 40,
+      embedText: (text) => [text.length],
+      emitTelemetry: (event, payload) => telemetry.push({ event, payload }),
+      now: 2000,
+    }
+  );
+
+  assert.equal(records.length, 2);
+  assert.equal(inserted.length, 2);
+  assert.equal(linked.length, 2);
+  assert.equal(telemetry.length, 2);
+  assert.ok(inserted.some((item) => item.content.includes('[redacted:email]')));
+  assert.ok(inserted.some((item) => item.content.includes('[redacted:phone]')));
+  assert.ok(telemetry.every((entry) => entry.event === 'memory:ingested'));
+});
+
+test('ingestMemoryWindow dedupes overlapping windows by hashed content + agent + timestamp bucket', async () => {
+  const inserted = [];
+
+  const repository = {
+    async insertMemory(memory) {
+      inserted.push(memory);
+      return memory;
+    },
+  };
+
+  const input = {
+    events: [
+      { agent_id: 'agent-2', content: 'stable content', timestamp: 10_000, tags: ['a'], metadata: {} },
+      { agent_id: 'agent-2', content: 'stable content two', timestamp: 10_200, tags: ['b'], metadata: {} },
+    ],
+    windowStart: 10_000,
+    windowEnd: 10_500,
+  };
+
+  const first = await ingestMemoryWindow(input, { repository, now: 10_500 });
+  const second = await ingestMemoryWindow(input, { repository, now: 10_500 });
+
+  assert.equal(first.length, 1);
+  assert.equal(second.length, 0);
+  assert.equal(inserted.length, 1);
+});