From 7cef8242a12ce1e7d554dd31c8be2cb0686de8fa Mon Sep 17 00:00:00 2001 From: thePR0M3TH3AN <53631862+PR0M3TH3AN@users.noreply.github.com> Date: Sat, 14 Feb 2026 15:15:15 -0500 Subject: [PATCH] Add memory admin CLI hooks and safe telemetry events --- src/lib.mjs | 112 +++++++++++++++ src/services/memory/db-adapter.js | 58 ++++++++ src/services/memory/index.js | 231 +++++++++++++++++++++++++++++- test/memory-admin.test.mjs | 106 ++++++++++++++ 4 files changed, 502 insertions(+), 5 deletions(-) create mode 100644 test/memory-admin.test.mjs diff --git a/src/lib.mjs b/src/lib.mjs index e32fa27..38dd2e4 100644 --- a/src/lib.mjs +++ b/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 ] [--type ] [--tags ] [--pinned ] + inspect-memory --id + pin-memory --id + unpin-memory --id + trigger-prune-dry-run [--retention-ms ] + memory-stats [--window-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 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 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 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(); diff --git a/src/services/memory/db-adapter.js b/src/services/memory/db-adapter.js index 4f770c1..2913173 100644 --- a/src/services/memory/db-adapter.js +++ b/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 ?? []; + }, }; } diff --git a/src/services/memory/index.js b/src/services/memory/index.js index 8752d9d..4922ee5 100644 --- a/src/services/memory/index.js +++ b/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} 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} 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, + }; +} diff --git a/test/memory-admin.test.mjs b/test/memory-admin.test.mjs new file mode 100644 index 0000000..4adbba8 --- /dev/null +++ b/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)); +});