diff --git a/test/memory-ingest.test.mjs b/test/memory-ingest.test.mjs index 50c71d4..eff7b6d 100644 --- a/test/memory-ingest.test.mjs +++ b/test/memory-ingest.test.mjs @@ -1,253 +1,165 @@ -import { test } from 'node:test'; +import { test, describe, beforeEach, afterEach } from 'node:test'; import assert from 'node:assert/strict'; import { ingestEvents, getRelevantMemories } from '../src/services/memory/index.js'; -// Mock repository helpers -function createMockRepository() { - const store = new Map(); - const fingerprints = new Set(); - const embeddings = new Map(); - let failNextInsert = false; +describe('ingestEvents', () => { + let repository; + let telemetryEvents = []; + let metricsEvents = []; + let storedMemories = []; + let storedFingerprints = new Map(); + let linkedEmbeddings = new Map(); - return { - _store: store, - _fingerprints: fingerprints, - _embeddings: embeddings, - setFailNextInsert(fail) { failNextInsert = fail; }, - - async insertMemory(memory) { - if (failNextInsert) { - throw new Error('Simulated insert failure'); - } - store.set(memory.id, memory); - return memory; + const mockOptions = { + embedText: async (text) => Array(128).fill(0.1), // Deterministic embedding + emitTelemetry: (event, payload) => telemetryEvents.push({ event, payload }), + metrics: { + emit: (metric, payload) => metricsEvents.push({ metric, payload }), }, - async hasIngestionFingerprint(key) { - return fingerprints.has(key); + // Mock environment to ensure enabled by default + env: { + TORCH_MEMORY_ENABLED: 'true', + TORCH_MEMORY_INGEST_ENABLED: 'true', }, - async storeIngestionFingerprint(key, payload) { - fingerprints.add(key); - }, - async linkEmbedding(id, embedding) { - embeddings.set(id, embedding); - }, - async listMemories() { - return [...store.values()]; - }, - async updateMemoryUsage(id) { - // no-op for this test - }, - async setPinned(id, pinned) { - const memory = store.get(id); - if (memory) { - memory.pinned = pinned; - store.set(id, memory); - } - return memory; - } - }; -} - -// Mock telemetry collector -function createTelemetryCollector() { - const events = []; - return { - events, - emit: (event, payload) => events.push({ event, payload }), - find: (eventName) => events.find(e => e.event === eventName), - filter: (eventName) => events.filter(e => e.event === eventName), - clear: () => { events.length = 0; } - }; -} - -// Mock metrics collector -function createMetricsCollector() { - const metrics = []; - return { - metrics, - emit: (metric, payload) => metrics.push({ metric, payload }), - find: (metricName) => metrics.find(m => m.metric === metricName), - clear: () => { metrics.length = 0; } - }; -} - -test('ingestEvents: Happy path ingestion', async () => { - const repository = createMockRepository(); - const telemetry = createTelemetryCollector(); - const metrics = createMetricsCollector(); - - const events = [ - { - agent_id: 'test-agent', - content: 'This is a test event for ingestion.', - timestamp: Date.now(), - tags: ['test'], - metadata: { session_id: 'session-1' } - } - ]; - - const options = { - repository, - emitTelemetry: telemetry.emit, - metrics: { emit: metrics.emit }, - embedText: async () => [0.1, 0.2, 0.3], // Mock embedding - agent_id: 'test-agent', - env: { TORCH_MEMORY_INGEST_ENABLED: 'true' } + now: 1600000000000, // Fixed time }; - const records = await ingestEvents(events, options); + beforeEach(() => { + telemetryEvents = []; + metricsEvents = []; + storedMemories = []; + storedFingerprints = new Map(); + linkedEmbeddings = new Map(); - // Assertions - assert.ok(Array.isArray(records), 'Should return an array of records'); - assert.equal(records.length, 1, 'Should return 1 record'); - assert.ok(records[0].id, 'Record should have an ID'); - assert.equal(records[0].content, events[0].content, 'Content should match'); - - // Verify repository - assert.ok(repository._store.has(records[0].id), 'Record should be stored in repository'); - - // Verify telemetry - const ingestedEvent = telemetry.find('memory:ingested'); - assert.ok(ingestedEvent, 'Should emit memory:ingested telemetry'); - assert.equal(ingestedEvent.payload.agent_id, 'test-agent'); - - // Verify metrics - const ingestedMetric = metrics.find('memory_ingested_total'); - assert.ok(ingestedMetric, 'Should emit memory_ingested_total metric'); - assert.equal(ingestedMetric.payload.count, 1); -}); - -test('ingestEvents: Skips ingestion when disabled via feature flag', async () => { - const repository = createMockRepository(); - const telemetry = createTelemetryCollector(); - - const events = [ - { - agent_id: 'test-agent', - content: 'Should be skipped', - timestamp: Date.now() - } - ]; - - const options = { - repository, - emitTelemetry: telemetry.emit, - agent_id: 'test-agent', - env: { TORCH_MEMORY_INGEST_ENABLED: 'false' } - }; - - const records = await ingestEvents(events, options); - - assert.deepEqual(records, [], 'Should return empty array'); - assert.equal(repository._store.size, 0, 'Should not store anything'); - - const skippedEvent = telemetry.find('memory:ingest_skipped'); - assert.ok(skippedEvent, 'Should emit memory:ingest_skipped'); - assert.equal(skippedEvent.payload.reason, 'flag_disabled'); -}); - -test('ingestEvents: Clears cache after ingestion', async () => { - const repository = createMockRepository(); - const telemetry = createTelemetryCollector(); - const agentId = 'cache-test-agent'; - - // Pre-populate repository - const existingMemory = { - id: 'mem-1', - agent_id: agentId, - content: 'Existing memory', - tags: ['test'], - last_seen: Date.now(), - type: 'event' - }; - await repository.insertMemory(existingMemory); - - // 1. Populate cache - await getRelevantMemories({ agent_id: agentId, repository, emitTelemetry: telemetry.emit }); - let retrieved = telemetry.filter('memory:retrieved'); - assert.equal(retrieved[0].payload.cache_hit, false, 'First retrieval should be a cache miss'); - telemetry.clear(); - - // 2. Verify cache hit - await getRelevantMemories({ agent_id: agentId, repository, emitTelemetry: telemetry.emit }); - retrieved = telemetry.filter('memory:retrieved'); - assert.equal(retrieved[0].payload.cache_hit, true, 'Second retrieval should be a cache hit'); - telemetry.clear(); - - // 3. Ingest new event (should trigger cache clear) - const events = [ - { - agent_id: agentId, - content: 'New memory triggers cache clear', - timestamp: Date.now() - } - ]; - - await ingestEvents(events, { - repository, - emitTelemetry: telemetry.emit, - embedText: async () => [0.1], - agent_id: agentId, - env: { TORCH_MEMORY_INGEST_ENABLED: 'true' } + repository = { + async insertMemory(memory) { + storedMemories.push(memory); + return memory; + }, + async hasIngestionFingerprint(key) { + return storedFingerprints.has(key); + }, + async storeIngestionFingerprint(key, payload) { + storedFingerprints.set(key, payload); + }, + async linkEmbedding(id, embedding) { + linkedEmbeddings.set(id, embedding); + }, + // Minimal implementation for getRelevantMemories to work for cache check + async listMemories() { + return storedMemories; + }, + async updateMemoryUsage() {}, + }; }); - // 4. Verify cache miss - telemetry.clear(); // Clear ingestion telemetry - await getRelevantMemories({ agent_id: agentId, repository, emitTelemetry: telemetry.emit }); - retrieved = telemetry.filter('memory:retrieved'); - assert.equal(retrieved[0].payload.cache_hit, false, 'Retrieval after ingestion should be a cache miss'); -}); - -test('ingestEvents: Deduplicates identical events within window', async () => { - const repository = createMockRepository(); - const agentId = 'dedupe-agent'; - const event = { - agent_id: agentId, - content: 'Duplicate content', - timestamp: 1000, - tags: ['test'] - }; - - const options = { - repository, - embedText: async () => [0.1], - agent_id: agentId, - env: { TORCH_MEMORY_INGEST_ENABLED: 'true' }, - windowBucketMs: 60000, // Fixed window - now: 2000 // Fixed time - }; - - // First ingestion - const records1 = await ingestEvents([event], options); - assert.equal(records1.length, 1, 'First ingestion should succeed'); - - // Second ingestion (same event, same window/bucket) - const records2 = await ingestEvents([event], options); - assert.equal(records2.length, 0, 'Second ingestion should be deduplicated (skipped)'); -}); - -test('ingestEvents: Handles repository errors', async () => { - const repository = createMockRepository(); - repository.setFailNextInsert(true); - - const events = [ - { - agent_id: 'error-agent', - content: 'Will fail', - timestamp: Date.now() - } - ]; - - const options = { - repository, - embedText: async () => [0.1], - agent_id: 'error-agent', - env: { TORCH_MEMORY_INGEST_ENABLED: 'true' } - }; - - await assert.rejects( - async () => ingestEvents(events, options), - /Simulated insert failure/, - 'Should propagate repository errors' - ); + test('happy path: successfully ingests events', async () => { + const events = [ + { + agent_id: 'agent-1', + content: 'User logged in', + timestamp: mockOptions.now - 1000, + metadata: { session_id: 'session-1', source: 'test' }, + }, + { + agent_id: 'agent-1', + content: 'User clicked button', + timestamp: mockOptions.now - 500, + metadata: { session_id: 'session-1', source: 'test' }, + }, + ]; + + const result = await ingestEvents(events, { ...mockOptions, repository }); + + assert.equal(result.length, 1); // Should be chunked into one memory + assert.equal(storedMemories.length, 1); + assert.equal(storedMemories[0].agent_id, 'agent-1'); + assert.match(storedMemories[0].content, /User logged in/); + assert.match(storedMemories[0].content, /User clicked button/); + + // Verify telemetry + const ingestedEvent = telemetryEvents.find((e) => e.event === 'memory:ingested'); + assert.ok(ingestedEvent, 'Should emit memory:ingested telemetry'); + assert.equal(ingestedEvent.payload.agent_id, 'agent-1'); + assert.equal(ingestedEvent.payload.chunk_events, 2); + + // Verify metrics + const ingestedMetric = metricsEvents.find((e) => e.metric === 'memory_ingested_total'); + assert.ok(ingestedMetric, 'Should emit memory_ingested_total metric'); + assert.equal(ingestedMetric.payload.count, 1); + + // Verify embedding link + assert.ok(linkedEmbeddings.has(storedMemories[0].id), 'Should link embedding'); + }); + + test('skips ingestion when feature flag is disabled', async () => { + const events = [{ agent_id: 'agent-1', content: 'test', timestamp: Date.now() }]; + const options = { + ...mockOptions, + repository, + env: { TORCH_MEMORY_INGEST_ENABLED: 'false' }, + }; + + const result = await ingestEvents(events, options); + + assert.equal(result.length, 0); + assert.equal(storedMemories.length, 0); + + const skippedEvent = telemetryEvents.find((e) => e.event === 'memory:ingest_skipped'); + assert.ok(skippedEvent, 'Should emit memory:ingest_skipped telemetry'); + assert.equal(skippedEvent.payload.reason, 'flag_disabled'); + }); + + test('clears cache after ingestion', async () => { + const agentId = 'agent-cache-test'; + + // 1. Populate cache + await getRelevantMemories({ agent_id: agentId, repository, ...mockOptions }); + + // 2. Verify cache hit (by checking telemetry from a second call) + telemetryEvents = []; // Reset telemetry + await getRelevantMemories({ agent_id: agentId, repository, ...mockOptions }); + const hitEvent = telemetryEvents.find(e => e.event === 'memory:retrieved'); + assert.ok(hitEvent.payload.cache_hit, 'Should hit cache before ingestion'); + + // 3. Ingest event + await ingestEvents( + [{ agent_id: agentId, content: 'new event', timestamp: mockOptions.now }], + { ...mockOptions, repository } + ); + + // 4. Verify cache miss + telemetryEvents = []; // Reset telemetry + await getRelevantMemories({ agent_id: agentId, repository, ...mockOptions }); + const missEvent = telemetryEvents.find(e => e.event === 'memory:retrieved'); + assert.equal(missEvent.payload.cache_hit, false, 'Should miss cache after ingestion (cache cleared)'); + }); + + test('skips duplicate windows', async () => { + const events = [{ agent_id: 'agent-dedupe', content: 'repeat', timestamp: mockOptions.now }]; + + // First ingestion + await ingestEvents(events, { ...mockOptions, repository }); + assert.equal(storedMemories.length, 1); + + // Second ingestion (should be duplicate) + await ingestEvents(events, { ...mockOptions, repository }); + assert.equal(storedMemories.length, 1, 'Should not insert duplicate memory'); + }); + + test('propagates repository errors', async () => { + const errorRepository = { + ...repository, + async insertMemory() { + throw new Error('Database connection failed'); + }, + }; + + const events = [{ agent_id: 'agent-err', content: 'fail', timestamp: mockOptions.now }]; + + await assert.rejects( + async () => ingestEvents(events, { ...mockOptions, repository: errorRepository }), + /Database connection failed/ + ); + }); });