From 8e1ff73cbb2aa8b1dbe746b8a2a09bab6c308669 Mon Sep 17 00:00:00 2001 From: thePR0M3TH3AN <53631862+PR0M3TH3AN@users.noreply.github.com> Date: Sat, 14 Feb 2026 14:48:19 -0500 Subject: [PATCH] Implement batched memory ingestion pipeline with dedupe --- src/services/memory/index.js | 44 +----- src/services/memory/ingestor.js | 240 +++++++++++++++++++++++++++++++- test/memory-ingestor.test.mjs | 90 ++++++++++++ 3 files changed, 331 insertions(+), 43 deletions(-) create mode 100644 test/memory-ingestor.test.mjs diff --git a/src/services/memory/index.js b/src/services/memory/index.js index d5d9d67..95a93c1 100644 --- a/src/services/memory/index.js +++ b/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; } diff --git a/src/services/memory/ingestor.js b/src/services/memory/ingestor.js index c3bfdb8..86d6707 100644 --- a/src/services/memory/ingestor.js +++ b/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, params?: { since?: number, limit?: number }) => Record[] }, + * runtimeScope?: Record, + * logSource?: { getEvents?: (params: { from: number, to: number, agent_id?: string }) => Promise[]> | Record[], fetchEvents?: (params: { from: number, to: number, agent_id?: string }) => Promise[]> | Record[] }, + * agent_id?: string, + * windowStart?: number, + * windowEnd?: number, + * }} input + * @param {{ + * repository?: { insertMemory: (memory: import('./schema.js').MemoryRecord) => Promise, hasIngestionFingerprint?: (dedupeKey: string) => Promise, storeIngestionFingerprint?: (dedupeKey: string, payload: Record) => Promise, linkEmbedding?: (memoryId: string, embedding: { id: string, vector: number[] }) => Promise }, + * maxSummaryLength?: number, + * maxChunkChars?: number, + * maxChunkTokens?: number, + * embedText?: (value: string) => Promise | number[], + * emitTelemetry?: (event: string, payload: Record) => void, + * telemetry?: { emit: (event: string, payload: Record) => void }, + * now?: number, + * windowBucketMs?: number, + * }} options + * @returns {Promise} + */ +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 }} repository * @param {import('./schema.js').MemoryRecord} memory diff --git a/test/memory-ingestor.test.mjs b/test/memory-ingestor.test.mjs new file mode 100644 index 0000000..95a31a9 --- /dev/null +++ b/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 user@email.com', 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); +});