Efficient Memory Consolidation at Scale

Running consolidation on millions of memories is completely different from running it on thousands.

The Problem

The original consolidation worker was simple: scan through recent memories, identify clusters using cosine-centroid similarity, compress related experiences into higher-level abstractions. This works well at small scale. At large scale, two things break.

First, the scan is expensive. Going through millions of memories to find consolidation candidates takes hours. Second, the worker falls behind. If new experiences arrive faster than the worker processes them, the consolidation lag grows unbounded and the system slowly fills with unprocessed raw experiences.

Here is how the current consolidation engine selects replay candidates. It uses a decay-aware query that prioritizes low-retrieval-count, high-importance events - the memories that have been formed but not yet processed:

// packages/consolidation-engine/src/engine.ts

async selectReplayCandidates(
  request: ConsolidationRequest,
): Promise<ReadonlyArray<ReplayCandidate>> {
  const baseQuery = buildReplaySelectionQuery({
    maxAge: request.maxAge,
    size: request.size ?? 100,
    minImportance: request.minImportance ?? 0.1,
    timestampField: "timestamp",
    usageField: "retrieval_count",
  });

  // Overlay required tag filters without touching sort/size.
  const query = request.requiredTags?.length
    ? {
        ...baseQuery,
        query: {
          bool: {
            must: [
              baseQuery["query"],
              ...request.requiredTags.map((tag) => ({ term: { tags: tag } })),
            ],
          },
        },
      }
    : baseQuery;

  const hits = await this.searchClient(this.openSearch, "experience_events", query);

  return hits.map((hit) => ({
    memoryId: hit._source.event_id ?? hit._id,
    summary: hit._source.summary ?? "",
    embedding: hit._source.embedding ?? [],
    importanceScore: hit._source.importance_score ?? 0,
    rewardScore: hit._source.reward_score ?? 0.5,
    retrievalCount: hit._source.retrieval_count ?? 0,
    tags: hit._source.tags ?? [],
  }));
}

After writing a consolidated memory, the engine bumps each source event's retrieval_count. This is what prevents the same events from being selected as candidates repeatedly - each consolidation run advances the watermark organically.

Incremental Consolidation

The first fix is incremental processing. Instead of re-scanning everything on every run, maintain a consolidation watermark that tracks which memories have already been processed. Each run advances the watermark, processing only new and recently modified memories.

This alone reduces per-run work by an order of magnitude once the memory store has matured, because most memories were consolidated in previous runs and do not need to be reconsidered.

Smarter Candidate Selection

Full cosine-centroid clustering on thousands of candidates is expensive. Before running it, use lightweight heuristics to identify likely cluster candidates: shared tags, overlapping time windows, similar source types, common entity references. Run full clustering only on these pre-filtered candidates.

The quality tradeoff is small - heuristic pre-filtering occasionally misses a cross-domain consolidation opportunity - but the compute savings are significant.

Distributed Execution

Multiple consolidation workers running in parallel across domain partitions dramatically reduce consolidation lag. A coordination layer - distributed locks via Redis - prevents workers from processing the same partition simultaneously.

With parallel workers across four domain partitions, consolidation lag drops from hours to minutes even as event volume grows.

The Result

These three changes together - incremental processing, smarter sampling, and distributed execution - transform consolidation from a batch process that struggles to keep up into a near-real-time component that stays current with the event stream.

Consolidation is one of the most compute-intensive parts of the system. Getting it right at scale is not optional.