Quality delta review 2026-07-14: moments N+1 batch, token-formula drift, DB atomicity (D1-D5)

lib/claude_memory/audit/checks.rb

@@ -382,14 +382,16 @@ def observation_status_corroboration(manager)
       def truncated_source_content(manager)
         detector = Distill::TruncationDetector.new
         flagged = []
-        total = 0
 
         %w[project global].each do |scope|
           store = manager.store_if_exists(scope)
           next unless store
-          rows = store.content_items.select(:id, :raw_text).all
-          total += rows.size
-          flagged.concat(rows.select { |r| detector.truncated?(r[:raw_text]) }.map { |r| r[:id] })
+          # Stream raw_text a page at a time — it is the largest column in the
+          # table (up to the ingest cap per row) and a mature DB holds thousands
+          # of rows, so materializing all of it at once is an unbounded load.
+          store.content_items.select(:id, :raw_text).order(:id).paged_each(rows_per_fetch: 500) do |row|
+            flagged << row[:id] if detector.truncated?(row[:raw_text])
+          end
         end
         return [] if flagged.empty?
 

lib/claude_memory/core/jaccard.rb

@@ -10,11 +10,11 @@ module Core
     # both-empty, so there is no 0/0).
     module Jaccard
       def self.score(a, b)
+        # The empty guard also covers both-empty, so the union below is always
+        # ≥ 1 and there is no 0/0 to defend against.
         return 0.0 if a.empty? || b.empty?
 
-        union = (a | b).size
-        return 0.0 if union.zero?
-        (a & b).size.to_f / union
+        (a & b).size.to_f / (a | b).size
       end
     end
   end

lib/claude_memory/core/token_budget.rb

@@ -30,6 +30,7 @@ def self.from_detail_json(json_values)
 
       def initialize(tokens)
         @sorted = tokens.sort.freeze
+        freeze
       end
 
       def empty?
@@ -54,10 +55,12 @@ def avg
       end
 
       def min
+        return 0 if empty?
         @sorted.first
       end
 
       def max
+        return 0 if empty?
         @sorted.last
       end
     end

lib/claude_memory/core/token_estimator.rb

@@ -2,7 +2,7 @@
 
 module ClaudeMemory
   module Core
-    class TokenEstimator
+    module TokenEstimator
       # Approximation: ~4 characters per token for English text
       # More accurate for Claude's tokenizer than simple word count
       CHARS_PER_TOKEN = 4.0
@@ -18,12 +18,9 @@ def self.from_chars(char_count)
       def self.estimate(text)
         return 0 if text.nil? || text.empty?
 
-        # Remove extra whitespace and count characters
-        normalized = text.strip.gsub(/\s+/, " ")
-        chars = normalized.length
-
-        # Return ceiling to avoid underestimation
-        (chars / CHARS_PER_TOKEN).ceil
+        # Normalize whitespace before counting, then defer to from_chars so the
+        # 4-chars/token arithmetic lives in exactly one place.
+        from_chars(text.strip.gsub(/\s+/, " ").length)
       end
 
       def self.estimate_fact(fact)

lib/claude_memory/dashboard/moments.rb

@@ -31,6 +31,9 @@ class Moments
         "sweep" => %w[hook_sweep]
       }.freeze
 
+      # Moment kinds whose detail carries scoped top-fact IDs to resolve.
+      SCOPED_FACT_KINDS = %w[context_injection recall_hit recall_empty].freeze
+
       def initialize(manager)
         @manager = manager
       end
@@ -70,7 +73,12 @@ def list(params = {})
         content_by_id = batch_content(store, content_ids)
         facts_by_content = batch_extracted_facts(store, content_ids)
 
-        moments = events.map { |e| build_moment(store, e, content_by_id, facts_by_content) }
+        # Preload scoped top-facts for every recall/context event in one query
+        # per scope (was a per-row facts + entities lookup in resolve_scoped_facts).
+        scoped_details = events.filter_map { |e| e[:details] if SCOPED_FACT_KINDS.include?(kind_for(e)) }
+        fact_index = ScopedFactResolver.build_fact_index(@manager, ScopedFactResolver.merge_scoped_ids(scoped_details))
+
+        moments = events.map { |e| build_moment(store, e, content_by_id, facts_by_content, fact_index) }
         moments = moments.select { |m| kinds.include?(m[:kind]) } unless kinds.empty?
         has_more = moments.size > limit
         moments = moments.first(limit)
@@ -130,7 +138,7 @@ def kind_for(event)
         end
       end
 
-      def build_moment(store, event, content_by_id, facts_by_content)
+      def build_moment(store, event, content_by_id, facts_by_content, fact_index)
         details = event[:details] || {}
         kind = kind_for(event)
         base = {
@@ -145,18 +153,18 @@ def build_moment(store, event, content_by_id, facts_by_content)
           details: details
         }
 
-        enrich(base, kind, details, content_by_id, facts_by_content)
+        enrich(base, kind, details, content_by_id, facts_by_content, fact_index)
       end
 
-      def enrich(moment, kind, details, content_by_id, facts_by_content)
+      def enrich(moment, kind, details, content_by_id, facts_by_content, fact_index)
         case kind
         when "context_injection"
           moment.merge(
             context_preview: details[:preview],
             context_length: details[:context_length],
             fact_count: details[:fact_count] || (details[:top_fact_ids] || []).size,
             top_subjects: details[:top_subjects] || [],
-            top_facts: resolve_scoped_facts(details),
+            top_facts: ScopedFactResolver.resolve_from_index(details, fact_index),
             truncated: details[:truncated]
           )
         when "recall_hit", "recall_empty"
@@ -165,7 +173,7 @@ def enrich(moment, kind, details, content_by_id, facts_by_content)
             query: details[:query],
             result_count: details[:result_count] || 0,
             scope: details[:scope],
-            top_facts: resolve_scoped_facts(details),
+            top_facts: ScopedFactResolver.resolve_from_index(details, fact_index),
             results_by_scope: details[:results_by_scope]
           )
         when "extraction"
@@ -196,11 +204,6 @@ def enrich(moment, kind, details, content_by_id, facts_by_content)
         end
       end
 
-      def resolve_scoped_facts(details)
-        scoped = ScopedFactResolver.scoped_ids_from_details(details)
-        ScopedFactResolver.resolve(@manager, scoped)
-      end
-
       # Collect the distinct content_item ids referenced by extraction/ingest
       # events, so their content + facts can be batch-loaded once per page.
       def content_item_ids_for(events)

lib/claude_memory/dashboard/scoped_fact_resolver.rb

@@ -63,6 +63,55 @@ def resolve(manager, scoped_ids)
         []
       end
 
+      # Merge the scoped-id hashes of many events into one {scope => [ids]}
+      # (deduped), so a whole page of recall/context events can be resolved
+      # with one query per scope instead of one per event.
+      def merge_scoped_ids(details_list)
+        merged = Hash.new { |h, k| h[k] = [] }
+        details_list.each do |details|
+          scoped_ids_from_details(details).each { |scope, ids| merged[scope.to_s].concat(ids) }
+        end
+        merged.transform_values(&:uniq)
+      end
+
+      # Batch-load presented facts for a merged {scope => [ids]} hash into a
+      # {scope => {fact_id => presented_fact}} index — one facts query and one
+      # FactPresenter entity load per scope for the entire page. Feed the
+      # result to resolve_from_index for each event. Replaces the per-event
+      # resolve (facts + entities query per row) with a per-scope batch.
+      def build_fact_index(manager, merged_scoped_ids)
+        return {} if merged_scoped_ids.nil? || merged_scoped_ids.empty?
+        index = {}
+        merged_scoped_ids.each do |scope, ids|
+          next if ids.nil? || ids.empty?
+          store = manager.store_if_exists(scope.to_s)
+          next unless store
+          rows = store.facts.where(id: ids.map(&:to_i)).all
+          next if rows.empty?
+          presented = FactPresenter.new(store).list_summary(rows)
+          index[scope.to_s] = presented.each_with_object({}) do |fact, acc|
+            acc[fact[:id]] = fact.merge(source: scope.to_s)
+          end
+        end
+        index
+      rescue Sequel::DatabaseError => e
+        ClaudeMemory.logger.debug("ScopedFactResolver#build_fact_index failed: #{e.message}")
+        {}
+      end
+
+      # Resolve one event's details against a prebuilt index (see
+      # build_fact_index), preserving per-scope input order. Pure — no I/O.
+      def resolve_from_index(details, index)
+        scoped = scoped_ids_from_details(details)
+        return [] if scoped.empty?
+        scoped.flat_map do |scope, ids|
+          per_scope = index[scope.to_s] || {}
+          # uniq so a repeated id in one event's list emits its fact once,
+          # matching the row-set dedup the query-based resolve gets for free.
+          ids.uniq.filter_map { |id| per_scope[id.to_i] }
+        end
+      end
+
       # Flat list of unique scoped pairs — handy for counting unique facts
       # referenced across a set of events.
       #

lib/claude_memory/index/vector_index.rb

@@ -109,22 +109,29 @@ def backfill_batch!(limit: 100)
         now = Time.now.utc.iso8601
         indexed_ids = []
 
-        rows.each do |row|
-          vector = JSON.parse(row[:embedding_json])
-          blob = vector.pack("f*")
-          # No DELETE needed: vec_indexed_at is nil so these rows can't be in vec0
-          execute_with_params(
-            "INSERT INTO facts_vec(fact_id, embedding) VALUES (?, ?)",
-            row[:id], blob
-          )
-          indexed_ids << row[:id]
-        rescue JSON::ParserError
-          next
+        # Atomic: the vec0 rows and their vec_indexed_at flags must land
+        # together. Without the transaction each INSERT self-commits, so a crash
+        # before the batch flag-update leaves vec0 rows with vec_indexed_at nil —
+        # the next backfill re-INSERTs them (the "no DELETE needed" invariant
+        # below only holds inside a transaction), duplicating embeddings that
+        # never self-heal.
+        @db.transaction do
+          rows.each do |row|
+            vector = JSON.parse(row[:embedding_json])
+            blob = vector.pack("f*")
+            # No DELETE needed: vec_indexed_at is nil so these rows can't be in vec0
+            execute_with_params(
+              "INSERT INTO facts_vec(fact_id, embedding) VALUES (?, ?)",
+              row[:id], blob
+            )
+            indexed_ids << row[:id]
+          rescue JSON::ParserError
+            next
+          end
+
+          @store.facts.where(id: indexed_ids).update(vec_indexed_at: now) if indexed_ids.any?
         end
 
-        # Batch-update timestamps
-        @store.facts.where(id: indexed_ids).update(vec_indexed_at: now) if indexed_ids.any?
-
         indexed_ids.size
       end
 

lib/claude_memory/observe/observation_stats.rb

@@ -75,7 +75,10 @@ def source_tokens_for(store)
         return 0 if ids.empty?
 
         bytes = store.content_items.where(id: ids).sum(:byte_len) || 0
-        (bytes / Core::TokenEstimator::CHARS_PER_TOKEN).round
+        # Same estimator (and rounding) the observation token_count denominator
+        # uses (observation_writes.rb), so the compression ratio compares like
+        # with like instead of ceil-vs-round.
+        Core::TokenEstimator.from_chars(bytes)
       end
     end
   end

lib/claude_memory/observe/reflector.rb

@@ -46,7 +46,12 @@ def initialize(store, info_ttl_days: DEFAULT_INFO_TTL_DAYS, matcher: TokenOverla
       def reflect!
         deduped = 0
         expired = 0
-        @store.db.transaction do
+        # Retryable transaction: reflection runs in the sweep hook, which races
+        # the ingest hook for the WAL writer (the documented contention gotcha).
+        # transaction_with_retry retries the whole transaction on SQLITE_BUSY —
+        # the individual mutators must NOT be wrapped, since retrying a single
+        # statement inside an open transaction can't clear the busy lock.
+        @store.transaction_with_retry do
           deduped = dedupe
           expired = expire_stale_info
         end

spec/claude_memory/core/token_budget_spec.rb

@@ -47,5 +47,14 @@
       expect(budget.avg).to eq(0)
       expect(budget.p50).to eq(0)
     end
+
+    it "returns 0 (not nil) for min and max, matching the other aggregates" do
+      expect(budget.min).to eq(0)
+      expect(budget.max).to eq(0)
+    end
+
+    it "is frozen" do
+      expect(budget).to be_frozen
+    end
   end
 end

spec/claude_memory/dashboard/scoped_fact_resolver_spec.rb

@@ -100,4 +100,53 @@ def insert(store, object:, scope:)
       expect(pairs).to contain_exactly(["project", 1], ["project", 2], ["global", 1], ["global", 3])
     end
   end
+
+  describe "batch resolution (build_fact_index + resolve_from_index)" do
+    it "merges many events' scoped ids into one deduped {scope => ids}" do
+      d1 = {"top_facts_by_scope" => {"project" => [1, 2]}}
+      d2 = {"top_facts_by_scope" => {"project" => [2, 3], "global" => [1]}}
+      merged = described_class.merge_scoped_ids([d1, d2])
+      expect(merged["project"]).to eq([1, 2, 3])
+      expect(merged["global"]).to eq([1])
+    end
+
+    it "resolves each event from a shared index with one query per scope, not per event" do
+      p = insert(manager.project_store, object: "P", scope: "project")
+      g = insert(manager.global_store, object: "G", scope: "global")
+      details = [
+        {"top_facts_by_scope" => {"project" => [p]}},
+        {"top_facts_by_scope" => {"global" => [g]}},
+        {"top_facts_by_scope" => {"project" => [p], "global" => [g]}}
+      ]
+      index = described_class.build_fact_index(manager, described_class.merge_scoped_ids(details))
+
+      # Two scopes touched ⇒ index has exactly the two scope buckets.
+      expect(index.keys).to contain_exactly("project", "global")
+      # build_fact_index does not re-query per event; resolve_from_index is pure.
+      expect(manager.project_store).not_to receive(:facts)
+      expect(described_class.resolve_from_index(details[0], index).map { |f| f[:object] }).to eq(%w[P])
+      expect(described_class.resolve_from_index(details[1], index).map { |f| f[:source] }).to eq(%w[global])
+      expect(described_class.resolve_from_index(details[2], index).map { |f| f[:object] }).to eq(%w[P G])
+    end
+
+    it "preserves per-scope input order when resolving from the index" do
+      a = insert(manager.project_store, object: "A", scope: "project")
+      b = insert(manager.project_store, object: "B", scope: "project")
+      c = insert(manager.project_store, object: "C", scope: "project")
+      details = {"top_facts_by_scope" => {"project" => [c, a, b]}}
+      index = described_class.build_fact_index(manager, described_class.merge_scoped_ids([details]))
+      expect(described_class.resolve_from_index(details, index).map { |f| f[:object] }).to eq(%w[C A B])
+    end
+
+    it "returns [] from resolve_from_index for events with no scoped facts" do
+      expect(described_class.resolve_from_index({}, {})).to eq([])
+    end
+
+    it "emits a repeated id once (matching the query-based resolve's row-set dedup)" do
+      p = insert(manager.project_store, object: "P", scope: "project")
+      details = {"top_facts_by_scope" => {"project" => [p, p]}}
+      index = described_class.build_fact_index(manager, described_class.merge_scoped_ids([details]))
+      expect(described_class.resolve_from_index(details, index).map { |f| f[:object] }).to eq(%w[P])
+    end
+  end
 end

spec/claude_memory/index/lexical_fts_spec.rb

@@ -101,6 +101,37 @@
     end
   end
 
+  describe "#rebuild!" do
+    before do
+      [
+        {text: "We are using PostgreSQL as our primary database.", hash: "r1"},
+        {text: "The authentication system uses JWT tokens.", hash: "r2"}
+      ].each do |item|
+        id = store.upsert_content_item(
+          source: "claude_code", session_id: "sess-1",
+          text_hash: item[:hash], byte_len: item[:text].bytesize, raw_text: item[:text]
+        )
+        fts.index_content_item(id, item[:text])
+      end
+    end
+
+    it "rebuilds the index from content_items" do
+      fts.rebuild!
+      expect(fts.search("PostgreSQL")).not_to be_empty
+    end
+
+    it "leaves the existing index intact when the rebuild fails mid-flight (transactional DDL rollback)" do
+      expect(fts.search("PostgreSQL")).not_to be_empty
+
+      # Fail after the DROP but before the reinserts complete; the whole
+      # transaction (incl. the DROP) must roll back, leaving the old index.
+      allow(fts).to receive(:create_contentless_table!).and_raise(RuntimeError, "boom")
+      expect { fts.rebuild! }.to raise_error(RuntimeError, "boom")
+
+      expect(fts.search("PostgreSQL")).not_to be_empty
+    end
+  end
+
   describe "corrupt rank-index handling (issue #7, Finding 2)" do
     it "translates a malformed-on-rank failure into a CorruptRankIndexError with a compact hint" do
       expect {