Add durable memory schema validation in ingest pipeline

src/services/memory/index.js

@@ -4,7 +4,7 @@ import { normalizeEvents } from './ingestor.js';
 import { selectPrunableMemories } from './pruner.js';
 import { filterAndRankMemories } from './retriever.js';
 import { startScheduler } from './scheduler.js';
-import { createMemoryRecord } from './schema.js';
+import { createMemoryRecord, validateMemoryItem } from './schema.js';
 import { summarizeEvents } from './summarizer.js';
 
 const memoryStore = new Map();
@@ -15,7 +15,7 @@ const cache = createMemoryCache();
  *
  * @param {import('./schema.js').MemoryEvent[]} events - Ordered list of raw events to persist.
  * @param {{ maxSummaryLength?: number, embedText?: (value: string) => Promise<number[]> | number[] }} [options] - Optional ingest tuning.
- * @returns {Promise<import('./schema.js').MemoryRecord[]>} Stored memory records with generated IDs/embeddings.
+ * @returns {Promise<import('./schema.js').MemoryRecord[]>} Stored memory records.
  */
 export async function ingestEvents(events, options = {}) {
   const normalizedEvents = normalizeEvents(events);
@@ -26,15 +26,32 @@ export async function ingestEvents(events, 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,
-      metadata: event.metadata,
-      embedding,
+      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,
-      updated_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');
+    }
+
     memoryStore.set(record.id, record);
     records.push(record);
   }
@@ -100,7 +117,7 @@ export function pinMemory(id) {
   const memory = memoryStore.get(id);
   if (!memory) return null;
 
-  const updated = { ...memory, pinned: true, updated_at: Date.now() };
+  const updated = { ...memory, pinned: true, last_seen: Date.now() };
   memoryStore.set(id, updated);
   cache.clear();
   return updated;
@@ -116,7 +133,7 @@ export function unpinMemory(id) {
   const memory = memoryStore.get(id);
   if (!memory) return null;
 
-  const updated = { ...memory, pinned: false, updated_at: Date.now() };
+  const updated = { ...memory, pinned: false, last_seen: Date.now() };
   memoryStore.set(id, updated);
   cache.clear();
   return updated;

src/services/memory/pruner.js

@@ -7,5 +7,5 @@ export function selectPrunableMemories(memories, options) {
   const now = options.now ?? Date.now();
   const cutoff = now - options.retentionMs;
 
-  return memories.filter((memory) => !memory.pinned && memory.updated_at < cutoff);
+  return memories.filter((memory) => !memory.pinned && memory.last_seen < cutoff);
 }

src/services/memory/retriever.js

@@ -30,7 +30,7 @@ export function filterAndRankMemories(memories, params) {
       const haystack = `${memory.summary} ${memory.content}`.toLowerCase();
       const queryScore = loweredQuery && haystack.includes(loweredQuery) ? 1 : 0;
       const pinBoost = memory.pinned ? 1 : 0;
-      const recencyScore = memory.updated_at;
+      const recencyScore = Number.isFinite(memory.last_seen) ? memory.last_seen : memory.created_at;
 
       return {
         memory,

src/services/memory/schema.js

@@ -7,18 +7,26 @@
  * @property {Record<string, unknown>} [metadata]
  */
 
+export const SCHEMA_VERSION = 1;
+
 /**
  * @typedef {Object} MemoryRecord
+ * @property {number} schema_version
  * @property {string} id
  * @property {string} agent_id
- * @property {string} summary
+ * @property {string} session_id
+ * @property {string} type
  * @property {string} content
- * @property {number} created_at
- * @property {number} updated_at
+ * @property {string} summary
  * @property {string[]} tags
- * @property {number[] | null} embedding
+ * @property {number} importance
+ * @property {string | null} embedding_id
+ * @property {number} created_at
+ * @property {number} last_seen
+ * @property {string} source
+ * @property {number | null} ttl_days
+ * @property {string | null} merged_into
  * @property {boolean} pinned
- * @property {Record<string, unknown>} metadata
  */
 
 /**
@@ -28,28 +36,99 @@
 export function normalizeEvent(event) {
   return {
     ...event,
-    timestamp: Number.isFinite(event?.timestamp) ? event.timestamp : Date.now(),
-    tags: Array.isArray(event?.tags) ? event.tags : [],
+    timestamp: Number.isFinite(event?.timestamp) ? Number(event.timestamp) : Date.now(),
+    tags: Array.isArray(event?.tags)
+      ? event.tags.map((tag) => String(tag)).filter((tag) => tag.length > 0)
+      : [],
     metadata: event?.metadata && typeof event.metadata === 'object' ? event.metadata : {},
   };
 }
 
+/**
+ * @param {Partial<MemoryRecord>} input
+ * @returns {MemoryRecord}
+ */
+export function normalizeMemoryItem(input = {}) {
+  const now = Date.now();
+
+  return {
+    schema_version: SCHEMA_VERSION,
+    id: typeof input.id === 'string' && input.id.trim() ? input.id.trim() : crypto.randomUUID(),
+    agent_id: typeof input.agent_id === 'string' ? input.agent_id.trim() : '',
+    session_id: typeof input.session_id === 'string' ? input.session_id.trim() : 'unknown',
+    type: typeof input.type === 'string' && input.type.trim() ? input.type.trim() : 'event',
+    content: typeof input.content === 'string' ? input.content : '',
+    summary: typeof input.summary === 'string' ? input.summary : '',
+    tags: Array.isArray(input.tags)
+      ? input.tags.map((tag) => String(tag).trim()).filter((tag) => tag.length > 0)
+      : [],
+    importance: Number.isFinite(input.importance)
+      ? Math.max(0, Math.min(1, Number(input.importance)))
+      : 0.5,
+    embedding_id: typeof input.embedding_id === 'string' && input.embedding_id.trim()
+      ? input.embedding_id.trim()
+      : null,
+    created_at: Number.isFinite(input.created_at) ? Number(input.created_at) : now,
+    last_seen: Number.isFinite(input.last_seen) ? Number(input.last_seen) : now,
+    source: typeof input.source === 'string' && input.source.trim() ? input.source.trim() : 'ingest',
+    ttl_days: Number.isFinite(input.ttl_days) ? Math.max(0, Math.floor(Number(input.ttl_days))) : null,
+    merged_into: typeof input.merged_into === 'string' && input.merged_into.trim()
+      ? input.merged_into.trim()
+      : null,
+    pinned: Boolean(input.pinned),
+  };
+}
+
+/**
+ * @param {unknown} item
+ * @returns {{ valid: boolean, errors: Array<{ field: string, expected: string, received: string, value: unknown }> }}
+ */
+export function validateMemoryItem(item) {
+  const errors = [];
+
+  const check = (condition, field, expected, value) => {
+    if (!condition) {
+      errors.push({
+        field,
+        expected,
+        received: Array.isArray(value) ? 'array' : typeof value,
+        value,
+      });
+    }
+  };
+
+  check(item && typeof item === 'object', 'item', 'object', item);
+  if (!item || typeof item !== 'object') {
+    return { valid: false, errors };
+  }
+
+  check(item.schema_version === SCHEMA_VERSION, 'schema_version', `number (${SCHEMA_VERSION})`, item.schema_version);
+  check(typeof item.id === 'string' && item.id.length > 0, 'id', 'non-empty string', item.id);
+  check(typeof item.agent_id === 'string' && item.agent_id.length > 0, 'agent_id', 'non-empty string', item.agent_id);
+  check(typeof item.session_id === 'string' && item.session_id.length > 0, 'session_id', 'non-empty string', item.session_id);
+  check(typeof item.type === 'string' && item.type.length > 0, 'type', 'non-empty string', item.type);
+  check(typeof item.content === 'string', 'content', 'string', item.content);
+  check(typeof item.summary === 'string', 'summary', 'string', item.summary);
+  check(Array.isArray(item.tags) && item.tags.every((tag) => typeof tag === 'string'), 'tags', 'string[]', item.tags);
+  check(Number.isFinite(item.importance) && item.importance >= 0 && item.importance <= 1, 'importance', 'number between 0 and 1', item.importance);
+  check(item.embedding_id === null || (typeof item.embedding_id === 'string' && item.embedding_id.length > 0), 'embedding_id', 'null or non-empty string', item.embedding_id);
+  check(Number.isFinite(item.created_at), 'created_at', 'finite number', item.created_at);
+  check(Number.isFinite(item.last_seen), 'last_seen', 'finite number', item.last_seen);
+  check(typeof item.source === 'string' && item.source.length > 0, 'source', 'non-empty string', item.source);
+  check(item.ttl_days === null || (Number.isInteger(item.ttl_days) && item.ttl_days >= 0), 'ttl_days', 'null or non-negative integer', item.ttl_days);
+  check(item.merged_into === null || (typeof item.merged_into === 'string' && item.merged_into.length > 0), 'merged_into', 'null or non-empty string', item.merged_into);
+  check(typeof item.pinned === 'boolean', 'pinned', 'boolean', item.pinned);
+
+  return {
+    valid: errors.length === 0,
+    errors,
+  };
+}
+
 /**
  * @param {Partial<MemoryRecord>} record
  * @returns {MemoryRecord}
  */
 export function createMemoryRecord(record) {
-  const now = Date.now();
-  return {
-    id: record.id || crypto.randomUUID(),
-    agent_id: record.agent_id || '',
-    summary: record.summary || '',
-    content: record.content || '',
-    created_at: Number.isFinite(record.created_at) ? record.created_at : now,
-    updated_at: Number.isFinite(record.updated_at) ? record.updated_at : now,
-    tags: Array.isArray(record.tags) ? record.tags : [],
-    embedding: Array.isArray(record.embedding) ? record.embedding : null,
-    pinned: Boolean(record.pinned),
-    metadata: record.metadata && typeof record.metadata === 'object' ? record.metadata : {},
-  };
+  return normalizeMemoryItem(record);
 }

test/memory-schema.test.mjs

@@ -0,0 +1,78 @@
+import { test } from 'node:test';
+import assert from 'node:assert/strict';
+import { ingestEvents } from '../src/services/memory/index.js';
+import {
+  SCHEMA_VERSION,
+  normalizeMemoryItem,
+  validateMemoryItem,
+} from '../src/services/memory/schema.js';
+
+test('normalizeMemoryItem applies durable schema defaults and coercions', () => {
+  const item = normalizeMemoryItem({
+    agent_id: 'agent-1',
+    content: 'hello',
+    summary: 'hello',
+    tags: ['a', 2, ''],
+    importance: 2,
+    ttl_days: 3.8,
+    pinned: 'yes',
+  });
+
+  assert.equal(item.schema_version, SCHEMA_VERSION);
+  assert.equal(item.agent_id, 'agent-1');
+  assert.deepEqual(item.tags, ['a', '2']);
+  assert.equal(item.importance, 1);
+  assert.equal(item.ttl_days, 3);
+  assert.equal(item.pinned, true);
+  assert.equal(item.type, 'event');
+  assert.equal(item.source, 'ingest');
+  assert.ok(item.id);
+});
+
+test('validateMemoryItem reports field-level errors', () => {
+  const result = validateMemoryItem({
+    schema_version: SCHEMA_VERSION,
+    id: 'id-1',
+    agent_id: '',
+    session_id: 's-1',
+    type: 'event',
+    content: 'content',
+    summary: 'summary',
+    tags: ['tag'],
+    importance: 0.4,
+    embedding_id: null,
+    created_at: Date.now(),
+    last_seen: Date.now(),
+    source: 'ingest',
+    ttl_days: null,
+    merged_into: null,
+    pinned: false,
+  });
+
+  assert.equal(result.valid, false);
+  assert.ok(result.errors.some((error) => error.field === 'agent_id'));
+});
+
+test('ingestEvents rejects invalid writes and logs structured validation details', async () => {
+  const originalError = console.error;
+  const entries = [];
+  console.error = (...args) => {
+    entries.push(args);
+  };
+
+  try {
+    await assert.rejects(
+      () => ingestEvents([{ content: 'missing agent id' }]),
+      /Memory ingest rejected/
+    );
+  } finally {
+    console.error = originalError;
+  }
+
+  assert.equal(entries.length, 1);
+  const [tag, payload] = entries[0];
+  assert.equal(tag, 'memory_validation_error');
+  assert.equal(payload.stage, 'ingest');
+  assert.ok(Array.isArray(payload.fields));
+  assert.ok(payload.fields.some((field) => field.field === 'agent_id'));
+});