feat: move admin analytics reads to daily facts

dashboard/backend/api/routers/admin_analytics.py

@@ -49,7 +49,6 @@
 _PROVIDER_ID_PATTERN = re.compile(r"^[a-z0-9_]{2,64}$")
 _MODEL_ID_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._/\-:]{0,255}$")
 _POSITIVE_INTEGER_PATTERN = re.compile(r"^[0-9]+$")
-_USER_STATES = {"blocked", "needs_attention", "dormant", "onboarding", "active"}
 _ACTIVITY_SECTIONS = {"timeline", "runs", "usage", "sessions"}
 _LIFECYCLE_SEGMENTS = {"new", "onboarding", "growing", "core", "at_risk", "dormant"}
 _OPERATIONAL_STATES = {"blocked", "needs_attention", "healthy"}
@@ -209,7 +208,6 @@ def _value_user_filters(request: Request) -> tuple[UserValueFilters, int, int]:
         request,
         {
             "q",
-            "status",
             "lifecycle_segment",
             "operational_state",
             "commercial_tier",
@@ -241,9 +239,6 @@ def _value_user_filters(request: Request) -> tuple[UserValueFilters, int, int]:
             user_group = parse_user_group(user_group)
         except ValueError:
             _invalid_query()
-    legacy_status = values.get("status")
-    if legacy_status is not None and legacy_status not in _USER_STATES:
-        _invalid_query()
 
     from_date = (
         _parse_date(values["last_meaningful_activity_from"])
@@ -278,7 +273,6 @@ def _value_user_filters(request: Request) -> tuple[UserValueFilters, int, int]:
             priority=(
                 _parse_bool(values["priority"]) if "priority" in values else False
             ),
-            legacy_status=legacy_status,
             include_internal=(
                 _parse_bool(values["include_internal"])
                 if "include_internal" in values

dashboard/backend/api/routers/backtests.py

@@ -1194,7 +1194,7 @@ def _finalize_slot_locked(
         # Still `backtest_failed`: the user asked for a backtest, got nothing,
         # and was billed. A separate event name would need nine registrations
         # and would drop timeouts out of `metrics.py`'s `_TERMINAL_FAILURE`,
-        # `states.py`'s consecutive-failure alert and `backfill.py`'s terminal
+        # `lifecycle.py`'s consecutive-failure alert and `backfill.py`'s terminal
         # map -- the three things that exist to notice runs failing. What was
         # missing is the reason, not the outcome.
         return {

dashboard/backend/app.py

@@ -355,34 +355,18 @@ def bar_cache_background():
     except Exception as e:
         print(f"⚠️ Research sweeper start error: {e}")
 
-    try:
-        from dashboard.backend.domain.analytics.maintenance import (
-            run_analytics_maintenance,
-        )
-        from dashboard.backend.domain.runs.service import register_reaper_sweep
-
-        register_reaper_sweep(run_analytics_maintenance)
-        print("🧹 Analytics maintenance sweep registered with the reaper")
-    except Exception as e:
-        print(
-            "WARNING: analytics.maintenance_registration_failed "
-            f"category={type(e).__name__}"
-        )
-
     try:
         # Admin layer redesign PR A (design D23, SS6.9): the daily-facts job
         # runs on its own thread, not as a reaper sweep -- a whole-population
         # batch across three databases on the heartbeat thread would let a
         # slow analytics night mark live runs as orphaned. The worker also
-        # owns rollup_day and the retention coordinator now, and runs the
-        # idempotent user_activity seed + history copy once before ticking.
+        # owns rollup_day and the retention coordinator now, and PR B's
+        # startup migration (re-seed user_activity, roll up and derive the
+        # legacy history, drop the snapshot tables) runs on it before its
+        # first tick -- never here, where it would block the event loop.
         from dashboard.backend.domain.analytics.daily_job import (
             start_daily_facts_worker,
         )
-        from dashboard.backend.domain.analytics.facts_migration import (
-            run_startup_migrations,
-        )
-        run_startup_migrations()
         start_daily_facts_worker()
         print("📊 Analytics daily-facts worker started")
     except Exception as e:

dashboard/backend/database.py

@@ -1018,6 +1018,12 @@ def aggregate_operator_cost_for_day(self, day: date) -> Dict[int, int]:
         ``est_cost_usd`` is a float; converted with ``round(value * 1_000_000)``
         and clamped at zero so a negative stored value cannot violate
         ``user_daily_facts.operator_cost_micro``'s CHECK.
+
+        BYOK runs are excluded: ``engine.py`` stamps ``est_cost_usd`` on them
+        too, from the user's own provider bill, which is not operator spend
+        (design SS6.7 "for operator-funded runs"). The lane is read from
+        ``metadata.llm_execution.billing_mode``; a run that records none (rule
+        based, pre-handoff) or whose metadata is not valid JSON keeps counting.
         """
         start = f"{day.isoformat()} 00:00:00"
         end = f"{(day + timedelta(days=1)).isoformat()} 00:00:00"
@@ -1029,6 +1035,12 @@ def aggregate_operator_cost_for_day(self, day: date) -> Dict[int, int]:
             FROM agent_runs
             WHERE owner_user_id IS NOT NULL
               AND updated_at >= ? AND updated_at < ?
+              AND COALESCE(
+                    CASE WHEN json_valid(metadata)
+                         THEN json_extract(metadata, '$.llm_execution.billing_mode')
+                    END,
+                    ''
+                  ) <> 'byok'
             GROUP BY owner_user_id
             """,
             (start, end),

dashboard/backend/database_postgres.py

@@ -1105,7 +1105,9 @@ def get_runs_by_mode(self, mode: str) -> List[Dict]:
         return [BacktestDatabase._parse_run_row(row) for row in rows]
 
     def aggregate_operator_cost_for_day(self, day: date) -> Dict[int, int]:
-        """See the SQLite twin."""
+        """See the SQLite twin. ``metadata`` is TEXT here as well; ``IS JSON
+        OBJECT`` (PostgreSQL 16+, prod and CI run 18) guards the cast so one
+        malformed row cannot fail the whole day's statement."""
         start = f"{day.isoformat()} 00:00:00"
         end = f"{(day + timedelta(days=1)).isoformat()} 00:00:00"
         with self._get_connection() as conn:
@@ -1116,6 +1118,12 @@ def aggregate_operator_cost_for_day(self, day: date) -> Dict[int, int]:
                     FROM agent_runs
                     WHERE owner_user_id IS NOT NULL
                       AND updated_at >= %s AND updated_at < %s
+                      AND COALESCE(
+                            CASE WHEN metadata IS JSON OBJECT
+                                 THEN metadata::jsonb -> 'llm_execution' ->> 'billing_mode'
+                            END,
+                            ''
+                          ) <> 'byok'
                     GROUP BY owner_user_id
                     """,
                     (start, end),

dashboard/backend/domain/analytics/backfill.py

@@ -108,7 +108,6 @@ class BackfillReport(BaseModel):
     write_failures: int = Field(default=0, ge=0)
     repair_failures: int = Field(default=0, ge=0)
     affected_users: int = Field(default=0, ge=0)
-    repaired_snapshots: int = Field(default=0, ge=0)
     rebuilt_rollup_days: int = Field(default=0, ge=0)
     source_event_ids: list[str] = Field(default_factory=list)
 
@@ -509,13 +508,20 @@ def backfill_analytics(
     dry_run: bool = False,
     source: BackfillSource | None = None,
     store: Any = analytics_store,
-    recalculate_snapshot: Any | None = None,
     rebuild_rollup: Any | None = None,
 ) -> BackfillReport:
     """Reconstruct safe events inside ``[now - days, before]``.
 
-    Bulk repairs run once per affected subject/day after all inserts complete.
+    Rollup repairs run once per affected day after all inserts complete.
     Dry runs enumerate and validate sources but perform no writes at all.
+
+    Inserted events reach the raw log and the day rollups only. They do not
+    update ``user_activity`` (activation date, last meaningful activity) or
+    ``user_daily_facts``: events go in through ``store.append_event``, not the
+    ingestion path that maintains ``user_activity``, and the daily job's
+    late-arrival sweep recomputes facts only for the last
+    ``RECOMPUTE_LOOKBACK_DAYS`` days. A historical backfill therefore moves no
+    one's activation date or lifecycle segment.
     """
 
     if isinstance(days, bool) or not isinstance(days, int) or not 1 <= days <= 180:
@@ -593,14 +599,6 @@ def backfill_analytics(
         if candidate.occurred_at.date() < current.date():
             affected_days.add(candidate.occurred_at.date())
 
-    if recalculate_snapshot is None:
-        from .states import AnalyticsStateStore, recalculate_user_snapshot
-
-        state_store = AnalyticsStateStore(store)
-
-        def recalculate_snapshot(user_id: int, **kwargs: Any) -> Any:
-            return recalculate_user_snapshot(user_id, store=state_store, **kwargs)
-
     if rebuild_rollup is None:
         from .rollups import AnalyticsRollupStore, rollup_day
 
@@ -610,14 +608,7 @@ def rebuild_rollup(day: date, **kwargs: Any) -> Any:
             return rollup_day(day, store=rollup_store, **kwargs)
 
     repair_failures = 0
-    repaired_snapshots = 0
     rebuilt_rollup_days = 0
-    for user_id in sorted(affected_users):
-        try:
-            recalculate_snapshot(user_id, now=current)
-            repaired_snapshots += 1
-        except Exception:
-            repair_failures += 1
     for day in sorted(affected_days):
         try:
             rebuild_rollup(day, now=datetime.combine(
@@ -640,7 +631,6 @@ def rebuild_rollup(day: date, **kwargs: Any) -> Any:
         write_failures=write_failures,
         repair_failures=repair_failures,
         affected_users=len(affected_users),
-        repaired_snapshots=repaired_snapshots,
         rebuilt_rollup_days=rebuilt_rollup_days,
         source_event_ids=inserted_ids,
     )

dashboard/backend/domain/analytics/daily_facts.py

@@ -2,7 +2,7 @@
 
 Every query here is parameterised by a date, never by a user id. That is the
 property the read-budget test enforces, and it is the difference between this
-module and the snapshot sweep it replaces: cost grows with days, not with
+module and the per-user snapshot sweep it replaced: cost grows with days, not with
 users times minutes (design SS6.9, SS6.12).
 
 Scheduling lives in ``daily_job.py`` (its own worker thread, D23). This module
@@ -110,19 +110,62 @@ def _count(totals: Any, field: str) -> int:
     return int(getattr(totals, field)) if totals is not None else 0
 
 
+def _rederive_next_day_transitions(
+    day: date, facts: list[UserDailyFact], *, store: Any, now: datetime
+) -> int:
+    """Rebuild ``day + 1``'s transition log against ``day``'s rewritten facts.
+
+    A transition's ``from_segment`` is the previous day's segment, so
+    recomputing a past day can invalidate the next day's log without
+    touching its facts. Only the late-arrival sweep calls this (the target
+    day's successor is today, which has no facts). ``inactive_days`` belongs
+    to the later day's own evaluation, so an existing row keeps its value; a
+    move that only appears now has none to keep and takes the column's 0.
+    """
+    next_day = day + timedelta(days=1)
+    later = store.list_facts_for_date(next_day)
+    if not later:
+        return 0
+    existing = {row.user_id: row for row in store.list_transitions_for_date(next_day)}
+    segment_on_day = {row.user_id: row.lifecycle_segment for row in facts}
+    rows: list[LifecycleTransitionRow] = []
+    for fact in later:
+        before = segment_on_day.get(fact.user_id)
+        if before is None or before == fact.lifecycle_segment:
+            continue
+        kept = existing.get(fact.user_id)
+        rows.append(
+            LifecycleTransitionRow(
+                user_id=fact.user_id,
+                snapshot_date=next_day,
+                from_segment=before,
+                to_segment=fact.lifecycle_segment,
+                inactive_days=kept.inactive_days if kept is not None else 0,
+                data_quality=fact.data_quality,
+                created_at=now,
+            )
+        )
+    return store.replace_lifecycle_transitions(next_day, rows)
+
+
 def _compute_day(
     day: date,
     *,
     now: datetime,
     store: Any,
     run_history_store: Any,
     rollup: Callable[..., Any],
+    rederive_next: bool = False,
 ) -> _DayOutcome:
     """Steps 1-6 of design SS6.9 for one day, each wrapped individually.
 
     Step order differs from the design's numbering in one place: the eligible
     population is read first, because the operational step (5) answers for
     exactly those ids. Query count is unchanged.
+
+    ``rederive_next`` is set for a late-arrival recompute, whose successor
+    day already has facts and therefore a transition log keyed to this
+    day's segments.
     """
     from .rollups import AnalyticsRollupStore
 
@@ -158,7 +201,7 @@ def subjects_step() -> list[dict[str, Any]]:
         return _DayOutcome(failed_steps=tuple(failed))
     eligible_ids = [int(subject["id"]) for subject in subjects]
 
-    # 1. Anonymous rollups for D (moved here from run_analytics_maintenance).
+    # 1. Anonymous rollups for D.
     step(
         "rollup",
         lambda: rollup(day, store=AnalyticsRollupStore(analytics), now=_midnight(now.date())),
@@ -320,7 +363,14 @@ def transitions_step() -> int:
                     created_at=now,
                 )
             )
-        return store.append_lifecycle_transitions(changed)
+        # Whole-day replace: a recompute that puts a user back on yesterday's
+        # segment must also remove the move the first run recorded.
+        written = store.replace_lifecycle_transitions(day, changed)
+        if rederive_next:
+            written += _rederive_next_day_transitions(
+                day, [row for row, _days in rows_with_days], store=store, now=now
+            )
+        return written
 
     transitions = step("transitions", transitions_step, 0)
     return _DayOutcome(
@@ -385,14 +435,22 @@ def run_daily_facts(
                 "WARNING: analytics.daily_facts.recompute_scan_failed "
                 f"category={type(exc).__name__[:80]}"
             )
-        # One store call after the loop, never inside it: the lease belongs to
+        # Store calls after the loop, never inside it: the lease belongs to
         # the job row, not the day, and the discipline guard forbids store
-        # calls inside loops (design SS6.11 rule 7). A failed stale day stays
-        # in list_days_needing_recompute, so the next tick retries it.
+        # calls inside loops (design SS6.11 rule 7). A failed stale day is
+        # re-queued explicitly: a step that failed after the facts upsert had
+        # already refreshed calculated_at, which is what takes a day off
+        # list_days_needing_recompute -- so without it, the day's partial rows
+        # (or a stale next-day transition log) would stand for good.
         failed_recomputes: list[date] = []
         for stale_day in stale_days[:MAX_RECOMPUTES_PER_TICK]:
             stale_outcome = _compute_day(
-                stale_day, now=current, store=store, run_history_store=runs, rollup=rollup_fn
+                stale_day,
+                now=current,
+                store=store,
+                run_history_store=runs,
+                rollup=rollup_fn,
+                rederive_next=True,
             )
             if stale_outcome.failed_steps:
                 failed_recomputes.append(stale_day)
@@ -403,6 +461,7 @@ def run_daily_facts(
                 continue
             recomputed.append(stale_day)
         if failed_recomputes:
+            store.requeue_days_for_recompute(failed_recomputes)
             store.release_projection_day(DAILY_FACTS_JOB, now=current)
 
         report = DailyFactsReport(

dashboard/backend/domain/analytics/daily_job.py

@@ -7,9 +7,11 @@
 inside the single web process (design SS2.3): no external scheduler, no second
 service.
 
-Every tick is ``daily_facts.run_daily_facts``; the first thing the thread does,
-once, is ``facts_migration.run_startup_migrations`` (seed ``user_activity``,
-copy the legacy history). Both are idempotent, so a restart costs nothing.
+Every tick is ``daily_facts.run_daily_facts``; the first thing the thread does
+is ``facts_migration.run_startup_migrations`` (re-seed ``user_activity``, roll
+up and derive the legacy history, then drop the legacy snapshot tables), and
+until that succeeds it is retried before each tick. Both are idempotent, so a
+restart costs nothing.
 """
 
 from __future__ import annotations
@@ -98,15 +100,29 @@ def start_daily_facts_worker(
         tick_fn = tick if tick is not None else _default_tick
         prepare_fn = prepare if prepare is not None else _default_prepare
 
-        def _loop() -> None:
+        def _prepare() -> bool:
             try:
-                prepare_fn()
+                report = prepare_fn()
             except Exception as exc:
                 print(
                     "WARNING: analytics.facts_migration_failed "
                     f"category={type(exc).__name__[:80]}"
                 )
+                return False
+            if report is not None:
+                print(f"🧹 Analytics startup migration: {report!r}", flush=True)
+            return True
+
+        def _loop() -> None:
+            # The only caller of the startup migration (app.py used to run it
+            # synchronously as well, on the event loop). A failure is retried
+            # before each tick rather than only at the next boot: the drop it
+            # guards is the step most likely to lose a lock to the outgoing
+            # instance on a deploy.
+            prepared = _prepare()
             while not stop.wait(interval):
+                if not prepared:
+                    prepared = _prepare()
                 try:
                     tick_fn()
                 except Exception as exc: