Fix filtered dashboard durations

app/services/dashboard_data/snapshots.rb

@@ -2,6 +2,8 @@ module DashboardData
   module Snapshots
     GROUPED_DIMENSIONS = %i[project language editor operating_system category].freeze
     WEEKLY_PROJECT_DIMENSION = "weekly_project".freeze
+    # Keep indexed predecessor probes bounded; broad selections favour one scan.
+    PREDECESSOR_LOOKUP_LIMIT = 1_000
 
     module_function
 
@@ -229,7 +231,102 @@ def activity_graph_date_range(timezone)
       end
     end
 
-    # Live aggregate snapshot used by the filtered (non-rollup) dashboard path.
+    def adaptive_filtered_snapshot(user:, scope:)
+      matches = yield scope.with_valid_timestamps
+      attributed = if matches.reorder(nil).limit(PREDECESSOR_LOOKUP_LIMIT + 1).count <= PREDECESSOR_LOOKUP_LIMIT
+        predecessor_dashboard_scope(timeline: scope, matches: matches)
+      else
+        yield attributed_dashboard_scope(scope)
+      end
+      filtered_query_snapshot(user: user, scope: attributed)
+    end
+
+    def predecessor_dashboard_scope(timeline:, matches:)
+      current_sql = matches.reorder(nil).select(:id, :time, *GROUPED_DIMENSIONS).to_sql
+      previous_sql = timeline.with_valid_timestamps
+        .where("(heartbeats.time, heartbeats.id) < (current_heartbeat.time, current_heartbeat.id)")
+        .reorder(time: :desc, id: :desc).limit(1).select(:time).to_sql
+      timeout = Heartbeat.heartbeat_timeout_duration.to_i
+      attributed_sql = <<~SQL.squish
+        SELECT current_heartbeat.*,
+               LEAST(COALESCE(current_heartbeat.time - previous_heartbeat.time, 0), #{timeout}) AS duration
+        FROM (#{current_sql}) current_heartbeat
+        LEFT JOIN LATERAL (#{previous_sql}) previous_heartbeat ON TRUE
+      SQL
+      Heartbeat.unscoped.from("(#{attributed_sql}) heartbeats")
+    end
+
+    # Date range and archive eligibility belong inside this window; dashboard
+    # dimension filters belong outside it. Each gap belongs to the current row.
+    def attributed_dashboard_scope(scope)
+      timeout = Heartbeat.heartbeat_timeout_duration.to_i
+      timeline = scope.with_valid_timestamps.reorder(nil).select(
+        :id, :time, *GROUPED_DIMENSIONS,
+        Arel.sql("LEAST(COALESCE(time - LAG(time) OVER (ORDER BY time, id), 0), #{timeout}) AS duration")
+      )
+      Heartbeat.unscoped.from(timeline, :heartbeats)
+    end
+
+    def filtered_query_snapshot(user:, scope:)
+      # Materialize after filtering so every aggregate reuses the same small set
+      # of attributed rows instead of sorting/windowing the full timeline again.
+      relation_sql = scope.select(:time, *GROUPED_DIMENSIONS, :duration).to_sql
+      timezone = Heartbeat.connection.quote(user.timezone)
+      local_time = "to_timestamp(time) AT TIME ZONE #{timezone}"
+      ranges = week_ranges(user.timezone)
+      week = "TO_CHAR(DATE_TRUNC('week', #{local_time}), 'YYYY-MM-DD')"
+      slot = "CONCAT(EXTRACT(ISODOW FROM #{local_time})::integer, '-', EXTRACT(HOUR FROM #{local_time})::integer)"
+      aggregates = [ <<~SQL.squish ]
+        SELECT 'total' AS dimension, NULL::text AS bucket, NULL::text AS week_key,
+               COALESCE(SUM(duration), 0) AS duration, COUNT(*) AS heartbeats
+        FROM filtered
+      SQL
+      GROUPED_DIMENSIONS.each do |field|
+        aggregates << <<~SQL.squish
+          SELECT '#{field}', #{field}::text, NULL::text, SUM(duration), NULL::bigint
+          FROM filtered GROUP BY #{field}
+        SQL
+      end
+      aggregates << <<~SQL.squish
+        SELECT 'weekly_project', project::text, #{week}, SUM(duration), NULL::bigint
+        FROM filtered
+        WHERE time BETWEEN #{Heartbeat.connection.quote(ranges.last[1])} AND #{Heartbeat.connection.quote(ranges.first[2])}
+        GROUP BY #{week}, project
+      SQL
+      aggregates << <<~SQL.squish
+        SELECT 'coding_rhythm', #{slot}, NULL::text, SUM(duration), NULL::bigint
+        FROM filtered GROUP BY #{slot}
+      SQL
+
+      rows = Heartbeat.connection.select_all(<<~SQL.squish)
+        WITH filtered AS MATERIALIZED (#{relation_sql})
+        #{aggregates.join(' UNION ALL ')}
+      SQL
+      snapshot = {
+        total_time: 0,
+        total_heartbeats: 0,
+        grouped_durations: GROUPED_DIMENSIONS.index_with { {} },
+        weekly_project_stats: ranges.to_h { |key, *_| [ key, {} ] },
+        coding_rhythm: { timezone: user.timezone, duration_by_slot: {} }
+      }
+      rows.each do |row|
+        duration = row["duration"].to_i
+        case row["dimension"]
+        when "total"
+          snapshot[:total_time] = duration
+          snapshot[:total_heartbeats] = row["heartbeats"].to_i
+        when "weekly_project"
+          snapshot[:weekly_project_stats].fetch(row["week_key"])[row["bucket"]] = duration
+        when "coding_rhythm"
+          snapshot[:coding_rhythm][:duration_by_slot][row["bucket"]] = duration
+        else
+          snapshot[:grouped_durations].fetch(row["dimension"].to_sym)[row["bucket"]] = duration
+        end
+      end
+      snapshot
+    end
+
+    # Live aggregate snapshot used by the unfiltered non-rollup dashboard path.
     # Returns the same shape as the rollup-derived aggregate snapshot.
     def aggregate_query_snapshot(user:, scope:)
       {

app/services/dashboard_stats.rb

@@ -16,7 +16,7 @@ def filterable_dashboard_data
     interval = params[:interval]
     return build_filterable_dashboard_data(interval) if rollup_eligible?
 
-    key = [ user, archived_project_names ] + FILTERS.map { |field| params[field] } + [ interval.to_s, params[:from], params[:to] ]
+    key = [ "attributed_dashboard_v1", user, archived_project_names ] + FILTERS.map { |field| params[field] } + [ interval.to_s, params[:from], params[:to] ]
     Rails.cache.fetch(key, expires_in: 5.minutes) { build_filterable_dashboard_data(interval) }
   end
 
@@ -141,19 +141,24 @@ def query_result(raw_filter_options, archived)
     h = ApplicationController.helpers
 
     Time.use_zone(user.timezone) do
-      hb = filtered_dashboard_heartbeats(raw_filter_options, result: result)
-      hb = hb.filter_by_time_range(params[:interval], params[:from], params[:to])
-      snapshot = DashboardData::Snapshots.aggregate_query_snapshot(user: user, scope: hb)
+      hb = dashboard_heartbeats.filter_by_time_range(params[:interval], params[:from], params[:to])
+      snapshot = if FILTERS.any? { |field| params[field].present? }
+        DashboardData::Snapshots.adaptive_filtered_snapshot(user: user, scope: hb) do |scope|
+          filtered_dashboard_heartbeats(raw_filter_options, result: result, scope: scope)
+        end
+      else
+        DashboardData::Snapshots.aggregate_query_snapshot(user: user, scope: hb)
+      end
       DashboardData::Snapshots.fill_aggregate_result(result: result, snapshot: snapshot, archived: archived, helpers: h)
     end
 
     result
   end
 
-  def filtered_dashboard_heartbeats(filter_options, result: nil)
+  def filtered_dashboard_heartbeats(filter_options, result: nil, scope: dashboard_heartbeats)
     helpers = ApplicationController.helpers
 
-    FILTERS.each_with_object(dashboard_heartbeats) do |field, heartbeats|
+    FILTERS.each_with_object(scope) do |field, heartbeats|
       next unless params[field].present?
 
       selected = params[field].split(",")

test/services/dashboard_stats_test.rb

@@ -152,13 +152,133 @@ def stats.rollups_available? = false
           params: { interval: "custom", to: "2026-04-14", project: "beta" }
         ).filterable_dashboard_data
 
-        assert_equal 60, result[:total_time]
+        # The first beta row owns the capped gap from alpha, then adds 60s.
+        assert_equal 180, result[:total_time]
         assert_equal 2, result.dig(:coding_time_average, :day_count)
-        assert_equal 30.0, result.dig(:coding_time_average, :average_seconds)
+        assert_equal 90.0, result.dig(:coding_time_average, :average_seconds)
       end
     end
   end
 
+  test "language filters retain durations from the complete heartbeat timeline" do
+    with_memory_cache_store do
+      Rails.cache.clear
+      user = create(:user)
+
+      create_heartbeat_at(user, "2026-04-14 09:00:00 UTC", project: "alpha", language: "XML", editor: "vscode", operating_system: "macos", category: "coding")
+      create_heartbeat_at(user, "2026-04-14 09:01:00 UTC", project: "beta", language: "Ruby", editor: "vscode", operating_system: "macos", category: "coding")
+      create_heartbeat_at(user, "2026-04-14 09:02:00 UTC", project: "alpha", language: "XML", editor: "vscode", operating_system: "macos", category: "coding")
+
+      stats = build_stats(user, params: { language: "XML" })
+      def stats.rollups_available? = false
+
+      result = stats.filterable_dashboard_data
+
+      assert_equal 60, result[:total_time]
+      assert_equal({ "alpha" => 60 }, result[:project_durations])
+    end
+  end
+
+  test "combined filters sum attributed rows consistently across every dashboard aggregate" do
+    user = create(:user, timezone: "UTC")
+    travel_to Time.utc(2026, 4, 14, 12) do
+      attributes = { project: "alpha", language: "XML", editor: "vscode", operating_system: "macos", category: "coding" }
+      create_heartbeat_at(user, "2026-04-13 23:59:30 UTC", **attributes)
+      create_heartbeat_at(user, "2026-04-14 09:00:00 UTC", **attributes)
+      create_heartbeat_at(user, "2026-04-14 09:01:00 UTC", **attributes.merge(project: "beta", language: "Ruby"))
+      create_heartbeat_at(user, "2026-04-14 09:02:00 UTC", **attributes)
+      create_heartbeat_at(user, "2026-04-14 09:03:00 UTC", **attributes.merge(editor: "zed"))
+      create_heartbeat_at(user, "2026-04-14 09:04:00 UTC", **attributes.merge(project: "gamma", language: "JSON"))
+      create_heartbeat_at(user, "2026-04-14 09:10:00 UTC", **attributes.merge(project: "gamma", language: "JSON"))
+
+      stats = build_stats(user, params: {
+        interval: "today", project: "alpha,gamma", language: "XML,JSON",
+        editor: "VSCode", operating_system: "macOS", category: "coding"
+      })
+      result = stats.build_filterable_dashboard_data("today")
+
+      assert_equal 240, result[:total_time]
+      assert_equal 4, result[:total_heartbeats]
+      assert_equal({ "alpha" => 60, "gamma" => 180 }, result[:project_durations])
+      assert_equal({ "XML" => 60, "JSON" => 180 }, result["language_stats"])
+      assert_equal({ "coding" => 240 }, result[:coding_category_stats])
+      assert_equal({ "alpha" => 60, "gamma" => 180 }, result[:weekly_project_stats].fetch("2026-04-13"))
+      assert_equal({ "2-9" => 240 }, result[:coding_rhythm][:duration_by_slot])
+    end
+  end
+
+  test "filtered duration timeline excludes archived deleted and other users heartbeats" do
+    user = create(:user)
+    attributes = { project: "alpha", language: "XML", editor: "vscode", operating_system: "macos", category: "coding" }
+    create_heartbeat_at(user, "2026-04-14 09:00:00 UTC", **attributes)
+    create_heartbeat_at(user, "2026-04-14 09:01:00 UTC", **attributes.merge(project: "archived"))
+    create(:project_repo_mapping, user: user, project_name: "archived").archive!
+    deleted = create_heartbeat_at(user, "2026-04-14 09:01:30 UTC", **attributes)
+    deleted.soft_delete
+    create_heartbeat_at(create(:user), "2026-04-14 09:01:45 UTC", **attributes)
+    create_heartbeat_at(user, "2026-04-14 09:02:00 UTC", **attributes)
+
+    result = build_stats(user, params: { language: "XML" }).build_filterable_dashboard_data(nil)
+    assert_equal 120, result[:total_time]
+    assert_equal 2, result[:total_heartbeats]
+    assert_equal({ "alpha" => 120 }, result[:project_durations])
+
+    empty = build_stats(user, params: { language: "Ruby" }).build_filterable_dashboard_data(nil)
+    assert_equal 0, empty[:total_time]
+    assert_equal 0, empty[:total_heartbeats]
+    assert_empty empty[:project_durations]
+    assert_empty empty[:coding_rhythm][:duration_by_slot]
+  end
+
+  test "small and large filtered timelines preserve duration attribution" do
+    [ 3, 1_002 ].each do |heartbeat_count|
+      user = create(:user, timezone: "UTC")
+      first = create(
+        :heartbeat,
+        user: user,
+        time: Time.utc(2026, 4, 14, 9).to_f,
+        project: "alpha",
+        language: "XML",
+        editor: "vscode",
+        operating_system: "macos",
+        category: "coding"
+      )
+      Heartbeat.insert_all!((1...heartbeat_count).map do |offset|
+        first.attributes.except("id").merge(
+          "time" => first.time + offset,
+          "language" => offset == 1 ? "Ruby" : "XML",
+          "fields_hash" => Digest::SHA256.hexdigest("filtered-timeline-#{first.id}-#{offset}")
+        )
+      end)
+
+      result = build_stats(user, params: { language: "XML" }).filterable_dashboard_data
+
+      assert_equal heartbeat_count - 2, result[:total_time]
+      assert_equal heartbeat_count - 1, result[:total_heartbeats]
+      assert_equal({ "alpha" => heartbeat_count - 2 }, result[:project_durations])
+      assert_equal heartbeat_count - 2, result[:coding_category_stats].values.sum
+      assert_equal heartbeat_count - 2, result[:coding_rhythm][:duration_by_slot].values.sum
+    end
+  end
+
+  test "predecessor lookups preserve timestamp ties nil buckets and fractional durations" do
+    user = create(:user, timezone: "Europe/London")
+    time = Time.utc(2026, 4, 13, 22, 59, 59).to_f
+    create(:heartbeat, user: user, time: time, language: "Ruby", project: "beta")
+    create(:heartbeat, user: user, time: time + 1.5, language: "XML", project: nil)
+    create(:heartbeat, user: user, time: time + 1.5, language: "Ruby", project: "beta")
+    create(:heartbeat, user: user, time: time + 1.5, language: "XML", project: nil, entity: "other.xml")
+    create(:heartbeat, user: user, time: time + 2.25, language: "XML", project: nil)
+    expected = DashboardData::Snapshots.filtered_query_snapshot(
+      user: user, scope: DashboardData::Snapshots.attributed_dashboard_scope(user.heartbeats).where(language: "XML")
+    )
+    actual = DashboardData::Snapshots.adaptive_filtered_snapshot(user: user, scope: user.heartbeats) { |scope| scope.where(language: "XML") }
+    assert_equal expected, actual
+    assert_equal 2, actual[:total_time]
+    assert_equal({ nil => 2 }, actual[:grouped_durations][:project])
+    assert_equal({ "2-0" => 2 }, actual[:coding_rhythm][:duration_by_slot])
+  end
+
   test "homepage rollup path falls back to live filter options when filter option rollup is missing" do
     with_memory_cache_store do
       Rails.cache.clear