UN-4515 [FIX] Dashboard Overview: repair window, local-day date range, Prompt Studio-only orgs

backend/dashboard_metrics/management/commands/backfill_metrics.py

@@ -15,10 +15,12 @@
     python manage.py backfill_metrics --days=90 --org-id=5
     python manage.py backfill_metrics --days=7 --dry-run
     python manage.py backfill_metrics --days=90 --active-only
+    python manage.py backfill_metrics --start-date=2026-03-12 --end-date=2026-04-08 \
+        --skip-hourly --skip-monthly
 """
 
 import logging
-from datetime import datetime, timedelta
+from datetime import UTC, date, datetime, time, timedelta
 from typing import Any
 
 from account_v2.models import Organization
@@ -81,6 +83,21 @@ def add_arguments(self, parser):
             default=30,
             help="Number of days to backfill (default: 30)",
         )
+        parser.add_argument(
+            "--start-date",
+            type=str,
+            default=None,
+            help="First UTC day to backfill (YYYY-MM-DD). Overrides --days.",
+        )
+        parser.add_argument(
+            "--end-date",
+            type=str,
+            default=None,
+            help=(
+                "Last UTC day to backfill, inclusive (YYYY-MM-DD). Requires "
+                "--start-date. Default: now."
+            ),
+        )
         parser.add_argument(
             "--org-id",
             type=str,
@@ -137,14 +154,22 @@ def handle(self, *args, **options):
         skip_monthly = options["skip_monthly"]
         active_only = options["active_only"]
 
-        end_date = timezone.now()
-        # Truncated to match the cron's daily_start: an untruncated boundary writes the
-        # oldest day covering only part of it, and the monthly rollup now sums the
-        # persisted daily tier rather than recomputing that day from source.
-        start_date = truncate_to_day(end_date - timedelta(days=days))
+        start_date, end_date = self._resolve_window(
+            days, options["start_date"], options["end_date"]
+        )
+
+        if (
+            options["start_date"]
+            and not skip_monthly
+            and not self._covers_whole_months(start_date, end_date, options["end_date"])
+        ):
+            raise CommandError(
+                "The range covers part of a month, and its monthly total would be "
+                "replaced with only these days. Pass --skip-monthly, or give whole "
+                "months."
+            )
 
         self.stdout.write(f"Backfill period: {start_date.date()} to {end_date.date()}")
-        self.stdout.write(f"Days: {days}")
 
         if skip_daily and not skip_monthly:
             self.stdout.write(
@@ -254,6 +279,51 @@ def handle(self, *args, **options):
                 "incomplete. Re-run before the next aggregation."
             )
 
+    @staticmethod
+    def _resolve_window(
+        days: int, start_day: str | None, end_day: str | None
+    ) -> tuple[datetime, datetime]:
+        """Return the (start, end) instants to backfill, both whole UTC days."""
+        if end_day and not start_day:
+            raise CommandError("--end-date requires --start-date")
+        if not start_day:
+            end_date = timezone.now()
+            # Truncated to match the cron's daily_start: an untruncated boundary
+            # writes the oldest day covering only part of it, and the monthly rollup
+            # sums the persisted daily tier rather than recomputing it from source.
+            return truncate_to_day(end_date - timedelta(days=days)), end_date
+
+        try:
+            start_date = datetime.combine(date.fromisoformat(start_day), time.min, UTC)
+            end_date = timezone.now()
+            if end_day:
+                # The source queries are end-inclusive, so stop just short of the
+                # next midnight: ending on it would rewrite that day with only the
+                # rows stamped at exactly 00:00.
+                next_day = date.fromisoformat(end_day) + timedelta(days=1)
+                end_date = datetime.combine(next_day, time.min, UTC) - timedelta(
+                    microseconds=1
+                )
+        except ValueError as e:
+            raise CommandError(f"Dates must be YYYY-MM-DD: {e}") from e
+        if start_date > end_date:
+            raise CommandError("--start-date is after --end-date")
+        return start_date, end_date
+
+    @staticmethod
+    def _covers_whole_months(
+        start_date: datetime, end_date: datetime, end_day: str | None
+    ) -> bool:
+        """Whether every month in the range is complete in it.
+
+        Without --end-date the range ends now, in the current month, which the
+        aggregation task rederives from daily anyway.
+        """
+        ends_on_month_end = end_day is None or (
+            (end_date + timedelta(microseconds=1)).day == 1
+        )
+        return start_date.day == 1 and ends_on_month_end
+
     def _resolve_org_ids(
         self,
         org_id: str | None,

backend/dashboard_metrics/tasks.py

@@ -15,6 +15,7 @@
 from typing import Any
 from uuid import uuid4
 
+from account_usage.models import PageUsage
 from account_v2.models import Organization
 from celery import shared_task
 from celery.exceptions import SoftTimeLimitExceeded
@@ -854,21 +855,33 @@ def _aggregate_org(
 
 
 def _active_org_ids(end_date: datetime, window_start: datetime) -> set:
-    """Organizations with execution activity in the prefilter lookback.
+    """Organizations with execution or page activity in the prefilter lookback.
 
     Never narrower than the caller's own query window: a widened
     source_window_days must not be prefiltered back down to the default
     lookback, or the reconciliation pass skips the orgs it exists to repair.
+
+    Page usage is unioned in because Prompt Studio records pages without a
+    workflow execution; an org using only Prompt Studio would get no rows.
     """
     cutoff = min(
         window_start,
         end_date - timedelta(days=DASHBOARD_ACTIVE_ORG_LOOKBACK_DAYS),
     )
-    return set(
+    executing = set(
         WorkflowExecution.objects.filter(created_at__gte=cutoff)
         .values_list("workflow__organization_id", flat=True)
         .distinct()
     )
+    # PageUsage keys orgs by their string identifier, not the PK.
+    paging = set(
+        Organization.objects.filter(
+            organization_id__in=PageUsage.objects.filter(created_at__gte=cutoff).values(
+                "organization_id"
+            )
+        ).values_list("id", flat=True)
+    )
+    return executing | paging
 
 
 def _build_result(

backend/dashboard_metrics/tests/test_backfill_metrics.py

@@ -4,7 +4,7 @@
 tier is now derived from what it writes, so a regression here is not self-healing.
 """
 
-from datetime import timedelta
+from datetime import date, timedelta
 from io import StringIO
 from unittest.mock import patch
 
@@ -19,7 +19,7 @@
 from workflow_manager.workflow_v2.models.workflow import Workflow
 
 from dashboard_metrics.management.commands.backfill_metrics import Command
-from dashboard_metrics.models import EventMetricsDaily, Granularity
+from dashboard_metrics.models import EventMetricsDaily, EventMetricsMonthly, Granularity
 from dashboard_metrics.tasks import truncate_to_day
 
 
@@ -238,3 +238,138 @@ def test_a_clean_run_still_reports_complete(self):
             "backfill_metrics", days=1, skip_hourly=True, skip_monthly=True, stdout=out
         )
         assert "BACKFILL COMPLETE" in out.getvalue()
+
+
+class TestExplicitDateRange(TestCase):
+    """--start-date/--end-date repairs a past window that --days cannot reach.
+
+    The source queries are end-inclusive, so a range ending at the next midnight
+    would rewrite the day after it with only the rows stamped at exactly 00:00.
+    """
+
+    def setUp(self):
+        self.org = Organization.objects.create(
+            organization_id="range-org", name="range", display_name="Range"
+        )
+        self.workflow = Workflow.objects.create(
+            workflow_name="range-wf", organization=self.org
+        )
+        self.day = truncate_to_day(timezone.now() - timedelta(days=200))
+
+    def _seed(self, stamp):
+        execution = WorkflowExecution.objects.create(
+            workflow=self.workflow, status=ExecutionStatus.COMPLETED
+        )
+        fe = WorkflowFileExecution.objects.create(
+            workflow_execution=execution,
+            file_name=f"{stamp.isoformat()}.pdf",
+            status=ExecutionStatus.COMPLETED.value,
+        )
+        WorkflowFileExecution.objects.filter(pk=fe.pk).update(created_at=stamp)
+        WorkflowExecution.objects.filter(pk=execution.pk).update(created_at=stamp)
+
+    def _daily(self, day):
+        return EventMetricsDaily._base_manager.get(
+            organization=self.org, date=day.date(), metric_name="documents_processed"
+        ).metric_value
+
+    def test_the_range_is_written_whole_and_the_next_day_is_untouched(self):
+        next_day = self.day + timedelta(days=1)
+        self._seed(self.day + timedelta(hours=1))
+        self._seed(self.day + timedelta(hours=23))
+        self._seed(next_day)  # exactly midnight: inside an end-inclusive midnight end
+        EventMetricsDaily._base_manager.create(
+            organization=self.org,
+            date=next_day.date(),
+            metric_name="documents_processed",
+            project="default",
+            tag="",
+            metric_value=5,
+            metric_count=1,
+        )
+
+        call_command(
+            "backfill_metrics",
+            start_date=self.day.date().isoformat(),
+            end_date=self.day.date().isoformat(),
+            skip_hourly=True,
+            skip_monthly=True,
+            stdout=StringIO(),
+        )
+
+        assert self._daily(self.day) == 2
+        assert self._daily(next_day) == 5, "the day after the range was overwritten"
+
+    def test_end_date_without_start_date_is_refused(self):
+        with self.assertRaises(CommandError):
+            call_command("backfill_metrics", end_date="2026-04-08", stdout=StringIO())
+
+
+class TestAPartialMonthRangeKeepsTheMonthlyTotal(TestCase):
+    """A range covering part of a month must not replace that month's total.
+
+    The monthly row would be rewritten with only the days in the range, and a
+    month outside the cron's current-and-previous window is never rederived.
+    """
+
+    def setUp(self):
+        self.org = Organization.objects.create(
+            organization_id="month-org", name="month", display_name="Month"
+        )
+        EventMetricsMonthly._base_manager.create(
+            organization=self.org,
+            month=date(2026, 3, 1),
+            metric_name="documents_processed",
+            project="default",
+            tag="",
+            metric_value=1000,
+            metric_count=31,
+        )
+
+    def _monthly(self):
+        return EventMetricsMonthly._base_manager.get(
+            organization=self.org,
+            month=date(2026, 3, 1),
+            metric_name="documents_processed",
+        ).metric_value
+
+    def test_a_partial_month_without_skip_monthly_is_refused(self):
+        """A partial first month, and a partial last month."""
+        for start, end in (("2026-03-12", "2026-04-08"), ("2026-03-01", "2026-03-20")):
+            with self.subTest(start=start, end=end):
+                with self.assertRaises(CommandError) as caught:
+                    call_command(
+                        "backfill_metrics",
+                        start_date=start,
+                        end_date=end,
+                        skip_hourly=True,
+                        stdout=StringIO(),
+                    )
+                assert "--skip-monthly" in str(caught.exception)
+                assert self._monthly() == 1000
+
+    def test_whole_months_may_rewrite_the_monthly_total(self):
+        """The control: a range of whole months is a complete source for them."""
+        workflow = Workflow.objects.create(
+            workflow_name="month-wf", organization=self.org
+        )
+        execution = WorkflowExecution.objects.create(
+            workflow=workflow, status=ExecutionStatus.COMPLETED
+        )
+        fe = WorkflowFileExecution.objects.create(
+            workflow_execution=execution,
+            file_name="march.pdf",
+            status=ExecutionStatus.COMPLETED.value,
+        )
+        stamp = timezone.make_aware(timezone.datetime(2026, 3, 15, 12))
+        WorkflowFileExecution.objects.filter(pk=fe.pk).update(created_at=stamp)
+        WorkflowExecution.objects.filter(pk=execution.pk).update(created_at=stamp)
+
+        call_command(
+            "backfill_metrics",
+            start_date="2026-03-01",
+            end_date="2026-03-31",
+            skip_hourly=True,
+            stdout=StringIO(),
+        )
+        assert self._monthly() == 1