fix(agent-memory): push consolidate predicates into the query and skip summaries

- olderThanSeconds/maxImportance are now NUMERIC range clauses in the
  FT.SEARCH filter, so the scan limit applies to actual matches instead of an
  arbitrary first window, and content isn't transferred for discarded rows
- the candidate scan excludes @source:{summary}, so a default run no longer
  re-folds prior summaries into a new summary (target them explicitly to do so)
- extract buildConsolidateFilter (shares scope-clause logic with buildScopeFilter)

Author: jamby77

packages/agent-memory/src/MemoryStore.ts

@@ -1,7 +1,12 @@
 import { randomUUID } from 'node:crypto';
 import { encodeFloat32, parseFtSearchResponse } from '@betterdb/valkey-search-kit';
 import { buildMemoryRecord } from './buildMemoryRecord';
-import { buildRecallQuery, buildScopeFilter, SCORE_FIELD } from './buildRecallQuery';
+import {
+  buildConsolidateFilter,
+  buildRecallQuery,
+  buildScopeFilter,
+  SCORE_FIELD,
+} from './buildRecallQuery';
 import { parseMemoryItem } from './parseMemoryItem';
 import { compositeScore, similarityFromDistance, type RecallWeights } from './compositeScore';
 import { selectEvictions, type EvictionCandidate } from './selectEvictions';
@@ -10,7 +15,6 @@ import type {
   ConsolidateResult,
   EmbedFn,
   MemoryHit,
-  MemoryItem,
   MemoryScope,
   MemoryStoreClient,
   RecallOptions,
@@ -27,6 +31,7 @@ const FORGET_MAX_BATCHES = 10000;
 const EVICTION_SCAN_LIMIT = 10000;
 const CONSOLIDATE_SCAN_LIMIT = 10000;
 const DEFAULT_SUMMARY_IMPORTANCE = 0.7;
+const SUMMARY_SOURCE = 'summary';
 
 export interface MemoryStoreOptions {
   client: MemoryStoreClient;
@@ -231,10 +236,23 @@ export class MemoryStore {
       );
     }
 
+    // Push olderThanSeconds/maxImportance into the query (both are NUMERIC
+    // indexed) so the scan limit applies to actual matches, not an arbitrary
+    // first window, and we don't transfer rows we'd only discard. Prior
+    // summaries are excluded so a default run doesn't re-fold them into a new
+    // summary; target them explicitly to re-consolidate.
+    const filter = buildConsolidateFilter(scope, tags, {
+      maxCreatedAt:
+        options.olderThanSeconds !== undefined
+          ? now - options.olderThanSeconds * 1000
+          : undefined,
+      maxImportance: options.maxImportance,
+      excludeSource: SUMMARY_SOURCE,
+    });
     const raw = await this.client.call(
       'FT.SEARCH',
       `${this.name}:mem:idx`,
-      buildScopeFilter(scope, tags),
+      filter,
       'RETURN',
       '10',
       'content',
@@ -253,9 +271,7 @@ export class MemoryStore {
       'DIALECT',
       '2',
     );
-    const candidates = parseFtSearchResponse(raw)
-      .map((hit) => parseMemoryItem(this.name, hit))
-      .filter((item) => isConsolidationCandidate(item, options, now));
+    const candidates = parseFtSearchResponse(raw).map((hit) => parseMemoryItem(this.name, hit));
 
     if (candidates.length === 0) {
       return { consolidated: 0, created: [], deleted: 0 };
@@ -267,7 +283,7 @@ export class MemoryStore {
     const summaryId = await this.remember(summary, {
       ...scope,
       tags,
-      source: 'summary',
+      source: SUMMARY_SOURCE,
       importance: options.summaryImportance ?? DEFAULT_SUMMARY_IMPORTANCE,
     });
 
@@ -400,19 +416,3 @@ function ftSearchTotal(raw: unknown): number {
   return Number.isFinite(total) && total > 0 ? total : 0;
 }
 
-function isConsolidationCandidate(
-  item: MemoryItem,
-  options: ConsolidateOptions,
-  now: number,
-): boolean {
-  if (options.maxImportance !== undefined && item.importance > options.maxImportance) {
-    return false;
-  }
-  if (options.olderThanSeconds !== undefined) {
-    const ageSeconds = (now - item.createdAt) / 1000;
-    if (ageSeconds < options.olderThanSeconds) {
-      return false;
-    }
-  }
-  return true;
-}

packages/agent-memory/src/__tests__/MemoryStore.consolidate.test.ts

@@ -89,34 +89,40 @@ describe('MemoryStore.consolidate', () => {
     expect(del?.slice(1).sort()).toEqual(['mem:mem:a', 'mem:mem:b']);
   });
 
-  it('selects only candidates older than olderThanSeconds', async () => {
+  it('pushes olderThanSeconds into the query as a created_at upper bound', async () => {
     const summarize = vi.fn(async () => 'summary');
-    const client = consolidatingClient([
-      itemHit('old', { importance: 0.2, ageSeconds: 100000 }),
-      itemHit('recent', { importance: 0.2, ageSeconds: 10 }),
-    ]);
+    const client = consolidatingClient([itemHit('a', { importance: 0.2, ageSeconds: 100000 })]);
     const store = new MemoryStore({ client, name: 'mem', embedFn: fakeEmbed(8) });
 
-    const result = await store.consolidate({ olderThanSeconds: 3600, summarize });
+    await store.consolidate({ olderThanSeconds: 3600, summarize });
 
-    expect(summarize.mock.calls[0][0].map((i: { id: string }) => i.id)).toEqual(['old']);
-    expect(result.consolidated).toBe(1);
-    const del = client.call.mock.calls.find((c) => c[0] === 'DEL');
-    expect(del?.slice(1)).toEqual(['mem:mem:old']);
+    const search = client.call.mock.calls.find((c) => c[0] === 'FT.SEARCH');
+    const filter = search?.[2] as string;
+    // Server-side range so the scan limit applies to actual matches, not an
+    // arbitrary first window.
+    expect(filter).toMatch(/@created_at:\[-inf \d+\]/);
   });
 
-  it('selects only candidates at or below maxImportance', async () => {
+  it('pushes maxImportance into the query as an importance upper bound', async () => {
     const summarize = vi.fn(async () => 'summary');
-    const client = consolidatingClient([
-      itemHit('low', { importance: 0.2, ageSeconds: 100000 }),
-      itemHit('high', { importance: 0.9, ageSeconds: 100000 }),
-    ]);
+    const client = consolidatingClient([itemHit('a', { importance: 0.2, ageSeconds: 100000 })]);
+    const store = new MemoryStore({ client, name: 'mem', embedFn: fakeEmbed(8) });
+
+    await store.consolidate({ maxImportance: 0.5, summarize });
+
+    const search = client.call.mock.calls.find((c) => c[0] === 'FT.SEARCH');
+    expect(search?.[2] as string).toContain('@importance:[-inf 0.5]');
+  });
+
+  it('excludes prior summaries from the candidate scan so a default run does not re-fold them', async () => {
+    const summarize = vi.fn(async () => 'summary');
+    const client = consolidatingClient([itemHit('a', { importance: 0.2, ageSeconds: 100000 })]);
     const store = new MemoryStore({ client, name: 'mem', embedFn: fakeEmbed(8) });
 
-    const result = await store.consolidate({ maxImportance: 0.5, summarize });
+    await store.consolidate({ namespace: 'u1', summarize });
 
-    expect(summarize.mock.calls[0][0].map((i: { id: string }) => i.id)).toEqual(['low']);
-    expect(result.consolidated).toBe(1);
+    const search = client.call.mock.calls.find((c) => c[0] === 'FT.SEARCH');
+    expect(search?.[2] as string).toContain('-@source:{summary}');
   });
 
   it('writes the summary scoped to the request at summaryImportance', async () => {

packages/agent-memory/src/__tests__/buildRecallQuery.test.ts

@@ -1,5 +1,5 @@
 import { describe, it, expect } from 'vitest';
-import { buildRecallQuery } from '../buildRecallQuery';
+import { buildRecallQuery, buildConsolidateFilter } from '../buildRecallQuery';
 
 describe('buildRecallQuery', () => {
   it('builds a bare KNN query when there are no filters', () => {
@@ -18,3 +18,25 @@ describe('buildRecallQuery', () => {
     );
   });
 });
+
+describe('buildConsolidateFilter', () => {
+  it('appends inclusive NUMERIC ranges and a source exclusion to the scope filter', () => {
+    expect(
+      buildConsolidateFilter({ namespace: 'u1' }, ['pref'], {
+        maxCreatedAt: 1000,
+        maxImportance: 0.5,
+        excludeSource: 'summary',
+      }),
+    ).toBe('(@namespace:{u1} @tags:{pref} @created_at:[-inf 1000] @importance:[-inf 0.5] -@source:{summary})');
+  });
+
+  it('omits absent predicates', () => {
+    expect(buildConsolidateFilter({ threadId: 't' }, [], { excludeSource: 'summary' })).toBe(
+      '(@threadId:{t} -@source:{summary})',
+    );
+  });
+
+  it('still constrains by range when there is no scope', () => {
+    expect(buildConsolidateFilter({}, [], { maxImportance: 0.3 })).toBe('(@importance:[-inf 0.3])');
+  });
+});

packages/agent-memory/src/buildRecallQuery.ts

@@ -4,7 +4,7 @@ import type { MemoryScope } from './types';
 export const SCORE_FIELD = '__score';
 export const VECTOR_FIELD = 'vector';
 
-export function buildScopeFilter(scope: MemoryScope, tags: string[]): string {
+function scopeClauses(scope: MemoryScope, tags: string[]): string[] {
   const clauses: string[] = [];
   if (scope.threadId !== undefined) {
     clauses.push(`@threadId:{${escapeTag(scope.threadId)}}`);
@@ -18,12 +18,44 @@ export function buildScopeFilter(scope: MemoryScope, tags: string[]): string {
   for (const tag of tags) {
     clauses.push(`@tags:{${escapeTag(tag)}}`);
   }
+  return clauses;
+}
+
+function joinClauses(clauses: string[]): string {
   if (clauses.length === 0) {
     return '*';
   }
   return `(${clauses.join(' ')})`;
 }
 
+export function buildScopeFilter(scope: MemoryScope, tags: string[]): string {
+  return joinClauses(scopeClauses(scope, tags));
+}
+
+export interface ConsolidateFilterOptions {
+  maxCreatedAt?: number;
+  maxImportance?: number;
+  excludeSource?: string;
+}
+
+export function buildConsolidateFilter(
+  scope: MemoryScope,
+  tags: string[],
+  options: ConsolidateFilterOptions,
+): string {
+  const clauses = scopeClauses(scope, tags);
+  if (options.maxCreatedAt !== undefined) {
+    clauses.push(`@created_at:[-inf ${options.maxCreatedAt}]`);
+  }
+  if (options.maxImportance !== undefined) {
+    clauses.push(`@importance:[-inf ${options.maxImportance}]`);
+  }
+  if (options.excludeSource !== undefined) {
+    clauses.push(`-@source:{${escapeTag(options.excludeSource)}}`);
+  }
+  return joinClauses(clauses);
+}
+
 export function buildRecallQuery(k: number, scope: MemoryScope, tags: string[]): string {
   return `${buildScopeFilter(scope, tags)}=>[KNN ${k} @${VECTOR_FIELD} $vec AS ${SCORE_FIELD}]`;
 }