"""Scheduled handlers for the AI Spending dashboard. Two jobs (the legacy ``ai_spending.azure_openai_resources_sync`` cron was retired in PR #253 — its scheduled_jobs row is deleted by migration 153, and the per-resource ``cmdb_ai_resources`` ledger it populated was dropped by migration 151. AI cost classification now flows through the configurable ``ai_spending.service_whitelist`` applied to ``cmdb_cost_facts.service`` — no second sync needed): * ``ai_spending.monthly_summary_generate`` — every 1st of the month at 03:05 Asia/Shanghai. Generates the previous month's bilingual summary via the Bailian LLM and persists it to ``ai_spending_summaries``. No-op if ``ai_spending.llm.api_key`` isn't configured. * ``ai_spending.cny_rate_refresh`` — every Monday at 05:00 UTC. Pulls a fresh CNY/USD rate from frankfurter.dev (ECB-published, free, no auth). On a transient fetch failure it leaves the existing value untouched and soft-skips, but if the stored rate is older than ``_RATE_STALE_AFTER_DAYS`` (14, two missed weekly runs) it raises so the failure_notifier alarms — silent staleness on a value that drives both AI-spending USD totals and cloud cost USD totals would be a real reporting integrity issue. """ from __future__ import annotations import logging from datetime import UTC, date, datetime, timedelta from typing import Any from zoneinfo import ZoneInfo import httpx from sqlalchemy.ext.asyncio import AsyncSession from app.core.exceptions import format_exception_for_log from app.core.redis_client import get_redis from app.domains.ai_spending.locks import ( RollupLockBusy, rollup_write_lock, ) from app.scheduled_jobs.registry import register_scheduled_job logger = logging.getLogger(__name__) # The monthly-summary cron schedule (``5 3 1 * *``) is anchored on # Asia/Shanghai in ``scheduled_jobs.timezone`` (set by migration 079). # "Previous calendar month" MUST be computed in the same timezone — at # 03:05 SH on day 1, UTC is still 19:05 of the LAST day of the prior # month in the UTC calendar, so ``now(UTC).date().replace(day=1) - 1`` # resolves to the month BEFORE the one that just ended (e.g. April when # the cron means to summarise May). This wrong period overwrote prior # months for 7 days in 2026-05 before being caught. Pin the constant # here so the test suite covers the boundary explicitly. _PERIOD_TZ = ZoneInfo("Asia/Shanghai") # ─── Monthly summary auto-generation ──────────────────────────── @register_scheduled_job( key="ai_spending.monthly_summary_generate", display_name="AI 支出 · 月度总结自动生成", description=( "每月 1 号生成上个自然月的双语 AI 支出总结。需要 " "platform_settings.ai_spending.llm.api_key 已配置;否则任务" "记录跳过原因后退出。" ), ) async def run_ai_spending_monthly_summary( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.domains.ai_spending.service import ( SETTING_LLM_API_KEY, AiSpendingService, ) service = AiSpendingService(db) api_key = await service._get_setting_str(SETTING_LLM_API_KEY) if not api_key: return { "generated": False, "skipped_reason": ( "ai_spending.llm.api_key not configured; admin must " "set this in /admin/ai-spending/settings before the " "auto-summary can run." ), } # Previous calendar month, computed in the SAME timezone as the # cron schedule (Asia/Shanghai). Round-22 (#443) had anchored this # on UTC to "decouple from worker TZ", but that broke the meaning # of "previous month" for an SH-anchored cron: when the cron fires # at 03:05 SH on day 1, UTC is still 19:05 of the last day of # what-was-supposed-to-be-the-just-ended month, so the UTC-derived # "previous month" rolls one month TOO FAR back. Concretely on # 2026-06-01 03:05 SH (= 2026-05-31 19:05 UTC) the handler wrote # "2026-04" instead of "2026-05" and overwrote the existing April # summary. Re-pinning the period TZ to match the cron TZ keeps the # two coherent. See ``test_monthly_summary_handler.py`` for the # boundary pin. today = datetime.now(_PERIOD_TZ).date() first_of_month = today.replace(day=1) last_month_end = first_of_month - timedelta(days=1) period = f"{last_month_end.year:04d}-{last_month_end.month:02d}" # Allow opt-out via config so an admin can pause auto-gen without # disabling the row in scheduled_jobs (helpful while the LLM # endpoint is intermittently flaky). if not config.get("enabled", True): return {"generated": False, "skipped_reason": "config.enabled=false"} trace_id = f"ai_spending.auto.{period}" response = await service.auto_generate_summary( period=period, trace_id=trace_id, ) return { "generated": True, "period": period, "model_used": response.model_used, "input_tokens": response.input_token_count, "output_tokens": response.output_token_count, } # ─── CNY → USD rate refresh ───────────────────────────────────── # If a fetch attempt fails AND the stored rate hasn't been # refreshed in this many days, escalate the cron to a hard failure # so the failure_notifier alarms. 14 days = two missed weekly runs; # one missed run still soft-skips so a single transient outage # doesn't page anyone, but a sustained outage doesn't silently # drift either. Tuned vs the cron's weekly cadence. _RATE_STALE_AFTER_DAYS = 14 # FX source. frankfurter.dev publishes ECB reference rates daily, # free, no auth, returns ``rates.USD`` for ``base=CNY`` — same # response shape as the previous (now-paid) ``exchangerate.host``, # so the parser is unchanged. ECB rates are the most operationally # stable free source for CNY/USD; if frankfurter ever stops # responding the cron alarms via the staleness escalation below # and operators set the rate manually on /admin/ai-spending/settings. _FX_SOURCE_URL = "https://api.frankfurter.dev/v1/latest?base=CNY&symbols=USD" _FX_SOURCE_NAME = "frankfurter.dev" @register_scheduled_job( key="ai_spending.cny_rate_refresh", display_name="AI 支出 · 人民币汇率刷新", description=( "每周一从 frankfurter.dev (ECB) 拉一次 CNY→USD 汇率写入 " "platform_settings.ai_spending.cny_to_usd_rate。单次 " "失败软跳过,但当前值若超过 14 天未刷新则硬失败告警 " "(防止静默漂移影响 AI 支出 + 云成本 USD 报表)。" ), ) async def run_ai_spending_cny_rate_refresh( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.domains.ai_spending.service import ( SETTING_CNY_RATE, SETTING_CNY_RATE_REFRESH_AT, ) from app.integrations.models import PlatformSetting # frankfurter.dev emits CNY→USD as ``rates.USD`` when base=CNY. # We deliberately use exactly one source — adding a second free # source as a "fallback" sounds robust but in practice masks the # primary going stale. If frankfurter stops responding, the # staleness escalation below alarms after 14 days and the admin # either swaps the URL via the constants above or sets the rate # manually on /admin/ai-spending/settings. rate: float | None = None error_reason: str | None = None try: # ``trust_env=False`` blocks HTTP_PROXY env-var injection — # a compromised sidecar / debug override could otherwise # tunnel this rate-refresh request and mess with subsequent # CNY dashboards. Defense-in-depth alongside the URL being # hardcoded (so SSRF allowlist isn't needed, but proxy # isolation IS). async with httpx.AsyncClient(timeout=20, trust_env=False) as client: resp = await client.get(_FX_SOURCE_URL) if resp.status_code != 200: error_reason = ( f"{_FX_SOURCE_NAME} HTTP {resp.status_code}: " f"{resp.text[:200]}" ) else: data = resp.json() usd = (data.get("rates") or {}).get("USD") if usd is None: error_reason = f"{_FX_SOURCE_NAME}: rates.USD missing" else: try: rate = float(usd) except (TypeError, ValueError): error_reason = ( f"{_FX_SOURCE_NAME}: rates.USD not a number " f"(got {usd!r})" ) except Exception as exc: # noqa: BLE001 error_reason = ( f"{_FX_SOURCE_NAME}: {format_exception_for_log(exc)}" ) # Plausibility clamp: USD-per-CNY has historically been within # 0.10–0.20 (1 USD = 5–10 CNY). Reject values outside a wide # safety band so a hijacked / misbehaving rate source can't # push all dashboard numbers ~10× off. Operator can manually # set any value via /admin/ai-spending/settings if the band # proves too tight in some future regime. Bounds are INCLUSIVE # (``<=``) — pre-launch hardening 2026-05-16 (LS1 P1) flipped # the previously-exclusive ``<`` because an ECB rate of exactly # 0.05 or 1.0 (highly implausible but theoretically possible # during a regime shift) would otherwise reject the write AND # the corresponding ``resolve_cny_per_usd`` read, taking down # every CN-cloud ingestion simultaneously. if rate is not None and not (0.05 <= rate <= 1.0): error_reason = ( f"{_FX_SOURCE_NAME}: rate {rate} outside the " "0.05 <= rate <= 1.0 plausibility band; " "leaving the existing setting unchanged" ) rate = None if rate is not None and rate <= 0: error_reason = ( f"{_FX_SOURCE_NAME}: returned non-positive rate {rate}" ) rate = None if rate is None: # Fetch failed (or value rejected). Decide soft-skip vs # hard-fail by reading how stale the existing rate row is. # Use the row's ``updated_at`` (SQLAlchemy ``onupdate=now``) # as the truth — the explicit ``cny_rate_last_refreshed_at`` # value-row has historically been unreliable (PR 2026-05-09 # found prod's row stuck at JSON null even after successful # refreshes). updated_at can't lie because it ticks on every # actual UPDATE. existing = await db.get(PlatformSetting, SETTING_CNY_RATE) last_refreshed = existing.updated_at if existing is not None else None if last_refreshed is None: # No existing rate at all (fresh install before # migration 079, or someone deleted the row). Hard-fail # so the operator sees the alarm immediately. raise RuntimeError( f"FX refresh failed and no existing rate is stored " f"({SETTING_CNY_RATE} row is missing). " f"Source error: {error_reason}", ) # MySQL/TiDB store DATETIME as naive UTC (the schema's # ``DateTime(timezone=True)`` is honoured by Postgres but not # by MySQL — the column is naive in the driver round-trip). # Treat naive as UTC explicitly; do not let a future Postgres # migration silently change semantics. if last_refreshed.tzinfo is None: last_refreshed = last_refreshed.replace(tzinfo=UTC) age = datetime.now(UTC) - last_refreshed if age >= timedelta(days=_RATE_STALE_AFTER_DAYS): raise RuntimeError( f"FX refresh failed and stored rate is " f"{age.days} days stale (≥ " f"{_RATE_STALE_AFTER_DAYS} day threshold). " f"AI-spending and cloud-cost USD totals are " f"increasingly inaccurate. Fix the upstream " f"source or set the rate manually on " f"/admin/ai-spending/settings. Source error: " f"{error_reason}", ) return { "refreshed": False, "skipped_reason": error_reason, "stale_days": age.days, "stale_threshold_days": _RATE_STALE_AFTER_DAYS, } # Persist. SQLAlchemy ``onupdate=func.now()`` on the row updates # ``updated_at`` automatically — this is what the staleness # check above reads, so the two paths are coherent. BUT the # update only fires when SQLAlchemy sees the attribute as dirty; # for a JSON column, assigning the same float twice (e.g. ECB # didn't move that week, especially across CN holidays) compares # equal and skips the UPDATE → ``updated_at`` doesn't tick → the # staleness window keeps counting from whenever the value actually # last changed → eventually the next failed fetch escalates to a # spurious hard-fail. ``flag_modified`` forces the dirty marker # so every successful refresh ticks ``updated_at`` even when the # rate value is unchanged. Pinned by a regression test. from sqlalchemy.orm.attributes import flag_modified row = await db.get(PlatformSetting, SETTING_CNY_RATE) if row is None: row = PlatformSetting(key=SETTING_CNY_RATE, value=rate) db.add(row) else: row.value = rate flag_modified(row, "value") refreshed_at = datetime.now(UTC).isoformat() ts_row = await db.get(PlatformSetting, SETTING_CNY_RATE_REFRESH_AT) if ts_row is None: ts_row = PlatformSetting( key=SETTING_CNY_RATE_REFRESH_AT, value=refreshed_at, ) db.add(ts_row) else: ts_row.value = refreshed_at await db.flush() return { "refreshed": True, "cny_to_usd_rate": rate, "refreshed_at": refreshed_at, "source": _FX_SOURCE_NAME, } # ─── Period-rollup materialisation refresh ────────────────────── @register_scheduled_job( key="ai_spending.period_rollup_refresh", display_name="AI 支出 · 周期物化表刷新", description=( "每天 06:30 UTC(在 cmdb.cost_facts_sync / bailian.usage_sync " "之后)重算并物化已关闭月份的 BucketRow 集到 " "ai_spending_period_rollup。最近 recent_months 个闭月每跳必算" "(吸收迟到 facts + 月初新关闭月预热),更早的闭月仅在缺当前" "指纹物化时补算。闭月读直接命中物化表,免去每请求扫 270 万行 " "cmdb_cost_facts。" ), # Multi-minute scan over the 2.7M-row cmdb_cost_facts table per # rebuilt window (up to ``recent_months + 1`` windows every run) — a # long sweep by the 2026-07-19 pool-gate post-mortem's definition, so # it must count against the heavy sub-gate (global 8 / heavy 4) instead # of competing with fast critical crons for global slots. # # Both rollup crons being heavy DOES make either start queueable behind # the other, i.e. more overlap than the 30-minute schedule gap suggests. # That is accepted, not a reason to demote one: the gate exists to keep # long sweeps from starving fast crons, and both jobs genuinely are long # sweeps — misclassifying one to buy schedule luck trades a real # resource invariant for a coincidence. Overlap is handled by # ``ROLLUP_WRITE_LOCK`` (correct regardless of start times), and BOTH # crons escalate a chronic lock-busy streak so that structurally-likelier # overlap cannot starve either direction silently. Both escalation # messages are worded for a gate-delayed world — they point at the run # history, not at cron spacing. weight="heavy", ) async def run_ai_spending_period_rollup_refresh( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: # Time-anchored (06:30 UTC), NOT event-chained off cost_facts_sync: # that cron runs per-integration with no global completion event, so # there's nothing to hook. 06:30 sits after cost_facts_sync (~04:00) # and bailian.usage_sync (~05:00) so the day's facts have landed. # How many closed months to keep materialised (matches the # dashboard's ~13-month lookback). The most-recent ``recent_months`` # are ALWAYS recomputed (late facts can still post to a recently # closed month + the month that just rolled over needs warming); # older months are only (re)computed when their current-fingerprint # materialisation is missing — so steady-state daily work is bounded # to ~recent_months heavy scans, not months_back. # Clamp config to sane bounds: ``months_back`` ≥ 1 (a 0/negative value # would silently materialise nothing and quietly revert the dashboard # to per-request full scans); ``recent_months`` within [0, months_back] # (recent > back would make the "skip if exists" optimisation never # fire → every window heavy-recomputed daily). months_back = max(1, int(config.get("months_back", 13))) recent_months = max(0, min(int(config.get("recent_months", 2)), months_back)) # Serialise against the OTHER writers of these windows — the drift # check's auto-heal (#2083) and the manual refresh (router / MCP). The # runner's single-flight lock is keyed per job, so it does not cover # cross-job overlap; two overlapping DELETE+INSERTs leave duplicate # rows that every read SUMs. Only ACQUIRE-time busy is benign; a lock # LOST mid-hold surfaces as the core ``LockBusy`` and must propagate # so the run is recorded ``failed`` (the rebuild may have interleaved # with a new holder — see ``locks.RollupLockBusy``). # # ``deferred`` gates the streak reset in the ``finally`` — same shape as # the drift check's heal defer. Anything that is NOT an acquire-busy skip # (a completed rebuild, or a crash inside it) breaks the streak. deferred = False try: try: async with rollup_write_lock(): return await _refresh_rollup_windows( db, months_back=months_back, recent_months=recent_months, ) except RollupLockBusy: # Another writer is rebuilding right now. It rebuilds at the # current fingerprint too, so skipping ONCE OR TWICE is not a # gap — and returning (rather than raising) keeps a benign # overlap out of the daily failure alarm. # # Chronic skipping is a different animal (2026-07-29 review, F3): # this cron is the ONLY writer of the current (open) month, which # no other writer covers — the drift check reconciles closed # months only, and the manual refresh is a human click. Skip # every day and the dashboard's default view serves an ever- # staler open month while the run history reads all-green. # Streak-count and escalate on the same threshold as the heal. deferred = True streak = await _bump_defer_streak(_REFRESH_DEFER_STREAK_KEY) if streak >= _HEAL_DEFER_MAX_CONSECUTIVE: # What the counter supports: "the lock was held by SOME other # rollup writer at acquire time on each of the last N runs." # It cannot name the writer, and it cannot distinguish a # slow/queued sibling from a crashed holder — so the message # points at the run history rather than at cron spacing. raise RuntimeError( f"AI spending rollup refresh skipped on {streak} " f"consecutive runs — the rollup write lock was already " f"held at acquire time every time, so this job has " f"re-materialised NO window across those runs and the " f"current open month (which only this job rebuilds; the " f"drift check reconciles closed months only) is stale " f"since the last successful one. Which writer held the " f"lock is NOT knowable from this counter: the candidates " f"are cost_drift_check's heal and an admin/MCP manual " f"refresh. Both rollup crons are weight='heavy' and " f"share the 4-slot heavy sub-gate, so neither " f"necessarily starts on its nominal schedule — read the " f"scheduled-jobs run history for the ACTUAL start times " f"and durations (and the gate queueing) before changing " f"any cron expression.", ) return { "status": "skipped", "reason": "rollup write lock held by another rollup writer", "consecutive_deferrals": streak, "periods_refreshed": 0, "rows_written": 0, } finally: if not deferred: await _reset_defer_streak(_REFRESH_DEFER_STREAK_KEY) async def _refresh_rollup_windows( db: AsyncSession, *, months_back: int, recent_months: int, ) -> dict[str, Any]: """The rebuild loop. Caller holds ``ROLLUP_WRITE_LOCK``. Commits per window, INSIDE the lock (2026-07-29 review). Relying on the runner's post-handler commit released the lock with every rebuilt window still uncommitted, so a second writer could acquire it and start its own DELETE+INSERT on the same window before these rows landed — the duplicate-row/double-count race the lock exists to prevent. Per window rather than one commit at the end also keeps each transaction short (same posture as ``cost_service.ingest_cost_facts``' per-batch commit); each window's DELETE+INSERT is internally atomic, so a failure partway through leaves earlier windows correctly materialised and still raises. """ from app.domains.ai_spending.service import ( AiSpendingService, _current_period, _parse_period, _shift_period, ) svc = AiSpendingService(db) fp = await svc._rollup_fingerprint() # Anchor "current (open) month" on Asia/Shanghai via the SAME helper # the drift check, the manual refresh and the dashboard use # (``service._current_period``), so all four writers agree on which # month is open. # # 2026-07-29 correction: on this cron's REAL schedule the two calendars # never disagree — 06:30 UTC is 14:30 the same day in Asia/Shanghai, and # they only diverge from 16:00 UTC onward. The anchor therefore changes # behaviour only for a run that starts between 16:00–23:59 UTC: a manual # trigger, or a start pushed there by the heavy concurrency sub-gate. On # the LAST UTC day of a month such a run sees the SH calendar already in # the next month and treats the still-unfinished UTC month as closed. # # Materialising it is harmless (the rollup is a cache and that month sits # inside ``recent_months``, so it is recomputed on every later tick, and # the 07:00 drift check heals it). What would NOT be harmless is the # skip-if-exists optimisation freezing that partial window forever when # an operator sets ``recent_months=0``. Hence the explicit UTC guard on # the skip below. current_period = _current_period() utc_period = datetime.now(UTC).strftime("%Y-%m") refreshed: list[str] = [] skipped = 0 total_rows = 0 oldest_start: date | None = None # i=0 is the CURRENT (open) month — always refreshed so the default # dashboard view is served fast from the rollup (perf fix 2026-06); # i=1..recent_months are recently-closed months (late-arriving facts); # older months are only (re)computed when their materialisation is # missing/stale-fingerprint. for i in range(0, months_back + 1): period = _shift_period(current_period, -i) start, end_excl = _parse_period(period) oldest_start = start # last iteration = oldest retained window # ``period > utc_period`` never happens; ``period == utc_period`` # means the UTC month is still receiving facts even if Shanghai has # rolled over — never freeze that window behind skip-if-exists. if ( i > recent_months and period < utc_period and await svc._rollup_window_exists(start, end_excl, fp) ): skipped += 1 continue # No try/except: a DB error must propagate so the run is recorded # ``failed`` and the failure_notifier alarms (silent rollup # staleness would just degrade dashboard latency, but a persistent # write failure is worth surfacing). total_rows += await svc.refresh_period_rollup(start, end_excl) # Commit inside the lock, per window — see the docstring. await db.commit() refreshed.append(period) # Retention prune: windows entirely older than the retained range are # never read (dashboard lookback ≈ months_back) and never re-touched by # this loop, so their rows (incl. stale-fingerprint generations left by # past whitelist/FX edits) would accumulate forever. Bound the table by # deleting anything whose window ends on/before the oldest retained # window's start. pruned = 0 if oldest_start is not None: pruned = await svc.prune_rollup_before(oldest_start) await db.commit() return { "periods_refreshed": len(refreshed), "rows_written": total_rows, "periods_skipped": skipped, "rows_pruned": pruned, "fingerprint": fp[:12], "periods": refreshed, } # ─── Cost rollup↔source drift check ───────────────────────────── # Consecutive-deferral visibility for the drift check's heal path. A # single acquire-busy defer is benign (some other rollup writer was mid- # rebuild once); but if EVERY drift run that finds drift defers, the job # returns success forever — drift on months outside the refresh cron's # ``recent_months`` window would then never be healed OR alarmed. Track # consecutive deferrals in Redis; at the threshold, raise so the # failure_notifier pages. # # What the counter can support (2026-07-29 review): "the rollup write lock # was held by SOME other rollup writer on each of the last N runs that had # findings". It cannot identify WHICH writer, and it cannot conclude # anything about cron spacing — both rollup crons are ``weight="heavy"`` # and share a 4-slot sub-gate with ~14 heavy jobs, so either start can be # queued minutes past its schedule. The escalation message must therefore # point the operator at the run history (actual starts + durations + gate # queueing), not at the 06:30/07:00 schedule. _HEAL_DEFER_STREAK_KEY = "helix:ai_spending:cost_drift_check:heal_defer_streak" _HEAL_DEFER_MAX_CONSECUTIVE = 3 # Safety TTL — an orphaned counter (job disabled / renamed) must not pin # a stale streak forever. Comfortably longer than the daily cadence. _HEAL_DEFER_STREAK_TTL = 14 * 24 * 3600 # The refresh cron's mirror counter (2026-07-29 review, F3). Same mechanism, # separate key: the two crons defer for the same reason but starve different # things, so one streak must not reset or escalate the other. _REFRESH_DEFER_STREAK_KEY = ( "helix:ai_spending:period_rollup_refresh:defer_streak" ) async def _bump_defer_streak(key: str) -> int: redis = get_redis() streak = int(await redis.incr(key)) await redis.expire(key, _HEAL_DEFER_STREAK_TTL) return streak async def _reset_defer_streak(key: str) -> None: """Clear the counter. Best-effort by design — see the callers: this runs on the way out of EVERY non-deferred outcome, including the failure paths, and a Redis hiccup here must not replace the real error.""" try: await get_redis().delete(key) except Exception: # noqa: BLE001 — never mask the outcome being reported logger.warning( "ai_spending: could not reset the defer streak counter %s; a " "stale streak may over-count", key, exc_info=True, ) async def _heal_and_recheck( db: AsyncSession, findings: list[dict[str, Any]], *, months_back: int, ) -> tuple[list[dict[str, Any]], list[str]]: """Re-materialise every drifted window, then re-run the comparison. Caller holds ``ROLLUP_WRITE_LOCK``. Commits immediately so the rebuilt cache survives the caller's subsequent ``raise`` — the heal is worth keeping even when the re-check still disagrees. Returns ``(remaining_findings, healed_periods)``. """ from app.domains.ai_spending.cost_drift import cost_rollup_drift from app.domains.ai_spending.service import AiSpendingService, _parse_period svc = AiSpendingService(db) healed: list[str] = [] # Findings only ever cover closed months by construction. for period in sorted({f["period"] for f in findings}): start, end_excl = _parse_period(period) await svc.refresh_period_rollup(start, end_excl) healed.append(period) await db.commit() recheck = await cost_rollup_drift(db, months_back=months_back) return recheck["findings"], healed @register_scheduled_job( key="ai_spending.cost_drift_check", display_name="AI 支出 · 成本漂移对账", description=( "每天 07:00 UTC(在 ai_spending.period_rollup_refresh 06:30 之后)" "对最近 N 个闭月逐桶比对物化 rollup 与实时重算值。发现超阈值漂移时" "先自动重物化漂移窗口再复检(吸收迟到/重述 facts —— 尤其是落在 " "recent_months 之外、刷新 cron 永远不会重算的更早闭月);复检仍超" "阈值才记为失败并告警(失败通知器去重)。仅闭月对账;不对各云/百炼" "控制台口径做交叉核对(无 API 对等口径,以控制台为准)。" ), # The live-recompute side re-scans the same 2.7M-row cmdb_cost_facts # table per checked month (``months_back``, default 3) and again after a # heal — long sweep, so heavy sub-gate. Same rationale (and same # accepted overlap consequence) as the refresh cron above. weight="heavy", ) async def run_ai_spending_cost_drift_check( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: """Reconcile materialised rollup vs live recompute for closed months. Scheduled at 07:00 UTC — AFTER ``ai_spending.period_rollup_refresh`` (06:30) so the cron has already rebuilt the current + recent closed months at the current fingerprint; reconciling before that would flag the just-refreshed window as drifted on every run. Time-anchored, not event-chained (the rollup refresh emits no global completion event — same rationale as the refresh cron itself). Self-heals before alarming: any drifted window is re-materialised via ``refresh_period_rollup`` (committed immediately so the heal survives a subsequent raise) and the check re-runs. Rationale — the rollup is a faithful CACHE, so "cache ≠ source" has exactly one correct fix: rebuild the cache. This closes a real 2026-07-28 incident shape: an out-of-band cost-facts backfill restated a month older than the refresh cron's ``recent_months`` always-recompute window, so the stale rollup would have alarmed daily *forever* (the refresh cron skips materialised windows at the current fingerprint; the facts themselves are not part of the fingerprint). Only a still-drifted re-check — i.e. the live compute disagreeing with a rollup built from itself seconds earlier, which means facts are churning mid-check or something is genuinely broken — RAISES so the post-handler write-back fires ``maybe_notify_failure`` (the standard in-app + Feishu alarm with signature-based dedup — consecutive days with the same drifting buckets collapse to one alert until an admin acks). A clean run returns the per-period counts (plus any auto-healed windows) so the run history shows what was checked and repaired. """ from app.domains.ai_spending.cost_drift import cost_rollup_drift months_back = max(1, int(config.get("months_back", 3))) # ``deferred`` gates the streak reset in the ``finally`` below. Anything # that is NOT an acquire-busy defer — clean, healed, still-drifted, or a # crash in the drift computation / heal — breaks the streak. Pre-fix the # reset sat on the fall-through path only, so a run that RAISED before # reaching it left the counter intact and a # ``defer, defer, heal-crash, defer`` sequence escalated as "chronically # overlapping refresh" — the wrong diagnosis handed to the operator. deferred = False try: result = await cost_rollup_drift(db, months_back=months_back) findings = result["findings"] healed_periods: list[str] = [] findings_before_heal = len(findings) if findings: # Same lock the refresh cron holds — see locks.py. Without it the # heal and a slow/delayed refresh can rebuild one window # concurrently and leave it double-counted. Catch ONLY the # acquire-time busy (``RollupLockBusy``); a lock LOST mid-heal is # the core ``LockBusy`` and must propagate so the run fails loudly # (handlers-raise-on-failure invariant — the half-done rebuild may # have interleaved with a new holder). try: async with rollup_write_lock(): findings, healed_periods = await _heal_and_recheck( db, findings, months_back=months_back, ) except RollupLockBusy: # Another rollup writer is rebuilding right now, which is # exactly the repair the heal would perform. Report instead # of alarming ONCE OR TWICE: raising here would page IT for a # benign overlap, and the next run (24h) re-checks against the # rebuilt cache. But deferring EVERY time means the findings # are never re-checked against a settled cache — drift # outside the refresh cron's recent_months window would then # never be healed OR alarmed — so the third consecutive # defer escalates to a hard failure. deferred = True streak = await _bump_defer_streak(_HEAL_DEFER_STREAK_KEY) if streak >= _HEAL_DEFER_MAX_CONSECUTIVE: raise RuntimeError( f"AI cost drift heal deferred on {streak} consecutive " f"runs with findings — the rollup write lock was held " f"by another rollup writer every time, so the " f"{findings_before_heal} drift finding(s) are never " f"re-checked against a settled cache. Which writer " f"held it is NOT knowable from this counter: the " f"candidates are period_rollup_refresh, an admin/MCP " f"manual refresh, and a previous heal. Both rollup " f"crons are weight='heavy' and share the 4-slot heavy " f"sub-gate, so neither necessarily starts on its " f"nominal schedule — read the scheduled-jobs run " f"history for the ACTUAL start times and durations " f"(and the gate queueing) before changing any cron " f"expression.", ) return { "status": "heal_deferred", "reason": ( "rollup write lock held by another rollup writer" ), "consecutive_deferrals": streak, "drift_findings": findings_before_heal, "periods_checked": result["periods_checked"], "months_back": months_back, } finally: # Every non-deferred outcome — including an exception — breaks the # streak. Only an acquire-busy defer may accumulate. if not deferred: await _reset_defer_streak(_HEAL_DEFER_STREAK_KEY) if findings: # Deterministic, dedup-friendly summary. The failure_notifier # signature hashes ONLY the first line and strips digits, so the # stable dedup key must live on line 1: the sorted set of # "bucket(cause)" pairs (word-only, digit-free) — "which buckets are # drifting and why". A new bucket joining the set changes the # signature → fresh alert; the same set recurring daily dedups to one. # The volatile detail (which bucket is worst, exact $ and period — # these reorder day to day) goes on LINE 2 so a flapping "worst" # doesn't churn the signature and re-alert on an unchanged set. pairs = sorted({ f"{f['bucket']}({f['likely_cause']})" for f in findings }) worst = max(findings, key=lambda f: f["diff_usd"]) raise RuntimeError( f"AI cost rollup drift over threshold in buckets: {pairs}\n" f"{len(findings)} bucket-month(s) STILL drifted after " f"auto re-materialising {healed_periods}. Worst: " f"{worst['bucket']} {worst['period']} rollup=" f"${worst['rollup_usd']} vs live=${worst['live_usd']} " f"(diff ${worst['diff_usd']}, cause {worst['likely_cause']}). " f"A rollup rebuilt from the live compute seconds earlier should " f"match it — facts are churning mid-check (an in-flight " f"backfill?) or the compute is non-deterministic. Review " f"/admin/reconciliation cost lens." ) return { "drift_findings": 0, "drift_findings_before_heal": findings_before_heal, "auto_healed_periods": healed_periods, "periods_checked": result["periods_checked"], "months_back": months_back, } # ``_get_cny_rate`` helper removed in PR #253 — its only caller was # the retired ``ai_spending.azure_openai_resources_sync`` cron. The # remaining handlers in this file (``run_ai_spending_monthly_summary``, # ``run_ai_spending_cny_rate_refresh``) get the rate via # ``app.core.fx_rate`` or ``AiSpendingService._get_cny_rate`` directly.