Add memory rollout feature flags and docs

src/services/memory/feature-flags.js

@@ -0,0 +1,61 @@
+const FALSEY = new Set(['0', 'false', 'off', 'no', 'disabled']);
+const TRUTHY = new Set(['1', 'true', 'on', 'yes', 'enabled', 'all']);
+
+function normalize(value) {
+  return String(value ?? '').trim();
+}
+
+function parseAgentToggle(value) {
+  const raw = normalize(value);
+  if (!raw) return { kind: 'unset' };
+
+  const lower = raw.toLowerCase();
+  if (FALSEY.has(lower)) return { kind: 'off' };
+  if (TRUTHY.has(lower)) return { kind: 'all' };
+
+  const allowList = raw
+    .split(',')
+    .map((item) => item.trim())
+    .filter(Boolean);
+
+  return allowList.length > 0 ? { kind: 'allow-list', allowList } : { kind: 'off' };
+}
+
+function isToggleEnabledForAgent(value, agentId, fallbackWhenUnset) {
+  const parsed = parseAgentToggle(value);
+  if (parsed.kind === 'unset') return fallbackWhenUnset;
+  if (parsed.kind === 'off') return false;
+  if (parsed.kind === 'all') return true;
+  if (!agentId) return false;
+  return parsed.allowList.includes(String(agentId));
+}
+
+export function isMemoryEnabled(env = process.env) {
+  return isToggleEnabledForAgent(env.TORCH_MEMORY_ENABLED, undefined, true);
+}
+
+export function isMemoryIngestEnabled(agentId, env = process.env) {
+  if (!isMemoryEnabled(env)) return false;
+  return isToggleEnabledForAgent(env.TORCH_MEMORY_INGEST_ENABLED, agentId, true);
+}
+
+export function isMemoryRetrievalEnabled(agentId, env = process.env) {
+  if (!isMemoryEnabled(env)) return false;
+  return isToggleEnabledForAgent(env.TORCH_MEMORY_RETRIEVAL_ENABLED, agentId, false);
+}
+
+export function getMemoryPruneMode(env = process.env) {
+  if (!isMemoryEnabled(env)) return 'off';
+
+  const raw = normalize(env.TORCH_MEMORY_PRUNE_ENABLED);
+  if (!raw) return 'active';
+
+  const lower = raw.toLowerCase();
+  if (FALSEY.has(lower)) return 'off';
+  if (['dry-run', 'dryrun', 'report', 'audit'].includes(lower)) return 'dry-run';
+  return 'active';
+}
+
+export function isMemoryPruneEnabled(env = process.env) {
+  return getMemoryPruneMode(env) === 'active';
+}

src/services/memory/formatter.js

@@ -1,3 +1,5 @@
+import { isMemoryRetrievalEnabled } from './feature-flags.js';
+export { isMemoryRetrievalEnabled } from './feature-flags.js';
 const DEFAULT_TOKEN_BUDGET = 1000;
 const DEFAULT_MAX_SUMMARY_CHARS = 240;
 const DEFAULT_MAX_CONTEXT_CHARS = 320;
@@ -130,32 +132,6 @@ function toContextString(agentContext) {
   return String(agentContext).trim();
 }
 
-/**
- * Supports safe rollout and canarying with one env var:
- * - true/1/on/all => enabled for all agents
- * - false/0/off/no/empty => disabled
- * - comma-separated agent ids => enabled only for matching ids
- *
- * @param {string} agentId
- * @param {Record<string, string | undefined>} [env]
- */
-export function isMemoryRetrievalEnabled(agentId, env = process.env) {
-  const raw = String(env.TORCH_MEMORY_RETRIEVAL_ENABLED ?? '').trim();
-  if (!raw) return false;
-
-  const normalized = raw.toLowerCase();
-  if (['0', 'false', 'off', 'no', 'disabled'].includes(normalized)) return false;
-  if (['1', 'true', 'on', 'yes', 'enabled', 'all'].includes(normalized)) return true;
-
-  const allowList = raw
-    .split(',')
-    .map((value) => value.trim())
-    .filter(Boolean);
-
-  if (allowList.length === 0) return false;
-  return allowList.includes(agentId);
-}
-
 /**
  * Builds a final prompt with retrieved memory context inserted ahead of the base prompt.
  *

src/services/memory/index.js

@@ -3,6 +3,7 @@ import { ingestMemoryWindow } from './ingestor.js';
 import { listPruneCandidates, selectPrunableMemories } from './pruner.js';
 import { filterAndRankMemories, updateMemoryUsage } from './retriever.js';
 import { startScheduler } from './scheduler.js';
+import { getMemoryPruneMode, isMemoryIngestEnabled } from './feature-flags.js';
 
 const memoryStore = new Map();
 const cache = createMemoryCache();
@@ -133,6 +134,10 @@ function recordThroughput(samples, now) {
 export async function ingestEvents(events, options = {}) {
   const repository = options.repository ?? memoryRepository;
   const telemetry = buildTelemetryEmitter(options);
+  if (!isMemoryIngestEnabled(options.agent_id, options.env)) {
+    telemetry('memory:ingest_skipped', { reason: 'flag_disabled', agent_id: options.agent_id ?? null });
+    return [];
+  }
   const records = await ingestMemoryWindow({ events }, {
     ...options,
     repository,
@@ -201,6 +206,7 @@ export async function runPruneCycle(options = {}) {
   const repository = options.repository ?? memoryRepository;
   const telemetry = buildTelemetryEmitter(options);
   const retentionMs = options.retentionMs ?? (1000 * 60 * 60 * 24 * 30);
+  const pruneMode = options.pruneMode ?? getMemoryPruneMode(options.env);
   const dbCandidates = await listPruneCandidates(repository, {
     retentionMs,
     now: options.now,
@@ -213,6 +219,21 @@ export async function runPruneCycle(options = {}) {
       now: options.now,
     });
 
+  if (pruneMode === 'off') {
+    telemetry('memory:prune_skipped', { reason: 'flag_disabled', retention_ms: retentionMs });
+    return { pruned: [], mode: 'off' };
+  }
+
+  if (pruneMode === 'dry-run') {
+    telemetry('memory:pruned', {
+      pruned_count: prunable.length,
+      retention_ms: retentionMs,
+      dry_run: true,
+    });
+    emitMetric(options, 'memory_pruned_total', { count: 0, retention_ms: retentionMs, dry_run: true });
+    return { pruned: [], candidates: prunable, mode: 'dry-run' };
+  }
+
   for (const memory of prunable) {
     memoryStore.delete(memory.id);
   }
@@ -229,7 +250,7 @@ export async function runPruneCycle(options = {}) {
 
   const result = { pruned: prunable };
   if (Number.isFinite(options.scheduleEveryMs) && options.scheduleEveryMs > 0) {
-    result.scheduler = startScheduler(() => runPruneCycle({ retentionMs, repository }), {
+    result.scheduler = startScheduler(() => runPruneCycle({ retentionMs, repository, env: options.env, pruneMode: options.pruneMode }), {
       intervalMs: options.scheduleEveryMs,
     });
   }

src/services/memory/scheduler.js

@@ -1,3 +1,4 @@
+import { getMemoryPruneMode, isMemoryEnabled, isMemoryIngestEnabled } from './feature-flags.js';
 const MINUTE_MS = 60_000;
 const HOUR_MS = 60 * MINUTE_MS;
 const DAY_MS = 24 * HOUR_MS;
@@ -182,6 +183,21 @@ async function runScheduledJob(options) {
   }
 }
 
+function shouldRunJob(jobName, options) {
+  const env = options.env ?? process.env;
+  if (!isMemoryEnabled(env)) return false;
+
+  if (jobName === 'ingestRecentRuntimeEvents') {
+    return isMemoryIngestEnabled(options.ingestAgentId, env);
+  }
+
+  if (jobName === 'pruningCycle') {
+    return getMemoryPruneMode(env) !== 'off';
+  }
+
+  return true;
+}
+
 /**
  * @param {{
  *   handlers: {
@@ -218,6 +234,17 @@ export function startMemoryMaintenanceScheduler(options) {
 
   const startFixedIntervalJob = (jobName, definition, handler) => {
     const run = () => {
+      if (!shouldRunJob(jobName, options)) {
+        options.emitMetric?.('memory_scheduler_job', {
+          job: jobName,
+          status: 'skipped_flag_disabled',
+          durationMs: 0,
+          itemCount: 0,
+          failures: 0,
+          retries: 0,
+        });
+        return;
+      }
       void runJob(jobName, definition, handler);
     };
 
@@ -232,6 +259,18 @@ export function startMemoryMaintenanceScheduler(options) {
       const delayMs = definition.minIntervalMs + Math.floor((definition.maxIntervalMs - definition.minIntervalMs) * jitterRatio);
       const timeout = setTimeout(async () => {
         timeouts.delete(timeout);
+        if (!shouldRunJob(jobName, options)) {
+          options.emitMetric?.('memory_scheduler_job', {
+            job: jobName,
+            status: 'skipped_flag_disabled',
+            durationMs: 0,
+            itemCount: 0,
+            failures: 0,
+            retries: 0,
+          });
+          scheduleNext();
+          return;
+        }
         await runJob(jobName, definition, handler);
         scheduleNext();
       }, delayMs);
@@ -239,6 +278,18 @@ export function startMemoryMaintenanceScheduler(options) {
     };
 
     if (options.runImmediately !== false) {
+      if (!shouldRunJob(jobName, options)) {
+        options.emitMetric?.('memory_scheduler_job', {
+          job: jobName,
+          status: 'skipped_flag_disabled',
+          durationMs: 0,
+          itemCount: 0,
+          failures: 0,
+          retries: 0,
+        });
+        scheduleNext();
+        return;
+      }
       void runJob(jobName, definition, handler).then(() => {
         scheduleNext();
       });