feat(data): auto-refresh stale data on the analyzer entrypoint

After every code update that adds a new field to the importer's
output shape (e.g. v0.3.0 boss_casts on fights, mana_recovery hook
on healers, the v0.2.0 stats/talent_ids on players), already-imported
data lacks the new fields and the AI prompt silently loses features.
This adds an automatic version-tag check so the next analysis touches
the affected rows and triggers a fresh fetch transparently.

Design
- Two integer-version constants in code, kept next to their importers:
    ``REPORT_DATA_VERSION = 3``        in ``report_service``
    ``TOP_LOG_DETAIL_VERSION = 2``     in ``top_logs_service``
  Each is documented with a per-version changelog. Devs bump when they
  change the shape — same discipline as alembic migration numbers.
  ``wow_localizations`` already has the equivalent via
  ``EXPECTED_KINDS`` (see ``workers/tasks/wow_data``); no extra
  constant needed.
- Every write path stamps ``extras['_v']`` / ``detail_payload['_v']``
  with the current constant. Missing ``_v`` reads as version 0 →
  always stale.
- ``run_report_import`` is now smart-idempotent: if a report's
  fights all carry the current ``_v`` it short-circuits as before; if
  any are below, it wipes the fights (cascade to players/casts/gear/
  analyses) and re-imports cleanly.
- ``refresh_top_logs_for_spec_encounter`` 's incremental cache
  snapshot drops stale detail blobs so the fetch loop re-pulls them;
  the post-refresh DELETE+INSERT pass then writes them back with the
  current version. The analyzer's ``_fetch_top_log_references`` also
  filters stale rows out at query time, so a fully-stale (spec,
  encounter, metric) bucket trips the empty-refs path and naturally
  triggers ``_ensure_references`` → fresh re-seed.
- ``request_analysis`` runs ``report_data_is_current`` as its first
  step. If stale, it auto-invokes ``run_report_import`` and re-loads
  the fight/player rows. Failures are logged and fall through with
  whatever data we have rather than blocking the analysis.

Verified end-to-end against a real WCL report:
1) Fresh import: fight + player extras carry ``_v=3``.
2) Manually downgrade ``_v=1`` on every fight → ``report_data_is_current``
   reports False as expected.
3) Re-run ``run_report_import`` on the stale report → fights are wiped,
   re-imported, and the new fight extras carry ``_v=3`` again.
4) Same loop on top_logs: fresh refresh stamps ``_v=2``, downgrade to
   ``_v=1`` is detected, the next incremental refresh skips stale
   cache entries and re-fetches detail; all rows end up at ``_v=2``.

User-visible effect: first analysis after an upgrade on a previously-
imported report or stale (spec, encounter, metric) takes one cycle
longer (~30-90s) while the refresh runs; subsequent analyses are
unchanged. No manual ``DELETE FROM`` dance after every release.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: babatonga

backend/app/services/ai/analyzer.py

@@ -224,6 +224,18 @@ async def _fetch_top_log_references(
         .limit(limit)
     )
     rows = (await session.execute(stmt)).scalars().all()
+    # Drop rows whose detail blob was written by an older code shape —
+    # they'd feed the AI a payload missing fields (e.g. boss_casts on
+    # pre-v0.3.0 detail). When all rows are stale this returns ``[]``,
+    # which makes the caller fall through to ``_ensure_references`` →
+    # incremental refresh → fresh detail with the current version stamp.
+    from app.services.top_logs_service import TOP_LOG_DETAIL_VERSION as _DV
+
+    rows = [
+        r
+        for r in rows
+        if int((r.detail_payload or {}).get("_v") or 0) >= _DV
+    ]
     out: list[dict[str, Any]] = []
     for r in rows:
         detail = r.detail_payload or {}
@@ -634,6 +646,67 @@ async def request_analysis(
     if not report or report.id != fight.report_id:
         raise NotFoundError("Report mismatch.")
 
+    # ----- Auto-refresh stale data before running the analysis ---------------
+    # When the code shipped a new field since this report (or its cached
+    # top-log references) was imported, the analyser would otherwise read
+    # missing fields and the AI prompt would silently lose features.
+    # ``REPORT_DATA_VERSION`` / ``TOP_LOG_DETAIL_VERSION`` are bumped by the
+    # importer code when the shape changes; we compare against the stored
+    # ``_v`` stamp and re-fetch when behind. Failures are logged but never
+    # blocking — we proceed with whatever data we have, the AI degrades
+    # gracefully on missing fields.
+    from app.services.report_service import (
+        REPORT_DATA_VERSION as _RDV,
+        report_data_is_current,
+        run_report_import,
+    )
+
+    if not await report_data_is_current(session, report.id):
+        logger.info(
+            "report %s data older than v%d — auto-refreshing before analysis",
+            report.id,
+            _RDV,
+        )
+        try:
+            async with WclClient() as wcl:
+                await run_report_import(session, report_id=report.id, wcl_client=wcl)
+            await session.commit()
+            # Re-load fight + player after the import wiped + re-inserted them.
+            fight = (
+                await session.execute(
+                    select(ReportFight).where(
+                        ReportFight.report_id == report.id,
+                        ReportFight.fight_id == fight.fight_id,
+                    )
+                )
+            ).scalar_one_or_none()
+            player = (
+                await session.execute(
+                    select(ReportPlayer)
+                    .where(
+                        ReportPlayer.fight_id == fight.id if fight else False,
+                        ReportPlayer.actor_id == player.actor_id,
+                    )
+                    .options(
+                        selectinload(ReportPlayer.casts),
+                        selectinload(ReportPlayer.gear),
+                    )
+                )
+            ).scalar_one_or_none()
+            if not fight or not player:
+                raise UpstreamError(
+                    "Report data was refreshed but the original fight/player "
+                    "rows could not be re-located — the underlying WCL log "
+                    "may have changed."
+                )
+        except UpstreamError:
+            raise
+        except Exception:  # noqa: BLE001
+            logger.exception(
+                "Auto-refresh of report %s failed — continuing with stale data",
+                report.id,
+            )
+
     role_focus = "healer" if player.role == "healer" else ("tank" if player.role == "tank" else "dps")
 
     # Lazy-fetch extras (buffs / debuffs / damage_taken) from WCL the first time

backend/app/services/report_service.py

@@ -4,7 +4,7 @@
 import logging
 from typing import Any
 
-from sqlalchemy import select
+from sqlalchemy import delete, select
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import selectinload
 
@@ -38,6 +38,23 @@
 
 logger = logging.getLogger(__name__)
 
+# Bumped whenever the shape of ``ReportFight.extras`` / ``ReportPlayer.extras``
+# / ``ReportPlayerGear`` / ``ReportPlayerCast`` changes in a way the AI prompt
+# or analyser needs. Stored verbatim into ``extras['_v']`` on every fresh
+# import; ``run_report_import`` re-imports a report when any fight's stored
+# ``_v`` is below this value. Missing ``_v`` is treated as version 0
+# (= ancient), which forces a refresh.
+#
+# History:
+#   v1: initial shape (basic per-fight/per-player rollups).
+#   v2 (v0.2.0): added talent_ids/stats/active_time_ms in player.extras,
+#                gear rows in report_player_gear, talents_loadout JSONB.
+#   v3 (v0.3.0): added wcl_fight_start_ms + fight-relative-second
+#                phase_transitions + boss_casts on fight.extras;
+#                mana_recovery is lazy-fetched at analysis time so it
+#                doesn't gate this version.
+REPORT_DATA_VERSION = 3
+
 
 async def fetch_boss_cast_timeline(
     client: WclClient,
@@ -114,6 +131,28 @@ async def create_import_skeleton(
     return report
 
 
+async def report_data_is_current(session: AsyncSession, report_id: Any) -> bool:
+    """Return ``True`` iff every fight on the report carries the current
+    ``REPORT_DATA_VERSION`` stamp. Empty (no fights) → ``False`` because
+    a freshly-skeletoned but never-imported report counts as not current.
+
+    Missing ``_v`` (older code's writes) is treated as version 0, which is
+    always below the current constant — so legacy data automatically
+    triggers a refresh on the next analysis without any per-row migration.
+    """
+    fights = (
+        await session.execute(
+            select(ReportFight.extras).where(ReportFight.report_id == report_id)
+        )
+    ).all()
+    if not fights:
+        return False
+    return all(
+        int((extras or {}).get("_v") or 0) >= REPORT_DATA_VERSION
+        for (extras,) in fights
+    )
+
+
 async def run_report_import(
     session: AsyncSession,
     *,
@@ -124,15 +163,34 @@ async def run_report_import(
 
     Drives ``import_status``: ``importing`` → ``ready`` (or ``failed`` on
     exception). Idempotent — calling on a ``ready`` row that already has
-    fights does nothing besides refreshing the status.
+    fights does nothing besides refreshing the status, UNLESS the stored
+    data was written by an older code version (different
+    ``REPORT_DATA_VERSION``). In that case the existing fights are wiped
+    and a full re-import runs so the analyser can rely on the current
+    fight/player extras shape.
     """
     report = (
         await session.execute(select(Report).where(Report.id == report_id))
     ).scalar_one_or_none()
     if report is None:
         raise NotFoundError("Report not found.")
     if report.import_status == "ready" and report.start_time is not None:
-        return  # already populated; nothing to do
+        if await report_data_is_current(session, report_id):
+            return
+        logger.info(
+            "report %s data is older than current version %d — re-importing",
+            report_id,
+            REPORT_DATA_VERSION,
+        )
+        # Cascade through fights → players → casts/gear/analyses. We keep
+        # the Report row itself (analyses referencing it survive the
+        # version bump cleanly — their historical structured output is
+        # unchanged, only the *next* analysis sees fresh data).
+        await session.execute(
+            delete(ReportFight).where(ReportFight.report_id == report_id)
+        )
+        report.import_status = "importing"
+        await session.flush()
 
     code = report.wcl_code
     owner_user_id = report.owner_user_id
@@ -178,8 +236,12 @@ async def run_report_import(
                     fight_start_ms=fight_start_ms,
                 )
                 # Mutate in place — JSONB will pick this up on the next flush.
+                # ``_v`` is the version stamp the analyser checks to decide
+                # whether this row's shape matches what the current code
+                # expects to read.
                 merged_extras = dict(fight_row.extras or {})
                 merged_extras["boss_casts"] = boss_casts
+                merged_extras["_v"] = REPORT_DATA_VERSION
                 fight_row.extras = merged_extras
             await session.flush()
 
@@ -360,6 +422,14 @@ async def _populate_players(
                         "rank": None,
                         "out_of": None,
                     },
+                    # Version stamp matched against ``REPORT_DATA_VERSION`` —
+                    # if a future code update bumps the constant, the
+                    # analyser sees this player as stale and triggers a
+                    # full re-import for the parent report. Lazy-fetched
+                    # fields (buffs/debuffs/casts/mana_recovery) are NOT
+                    # gated on this number; only the shape produced by the
+                    # import flow is.
+                    "_v": REPORT_DATA_VERSION,
                 },
             )
             session.add(db_player)

backend/app/services/top_logs_service.py

@@ -46,6 +46,22 @@
 
 logger = logging.getLogger(__name__)
 
+# Bumped whenever ``TopLog.detail_payload`` shape gains a new field the
+# analyser depends on. The incremental refresh path snapshots existing
+# detail rows for cache-reuse — stale rows (``_v`` below this constant,
+# or missing entirely) are skipped from the snapshot so they get re-
+# fetched freshly. The analyser also filters stale rows out of its
+# reference query, so a fully-stale (spec, encounter, metric) bucket
+# triggers a clean re-seed instead of feeding the AI old detail.
+#
+# History:
+#   v1: initial detail shape (casts / gear / buffs / debuffs / damage_taken).
+#   v2 (v0.3.0): added stats, talent_ids, boss_casts. Phase-transition
+#                shape on the parent report is also a v3 thing but not
+#                duplicated into detail_payload, so v2 stays the right
+#                stamp.
+TOP_LOG_DETAIL_VERSION = 2
+
 
 def _metric_for_role(role: Role) -> str:
     if role == Role.healer:
@@ -103,8 +119,17 @@ async def refresh_top_logs_for_spec_encounter(
             )
         ).scalars().all()
         for r in existing_rows:
-            if r.detail_payload:
-                cached_detail[(r.wcl_report_code, int(r.wcl_fight_id))] = r.detail_payload
+            detail = r.detail_payload
+            if not detail:
+                continue
+            # Only reuse detail blobs written by the current code shape.
+            # Stale ones (``_v`` below the current ``TOP_LOG_DETAIL_VERSION``
+            # — or missing entirely, which counts as version 0) get dropped
+            # from the cache so the fetch loop below re-pulls them, and
+            # the row gets re-inserted with fresh data + current version.
+            if int(detail.get("_v") or 0) < TOP_LOG_DETAIL_VERSION:
+                continue
+            cached_detail[(r.wcl_report_code, int(r.wcl_fight_id))] = detail
 
     try:
         spec_name = _wcl_spec_name(spec.name_en)
@@ -354,6 +379,11 @@ async def _fetch_detail_for_top_log(
         # between the user's pull and the reference kill directly.
         "boss_casts": boss_casts,
         "metric": metric,
+        # Version stamp — the incremental cache snapshot in
+        # ``refresh_top_logs_for_spec_encounter`` only reuses a detail
+        # blob when this matches ``TOP_LOG_DETAIL_VERSION``. Bumping the
+        # constant forces a clean re-fetch.
+        "_v": TOP_LOG_DETAIL_VERSION,
     }