[privacy] surface local filtered-count request summary sink

In-process, thread-safe last-request-summary sink (per account_id) with a
public last_request_summary(account_id:) reader so FilteredCountInstrumentation's
documented local observability signals are consumable instead of discarded.
No APM/New Relic or any off-box egress; sink and reader stay fully in-process.
Adds regression specs asserting the completed summary (status, counts,
duration) is readable and an unknown account returns nil.

app/services/conversations/unread_counts/filtered_count_instrumentation.rb

@@ -65,8 +65,31 @@ class Conversations::UnreadCounts::FilteredCountInstrumentation
       record_increment_summary(operation, attributes) if aggregated_increment?(operation)
     end
 
+    # Local-only observability sink. Consumers may read the most recent completed
+    # request summary for an account in this process; nothing is emitted off-box.
+    def last_request_summary(account_id:)
+      sink = summary_sink
+      summary = sink[:lock].synchronize { sink[:store][account_id] }
+      summary&.dup
+    end
+
     private
 
+    def summary_sink
+      init_lock = (@initialization_lock ||= Mutex.new)
+      init_lock.synchronize do
+        @summary_store_lock ||= Mutex.new
+        @summary_store ||= {}
+      end
+      { lock: @summary_store_lock, store: @summary_store }
+    end
+
+    def store_request_summary(summary)
+      key = summary[:account_id]
+      sink = summary_sink
+      sink[:lock].synchronize { sink[:store][key] = summary.dup }
+    end
+
     def record_observation(operation, attributes, started_at, status:)
       duration_ms = elapsed_ms_since(started_at)
       record_observation_summary(operation, attributes, status: status)
@@ -76,6 +99,7 @@ class Conversations::UnreadCounts::FilteredCountInstrumentation
     def record_request_summary(summary, status, started_at)
       summary[:status] = status
       summary[:duration_ms] = elapsed_ms_since(started_at)
+      store_request_summary(summary)
     end
 
     def request_summary(account_id)

spec/services/conversations/unread_counts/filtered_count_instrumentation_spec.rb

@@ -44,5 +44,25 @@ RSpec.describe Conversations::UnreadCounts::FilteredCountInstrumentation do
         described_class.summarize_request(account_id: 1) { raise error }
       end.to raise_error(error)
     end
+
+    it 'makes the completed request summary readable from the local sink' do
+      described_class.summarize_request(account_id: 99) do
+        described_class.increment(:snapshot_state, account_id: 99, snapshot_scope: :filter, snapshot_status: :missing)
+        described_class.observe(:snapshot_build, account_id: 99, snapshot_scope: :filter) { 'built' }
+        'ok'
+      end
+
+      summary = described_class.last_request_summary(account_id: 99)
+      expect(summary).not_to be_nil
+      expect(summary[:account_id]).to eq(99)
+      expect(summary[:status]).to eq(:success)
+      expect(summary[:snapshot_status_missing_count]).to eq(1)
+      expect(summary[:snapshot_build_success_count]).to eq(1)
+      expect(summary[:duration_ms]).to be_a(Numeric)
+    end
+
+    it 'returns nil from the local sink for an account with no recorded summary' do
+      expect(described_class.last_request_summary(account_id: 999_999)).to be_nil
+    end
   end
 end