"""Daily deterministic Intelligence snapshot production (zero LLM calls).""" from __future__ import annotations import logging import time import uuid from datetime import UTC, datetime from typing import Any from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from app.core.exceptions import AppException from app.domains.cmdb.models import SCOPE_STATUS_ACTIVE, CmdbCloudScope from app.intelligence.cloud_scope_facts import ScopeAnomalyIndex from app.intelligence.cloud_scope_snapshot import build_cloud_scope_snapshot from app.intelligence.critical_health_notifications import notify_critical_health_snapshot from app.intelligence.generation import generate_analysis from app.intelligence.models import IntelligenceAnalysis from app.intelligence.provider_registry import ( BUSINESS_AUDIENCE_KEY, platform_providers, ) from app.intelligence.providers import AssetFactProvider from app.intelligence.recommendations import materialize_analysis_recommendations from app.intelligence.regional_audiences import regional_asset_audiences from app.intelligence.settings import intelligence_enabled from app.intelligence.snapshot_service import SnapshotSubject, build_snapshot, latest_snapshot from app.scheduled_jobs.models import ScheduledJob, ScheduledJobRun from app.scheduled_jobs.registry import register_scheduled_job from app.scheduled_jobs.slo import SLO_DAILY_DIGEST logger = logging.getLogger(__name__) _CLOUD_SCOPE_BASELINE_BOOTSTRAP_KEY = "intelligence.cloud_scope_baseline_bootstrap" # Wall-clock a single bootstrap tick may spend sweeping scopes. Sized # well under the worker watchdog's 300s startup grace so a slow sweep # can never again be mistaken for a wedged scheduler, and under the # scheduler's own 1800s default handler timeout. At the ~12s/scope # measured against prod on 2026-08-18 this is ~10 scopes per tick; on a # 60s interval the 271-scope fleet drains in roughly half an hour. _BOOTSTRAP_BUDGET_SECS = 120.0 _BOOTSTRAP_BUDGET_CEILING_SECS = 240.0 # The recurring fleet refresh is a heavy nightly job, so it gets a much # larger slice than a bootstrap tick — but still a ceiling, kept under # the scheduler's 1800s default handler timeout so the run always ends # by returning rather than by being killed. _DAILY_BUDGET_SECS = 1500.0 _DAILY_BUDGET_CEILING_SECS = 1680.0 @register_scheduled_job( key="intelligence.snapshot_daily", display_name="AI 大脑每日确定性快照", description=( "每日汇总人员、入转离、资产、软件/API、云网络、成本、安全和平台运行" "八维确定性事实；同事实 Hash 不写新版本。只计算评分和匿名证据，不调用模型。" ), slo=SLO_DAILY_DIGEST, weight="heavy", ) async def run_intelligence_snapshot_daily( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: """Build one immutable shared-business snapshot at a single time anchor.""" run_at = datetime.now(UTC) result = await build_snapshot( db, subject=SnapshotSubject("platform", "platform", BUSINESS_AUDIENCE_KEY), providers=platform_providers(now=lambda: run_at), ) regional_results = [] for audience in await regional_asset_audiences(db): regional_results.append( await build_snapshot( db, subject=SnapshotSubject("domain", "assets", audience.key), providers=( AssetFactProvider( managed_location_ids=audience.location_ids, now=lambda: run_at, ), ), ) ) critical_notifications = ( await notify_critical_health_snapshot(db, result.snapshot) if result.created else 0 ) return { "snapshot_id": str(result.snapshot.id), "created": result.created, "facts_hash": result.snapshot.facts_hash, "score_version": result.snapshot.score_version, "overall_score": str(result.snapshot.overall_score), "confidence_score": str(result.snapshot.confidence_score), "provider_failures": list(result.provider_failures), "regional_audience_count": len(regional_results), "regional_snapshots_created": sum(row.created for row in regional_results), "regional_provider_failure_count": sum( bool(row.provider_failures) for row in regional_results ), "critical_notifications_queued": critical_notifications, "model_calls": 0, "run_at": run_at.isoformat(), } @register_scheduled_job( key="intelligence.cloud_scope_baseline_daily", display_name="AI 大脑云订阅每日确定性基线", description=( "为有效云订阅预计算共享健康分、数据可信度和逐项数据缺口。" "页面读取不调用模型；用户明确点击后才生成 AI 解读。\n" "每次运行有 wall-clock 上限（config.budget_secs，默认 1500s），到点在" "订阅边界停下并把游标写回本行，下次运行接着扫 —— 整个机队按轮转覆盖，" "而不是每天从头重扫。2026-08-18 实测每订阅约 12s、271 个订阅约 60 分钟，" "超过 handler 默认 1800s 超时；此前它没有上限,从 2026-08-17 播种起" "一次都没跑完过（每次都被 6 小时 stale-sweeper 判死）。" ), slo=SLO_DAILY_DIGEST, weight="heavy", ) async def run_cloud_scope_baseline_daily( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: """Refresh active subscriptions' deterministic, shared baseline. The subscription analysis GET route deliberately never builds a snapshot: it must remain read-only and spend no LLM tokens. This job gives every eligible subscription a fresh baseline before anyone opens its AI tab, so ``数据不足`` can name the exact missing source instead of becoming a generic empty state. A failed subscription is isolated from its peers. **Bounded + resumable (2026-08-18).** This used to walk the whole fleet in one call with no ceiling. Measured against prod that is ~60 minutes even after ``ScopeAnomalyIndex``, i.e. past the scheduler's 1800s default handler timeout — and before the index it was ~3.9 hours, which is why this job had never once completed since being seeded: every run was eventually declared dead by the 6h stale-run sweeper. It now spends at most ``budget_secs`` per run and carries a cursor on its own job row, so the fleet is covered by rotation. The trade is deliberate and stated on the row: bounded, always-terminating coverage over roughly two runs beats an unbounded sweep that never lands a single complete pass. """ try: budget = float(config.get("budget_secs", _DAILY_BUDGET_SECS)) except (TypeError, ValueError): budget = _DAILY_BUDGET_SECS budget = max(60.0, min(budget, _DAILY_BUDGET_CEILING_SECS)) job = ( await db.execute( select(ScheduledJob).where( ScheduledJob.handler_key == "intelligence.cloud_scope_baseline_daily", ) ) ).scalar_one_or_none() cursor = str((job.config or {}).get("resume_after_scope_id") or "").strip() if job else "" summary = await _refresh_cloud_scope_baselines( db, budget_secs=budget, after_scope_id=cursor or None, ) if job is not None: job_config = dict(job.config or {}) if summary["complete"]: # Wrapped the fleet — next run starts a fresh pass from the top. job_config.pop("resume_after_scope_id", None) else: job_config["resume_after_scope_id"] = summary["next_scope_id"] job.config = job_config await db.commit() return summary async def _refresh_cloud_scope_baselines( db: AsyncSession, *, budget_secs: float | None = None, after_scope_id: str | None = None, ) -> dict[str, Any]: """Build current scope baselines without a model call or sampling. ``budget_secs`` bounds the wall-clock a single call may spend. When the budget runs out the sweep stops on a scope boundary and reports ``next_scope_id`` so the caller can resume exactly where it stopped; ``complete`` says whether the fleet was finished. ``None`` (the default) keeps the historical "walk the whole fleet in one call" behaviour for callers that want it. Why resumability, not just speed: per-scope cost is dominated by work this module does not own (``resolve_scope_servers`` and the per-scope cost-fact reads measured at ~12s/scope against prod on 2026-08-18, even after ``ScopeAnomalyIndex`` removed the redundant per-integration anomaly scan). A fleet-wide sweep is therefore inherently long and grows with the fleet. Making the sweep resumable puts a ceiling on any single run regardless of how slow an individual scope becomes — which is the property that was missing when this code path outlived the worker watchdog's grace and crash-looped prod for ~16h. The cursor is the scope id, and the scan is ordered by scope id, so resumption is stable: a scope added mid-sweep with a lower id is picked up on the next full pass rather than shifting the window. """ started = time.monotonic() scope_ids = list( ( await db.execute( select(CmdbCloudScope.id) .where( CmdbCloudScope.deleted_at.is_(None), CmdbCloudScope.status == SCOPE_STATUS_ACTIVE, *( [CmdbCloudScope.id > after_scope_id] if after_scope_id else [] ), ) .order_by(CmdbCloudScope.id) ) ).scalars() ) created = 0 reused = 0 processed = 0 next_scope_id: str | None = None failed_scope_ids: list[str] = [] run_at = datetime.now(UTC) # One shared anomaly memo for the whole fleet pass. Without it every # scope re-ran its integration's 30-day anomaly scan — measured at # ~51s/scope against prod, i.e. ~3.9h for 271 scopes, which is why # this job had never once completed. See ``ScopeAnomalyIndex``. # Scoped to this call so it can never outlive ``run_at``. anomaly_index = ScopeAnomalyIndex() for position, scope_id in enumerate(scope_ids): # Budget is checked BEFORE each scope, never mid-scope: a snapshot # is written by ``persist_compiled_snapshot`` as one unit, so the # only safe stopping point is a scope boundary. The first scope # always runs — a budget smaller than one scope must still make # forward progress, or a resumable sweep would spin forever # re-reading the same cursor. if ( budget_secs is not None and position > 0 and time.monotonic() - started >= budget_secs ): next_scope_id = str(scope_ids[position - 1]) logger.info( "intelligence cloud scope baseline: budget %.0fs spent after " "%d/%d scope(s); resuming after scope_id=%s", budget_secs, processed, len(scope_ids), next_scope_id, ) break try: result = await build_cloud_scope_snapshot( db, str(scope_id), now=run_at, anomaly_index=anomaly_index, ) except Exception: # A provider/data anomaly for one subscription must not leave all # other employees with an unexplained empty analysis page. await db.rollback() failed_scope_ids.append(str(scope_id)) logger.exception( "intelligence cloud scope baseline failed scope_id=%s", scope_id, ) processed += 1 continue created += int(result.created) reused += int(not result.created) processed += 1 return { # ``scope_count`` stays the number of scopes this call was # responsible for (the remaining tail when resuming), so # ``processed == scope_count`` is exactly "this call finished its # share" and the bootstrap's completeness check below reads # naturally off ``complete``. "scope_count": len(scope_ids), "scopes_processed": processed, "complete": next_scope_id is None, "next_scope_id": next_scope_id, "resumed_after_scope_id": after_scope_id, "snapshots_created": created, "snapshots_reused": reused, "failed_scope_count": len(failed_scope_ids), # IDs remain in the scheduler audit record only. They are never # surfaced through employee-facing AI responses. "failed_scope_ids": failed_scope_ids, "model_calls": 0, "elapsed_secs": round(time.monotonic() - started, 1), "run_at": run_at.isoformat(), } @register_scheduled_job( key=_CLOUD_SCOPE_BASELINE_BOOTSTRAP_KEY, display_name="AI 大脑云订阅基线首次回填", description=( "一次性回填所有有效云订阅的确定性健康基线；成功后自动停用。" "仅读取既有事实和写快照，不调用模型或消耗 Token。" ), slo=SLO_DAILY_DIGEST, # This is a one-shot release warm-up. It only reads existing domain facts # and writes immutable snapshots, so it must not sit behind long-running # cloud vendor sweeps while holding its distributed single-flight lock. # The recurring full fleet refresh remains ``heavy``. weight="light", ) async def run_cloud_scope_baseline_bootstrap( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: """Warm a release's pre-existing subscriptions without waiting for tomorrow. A daily cron added after its fire time would otherwise leave every pre-existing subscription on the generic first-read state for almost a day. The interval row is deliberately self-disabling: it retries a bounded number of times on an exceptional scope, then the recurring daily job takes over and the scheduler remains transparent to operators. The sweep is **budgeted and resumable**. Before 2026-08-18 this walked all 271 scopes in one call — measured at ~3.9 hours, so it never once reached its own completion branch, and because each tick outlived the worker watchdog's 300s grace it crash-looped prod for ~16h. Now each tick spends at most ``budget_secs`` and stores the cursor on the job row, so the fleet drains over several ticks and the job disables itself only when the sweep genuinely reaches the end. """ cursor = str(config.get("resume_after_scope_id") or "").strip() or None try: budget = float(config.get("budget_secs", _BOOTSTRAP_BUDGET_SECS)) except (TypeError, ValueError): budget = _BOOTSTRAP_BUDGET_SECS # Keep every tick comfortably inside the watchdog's startup grace so a # slow sweep can never again read as a wedged scheduler. budget = max(30.0, min(budget, _BOOTSTRAP_BUDGET_CEILING_SECS)) summary = await _refresh_cloud_scope_baselines( db, budget_secs=budget, after_scope_id=cursor, ) job = ( await db.execute( select(ScheduledJob).where( ScheduledJob.handler_key == _CLOUD_SCOPE_BASELINE_BOOTSTRAP_KEY, ) ) ).scalar_one_or_none() if job is None: raise RuntimeError("Cloud-scope baseline bootstrap job row is missing") failed = int(summary["failed_scope_count"]) attempt = 0 max_attempts = 6 if failed: try: max_attempts = max(1, min(int(config.get("max_attempts", 6)), 12)) except (TypeError, ValueError): max_attempts = 6 # Count DISPATCHES, not completions. ``job.run_count`` is bumped # only in the scheduler's terminal write-back, so a handler that # is killed mid-run (worker restart, deploy, OOM) never # increments it — and this bounded-retry give-up, the only thing # that ever disables this one-shot row, becomes unreachable. On # 2026-08-17 that turned a slow backfill (271 scopes vs the # watchdog's 300s grace) into a 16h worker crash-loop: 874 # dispatches, ``run_count`` still 0, attempt permanently 1. # Non-skipped run rows count every real attempt, including the # abandoned ones, so the escape hatch stays reachable. attempt = int( ( await db.execute( select(func.count()) .select_from(ScheduledJobRun) .where( ScheduledJobRun.job_id == job.id, ScheduledJobRun.status != "skipped", ) ) ).scalar_one() or 0 ) gave_up = bool(failed) and attempt >= max_attempts swept_clean = not failed and summary["complete"] # Persist sweep progress BEFORE any failure verdict, and in the SAME # commit that may disable the row. A failing scope must not throw away # the ground this tick already covered: the failure paths below raise, # so a cursor written after them would never land, the next tick would # restart from the top, and the sweep would re-walk the same prefix # forever — the exact "never completes" shape this job was stuck in, # just wearing a tidier summary. job_config = dict(job.config or {}) if summary["complete"]: # Clear the cursor so a future re-enable starts a clean pass rather # than resuming from a stale mid-fleet position. job_config.pop("resume_after_scope_id", None) else: job_config["resume_after_scope_id"] = summary["next_scope_id"] job.config = job_config if gave_up or swept_clean: job.enabled = False await db.commit() if failed: if gave_up: raise RuntimeError( "Cloud-scope baseline bootstrap gave up after " f"{attempt} attempt(s): {failed}/{summary['scope_count']} scope(s) failed; " "daily baseline remains enabled for the next retry." ) raise RuntimeError( "Cloud-scope baseline bootstrap incomplete: " f"{failed}/{summary['scope_count']} scope(s) failed; retrying automatically." ) if not summary["complete"]: # Budget ran out with scopes still ahead — stay enabled so the next # tick resumes from the cursor just committed. return { **summary, "bootstrap_complete": False, "bootstrap_disabled": False, } return {**summary, "bootstrap_complete": True, "bootstrap_disabled": True} async def _analysis_failure_reason( db: AsyncSession, *, snapshot_id: str, analysis_kind: str, ) -> str | None: """The sentence that says WHY the generation failed, for the alert. ``generate_analysis`` records the real cause on the ``intelligence_analyses`` row (which rule rejected the output, which fact reference was unknown, the vendor's own sentence) and then raises an AppException whose message is a deliberately generic one-liner. The operator alert was built from the CODE alone, so every P1 card for this job read identically — ``INTELLIGENCE_OUTPUT_INVALID (snapshot …)`` — and finding out what actually broke meant a direct prod-TiDB query. On 2026-08-25 that query returned "Unknown fact reference: /facts/cloud_network.resource_waste_rate.value", which is the whole diagnosis; the card should have carried it. Best-effort by construction: a failure that never got as far as a pending row (no API key, lease contention, a poisoned session) has no reason to read, and the alert still names the code. Nothing here may replace the original exception, so every fault is swallowed — this runs on a path that is already failing. """ # Rolls back first, then runs a single SELECT and no writes at all, so # there is no flush whose failure could deactivate the caller's # transaction — the condition ``best_effort`` exists to contain. # Returning None on any fault is the documented contract above: this # runs on an already-failing path and must never replace the original # exception. The marker has to sit within two lines of the ``try:``. # savepoint-exempt: read-only (rollback + one SELECT, performs no writes) try: # The generation failure path ends in ``_mark_failed``, which # rolls back and commits its own row; other paths may leave a # poisoned transaction behind, and a SELECT on one raises. await db.rollback() row = (await db.execute( select(IntelligenceAnalysis) .where( IntelligenceAnalysis.snapshot_id == uuid.UUID(snapshot_id), IntelligenceAnalysis.analysis_kind == analysis_kind, IntelligenceAnalysis.status == "failed", ) .order_by(IntelligenceAnalysis.created_at.desc()) .limit(1) )).scalar_one_or_none() except Exception: # noqa: BLE001 — never fail the failure path logger.warning( "could not read the intelligence failure reason for snapshot %s", snapshot_id, exc_info=True, ) return None message = (row.failure_message or "").strip() if row is not None else "" return message.split("\n", 1)[0] or None @register_scheduled_job( key="intelligence.analysis_daily", display_name="AI 大脑每周跨域简报", description=( "仅在 Intelligence Center 发布闸门打开时，每周为事实变化后的平台快照生成一次" "中英双语简报；事实未变化复用已有结果，失败不影响确定性快照。" ), slo=SLO_DAILY_DIGEST, # The gate has two classes, not three: an earlier ``weight="normal"`` was # not a ``SchedJobWeight`` and silently fell through to ``light``. This # handler holds its pool connection across a 120s-timeout LLM call # (``BailianLLMClient`` in ``generation.py``), which is the documented # ``heavy`` criterion. weight="heavy", ) async def run_intelligence_analysis_daily( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: if not await intelligence_enabled(db): return {"status": "skipped", "reason": "release_gate_closed", "model_calls": 0} snapshot = await latest_snapshot( db, subject=SnapshotSubject("platform", "platform", BUSINESS_AUDIENCE_KEY), ) if snapshot is None: return {"status": "skipped", "reason": "no_snapshot", "model_calls": 0} # Read both identifiers while the instance is guaranteed loaded. On the # failure path ``generate_analysis`` calls ``_mark_failed``, whose # ``db.rollback()`` expires every instance in the session — a rollback # expires regardless of ``expire_on_commit=False``. Touching ``snapshot`` # after that point is a sync attribute access that emits a refresh SELECT # and raises ``MissingGreenlet``, which is exactly how the 2026-08-17 run # alerted with an ORM fault while ``intelligence_analyses`` held the real # cause (``INTELLIGENCE_OUTPUT_INVALID`` from qwen3-max). snapshot_id = str(snapshot.id) facts_hash = snapshot.facts_hash try: result = await generate_analysis( db, snapshot=snapshot, analysis_kind="platform_daily_brief", actor_id=None, enforce_user_quota=False, ) except Exception as exc: # This job's entire output is the narrative, so a generation failure # IS a job failure: raise so the run row and the operator alert name # the model-side cause instead of recording a silent ``ok``. The # audited ``intelligence_analyses`` failure row is already committed # by ``_mark_failed`` and survives this raise. # # Only an ``AppException`` has a ``code`` that MEANS anything. The # previous ``getattr(exc, "code", None)`` looked duck-typed and safe, # but every SQLAlchemy DBAPI error also carries ``.code`` — holding a # documentation-anchor slug, not an error code. This handler catches # ``Exception`` around a call that touches the database across a 120s # model call, so those are exactly the ones that reach it: two real # faults injected on 2026-09-06 (a dropped pooled connection, an # over-wide vendor model id) paged with ``…失败:7s2a`` and # ``…失败:9h9h``. The class name is not a Helix error code either, # but ``PendingRollbackError`` tells an operator what to look at and # ``7s2a`` tells them to search the SQLAlchemy docs. code = exc.code if isinstance(exc, AppException) else type(exc).__name__ reason = await _analysis_failure_reason( db, snapshot_id=snapshot_id, analysis_kind="platform_daily_brief", ) if not reason and not isinstance(exc, AppException): # No audited row to quote — which is itself the signature of a # DB-side fault, the one case where the exception's own sentence # is the only description of what broke. Scrubbed and clipped # like the vendor detail in ``generate_analysis``: these strings # occasionally echo credential-shaped fragments, and this one # lands in an operator alert. try: from app.integrations.service import _scrub_message # noqa: PLC0415 reason = _scrub_message(str(exc) or "").strip().split("\n", 1)[0][:300] except Exception: # noqa: BLE001 — never fail the failure path reason = None raise RuntimeError( f"AI 大脑每日跨域简报生成失败:{code}" + (f" —— {reason}" if reason else "") + f" (snapshot {snapshot_id})" ) from exc materialized = await materialize_analysis_recommendations(db, result.analysis) return { "status": "cached" if result.cache_reused else "generated", "analysis_id": str(result.analysis.id), "snapshot_id": snapshot_id, "facts_hash": facts_hash, "recommendations_created": materialized.created, # ``refreshed`` is how many standing suggestions had their prose # re-stated from this run's facts; ``superseded`` is how many the # analyses stopped emitting long enough to archive. Both used to be # invisible, which is how a card quoting a metric fixed weeks earlier # survived every weekly run without anyone being able to see it. "recommendations_refreshed": materialized.refreshed, "recommendations_superseded": materialized.superseded, "model_calls": 0 if result.cache_reused else 1, } __all__ = [ "run_intelligence_snapshot_daily", "run_cloud_scope_baseline_daily", "run_cloud_scope_baseline_bootstrap", "run_intelligence_analysis_daily", ]