diff --git a/backend/dashboard_metrics/management/commands/backfill_metrics.py b/backend/dashboard_metrics/management/commands/backfill_metrics.py index 0a88a1be0b..694e656fb6 100644 --- a/backend/dashboard_metrics/management/commands/backfill_metrics.py +++ b/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, diff --git a/backend/dashboard_metrics/tasks.py b/backend/dashboard_metrics/tasks.py index a5dbba5ce4..d52a5a0e84 100644 --- a/backend/dashboard_metrics/tasks.py +++ b/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( diff --git a/backend/dashboard_metrics/tests/test_backfill_metrics.py b/backend/dashboard_metrics/tests/test_backfill_metrics.py index 1e2a77d32e..69d0783d62 100644 --- a/backend/dashboard_metrics/tests/test_backfill_metrics.py +++ b/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 diff --git a/backend/dashboard_metrics/tests/test_tasks.py b/backend/dashboard_metrics/tests/test_tasks.py index 6b238825f9..0e7ab136c4 100644 --- a/backend/dashboard_metrics/tests/test_tasks.py +++ b/backend/dashboard_metrics/tests/test_tasks.py @@ -7,6 +7,7 @@ from types import SimpleNamespace from unittest.mock import patch +from account_usage.models import PageUsage from account_v2.models import Organization from celery.exceptions import SoftTimeLimitExceeded from django.apps import apps @@ -511,6 +512,24 @@ def test_the_floor_keeps_an_org_older_than_the_source_window(self): window_start = self.now - timedelta(days=DASHBOARD_SOURCE_WINDOW_DAYS) assert stale_org.id in _active_org_ids(self.now, window_start) + def test_an_org_with_only_page_usage_is_kept(self): + """Prompt Studio writes page_usage without a workflow execution. + + Shortlisting on executions alone left those orgs with no daily rows at all. + """ + ps_org = Organization.objects.create( + organization_id="ps-only-org", name="ps", display_name="PS only" + ) + usage = PageUsage.objects.create( + organization_id=ps_org.organization_id, pages_processed=3, file_size=1 + ) + PageUsage.objects.filter(pk=usage.pk).update( + created_at=self.now - timedelta(days=1) + ) + + window_start = self.now - timedelta(days=DASHBOARD_SOURCE_WINDOW_DAYS) + assert ps_org.id in _active_org_ids(self.now, window_start) + class TestMonthlyRollupFailurePosture(TestCase): """The rollup's errors must be counted, so success is False without a retry.""" @@ -835,9 +854,7 @@ def test_the_comparison_can_fail_when_the_daily_tier_is_wrong(self): ) assert self._written() == expected - last_month_day = ( - self.now - timedelta(days=self.days_to_last_month_end) - ).date() + last_month_day = (self.now - timedelta(days=self.days_to_last_month_end)).date() corrupted = EventMetricsDaily._base_manager.filter( date=last_month_day, metric_name="documents_processed" ).update(metric_value=99) @@ -1209,15 +1226,25 @@ def test_a_healthy_sibling_metric_is_still_updated(self): """ self._monthly(self.short, value=80) EventMetricsMonthly._base_manager.create( - organization=self.short, month=self.month, metric_name="pages_processed", - metric_type=MetricType.COUNTER, metric_value=5, metric_count=1, - project="default", tag="", + organization=self.short, + month=self.month, + metric_name="pages_processed", + metric_type=MetricType.COUNTER, + metric_value=5, + metric_count=1, + project="default", + tag="", ) self._daily(self.short, self.month, value=70) # short — will be skipped EventMetricsDaily._base_manager.create( - organization=self.short, date=self.month, metric_name="pages_processed", - metric_type=MetricType.COUNTER, metric_value=50, metric_count=1, - project="default", tag="", + organization=self.short, + date=self.month, + metric_name="pages_processed", + metric_type=MetricType.COUNTER, + metric_value=50, + metric_count=1, + project="default", + tag="", ) lowered = _pairs_the_rollup_would_lower(self.month) @@ -1338,12 +1365,14 @@ def test_the_soft_time_limit_is_not_swallowed_by_the_diagnostic(self): then overwrite exactly the totals the guard exists to preserve, and once a short total is stored the lowering check can never see it again. """ - with patch( - "dashboard_metrics.tasks._rollup_monthly_from_daily" - ) as rollup, patch( - "dashboard_metrics.tasks._pairs_the_rollup_would_lower", - side_effect=SoftTimeLimitExceeded(), - ), patch("dashboard_metrics.tasks.WorkflowExecution") as mock_execution: + with ( + patch("dashboard_metrics.tasks._rollup_monthly_from_daily") as rollup, + patch( + "dashboard_metrics.tasks._pairs_the_rollup_would_lower", + side_effect=SoftTimeLimitExceeded(), + ), + patch("dashboard_metrics.tasks.WorkflowExecution") as mock_execution, + ): prefilter = mock_execution.objects.filter.return_value prefilter.values_list.return_value.distinct.return_value = [1] with self.assertRaises(SoftTimeLimitExceeded): @@ -1386,9 +1415,9 @@ def test_a_stale_owner_does_not_release_the_current_holder(self): _release_aggregation_locks([self.key], "older-run") - assert cache.get(self.key) is not None, ( - "a run that no longer owns the key deleted the current holder's lock" - ) + assert ( + cache.get(self.key) is not None + ), "a run that no longer owns the key deleted the current holder's lock" def test_the_owner_does_release_its_own_key(self): """The control: ownership gates the delete, it does not disable it.""" @@ -1418,21 +1447,31 @@ def _run(self, **kwargs): def test_a_lowered_total_reaches_the_result_dict(self): EventMetricsMonthly._base_manager.create( - organization=self.org, month=self.month, - metric_name="documents_processed", metric_type=MetricType.COUNTER, - metric_value=999, metric_count=1, project="default", tag="", + organization=self.org, + month=self.month, + metric_name="documents_processed", + metric_type=MetricType.COUNTER, + metric_value=999, + metric_count=1, + project="default", + tag="", ) EventMetricsDaily._base_manager.create( - organization=self.org, date=self.month, - metric_name="documents_processed", metric_type=MetricType.COUNTER, - metric_value=1, metric_count=1, project="default", tag="", + organization=self.org, + date=self.month, + metric_name="documents_processed", + metric_type=MetricType.COUNTER, + metric_value=1, + metric_count=1, + project="default", + tag="", ) result = self._run(tier=AggregationTier.DAILY_MONTHLY) - assert result["monthly"]["needs_daily_repair"], ( - "the rollup lowered a total and the result dict did not say so" - ) + assert result["monthly"][ + "needs_daily_repair" + ], "the rollup lowered a total and the result dict did not say so" def test_an_incomplete_daily_tier_reaches_the_result_dict(self): """The sibling of the assertion above, for the other check. @@ -1451,16 +1490,21 @@ def test_an_incomplete_daily_tier_reaches_the_result_dict(self): # A row for the month's first day and nothing since: every other whole day # of the month is missing. EventMetricsDaily._base_manager.create( - organization=self.org, date=self.month, - metric_name="documents_processed", metric_type=MetricType.COUNTER, - metric_value=1, metric_count=1, project="default", tag="", + organization=self.org, + date=self.month, + metric_name="documents_processed", + metric_type=MetricType.COUNTER, + metric_value=1, + metric_count=1, + project="default", + tag="", ) result = self._run(tier=AggregationTier.DAILY_MONTHLY) - assert result["monthly"]["incomplete_daily_coverage"], ( - "the daily tier is missing whole days and the result dict did not say so" - ) + assert result["monthly"][ + "incomplete_daily_coverage" + ], "the daily tier is missing whole days and the result dict did not say so" def test_a_failed_coverage_check_is_reported_as_unavailable(self): """The sibling of the lowering check's arm, for the other key. @@ -1627,16 +1671,26 @@ def setUp(self): def _daily(self, day, value): EventMetricsDaily._base_manager.create( - organization=self.org, date=day, metric_name="documents_processed", - metric_type=MetricType.COUNTER, metric_value=value, metric_count=1, - project="default", tag="", + organization=self.org, + date=day, + metric_name="documents_processed", + metric_type=MetricType.COUNTER, + metric_value=value, + metric_count=1, + project="default", + tag="", ) def _monthly(self, month, value): EventMetricsMonthly._base_manager.create( - organization=self.org, month=month, metric_name="documents_processed", - metric_type=MetricType.COUNTER, metric_value=value, metric_count=1, - project="default", tag="", + organization=self.org, + month=month, + metric_name="documents_processed", + metric_type=MetricType.COUNTER, + metric_value=value, + metric_count=1, + project="default", + tag="", ) def test_a_short_month_is_flagged_even_when_its_neighbour_is_whole(self): @@ -1693,7 +1747,9 @@ def setUp(self): self.addCleanup(override.disable) cache.clear() self.addCleanup(cache.clear) - self.keys = _aggregation_lock_keys(AggregationTier.ALL, DASHBOARD_SOURCE_WINDOW_DAYS) + self.keys = _aggregation_lock_keys( + AggregationTier.ALL, DASHBOARD_SOURCE_WINDOW_DAYS + ) def _held(self): return [k for k in self.keys if cache.get(k) is not None] @@ -1796,12 +1852,14 @@ def test_the_fallback_still_applies_with_no_row_to_inherit_from(self): """The control, so the assertions above cannot pass by coincidence.""" self.migration.create_reconciliation_task(apps, None) - assert PeriodicTask.objects.get( - name=self.migration.RECONCILE_TASK_NAME - ).enabled is True - assert PgPeriodicTask.objects.get( - name=self.migration.RECONCILE_TASK_NAME - ).pg_owned is False + assert ( + PeriodicTask.objects.get(name=self.migration.RECONCILE_TASK_NAME).enabled + is True + ) + assert ( + PgPeriodicTask.objects.get(name=self.migration.RECONCILE_TASK_NAME).pg_owned + is False + ) class TestTheTruncatedListNamesTheTotal(TestCase): diff --git a/frontend/src/components/metrics-dashboard/MetricsDashboard.jsx b/frontend/src/components/metrics-dashboard/MetricsDashboard.jsx index 3e814bf027..fe0be77598 100644 --- a/frontend/src/components/metrics-dashboard/MetricsDashboard.jsx +++ b/frontend/src/components/metrics-dashboard/MetricsDashboard.jsx @@ -114,21 +114,20 @@ function MetricsDashboard() { // Fixed 30-day range for activity charts (independent of date picker) const chartStart = useMemo( - () => stableNow.subtract(30, "day").toISOString(), + () => stableNow.subtract(30, "day").format(), [stableNow], ); - const chartEnd = useMemo(() => stableNow.toISOString(), [stableNow]); + const chartEnd = useMemo(() => stableNow.format(), [stableNow]); - // API hooks - pass date range to overview (for summary cards) + // API hooks - pass date range to overview (for summary cards). + // format() keeps the local UTC offset; toISOString() would shift the range a + // day back for users ahead of UTC, because the backend takes .date() on it. const { data: overviewData, loading: overviewLoading, error: overviewError, refetch: refetchOverview, - } = useMetricsOverview( - dateRange[0]?.toISOString(), - dateRange[1]?.toISOString(), - ); + } = useMetricsOverview(dateRange[0]?.format(), dateRange[1]?.format()); // Fixed 30-day data for charts const {