Add memory admin CLI hooks and safe telemetry events

src/lib.mjs

@@ -22,6 +22,14 @@ import { cmdInit, cmdUpdate } from './ops.mjs';
 import { getRoster as _getRoster } from './roster.mjs';
 import { queryLocks as _queryLocks, publishLock as _publishLock, parseLockEvent } from './lock-ops.mjs';
 import { cmdDashboard } from './dashboard.mjs';
+import {
+  inspectMemory,
+  listMemories,
+  memoryStats,
+  pinMemory,
+  triggerPruneDryRun,
+  unpinMemory,
+} from './services/memory/index.js';
 import { ExitError } from './errors.mjs';
 import { todayDateStr, nowUnix, getIsoWeekStr } from './utils.mjs';
 import fs from 'node:fs/promises';
@@ -442,6 +450,14 @@ function parseArgs(argv) {
     port: DEFAULT_DASHBOARD_PORT,
     logDir: 'task-logs',
     ignoreLogs: false,
+    id: null,
+    type: null,
+    tags: [],
+    pinned: null,
+    limit: null,
+    offset: null,
+    retentionMs: null,
+    windowMs: null,
   };
   let i = 0;
 
@@ -470,6 +486,40 @@ function parseArgs(argv) {
       args.logDir = argv[++i];
     } else if (arg === '--ignore-logs') {
       args.ignoreLogs = true;
+    } else if (arg === '--id') {
+      args.id = argv[++i];
+    } else if (arg.startsWith('--id=')) {
+      args.id = arg.split('=')[1];
+    } else if (arg === '--type') {
+      args.type = argv[++i];
+    } else if (arg.startsWith('--type=')) {
+      args.type = arg.split('=')[1];
+    } else if (arg === '--tags') {
+      args.tags = String(argv[++i]).split(',').map((tag) => tag.trim()).filter(Boolean);
+    } else if (arg.startsWith('--tags=')) {
+      args.tags = String(arg.split('=')[1]).split(',').map((tag) => tag.trim()).filter(Boolean);
+    } else if (arg === '--pinned') {
+      const value = String(argv[++i]).toLowerCase();
+      args.pinned = value === 'true' ? true : value === 'false' ? false : null;
+    } else if (arg.startsWith('--pinned=')) {
+      const value = String(arg.split('=')[1]).toLowerCase();
+      args.pinned = value === 'true' ? true : value === 'false' ? false : null;
+    } else if (arg === '--limit') {
+      args.limit = parseInt(argv[++i], 10);
+    } else if (arg.startsWith('--limit=')) {
+      args.limit = parseInt(arg.split('=')[1], 10);
+    } else if (arg === '--offset') {
+      args.offset = parseInt(argv[++i], 10);
+    } else if (arg.startsWith('--offset=')) {
+      args.offset = parseInt(arg.split('=')[1], 10);
+    } else if (arg === '--retention-ms') {
+      args.retentionMs = parseInt(argv[++i], 10);
+    } else if (arg.startsWith('--retention-ms=')) {
+      args.retentionMs = parseInt(arg.split('=')[1], 10);
+    } else if (arg === '--window-ms') {
+      args.windowMs = parseInt(argv[++i], 10);
+    } else if (arg.startsWith('--window-ms=')) {
+      args.windowMs = parseInt(arg.split('=')[1], 10);
     }
   }
 
@@ -488,6 +538,13 @@ Commands:
   init      [--force]                              Initialize torch/ directory in current project
   update    [--force]                              Update torch/ configuration (backups, merges)
 
+  list-memories           [--agent <id>] [--type <type>] [--tags <a,b>] [--pinned <true|false>]
+  inspect-memory          --id <memoryId>
+  pin-memory              --id <memoryId>
+  unpin-memory            --id <memoryId>
+  trigger-prune-dry-run   [--retention-ms <ms>]
+  memory-stats            [--window-ms <ms>]
+
 Options:
   --dry-run       Build and sign the event but do not publish
   --force         Overwrite existing files (for init) or all files (for update)
@@ -580,6 +637,61 @@ export async function main(argv) {
         break;
       }
 
+      case 'list-memories': {
+        const result = await listMemories({
+          agent_id: args.agent,
+          type: args.type,
+          tags: args.tags,
+          pinned: args.pinned,
+          limit: args.limit,
+          offset: args.offset,
+        });
+        console.log(JSON.stringify(result, null, 2));
+        break;
+      }
+
+      case 'inspect-memory': {
+        if (!args.id) {
+          console.error('ERROR: --id <memoryId> is required for inspect-memory');
+          throw new ExitError(1, 'Missing memory id');
+        }
+        const result = await inspectMemory(args.id);
+        console.log(JSON.stringify(result, null, 2));
+        break;
+      }
+
+      case 'pin-memory': {
+        if (!args.id) {
+          console.error('ERROR: --id <memoryId> is required for pin-memory');
+          throw new ExitError(1, 'Missing memory id');
+        }
+        const result = await pinMemory(args.id);
+        console.log(JSON.stringify(result, null, 2));
+        break;
+      }
+
+      case 'unpin-memory': {
+        if (!args.id) {
+          console.error('ERROR: --id <memoryId> is required for unpin-memory');
+          throw new ExitError(1, 'Missing memory id');
+        }
+        const result = await unpinMemory(args.id);
+        console.log(JSON.stringify(result, null, 2));
+        break;
+      }
+
+      case 'trigger-prune-dry-run': {
+        const result = await triggerPruneDryRun({ retentionMs: args.retentionMs ?? undefined });
+        console.log(JSON.stringify(result, null, 2));
+        break;
+      }
+
+      case 'memory-stats': {
+        const result = await memoryStats({ windowMs: args.windowMs ?? undefined });
+        console.log(JSON.stringify(result, null, 2));
+        break;
+      }
+
       default:
         console.error(`ERROR: Unknown command: ${args.command}`);
         usage();

src/services/memory/db-adapter.js

@@ -137,5 +137,63 @@ export function createMemoryRepository(db) {
 
       return result.rows?.[0] ?? null;
     },
+
+    /**
+     * @param {string} id
+     */
+    async getMemoryById(id) {
+      const result = await db.query(
+        `SELECT *
+         FROM memories
+         WHERE id = $1
+         LIMIT 1`,
+        [id]
+      );
+
+      return result.rows?.[0] ?? null;
+    },
+
+    /**
+     * @param {{ agent_id?: string, type?: string, pinned?: boolean, limit?: number, offset?: number }} [filters]
+     */
+    async listMemories(filters = {}) {
+      const where = [];
+      const params = [];
+
+      if (typeof filters.agent_id === 'string' && filters.agent_id.length > 0) {
+        params.push(filters.agent_id);
+        where.push(`agent_id = $${params.length}`);
+      }
+
+      if (typeof filters.type === 'string' && filters.type.length > 0) {
+        params.push(filters.type);
+        where.push(`type = $${params.length}`);
+      }
+
+      if (typeof filters.pinned === 'boolean') {
+        params.push(filters.pinned);
+        where.push(`pinned = $${params.length}`);
+      }
+
+      const limit = Number.isFinite(filters.limit) ? Math.max(1, Number(filters.limit)) : 100;
+      const offset = Number.isFinite(filters.offset) ? Math.max(0, Number(filters.offset)) : 0;
+      params.push(limit);
+      const limitPos = params.length;
+      params.push(offset);
+      const offsetPos = params.length;
+
+      const whereClause = where.length > 0 ? `WHERE ${where.join(' AND ')}` : '';
+      const result = await db.query(
+        `SELECT *
+         FROM memories
+         ${whereClause}
+         ORDER BY last_seen DESC
+         LIMIT $${limitPos}
+         OFFSET $${offsetPos}`,
+        params
+      );
+
+      return result.rows ?? [];
+    },
   };
 }

src/services/memory/index.js

@@ -35,8 +35,94 @@ const memoryRepository = {
     memoryStore.set(id, updated);
     return updated;
   },
+  async getMemoryById(id) {
+    return memoryStore.get(id) ?? null;
+  },
+  async listMemories() {
+    return [...memoryStore.values()];
+  },
 };
 
+const memoryStatsState = {
+  ingested: [],
+  retrieved: [],
+  pruned: [],
+  archived: 0,
+  deleted: 0,
+};
+
+const ESTIMATED_INDEX_BYTES_PER_RECORD = 512;
+
+function toSafeTelemetryPayload(payload = {}) {
+  const blockedFields = new Set(['content', 'summary', 'query', 'raw']);
+  return Object.entries(payload).reduce((safe, [key, value]) => {
+    if (blockedFields.has(key)) return safe;
+    if (typeof value === 'string' && value.length > 300) {
+      safe[key] = `${value.slice(0, 300)}…`;
+      return safe;
+    }
+    safe[key] = value;
+    return safe;
+  }, {});
+}
+
+function buildTelemetryEmitter(options = {}) {
+  if (typeof options.emitTelemetry === 'function') {
+    return (event, payload) => options.emitTelemetry(event, toSafeTelemetryPayload(payload));
+  }
+
+  if (typeof options.telemetry?.emit === 'function') {
+    return (event, payload) => options.telemetry.emit(event, toSafeTelemetryPayload(payload));
+  }
+
+  return (event, payload) => {
+    console.log('memory_telemetry', {
+      event,
+      payload: toSafeTelemetryPayload(payload),
+      ts: Date.now(),
+    });
+  };
+}
+
+function emitMetric(options = {}, metric, payload) {
+  if (typeof options.emitMetric === 'function') {
+    options.emitMetric(metric, payload);
+    return;
+  }
+
+  if (typeof options.metrics?.emit === 'function') {
+    options.metrics.emit(metric, payload);
+    return;
+  }
+
+  console.log('memory_metric', { metric, payload, ts: Date.now() });
+}
+
+function applyMemoryFilters(memories, filters = {}) {
+  const tags = Array.isArray(filters.tags)
+    ? filters.tags.map((tag) => String(tag).trim()).filter(Boolean)
+    : [];
+
+  return memories.filter((memory) => {
+    if (filters.agent_id && memory.agent_id !== filters.agent_id) return false;
+    if (filters.type && memory.type !== filters.type) return false;
+    if (typeof filters.pinned === 'boolean' && memory.pinned !== filters.pinned) return false;
+    if (typeof filters.includeMerged === 'boolean' && !filters.includeMerged && memory.merged_into) return false;
+    if (tags.length > 0) {
+      const memoryTags = new Set(memory.tags);
+      if (!tags.every((tag) => memoryTags.has(tag))) return false;
+    }
+    return true;
+  });
+}
+
+function recordThroughput(samples, now) {
+  samples.push(now);
+  if (samples.length > 10_000) {
+    samples.splice(0, samples.length - 10_000);
+  }
+}
+
 /**
  * Ingests agent events and stores them as memory records.
  *
@@ -46,7 +132,18 @@ const memoryRepository = {
  */
 export async function ingestEvents(events, options = {}) {
   const repository = options.repository ?? memoryRepository;
-  const records = await ingestMemoryWindow({ events }, { ...options, repository });
+  const telemetry = buildTelemetryEmitter(options);
+  const records = await ingestMemoryWindow({ events }, {
+    ...options,
+    repository,
+    emitTelemetry: (event, payload) => {
+      telemetry(event, payload);
+      if (event === 'memory:ingested') {
+        recordThroughput(memoryStatsState.ingested, Date.now());
+        emitMetric(options, 'memory_ingested_total', { count: 1 });
+      }
+    },
+  });
   cache.clear();
   return records;
 }
@@ -59,15 +156,38 @@ export async function ingestEvents(events, options = {}) {
  */
 export async function getRelevantMemories(params) {
   const { repository = memoryRepository, ...queryParams } = params;
+  const telemetry = buildTelemetryEmitter(params);
   const cacheKey = JSON.stringify(queryParams);
   const cached = cache.get(cacheKey);
   if (cached) {
+    telemetry('memory:retrieved', {
+      agent_id: queryParams.agent_id,
+      result_count: cached.length,
+      cache_hit: true,
+      query_present: Boolean(queryParams.query),
+      query_length: typeof queryParams.query === 'string' ? queryParams.query.length : 0,
+      tags_count: Array.isArray(queryParams.tags) ? queryParams.tags.length : 0,
+    });
+    emitMetric(params, 'memory_retrieved_total', { count: cached.length, cache_hit: true });
+    recordThroughput(memoryStatsState.retrieved, Date.now());
     return cached;
   }
 
   const ranked = await filterAndRankMemories([...memoryStore.values()], queryParams);
   await updateMemoryUsage(repository, ranked.map((memory) => memory.id));
   cache.set(cacheKey, ranked);
+
+  telemetry('memory:retrieved', {
+    agent_id: queryParams.agent_id,
+    result_count: ranked.length,
+    cache_hit: false,
+    query_present: Boolean(queryParams.query),
+    query_length: typeof queryParams.query === 'string' ? queryParams.query.length : 0,
+    tags_count: Array.isArray(queryParams.tags) ? queryParams.tags.length : 0,
+  });
+  emitMetric(params, 'memory_retrieved_total', { count: ranked.length, cache_hit: false });
+  recordThroughput(memoryStatsState.retrieved, Date.now());
+
   return ranked;
 }
 
@@ -79,6 +199,7 @@ export async function getRelevantMemories(params) {
  */
 export async function runPruneCycle(options = {}) {
   const repository = options.repository ?? memoryRepository;
+  const telemetry = buildTelemetryEmitter(options);
   const retentionMs = options.retentionMs ?? (1000 * 60 * 60 * 24 * 30);
   const dbCandidates = await listPruneCandidates(repository, {
     retentionMs,
@@ -96,6 +217,14 @@ export async function runPruneCycle(options = {}) {
     memoryStore.delete(memory.id);
   }
 
+  memoryStatsState.deleted += prunable.length;
+  recordThroughput(memoryStatsState.pruned, Date.now());
+  telemetry('memory:pruned', {
+    pruned_count: prunable.length,
+    retention_ms: retentionMs,
+  });
+  emitMetric(options, 'memory_pruned_total', { count: prunable.length, retention_ms: retentionMs });
+
   cache.clear();
 
   const result = { pruned: prunable };
@@ -114,8 +243,9 @@ export async function runPruneCycle(options = {}) {
  * @param {string} id - Memory record identifier.
  * @returns {Promise<import('./schema.js').MemoryRecord | null>} Updated memory record or null when not found.
  */
-export async function pinMemory(id) {
-  const updated = await memoryRepository.setPinned(id, true);
+export async function pinMemory(id, options = {}) {
+  const repository = options.repository ?? memoryRepository;
+  const updated = await repository.setPinned(id, true);
   cache.clear();
   return updated;
 }
@@ -126,8 +256,9 @@ export async function pinMemory(id) {
  * @param {string} id - Memory record identifier.
  * @returns {Promise<import('./schema.js').MemoryRecord | null>} Updated memory record or null when not found.
  */
-export async function unpinMemory(id) {
-  const updated = await memoryRepository.setPinned(id, false);
+export async function unpinMemory(id, options = {}) {
+  const repository = options.repository ?? memoryRepository;
+  const updated = await repository.setPinned(id, false);
   cache.clear();
   return updated;
 }
@@ -143,3 +274,93 @@ export async function markMemoryMerged(id, mergedInto) {
   cache.clear();
   return merged;
 }
+
+export async function listMemories(filters = {}, options = {}) {
+  const repository = options.repository ?? memoryRepository;
+  const source = typeof repository.listMemories === 'function'
+    ? await repository.listMemories(filters)
+    : [...memoryStore.values()];
+
+  const filtered = applyMemoryFilters(source, filters)
+    .sort((left, right) => right.last_seen - left.last_seen);
+
+  const offset = Number.isFinite(filters.offset) ? Math.max(0, Number(filters.offset)) : 0;
+  const limit = Number.isFinite(filters.limit)
+    ? Math.max(1, Math.floor(Number(filters.limit)))
+    : filtered.length;
+
+  return filtered.slice(offset, offset + limit);
+}
+
+export async function inspectMemory(id, options = {}) {
+  const repository = options.repository ?? memoryRepository;
+  if (typeof repository.getMemoryById === 'function') {
+    return repository.getMemoryById(id);
+  }
+  return memoryStore.get(id) ?? null;
+}
+
+export async function triggerPruneDryRun(options = {}) {
+  const repository = options.repository ?? memoryRepository;
+  const retentionMs = options.retentionMs ?? (1000 * 60 * 60 * 24 * 30);
+  const dbCandidates = await listPruneCandidates(repository, {
+    retentionMs,
+    now: options.now,
+  });
+
+  const candidates = dbCandidates.length > 0
+    ? dbCandidates
+    : selectPrunableMemories(await listMemories({}, { repository }), {
+      retentionMs,
+      now: options.now,
+    });
+
+  return {
+    dryRun: true,
+    retentionMs,
+    candidateCount: candidates.length,
+    candidates: candidates.map((memory) => ({
+      id: memory.id,
+      agent_id: memory.agent_id,
+      type: memory.type,
+      pinned: memory.pinned,
+      last_seen: memory.last_seen,
+      merged_into: memory.merged_into,
+    })),
+  };
+}
+
+export async function memoryStats(options = {}) {
+  const repository = options.repository ?? memoryRepository;
+  const now = options.now ?? Date.now();
+  const windowMs = Number.isFinite(options.windowMs) ? Math.max(1, Number(options.windowMs)) : (60 * 60 * 1000);
+  const memories = await listMemories({}, { repository });
+
+  const countsByType = memories.reduce((acc, memory) => {
+    acc[memory.type] = (acc[memory.type] ?? 0) + 1;
+    return acc;
+  }, {});
+
+  const archivedCount = memories.filter((memory) => Boolean(memory.merged_into)).length;
+  const deletedRate = memoryStatsState.deleted / Math.max(1, memoryStatsState.ingested.length);
+  const archivedRate = archivedCount / Math.max(1, memories.length);
+
+  const ingestedInWindow = memoryStatsState.ingested.filter((timestamp) => timestamp >= (now - windowMs)).length;
+  const ingestThroughputPerMinute = ingestedInWindow / Math.max(1, (windowMs / 60_000));
+
+  return {
+    countsByType,
+    totals: {
+      total: memories.length,
+      pinned: memories.filter((memory) => memory.pinned).length,
+      archived: archivedCount,
+      deletedObserved: memoryStatsState.deleted,
+    },
+    rates: {
+      archivedRate,
+      deletedRate,
+    },
+    indexSizeEstimateBytes: memories.length * ESTIMATED_INDEX_BYTES_PER_RECORD,
+    ingestThroughputPerMinute,
+  };
+}

test/memory-admin.test.mjs

@@ -0,0 +1,106 @@
+import test from 'node:test';
+import assert from 'node:assert/strict';
+
+import {
+  getRelevantMemories,
+  ingestEvents,
+  inspectMemory,
+  listMemories,
+  memoryStats,
+  pinMemory,
+  triggerPruneDryRun,
+  unpinMemory,
+} from '../src/services/memory/index.js';
+
+test('listMemories, inspectMemory, pinMemory, and unpinMemory support admin hooks', async () => {
+  const records = [
+    { id: 'm1', agent_id: 'agent-a', type: 'event', tags: ['ops'], pinned: false, last_seen: 10, merged_into: null },
+    { id: 'm2', agent_id: 'agent-a', type: 'summary', tags: ['ops', 'prod'], pinned: true, last_seen: 20, merged_into: null },
+    { id: 'm3', agent_id: 'agent-b', type: 'event', tags: ['dev'], pinned: false, last_seen: 5, merged_into: 'm2' },
+  ];
+
+  const repository = {
+    async listMemories() {
+      return records;
+    },
+    async getMemoryById(id) {
+      return records.find((record) => record.id === id) ?? null;
+    },
+    async setPinned(id, pinned) {
+      const index = records.findIndex((record) => record.id === id);
+      if (index < 0) return null;
+      records[index] = { ...records[index], pinned };
+      return records[index];
+    },
+  };
+
+  const listed = await listMemories({ agent_id: 'agent-a', tags: ['ops'], includeMerged: false }, { repository });
+  assert.deepEqual(listed.map((entry) => entry.id), ['m2', 'm1']);
+
+  const inspected = await inspectMemory('m2', { repository });
+  assert.equal(inspected?.id, 'm2');
+
+  const pinned = await pinMemory('m1', { repository });
+  assert.equal(pinned?.pinned, true);
+
+  const unpinned = await unpinMemory('m2', { repository });
+  assert.equal(unpinned?.pinned, false);
+});
+
+test('triggerPruneDryRun returns candidate metadata without deleting records', async () => {
+  const repository = {
+    async listPruneCandidates() {
+      return [
+        { id: 'old-1', agent_id: 'agent-z', type: 'event', pinned: false, last_seen: 1, merged_into: null },
+      ];
+    },
+    async listMemories() {
+      return [];
+    },
+  };
+
+  const dryRun = await triggerPruneDryRun({ repository, retentionMs: 1000 });
+  assert.equal(dryRun.dryRun, true);
+  assert.equal(dryRun.candidateCount, 1);
+  assert.deepEqual(dryRun.candidates[0], {
+    id: 'old-1',
+    agent_id: 'agent-z',
+    type: 'event',
+    pinned: false,
+    last_seen: 1,
+    merged_into: null,
+  });
+});
+
+test('memoryStats reports counts/rates and retrieval telemetry remains redaction-safe', async () => {
+  const telemetry = [];
+  await ingestEvents([
+    {
+      agent_id: 'agent-telemetry',
+      content: 'customer token ABC123 secret',
+      timestamp: Date.now(),
+      tags: ['incident'],
+      metadata: { session_id: 's-telemetry' },
+    },
+  ]);
+
+  const retrieved = await getRelevantMemories({
+    agent_id: 'agent-telemetry',
+    query: 'show customer token ABC123 secret',
+    emitTelemetry: (event, payload) => telemetry.push({ event, payload }),
+  });
+
+  assert.ok(retrieved.length >= 1);
+  const retrievalEvent = telemetry.find((entry) => entry.event === 'memory:retrieved');
+  assert.ok(retrievalEvent);
+  assert.equal('query' in retrievalEvent.payload, false);
+  assert.equal(retrievalEvent.payload.query_present, true);
+  assert.ok(retrievalEvent.payload.query_length > 0);
+
+  const stats = await memoryStats({ windowMs: 60 * 60 * 1000 });
+  assert.ok(stats.totals.total >= 1);
+  assert.ok(Number.isFinite(stats.rates.archivedRate));
+  assert.ok(Number.isFinite(stats.rates.deletedRate));
+  assert.ok(stats.indexSizeEstimateBytes >= 0);
+  assert.ok(Number.isFinite(stats.ingestThroughputPerMinute));
+});