"""Built-in scheduled job handlers. Importing this module registers every built-in recurring task under the scheduler's handler registry. Keep new handlers thin: they orchestrate a domain service and return a compact dict summary. Long-running work, DB transactions, error handling — all happen in the domain service, not here. """ from __future__ import annotations import asyncio import re import uuid from collections.abc import Awaitable, Callable from datetime import UTC, date, datetime, timedelta from typing import Any from sqlalchemy import and_, func, or_, select, text from sqlalchemy.exc import OperationalError from sqlalchemy.ext.asyncio import AsyncSession from app.core.exceptions import ( ConflictError, format_exception_for_log, is_connection_lost_db_error, is_transaction_aborting_db_error, ) from app.core.safe_logging import get_safe_logger from app.core.txn import savepoint # Value-only import (a frozen dataclass, no service wiring) so the # module-level card-chrome constants below can be built at import time. # The rest of this module keeps importing ``app.notifications.service`` # lazily inside handlers — that convention exists for the SERVICE, which # pulls the integration/driver chain. from app.notifications.service import LocalizedText from app.scheduled_jobs._seat_sync_helpers import raise_if_total_failure from app.scheduled_jobs.registry import register_scheduled_job from app.scheduled_jobs.slo import ( SLO_CMDB_COST_FACTS_SYNC, SLO_DAILY_DIGEST, SLO_LEAVER_ESCALATION_TICK, SLO_NOTIFICATION_OUTBOX_RETRY, SLO_OVERDUE_LEAVERS_CHECK, SLO_PROCESS_ADVANCE, ) logger = get_safe_logger(__name__) # MySQL lock-wait errno when InnoDB times out waiting on a row lock # (default 50s; we bump per-session to 120s for the HRIS confirm tx). _MYSQL_LOCK_WAIT_TIMEOUT = 1205 # Deadlock detected — InnoDB rolled back our transaction. Retryable # for the same reason as lock-wait: a fresh session + fresh cursor # read usually resolves it. _MYSQL_DEADLOCK = 1213 def _is_retryable_lock_error(exc: OperationalError) -> bool: """True if a fresh-session re-run of an idempotent unit of work can recover from ``exc``. Two transient families qualify: * lock-wait (1205) / deadlock (1213) / orphaned-savepoint (1305) — the transaction was rolled back server-side (delegates to :func:`app.core.exceptions.is_transaction_aborting_db_error`, the same errno set the domain-layer ``begin_nested()`` catch boundaries use, so there's a single source of truth); * connection-drop (2006 server-gone-away / 2013 lost-connection) — the socket itself died (e.g. a TiDB Cloud node recycled / drained under an in-flight query and RST'd it). ``pool_pre_ping`` only probes at checkout so it can't catch a mid-query drop; a fresh connection re-running the work is the canonical fix. Everything else (bad SQL, FK violation) is a real bug retry can't fix.""" return is_transaction_aborting_db_error(exc) or is_connection_lost_db_error( exc ) def _resolve_recurring_billing_window( config: dict[str, Any], ) -> tuple[date | None, date | None, str | None]: """Parse an optional ``start_date``/``end_date`` catch-up window from a recurring cloud-billing job's config, self-healing a *stale* pin. ``cmdb.cost_facts_sync`` / ``cmdb.billing_daily_sync`` accept a ``{start_date, end_date}`` window in their config to widen a one-off backfill. But these are *recurring daily* jobs: a window whose ``end_date`` has fallen into the past is a forgotten backfill pin that silently FREEZES ongoing ingestion — the nightly run keeps re-ingesting the same stale window and never advances toward "yesterday", so the truth table stops growing while the job still reports ``status=ok``. Prod incident 2026-06-16: ``cmdb.cost_facts.azure-primary`` was pinned to ``2026-04-01..2026-05-10``, so ``cmdb_cost_facts`` held zero rows after mid-May while ``cmdb_cloud_billing_daily`` (unpinned) stayed current — the FinOps overview showed no Azure spend in any tab even though the per-scope billing card did. Rule: a window is honoured only while its ``end_date`` is still recent (>= yesterday in UTC). An expired pin is dropped (the sync falls back to its rolling default) and the dropped window is returned as the third element so the caller can WARN-log it and surface it on the run row. Returns ``(start, end, None)`` when a window is honoured, ``(None, None, note)`` when a stale window is dropped, and ``(None, None, None)`` when no window was configured. Raises ``ValueError`` on a malformed ISO date so the caller can map it to a ``skipped`` outcome. """ start_raw = str(config.get("start_date") or "").strip() end_raw = str(config.get("end_date") or "").strip() start = date.fromisoformat(start_raw) if start_raw else None end = date.fromisoformat(end_raw) if end_raw else None if start is None and end is None: return None, None, None # Explicit past-window override: the ``cmdb.cost_facts_monthclose`` # dispatcher (and any deliberate historical backfill) fans a # *previous-month* window out through this handler via # ``execute_job(config_override=...)``. That window's ``end_date`` is # by definition older than yesterday, so the stale-pin self-heal # below would otherwise silently discard it and re-run the rolling # 3-day default — turning the whole month-close into a no-op. This # flag is only ever set on an EPHEMERAL config_override (never # persisted to a recurring job's ``config`` row), so a forgotten # backfill pin on the daily cron still self-heals as before — # nothing on disk carries this bypass. Requires both dates present; # a half-set window still falls through to validation. if ( config.get("allow_past_window") and start is not None and end is not None and start <= end ): return start, end, None yesterday = datetime.now(UTC).date() - timedelta(days=1) # A recurring daily job must always be allowed to reach yesterday. If # the pin's end is older than that, it can never catch up on its own — # treat it as expired and revert to the rolling default. if end is not None and end < yesterday: return None, None, ( f"ignored stale pinned window {start_raw or '?'}..{end_raw} " f"(end_date older than {yesterday.isoformat()}); fell back to " "rolling default to avoid frozen ingestion" ) return start, end, None # ``WorkResultT`` is whatever the retried unit of work returns. The # lock-retry helpers are transport, not policy: they hand the caller's own # result straight back, so a handler that returns ``dict[str, Any]`` keeps # that type through the wrapper instead of decaying to ``Any`` at every # call site. async def _run_in_fresh_session_with_lock_retry[WorkResultT]( work: Callable[[AsyncSession], Awaitable[WorkResultT]], *, label: str, retry_delays_sec: tuple[int, ...] = (0, 5, 15), ) -> WorkResultT: """Run ``await work(session)`` in a fresh, committed session, retrying on transient MySQL/TiDB errors (lock-wait / deadlock / connection-drop — see :func:`_is_retryable_lock_error`) with a brand-new session each attempt and returning ``work``'s result. InnoDB tells the deadlock victim to "restart transaction"; a fresh session + fresh cursor read is the canonical response. The whole transaction rolled back on a 1213, so re-running is clean — the caller's ``work`` must be idempotent (re-scan from current DB state), which both the leaver-escalation tick and the overdue-leaver transition are (they re-filter on live ``leaver_pending`` rows). The same idempotence is what makes re-running safe after a 2006/2013 connection drop killed the previous attempt's connection. Non-retryable ``OperationalError`` (bad SQL, FK violation) and all other exceptions propagate immediately (fail-fast). On exhausting all retries the last error is re-raised so the scheduler records a real failure. Mirrors the per-instance retry loop in ``run_hris_incremental_sync``. """ from app.core.database import async_session_factory last_exc: OperationalError | None = None for attempt_idx, delay in enumerate(retry_delays_sec, start=1): if delay > 0: logger.info( "%s retry %d/%d after %ds (prev: %s)", label, attempt_idx, len(retry_delays_sec), delay, type(last_exc).__name__ if last_exc else "n/a", ) await asyncio.sleep(delay) try: async with async_session_factory() as session: # Give the batch room to finish under transient # contention before InnoDB times out (default 50s is # tight when a concurrent cron / admin save holds rows). await session.execute( text("SET SESSION innodb_lock_wait_timeout = 120") ) result = await work(session) await session.commit() return result except OperationalError as exc: if _is_retryable_lock_error(exc): last_exc = exc errno = getattr(getattr(exc, "orig", None), "args", [None])[0] logger.warning( "%s hit MySQL errno %s, attempt %d/%d", label, errno, attempt_idx, len(retry_delays_sec), ) continue raise # Exhausted retries on a lock-wait / deadlock — surface as failure. assert last_exc is not None # loop ran ≥1 attempt logger.error("%s EXHAUSTED lock-retries (last: %s)", label, last_exc) raise last_exc async def run_with_lock_retry[WorkResultT]( db: AsyncSession, work: Callable[[AsyncSession], Awaitable[WorkResultT]], *, label: str, retry_delays_sec: tuple[int, ...] = (5, 15), ) -> WorkResultT: """Run ``await work(db)`` on the caller's session; on a MySQL lock-wait (1205) / deadlock (1213) roll the poisoned session back and re-run ``work`` on fresh sessions via :func:`_run_in_fresh_session_with_lock_retry`. The caller-session first attempt keeps the scheduler's contract intact (every handler receives a working session the framework commits afterwards) and the zero-contention happy path identical to a direct call — only the deadlock-victim path pays for new sessions. ``work`` must be idempotent (re-scan from current DB state) and own its commits: after a 1213 InnoDB has already rolled the whole transaction back, so a re-run starts from clean state. Born from the 2026-06-12 incident: a worker restart double-ran ``jira.seat_sync``, the loser of the INSERT deadlock (errno 1213) had its transaction rolled back server-side, the session turned into PendingRollbackError soup, and the whole run failed — daily snapshot unwritten, one auto-import lost. The canonical InnoDB response is "restart transaction", which for an idempotent reconcile means a fresh session re-running the whole thing. """ try: return await work(db) except OperationalError as exc: if not _is_retryable_lock_error(exc): raise errno = getattr(getattr(exc, "orig", None), "args", [None])[0] logger.warning( "%s hit MySQL errno %s on the scheduler-provided session; " "retrying on fresh sessions", label, errno, ) # Leave the caller's session clean — the framework commits it # after the handler returns, and a poisoned session would raise # PendingRollbackError there and mask the real error. await db.rollback() return await _run_in_fresh_session_with_lock_retry( work, label=label, retry_delays_sec=retry_delays_sec, ) # ── HRIS incremental sync ───────────────────────────────────────── @register_scheduled_job( key="hris.incremental_sync", display_name="HRIS 增量同步", description=( "遍历所有已启用的 HRIS 集成(中国 + 欧洲 SF 等),基于上次成功" "同步的游标拉取 lastModifiedDateTime 之后的所有人员变更," "自动确认批次并落入 identity_persons。全量同步仅在系统初始化" "由管理员手动触发,不在此定时任务内。" ), ) async def run_hris_incremental_sync( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.core.database import async_session_factory from app.domains.hr.service import SuccessFactorsLifecycleService from app.integrations.compat import list_hris_instances instances = await list_hris_instances(session=db) if not instances: return { "instances_synced": 0, "total_new": 0, "total_updated": 0, "total_deactivated": 0, "errors": [], "message": "未配置任何 HRIS 集成", } summary = { "instances_synced": 0, "total_new": 0, "total_updated": 0, "total_reactivated": 0, "total_deactivated": 0, "errors": [], # #1745: instances whose auto-confirm was halted by the mass-leaver # brake — the batch is staged + left in ``pending_review`` awaiting # manual IT review, NOT auto-applied. Folded into the terminal raise # below so the run lands ``status='failed'`` and the scheduled-job # failure-notifier pushes an actionable IT-admin card. "blocked": [], } lookback_hours = int(config.get("lookback_hours", 25)) # Retry-with-backoff: each instance gets up to 3 attempts. MySQL row # locks on ``identity_persons`` / ``hr_sync_batches`` can timeout at # the 50s default ``innodb_lock_wait_timeout`` if a concurrent # transaction (admin UI save, deploy rollover, a prior failed run # holding an undo segment) is holding the row. Retry gives the # contending transaction time to commit / rollback. Each attempt # uses a FRESH session so we don't inherit a poisoned state. retry_delays_sec = (0, 10, 30) # first attempt immediate, then 10s, then 30s for row, instance_config in instances: alias = row.slug or str(row.id) last_exc: Exception | None = None for attempt_idx, delay in enumerate(retry_delays_sec, start=1): if delay > 0: logger.info( "hris sync retry %d/%d for instance %s after %ds (prev: %s)", attempt_idx, len(retry_delays_sec), alias, delay, type(last_exc).__name__ if last_exc else "n/a", ) await asyncio.sleep(delay) try: async with async_session_factory() as session: # Bump the per-session lock-wait timeout so the # confirm_batch SELECT FOR UPDATE + chained upserts # have room to finish even if a chunk of the batch # is large. Default (50s) is tight for batches # >1k staged events; 120s leaves headroom without # hiding real deadlocks. await session.execute(text("SET SESSION innodb_lock_wait_timeout = 120")) service = SuccessFactorsLifecycleService.for_instance( session, integration_id=row.id, config=instance_config, ) result = await service.run_scheduled_incremental( triggered_by=f"system:scheduler:{alias}", lookback=timedelta(hours=lookback_hours), auto_confirm=True, ) await session.commit() # #1745: the mass-leaver brake halts auto-confirm WITHOUT # raising — it returns a result carrying # ``auto_confirm_blocked_reason`` and leaves the batch in # ``pending_review``. Pre-fix the handler discarded that # reason, counted the tick a full success, and offboarding # stalled silently (no notification, green dashboards). Treat # a brake trip as a distinct outcome: record it, do NOT count # it as a synced instance, and fold it into the terminal raise # so the failure-notifier pushes an actionable review card. blocked_reason = result.get("auto_confirm_blocked_reason") if blocked_reason: logger.warning( "hris sync mass-leaver brake tripped for instance %s " "— batch %s left in pending_review: %s", alias, result.get("batch_id"), blocked_reason, ) summary["blocked"].append({ "slug": alias, "batch_id": result.get("batch_id"), "reason": blocked_reason, }) else: summary["instances_synced"] += 1 summary["total_new"] += int(result.get("new_count", 0)) summary["total_updated"] += int(result.get("updated_count", 0)) summary["total_reactivated"] += int(result.get("reactivated_count", 0)) summary["total_deactivated"] += int(result.get("deactivated_count", 0)) last_exc = None break # success — move to next instance except OperationalError as exc: # Only retry on MySQL lock-wait / deadlock. Everything # else (connection drop, bad SQL, FK violation) is a # real bug that retry can't fix — fail fast. orig = exc.orig errno = getattr(orig, "args", [None])[0] if orig else None if errno in (_MYSQL_LOCK_WAIT_TIMEOUT, _MYSQL_DEADLOCK): last_exc = exc logger.warning( "hris sync hit MySQL errno %s on instance %s attempt %d/%d", errno, alias, attempt_idx, len(retry_delays_sec), ) continue logger.exception("hris sync failed for instance %s (non-retryable)", alias) summary["errors"].append({"slug": alias, "error": format_exception_for_log(exc)}) last_exc = None break except Exception as exc: # noqa: BLE001 logger.exception("hris sync failed for instance %s", alias) summary["errors"].append({"slug": alias, "error": format_exception_for_log(exc)}) last_exc = None break if last_exc is not None: # Exhausted all retries on a lock-wait / deadlock. logger.error( "hris sync EXHAUSTED retries for instance %s (last: %s)", alias, last_exc, ) summary["errors"].append({ "slug": alias, "error": f"lock_wait_exhausted after {len(retry_delays_sec)} attempts: {format_exception_for_log(last_exc, max_len=400)}", }) # Raise on systemic failure: zero instances synced AND we tried at # least one. A multi-day SuccessFactors outage / credential rotation # otherwise produces zero pages because per-instance errors are # captured into ``summary["errors"]`` but the handler returns ok. # Roster goes stale silently; downstream leaver / hire workflows # quietly stall. # Pre-launch hardening 2026-05-17 (Round-19 audit, finding I): # also raise on partial-failure when a majority of instances fail # in one tick. Pre-fix: 9 of 10 SF tenants failing (e.g. a region # outage hitting most of the multi-tenant roster) landed # ``status='ok'`` because instances_synced==1, and downstream # leaver / hire flows quietly stalled on the 9 stale rosters. total_attempted = summary["instances_synced"] + len(summary["errors"]) majority_failed = ( total_attempted >= 3 and len(summary["errors"]) * 2 >= total_attempted ) if (summary["errors"] and summary["instances_synced"] == 0) or majority_failed: err_preview = summary["errors"][:3] raise RuntimeError( f"hris.incremental_sync: instances_synced=" f"{summary['instances_synced']} errors={len(summary['errors'])} " f"of {len(instances)} instances — systemic failure: " f"{err_preview}" ) # #1745: raise on any brake trip so the run lands ``status='failed'`` and # the scheduled-job failure-notifier fires its standard IT-admin card # (dedup-aware: consecutive trips collapse to one signature, so admins # aren't spammed every hour — the numbers are stripped from the # signature). Offboarding is genuinely stalled behind a pending_review # batch until a human confirms it, so a silent 'ok' tick is wrong. The # per-instance work already committed in its own session above, so this # raise only marks the RUN failed — it does not roll back the staged # batches (which is exactly what we want: they stay in pending_review for # review). Systemic hard-failures take priority above; this surfaces the # softer "awaits manual review" condition. if summary["blocked"]: blocked_preview = [ f"{b['slug']} (batch {b['batch_id']}): {b['reason']}" for b in summary["blocked"][:3] ] raise RuntimeError( f"hris.incremental_sync: mass-leaver brake tripped on " f"{len(summary['blocked'])} of {len(instances)} instance(s) — " f"batch(es) left in pending_review awaiting manual IT review. " f"Verify the SF tenant scope / offboard volume, then confirm or " f"cancel the staged batch in the sync-batches admin page. " f"{blocked_preview}" ) return summary # ── HRIS full-roster sync (catches historical leavers beyond the # incremental 30-day window) ────────────────────────────────── @register_scheduled_job( key="hris.full_sync", weight="heavy", # 全量 roster 拉取,分钟级 — 见 registry.SchedJobWeight display_name="HRIS 全量同步(历史离职兜底)", description=( "每周对每个已启用的 HRIS 实例做一次 ``list_all_users`` 全量同步:" "拉取 SF 所有 ``status='t'`` + ``status='f'`` 用户,按 status 路由" "为 onboard / offboard 事件,自动确认。" "为什么需要它:增量同步硬上限回看 30 天,任何 SF " "``lastModifiedDateTime`` 在窗外的人(典型场景:2+ 月前离职、" "SF 数据迁移漏推)永远进不来。2026-06-01 wangdi@pingcap.com (P0175, " "离职 2026-03-20,73 天前)即为线上实例 —— PR #588 已在代码上修复 " "bootstrap 路径,本任务把同一路径接进调度器,让漂移再发生时 7 天内自愈。" "Mass-leaver brake 在 ``sync_mode='full'`` 路径已豁免(confirm_batch " "L560);historical leavers 通过 ``_upsert_person`` INSERT 时即" "落 ``employment_status='left'``,不触发 offboarding 副作用。" ), ) async def run_hris_full_sync( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.core.database import async_session_factory from app.domains.hr.service import SuccessFactorsLifecycleService from app.integrations.compat import list_hris_instances instances = await list_hris_instances(session=db) if not instances: return { "instances_synced": 0, "total_new": 0, "total_updated": 0, "total_deactivated": 0, "errors": [], "message": "未配置任何 HRIS 集成", } summary: dict[str, Any] = { "instances_synced": 0, "total_new": 0, "total_updated": 0, "total_reactivated": 0, "total_deactivated": 0, "errors": [], } # Each instance uses a FRESH session: a full-sync batch can stage # hundreds of events; sharing the worker's outer session with the # incremental tick's retry path or with audit-flush listeners would # blur transaction boundaries. Mirror the incremental handler's # session-per-instance shape so behaviour is symmetric. for row, instance_config in instances: alias = row.slug or str(row.id) try: async with async_session_factory() as session: # Same SESSION lock-wait bump as incremental — a full # sync that confirms 400+ events holds row locks on # identity_persons / identity_external_identities / # hr_sync_staged_events for several seconds at a time. await session.execute(text("SET SESSION innodb_lock_wait_timeout = 120")) service = SuccessFactorsLifecycleService.for_instance( session, integration_id=row.id, config=instance_config, ) result = await service.run_scheduled_full( triggered_by=f"system:scheduler:{alias}", auto_confirm=True, ) await session.commit() summary["instances_synced"] += 1 summary["total_new"] += int(result.get("new_count", 0)) summary["total_updated"] += int(result.get("updated_count", 0)) summary["total_reactivated"] += int(result.get("reactivated_count", 0)) summary["total_deactivated"] += int(result.get("deactivated_count", 0)) except ConflictError as exc: # A volume brake (migration 565) is not a partial failure to # tolerate — it is the platform refusing to auto-apply a batch # that would create or deactivate people at a rate that looks # like a bug, and parking it at ``pending_review`` for a human. # Reported separately from generic errors because the # partial-failure policy below would let a single braked # instance pass as "ok" on a two-instance install: the batch # then waits for an admin who was never told, which is the # exact silence the brake exists to break. logger.warning( "hris full sync BRAKED for instance %s — batch left at " "pending_review for human review: %s", alias, exc, ) summary.setdefault("braked", []).append( {"slug": alias, "reason": str(exc)}, ) except Exception as exc: # noqa: BLE001 logger.exception("hris full sync failed for instance %s", alias) summary["errors"].append({"slug": alias, "error": format_exception_for_log(exc)}) # A brake always raises, regardless of how many other instances # succeeded. Handler-status discipline: the job row is the only # surface that says a roster-sized change is sitting unapplied. if summary.get("braked"): raise RuntimeError( f"hris.full_sync: {len(summary['braked'])} instance(s) hit a " f"volume brake and are waiting at pending_review for a human " f"— {summary['braked'][:3]}" ) # Same fail-loud contract as incremental: zero-success-but-tried OR # majority-failed must raise so the scheduled-job row shows ``failed`` # and the failure notifier pages the admin. Silent partial success # is what got wangdi stuck in the first place. total_attempted = summary["instances_synced"] + len(summary["errors"]) majority_failed = ( total_attempted >= 3 and len(summary["errors"]) * 2 >= total_attempted ) if (summary["errors"] and summary["instances_synced"] == 0) or majority_failed: err_preview = summary["errors"][:3] raise RuntimeError( f"hris.full_sync: instances_synced=" f"{summary['instances_synced']} errors={len(summary['errors'])} " f"of {len(instances)} instances — systemic failure: " f"{err_preview}" ) return summary # ── HRIS full-roster reconciliation (drift detection) ──────────── # How long a future hire must stay absent from SF's ``get_users`` before the # daily cross-check soft-deletes them. The check runs DAILY, so 20h guarantees # the absence spanned ≥2 consecutive runs (with cron jitter tolerance) — a # transient one-day SF-Onboarding flicker (P10019 陈建华) never accumulates to # a delete. Small enough that a genuinely cancelled requisition still clears # within ~a day. _FUTURE_HIRE_ABSENCE_GRACE = timedelta(hours=20) @register_scheduled_job( key="hris.absent_person_check", display_name="HRIS 全量名册对账(消失员工 → 软坟头标记)", description=( "每日比对 SuccessFactors 当前活跃名册 vs Helix 中 employment_" "status='active'、source_system='successfactors' 且 hris_record_" "state='present' 的员工。如果某人在 Helix 里活跃但 SF 已经查不到 " "user_id(管理员直接删了 SF 用户、数据迁移漏推、webhook 吞事件等)," "增量同步永远感知不到。" "本任务在 mass-leaver brake 内(25 / 5% 双阈值 + 50% 上限)给消失的" "员工打软坟头:hris_record_state='purged' + hris_purged_at + hris_" "purged_reason,并写一条 hr_lifecycle_events(event_type='hris_purged')" "保留可追溯记录。绝不物理删除——资产、权限、审计 FK 全部保留。" "管理员看到 admin/users 列表里出现「已从 HRIS 抹除」徽章,可以决定" "是否再进一步走离职流程(employment_status 不在本任务范围内)。" "另外:未来入职者(hire_date 尚未到)在当前名册里天然缺席,不能按名册" "差异打坟头(会误伤 P10012/P10014 这类真·待入职)。本任务改为对这批人" "逐个按 user_id 回查 SF(get_users,ANY-status),SF 已查不到的= 被撤销的" "招聘需求(如 P10019),因其尚未到岗、无账号/资产/SSO 可回收,直接软删+" "打坟头,彻底从列表消失。" ), ) async def run_hris_absent_person_check( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: """Detect Helix-active persons who are missing from the live SF roster. Why this job exists: ``run_hris_incremental_sync`` is event-driven — it walks SF's ``lastModifiedDateTime > cursor`` slice and only acts on persons SF emits events for. If SF stops emitting (admin hard- deletes the user, webhook silently 5xx'd, the row gets purged during a tenant migration), Helix has zero visibility — the person stays ``employment_status='active'`` forever and continues to occupy paid seats / hold permissions / appear in mail-group pickers. What the job does (migration 222 upgrade, 2026-05-30): 1. Brakes refuse to act on implausible candidate counts (>5% or >25 per instance, or >50% of the instance roster) — same shape as the webhook + manual-confirm mass-leaver brake. 2. For every candidate that passes the brake, stamp a **soft tombstone** on the IdentityPerson row: ``hris_record_state= 'purged'``, ``hris_purged_at=now``, ``hris_purged_reason= 'absent_from_full_sync'``. The row is NEVER hard-deleted — downstream FKs (asset allocations, audit logs, account history, email-group membership) stay intact. 3. Write an ``hr_lifecycle_events`` row (event_type=``hris_purged``) so the trail is queryable alongside onboard / offboard events. 4. Write the existing per-person audit row (action=``hris.absent_ person_detected``) for cross-system observability. Idempotency: candidates are filtered by ``hris_record_state='present'`` so an already-purged person is never re-scanned and never re-audited. A second run after every absentee has been marked returns ``candidates_total=0`` — the signal stops repeating. Employment-status flip is intentionally OUT of scope: marking ``leaver_pending`` triggers offboarding side-effects (asset transfer scan, SSO suspend, Feishu notice) which the user must consciously initiate. The tombstone here only records the fact that HRIS has lost the row; admins decide whether to follow up with offboarding. Future-hire cross-check (2026-07-17, P10019 post-mortem): the roster diff in step 2 skips any hire whose ``hire_date`` is still in the future — a hire whose EmpJob starts later has no effective-today row, so ``list_current_jobs`` never carries them and a naive diff would tombstone every legit upcoming hire (P10012/P10014). That exemption is also a blind spot: a future hire SF *cancelled* looks identical. So future hires are collected separately and each is verified against SF by userId (``get_users``, ANY-status filter — returns future ``active_external_suite`` hires too). A userId SF no longer returns was genuinely removed; since a never-started hire has nothing downstream to offboard, it is **soft-deleted** (``deleted_at`` + ``purged`` tombstone + SF identity soft-deleted), not merely tombstoned — so it disappears from the directory entirely. Guarded by the same brake plus a "get_users returned nothing for ≥2 asks = treat as SF scope/outage" check so a blip can't mass-delete real incoming hires. Reversible: the SF upsert path revives a soft-deleted person/identity if the same userId is ever re-hired. """ from zoneinfo import ZoneInfo from app.audit.service import AuditService from app.domains.hr.models import HrLifecycleEvent from app.identity.person_models import ( IdentityExternalIdentity, IdentityPerson, ) from app.identity.successfactors_auth import build_auth_from_config from app.identity.successfactors_client import SuccessFactorsClient from app.integrations.compat import list_hris_instances instances = await list_hris_instances(session=db) summary: dict[str, Any] = { "instances_checked": 0, "candidates_total": 0, # Number of rows whose ``hris_record_state`` was actually flipped # from ``present`` → ``purged`` this run. Diverges from # ``candidates_total`` only when a per-person SAVEPOINT failed # (DB hiccup mid-row) — those rows still appear in alerted_persons # but didn't get the tombstone stamp; admins should re-run. "purged_persons": 0, # Future hires SF has genuinely removed (cancelled/withdrawn # requisitions), verified by a per-userId ``get_users`` lookup and # soft-deleted here. Kept separate from ``purged_persons`` because # the action differs — a never-started hire is soft-deleted, not # just tombstoned (see the future-hire cross-check below). "cancelled_future_hires": 0, # Future-hire absence debounce (2026-07-20, P10019): a first/within- # grace absence is recorded but NOT deleted; a return clears it. "future_hire_absence_grace_started": 0, "future_hire_absence_grace_pending": 0, "future_hire_absence_cleared": 0, "alerted_persons": [], "brake_tripped": [], "errors": [], } if not instances: summary["message"] = "未配置任何 HRIS 集成" return summary audit = AuditService(db) trace_id = f"hris.absent_person_check:{datetime.now(UTC).isoformat()}" # Brake threshold: matches the webhook + manual confirm flows so # a corrupted SF roster (or a single bad tick) can't generate a # mass-leaver alert that scares the admin into bulk-flipping # everyone. ``max(25, 5%)`` floor anchors the brake at "25 or # 5% of active, whichever is larger" — same shape as the # ``process_webhook_events`` guard in hr/service.py. active_total_global = await db.scalar( select(func.count()) .select_from(IdentityPerson) .where( IdentityPerson.employment_status == "active", IdentityPerson.deleted_at.is_(None), ) ) or 0 threshold = max(25, int(active_total_global * 0.05)) for row, instance_config in instances: alias = row.slug or str(row.id) try: # SAVEPOINT (2026-08-06 review): scope this iteration's writes so a # DB-level failure rolls back only its own work. ``savepoint`` # re-raises everything, so the handler chain below keeps its exact # semantics — it only stops a failure from poisoning the session # for the remaining items and the run's trailing commit. async with savepoint(db, label="cron item"): base_url = (instance_config.get("base_url") or "").rstrip("/") if not base_url: summary["errors"].append({ "slug": alias, "error": "missing base_url", }) continue sf_auth = build_auth_from_config(instance_config) client = SuccessFactorsClient(base_url=base_url, auth=sf_auth) # Pull the full SF active roster. ``limit=None`` walks every # page (the underlying ``_list_entities`` paginates); the # default 10 in the signature is a small-batch test shortcut. sf_jobs = await client.list_current_jobs(limit=None) sf_user_ids: set[str] = { str(j["userId"]) for j in sf_jobs if j.get("userId") } if not sf_user_ids: summary["errors"].append({ "slug": alias, "error": ( "SF returned 0 active jobs — refusing to flag " "candidates (likely an SF outage / scope issue)" ), }) continue # Helix-active persons sourced from THIS instance. # ``hris_record_state='present'`` filter (migration 222) skips # rows we've already tombstoned in a previous run so a person # SF lost two weeks ago doesn't keep showing up as a fresh # candidate every night. helix_active = list((await db.execute( select(IdentityPerson).where( IdentityPerson.source_system == "successfactors", IdentityPerson.employment_status == "active", IdentityPerson.deleted_at.is_(None), IdentityPerson.hris_record_state == "present", IdentityPerson.hris_instance_id == row.id, ) )).scalars().all()) absent: list[IdentityPerson] = [] # Future / day-one hires are structurally absent from the # effective-today roster whether they are alive OR cancelled # in SF — collected here and verified directly against SF by # userId after the roster diff (future-hire cross-check). future_hires: list[IdentityPerson] = [] # Company business day, Asia/Shanghai — the SAME ``>=`` # boundary convention ``classify_inactive_sf_user`` and the # #1952 full/incremental fix use for pre-start hires. Strict # ``>`` on a UTC date (the pre-fix shape here) left a day-one # hire unprotected: SF-DC/UTC date skew can keep their EmpJob # row out of the effective-today roster on their first day, # and they'd be tombstoned the morning they start. today_sh = datetime.now(ZoneInfo("Asia/Shanghai")).date() for p in helix_active: if not p.hris_person_id: # Nothing to compare against — likely a non-SF row # that someone mislabeled with source_system. Skip # rather than flag. continue if p.hire_date is not None and p.hire_date >= today_sh: # Future / day-one hire (2026-07-12 roger.zhou02 # post-mortem; boundary widened to ``>=`` + Shanghai # date per the #1952 convention, 2026-07-16): the # roster above is ``list_current_jobs`` — EmpJob # rows effective AS OF TODAY. A hire whose job starts # in the future has no effective row yet, so they are # structurally absent from this roster while being # perfectly alive in SF (静文 P10012 / 亦寒 P10014 were # tombstoned exactly this way, which then blocked # their day-1 provisioning). Absence only means # something once the start date has passed. # # But that exemption is a blind spot: a future hire SF # *cancelled* is byte-identical to a live one from the # roster's view (P10019, 2026-07-17). Don't tombstone # from the roster diff — hand these to the future-hire # cross-check below, which verifies each against SF by # userId and can tell alive from removed. future_hires.append(p) continue if p.hris_person_id not in sf_user_ids: absent.append(p) summary["instances_checked"] += 1 # Per-instance brake — even though the global threshold is # set above. A single instance returning a tiny roster # shouldn't be allowed to mass-flag because the GLOBAL active # count is huge. Per-instance check uses the instance's own # active count. # # Two-layered brake (Round 47 hardening, 2026-05-23): # 1. Absolute floor: ``max(25, 5% global)`` ∩ ``max(25, 5% # instance)`` — same shape as confirm_batch + webhook. # 2. Percentage ceiling: regardless of the floor, refuse to # surface alerts when > 50% of an instance has vanished # in a single tick. A 10-person instance with 9 absent # previously slipped the 25-floor and was about to # mass-flag 90% of the roster as leavers. The ceiling # only applies when the instance has ≥ 4 active rows, # otherwise a legitimate 1-of-2 hard-delete would be # braked away (50% absent on a 2-person tenant IS # exactly the detection case this job was built for). instance_active_count = len(helix_active) instance_threshold = max(25, int(instance_active_count * 0.05)) effective_threshold = min(threshold, instance_threshold) percentage_ceiling_tripped = ( instance_active_count >= 4 and len(absent) * 2 > instance_active_count # > 50% ) if ( len(absent) > effective_threshold or percentage_ceiling_tripped ): summary["brake_tripped"].append({ "slug": alias, "candidate_count": len(absent), "threshold": effective_threshold, "instance_active": instance_active_count, "sf_active": len(sf_user_ids), }) logger.warning( "hris.absent_person_check: instance %s tripped brake " "— %d candidates vs threshold %d (instance_active=%d, " "sf_active=%d). Refusing to alert; investigate the " "SF roster integrity before re-running.", alias, len(absent), effective_threshold, instance_active_count, len(sf_user_ids), ) # Audit row for the brake itself so admin sees it in # /admin/audit even if the per-person rows below didn't # fire. Wrap in a SAVEPOINT so a malformed audit row # (UNIQUE collision, JSON-encode failure on weird names) # doesn't poison the outer session and silently corrupt # the next instance's writes — same pattern as the # apikey ``_record_dedup_send`` fix from Round 44. try: async with db.begin_nested(): await audit.log( trace_id=trace_id, actor_type="system", actor_user_id=None, action="hris.absent_check_brake_tripped", status="partial", resource_type="integration", resource_id=str(row.id), detail={ "slug": alias, "candidate_count": len(absent), "threshold": effective_threshold, "instance_active": instance_active_count, "sf_active": len(sf_user_ids), }, ) except Exception: # noqa: BLE001 logger.exception( "hris.absent_person_check: brake-audit write failed", ) continue purged_now = datetime.now(UTC) purged_effective_date = purged_now.date() for p in absent: summary["alerted_persons"].append({ "person_id": str(p.id), "hris_person_id": p.hris_person_id, "work_email": p.work_email, "display_name": p.display_name, "instance_slug": alias, }) # SAVEPOINT per row so a single bad audit insert (UNIQUE # collision, JSON serialization issue on an unusual # display_name) or a per-row tombstone write doesn't # roll back the outer session and silently drop the # remaining persons' marks. One SAVEPOINT covers BOTH # the tombstone update and the audit + lifecycle rows # so they always commit atomically — admins reading # alerted_persons can trust that those rows are marked. try: async with db.begin_nested(): # 1. Soft tombstone (migration 222). ``p`` is # already attached to ``db`` from the SELECT # above, so the assignments below land via the # session's dirty-tracking — no explicit add() # needed. We do NOT touch employment_status: # following up with leaver_pending fires # offboarding side-effects (asset transfer, # SSO suspend, Feishu notice) and must be a # deliberate human decision. p.hris_record_state = "purged" p.hris_purged_at = purged_now p.hris_purged_reason = "absent_from_full_sync" # 2. Lifecycle event — queryable from the same # ``hr_lifecycle_events`` table that already # carries onboard / offboard, so reporting # tools that already pivot on ``event_type`` # pick this up without schema changes. db.add(HrLifecycleEvent( person_id=p.id, event_type="hris_purged", effective_date=purged_effective_date, detail={ "hris_person_id": p.hris_person_id, "work_email": p.work_email, "display_name": p.display_name, "instance_slug": alias, "instance_id": str(row.id), "reason": "absent_from_full_sync", }, )) # 3. Audit row (pre-existing behaviour) — kept # for cross-system observability so the audit # console shows the action timeline even when # the lifecycle table is queried separately. await audit.log( trace_id=trace_id, actor_type="system", actor_user_id=None, action="hris.absent_person_detected", status="success", resource_type="identity_person", resource_id=str(p.id), detail={ "hris_person_id": p.hris_person_id, "work_email": p.work_email, "display_name": p.display_name, "instance_slug": alias, "instance_id": str(row.id), "tombstone": "applied", }, ) summary["purged_persons"] += 1 except Exception as exc: # noqa: BLE001 logger.exception( "hris.absent_person_check: per-person tombstone/" "audit write failed for person=%s — row will be " "re-attempted next run since hris_record_state " "stays 'present' on rollback", p.id, ) # Feed the existing tail-raise (below): per-person # failures are the SOLE condition that flips the run # to status='failed'. Without appending here, a # deterministically-failing tombstone write would be # logged every tick while the run stays 'ok' and the # failure-notifier never fires. summary["errors"].append({ "slug": alias, "person_id": str(p.id), "error": format_exception_for_log(exc), }) summary["candidates_total"] += len(absent) # ── Future-hire cross-check (2026-07-17, P10019 post-mortem) ── # The roster diff above skips future hires on purpose (a hire # whose EmpJob starts later has no effective-today row, so a # naive diff would tombstone every legit upcoming hire — the # P10012/P10014 incident). That exemption is also the exact # blind spot that left P10019 — a future hire SF *cancelled* — # sitting ``active``/``present`` with nobody able to clean it # up. Resolve it directly: ask SF for each future hire by # userId. ``get_users`` uses the ANY-status filter, so it # returns active, forfeited AND future ``active_external_suite`` # hires — a userId it no longer returns has been genuinely # removed from SF. Runs only when the roster brake did NOT trip # (the ``continue`` above skips this), so an SF outage never # reaches here. if future_hires: try: sf_future_users = await client.get_users( [p.hris_person_id for p in future_hires] ) except Exception as exc: # noqa: BLE001 # Never cancel on an unverified signal. Skip the # cross-check this run; it retries on the next tick. logger.warning( "hris.absent_person_check: get_users failed for %d " "future hires on instance %s — skipping future-hire " "cross-check this run: %s", len(future_hires), alias, format_exception_for_log(exc), ) sf_future_users = None if sf_future_users is not None: returned_ids = { str(u["userId"]) for u in sf_future_users if u.get("userId") } cancelled_future = [ p for p in future_hires if p.hris_person_id not in returned_ids ] # A future hire SF returned again clears any prior # absence-grace marker — the earlier absence was a # transient SF-Onboarding flicker, not a cancellation. cleared_any = False for p in future_hires: if ( p.hris_person_id in returned_ids and p.hris_absent_since is not None ): p.hris_absent_since = None summary["future_hire_absence_cleared"] += 1 cleared_any = True if cleared_any: # Persist the cleared markers into the transaction (the # scheduled-job runner commits after the handler). await db.flush() # SF-blip protection. ``list_current_jobs`` already # proved SF is up (we ``continue``d on an empty roster # earlier), but guard the future-hire lookup too: an # empty ``get_users`` result for a set of ≥2 is a # scope/outage artifact, not a real mass-cancellation; # and reuse the roster's >50%-with-≥4 ceiling. A lone # cancelled future hire (the P10019 shape) still passes. blip = ( (not returned_ids and len(future_hires) >= 2) or ( len(future_hires) >= 4 and len(cancelled_future) * 2 > len(future_hires) ) ) if cancelled_future and blip: logger.warning( "hris.absent_person_check: future-hire cross-" "check braked on instance %s — %d/%d future " "hires absent from get_users (likely SF scope/" "outage, not mass cancellation); not removing.", alias, len(cancelled_future), len(future_hires), ) summary["brake_tripped"].append({ "slug": alias, "future_hire_cancel_candidates": len( cancelled_future ), "future_hires": len(future_hires), "kind": "future_hire_cross_check", }) elif cancelled_future: cancel_now = datetime.now(UTC) # Per-person absence DEBOUNCE (2026-07-20, P10019 陈建华). # SF-Onboarding pre-hires flicker in/out of get_users # while their requisition is in flux, so NEVER soft- # delete on a single absence (the old code deleted a # lone absent future hire immediately — it tombstoned a # REAL pre-hire that SF returned again the next day). # Record first-seen-absent; only soft-delete once the # absence has persisted past the grace window (≥2 # consecutive daily runs). A return clears the marker # (handled above). deleted_at is now revived on onboard # re-appearance too (_upsert_person), so even a delete # that later proves wrong self-heals. to_delete: list[IdentityPerson] = [] for p in cancelled_future: since = p.hris_absent_since if since is None: p.hris_absent_since = cancel_now summary["future_hire_absence_grace_started"] += 1 continue if since.tzinfo is None: since = since.replace(tzinfo=UTC) if cancel_now - since >= _FUTURE_HIRE_ABSENCE_GRACE: to_delete.append(p) else: summary["future_hire_absence_grace_pending"] += 1 # Persist the newly-recorded grace markers before the # per-person delete SAVEPOINTs (runner commits after). await db.flush() for p in to_delete: # One SAVEPOINT per person: soft-delete + # tombstone + identity + lifecycle + audit # commit atomically, and a single bad row # doesn't roll back the others. try: async with db.begin_nested(): p.hris_record_state = "purged" p.hris_purged_at = cancel_now p.hris_purged_reason = ( "cancelled_future_hire_absent_from_sf" ) # A never-started hire has NO accounts, # assets or SSO to offboard (the future- # hire provisioning gates deferred all of # it), so — unlike a mid-employment purge # — hide it entirely rather than leaving # an ``active``-looking row in the # directory. Soft-delete is reversible: # ``_upsert_external_identity`` / # ``_upsert_person`` are soft-delete- # tolerant, so a genuine re-hire of the # same userId later revives cleanly (no # 1062 on the provider-unique index). p.deleted_at = cancel_now ext = await db.scalar( select(IdentityExternalIdentity).where( IdentityExternalIdentity.person_id == p.id, IdentityExternalIdentity.provider == "successfactors", IdentityExternalIdentity.deleted_at .is_(None), ) ) if ext is not None: ext.deleted_at = cancel_now db.add(HrLifecycleEvent( person_id=p.id, event_type="hris_purged", effective_date=cancel_now.date(), detail={ "hris_person_id": p.hris_person_id, "work_email": p.work_email, "display_name": p.display_name, "hire_date": ( p.hire_date.isoformat() if p.hire_date else None ), "instance_slug": alias, "instance_id": str(row.id), "reason": ( "cancelled_future_hire_" "absent_from_sf" ), "soft_deleted": True, }, )) await audit.log( trace_id=trace_id, actor_type="system", actor_user_id=None, action=( "hris.cancelled_future_hire_" "removed" ), status="success", resource_type="identity_person", resource_id=str(p.id), detail={ "hris_person_id": p.hris_person_id, "display_name": p.display_name, "hire_date": ( p.hire_date.isoformat() if p.hire_date else None ), "instance_slug": alias, "tombstone": "applied", "soft_deleted": True, }, ) summary["cancelled_future_hires"] += 1 except Exception as exc: # noqa: BLE001 logger.exception( "hris.absent_person_check: future-hire " "removal failed for person=%s — will be " "re-attempted next run", p.id, ) summary["errors"].append({ "slug": alias, "person_id": str(p.id), "error": format_exception_for_log(exc), }) except Exception as exc: # noqa: BLE001 logger.exception( "hris.absent_person_check failed for instance %s", alias, ) summary["errors"].append({ "slug": alias, "error": format_exception_for_log(exc), }) # Trim alerted_persons in the returned summary so the # scheduled_job_runs row doesn't blow past its TEXT/JSON budget on a # 500-person tenant. Full set is in the audit log via the per-row # rows we just wrote. if len(summary["alerted_persons"]) > 50: truncated = len(summary["alerted_persons"]) - 50 summary["alerted_persons"] = summary["alerted_persons"][:50] summary["alerted_persons_truncated"] = truncated # Raise whenever ANY instance failed — partial failure is still a # failure per the platform-wide rule in # ``feedback_handler_status_must_raise``: scheduled-job handlers # must raise on failure so ``maybe_notify_failure`` actually fires. # Returning a summary that contains ``errors`` while letting # ``runs.status='ok'`` get written silently lost the ``positionNameEN`` # 400 storm for 10 days (2026-05-13 → 2026-05-23). The summary is # still attached to the run via ``scheduled_jobs/service.py``'s # error-result handling, so admins still see ``instances_checked`` # / ``alerted_persons`` from successful instances when they open # the failed run. if summary["errors"]: successful = summary["instances_checked"] total = successful + len(summary["errors"]) scope = ( "all instances failed" if successful == 0 else f"{len(summary['errors'])}/{total} instances failed" ) raise RuntimeError( f"hris.absent_person_check: {scope} — " f"{summary['errors'][:3]}" ) return summary # ── HRIS async-apply restart safety ─────────────────────────────── @register_scheduled_job( key="hris.resume_stuck_apply", display_name="HRIS 批次异步应用 · 重启兜底", description=( "每 60 秒扫描 status='applying' 且心跳 (apply_heartbeat_at) 已过期" "的 HRIS 同步批次 —— 这类批次由 confirm 接口 flip 到 applying 后用 " "background.spawn 起后台 apply,但后台任务因 backend 进程重启/崩溃" "中途丢失。重跑幂等的逐事件 apply 把剩余事件补完并落终态,保留原 " "reviewed_by。心跳新鲜的在途任务不会被打断(避免双跑)。这是 " "inline 主路径之外抗重启的第二条腰带,对齐迁移 239 / PR #279。" ), ) async def run_hris_resume_stuck_apply( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.domains.hr.service import SuccessFactorsLifecycleService stale_minutes = int(config.get("stale_after_minutes", 10)) limit = int(config.get("limit", 5)) async def _work(session: AsyncSession) -> dict[str, Any]: service = SuccessFactorsLifecycleService(session) return await service.resume_stale_applying_batches( stale_after=timedelta(minutes=stale_minutes), limit=limit, ) return await _run_in_fresh_session_with_lock_retry( _work, label="hris.resume_stuck_apply", ) # ── Overdue leaver check ────────────────────────────────────────── @register_scheduled_job( key="offboarding.reconcile_processes", display_name="离职流程对账", description=( "扫描所有处于 BLOCKED 或 READY 状态的离职流程,对每个 process " "调用 reconcile_blockers:把已 COMPLETED 的 provisioning task 关联的 " "blocker 翻为 revoked_auto,把已 COMPLETED 的 transfer 关联的 " "blocker 翻为 transferred,并按 _rescan 重新扫描以捕获离上次扫描" "之后新增的资源占用。这样运营无需手动点 'Reconcile' 按钮," "处理过程异步完成后状态自动收敛。建议 5 分钟一次。" ), ) async def run_offboarding_reconcile( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: """D-4 (2026-05-08): close the gap between async provisioning task completion and the offboarding blocker state. Provisioning tasks spawned by ``_apply_software_policy`` complete on their own, but nothing pushes their COMPLETED status back into the owning blocker until an admin manually clicks reconcile. This cron does that automatically on a 5-minute cadence so the offboarding dashboard keeps moving without manual prodding. """ from app.domains.offboarding.service import OffboardingService trace_id = f"system:cron:offboarding.reconcile:{uuid.uuid4()}" # reconcile_all_active_processes re-scans live BLOCKED/READY state, so a # fresh-session re-run after a transient lock / connection drop (a TiDB # Cloud node recycle can RST an in-flight query — errno 2013) is clean # and idempotent. summary = await _run_in_fresh_session_with_lock_retry( lambda session: OffboardingService(session).reconcile_all_active_processes( trace_id=trace_id ), label="offboarding.reconcile_processes", ) # Handler-must-raise contract: reconcile_all_active_processes collects # per-process failures into ``errors`` instead of propagating (so one # bad process can't abort the whole tick). But returning the summary # verbatim lands ``scheduled_job_runs.status='ok'`` even when processes # failed to reconcile every tick — silently, with no failure-notifier # page. Re-raise on any failure so the run records 'failed' and ops is # alerted; the successful processes are already committed by the # fresh-session wrapper, so this only affects run status, not the work. failed = summary.get("failed_count", 0) if isinstance(summary, dict) else 0 if failed: raise RuntimeError( f"offboarding.reconcile: {failed} process(es) failed to " f"reconcile — see errors in the run summary: {summary.get('errors')}" ) return summary @register_scheduled_job( key="hr.overdue_leavers_check", display_name="逾期离职员工状态推进", slo=SLO_OVERDUE_LEAVERS_CHECK, description=( "检查所有 leaver_pending 状态且 termination_date 已过期的员工," "尝试把他们推进到 left 状态。比较使用 [b]员工本地时区[/b] " "(identity_persons.country → IANA zone, 未知回退 Asia/Shanghai), " "确保美国员工在 PDT 23:59 之前都还活着,不会因为 UTC cron 时钟提前" "丢失邮箱 / 无法发告别信。被 Leaver Blocked(资产/软件未清)" "保留的员工出现在 held_user_ids;还在本地最后工作日(local" "尚未过 termination_date)的员工出现在 deferred_user_ids 等下一" "个 tick。同时跑经理升级 tick(reminder / auto-transfer)。" ), ) async def run_overdue_leavers_check( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.domains.hr.service import SuccessFactorsLifecycleService from app.domains.offboarding.escalation import LeaverEscalationService # Both phases run as one big multi-table mutation (escalation # auto-transfers + per-person ``flip_to_left`` touch identity rows, # software seats, cloud-scope cc lists, overrides, …). Under a # concurrent cron / admin save / HRIS sync grabbing the same rows in # a different order, InnoDB can pick this batch as the deadlock # victim (errno 1213) and roll the whole tick back. Retry each phase # on a fresh session — the canonical "restart transaction" response; # both phases re-scan live ``leaver_pending`` state so a re-run is # idempotent. escalation_summary: dict[str, Any] = ( await _run_in_fresh_session_with_lock_retry( lambda session: LeaverEscalationService(session).run_tick(), label="overdue_leavers_check:escalation_tick", ) ) transition_summary: dict[str, Any] = ( await _run_in_fresh_session_with_lock_retry( lambda session: SuccessFactorsLifecycleService( session ).transition_overdue_leavers(), label="overdue_leavers_check:transition", ) ) return { "escalation": escalation_summary, "transition": { "transitioned_count": transition_summary.get("transitioned_count", 0), "held_count": len(transition_summary.get("held_user_ids", [])), "held_user_ids": transition_summary.get("held_user_ids", []), "deferred_count": len( transition_summary.get("deferred_user_ids", []), ), "deferred_user_ids": transition_summary.get( "deferred_user_ids", [], ), }, } # ── Leaver platform-role cleanup ─────────────────────────────────── @register_scheduled_job( key="offboarding.revoke_leaver_platform_roles", display_name="离职人员系统角色清理", description=( "扫描所有 employment_status='left' 且仍持有特权平台角色" "(global_admin / auditor / cloud_admin / hr / finance / " "management / asset_global_admin)的人员,逐一吊销这些授予。" "这是离职流程内系统权限清理的兜底:offboarding 完成时已就地" "吊销(complete_process),该 cron 负责补撤历史遗留离职人员" "(本特性上线前已 left 的账号)以及离职后被误授予的漂移。" "复用 AuthService.revoke_platform_roles_for_leaver,幂等;" "刻意不动 employee(自身基础角色)与 asset_admin(由 " "location_admin_sync 协调派生,删了会被日对账重新加回)。" "末位 global_admin 会被跳过并告警而非吊销,避免锁死系统。" "建议每日一次。" ), ) async def run_revoke_leaver_platform_roles( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: trace_id = f"system:cron:offboarding.revoke_leaver_roles:{uuid.uuid4()}" # The scan re-reads live ``employment_status='left'`` + role state, # so a fresh-session re-run after a transient lock / connection drop # is clean and idempotent. return await _run_in_fresh_session_with_lock_retry( lambda session: _revoke_leaver_platform_roles(session, trace_id=trace_id), label="offboarding.revoke_leaver_platform_roles", ) async def _revoke_leaver_platform_roles( db: AsyncSession, *, trace_id: str, ) -> dict[str, Any]: from app.auth.models import AuthRole, AuthUserRole from app.auth.service import AuthService from app.core.rbac import LEAVER_REVOCABLE_ROLES, db_role_names_for from app.identity.person_models import IdentityPerson # DB-side role-name set, expanded to cover legacy aliases # (``it_admin`` / ``security_auditor`` rows) so a leaver who still # carries a pre-canonicalization grant is also caught. revocable_db_names: set[str] = set() for canonical in LEAVER_REVOCABLE_ROLES: revocable_db_names.update(db_role_names_for(canonical)) person_ids = ( await db.execute( select(IdentityPerson.id) .join(AuthUserRole, AuthUserRole.user_id == IdentityPerson.id) .join(AuthRole, AuthRole.id == AuthUserRole.role_id) .where( IdentityPerson.employment_status == "left", AuthRole.name.in_(revocable_db_names), ) .distinct() ) ).scalars().all() service = AuthService(db) cleaned = 0 roles_revoked = 0 details: list[dict[str, Any]] = [] swallowed_errors: list[str] = [] for pid in person_ids: # Per-person SAVEPOINT: one person's revoke failing must not # poison the whole batch's final commit (per-process isolation # pattern for cron loops over rows on a shared session). The # helper itself swallows the expected LAST_GLOBAL_ADMIN skip; # this guards against an unexpected per-row error. try: async with db.begin_nested(): revoked = await service.revoke_platform_roles_for_leaver( pid, trace_id=trace_id, ) except Exception as exc: # noqa: BLE001 # A deadlock / lock-wait (errno 1205/1213) aborts the whole # transaction server-side; swallowing it here orphans the # SAVEPOINT and the helper's final ``commit()`` then raises # ``PendingRollbackError`` — NOT an ``OperationalError``, so # the fresh-session lock-retry never fires and the real errno # is masked while every remaining leaver this tick is silently # skipped. Re-raise so ``_run_in_fresh_session_with_lock_retry`` # restarts cleanly (the scan re-reads live state and is # idempotent). Same boundary the sibling per-row loops use. if is_transaction_aborting_db_error(exc): raise logger.exception( "revoke_leaver_platform_roles: failed for person %s; " "continuing batch", pid, ) # Collect for the tail-raise: a leaver whose privileged-role # revoke deterministically fails would otherwise be logged # daily while the run stays status='ok' and the failure- # notifier never fires — the leaver keeps privileged RBAC # indefinitely. We finish the batch (other leavers may still # be cleanable) then raise so the run lands status='failed'. swallowed_errors.append(f"{pid}: {format_exception_for_log(exc)}") continue if revoked: cleaned += 1 roles_revoked += len(revoked) details.append({"person_id": str(pid), "roles": revoked}) summary = { "scanned_leavers_with_roles": len(person_ids), "leavers_cleaned": cleaned, "roles_revoked": roles_revoked, "detail": details[:50], "errors": swallowed_errors[:50], } # Tail-raise (feedback_handler_status_must_raise): any per-leaver # revoke that failed for a non-transient reason must flip the run to # status='failed' so the dedup-aware failure-notifier pages. The # summary above is still attached to the failed run, so admins see # how many leavers WERE cleaned alongside the failures. Mirrors the # tail-raise in hr.process_advance / run_recover_orphaned_left. if swallowed_errors: raise RuntimeError( f"offboarding.revoke_leaver_platform_roles: " f"{len(swallowed_errors)} of {len(person_ids)} leaver(s) " f"raised during privileged-role revoke — " f"{swallowed_errors[:3]}" ) return summary # ── CMDB domain expiry alert ────────────────────────────────────── # Role label per language for the expiry card. ``role`` is the # expiry-engine's recipient kind, not an RBAC role. _EXPIRY_ROLE_LABELS: dict[str, dict[str, str]] = { "owner": {"zh": "负责人", "en": "Owner", "ja": "担当者"}, "domain_owner": { "zh": "域名负责人", "en": "Domain owner", "ja": "ドメイン担当者", }, "lead": {"zh": "团队 lead", "en": "Team lead", "ja": "チームリード"}, } _EXPIRY_ACK_BUTTON = LocalizedText( zh="标记续期中(暂停提醒)", en="Mark as renewing (snooze)", ja="更新手続き中にする（通知を停止）", ) _EXPIRY_DETAIL_BUTTON = LocalizedText( zh="查看详情", en="View details", ja="詳細を見る", ) def _make_expiry_notifier( session, *, kind_label_zh: str, kind_label_en: str, kind_label_ja: str, ): """Build the ``notify(item, recipients, bucket)`` callable the expiry engine fans out with: an in-app inbox bell + a best-effort Feishu card per recipient, the card carrying a one-click "标记续期中" magic link (action token) + a "查看详情" deep link. Lives in the orchestration layer because it imports the notifications / approval domains. The card copy comes from the ``cmdb.expiry_alert`` template (zh / en / ja). Several pieces of that sentence have to be *translated* rather than *substituted* — the resource noun, the "expires in N days" clause, the recipient's role, the optional host/registrar line. Those are composed here in all three languages and passed as locale fragments (``when_zh`` / ``when_en`` / ``when_ja``, …), the convention ``cloud.billing_alert`` established: the stored context then carries every language, so the in-app bell still renders a coherent row if the reader later switches their language preference. The card's own chrome — subtitle and action-button labels — is NOT template-rendered (the loader has nowhere to put a ``CardAction``), so it is resolved per recipient via :func:`resolve_lang` instead. ``kind_label_*`` is the resource noun ("域名" / "Domain" / "ドメイン"). """ from datetime import UTC as _UTC from datetime import datetime as _dt from app.core.locale import resolve_lang from app.core.security import create_action_token from app.domains.cmdb.models import EXPIRY_RESOURCE_CERTIFICATE from app.notifications.service import ( CardAction, NotificationEnvelope, NotificationService, ) notif = NotificationService(session) kind_labels = { "zh": kind_label_zh, "en": kind_label_en, "ja": kind_label_ja, } async def notify(item, recipients, bucket) -> bool: today = _dt.now(_UTC).date() if item.expiry_date is None: return False days = (item.expiry_date - today).days if bucket == "expired": when = { "zh": f"已过期 {abs(days)} 天(到期日 {item.expiry_date})", "en": ( f"Expired {abs(days)} day(s) ago " f"(expiry date {item.expiry_date})" ), "ja": ( f"{abs(days)} 日前に有効期限が切れました" f"（有効期限 {item.expiry_date}）" ), } else: when = { "zh": f"还有 {days} 天到期(到期日 {item.expiry_date})", "en": ( f"Expires in {days} day(s) " f"(expiry date {item.expiry_date})" ), "ja": ( f"あと {days} 日で有効期限が切れます" f"（有効期限 {item.expiry_date}）" ), } if item.detail: detail_line = { "zh": f"\n主机/注册商:{item.detail}", "en": f"\nHost / registrar: {item.detail}", "ja": f"\nホスト / レジストラ：{item.detail}", } else: detail_line = {"zh": "", "en": "", "ja": ""} is_cert = item.resource_type == EXPIRY_RESOURCE_CERTIFICATE detail_url = ( f"/admin/cloud/certificates?search={item.label}" if is_cert else f"/admin/cloud/domains?search={item.label}" ) # Dedup is 本人-centric: the engine only advances last_alerted_on # when the OWNER (本人) was delivered, so a flaky Feishu push to an # escalation recipient never silently marks the owner as alerted. # The inbox bell row is written regardless (push_inbox_and_feishu # is bell-first), so "delivered" here means the owner's send didn't # raise. No owner recipient (shouldn't happen) → fall back to any. owner_ok = False any_ok = False has_owner = any(r.role == "owner" for r in recipients) for r in recipients: # Card buttons + subtitle are NOT template-rendered (the # template loader has nowhere to put a CardAction), so they # have to be localised right here or a Japanese reader gets a # Japanese body under Chinese buttons. Button labels are # ``LocalizedText`` (resolved per recipient by the notification # layer); the subtitle still picks its variant here because # ``when`` is composed from this item's dates. lang = resolve_lang(r.person) roles = _EXPIRY_ROLE_LABELS.get(r.role) actions = [] # The ack (snooze) action is offered to the owner + 域名负责人 — # the people who can actually act on the renewal. if r.role in ("owner", "domain_owner"): tok = create_action_token( purpose="expiry_ack", subject_id=str(r.person.id), resource_id=f"{item.resource_type}:{item.resource_id}", ttl_days=30, ) actions.append(CardAction( label=_EXPIRY_ACK_BUTTON, url=f"/cmdb/expiry/ack?token={tok}", style="primary", )) actions.append(CardAction( label=_EXPIRY_DETAIL_BUTTON, url=detail_url, )) env = NotificationEnvelope( recipient=r.person, # Fallback only — ``cmdb.expiry_alert`` is registered, so # the catalog's localised copy is what actually ships. title=f"[{kind_label_zh}到期提醒] {item.label}", body_text=( f"{when['zh']}\n" f"你的角色:{(roles or {}).get('zh', r.role)}" f"{detail_line['zh']}" ), channels=["feishu"], template_key="cmdb.expiry_alert", url=detail_url, card_subtitle=when[lang], card_actions=actions, related_resource_type=item.resource_type, context={ "label": item.label, # Locale fragments — all three languages travel on the # context so the bell can re-render the row correctly # after the reader switches their language preference. # ``roles`` is None for a recipient kind the engine # grew without a label here: degrade to the raw key # rather than blanking the line. **{ f"kind_label_{lg}": kind_labels[lg] for lg in ("zh", "en", "ja") }, **{f"when_{lg}": when[lg] for lg in ("zh", "en", "ja")}, **{ f"role_{lg}": (roles or {}).get(lg, r.role) for lg in ("zh", "en", "ja") }, **{ f"detail_line_{lg}": detail_line[lg] for lg in ("zh", "en", "ja") }, }, ) try: await notif.push_inbox_and_feishu(env) any_ok = True if r.role == "owner": owner_ok = True except Exception: # noqa: BLE001 logger.exception( "expiry_alert: notify failed for %s recipient %s", item.resource_id, r.person.id, ) return owner_ok if has_owner else any_ok return notify @register_scheduled_job( key="cmdb.domain_expiry_alert", display_name="域名到期提醒", description=( "域名(注册商)到期的阶梯式提醒 + 临期升级,委托给统一到期告警引擎" "(T-30 / T-14 各一次,最后 7 天起每天,过期后每天;最后一公里升级给" "owner + lead)。owner 可一键「标记续期中」暂停提醒。environment=prod " "的生产域名不在提醒范围(生产续期由 Helix 自助流程之外的团队负责)。受 " "network.user_notifications_enabled 开关控制。" ), ) async def run_domain_expiry_alert( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.core.database import async_session_factory from app.domains.cmdb.expiry_alert_engine import ( build_domain_expiry_items, run_expiry_alerts, ) from app.domains.cmdb.models import EXPIRY_RESOURCE_DOMAIN from app.domains.cmdb.network_notifications import ( network_user_notifications_enabled, ) if not await network_user_notifications_enabled(db): return { "status": "skipped", "reason": "network.user_notifications_enabled is false", } async with async_session_factory() as session: items = await build_domain_expiry_items(session) notify = _make_expiry_notifier( session, kind_label_zh="域名", kind_label_en="Domain", kind_label_ja="ドメイン", ) summary = await run_expiry_alerts( session, resource_type=EXPIRY_RESOURCE_DOMAIN, items=items, config=config, notify=notify, ) await session.commit() if summary["owners_notified"] == 0 and summary["send_failures"] > 0: raise RuntimeError( "cmdb.domain_expiry_alert: every notification failed " f"(failures={summary['send_failures']}). Check Feishu credentials." ) return summary # ── Certificate probe sweep ─────────────────────────────────────── @register_scheduled_job( key="cmdb.cert_probe_sweep", display_name="证书探测扫描", description=( "对启用租户下 DNS(A/AAAA/CNAME)+ 域名(apex)凑出的主机清单逐个发起 " "443 TLS 握手,读取对端实际下发的叶子证书并 upsert 到 cmdb_certificates。" "并发受限 + 单主机超时;探测不到的主机标记 unreachable(绝不当作 expired)," "不为不可达主机凭空建行。非用户通知类任务,不受 " "network.user_notifications_enabled 开关影响。" ), ) async def run_cert_probe_sweep( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: import asyncio as _asyncio from app.core.database import async_session_factory from app.domains.cmdb.cert_probe import probe_certificate from app.domains.cmdb.cert_service import ( build_probe_targets, mark_orphaned_probe_targets, upsert_probed_certificate, ) concurrency = max(1, min(50, int(config.get("concurrency") or 20))) timeout = max(1.0, min(30.0, float(config.get("timeout_seconds") or 8.0))) summary: dict[str, Any] = { "targets": 0, "created": 0, "updated": 0, "unreachable": 0, "skipped": 0, # Probed certs whose host has LEFT the target set (DNS record disabled # or removed at the provider). Not a probe failure — the sweep simply # has nothing to probe — but the row's data is frozen from here on, so # it must not stay silent. See ``mark_orphaned_probe_targets``. "orphaned_targets": 0, "errors": [], } async with async_session_factory() as session: targets = await build_probe_targets(session) summary["targets"] = len(targets) if not targets: # No targets at all is a systemic condition (DNS sync broken / # every tenant disabled), NOT "nothing to do" — flagging every # certificate as orphaned here would be pure noise, so skip it. return {**summary, "status": "ok", "reason": "no probe targets"} sem = _asyncio.Semaphore(concurrency) async def _probe(t: dict): async with sem: return t, await probe_certificate( t["host"], t["port"], timeout, ) results = await _asyncio.gather(*[_probe(t) for t in targets]) # Upserts run sequentially on the single session (probing already # happened concurrently above without touching the session). # Each upsert is isolated in a SAVEPOINT so one unexpected flush # failure (e.g. a hostile cert field that slips past clamping) # cannot poison the shared session and lose the whole batch at the # final commit — the per-process-isolation invariant. for t, probed in results: try: async with session.begin_nested(): _row, action = await upsert_probed_certificate( session, probed, environment=t["environment"], ) summary[action] = summary.get(action, 0) + 1 except Exception as exc: # noqa: BLE001 logger.exception( "cert_probe_sweep: upsert failed for %s:%s; skipping", probed.host, probed.port, ) summary["skipped"] = summary.get("skipped", 0) + 1 summary["errors"].append({ "host": t["host"], "error": format_exception_for_log(exc), }) # Certificates whose probe host is no longer a target: the DNS record # was disabled or removed, so the sweep above never reached them and # their stored data is frozen at whatever the last probe wrote. summary["orphaned_targets"] = await mark_orphaned_probe_targets( session, {t["host"] for t in targets}, ) await session.commit() # Raise on systemic failure: every target upsert failed and nothing # advanced. Pre-fix an all-fail run (e.g. a poisonous cert field that # trips every flush) folded into ``skipped`` and returned normally — # the run wrote ``status='ok'`` and the dedup-aware failure-notifier # never fired. ``unreachable`` is a SUCCESSFUL probe outcome (the host # was reached or correctly classified and the row upserted), so it # counts as progress alongside created / updated. raise_if_total_failure( job_key="cmdb.cert_probe_sweep", summary=summary, progress_keys=("created", "updated", "unreachable"), ) return summary # ── Certificate expiry alert ────────────────────────────────────── @register_scheduled_job( key="cmdb.cert_expiry_alert", display_name="证书到期提醒", description=( "TLS 证书到期的阶梯式提醒 + 临期升级,委托给统一到期告警引擎" "(T-30 / T-14 各一次,最后 7 天起每天,过期后每天)。最后一公里" "升级:除本人外还推送给域名负责人 + 本人 lead。本人可一键「标记续期中」" "暂停提醒。owner 实时解析自 DNS 记录/域名。environment=prod 的生产证书" "不在提醒范围(生产续期由 Helix 自助流程之外的团队负责)。受 " "network.user_notifications_enabled 开关控制。" ), ) async def run_cert_expiry_alert( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.core.database import async_session_factory from app.domains.cmdb.expiry_alert_engine import ( build_cert_expiry_items, run_expiry_alerts, ) from app.domains.cmdb.models import EXPIRY_RESOURCE_CERTIFICATE from app.domains.cmdb.network_notifications import ( network_user_notifications_enabled, ) if not await network_user_notifications_enabled(db): return { "status": "skipped", "reason": "network.user_notifications_enabled is false", } async with async_session_factory() as session: items = await build_cert_expiry_items(session) notify = _make_expiry_notifier( session, kind_label_zh="证书", kind_label_en="Certificate", kind_label_ja="証明書", ) summary = await run_expiry_alerts( session, resource_type=EXPIRY_RESOURCE_CERTIFICATE, items=items, config=config, notify=notify, ) await session.commit() if summary["owners_notified"] == 0 and summary["send_failures"] > 0: raise RuntimeError( "cmdb.cert_expiry_alert: every notification failed " f"(failures={summary['send_failures']}). Check Feishu credentials." ) return summary # ── Hire-day IT info push ───────────────────────────────────────── @register_scheduled_job( key="hr.hire_day_it_notify", display_name="入职当天 IT 信息推送", description=( "在员工的 hire_date 当天给他推送 IT 基础信息(登录入口 / 已分配软件 / " "设备领取 / 2FA / IT 支持邮箱),Email + Feishu 双通道," "通过 identity_persons.it_info_sent_at 幂等。" ), ) async def run_hire_day_it_notify( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.core.database import async_session_factory from app.hr_lifecycle.hire_day_service import HireDayNotifyService async with async_session_factory() as session: service = HireDayNotifyService(session) summary = await service.run_for_today() await session.commit() # feedback_handler_status_must_raise: a per-hire dispatch exception # accumulates in ``summary["errors"]`` but run_for_today returns # normally — surface it as runs.status='failed' so an admin is paged # rather than a new hire silently getting no Day-1 IT info. A hire # skipped for "no email" is NOT an error (separate counter). if summary["errors"]: raise RuntimeError( f"hr.hire_day_it_notify: {len(summary['errors'])} of " f"{summary['candidates']} hire(s) raised — " f"{summary['errors'][:3]}" ) return summary @register_scheduled_job( key="hr.hire_day_feishu_welcome", display_name="入职欢迎 · 飞书卡片补发", description=( "每 30 分钟扫最近 3 天已收到 Day-1 欢迎邮件、但飞书欢迎卡片" "还没送达的新员工;飞书账号一建好(identity-sync feishu_contact)" "就补发富卡片版第一天指南。幂等键 identity_persons." "it_info_sent_channels;3 天窗口自然放弃永远不建飞书号的人员。" ), ) async def run_hire_day_feishu_welcome( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.core.database import async_session_factory from app.hr_lifecycle.hire_day_service import HireDayNotifyService async with async_session_factory() as session: service = HireDayNotifyService(session) summary = await service.run_feishu_welcome_catchup() await session.commit() # feedback_handler_status_must_raise: surface per-person dispatch # exceptions as runs.status='failed'. "awaiting_account" is the # expected steady state (Feishu account not minted yet), never an # error. if summary["errors"]: raise RuntimeError( f"hr.hire_day_feishu_welcome: {len(summary['errors'])} of " f"{summary['candidates']} hire(s) raised — " f"{summary['errors'][:3]}" ) return summary @register_scheduled_job( key="hr.day90_survey_notify", display_name="入职 90 天 · 体验问卷推送", description=( "按员工本地日历,在入职满 90 天时推送入职体验问卷邀请" "(飞书卡片 + 邮件,双语);幂等键 identity_persons." "day90_survey_sent_at;90–97 天补发窗口兜底漏跑," "上线时不会打扰在职已久的员工。" ), ) async def run_day90_survey_notify( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.core.database import async_session_factory from app.hr_lifecycle.hire_day_service import HireDayNotifyService async with async_session_factory() as session: service = HireDayNotifyService(session) summary = await service.run_day90_survey() await session.commit() # feedback_handler_status_must_raise: a per-person dispatch failure # must land runs.status='failed', not a green tick. if summary["errors"]: raise RuntimeError( f"hr.day90_survey_notify: {len(summary['errors'])} of " f"{summary['candidates']} invite(s) raised — " f"{summary['errors'][:3]}" ) return summary @register_scheduled_job( key="hr.pending_leaver_notice", display_name="离职告知信推送(HRIS 审批完成即推)", description=( "SF 离职审批一批完就把 HR 的《离职流程指引》推给本人(飞书卡片 + 邮件," "三语)。判据:EmpEmployment.endDate 在未来 + 有 EmpEmploymentTermination " "离职原因行 + User.status 仍是 't' —— 等价于「审批已完成、尚未到最后" "工作日」。既有离职流程要等 SF 翻 status 才建,那已经是最后工作日当天" "或之后了,所以整个通知期几乎无触达 —— 生产 87 条流程里只有 6 位离职者在 LWD " "之前收到过任何离职域通知,其中只有 2 位真的 delivered。\n" "默认范围:中国员工 + 雇佣类型 1070 正式 / 1072 实习生;命中连接器 " "excluded_termination_reason_codes(人员分类转换 / 跨 DC 调动 / 兼岗结束)" "的不推。幂等键 identity_persons.offboarding_notice_sent_at + " "offboarding_notice_last_day,HR 改期才重推。\n" 'config:{"employee_type_codes":[...],"china_only":true,' '"lookahead_days":365,"dry_run":false}。' ), ) async def run_pending_leaver_notice( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.core.database import async_session_factory from app.hr_lifecycle.pending_leaver_notice_service import ( PendingLeaverNoticeService, ) async with async_session_factory() as session: summary = await PendingLeaverNoticeService(session).run(config) await session.commit() # feedback_handler_status_must_raise: a per-person dispatch failure — # or an SF instance we could not read at all — must land # runs.status='failed'. A silent green tick here means the notice # period passes with nobody noticing the letter never went out, which # is exactly the failure this job exists to end. instance_errors = [d for d in summary["instances"] if d.get("error")] if summary["errors"] or instance_errors: raise RuntimeError( f"hr.pending_leaver_notice: {len(summary['errors'])} of " f"{summary['candidates']} notice(s) raised, " f"{len(instance_errors)} HRIS instance(s) unreadable — " f"{(summary['errors'] + instance_errors)[:3]}" ) return summary @register_scheduled_job( key="hr.email_name_reconcile", display_name="邮箱名对账(HR 指定的邮箱名 vs 实际邮箱)", description=( "每天对账 identity_persons.hris_email_name(HR 在 SF " "PerPersonal.customString4 指定的邮箱名)与实际 work_email 的 " "local-part,只读 Helix 自己的两列,不打 SF / Google。\n" "① 尚未建 Google 号且不一致 —— 自动重跑 reserve_work_email_for," "把预告地址和 SF PerEmail 回写一起纠正。\n" "② 已建号且不一致 —— 建号后只能人工 rename,推卡给 global_admin。\n" "为什么存在:P10026(2026-08-31)HR 在建人 2 天后才填邮箱名,正确值此后" "每 4 小时被读到一次,却因为 reserve 的 dispatch 闸只认 create/reactivate " "而一次都没生效,一直到 Day-1 建号才被员工本人发现。主修复是让建号那一刻" "直接读 hris_email_name;这道 cron 不依赖任何一条具体路径正确,是兜底网。\n" 'config:{"dry_run":false,"lookback_days":180}。' ), ) async def run_email_name_reconcile( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.hr_lifecycle.email_name_reconcile_service import ( EmailNameReconcileService, ) summary = await EmailNameReconcileService(db).run(config) await db.commit() # feedback_handler_status_must_raise: a per-person re-reserve failure must # land runs.status='failed'. A silent green tick here would recreate the # exact blind spot this job exists to close — nobody checking the result. if summary["errors"]: raise RuntimeError( f"hr.email_name_reconcile: {len(summary['errors'])} of " f"{summary['scanned']} person(s) failed to re-reserve — " f"{summary['errors'][:3]}" ) return summary # ── Offboarding state machine advance (the data-freshness tick) ── @register_scheduled_job( key="hr.process_advance", slo=SLO_PROCESS_ADVANCE, display_name="离职流程自动推进", description=( "数据保鲜核心:对 termination_date <= today 且仍有 blocker 的离职人," "立即强制转让(manager → manager 的 manager → IT Admin 兜底人)," "再原子地 suspend_sso + status=left。每 15 分钟一次。" ), ) async def run_process_advance( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.hr_lifecycle.advance_service import ProcessAdvanceService async def _work(session: AsyncSession) -> dict[str, Any]: service = ProcessAdvanceService(session) return await service.run_tick( fallback_email=config.get("escalation_fallback_email"), ) # A lock-wait / deadlock / orphaned-savepoint inside the tick aborts # the whole transaction server-side and propagates out of run_tick # (the per-person/per-blocker savepoint catches re-raise it rather than # cascading into a ``RELEASE SAVEPOINT ... does not exist`` 1305). The # tick is idempotent — it re-scans live ``leaver_pending`` state — so # the canonical InnoDB response is to restart on a fresh session. The # helper owns the commit (after work returns, before any raise below). summary = await _run_in_fresh_session_with_lock_retry( _work, label="hr.process_advance", ) # Scheduled-job convention (feedback_handler_status_must_raise): a # genuine per-leaver exception accumulates in ``summary["errors"]`` # but run_tick returns normally, so without this the run lands # status='ok' and maybe_notify_failure never fires — a leaver whose # force-transfer/suspend raises every 15 min retains access silently. # A leaver merely held for grace / missing fallback is NOT an error # (separate counters + a compliance incident already raised), so only # ``errors`` trips the raise. Mirrors run_recover_orphaned_left. if summary["errors"]: raise RuntimeError( f"hr.process_advance: {len(summary['errors'])} of " f"{summary['processed']} leaver(s) raised — " f"{summary['errors'][:3]}" ) return summary @register_scheduled_job( key="hr.recover_orphaned_left", display_name="离职孤儿流程自愈", description=( "自愈不变量:已经 employment_status='left' 的人不应再有卡在 BLOCKED " "且含 pending blocker 的离职流程。run_tick 只选 leaver_pending," "一旦有人(如 flip-to-left 撞上并发 start_process 扫描的竞态、" "HRIS 直写、手工改库)在带 pending blocker 的情况下变成 left," "其 owner 绑定的基础设施(cloud_server / DNS / IAM / 邮件组 / 硬件)" "永远无人转让、流程永远卡死。本任务对这类孤儿跑 force-transfer" "(bypass_grace,人已离开宽限期无意义)转给交接人/manager 兜底并" "完成流程。幂等。建议 15 分钟一次。" ), ) async def run_recover_orphaned_left( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.hr_lifecycle.advance_service import ProcessAdvanceService async def _work(session: AsyncSession) -> dict[str, Any]: service = ProcessAdvanceService(session) return await service.recover_orphaned_left_processes( fallback_email=config.get("escalation_fallback_email"), ) # Same fresh-session lock-retry as hr.process_advance — this sweep runs # every 15 min on the SAME tables (offboarding processes / blockers / # asset transfers) and is the most likely deadlock counterparty. The # helper commits on success so orphans that DID heal in their per-person # savepoints are persisted even when a sibling orphan recorded a # (non-fatal) error; a transaction-aborting error re-runs the whole # idempotent sweep on a new session instead. summary = await _run_in_fresh_session_with_lock_retry( _work, label="hr.recover_orphaned_left", ) # Scheduled-job convention (feedback_handler_status_must_raise): an # UNEXPECTED per-person exception must surface as runs.status='failed' # so maybe_notify_failure fires — otherwise the run lands 'ok' and the # error hides in last_result_summary forever. NOTE: a person who # merely can't be auto-resolved (no fallback owner) is NOT an error — # _advance_single already raised a CRITICAL compliance incident for it # and it's counted under ``still_blocked``; only ``errors`` (genuine # exceptions) trip the raise. The committed summary is still attached # to the failed run so admins see ``recovered`` / ``still_blocked``. if summary["errors"]: raise RuntimeError( f"hr.recover_orphaned_left: {len(summary['errors'])} of " f"{summary['scanned']} orphan(s) raised — " f"{summary['errors'][:3]}" ) return summary # ── Terminal Google-account delete (离职号最终删除) ──────────────── @register_scheduled_job( key="hr.offboarding_google_account_delete", display_name="离职 Google 账号最终删除", description=( "2026-07-01 翻案后离职者 Google 账号在末日次日被自动挂起,但此前" "永久停留在挂起态。本任务是终态处置:对已 COMPLETED 的离职流程," "在 leaver_date + delete_lead_days(默认 2,员工本地时区)到达后" "硬删除已挂起的 Google 账号。Google 保留 ~20 天可 undelete 恢复窗口" "作为误删安全网。保留号(external_account_registry 服务/共享/外部)" "永不删除;返聘/活跃/未确认挂起的账号跳过。幂等键 " "session_enforcement.google_account_deletion(deleted/already_absent/" "retained 为终态)。每日一次。" ), ) async def run_offboarding_google_account_delete( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.domains.offboarding.service import ( DEFAULT_GOOGLE_DELETE_LEAD_DAYS, OffboardingService, ) trace_id = f"system:cron:offboarding.google_delete:{uuid.uuid4()}" lead = config.get("delete_lead_days") lead = int(lead) if lead is not None else DEFAULT_GOOGLE_DELETE_LEAD_DAYS # The sweep re-scans live COMPLETED state and re-checks every gate under a # row lock, so a fresh-session re-run after a transient lock/connection # drop is clean and idempotent — same wrapper as reconcile/advance. summary = await _run_in_fresh_session_with_lock_retry( lambda session: OffboardingService(session).sweep_google_account_deletions( trace_id=trace_id, delete_lead_days=lead, ), label="hr.offboarding_google_account_delete", ) # Handler-must-raise (feedback_handler_status_must_raise): a genuine vendor # delete failure accumulates in ``errors`` but the sweep returns normally, # so without this the run lands 'ok' and the failure notifier never fires — # a leaver whose users.delete 403s (SA scope gap) every day would sit # suspended-but-not-deleted silently. Retention skips / not-yet-due are NOT # errors. Successful deletes are already committed by the wrapper. if summary["errors"]: raise RuntimeError( f"hr.offboarding_google_account_delete: {len(summary['errors'])} " f"of {summary['scanned']} scanned failed — {summary['errors'][:3]}" ) return summary # ── Terminal Zoom-account delete (离职 Zoom 号最终删除) ───────────── @register_scheduled_job( key="hr.offboarding_zoom_account_delete", display_name="离职 Zoom 账号最终删除", description=( "离职者 Zoom 账号两段式终态处置(对标 Google):末日次日" "(leaver_date+1)自动停用,本任务在 leaver_date + delete_lead_days" "(默认 2,员工本地时区)到达后硬删除已停用的 Zoom 账号。" "⚠️ Zoom 删除是永久的、无 undelete 恢复窗口,停用→删除的间隔即" "唯一纠错窗口。保留号(external_account_registry 服务/共享/外部)" "永不删除;返聘/活跃/未确认停用的账号跳过。幂等键 " "session_enforcement.zoom_account_deletion(deleted/already_absent/" "retained 为终态)。每日一次。" ), ) async def run_offboarding_zoom_account_delete( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.domains.offboarding.service import ( DEFAULT_ZOOM_DELETE_LEAD_DAYS, OffboardingService, ) trace_id = f"system:cron:offboarding.zoom_delete:{uuid.uuid4()}" lead = config.get("delete_lead_days") lead = int(lead) if lead is not None else DEFAULT_ZOOM_DELETE_LEAD_DAYS # The sweep re-scans live COMPLETED state and re-checks every gate under a # row lock, so a fresh-session re-run after a transient lock/connection # drop is clean and idempotent — same wrapper as the Google delete job. # Annotate the local so the ``Any`` the retry wrapper returns is laundered # into a concrete dict (mypy no-any-return ratchet). summary: dict[str, Any] = await _run_in_fresh_session_with_lock_retry( lambda session: OffboardingService(session).sweep_zoom_account_deletions( trace_id=trace_id, delete_lead_days=lead, ), label="hr.offboarding_zoom_account_delete", ) # Handler-must-raise (feedback_handler_status_must_raise): a genuine vendor # delete failure accumulates in ``errors`` but the sweep returns normally, # so without this the run lands 'ok' and the failure notifier never fires. if summary["errors"]: raise RuntimeError( f"hr.offboarding_zoom_account_delete: {len(summary['errors'])} " f"of {summary['scanned']} scanned failed — {summary['errors'][:3]}" ) return summary # ── Deferred Feishu-account delete retry (离职飞书号延迟删除) ────────── @register_scheduled_job( key="hr.offboarding_feishu_deferred_delete", display_name="离职飞书账号延迟删除重试", description=( "离职飞书号终态删除的延迟重试。离职者若有在途的自提审批实例" "(回购模板第二步停在提交人本人节点),末日+1 的删号会把该节点" "永久卡死(Helix 补传兜底也需要提交人飞书账号存在),因此删除被" "推迟、账号先冻结(访问已切断)。本任务每日扫描 " "session_enforcement.feishu_account 处于 " "delete_deferred_open_approvals 状态的流程:审批已终结 → 执行删除;" "仍在途且未超窗 → 继续推迟;超过 defer_max_days(默认 30,员工本地" "时区)→ 视为废弃审批,照删并审计记录 deferral_expired。瞬时删除" "失败保持推迟态,次日自动重试。" ), ) async def run_offboarding_feishu_deferred_delete( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.domains.offboarding.service import ( FEISHU_DELETE_DEFER_MAX_DAYS, OffboardingService, ) trace_id = f"system:cron:offboarding.feishu_deferred_delete:{uuid.uuid4()}" days = config.get("defer_max_days") days = int(days) if days is not None else FEISHU_DELETE_DEFER_MAX_DAYS summary = await _run_in_fresh_session_with_lock_retry( lambda session: ( OffboardingService(session).sweep_deferred_feishu_account_deletions( trace_id=trace_id, defer_max_days=days, ) ), label="hr.offboarding_feishu_deferred_delete", ) # Handler-must-raise (feedback_handler_status_must_raise): a vendor delete # failure keeps the row deferred (retried next tick) but must still land # runs.status='failed' so the failure notifier surfaces a leaver whose # delete errors every day instead of hiding behind a green tick. if summary["errors"]: raise RuntimeError( f"hr.offboarding_feishu_deferred_delete: {len(summary['errors'])} " f"of {summary['scanned']} scanned failed — {summary['errors'][:3]}" ) return summary # ── Default-handover reconcile (离职交接人回填) ──────────────────── @register_scheduled_job( key="hr.offboarding_handover_reconcile", display_name="离职默认交接人回填", description=( "离职流程的「默认交接人」只在建流程那一刻从 HRIS 直属上级预填一次。" "若当时 manager 链还没解析出来(后续 backfill_manager_links 才补上)," "这个值就永远是空的,再也没人回头看 —— 生产上因此有 12 个已完成、" "2 个在途流程没有交接人,hice@pingcap.com 的 Google 号就是在这种状态" "下被删的。本任务每日重扫仍然「有意义」的流程(未完成的,以及虽已完成" "但终删还没落地的),用同一套解析链(上级 → 上级链 → 平台兜底人)回填;" "确实解析不出来的,给 IT 发一次性汇总告警(每个流程只发一次)," "而不是继续沉默。只填空值,不覆盖 IT 已经设好的人。每日一次。" ), ) async def run_offboarding_handover_reconcile( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.domains.offboarding.service import OffboardingService trace_id = f"system:cron:offboarding.handover_reconcile:{uuid.uuid4()}" limit = config.get("batch_limit") batch_limit = int(limit) if limit is not None else 500 # Same fresh-session + lock-retry wrapper as the sibling offboarding # sweeps: the reconcile re-reads live state and only ever fills a NULL, # so a re-run after a transient lock/connection drop is idempotent. summary: dict[str, Any] = await _run_in_fresh_session_with_lock_retry( lambda session: OffboardingService(session).reconcile_default_handovers( trace_id=trace_id, batch_limit=batch_limit, ), label="hr.offboarding_handover_reconcile", ) # Handler-must-raise (feedback_handler_status_must_raise): a row that # raises every day must not read as a green tick. Note ``unresolved`` is # NOT an error — it is the expected, alerted-on state for a leaver whose # whole management chain has left. if summary["errors"]: raise RuntimeError( f"hr.offboarding_handover_reconcile: {len(summary['errors'])} " f"of {summary['scanned']} scanned failed — {summary['errors'][:3]}" ) return summary # ── T-1 terminal-delete preview (离职删号预告) ────────────────────── @register_scheduled_job( key="hr.offboarding_delete_preview", display_name="离职删号 T-1 预告", description=( "两个终删任务(03:00 / 03:10 UTC)成功时不通知任何人 —— 一个账号" "被永久删除(Zoom)或进入 ~20 天恢复窗口(Google),全程没有人有机会" "说「等一下」。失败通知器只在出错时响,也就是说**正确**的那条路径是" "静默的,这对不可逆操作正好反了。本任务在删除前一天给 IT 发一条汇总:" "列出明天要删的账号,以及每个账号的交接是否真的到位。并写明 API 修不了" "的那件事 —— Gmail 邮件内容不可转移(Data Transfer 只搬 Drive 和日历)," "要保留必须在删除前由人把邮箱标为保留。每个账号只预告一次(不是每天)," "被卡住的删除另有自己的一次性告警。每日一次。" ), ) async def run_offboarding_delete_preview( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.domains.offboarding.service import ( DEFAULT_GOOGLE_DELETE_LEAD_DAYS, DEFAULT_ZOOM_DELETE_LEAD_DAYS, OffboardingService, ) trace_id = f"system:cron:offboarding.delete_preview:{uuid.uuid4()}" g_lead = config.get("google_lead_days") z_lead = config.get("zoom_lead_days") summary: dict[str, Any] = await _run_in_fresh_session_with_lock_retry( lambda session: OffboardingService( session, ).preview_pending_terminal_deletes( trace_id=trace_id, google_lead_days=( int(g_lead) if g_lead is not None else DEFAULT_GOOGLE_DELETE_LEAD_DAYS ), zoom_lead_days=( int(z_lead) if z_lead is not None else DEFAULT_ZOOM_DELETE_LEAD_DAYS ), ), label="hr.offboarding_delete_preview", ) # Handler-must-raise: a row that blows up every day must not read as a # green tick — a warning job that silently stops warning is the same # failure mode this job exists to fix. if summary["errors"]: raise RuntimeError( f"hr.offboarding_delete_preview: {len(summary['errors'])} of " f"{summary['scanned']} scanned failed — {summary['errors'][:3]}" ) return summary # ── Post-leaver verification (T+1) ──────────────────────────────── @register_scheduled_job( key="hr.post_leaver_verify", display_name="离职后 T+1 资源复查", description=( "对最近 N 天前转为 left 状态的人重新扫描所有资源指向," "发现残留立即生成 critical compliance_incident 并推送 IT Admin。" ), ) async def run_post_leaver_verify( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.core.database import async_session_factory from app.hr_lifecycle.post_leaver_service import PostLeaverVerifyService recent_days = int(config.get("recent_days", 7)) async with async_session_factory() as session: service = PostLeaverVerifyService(session) summary = await service.run(recent_days=recent_days) await session.commit() return summary # ── Daily digest to IT Admin ────────────────────────────────────── @register_scheduled_job( key="hr.daily_digest", slo=SLO_DAILY_DIGEST, display_name="入离职每日摘要", description=( "每天 09:00 给 IT Admin 发 Email + Feishu:今日应离 / 3 日内入职 / " "当前未解 blocker / 昨日新 compliance incidents。" ), ) async def run_daily_digest( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.core.database import async_session_factory from app.hr_lifecycle.digest_service import DailyDigestService async with async_session_factory() as session: service = DailyDigestService(session) summary = await service.run() await session.commit() return summary # ── Notification outbox retry ───────────────────────────────────── @register_scheduled_job( key="notification.outbox_retry", slo=SLO_NOTIFICATION_OUTBOX_RETRY, display_name="通知失败重投", description=( "每小时扫 notification_outbox,对 status in (pending/partial/failed) " "且 next_attempt_at 已到期的行重试投递。" "超过 max_attempts 标记 dead,需要管理员手动处理。" ), ) async def run_outbox_retry( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.core.database import async_session_factory from app.notifications.service import NotificationService limit = int(config.get("limit", 100)) async with async_session_factory() as session: service = NotificationService(session) summary = await service.retry_pending_outbox(limit=limit) await session.commit() # Total-failure escalation. ``retry_pending_outbox`` has no # ``errors`` key (so ``raise_if_total_failure`` doesn't apply) — # rows picked, a real delivery was attempted, none landed, and at # least one permanently dead means the outbound channel (including # the one that delivers failure alerts itself) is down: must page. # An all-still_pending batch is transient backoff and will retry → no # raise. # # ``retried > 0`` is load-bearing: a departed/deleted-recipient row is # marked ``dead`` (+ ``skipped_left``) WITHOUT ever reaching # ``_deliver`` — a normal, healthy outcome right after an offboarding # wave. Only a genuine ``_deliver`` attempt increments ``retried``. # Without this gate a tick that drains only departed-recipient rows # (delivered=0, dead>0, retried=0) would false-page on a working # system. if ( summary.get("picked", 0) > 0 and summary.get("retried", 0) > 0 and summary.get("delivered", 0) == 0 and summary.get("dead", 0) > 0 ): raise RuntimeError( f"notification.outbox_retry total failure: picked={summary['picked']} " f"retried={summary['retried']} delivered=0 dead={summary['dead']} " f"still_pending={summary.get('still_pending', 0)}" ) return summary # ── CMDB resource sync (per integration × capability) ──────────── @register_scheduled_job( key="cmdb.resource_directory_refresh", weight="heavy", display_name="云资源目录索引刷新", description=( "每15分钟投影现有CMDB专用表，幂等记录新增、变化、消失与恢复；" "只读Helix数据库，不调用或修改Provider资源。" ), ) async def run_resource_directory_refresh( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.domains.cmdb.resource_directory import refresh_all try: retention = int(config.get("history_retention_days", 730)) except (TypeError, ValueError): retention = 730 try: batch_size = int(config.get("batch_size", 100)) except (TypeError, ValueError): batch_size = 100 return await refresh_all( db, history_retention_days=retention, batch_size=batch_size, ) @register_scheduled_job( key="cmdb.resource_inventory_refresh", weight="heavy", display_name="云资源广覆盖发现", description=( "每小时读取显式开启的AWS Resource Explorer、Azure Resources与GCP Cloud Asset；" "只读、不创建角色、不修改云资源，不完整扫描禁止消失清理。" ), ) async def run_resource_inventory_refresh( db: AsyncSession, config: dict[str, Any], # noqa: ARG001 ) -> dict[str, Any]: from app.domains.cmdb.resource_directory import refresh_inventories return await refresh_inventories(db) @register_scheduled_job( key="cmdb.resource_governance_evaluate", weight="heavy", display_name="云资源治理规则评估", description="每日评估统一目录治理规则并写入永久聚合趋势；只读云资源。", ) async def run_resource_governance_evaluate( db: AsyncSession, config: dict[str, Any], # noqa: ARG001 ) -> dict[str, Any]: from app.domains.cmdb.resource_governance import evaluate_and_rollup return await evaluate_and_rollup(db) @register_scheduled_job( key="cmdb.cost_facts_raw_purge", weight="light", display_name="云成本明细 raw 保留期清理", description=( "每日把超出保留期的 cmdb_cost_facts.raw(厂商原始账单行)置空；" "只清 raw 一列,金额与所有维度列不变,不删除任何行。" "按天推进并逐天提交 watermark,任何时刻被打断都不丢进度;" "``max_seconds_per_run``(默认 600s)是软预算,到点干净收工" "并在 summary 里标 stopped_early,下一跳从 watermark 继续。" ), ) async def run_cost_facts_raw_purge( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.domains.cmdb.cost_service import purge_cost_fact_raw def _int(key: str, default: int) -> int: try: return int(config.get(key, default)) except (TypeError, ValueError): return default # ``retention_days`` deliberately defaults to None so the DB-first # platform_setting wins; the cron config is only an override. retention_raw = config.get("retention_days") retention: int | None try: retention = int(retention_raw) if retention_raw is not None else None except (TypeError, ValueError): retention = None # Soft budget well under the scheduler's 1800s hard timeout. The # hard timeout is a kill that loses the run's summary; this is a # clean stop that keeps the watermark and reports why it stopped. return await purge_cost_fact_raw( db, retention_days=retention, batch_size=_int("batch_size", 2000), max_days_per_run=_int("max_days_per_run", 30), max_seconds_per_run=float(_int("max_seconds_per_run", 600)), ) @register_scheduled_job( key="cmdb.resource_report_dispatch", weight="light", display_name="云资源目录报表投递", description="每小时检查到期订阅，发送前重新计算权限范围并冻结报表；不修改云资源。", ) async def run_resource_report_dispatch( db: AsyncSession, config: dict[str, Any], # noqa: ARG001 ) -> dict[str, Any]: from app.domains.cmdb.resource_reporting import dispatch_due_reports return await dispatch_due_reports(db) @register_scheduled_job( key="cmdb.resource_sync", weight="heavy", # 多分钟云 API 扫描 — 见 registry.SchedJobWeight display_name="云资源同步", description=( "每个 (云账号 × 资源类型) 对应一条 scheduled_jobs 行,都挂到这个" "handler;config 里的 integration_id + capability 决定同步什么。" "替代 migration 046 之前的全局 cmdb.cloud_sync —— 粒度到能力级," "管理员可以单独暂停某个账号的某一类资源同步,不再是 all-or-nothing。" ), ) async def run_resource_sync( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.core.database import async_session_factory from app.domains.cmdb import sync as cmdb_sync from app.domains.cmdb.service import CmdbService from app.integrations.models import Integration integration_id = str(config.get("integration_id") or "").strip() capability = str(config.get("capability") or "").strip() if not integration_id or not capability: return { "status": "skipped", "reason": "config missing integration_id or capability", } async with async_session_factory() as session: integration = await session.get(Integration, uuid.UUID(integration_id)) if integration is None or integration.kind != "cloud": return { "status": "skipped", "reason": "integration missing or not kind=cloud", "integration_id": integration_id, } if not integration.enabled: return { "status": "skipped", "reason": "integration disabled", "integration_id": integration_id, "alias": integration.name, } alias = integration.name or integration_id capabilities = (integration.capabilities or []) if capabilities and capability not in capabilities: return { "status": "skipped", "reason": "capability not enabled on this integration", "integration_id": integration_id, "alias": alias, "capability": capability, } job = await CmdbService.create_sync_job( session, integration_id=integration_id, resource_types=[capability], triggered_by_user_id=None, triggered_by="scheduled", ) job_id = str(job.id) await cmdb_sync.run_sync( integration_id=integration_id, job_id=job_id, capabilities=[capability], triggered_by_user_id=None, ) async with async_session_factory() as session: from app.domains.cmdb.models import CmdbSyncJob refreshed = await session.get(CmdbSyncJob, uuid.UUID(job_id)) summary = { "integration_id": integration_id, "alias": alias, "capability": capability, "job_id": job_id, "status": refreshed.status if refreshed else "unknown", "upserted": refreshed.total_upserted if refreshed else 0, "deleted": refreshed.total_deleted if refreshed else 0, "errors": refreshed.total_errors if refreshed else 0, "capability_results": ( refreshed.capability_results if refreshed else None ), } # Surface capability-level failures at the scheduled_job layer too. # Without this the handler returns normally and the run row lands on # ``status=ok`` (显示"成功") even though the sync actually dropped # rows or errored on every upsert — misleading in the admin UI's # execution history. Raising maps the run to ``status=failed``, # bumps failure_count, and puts the capability error text on # ``last_error`` / ``run.error_message`` so admins can click through. # # 2026-05-17 hardening: the prior raise text was # ``capability 'object_storage' status=partial_failure`` — true # but useless. The actual leg breakdown was sitting in # ``cap_info['partial_failures']`` and the admin had to re-run a # SQL query against ``cmdb_sync_jobs.capability_results`` to see # WHICH legs failed. Summarise the leg breakdown into the raised # message so the failure-notifier and the admin UI's execution- # history row carry the actionable detail directly. inner_status = summary["status"] if inner_status in ("failed", "partial_failure"): cap_results = summary.get("capability_results") or {} cap_info = cap_results.get(capability) if isinstance(cap_results, dict) else None cap_error: str | None = None if isinstance(cap_info, dict): cap_error = cap_info.get("error") or cap_info.get("reason") legs = cap_info.get("partial_failures") if legs and isinstance(legs, list): from app.integrations.drivers._cloud_shared import ( _summarize_leg_failures, ) leg_summary = _summarize_leg_failures([ leg for leg in legs if isinstance(leg, dict) ]) if leg_summary: cap_error = ( f"{cap_error or 'partial_failure'} · " f"{len(legs)} leg(s): {leg_summary}" ) if not cap_error: cap_error = f"capability {capability!r} status={inner_status}" raise RuntimeError( f"cmdb.resource_sync {integration_id}/{capability}: {cap_error}" ) return summary # ── CMDB cloud efficiency (utilization + cost-optimization) ────────── @register_scheduled_job( key="cmdb.server_metrics_sync", weight="heavy", # 拉云监控 14 天窗口 — 多分钟 display_name="云利用率采集", description=( "【云成本优化·① 利用率采集】每个云账号一条,从云监控(CloudWatch / " "CloudMonitor / Azure Monitor / GCP Cloud Monitoring)拉取过去 14 天各" "实例 CPU/网络(内存/磁盘需 agent,尽力而为)小时级利用率,按日聚合" "写入 cmdb_server_metrics_daily(保留 90 天)。只读监控 API。" ), ) async def run_server_metrics_sync( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.core.database import async_session_factory from app.domains.cmdb.metrics_sync import run_metrics_sync from app.integrations.models import Integration integration_id = str(config.get("integration_id") or "").strip() if not integration_id: return {"status": "skipped", "reason": "config missing integration_id"} async with async_session_factory() as session: integration = await session.get(Integration, uuid.UUID(integration_id)) if integration is None or integration.kind != "cloud": return { "status": "skipped", "reason": "integration missing or not kind=cloud", "integration_id": integration_id, } if not integration.enabled: return { "status": "skipped", "reason": "integration disabled", "integration_id": integration_id, } return await run_metrics_sync(session, integration_id) @register_scheduled_job( key="cmdb.efficiency_compute", weight="heavy", display_name="云效率评估", description=( "【云成本优化·② 效率评估】每个云账号一条,用 cmdb_server_metrics_daily " "滚动窗口 + cmdb_cost_facts 单资源成本 + 规则引擎,重算每台机器的效率" "画像(闲置/超配/健康 + 建议 + 预估月省)写入 cmdb_server_efficiency。" "重算前先跑控制台侧成效检测(被标记机器在云控制台被关停/删除/降配" "→ 记入已优化成本,source=console)。排在利用率采集/成本明细/日账单" "之后。除记账外纯只读洞察。" ), ) async def run_efficiency_compute( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.core.database import async_session_factory from app.domains.cmdb.efficiency import compute_efficiency_for_integration from app.domains.cmdb.savings_detector import ( detect_console_savings_for_integration, ) from app.integrations.models import Integration integration_id = str(config.get("integration_id") or "").strip() if not integration_id: return {"status": "skipped", "reason": "config missing integration_id"} async with async_session_factory() as session: integration = await session.get(Integration, uuid.UUID(integration_id)) if integration is None or integration.kind != "cloud": return { "status": "skipped", "reason": "integration missing or not kind=cloud", "integration_id": integration_id, } if not integration.enabled: return { "status": "skipped", "reason": "integration disabled", "integration_id": integration_id, } # Console-side outcome detection MUST run before the recompute — # it compares the freshly-synced server state against yesterday's # verdict + spec snapshot, which the recompute overwrites. # Best-effort: a detector failure never blocks the compute. try: console_savings = await detect_console_savings_for_integration( session, integration_id, ) except Exception as exc: # noqa: BLE001 — observer, not the job logger.warning( "console-savings detection failed for %s: %s", integration_id, str(exc)[:200], ) await session.rollback() console_savings = {"status": "failed", "error": str(exc)[:200]} result = await compute_efficiency_for_integration( session, integration_id, ) from app.intelligence.savings_reconciliation import reconcile_recommendation_savings result["intelligence_recommendation_savings_links"] = ( await reconcile_recommendation_savings(session) ) result["console_savings"] = console_savings return result @register_scheduled_job( key="cmdb.efficiency_owner_digest", weight="light", display_name="云成本优化 · Owner 周提醒", description=( "【默认关,真开关在效率页「触达设置」的平台配置】每周把每位机器" "owner / 订阅业主名下待处理的优化建议聚合成一张飞书卡片 + 站内" "通知(绝不逐台轰炸;已 triage/暂缓的项不算)。发送前可在效率页" "预览收件人;支持灰度名单。配置存 platform_settings[" "'cloud_efficiency.owner_digest'],cron 行保持启用、配置关闭时" "静默跳过——单一开关,避免两处状态打架。" ), ) async def run_efficiency_owner_digest( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.core.database import async_session_factory from app.domains.cmdb.owner_digest import send_owner_digest async with async_session_factory() as session: return await send_owner_digest(session) @register_scheduled_job( key="cmdb.efficiency_digest", weight="light", display_name="云成本优化 · IT 周报摘要", description=( "【IT-only,默认禁用/opt-in】每周把云资源效率洞察的 Top 可省项聚合成" "一条飞书卡片,只发给 IT 管理员(global_admin/it_admin),从不发员工/" "经理。当周没有可优化项时静默不发(不打扰)。上线前请在 " "/admin/scheduled-jobs 评估存量触发面后手动启用。" ), ) async def run_efficiency_digest( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.core.database import async_session_factory from app.domains.cmdb import efficiency_service from app.notifications.service import ( CardStat, NotificationEnvelope, NotificationService, ) from app.scheduled_jobs.failure_notifier import list_admin_recipients if not config.get("enabled", True): return {"status": "skipped", "reason": "digest disabled via config"} async with async_session_factory() as session: summary = await efficiency_service.efficiency_summary(session, visibility=None) totals = summary["totals"] savings = totals.get("estimated_monthly_savings_usd") or 0 actionable = totals.get("with_recommendation") or 0 # Don't spam an empty week — the whole point is signal, not noise. if actionable <= 0 or savings <= 0: return {"status": "skipped", "reason": "no actionable savings this week"} top, _total = await efficiency_service.list_instances( session, visibility=None, sort="savings", page=1, page_size=5, ) recipients = await list_admin_recipients(session) if not recipients: return {"status": "skipped", "reason": "no IT-admin recipients"} verdict_counts = {r["verdict"]: r["count"] for r in summary["by_verdict"]} lines = [] def _clean(v: object, limit: int = 80) -> str: # Cloud resource names/tags are attacker-influenceable; strip # control + Feishu-markdown chars before they enter the card, # and bound the length. s = str(v or "") s = re.sub(r"[\x00-\x1f\x7f`*_~\[\]<>|]", " ", s).strip() return s[:limit] or "(unnamed)" for r in top: sv = r.get("estimated_monthly_savings_usd") name = _clean(r.get("name") or r.get("instance_id")) provider = _clean(r.get("provider"), 20).upper() lines.append( f"· {name} ({provider}) — " f"{_clean(r['verdict'], 24)} → {_clean(r['recommendation'], 24)}" f"{f',预估月省 ${sv:,.0f}' if sv else ''}" ) body_text = ( f"本周云资源效率洞察:共 {totals.get('servers', 0)} 台纳入评估," f"其中 {actionable} 台可优化,按建议执行预计每月可省 " f"${savings:,.0f}(已归属成本 ${totals.get('attributed_monthly_cost_usd') or 0:,.0f}/月)。\n\n" f"Top 可省项:\n" + ("\n".join(lines) if lines else "(无)") + "\n\n此为只读洞察,执行(关停/降配)请在云服务器详情页人工操作。" ) card_stats = [ CardStat( label=LocalizedText( zh="预估可省 / 月", en="Est. savings / month", ja="想定削減額 / 月", ), value=f"${savings:,.0f}", ), CardStat( label=LocalizedText( zh="可优化机器", en="Optimizable instances", ja="最適化可能なインスタンス", ), value=str(actionable), ), CardStat( label=LocalizedText(zh="闲置", en="Idle", ja="アイドル"), value=str(verdict_counts.get("idle", 0)), ), CardStat( label=LocalizedText( zh="超配", en="Oversized", ja="オーバースペック", ), value=str(verdict_counts.get("oversized", 0)), ), ] # Annotated: an unannotated literal infers a narrow value type, and # since ``card_subtitle`` became a union (str | LocalizedText | None) # the inferred type no longer satisfies the constructor. envelope_kwargs: dict[str, Any] = { "title": f"💡 云成本优化周报 — 预计每月可省 ${savings:,.0f}", "body_text": body_text, "channels": ["feishu", "email"], "card_subtitle": LocalizedText( zh="IT 云成本优化 · 每周汇总(仅 IT 管理员可见,员工/经理不收到)", en=( "IT cloud cost optimisation · weekly digest " "(IT admins only — not sent to employees or managers)" ), ja=( "IT クラウドコスト最適化 · 週次サマリー" "(IT 管理者のみ。社員・マネージャーには送信されません)" ), ), "card_stats": card_stats, "url": "/admin/cloud/efficiency", "related_resource_type": "cloud_efficiency", } trace_id = f"cmdb.efficiency_digest:{datetime.now(UTC).strftime('%Y%m%d')}" sent = 0 for r in recipients: try: svc = NotificationService(session) await svc.dispatch( NotificationEnvelope(recipient=r, **envelope_kwargs), trace_id=trace_id, actor_user_id=None, ) sent += 1 except Exception as exc: # noqa: BLE001 — best-effort per recipient logger.warning( "efficiency digest dispatch to %s failed: %s", r.work_email, str(exc)[:160], ) await session.commit() return { "status": "ok", "recipients": len(recipients), "dispatched": sent, "estimated_monthly_savings_usd": float(savings), "actionable": actionable, } # ── CMDB monthly billing sync ───────────────────────────────────── async def _aggregate_monthly_billing_from_daily( session: AsyncSession, integration_id: str, target_month: date, ) -> dict[str, Any]: """Build a monthly-billing dict by aggregating ``cmdb_cloud_billing_daily`` rows for the target month. Used as the AWS ``billing_source=bigquery`` substitute for the Cost Explorer call: those tenants intentionally don't grant ``ce:GetCostAndUsage`` so the legacy provider path 100%-fails with ``AccessDeniedException``. The daily-sync cron already pulls the same BQ CUR data at day×scope×service granularity — summing those rows for the month produces the same number CE would have returned, without the second BQ scan and without requiring CE permissions. Returns the same shape ``provider.get_monthly_billing`` does so the downstream upsert + budget-alert code is unchanged. """ from calendar import monthrange from decimal import Decimal as _Decimal from sqlalchemy import select as _select from app.domains.cmdb.models import CmdbCloudBillingDaily month_start = target_month.replace(day=1) _, last_day = monthrange(month_start.year, month_start.month) month_end = month_start.replace(day=last_day) rows = (await session.execute( _select( CmdbCloudBillingDaily.total_cost_usd, CmdbCloudBillingDaily.cost_by_service, CmdbCloudBillingDaily.currency, ).where( CmdbCloudBillingDaily.integration_id == integration_id, CmdbCloudBillingDaily.date >= month_start, CmdbCloudBillingDaily.date <= month_end, ) )).all() # Round-27 pre-launch (2026-05-25): the column is named # ``total_cost_usd`` but Aliyun / Tencent ``list_billing_daily`` # implementations document writing native CNY into this field with # ``currency='CNY'`` (the conversion happens at read time in # ``lifecycle_service._to_usd``). Today this helper is AWS-only — # gated upstream — but if it's ever wired to a non-USD provider # without converting, we'd silently land CNY totals tagged as USD. # Refuse to sum any row whose ``currency`` is not ``USD`` so a wiring # mistake surfaces as a loud failure, not a 7×-inflated total. total = _Decimal("0") by_service: dict[str, float] = {} seen_currencies: set[str] = set() skipped_non_usd_rows = 0 for r in rows: row_currency = (r.currency or "USD").upper() if row_currency != "USD": skipped_non_usd_rows += 1 seen_currencies.add(row_currency) continue total += _Decimal(str(r.total_cost_usd or 0)) for svc, v in (r.cost_by_service or {}).items(): by_service[svc] = float( _Decimal(str(by_service.get(svc, 0))) + _Decimal(str(v)), ) seen_currencies.add("USD") if skipped_non_usd_rows: # Hard fail rather than silent under-count: the caller expected # a per-month USD rollup and the daily rows carry non-USD # values. The cron handler converts this to a job-row error # the failure_notifier paages — wiring bug must be fixed. raise RuntimeError( f"_aggregate_monthly_billing_from_daily({integration_id}, " f"{month_start.isoformat()}): {skipped_non_usd_rows} daily " f"rows carry non-USD currency {sorted(seen_currencies)}; " "this helper only sums USD-denominated rows. The driver's " "``list_billing_daily`` MUST write USD into " "``total_cost_usd`` or the caller must route through " "a currency-aware aggregator.", ) # ``total_cost_usd`` is the normalised total — when the daily rows # mix currencies (rare; happens on a mid-month integration rotation) # the rollup is denominated in USD, not in whichever currency the # last row carried. The earlier "last-wins" branch misled # downstream consumers about denomination. if len(seen_currencies) > 1: logger.warning( "cmdb.billing_sync rollup %s saw heterogeneous currencies %s " "in daily rows — reporting in normalised USD.", integration_id, sorted(seen_currencies), ) currency = "USD" else: currency = next(iter(seen_currencies), "USD") return { "month": month_start, "total_cost_usd": float(total), "cost_by_service": by_service, "currency": currency, "raw_response": { "source": "aggregated_from_cmdb_cloud_billing_daily", "month_range": [month_start.isoformat(), month_end.isoformat()], "row_count": len(rows), }, } async def _fetch_monthly_billing_via_provider( integration: Any, target_month: date, ) -> dict[str, Any]: """Call ``provider.get_monthly_billing`` and return its dict. Raises ``RuntimeError`` on any failure (credential decryption, provider missing the method, vendor API error) so the scheduled-job runner records ``status=failed`` — the failure notifier then fires. Returning a ``{"status": "failed"}`` dict here would land in ``scheduled_job_runs.status='ok'`` with the truth buried in ``result_summary``, which is how aws-primary silently swallowed Cost Explorer AccessDenied for three months. """ from app.domains.cmdb.credential_crypto import decrypt_fields from app.domains.cmdb.providers.base import ( ProviderCredentials, get_provider_class, ) integration_id = str(integration.id) try: cred_fields = decrypt_fields( integration.secrets_encrypted, integration.secrets_nonce, integration_id=integration_id, ) if integration.secrets_encrypted else {} except Exception as exc: # noqa: BLE001 raise RuntimeError( f"cmdb.billing_sync {integration_id}: credential decryption failed: {exc}", ) from exc merged: dict[str, str] = {} if integration.config_public: merged.update( {k: str(v) for k, v in integration.config_public.items() if v is not None}, ) merged.update(cred_fields) try: provider_cls = get_provider_class(integration.provider) except ValueError as exc: raise RuntimeError( f"cmdb.billing_sync {integration_id}: unknown provider {integration.provider!r}: {exc}", ) from exc provider = provider_cls(ProviderCredentials( provider_id=integration_id, provider_type=integration.provider, account_alias=integration.name, fields=merged, )) if not hasattr(provider, "get_monthly_billing"): await provider.close() raise RuntimeError( f"cmdb.billing_sync {integration_id}: provider class " f"{type(provider).__name__} does not implement " "get_monthly_billing — billing pipeline regression", ) try: return await provider.get_monthly_billing(target_month) except Exception as exc: # noqa: BLE001 logger.exception("billing fetch failed for %s", integration_id) raise RuntimeError( f"cmdb.billing_sync {integration_id}: {format_exception_for_log(exc, max_len=400)}", ) from exc finally: await provider.close() @register_scheduled_job( key="cmdb.billing_sync", weight="heavy", # 多分钟云 API 扫描 — 见 registry.SchedJobWeight display_name="云账单月度抓取", description=( "每月 2 号凌晨从 AWS Cost Explorer / 同类 API 拉取上月账单," "写入 cmdb_billing_monthly。供 /admin/cloud/overview?tab=billing 多云聚合" "报表和 billing.budget_usd 平台参数的超预算 Feishu 告警用。" ), ) async def run_billing_sync( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from datetime import date as _date from datetime import timedelta as _td from decimal import Decimal as _Decimal from zoneinfo import ZoneInfo from sqlalchemy.dialects.mysql import insert as mysql_insert from app.core.database import async_session_factory from app.domains.cmdb.models import CmdbBillingMonthly from app.integrations.models import Integration integration_id = str(config.get("integration_id") or "").strip() if not integration_id: return {"status": "skipped", "reason": "missing integration_id"} # Target last month unless config overrides. Anchor "today" to # Asia/Shanghai (the scheduled_jobs row's timezone) instead of # process-local — otherwise a UTC-container deployment running the # cron at day 1 early hours would compute "last month" as two # months ago. explicit_month = config.get("month") if explicit_month: target_month = _date.fromisoformat(str(explicit_month)) else: today_sh = datetime.now(ZoneInfo("Asia/Shanghai")).date() first_of_this = today_sh.replace(day=1) target_month = (first_of_this - _td(days=1)).replace(day=1) async with async_session_factory() as session: integration = await session.get(Integration, uuid.UUID(integration_id)) if integration is None or integration.kind != "cloud" or not integration.enabled: return { "status": "skipped", "reason": "integration missing/disabled/wrong kind", "integration_id": integration_id, } # AWS billing_source=bigquery: the daily-by-scope rows already # came from CUR-in-BigQuery via ``cmdb.billing_daily_sync`` and # cover the target month at finer granularity. Aggregating from # ``cmdb_cloud_billing_daily`` is both the single source of # truth (same upstream feed) AND avoids calling Cost Explorer # — these BQ-mode tenants intentionally don't grant # ``ce:GetCostAndUsage`` (PingCAP prod aws-primary, 2026-05). # Falling through to provider.get_monthly_billing() would 100% # AccessDeniedException on those tenants. Other providers and # AWS-on-Cost-Explorer stay on the provider path. config_public = integration.config_public or {} if ( integration.provider == "aws" and (config_public.get("billing_source") or "").strip() == "bigquery" ): billing = await _aggregate_monthly_billing_from_daily( session, str(integration.id), target_month, ) else: billing = await _fetch_monthly_billing_via_provider( integration, target_month, ) # ``total_cost_usd is None`` signals the driver couldn't convert # the native currency to USD (unknown FX rate — see Aliyun / # AliCloud-Intl drivers). Log loud and skip the row rather than # silently writing 0 (which would then suppress every budget # alert for this tenant for the month). Native amount + currency # are still on the dict for any consumer that wants to display # them, but the USD-headline-stored row is gated. raw_usd = billing.get("total_cost_usd") if raw_usd is None: logger.warning( "billing_sync: integration %s returned total_cost_usd=None " "(currency=%r, native=%s); skipping monthly upsert + alert", integration_id, billing.get("currency"), billing.get("total_cost_native"), ) return { "integration_id": integration_id, "skipped": "unknown_currency", "currency": billing.get("currency"), "total_native": billing.get("total_cost_native"), } total = _Decimal(str(raw_usd or 0)) # Pre-launch audit Round-29 fix: for CNY-native billings (国内 # 阿里云 + 一部分 AliCloud Intl tenants billed 人民币), the # provider's monthly rollup uses a HARDCODED ``_FX_TO_USD["CNY"] # = 0.14`` table. That table drifts: the ECB-published rate moves # 3-5% over a quarter, so a stale 0.14 produces silently-wrong # USD headline numbers that finance later trusts. The DAILY # cost_facts pipeline already resolves CNY/USD dynamically via # ``app.core.fx_rate.resolve_cny_per_usd`` (per-integration # override → weekly-refreshed platform setting); apply the same # resolver here so the monthly rollup matches what dashboards # show. Failure to resolve raises — the driver-level fallback # would otherwise be silently inaccurate. Other currencies (USD # / EUR) keep the provider's value. billing_currency = (billing.get("currency") or "USD").upper() total_native = billing.get("total_cost_native") # Round-27 pre-launch (2026-05-25): track whether the rebase # actually landed so we write ``currency='USD'`` on the row only # when the value IS USD. Previously the column claimed CNY even # after a successful rebase — every downstream consumer that # honours the ``currency`` flag (MCP / CSV / future FE) # mis-rendered "¥USD-value" as if CNY. Schema invariant: the # numeric value and the ``currency`` label MUST agree on units. rebase_landed = False if billing_currency == "CNY" and total_native is not None: try: from app.core import fx_rate as _fx_rate cny_per_usd = await _fx_rate.resolve_cny_per_usd( session, integration_config=integration.config_public, ) # ``cny_per_usd`` is e.g. 7.20 (1 USD = 7.20 CNY); to # convert native CNY → USD divide by it. rebased = _Decimal(str(total_native)) / _Decimal(str(cny_per_usd)) # Persist the freshly converted value. Round to 2dp to # match the column scale and audit-displayable form. from decimal import ROUND_HALF_UP total = rebased.quantize(_Decimal("0.01"), rounding=ROUND_HALF_UP) rebase_landed = True logger.info( "billing_sync: rebased CNY-native total for %s " "(native=%s CNY, cny_per_usd=%s, rebased=%s USD; " "driver value was %s)", integration_id, total_native, cny_per_usd, total, raw_usd, ) except RuntimeError as exc: # Platform setting / override unavailable. Keep the # driver's stale-but-bounded value so the budget alert # still fires; log loud so the FX cron failure is # actioned. The daily cost_facts ingestion failure- # notifier will already have paged. logger.warning( "billing_sync: CNY rebase skipped for %s — %s; " "falling back to driver-supplied total_cost_usd=%s", integration_id, exc, raw_usd, ) # Effective row currency for persistence + return value: USD on # successful rebase OR when the driver already labelled USD; # otherwise the driver's native label (CNY/EUR/etc) survives so # downstream consumers can branch on it. effective_currency = ( "USD" if rebase_landed else (billing.get("currency") or "USD") ) month = billing.get("month") or target_month # Drivers MUST return ``month`` as a ``date`` per the # ``BillingProvider`` contract. Coerce defensively when a # driver returns a string ("YYYY-MM" or "YYYY-MM-DD") so the # downstream ``maybe_send_budget_alert`` ``f"{month:%Y-%m}"`` # format-spec doesn't blow up with TypeError, swallowed by the # outer try/except → silent missing alerts. Fail loud on a # totally bogus value. if isinstance(month, str): from datetime import date as _date_t try: if len(month) == 10: month = _date_t.fromisoformat(month) elif len(month) == 7: month = _date_t.fromisoformat(month + "-01") else: raise ValueError(f"unparseable month: {month!r}") except Exception as exc: # noqa: BLE001 raise RuntimeError( f"billing driver returned non-date month " f"({month!r}) for integration {integration_id}" ) from exc stmt = mysql_insert(CmdbBillingMonthly).values( id=str(uuid.uuid4()), integration_id=str(integration.id), month=month, total_cost_usd=total, cost_by_service=billing.get("cost_by_service") or {}, currency=effective_currency, raw_response=billing.get("raw_response") or {}, fetched_at=datetime.now(UTC), ).on_duplicate_key_update( total_cost_usd=total, cost_by_service=billing.get("cost_by_service") or {}, currency=effective_currency, raw_response=billing.get("raw_response") or {}, fetched_at=datetime.now(UTC), ) await session.execute(stmt) await session.commit() # Post-fetch budget alert (Feishu) — threshold lives in platform_settings. # Round-51 L2-F1: the previous bare-except meant a budget-alert # dispatch failure (Feishu down, dedup-table FK race, template # missing) landed ``scheduled_job_runs.status='ok'`` and failure_ # notifier never paged. Since this is the MONTHLY rollup, the next # chance to re-page is +1 month — operators were structurally # blind to over-budget alerts being silently dropped. Surface # the error in the handler's return summary AND raise so the # cron tick is marked failed, which then triggers the failure # notifier on the next sweep. # # Round-51 regression-review R51-A: ``maybe_send_budget_alert`` # returns a status dict for the documented failure modes # (``exceeded-no-webhook``, ``exceeded-webhook-rejected``, # ``exceeded-post-failed``) — it does NOT raise. The previous # try/except only caught pathological code-path failures and the # documented silent-drop modes the Round-21 audit found stayed # silent. Inspect the returned dict's ``status`` field and treat # the documented failure sentinels as raise-shaped. budget_alert_error: str | None = None budget_alert_skipped: str | None = None _BUDGET_ALERT_FAILURE_STATUSES = ( "exceeded-no-webhook", "exceeded-webhook-rejected", "exceeded-post-failed", ) # Round-666 hardening: only fire the budget alert when ``total`` is a # trustworthy USD figure. ``effective_currency`` is 'USD' iff the row # is genuinely USD-denominated — a native-USD provider (AWS/GCP/Azure # + the AWS-BQ aggregator, which hard-fails on non-USD rows) OR a CNY # tenant whose dynamic FX rebase SUCCEEDED above (``rebase_landed``). # When the CNY rebase FALLS BACK (FX cron stale/failed), ``total`` is # the native CNY amount and ``effective_currency`` stays 'CNY'. The # budget threshold is USD, so comparing a native CNY (~7×) amount to # it would fire a FALSE over-budget alert (疯狂告警). Skip the # alert: the row is still persisted unit-consistently for the # dashboard, the FX-cron ``failure_notifier`` already pages on the FX # gap, and the next monthly tick re-evaluates once FX is restored. if effective_currency != "USD": budget_alert_skipped = ( f"non-USD effective currency {effective_currency!r} " "(no trustworthy USD figure to compare against USD budget)" ) logger.info( "billing_sync: skipping budget alert for %s %s — %s", integration_id, month, budget_alert_skipped, ) else: try: from app.domains.cmdb.billing_alerts import maybe_send_budget_alert alert_result = await maybe_send_budget_alert( integration_id=integration_id, integration_name=integration.name or integration_id, month=month, total_usd=float(total), ) except Exception as exc: # noqa: BLE001 logger.exception("budget alert failed for %s", integration_id) budget_alert_error = ( f"budget alert dispatch failed: {type(exc).__name__}: {exc}" ) else: alert_status = ( alert_result.get("status") if isinstance(alert_result, dict) else None ) if alert_status in _BUDGET_ALERT_FAILURE_STATUSES: budget_alert_error = ( f"budget alert dispatch returned failure status: {alert_status}" ) summary = { "integration_id": integration_id, "alias": integration.name or integration_id, "month": str(month), "total_cost_usd": float(total), # Same effective-currency contract as the DB write — see # ``effective_currency`` above. "currency": effective_currency, "top_services": sorted( (billing.get("cost_by_service") or {}).items(), key=lambda kv: kv[1], reverse=True, )[:5], } if budget_alert_skipped is not None: # Visible in the run row (status stays ok — this is an intentional # skip, not a failure) so an operator can see the alert was # deliberately not fired this tick and why. summary["budget_alert_skipped"] = budget_alert_skipped if budget_alert_error is not None: summary["budget_alert_error"] = budget_alert_error # Raise AFTER assembling the summary so the run row carries # the integration / month / total context the operator needs # to identify which alert was dropped. raise RuntimeError( f"cmdb.billing_sync {integration_id} {month}: " f"data ingested OK but {budget_alert_error}" ) return summary # ── Provisioning dispatch ───────────────────────────────────────── @register_scheduled_job( key="provisioning.dispatch", display_name="Provisioning 任务派发", description=( "每 30s 扫 provisioning_tasks,对 status=pending 的任务分发到" "对应驱动(Zoom / Slack / GitHub / Feishu / Google Workspace 等)。" "已 in_progress 的任务由原 tick 负责完成;超过 max_attempts 的" "任务 status=failed 后等待管理员 retry。这个 handler 是审批通过" "→ 自动开通、离职 auto_revoke → 自动回收的执行引擎。" ), ) async def run_provisioning_dispatch( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.core.database import async_session_factory from app.domains.provisioning.service import ProvisioningService limit = int(config.get("limit", 50)) async with async_session_factory() as session: service = ProvisioningService(session) summary = await service.execute_pending_batch(limit=limit) # ``execute_pending_batch`` now commits per-task (Round 1 fix # LG-004). The trailing commit here is a no-op when no task ran # but covers the unlikely case that a future change leaves # session state pending. Wrapped because a commit on an empty # session shouldn't raise — but the worker should never crash # the cron because of it. try: await session.commit() except Exception: # noqa: BLE001 await session.rollback() # Total-failure escalation: ``execute_task`` catches connector # errors and sets task.status=failed WITHOUT raising, so a full # vendor outage makes every task fail while the run still reports # ok and the dedup-aware failure-notifier never fires. Raise when # every task failed and nothing made progress. A ``transient_retried`` # task is legitimately in-flight (will be re-picked), not failed — # so it counts as progress alongside ``completed`` / ``notified``. raise_if_total_failure( job_key="provisioning.dispatch", summary=summary, progress_keys=("completed", "notified", "transient_retried"), ) return summary # ── Cloud account lifecycle (PR 2026-04-23) ─────────────────────── # How long ``scopes_sync`` / ``roles_sync`` / ``grants_sync`` wait for the # shared ``cloud_sync:{integration_id}`` lock before declaring the run # starved. All three serialize on ONE key per tenant but run as three # independent crons minutes apart, so the slowest member routinely still # holds the lock when the next one fires. # # 2026-08-06 production finding: ``cmdb.roles.aws-primary`` (04:09, ~881 s) # was still holding the lock when ``cmdb.scopes.aws-primary`` fired at # 04:18. Scope sync therefore no-opped on 29 of its last 31 runs — and # because the old code RETURNED ``status="skipped"`` (which the runner # records as ``ok``) neither the failure notifier nor the admin UI ever # showed a problem. AWS sub-account ``417183877957`` created in July only # became visible in Helix on 2026-08-03, when one cron happened to win the # race. 900 s comfortably covers the observed 15-minute roles run. _CLOUD_SYNC_LOCK_WAIT_SEC = 900.0 def raise_cloud_sync_lock_starved(job_key: str, integration_id: str) -> None: """Fail the run when the shared cloud-sync lock never came free. Deliberately RAISES rather than returning ``{"status": "skipped"}``: a returned dict lands ``scheduled_job_runs.status='ok'`` (``feedback_handler_status_must_raise``), which is precisely what hid the 29-day scope-sync outage. A sync that did not run is a failure — the whole point is that somebody gets paged instead of the job reporting success while doing nothing. The integration id is hyphen-stripped so ``compute_failure_signature`` doesn't ``_UUID_RE``-collapse two different integrations' failures into one dedup signature (same treatment the inner-failure raise below uses). """ raise RuntimeError( f"{job_key} {integration_id.replace('-', '')}: shared cloud-sync " f"lock still held after waiting {_CLOUD_SYNC_LOCK_WAIT_SEC:.0f}s — " "this tenant's sync did NOT run. A sibling cloud sync " "(scopes/roles/grants) for the same integration is overrunning its " "cron gap; re-space the crons or raise the wait." ) @register_scheduled_job( key="cmdb.scopes_sync", weight="heavy", # 多分钟云 API 扫描 — 见 registry.SchedJobWeight display_name="云子账号/订阅同步", description=( "每日 04:00 抓取 AWS Organizations / Azure Subscription / " "GCP Project / 阿里云资源目录下的全部子账号,写入 cmdb_cloud_scopes。" "Owner 为空的新 scope 会产生 compliance_incident;" "已关闭的 scope 置 status=closed(不删,保留账单历史)。" ), ) async def run_scopes_sync( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.core.distributed_lock import LockBusy, distributed_lock from app.domains.cmdb.lifecycle_service import sync_scopes_for_integration integration_id = str(config.get("integration_id") or "").strip() if not integration_id: return {"status": "skipped", "reason": "missing integration_id"} # Same key + ttl as the admin "sync now" path + the grants_sync # handler so cron and manual invocations serialize through one # lock. Without this, an admin "sync now" running concurrently # with this cron would stamp scope rows with two different # ``last_seen_sync_job_id`` values; the first-to-finish sweep # then closes scopes the slower run had touched — silent # mass-close of legitimate scopes (same hazard ``run_grants_sync`` # was protected against in 2026-04). # # ``acquire_timeout`` — see ``_CLOUD_SYNC_LOCK_WAIT_SEC``. This job # is the one that got starved in production: it shares the lock with # ``roles_sync`` (~15 min on aws-primary) but is scheduled only 9 min # after it, so it lost the lock every single day. try: async with distributed_lock( f"cloud_sync:{integration_id}", ttl=600, acquire_timeout=_CLOUD_SYNC_LOCK_WAIT_SEC, ): result = await sync_scopes_for_integration(integration_id) except LockBusy: raise_cloud_sync_lock_starved("cmdb.scopes_sync", integration_id) # Pre-launch hardening 2026-05-16 (CR4 P0): raise on inner # ``status=failed`` so ``scheduled_job_runs.status`` lands ``failed`` # and ``failure_notifier`` actually pages. Returning the dict # silently lands ``ok`` per ``feedback_handler_status_must_raise``. # Round-3 review follow-up (LS3 P2-C): include the integration id # in a non-UUID-shaped form (hyphens stripped) so # ``compute_failure_signature`` doesn't ``_UUID_RE``-collapse two # different integrations' failures into the same dedup signature. # Without this, integration A failing at tick N and integration B # failing at tick N+1 share a signature and the second page gets # suppressed. int_sig = (integration_id or "").replace("-", "") if result.get("status") == "failed": raise RuntimeError( f"cmdb.scopes_sync int_sig={int_sig} integration={integration_id} " f"failed: {result.get('error') or result}" ) return result @register_scheduled_job( key="cmdb.roles_sync", weight="heavy", # 多分钟云 API 扫描 — 见 registry.SchedJobWeight display_name="云角色目录同步", description=( "每日 04:30 抓取本 tenant 下每个 scope 的 role 目录(预定义 + 自定义)," "写入 cmdb_cloud_roles。申请 flow 的 role 下拉直接读这张表," "'刷新角色'按钮可以立即触发同一逻辑。" ), ) async def run_roles_sync( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.core.distributed_lock import LockBusy, distributed_lock from app.domains.cmdb.lifecycle_service import sync_roles_for_integration integration_id = str(config.get("integration_id") or "").strip() if not integration_id: return {"status": "skipped", "reason": "missing integration_id"} # Mirror ``run_grants_sync`` / ``run_scopes_sync`` — the # mark-and-sweep on ``cmdb_cloud_roles.last_seen_sync_job_id`` # is the same hazard. Without this lock, an admin "sync now" + # the post-provision per-scope pre-warm (sync_roles_for_scope, # 2026-05-13) can each stamp role rows with their own job_id; # the slower-to-sweep instance closes rows the other had just # inserted. Catalog goes empty for the affected scope until the # next clean cron run. try: async with distributed_lock( f"cloud_sync:{integration_id}", ttl=600, acquire_timeout=_CLOUD_SYNC_LOCK_WAIT_SEC, ): result = await sync_roles_for_integration(integration_id) except LockBusy: raise_cloud_sync_lock_starved("cmdb.roles_sync", integration_id) # Pre-launch hardening 2026-05-16 (CR4 P0) + Round-1 review (Review-D # P1-1): raise on inner ``status=failed``. For ``errors[]`` only # raise when EVERY scope failed (``upserted == 0``) — partial errors # (e.g. one flaky scope out of 50) are visible in the result summary # and the sweep guard at ``lifecycle_service.sync_roles_for_integration`` # (tolerance from ``platform_settings['cmdb_sync.sweep_drop_tolerance_percent']``, # default 20%) already catches systemic loss. Without this # threshold, a single rate-limited scope would page failure_notifier # every cron tick and SRE would mute the channel within a week. # Round-3 review follow-up (LS3 P2-C): non-UUID-shaped integration # id in the raise message so dedup signature differentiates across # integrations (see ``run_scopes_sync`` for the explanation). int_sig = (integration_id or "").replace("-", "") if result.get("status") == "failed": raise RuntimeError( f"cmdb.roles_sync int_sig={int_sig} integration={integration_id} " f"failed: " f"{result.get('error') or result.get('reason') or result}" ) errors = result.get("errors") or [] upserted = int(result.get("upserted") or 0) if errors and upserted == 0: detail = "; ".join(str(e) for e in errors)[:500] or "" raise RuntimeError( f"cmdb.roles_sync int_sig={int_sig} integration={integration_id} " f"fully failed (0 upserted, {len(errors)} errors): {detail}" ) return result @register_scheduled_job( key="cmdb.grants_sync", weight="heavy", # 多分钟云 API 扫描 — 见 registry.SchedJobWeight display_name="云端权限反向同步", description=( "每日 05:00 反向拉取云侧已有的 role/policy 绑定 —— " "Azure roleAssignments、GCP projects.getIamPolicy、" "AWS Identity Center list_account_assignments 或 IAM " "list_attached_user_policies、阿里云 RAM ListPoliciesForUser、" "腾讯云 CAM ListAttachedUserAllPolicies —— 把员工邮箱已经" "持有的权限灌回 cmdb_cloud_access_grants(source='discovered')," "避免 Helix 外带赋权看不见。scope 无 owner 且只有一人持有 " "owner-等价角色(Azure Owner / GCP roles/owner / AWS " "AdministratorAccess / 阿里云/腾讯云 AdministratorAccess)" "时,自动把该人填为 scope 主。" ), ) async def run_grants_sync( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.core.distributed_lock import LockBusy, distributed_lock from app.domains.cmdb.lifecycle_service import sync_grants_for_integration integration_id = str(config.get("integration_id") or "").strip() if not integration_id: return {"status": "skipped", "reason": "missing integration_id"} # Same key + ttl as the admin "sync now" path so cron and manual # invocations serialize through one lock. Without this, an admin # clicking sync-now while the cron is running (or vice-versa) # creates two concurrent runs that each stamp grant rows with # their own ``sync_job_id``; the first run's drift mark-and-sweep # then closes every grant the second run had touched because # ``last_seen_grants_sync_job_id`` no longer matches the first # run's id — silent mass-revoke of legitimate access. try: async with distributed_lock( f"cloud_sync:{integration_id}", ttl=600, acquire_timeout=_CLOUD_SYNC_LOCK_WAIT_SEC, ): result = await sync_grants_for_integration(integration_id) except LockBusy: raise_cloud_sync_lock_starved("cmdb.grants_sync", integration_id) # Surface partial failures AND full failures at the scheduled-job # layer (same convention scopes_sync / roles_sync / cost_facts_sync # adopted; resource_sync 2026-04-24). Both shapes must raise so # ``scheduled_job_runs.status='failed'`` lands and the # failure-notifier fires. int_sig form keeps dedup signatures # distinct per integration (matches scopes_sync / roles_sync). int_sig = integration_id.replace("-", "") inner_status = result.get("status") if inner_status in ("failed", "partial_failure"): errs = result.get("errors") or [] reason = result.get("reason") detail = "; ".join(errs)[:500] or reason or "" raise RuntimeError( f"cmdb.grants_sync int_sig={int_sig} {inner_status}: {detail}" ) return result @register_scheduled_job( key="cmdb.auto_owner_grant_backfill", display_name="云账号 Owner 自动授权补撤", description=( "每 10 分钟扫 Helix 自助开通(cmdb_cloud_scope_requests 溯源)且 " "active 有 owner 的 scope,若 owner 既没有 source='helix' ACTIVE " "grant、也没有任何来源的 owner 等价角色 ACTIVE grant,重跑 " "_post_provision_finalize 把云端 IAM + Helix grant 行都补上。" "同步导入的存量 scope 从不在可修集里(owner 多半早已在云侧持有 " "IAM,provision_initial_owner_grant 会撞 ConflictError 静默 " "no-op)——2026-07-03 prod 曾有 98 个存量候选把 batch_cap=20 永久" "打满,真候选被彻底挤出。" "兜底 PR #279 的 auto-Owner 流 —— 该流原本走 FastAPI " "BackgroundTask,2026-06-02 调查发现 prod uvicorn 下静默丢任务," "audit 表里 cloud.scope_request.auto_owner_grant 从 PR #279 上" "线起一条没有。inline 化是主修,该 cron 是第二条腰带,捕获" "inline 路径自身崩了(LockBusy、Identity Center 抖动)的 case。" "provision_initial_owner_grant 幂等(ConflictError 视为 no-op)," "重跑安全。" ), ) async def run_auto_owner_grant_backfill( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.core.database import async_session_factory from app.domains.cmdb.lifecycle_router import ( _AWS_DEFAULT_OWNER_ROLE_NAME_FALLBACK, _AWS_DEFAULT_OWNER_ROLE_NAME_SETTING, _DEFAULT_OWNER_ROLE_BY_PROVIDER, _OWNER_ROLE_UNRESOLVABLE_RETRY_HOURS, _post_provision_finalize, ) from app.domains.cmdb.models import ( GRANT_STATUS_ACTIVE, PROD_ENVIRONMENTS, SCOPE_STATUS_ACTIVE, SCOPE_TYPE_GCP_BILLING_ACCOUNT, CmdbCloudAccessGrant, CmdbCloudRole, CmdbCloudScope, CmdbCloudScopeRequest, ) from app.identity.person_models import IdentityPerson from app.integrations.models import Integration from app.integrations.service import get_platform_setting # Cap how many scopes we touch per tick — if backfill is genuinely # broken across hundreds of scopes (e.g. day-after-launch), we don't # want one tick to hold AWS Identity Center for 30 minutes. Next # tick picks up the remainder. Default 20 — comfortably above # "weekly self-service ask volume" while bounding wall-clock. batch_cap = max(1, min(200, int(config.get("batch_cap") or 20))) # Only providers with an auto-Owner-grant policy can ever be fixed # by this cron. Without this gate, a provider with no owner-role # entry (Azure until 2026-08-17; ``ksyun`` still, by design) matches # the candidate predicate forever — ``_resolve_default_owner_role`` # returns ``None`` and ``_post_provision_finalize`` audit-logs # ``cloud.scope_request.auto_owner_grant`` ``status=skipped`` every # 10 min on the same scope. Same noise/cost class as the leaver + # PROD_ENVIRONMENTS exclusions above; keep the candidate set # symmetric with what the finalize code path can actually act on. supported_providers = [p for p in _DEFAULT_OWNER_ROLE_BY_PROVIDER] # Scopes whose auto-Owner permission-set is DETERMINISTICALLY # unresolvable (AWS Identity Center has no perm-set matching the # configured name) get ``auto_owner_grant_unresolvable_at`` stamped by # ``_post_provision_finalize``. Without the marker such a scope is a # candidate FOREVER — no ``source='helix'`` grant ever lands, so # ``~sub_helix_owner_grant`` stays true and we re-run the paid # ``list_scope_roles`` call + write a skipped-audit row on every # 10-min tick. Skip scopes stamped within the retry window; after it # they re-qualify so an admin who fixed the perm-set name recovers # automatically (and a successful resolve clears the marker outright # via ``provision_initial_owner_grant``). Configurable via # ``unresolvable_retry_hours``; defaults to the router constant. retry_hours = max( 1, int(config.get("unresolvable_retry_hours") or _OWNER_ROLE_UNRESOLVABLE_RETRY_HOURS), ) unresolvable_stale_cutoff = datetime.now(UTC) - timedelta( hours=retry_hours, ) async with async_session_factory() as session: # Effective AWS owner permission-set NAME (overridable via # platform_settings). Both the helix rows this cron writes and # the discovered rows grants_sync writes carry the perm-set # name in ``role_display_name`` (``role_key`` is the runtime # ARN we can't know here), so the owner-equivalent exclusion # below matches AWS by display name. aws_owner_role_name = str( await get_platform_setting( _AWS_DEFAULT_OWNER_ROLE_NAME_SETTING, default=_AWS_DEFAULT_OWNER_ROLE_NAME_FALLBACK, session=session, ) or "", ).strip() or _AWS_DEFAULT_OWNER_ROLE_NAME_FALLBACK # Active scopes with owner set, where the owner does NOT # currently hold any source='helix' active/pending grant on # this scope. NOT EXISTS keeps the LEFT JOIN deduplication # simple even when an owner has many discovered rows. # ACTIVE-only (review-sweep #2/#7): a PENDING source='helix' row # means the driver bind failed (``_dispatch_grant`` leaves it # PENDING with detail.last_error — NO cloud-side access). Excluding # PENDING here would strand exactly the failure this cron exists to # recover: the owner has a bookkeeping row but no IAM, and the scope # never re-enters the batch. Excluding only ACTIVE lets the # stuck-PENDING scope re-enter; ``provision_initial_owner_grant`` # re-dispatches the existing PENDING row instead of ConflictError. sub_helix_owner_grant = ( select(CmdbCloudAccessGrant.id) .where( CmdbCloudAccessGrant.scope_id == CmdbCloudScope.id, CmdbCloudAccessGrant.person_id == CmdbCloudScope.owner_person_id, CmdbCloudAccessGrant.source == "helix", CmdbCloudAccessGrant.status == GRANT_STATUS_ACTIVE, ) .correlate(CmdbCloudScope) .exists() ) # Leaver-owned scopes match the candidate predicate forever: # ``provision_initial_owner_grant`` has a leaver gate # (lifecycle_service.py:2473) that returns ``None`` for owners # in ``left`` / ``leaver_pending``, so no grant row is ever # written — the scope re-enters every 10-min batch indefinitely, # burns AWS Identity Center API budget on guaranteed no-ops, # and consumes ``batch_cap`` slots while reporting `fixed=0`. # Exclude here so the candidate set matches the gate (symmetric # with the prod-environment exclusion above). The leaver gate # in ``provision_initial_owner_grant`` stays as defence-in-depth # — admins should run ``assign_scope_owner`` for orphaned scopes. sub_owner_is_leaver = ( select(IdentityPerson.id) .where( IdentityPerson.id == CmdbCloudScope.owner_person_id, IdentityPerson.employment_status.in_( ("left", "leaver_pending"), ), ) .correlate(CmdbCloudScope) .exists() ) # Only scopes the PR #279 self-service flow provisioned are in # this cron's repair mandate — that's the flow it exists to # backstop. Sync-imported inventory (the vast majority of # cmdb_cloud_scopes) was assigned owners during CMDB migration; # those owners typically ALREADY hold cloud-side IAM that # grants_sync mirrors as source='discovered' rows, so # ``provision_initial_owner_grant`` hits ConflictError on the # same (scope, person, role) triple and no-ops — no grant, no # audit, no backoff marker. Prod 2026-07-03: 98 such imported # scopes matched the old predicate, the oldest 20 permanently # saturated batch_cap (created_at ASC is deterministic), and # every genuinely-broken newer scope was starved out of the # batch forever. Provenance = the scope-request row the # provision path stamps (``created_scope_id``). sub_provisioned_by_helix = ( select(CmdbCloudScopeRequest.id) .where( CmdbCloudScopeRequest.created_scope_id == CmdbCloudScope.id, ) .correlate(CmdbCloudScope) .exists() ) # Belt for the residual ConflictError-stall class INSIDE the # self-service set: an owner who acquired the owner-equivalent # role out-of-band (cloud console) gets a source='discovered' # ACTIVE row from grants_sync; the helix-source predicate above # can't see it, and the finalize path ConflictErrors on the # exact triple every tick — silently, forever. An ACTIVE # owner-equivalent grant of ANY source means the cloud side is # already right; nothing for this cron to repair. Static # providers match on role_key (single source of truth: # ``_DEFAULT_OWNER_ROLE_BY_PROVIDER``). AWS role_key is the # runtime permission-set ARN we can't know in SQL, so match by # NAME two ways: the grant row's own ``role_display_name`` # (helix rows the cron writes carry the perm-set name) OR the # role-catalog translation of the grant's role_key — prod # discovered SSO grant rows carry ``role_display_name = NULL``, # so the ``cmdb_cloud_roles`` lookup (ARN → perm-set name, # populated by roles_sync) is the one that actually matches # them (verified on aws-pingcap-datalake, 2026-07-03: the # owner's discovered row for ps-e4c0638798671e00 == the # configured DBaaS-DevLeader-Role, display name NULL on the # grant row itself). sub_grant_role_is_aws_owner = ( select(CmdbCloudRole.id) .where( CmdbCloudRole.role_key == CmdbCloudAccessGrant.role_key, CmdbCloudRole.display_name == aws_owner_role_name, ) .correlate(CmdbCloudAccessGrant) .exists() ) owner_equiv_role_conds = [ and_( Integration.provider == prov, CmdbCloudAccessGrant.role_key == role_key, ) for prov, (role_key, _disp) in _DEFAULT_OWNER_ROLE_BY_PROVIDER.items() if role_key != "__resolve_at_runtime__" ] owner_equiv_role_conds.append( and_( Integration.provider == "aws", or_( CmdbCloudAccessGrant.role_display_name == aws_owner_role_name, sub_grant_role_is_aws_owner, ), ) ) sub_owner_equiv_grant = ( select(CmdbCloudAccessGrant.id) .where( CmdbCloudAccessGrant.scope_id == CmdbCloudScope.id, CmdbCloudAccessGrant.person_id == CmdbCloudScope.owner_person_id, CmdbCloudAccessGrant.status == GRANT_STATUS_ACTIVE, or_(*owner_equiv_role_conds), ) .correlate(CmdbCloudScope, Integration) .exists() ) rows = (await session.execute( select( CmdbCloudScope.id, CmdbCloudScope.display_name, CmdbCloudScope.owner_person_id, ) # Join Integration so we can filter on (a) provider has an # auto-Owner-grant policy and (b) integration is currently # enabled. Both gates match the runtime guards that # ``_post_provision_finalize`` itself would otherwise just # log+skip every tick — keeps the candidate set in sync with # what the finalize step can actually fulfil. .join( Integration, Integration.id == CmdbCloudScope.integration_id, ) .where( CmdbCloudScope.deleted_at.is_(None), CmdbCloudScope.status == SCOPE_STATUS_ACTIVE, CmdbCloudScope.owner_person_id.isnot(None), # Exclude production scopes — ``_reuse_or_create_grant_row`` # raises ``CLOUD_SCOPE_PROD_BLOCKED`` on these and the # outer ``except Exception`` in ``_post_provision_finalize`` # would log a failure-audit every 10 minutes forever. Prod # scopes are explicitly out of self-service scope. CmdbCloudScope.environment.notin_(PROD_ENVIRONMENTS), # Billing-account scopes have no owner-equivalent IAM role # to mint, so provision_initial_owner_grant returns None # for them every tick — exclude from the candidate set so # they don't churn fixed=0 forever (symmetric with the # prod / leaver exclusions). CmdbCloudScope.scope_type != SCOPE_TYPE_GCP_BILLING_ACCOUNT, # Aliyun RD master (payer) is owner-exempt infra (raw. # rd_master) — no owner-equivalent role to mint, so it # would churn fixed=0 forever like a billing account. func.json_extract( CmdbCloudScope.raw_data, "$.rd_master", ).is_(None), # Disabled integrations are out of operator scope — admin # disabled them intentionally, the cron must not keep # poking at them. Integration.enabled.is_(True), # Only providers we actually have an auto-Owner policy # for can be repaired by this cron. Integration.provider.in_(supported_providers), ~sub_helix_owner_grant, ~sub_owner_is_leaver, sub_provisioned_by_helix, ~sub_owner_equiv_grant, # Skip deterministically-unresolvable scopes still inside # the retry window (see ``unresolvable_stale_cutoff`` # above). NULL = resolvable / never attempted = scan. or_( CmdbCloudScope.auto_owner_grant_unresolvable_at.is_(None), CmdbCloudScope.auto_owner_grant_unresolvable_at < unresolvable_stale_cutoff, ), ) # Oldest first — anything that's been missing the longest # gets priority. Bounded by batch_cap. .order_by(CmdbCloudScope.created_at.asc()) .limit(batch_cap) )).all() candidates = [ (str(r.id), r.display_name, str(r.owner_person_id)) for r in rows ] if not candidates: return {"status": "success", "scanned": 0, "fixed": 0, "errors": []} fixed = 0 errors: list[str] = [] # Scopes whose owner role is deterministically unresolvable. These # are NOT crashes (the finalize path stamped the backoff marker and # returned cleanly) — but they're a config problem only an admin can # fix (wrong Identity Center permission-set name), so we surface them # via the failure-notifier exactly ONCE per re-scan: a marked scope # drops out of the candidate predicate for ``retry_hours``, so it # only re-appears (and re-notifies) after the window. Tracked by # scope_id (UUID) so the failure-notifier signature stays stable # across runs (``compute_failure_signature`` collapses UUIDs) and # the dedup gate suppresses re-spam until an admin acks. unresolvable: list[str] = [] for scope_id, scope_name, owner_person_id in candidates: try: # ``is_backfill=True``: this cron retroactively writes # Helix-side bookkeeping for scope owners who already have # cloud-side access. Suppresses the "[已开通]" member # notification (the owner didn't ask for this; sending a # "云访问已生效" card on an access they've had for months is # misleading and was the user-visible failure that drove # PR #6XX after PR #671 first deployed the cron — 14 # owners got the noise card before the gate landed) and # writes ``expires_at=NULL`` so the renewal-nag sweeper # leaves these rows alone. # Thread the cron's effective retry window into the marker # helper so the audit short-circuit uses the SAME window as # this candidate-skip scan — otherwise an admin override of # ``unresolvable_retry_hours`` away from the default desyncs # the two (a <6h override would suppress the audit row on a # legitimate re-surfacing). await _post_provision_finalize( scope_id, is_backfill=True, unresolvable_retry_hours=retry_hours, ) except Exception as exc: # noqa: BLE001 # _post_provision_finalize wraps its own work in # try/except; if anything still escaped here, log + count. errors.append( f"backfill scope={scope_name!r} ({scope_id}): " f"{type(exc).__name__}: {exc}"[:200] ) continue # Verify the OWNER's grant actually landed — _post_provision_finalize # is best-effort; a LockBusy / role-resolve-skipped path leaves the # owner row absent (count=0). Only count this scope as fixed when an # ACTIVE source='helix' row exists *for the owner* (mirroring the # candidate scan predicate exactly; a non-owner helix grant from # owner_add_member must not falsely count as fixed). ACTIVE-only # (review-sweep #2/#7): a PENDING row means the driver bind failed # and the owner has NO cloud access — counting it as fixed would # report success for a grant that never landed and is the exact # silent-loss class this cron exists to surface. async with async_session_factory() as verify_session: landed_grant_count = await verify_session.scalar( select(func.count(CmdbCloudAccessGrant.id)).where( CmdbCloudAccessGrant.scope_id == scope_id, CmdbCloudAccessGrant.person_id == owner_person_id, CmdbCloudAccessGrant.source == "helix", CmdbCloudAccessGrant.status == GRANT_STATUS_ACTIVE, ) ) # Did the finalize path stamp the deterministic-unresolvable # marker this tick? A candidate only reaches the loop when the # marker is NULL or stale, so a marker present after finalize # means it was (re-)stamped now — surface it. Read in the same # verify session to avoid a third round-trip. unresolvable_marker = await verify_session.scalar( select(CmdbCloudScope.auto_owner_grant_unresolvable_at) .where(CmdbCloudScope.id == uuid.UUID(scope_id)) ) if landed_grant_count: fixed += 1 elif unresolvable_marker is not None: unresolvable.append(scope_id) result = { "status": ( "success" if not errors and not unresolvable else "partial_failure" ), "scanned": len(candidates), "fixed": fixed, "unresolvable": len(unresolvable), "errors": errors[:10], # cap audit detail size "batch_cap": batch_cap, } # Surface partial failures at the scheduled-job layer — same # convention as run_grants_sync (handlers.py:2007-2015) and the # project-wide rule "Scheduled-job handlers raise on failure" # (returning status='failed'/'partial_failure' silently lands # scheduled_job_runs.status='ok' and the failure-notifier never # fires). Without this raise, the backfill cron's whole purpose # — being the visible safety-net for the auto-Owner grant flow — # is defeated when its own exceptions go unreported. # Unresolvable scopes are surfaced via the SAME raise → failure- # notifier channel as genuine crashes (a misconfigured permission-set # name needs an admin), but kept distinct in the message so ops can # tell "cron crashed" from "fix the perm-set name". The candidate # predicate's retry window means a given scope only re-enters (and # re-raises) once per ``retry_hours``, so this is at-most-once-per- # window, not every tick. if errors or unresolvable: parts: list[str] = [] if errors: parts.append("; ".join(errors)) if unresolvable: parts.append( f"auto-owner grant deterministically unresolvable for " f"{len(unresolvable)} scope(s) — permission-set name " f"missing (platform_settings " f"{_AWS_DEFAULT_OWNER_ROLE_NAME_SETTING!r}) or owner " f"principal not found at the provider IdP; see the " f"cloud.scope_request.auto_owner_grant skipped-audit " f"rows for the per-scope reason: " + ", ".join(unresolvable[:10]) ) detail = " | ".join(parts)[:500] raise RuntimeError( f"cmdb.auto_owner_grant_backfill partial_failure " f"scanned={len(candidates)} fixed={fixed} " f"unresolvable={len(unresolvable)}: {detail}" ) return result @register_scheduled_job( key="cmdb.billing_daily_sync", weight="heavy", # 多分钟云 API 扫描 — 见 registry.SchedJobWeight display_name="云账单每日抓取", description=( "每日 06:00 拉取昨天的 scope 级日级账单(AWS Cost Explorer GROUP BY " "LINKED_ACCOUNT / Azure Cost Management / GCP BigQuery billing export /" " 阿里云 BSS),写入 cmdb_cloud_billing_daily。每 scope 最后一天 " "与过去 7 日均值对比,超出 ``cloud.daily_spike_multiplier`` 触发 " "Feishu 告警(scope owner + IT)。" ), ) async def run_billing_daily_sync( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.core.distributed_lock import LockBusy, distributed_lock from app.domains.cmdb.lifecycle_service import ( sync_billing_daily_for_integration, ) integration_id = str(config.get("integration_id") or "").strip() if not integration_id: return {"status": "skipped", "reason": "missing integration_id"} # Per-integration inner lock — same pattern as ``run_grants_sync`` # (line 1135). The outer ``execute_job`` single-flight is keyed by # ``handler_key`` (shared across every integration's per-row cron), # so this inner lock does NOT enable cross-tenant parallel cron — # the outer lock still serialises that. What this inner lock DOES # solve: the HTTP "sync now" button (admin) and any future MCP / # cron-equivalent caller that bypasses ``execute_job`` enter # ``sync_billing_daily_for_integration`` directly. Without this # inner key the admin click can race a concurrent cron tick that # already passed the outer lock, double-call the vendor billing # API, and produce two ``last_seen_billing_daily_at`` updates from # different runs. Keying on integration_id keeps the per-cloud # serialisation tight without affecting the handler-wide cron lock. # Drop a stale ``{start_date, end_date}`` pin so a forgotten backfill # window can't freeze the recurring daily sync — see # ``_resolve_recurring_billing_window`` and the 2026-06-16 Azure # cost_facts incident (same footgun, sibling handler). try: start_date, end_date, dropped_window = ( _resolve_recurring_billing_window(config) ) except ValueError as exc: return { "status": "skipped", "reason": f"invalid date in config: {exc}", "integration_id": integration_id, } if dropped_window: logger.warning( "cmdb.billing_daily_sync %s: %s", integration_id, dropped_window, ) try: async with distributed_lock( f"cloud_sync:billing_daily:{integration_id}", ttl=600, ): result = await sync_billing_daily_for_integration( integration_id, start_date=start_date, end_date=end_date, ) except LockBusy: return { "status": "skipped", "reason": "another billing_daily sync is already running", "integration_id": integration_id, } # Same pattern as cost_facts_sync below: if the inner service # returns ``failed``/``partial_failure`` OR accumulated per-scope # errors, we must raise so ``scheduled_job_runs.status='failed'`` # and the failure-notifier fires. The inner service does NOT # currently set ``status="partial_failure"`` on per-scope errors # (it returns the success-shape dict with ``errors=[...]``); check # both shapes to catch BQ misconfig (status=failed) AND per-scope # cloud-side errors (errors list populated) — the latter was a # round-2 audit P2 hole. int_sig = integration_id.replace("-", "") inner_status = result.get("status") inner_errors = result.get("errors") or [] if inner_status in ("failed", "partial_failure") or inner_errors: reason = result.get("reason") detail = "; ".join(inner_errors)[:500] or reason or "" # Tag the surfaced status so the failure_notifier dedup key # distinguishes whole-pipeline failures from per-scope errors. surfaced_status = inner_status or "partial_failure" raise RuntimeError( f"cmdb.billing_daily_sync int_sig={int_sig} " f"{surfaced_status}: {detail}", ) if dropped_window and isinstance(result, dict): result.setdefault("window_note", dropped_window) return result @register_scheduled_job( key="cmdb.cost_facts_sync", slo=SLO_CMDB_COST_FACTS_SYNC, weight="heavy", # 多分钟云 API 扫描 — 见 registry.SchedJobWeight display_name="云成本明细抓取", description=( "每日凌晨拉取昨天的 line-item 成本明细(Azure Cost Management ResourceId " "GROUP BY / 阿里云 BSS DescribeInstanceBill / AWS Cost Explorer " "GetCostAndUsageWithResources / 腾讯 DescribeBillDetail / GCP " "BigQuery export),写入 cmdb_cost_facts 作为成本分析的真理表。" "回填窗口通过 config={\"start_date\":\"YYYY-MM-DD\"," "\"end_date\":\"YYYY-MM-DD\"} 控制。" ), ) async def run_cost_facts_sync( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.core.distributed_lock import LockBusy, distributed_lock from app.domains.cmdb.cost_service import ( sync_bedrock_profile_facts_for_integration, sync_cost_facts_for_integration, ) integration_id = str(config.get("integration_id") or "").strip() if not integration_id: return {"status": "skipped", "reason": "missing integration_id"} # A leftover ``{start_date, end_date}`` backfill pin in a *recurring* # job freezes ongoing ingestion (prod incident 2026-06-16, Azure): # ``_resolve_recurring_billing_window`` drops a stale pin and falls # back to the rolling default. try: start_date, end_date, dropped_window = ( _resolve_recurring_billing_window(config) ) except ValueError as exc: return { "status": "skipped", "reason": f"invalid date in config: {exc}", "integration_id": integration_id, } if dropped_window: logger.warning( "cmdb.cost_facts_sync %s: %s", integration_id, dropped_window, ) # Per-integration inner lock — same pattern as ``run_grants_sync`` # (line 1135) and ``run_billing_daily_sync`` above. The outer # ``execute_job`` single-flight already serialises every # integration's cron tick via ``sched_job:cmdb.cost_facts_sync`` # (handler_key shared across rows), so this inner lock does NOT # enable cross-tenant parallel cron. What it DOES solve: the HTTP # backfill endpoint (`/cmdb/cost/backfill`) and the MCP # ``cost_backfill`` tool both enter ``sync_cost_facts_for_integration`` # directly without going through ``execute_job``; without this # inner key, an admin-clicked backfill could race the 04:xx cron # tick on the SAME integration, double-call the vendor billing # API (BigQuery / Cost Explorer / BSS — all billed per-query), # and double-write upserts that ON DUP KEY UPDATE only resolves # at the row level (driver-side calls already burnt). All three # entry points (HTTP, MCP, cron) now serialise on the same # ``cloud_sync:cost_facts:{integration_id}`` key. try: async with distributed_lock( f"cloud_sync:cost_facts:{integration_id}", ttl=900, ): result = await sync_cost_facts_for_integration( integration_id, start_date=start_date, end_date=end_date, ) # Additive Bedrock per-profile (UsedBy tag) breakdown for the # SAME window, so the per-AK detail page's by-profile sum # reconciles to its mtd_usd. No-op unless the integration has # cost_facts.aws_profile_tag. configured. Failures here # are non-fatal — the profile breakdown is a supplementary # drill-down, NOT the cost truth (cmdb_cost_facts is). Surface # the outcome in the result detail without failing the main # cost-facts run. try: profile_result = ( await sync_bedrock_profile_facts_for_integration( integration_id, start_date=start_date, end_date=end_date, ) ) if isinstance(result, dict): result["profile_sync"] = profile_result except Exception as exc: # noqa: BLE001 logger.warning( "cmdb.cost_facts_sync %s: profile breakdown pass " "failed (non-fatal): %s", integration_id, format_exception_for_log(exc), ) if isinstance(result, dict): result["profile_sync"] = { "status": "failed", "reason": format_exception_for_log(exc, max_len=300), } except LockBusy: return { "status": "skipped", "reason": "another cost_facts sync is already running", "integration_id": integration_id, } # Raise on both ``partial_failure`` and ``failed`` so the # ``scheduled_job_runs`` row maps to ``status=failed`` and the # failure-notifier (Feishu) fires. Same posture as # ``cmdb.resource_sync`` (handlers.py:806). Without raising on # ``failed`` (driver load error, every-scope error with zero # upserts) the run silently lands on success — misleading the # admin UI's execution history. ``ok`` (incl. all-scopes # ``not_supported`` for stub drivers) and ``skipped`` (disabled / # missing integration) keep returning normally — those are # legitimate non-failure outcomes. inner_status = result.get("status") if inner_status in ("failed", "partial_failure"): errs = result.get("errors") or [] reason = result.get("reason") detail = "; ".join(errs)[:500] or reason or "" # Per-scope counts so the failure-notifier message shows # blast radius ("1 of 38 failed" vs "all failed") without the # operator having to dig the discarded result_summary out of # logs. ``run`` raises after this, so result_summary lands # null on the run row — counts in the message are the only # surviving observable. (Carryover fix from the 2026-05-21 # Azure cost_facts_sync incident — single-scope transient # left the operator unable to tell whether 1 or 38 scopes # actually failed.) Suppress the scopes-summary suffix when # ``per_scope`` is empty by design — early-return paths # ("driver load failed" / "FX rate unavailable" / "BQ billing # source misconfigured") never enter the iteration loop, so # emitting "(scopes ok=0 failed=0 total=0)" there would be # misleading. per_scope = result.get("per_scope") or [] if per_scope: ok_count = sum( 1 for r in per_scope if r.get("status") == "ok" ) fail_count = sum( 1 for r in per_scope if r.get("status") in ("failed", "timeout") ) scope_summary = ( f" (scopes ok={ok_count} failed={fail_count} " f"total={len(per_scope)})" ) else: scope_summary = "" raise RuntimeError( f"cmdb.cost_facts_sync {integration_id} {inner_status}" f"{scope_summary}: {detail}", ) if dropped_window and isinstance(result, dict): result.setdefault("window_note", dropped_window) return result @register_scheduled_job( key="cmdb.cost.nightly_recompute_allocation", display_name="云成本归属夜间重算", description=( "每日凌晨调用 recompute_allocation(days=90),对最近 90 天的 " "cmdb_cost_facts 重新应用 cmdb_cost_allocation_rules + 范围" "owner 兜底(PR #241),把 ingest 阶段尚未命中规则的存量数据" "重新归到所属人头上。手工触发与该任务等价——区别仅在于 cron 自动跑。" ), ) async def run_cost_nightly_recompute_allocation( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: """Nightly re-stamp of cost-fact allocation across the 90-day window. Why this exists: PR #241 added the scope-owner fallback inside ``ingest_cost_facts`` AND ``recompute_allocation`` so a fact whose rule ladder misses still attributes to ``cmdb_cloud_scopes.owner_person_id`` instead of landing as "未分摊". But the fallback only takes effect on NEW writes — every existing fact written under the old rule-only logic stayed unallocated until an admin clicked the "重算分摊" button on ``/admin/cloud-cost``. Prod observation that motivated this cron: on 2026-05-10 the dashboard surfaced $7,316.10 / 99.8% as 未分摊 even though every scope had ``owner_person_id`` populated — the data was fine, the recompute was simply never run after PR #241 landed. Daily recompute amortises the cost of re-stamping the table across all rule edits / scope-owner changes between runs, and means an admin no longer has to remember to click the button after editing rules. ``days=90`` matches the same rolling window the admin-button HTTP path uses. ``recompute_allocation`` streams the window in 10k-row scan batches, committing each batch, so it never holds row locks across the whole table. Each scan statement is additionally bounded to a 500k-wide id band (2026-07-04 fix): the unbounded keyset form let one statement range-scan the multi-million-row pre-cutoff history block and TiDB's instance memory controller killed it with error 8176. A steady-state run (nothing changed) is dominated by the scan — ~270s at the current 2.7M-row / 90-day window. A re-stamp batches its writes by target tuple (one ``UPDATE … WHERE id IN (…)`` per distinct owner per scan batch), so even a full ~500k-row re-stamp stays well inside the 1800s hard timeout. (Pre-2026-05-27 the writes were per-row — ~10ms each — so a 500k re-stamp needed >1h and timed out; see ``recompute_allocation``.) Failure semantics: any raise propagates to the scheduled-job runner which marks the run ``failed`` and triggers the existing platform-wide failure-notifier (in-app + dedup-ack flow per ``feedback_failure_notifier``). No bespoke wiring. Concurrency: serialized with the manual HTTP "重算分摊" button (cost_router.py) and the ``cost_recompute_allocation`` MCP tool via the same ``distributed_lock("cost_recompute_allocation")`` key. Without this, a cron tick that fires while an admin is holding the button (or vice versa) re-stamps the same 90-day table in parallel, racing on ON-DUPLICATE-KEY-UPDATE writes and producing non-deterministic ``allocation_rule_id`` / ``cost_owner_person_id`` values per fact. The lock TTL is set to the executor's hard timeout (1800s) so the lock outlives the longest run it could possibly cover — the old 600s TTL was sized against a stale "≈12s" assumption and silently expired mid-run on every real recompute (270s steady-state, longer on a re-stamp), which is what let the 2026-05-27 03:30 and 04:49 ticks run concurrently and contend. If a run is killed at the 1800s timeout the ``async with`` still unwinds and releases the lock; a hard worker death is covered by the TTL. While held, the LockBusy branch keeps any other holder (cron retry / admin button / MCP) from piling on. """ from app.core.distributed_lock import LockBusy, distributed_lock from app.domains.cmdb.cost_service import recompute_allocation days = int(config.get("days") or 90) try: async with distributed_lock( "cost_recompute_allocation", ttl=1800, ): summary = await recompute_allocation(db, days=days) await db.commit() return summary except LockBusy: # Another holder (admin button or MCP tool) is mid-recompute. # Returning a skipped summary keeps the scheduled_job_runs # row honest — the next cron tick will retry; we don't want # to pile on and double-write. return { "status": "skipped", "reason": ( "another recompute_allocation holder is active; " "next cron tick will retry" ), } @register_scheduled_job( key="cmdb.cost_facts_monthclose", display_name="云成本明细 · 月结重拉(对账)", description=( "每月 5 号 / 12 号 07:00(Asia/Shanghai)的【月结重拉调度器】。把【上一个" "完整自然月】的 line-item 成本明细整月重新拉取,追平云厂商在月末 " "finalize 时的晚到核价修订(日累积 S3/ECR/Cost Explorer、未打 Owner " "标签的 Bedrock 用量;2026-06 实测 AWS SG 5 月差 $3.84),使 " "cmdb_cost_facts 与云控制台对账到分。为什么需要:每日 " "cmdb.cost_facts. 只滚动重拉「昨天起往前 3 天」,一个使用日过了 " "3 天就冻结,再不追修订。\n" "⚠️ 本任务不再自己串行拉全部账号(旧实现:8 账号 × ~205 scope × 整月 " "line-item 塞进一个 30min 超时里,prod 从未跑完过 —— 2026-06 跑 >6h 被 " "stale-running 清扫、2026-07 命中 1800s 超时)。现在它是【扇出调度器】:" "对每个启用云账号,把目标月切成 chunk_days(默认 7)天子窗,逐个通过该" "账号自己的 cmdb.cost_facts. 任务(execute_job + 显式过去窗口 " "allow_past_window,绕过每日任务的 stale-pin 自愈)重拉。每个子窗复用" "该账号自己的超时/单飞锁/run 行/失败告警,慢账号或失败账号被隔离、可" "独立重试。调度本身以有限并发(max_parallel,默认 3)后台 drain,秒级" "返回;完成情况以 cmdb.cost_facts. 的 run 历史为准,本任务的 run " "行只记录派发计划。\n" "config 可覆写:{\"start_date\":\"YYYY-MM-DD\",\"end_date\":" "\"YYYY-MM-DD\"}(任意区间回填)、chunk_days、chunk_timeout_secs、" "max_parallel。" ), ) async def run_cost_facts_monthclose( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: """Monthly month-close DISPATCHER (not an inline worker). Re-pulls the PREVIOUS calendar month's line-item cost facts so ``cmdb_cost_facts`` reconciles to the cent against the cloud console after the vendor finalizes the month. The daily ``cmdb.cost_facts.`` cron only re-fetches a rolling 3-day window, so any late vendor re-rating (month-end finalization, late-arriving untagged usage, daily-accruing S3/ECR/Cost-Explorer infra) that lands after a usage day ages out of that window never reaches Helix. This closes that gap once a month. NOT the same as ``cmdb.cost.nightly_recompute_allocation`` (which re-APPLIES allocation rules to EXISTING facts — re-attribution); this re-PULLS amounts from the vendor (data freshness). Both are needed and neither subsumes the other. **Why a dispatcher, not an inline loop.** The original implementation iterated every enabled cloud integration (~8) × every tenant scope (~205) × the whole month of per-resource line items sequentially, all inside a single 30-minute handler timeout. That workload is hours of vendor-API-bound work — a single heavy account's 3-day window alone runs 400-1600s in prod, so its full month is multi-thousand seconds and the aggregate never fit any single timeout. It literally never completed once in prod (2026-06 ran >6h and was reaped by the 6h stale-running sweeper; 2026-07 hit the new 1800s timeout), so the "reconcile to the cent" it promised never actually happened. So this now FANS OUT. For each enabled cloud integration it splits the target month into ``chunk_days``-day (default 7) sub-windows and runs each through that integration's own ``cmdb.cost_facts.`` job via :func:`execute_job`, passing an explicit past-month window plus ``allow_past_window=True`` (an ephemeral override that bypasses the recurring stale-pin self-heal in :func:`_resolve_recurring_billing_window` — it is never persisted to the daily job's row, so that job still self-heals a forgotten pin as before). Each sub-run therefore reuses the per-account job's own timeout / single-flight lock / run row / failure-notifier, so a slow or failing account is ISOLATED and independently retryable — no single account can wedge the whole reconcile. Chunks of the SAME account run sequentially (they serialize on that job's ``sched_job:{id}`` + inner ``cloud_sync:cost_facts:{id}`` locks anyway, which is exactly what keeps them from racing the daily 04:xx tick and double-billing the vendor). Different accounts drain in parallel up to ``max_parallel`` (default 3). The dispatch itself runs as a retained background task and this handler returns in seconds with the dispatch plan — the per-account ``cmdb.cost_facts.`` run history is the source of truth for completion. This handler's run row is short-lived, so the 6h stale-running sweeper can never reap it again. Config overrides (all optional): ``start_date`` / ``end_date`` (ISO, both required together — arbitrary backfill range instead of the previous month), ``chunk_days`` (1-31, default 7), ``chunk_timeout_secs`` (per sub-run timeout, clamped to the executor's [60, 7200], default 7200), ``max_parallel`` (1-8 concurrent accounts, default 3). """ from datetime import date as _date from uuid import UUID from sqlalchemy import select as _select from app.core.background import spawn from app.integrations.models import Integration from app.scheduled_jobs.models import ScheduledJob from app.scheduled_jobs.service import execute_job # Window: explicit override for ad-hoc backfill, else the previous # complete calendar month (UTC, matching the UTC-anchored # ``CmdbCostFact.date`` column and the daily cron's UTC window). start_str = (config.get("start_date") or "").strip() if config else "" end_str = (config.get("end_date") or "").strip() if config else "" if start_str and end_str: try: start_date = _date.fromisoformat(start_str) end_date = _date.fromisoformat(end_str) except ValueError as exc: return {"status": "skipped", "reason": f"invalid date: {exc}"} else: today = datetime.now(UTC).date() first_of_this_month = today.replace(day=1) end_date = first_of_this_month - timedelta(days=1) # last day prev month start_date = end_date.replace(day=1) # first day prev month if start_date > end_date: return { "status": "skipped", "reason": f"inverted window {start_date}..{end_date}", } def _clamp_int(raw: Any, default: int, lo: int, hi: int) -> int: try: return max(lo, min(hi, int(raw))) except (TypeError, ValueError): return default chunk_days = _clamp_int(config.get("chunk_days"), 7, 1, 31) # Per-sub-run timeout. The executor re-clamps to [60, 7200]; keep the # default at the ceiling so a heavy account's weekly chunk (prod: a # 3-day Azure/Aliyun window runs up to ~1600s, so a 7-day one has # ample headroom under 7200) never times out on a normal month. chunk_timeout = _clamp_int(config.get("chunk_timeout_secs"), 7200, 60, 7200) max_parallel = _clamp_int(config.get("max_parallel"), 3, 1, 8) # Split [start, end] inclusive into consecutive chunk_days-day windows. chunks: list[tuple[date, date]] = [] cursor = start_date step = timedelta(days=chunk_days) one_day = timedelta(days=1) while cursor <= end_date: chunk_end = min(cursor + step - one_day, end_date) chunks.append((cursor, chunk_end)) cursor = chunk_end + one_day # Enabled cloud integrations, mapped to their per-account # ``cmdb.cost_facts.`` job row (keyed by config.integration_id). integrations = ( await db.execute( _select(Integration).where( Integration.kind == "cloud", Integration.enabled.is_(True), ) ) ).scalars().all() integ_by_id = {str(i.id): i for i in integrations} cost_facts_jobs = ( await db.execute( _select(ScheduledJob).where( ScheduledJob.handler_key == "cmdb.cost_facts_sync", ScheduledJob.enabled.is_(True), ) ) ).scalars().all() dispatched: list[dict[str, Any]] = [] skipped: list[dict[str, Any]] = [] plan: list[tuple[UUID, str, str]] = [] # (job_id, iid, integration name) seen_iids: set[str] = set() for job in cost_facts_jobs: iid = str((job.config or {}).get("integration_id") or "").strip() integ = integ_by_id.get(iid) if integ is None: # Job row for a disabled / non-existent cloud integration — # nothing to reconcile. continue seen_iids.add(iid) plan.append((job.id, iid, integ.name)) dispatched.append({ "integration_id": iid, "name": integ.name, "job": job.name, "chunks": len(chunks), }) # Enabled cloud integrations with no enabled cost_facts job row can't # be reconciled through the fan-out — surface them instead of # silently dropping (drivers without ``list_cost_facts``, or a job an # admin disabled). for iid, integ in integ_by_id.items(): if iid not in seen_iids: skipped.append({ "integration_id": iid, "name": integ.name, "reason": "no enabled cmdb.cost_facts. job row", }) window = {"start": start_date.isoformat(), "end": end_date.isoformat()} if not plan: # Nothing to dispatch — honest no-op rather than a spawned task # that does nothing. return { "status": "ok", "mode": "dispatched", "window": window, "chunk_days": chunk_days, "dispatched": [], "skipped": skipped, "note": "no enabled cloud cost_facts jobs to reconcile", } chunk_windows = [ {"start_date": s.isoformat(), "end_date": e.isoformat()} for s, e in chunks ] async def _drain_account( job_id: UUID, iid: str, name: str, sem: asyncio.Semaphore, ) -> None: """Run every month chunk for one account sequentially through its own ``cmdb.cost_facts.`` job. Sequential is required: the chunks share that job's single-flight lock. Per-chunk failures are logged and do not abort the remaining chunks (each chunk's own run row + failure-notifier already record the failure).""" async with sem: for cw in chunk_windows: try: await execute_job( job_id, triggered_by="cost_facts_monthclose", config_override={ **cw, "allow_past_window": True, "timeout_secs": chunk_timeout, }, ) except Exception: # noqa: BLE001 # execute_job already writes a failed run row + fires # the per-account notifier for handler failures; this # catch only covers dispatch-layer errors (lock/Redis) # so one bad chunk can't kill the account's remaining # chunks or sibling account drains. logger.exception( "cost_facts_monthclose: chunk %s..%s dispatch " "failed for %s (%s)", cw["start_date"], cw["end_date"], name, iid, ) async def _orchestrate() -> None: sem = asyncio.Semaphore(max_parallel) results = await asyncio.gather( *( _drain_account(job_id, iid, name, sem) for (job_id, iid, name) in plan ), return_exceptions=True, ) failures = [r for r in results if isinstance(r, Exception)] logger.info( "cost_facts_monthclose drain finished: %d account(s) × %d " "chunk(s) for %s..%s (%d drain-level error(s))", len(plan), len(chunks), window["start"], window["end"], len(failures), ) spawn(_orchestrate(), name="cost_facts_monthclose_drain") return { "status": "ok", "mode": "dispatched", "window": window, "chunk_days": chunk_days, "chunk_timeout_secs": chunk_timeout, "max_parallel": max_parallel, "dispatched": dispatched, "skipped": skipped, "note": ( "fanned out to per-account cmdb.cost_facts. jobs; " "completion + per-account failures are visible in each " "account's cost_facts run history" ), } @register_scheduled_job( key="apikey.budget_check", display_name="API Key 月度预算检查", description=( "每日上午检查每个 status='active' 的 ApiKeyAssignment 的 MTD " "已花费(按 AK 关联的产品 bucket 分别求和:Foundry 按 RG、" "Bedrock 按 IAM 用户 principal tag、Vertex 按 GCP 项目 scope;" "Bailian 因 BSS 仅提供账户级数据,自动跳过该 AK 不告警)对比 " "monthly_budget_usd。当跨过 80% 或 100% 阈值时,通过 " "NotificationService 给申请人 + 直属上级各发一条 in-app + " "Feishu 卡片告警。每个 (assignment, month, threshold) 在 " "apikey_budget_alerts_dedup 里只落一行,跨日不重发。" ), ) async def run_apikey_budget_check( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: """Sweep every active ApiKeyAssignment and fire 80%/100% alerts. Cron is single-flight at the handler-key level via the outer ``execute_job`` lock (``sched_job:apikey.budget_check``). No per-assignment inner lock today — there's no parallel admin "force budget check" entry point yet. The dedup row's UNIQUE index catches duplicate WRITES, but the SELECT-then-INSERT order inside ``evaluate_assignment`` means two concurrent runs racing past the SELECT would both DISPATCH the notification before either INSERTs (outer lock today prevents this). **REQUIRED if a manual-trigger entry point is ever added (HTTP "force budget check" / MCP equivalent)**: mirror the inner-lock pattern from ``cmdb.cost_facts_sync`` — a per-assignment ``apikey:budget_check:{assignment_id}`` lock OR re-order the flow to write the dedup row BEFORE dispatch (and roll it back on dispatch failure). Without the lock, a manual trigger racing the cron tick can fan out duplicate alerts even though the UNIQUE index keeps the dedup ledger consistent. Anchors "today" to Asia/Shanghai (the canonical FinOps tz used by cmdb crons) so the dedup ``billing_month`` matches the user-facing month even when the worker container runs UTC. """ from zoneinfo import ZoneInfo from app.core.database import async_session_factory from app.domains.apikey.budget_alerts import evaluate_assignment from app.domains.apikey.models import ApiKeyAssignment today = datetime.now(ZoneInfo("Asia/Shanghai")).date() summary: dict[str, Any] = { "evaluated": 0, "fired": 0, "skipped": 0, "errors": [], "today": today.isoformat(), "details": [], } async with async_session_factory() as session: # Only active assignments — pending_approval / rejected / # closed all carry stale budgets that aren't a meaningful # signal. ``status='active'`` is the post-grant state set by # the auto-grant flows (Bedrock / Bailian / Azure OpenAI). rows = (await session.execute( select(ApiKeyAssignment).where( ApiKeyAssignment.status == "active", ApiKeyAssignment.monthly_budget_usd > 0, ), )).scalars().all() # Round-49 (2026-05-25) crash-survivability fix: wrap the sweep # in try/finally so the tail commit fires even when the loop # raises mid-iteration (e.g. iteration 50 of 200 hits an # unhandled DB error: deadlock, conn loss, OOM, KeyboardInterrupt). # Pre-fix, the commit at the bottom is reached ONLY on the # happy path; an exception bubbling out of the for-loop skipped # the commit, so every dedup row claimed by iterations 1–49 in # this run was silently rolled back at session close — and the # next morning the cron re-evaluated the SAME (assignment, # month, threshold_pct) triple and re-fired the 80% / 100% # alert to the same users. Mirrors the dedup-preservation # invariant the Round-46 fix codified for the happy path. try: for assignment in rows: summary["evaluated"] += 1 try: result = await evaluate_assignment( session, assignment=assignment, today=today, ) except Exception as exc: # noqa: BLE001 — single-row failures must not abort the sweep logger.exception( "apikey.budget_check: evaluate failed for %s", assignment.request_number, ) summary["errors"].append({ "request_number": assignment.request_number, "error": format_exception_for_log(exc, max_len=300), }) continue fired_count = len(result.get("fired") or []) if fired_count: summary["fired"] += fired_count else: summary["skipped"] += 1 # Trim the per-assignment detail before persistence — at # ~500 active keys × full result blob the row would # bloat past the JSON column's working size. Keep just # the request_number + fired/skipped counts. summary["details"].append({ "request_number": result["request_number"], "fired": result.get("fired") or [], "mtd_usd": result.get("mtd_usd"), "budget_usd": result.get("budget_usd"), "pct_used": result.get("pct_used"), }) finally: # Round-46 (2026-05-20) critical fix: without this commit # every ``ApiKeyBudgetAlertsDedup`` row that # ``evaluate_assignment`` → ``_try_claim_dedup_slot`` writes # inside a ``begin_nested()`` SAVEPOINT is silently rolled # back when the session closes — the next morning's cron # evaluates the same (assignment, month, threshold_pct) # triple and re-fires the 80% / 100% alert, every day, all # month. Mirror the ``run_apikey_expiry_check`` tail (line # ~2100) which already commits explicitly. # # Round-49 (2026-05-25): moved into ``finally`` so a # mid-loop crash still preserves dedup rows for iterations # that did succeed (see try-block comment above for the # crash-replay storm the bare-tail path could fire). # savepoint-exempt: commit of an OWN session inside a `finally` # while the original exception may still be propagating. A # SAVEPOINT here would be opened on the WRONG session (`db`, the # job session — `session` is the dedicated one being committed), # buying no isolation at all, and `best_effort` re-raises # 1205/1213/1305, which in a `finally` would REPLACE the original # exception. Same call as the `RELEASE_LOCK` finallys in # `billing_service.close_month`. Losing this commit re-fires the # 80%/100% budget alert every morning all month, so it must not be # made more fragile. # savepoint-exempt: dedicated own session; savepoint in finally would mask try: await session.commit() except Exception: # noqa: BLE001 # If commit ITSELF fails (TiDB down, conn reset), there # is nothing we can do here — the session-scope # rollback will fire on context exit and the next # cron run will resync. Log + swallow so the original # exception (if any) keeps propagating to the # failure_notifier path. logger.error( "apikey.budget_check: dedup commit failed at sweep " "tail; next cron tick will re-evaluate but pre-" "commit dispatched alerts may have already fired", ) # Cap the persisted detail list at the first 200 assignments so # an org with >>200 active keys (we're at ~30 today) doesn't # eventually overflow the ``last_result_summary`` JSON column. # The full audit trail lives on ``audit_logs`` # (``apikey.budget_alert_sent``) — the cron summary is just a # quick-glance row. if len(summary["details"]) > 200: summary["details_truncated"] = True summary["details"] = summary["details"][:200] # Round-3 review follow-up (LS3 P2-A): when every evaluated # assignment hits "all-recipients-failed-dedup-released" (the # break-out branch in budget_alerts) we have a systemic dispatch # failure — open_id resolver down, Feishu webhook misconfigured, # notification service crashed. Without this signal the cron # silently logs "fired=0, skipped=N" and failure_notifier never # pages — alerts go silent. Raise when there's a definite # "we tried every assignment AND every dispatch failed" pattern # so the failure_notifier surfaces the outage immediately. total_evaluated = int(summary.get("evaluated") or 0) total_fired = int(summary.get("fired") or 0) all_release_count = 0 for detail in summary.get("details") or []: # Each ``fired`` entry has ``notified=False`` when the per- # recipient dispatch all failed and we released the slot. for slot in detail.get("fired") or []: if slot.get("notified") is False: all_release_count += 1 # Round-4 review follow-up (P2-3): threshold lowered to >=1 so # tenants with only one or two active API keys still page on # systemic dispatch failure. The earlier >=3 threshold would let # a small-tenant Feishu/open_id outage go silent for the entire # cycle. Single-assignment 100% release IS a clear systemic signal # — every recipient unreachable — so a 1-row trip is correct. # # Round-51 R2-P0-3 fix: the original condition # ``all_release_count >= total_evaluated`` only trips when EVERY # evaluated assignment was over budget AND every dispatch failed. # But assignments BELOW threshold never enter the ``fired`` list, # so a sweep of 100 keys with 5 over-budget all failing dispatch # had ``all_release_count=5, total_evaluated=100`` → 5 >= 100 is # False → no page. A Feishu outage hitting only the over-budget # subset stayed silent. # # Round-51 regression-review fix: the first attempt at the rewrite # was ``total_fired == 0 AND all_release_count >= 1``, but # ``total_fired`` is a counter that INCLUDES the ``notified=False`` # release rows (see handler line 2174-2178 where every entry in # ``result['fired']`` increments it). So ``all_release_count >= 1`` # implies ``total_fired >= 1``, making the new gate unreachable. # Real systemic signal: at least one assignment was over budget # AND zero of those crossed-threshold rows landed a notification. # ``total_fired - all_release_count`` is the dispatched-OK count; # zero-OK + at-least-one-release is the systemic outage shape. dispatched_ok = total_fired - all_release_count if dispatched_ok == 0 and all_release_count >= 1: raise RuntimeError( f"apikey.budget_check systemic dispatch failure: " f"{total_evaluated} assignments evaluated, 0 dispatched, " f"{all_release_count} all-recipients-failed slots released. " "Likely Feishu / open_id resolver / notification service " "outage — check NotificationService.dispatch logs." ) # Per-assignment evaluate failures were captured into ``summary["errors"]`` # and the loop continued so one bad key didn't abort the sweep — but a # swallowed evaluate error means that key's budget was NOT checked this # run (a runaway-spend alert silently not fired). Returning normally would # land ``runs.status='ok'`` and the failure notifier would never page. # Raise so the outage surfaces, matching the ``apikey.expiry_check`` / # ``billing`` sibling handlers (feedback_handler_status_must_raise). The # notifier is dedup-aware, so a persistent bad row pages once, not daily. if summary["errors"]: err_preview = summary["errors"][:3] raise RuntimeError( f"apikey.budget_check: {len(summary['errors'])}/{total_evaluated} " f"assignment(s) failed budget evaluation and were NOT checked " f"this run: {err_preview}" ) return summary @register_scheduled_job( key="apikey.expiry_check", display_name="API Key 到期提醒", description=( "每日 09:00 (Asia/Shanghai) 扫描每个 status='active' 的 " "ApiKeyAssignment;在 days_left == 7 或 days_left == 1 的边界给 " "申请人 + 直属上级各发一条 in-app + Feishu 提醒,促其在到期前 " "点击「续期」延长 valid_until。dedup 表 (assignment_id, " "valid_until_iso, threshold_days) 保证同一 validity 窗口内每档 " "只发一次;admin 延长后 valid_until 改变 → 自然产生新 dedup key " "→ 下一窗口照常发。同一任务把已越过 valid_until 的 active key " "持久化为 expired 并写一条系统审计;GET 只计算 effective status。" ), ) async def run_apikey_expiry_check( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: """Sweep every active ApiKeyAssignment with a valid_until date and fire expiry-warning cards on the 7-day and 1-day windows. Anchored to Asia/Shanghai so the dedup ``valid_until_iso`` key matches the user-facing date even when the worker runs UTC. Same posture as :func:`run_apikey_budget_check` — single-flight via the outer ``execute_job`` lock; per-assignment errors accumulate into ``summary['errors']`` and don't abort the sweep. Raises ``RuntimeError`` when systemic failure is detected (every evaluated assignment errored) so the ``scheduled_jobs.failure_notifier`` surfaces it. """ from zoneinfo import ZoneInfo from app.core.database import async_session_factory from app.domains.apikey.expiry_alerts import sweep_expiry_alerts from app.domains.apikey.service import ApiKeyService today = datetime.now(ZoneInfo("Asia/Shanghai")).date() # Keep the durable transition transaction short. The rows selected by # ``check_and_update_expired`` stay locked until commit; notification # delivery may perform slow external I/O and must not extend that lock # window or roll back an already-complete status+audit transition. async with async_session_factory() as transition_session: expired = await ApiKeyService( transition_session, ).check_and_update_expired(today=today) await transition_session.commit() # Reminder delivery and its dedup state have their own transaction. A # provider failure can be retried without redoing (or undoing) expiry. async with async_session_factory() as reminder_session: summary = await sweep_expiry_alerts(reminder_session, today=today) await reminder_session.commit() summary["expired"] = expired # Systemic failure detection: every evaluated assignment errored. # Mirrors run_apikey_budget_check's tail block. total_evaluated = int(summary.get("evaluated") or 0) total_errors = len(summary.get("errors") or []) if total_evaluated >= 1 and total_errors == total_evaluated: sample = "; ".join( str(e) for e in summary.get("errors", []) )[:500] raise RuntimeError( f"apikey.expiry_check systemic failure: " f"{total_evaluated} assignments evaluated, all errored. " f"Sample: {sample}" ) return summary @register_scheduled_job( key="apikey.rotation_retire", display_name="API Key · 轮换旧凭证退役", description=( "每小时扫描处于「轮换中」(rotation_retire_at 非空) 的 " "ApiKeyAssignment:到点 (retire_at <= now) 的退役旧凭证 " "(Bedrock delete_access_key / Bailian DeleteApiKey / Azure " "regenerateKey 旧槽),清空轮换状态并通知申请人「旧 key 已停用」;" "24h 内即将退役且未提醒过的发一条「即将停用」提醒。退役对云侧幂等 " "(凭证已被吊销/离职清理删除时 no-op);已 revoked 的行跳过。" ), ) async def run_apikey_rotation_retire( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: """Sweep every mid-rotation ApiKeyAssignment and retire due ones. Single-flight via the outer ``execute_job`` lock. The sweep commits per assignment (each retire performs an irreversible cloud delete, so we persist promptly) and accumulates per-assignment errors into ``summary['errors']``. Raises when ANY row errored — a failed retire leaves an old credential live, so the run must land failed and reach the ``scheduled_jobs.failure_notifier``. (This is strictly stricter than the earlier systemic-only gate on ``retire_attempted == retire_failed``, so the "scanned == errors dilution" failure mode that gate existed to avoid cannot occur; the counters are still kept in the summary/message for observability.) """ from app.core.database import async_session_factory from app.domains.apikey.rotation import sweep_rotation_retire async with async_session_factory() as session: summary = await sweep_rotation_retire(session) scanned = int(summary.get("scanned") or 0) attempted = int(summary.get("retire_attempted") or 0) failed = int(summary.get("retire_failed") or 0) errors = len(summary.get("errors") or []) if errors >= 1: # Any retire failure means an old credential that should be dead # is still live (or a mint tombstone had to be force-cleared) — # both are actionable, so the run must land failed and reach the # dedup-aware failure notifier. Per-assignment work is committed # row-by-row inside the sweep, so raising here loses nothing. sample = "; ".join(str(e) for e in summary.get("errors", []))[:500] raise RuntimeError( f"apikey.rotation_retire: {errors}/{scanned} rotations " f"errored ({failed}/{attempted} due retires failed; old " f"credentials may still be live). Sample: {sample}" ) return summary @register_scheduled_job( key="cmdb.access_expiry_sweep", display_name="云访问过期扫描", description=( "每日 08:00 扫 cmdb_cloud_access_grants:到期前 14 / 7 / 1 天" "提醒成员续期;到期当日调对应 driver 的 revoke_access 后置 " "status=expired(driver 失败也落盘 — 下一次 sweep 会重试 driver 侧)。" ), ) async def run_access_expiry_sweep( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.core.database import async_session_factory from app.domains.cmdb.lifecycle_service import sweep_expiring_grants async with async_session_factory() as session: summary = await sweep_expiring_grants(session) await session.commit() # Pre-launch hardening 2026-05-16 (CR4 P1) + Round-1 review (Review-D # P1-2): raise when error rate crosses 10% of scanned grants. # ``sweep_expiring_grants`` commits per-grant work BEFORE the handler # observes errors; raising on any single error would page on 95% # success and the page itself would be misleading ("expiry sweep # failed" when in fact 19 of 20 succeeded). Threshold gates against # systemic failure (rate-limit storm, IAM permission revoked, # widespread cloud outage) while letting per-grant noise drain # quietly into ``summary.errors`` for forensics. errors = summary.get("errors") or [] scanned = int(summary.get("scanned") or 0) err_threshold = max(5, int(scanned * 0.10)) # 5 minimum, or 10% of scanned if len(errors) >= err_threshold: detail = "; ".join(str(e) for e in errors)[:500] or "" raise RuntimeError( f"cmdb.access_expiry_sweep failed widely: {len(errors)} " f"errors of {scanned} scanned (threshold {err_threshold}); " f"sample: {detail}" ) return summary @register_scheduled_job( key="cmdb.renewal_lapse_sweep", display_name="云权限续期申请超期关单", description=( "每日 09:00 关闭超过 14 天仍未审批的云权限续期申请。" "关单既不是批准也不是驳回 —— grant 按原到期日正常回收,申请人收到" "通知。不关单会有两个后果:该 grant 的唯一「待审批」名额被永久占用," "申请人无法重新提交;以及这条申请几个月后仍可被批准,把一个大家早已" "忘记的权限复活。" ), ) async def run_renewal_lapse_sweep( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.core.database import async_session_factory from app.domains.cmdb.renewal_service import ( lapse_stale_renewal_requests, nudge_expiring_renewal_requests, ) async with async_session_factory() as session: # Nudge BEFORE lapsing, in the same tick: a request whose deadline # is 3 days out gets the warning, and one already past its # deadline gets closed. Reversing the order would nudge about # requests that are about to be closed one line later. nudge = await nudge_expiring_renewal_requests(session) summary = await lapse_stale_renewal_requests(session) await session.commit() summary = { **summary, "owners_nudged": nudge.get("nudged", 0), "errors": [*summary.get("errors", []), *nudge.get("errors", [])], } # Same adaptive threshold shape as the expiry sweep: the sweep commits # per-group, so raising on a single failure would page on 95% success. errors = summary.get("errors") or [] lapsed = int(summary.get("lapsed") or 0) threshold = max(5, int(lapsed * 0.10)) if len(errors) >= threshold: detail = "; ".join(str(e) for e in errors)[:500] or "" raise RuntimeError( f"cmdb.renewal_lapse_sweep failed widely: {len(errors)} " f"errors of {lapsed} lapsed (threshold {threshold}); " f"sample: {detail}" ) return summary @register_scheduled_job( key="cmdb.scope_idle_sweep", display_name="云账号闲置候选扫描", description=( "每周一 09:00 扫一遍 status='active' 的 cmdb_cloud_scopes:" "0 active grants 且过去 30 天 $0 花费的视为闲置候选," "汇总成一封邮件 + Feishu 发给 IT Admin 提醒考虑下线。" "仅提醒,不会自动 decommission(误报代价过高)。" ), ) async def run_scope_idle_sweep( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.core.database import async_session_factory from app.core.distributed_lock import LockBusy, distributed_lock from app.domains.cmdb.lifecycle_service import sweep_idle_scopes # Single-flight lock: matches the pattern used by every other # cmdb sweep handler (leaver_grant_sweep, manual_revoke_retry_sweep, # owner_semiannual_confirmation, etc.). A multi-worker deployment # or a quick cron-restart could otherwise spawn two concurrent # sweeps; even though the per-scope 14d dedup window absorbs # most of the duplicate-notification risk, a second concurrent # sweep wastes BigQuery / Feishu / DB cycles. ``LockBusy`` here is # not an error — the other run will do the work — so return a # zero-shape summary so the failure-notifier doesn't fire. try: async with distributed_lock( "cmdb.scope_idle_sweep", ttl=900, ): async with async_session_factory() as session: summary = await sweep_idle_scopes(session) await session.commit() except LockBusy: return { "candidates": 0, "notified_recipients": 0, "owners_notified": 0, "admins_notified": 0, "stamped_scopes": 0, "errors": [], "skipped_lock_busy": True, } errors = summary.get("errors") or [] if errors: detail = "; ".join(str(e) for e in errors)[:500] or "" raise RuntimeError( f"cmdb.scope_idle_sweep completed with errors: {detail}" ) return summary @register_scheduled_job( key="cmdb.decommission_cleanup_resume", display_name="下线前置清理 · 推进与重试", weight="light", description=( "每 5 分钟推进一次「管理员已确认、但云端还不能接受」的下线。云厂商拒绝" "关闭的常见原因是 scope 里还有东西(Azure " "ResourcesExistingOnSubscription / GCP liens_exist);这个作业按管理员" "确认时看到的那份清单删掉它们,再重试关闭。云上删除是异步的(Azure 资源组" "删除返回 202 后在 Deleting 里待几十秒到几十分钟),所以它重复推进直到" "云端接受,然后走既有的落库 + 停 grant + 审计路径,并给点击下线的人发**一条**" "完成通知。**只删确认过的 key**:确认之后新出现的资源不会被静默删掉," "而是把这次下线转成「需要重新确认」并通知人。重试预算(72 次 ≈ 6 小时)" "用尽、或碰到 Helix 删不掉的东西,同样只通知一次,不在页面上反复报错。" ), ) async def run_decommission_cleanup_resume( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.core.distributed_lock import LockBusy, distributed_lock from app.domains.cmdb.decommission_cleanup import ( MAX_SCOPES_PER_RUN, resume_pending_decommissions, ) limit = config.get("limit") try: limit = int(limit) if limit is not None else MAX_SCOPES_PER_RUN except (TypeError, ValueError): limit = MAX_SCOPES_PER_RUN if limit < 1: limit = MAX_SCOPES_PER_RUN # Single-flight: two overlapping ticks would both issue provider # deletes for the same keys. Those calls are idempotent, so the harm # is bounded, but the attempt counters would double-count and burn # the escalation budget at twice the intended rate. # # ``db`` is deliberately unused — ``resume_pending_decommissions`` # opens a session per scope so one scope's expected rollback (the # provider still refusing is a raise, not an error) cannot discard # the bookkeeping written for its siblings in the same tick. del db try: async with distributed_lock( "cmdb.decommission_cleanup_resume", ttl=600, ): summary = await resume_pending_decommissions(limit=limit) except LockBusy: return { "checked": 0, "closed": 0, "still_cleaning": 0, "needs_reconfirm": 0, "blocked": 0, "stalled": 0, "unreadable": 0, "notified": 0, "errors": [], "skipped_lock_busy": True, } # ``blocked`` / ``stalled`` / ``needs_reconfirm`` are FINDINGS, not # handler failures: each one has already notified the admin who # clicked, and going red every 5 minutes until they act would train # everyone to ignore this job. What deserves red is the job being # unable to do its own work — an exception per scope. errors = summary.get("errors") or [] if errors: detail = "; ".join(str(e) for e in errors)[:500] or "" raise RuntimeError( "cmdb.decommission_cleanup_resume hit errors on " f"{len(errors)} scope(s): {detail}" ) return summary @register_scheduled_job( key="cmdb.decommission_reconcile", display_name="已下线云账号的云端终态对账", weight="heavy", description=( "每日 04:20 复核每一个 Helix 已标记 closed、但云端关闭仍在进行中的 " "scope(AWS PENDING_CLOSURE / GCP DELETE_REQUESTED / Azure Disabled / " "阿里云 Checking·静默期),把云端当前说法写回 " "decommission_provider_state。关闭完成就落终态并停止复核;" "云端报告关闭失败(阿里云 CheckFailed / DeleteFailed)或账号又活了" "(GCP projects.undelete、Azure 订阅恢复,都是控制台自助操作)" "则写审计并给管理员发聚合告警。**只读云端,永不自动改回 status** —— " "撤销一个不可逆的管理操作是人的决定。" ), ) async def run_decommission_reconcile( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.core.database import async_session_factory from app.core.distributed_lock import LockBusy, distributed_lock from app.domains.cmdb.decommission_reconcile import ( MAX_SCOPES_PER_RUN, reconcile_decommissioned_scopes, ) limit = config.get("limit") try: limit = int(limit) if limit is not None else MAX_SCOPES_PER_RUN except (TypeError, ValueError): limit = MAX_SCOPES_PER_RUN if limit < 1: limit = MAX_SCOPES_PER_RUN # Single-flight, same as every other cmdb sweep. Two concurrent runs # would double the provider reads and could both append the same # contradiction alert. try: async with distributed_lock("cmdb.decommission_reconcile", ttl=1800): async with async_session_factory() as session: summary = await reconcile_decommissioned_scopes( session, limit=limit, ) await session.commit() except LockBusy: return { "checked": 0, "confirmed_closed": 0, "still_pending": 0, "state_advanced": 0, "failed_closures": 0, "reopened_alive": 0, "unreadable": 0, "unsupported": 0, "alerted_recipients": 0, "skipped_over_limit": 0, "errors": [], "skipped_lock_busy": True, } # Contradictions are NOT handler failures — they are findings, already # audited and already alerted, and raising on them would mark the cron # red every day until a human resolves the underlying account. What # DOES deserve red is the sweep being unable to see: if most of what # we tried to check came back unreadable, the reconciler is not # reconciling and its silence would otherwise look like "all clear". checked = int(summary.get("checked") or 0) unreadable = int(summary.get("unreadable") or 0) if checked and unreadable >= max(3, int(checked * 0.5)): detail = "; ".join(str(e) for e in summary.get("errors") or [])[:500] raise RuntimeError( f"cmdb.decommission_reconcile could not read {unreadable} of " f"{checked} in-flight closures; sample: {detail or ''}" ) return summary @register_scheduled_job( key="cmdb.leaver_grant_sweep", display_name="已离职员工云访问 grant 兜底回收", description=( "每日 01:30 扫描所有 employment_status='left' 的人," "把他们名下任何仍为 active / pending 的 cmdb_cloud_access_grants " "通过 driver 一并 revoke。这是 hr.process_advance T+0 强制转让的" "兜底网:历史离职(在 cloud_access_grant 自动 revoke 代码上线前" "就走完离职流程的人,P0546-class)、或者云侧后于离职日才被同步进来的" "drift grant,这里统一清掉。suppress_driver_error=True 保证 Helix" "侧的 revoke 落盘,云侧失败时置 requires_manual_cloud_revoke=True," "下一次 drift sweep 会重试。无 recent_days 截断 —— 老的离职也要扫到。" ), ) async def run_leaver_grant_sweep( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: # ``sweep_active_grants_on_left_persons`` opens its own per-grant # sessions internally — see the function's docstring. The ``db`` # passed in by the scheduled-job runner is therefore unused and we # don't commit on it. The handler signature keeps the standard # shape so the runner's contract is uniform across all crons. from app.domains.cmdb.lifecycle_service import ( sweep_active_grants_on_left_persons, ) summary = await sweep_active_grants_on_left_persons() errors = summary.get("errors") or [] if errors: detail = "; ".join(str(e) for e in errors)[:500] or "" raise RuntimeError( f"cmdb.leaver_grant_sweep completed with errors: {detail}" ) return summary @register_scheduled_job( key="cmdb.leaver_iam_residue_sweep", display_name="离职人员云 IAM 权限残留对账", description=( "每日 02:10 用云端 IAM 镜像 (cmdb_iam_users) 对账 HR 离职名单:" "已过最后工作日的人若在云上仍持有策略或用户组,登记为治理待办并" "通知 IT 管理员。跑在 cmdb.leaver_grant_sweep (01:30) 之后," "这样常规回收链路已经有过机会。\n\n" "为什么不能用 grant 表来判断:离职回收会把 grant 置 revoked,而" "所有既有清扫都只看 active / pending —— revoked 行没有任何清扫" "会再看第二眼。褚杰 2026-06-01 的四条 CAM 策略是组继承的,当时的" "驱动 DetachUserPolicy 是空操作(6-02 才修),Helix 记录显示已回收," "云上活了两个半月没人发现。云端镜像是唯一能戳穿这种「成功的空操作」" "的证据。\n\n" "只读:本任务不做任何云端写操作,清理动作在管理端人工确认后执行。" ), weight="heavy", # 全量扫 cmdb_iam_users + 通知扇出 ) async def run_leaver_iam_residue_sweep( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: # Opens its own session (and commits) internally, same contract as # ``run_leaver_grant_sweep``; the runner-supplied ``db`` is unused. from app.core.distributed_lock import LockBusy, distributed_lock from app.domains.cmdb.leaver_iam_residue import sweep_leaver_iam_residue try: async with distributed_lock("cmdb.leaver_iam_residue_sweep", ttl=900): summary = await sweep_leaver_iam_residue() except LockBusy: # Zero-shape summary so the failure notifier stays quiet — a # concurrent run is doing the work, this is not a failure. return { "detected": 0, "opened": 0, "still_open": 0, "auto_resolved": 0, "admins_notified": 0, "errors": [], "skipped_lock_busy": True, } errors = summary.get("errors") or [] if errors: detail = "; ".join(str(e) for e in errors)[:500] or "" raise RuntimeError( f"cmdb.leaver_iam_residue_sweep completed with errors: {detail}" ) return summary @register_scheduled_job( key="cmdb.manual_revoke_retry_sweep", display_name="云端权限手动 revoke 重试", description=( "每小时扫一遍 status=revoked 且 detail.requires_manual_cloud_revoke=true " "的 cmdb_cloud_access_grants:再次调 driver.revoke_access。云上确实" "已经回收(已 revoked / already_absent)就把标记清掉;还是失败就保留" "等下一轮。修复 _revoke_grant_locked(suppress_driver_error=True) 之后" "云侧永久残留的漏洞 —— 离职员工权限不应该永远挂在云上。" "min_age_seconds=3600 给原始 revoke 一小时窗口避免抖动直接重打。" ), ) async def run_manual_revoke_retry_sweep( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.domains.cmdb.lifecycle_service import ( sweep_pending_manual_cloud_revokes, ) # Round-2 review follow-up (SS1 P2-1): a misconfigured admin # could write ``min_age_seconds=-1`` (or 0) in the cron config # JSON; both are truthy-falsy edge cases that would bypass the # 1 h debounce window (-1 → cutoff lands in the future → every # revoked grant immediately re-tried). ``max(60, ...)`` floors at # 1 minute so the debounce never disappears even on operator # typo. Upper bound left unbounded by design — an explicit large # min_age (e.g. 86400) is a legitimate "quiet day" override. raw_min_age = int(config.get("min_age_seconds") or 3600) min_age = max(60, raw_min_age) summary = await sweep_pending_manual_cloud_revokes(min_age_seconds=min_age) # Pre-launch hardening + Round-1 review (Review-D P1-3): hourly cron # means any per-grant noise pages 24× a day. Raise ONLY on systemic # failure (every scanned grant errored OR every grant still failing # AND zero cleared). Per-grant transient failures (cloud propagation # delay, momentary IAM throttle) stay in ``summary.errors[]`` for # forensics without paging on-call every hour. errors = summary.get("errors") or [] scanned = int(summary.get("scanned") or 0) cleared = int(summary.get("cleared") or 0) if scanned > 0 and cleared == 0 and len(errors) >= scanned: detail = "; ".join(str(e) for e in errors)[:500] or "" raise RuntimeError( f"cmdb.manual_revoke_retry_sweep systemic failure: " f"{len(errors)} errors of {scanned} scanned, 0 cleared; " f"sample: {detail}" ) return summary # ── Software monthly billing — auto-close ───────────────────────── # # NOTE (2026-05-09): the ``cmdb.scope_inactivity_sweep`` cron used to # live here. It was removed because the ``last_login_at`` data the # sweep depended on is not reliably retrievable across our cloud # providers — the AWS Identity Center / Azure portal SSO logins we # wanted to mirror don't expose a query that would let us populate # ``CmdbCloudAccessGrant.last_login_at`` consistently. Stage-0 # notifications were firing on stale / NULL data and generating # false "60+ days inactive" alerts to scope owners. # # Going forward the platform relies solely on the expiry-based flow # (``cmdb.access_expiry_sweep``): a grant has an ``expires_at``, we # notify at T-14 / T-7 / T-1, the user renews if still needed, and # at T+0 the sweep auto-revokes (prod scopes excepted — held + owner # notified). Migration 138 deletes the orphaned ``scheduled_jobs`` # row so the runner stops surfacing "handler not registered" errors. # Lifecycle-service helpers (``sweep_inactive_grants``, # ``refresh_person_last_login_in_scope``, # ``disable_grant_due_to_inactivity``) remain in # :mod:`app.domains.cmdb.lifecycle_service` for the existing test # suite — they're no longer wired to any scheduled job. @register_scheduled_job( key="software_billing.auto_close", display_name="软件月账单自动关账", description=( "每月 1 号凌晨把上个月关账。对每条在上月有过活跃状态的 " "software_seat_allocation 做快照(价格/cycle/软件名/plan 名/" "用户名/部门/reason 全部冻结)。一次性收费只在分配当月出现。" "幂等:如果上月已被管理员手动关过账,本次直接跳过。" ), ) async def run_software_billing_auto_close( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from datetime import timedelta as _td from zoneinfo import ZoneInfo from app.core.database import async_session_factory from app.domains.asset import billing_service # Anchor "today" to Asia/Shanghai (the scheduled_jobs row's tz) so # a cron scheduled for "03:00 on the 1st" consistently targets # last month even when the worker container runs in UTC. Same # pattern as run_billing_sync above. explicit_period = str(config.get("period") or "").strip() if explicit_period: target_period = explicit_period else: today_sh = datetime.now(ZoneInfo("Asia/Shanghai")).date() first_of_this = today_sh.replace(day=1) last_month = (first_of_this - _td(days=1)).replace(day=1) target_period = last_month.strftime("%Y-%m") async with async_session_factory() as session: # ``close_month`` raises PERIOD_ALREADY_CLOSED if the admin got # there first — treat that as a success-shape no-op rather than # bubbling the exception (which would flag the job as failed). # PERIOD_NOT_ENDED shouldn't fire in the normal cron path (we # compute last month) but guard anyway so a hand-typed config # with a current-or-future period doesn't page an oncall. try: result = await billing_service.close_month( session, target_period, actor_id=None, ) except Exception as exc: # noqa: BLE001 code = getattr(exc, "code", None) if code == "PERIOD_ALREADY_CLOSED": await session.rollback() return { "status": "skipped", "period": target_period, "reason": "already_closed", } if code == "PERIOD_NOT_ENDED": await session.rollback() return { "status": "skipped", "period": target_period, "reason": "period_not_ended", } await session.rollback() logger.exception("software_billing.auto_close failed") # Must raise so failure_notifier fires — silent monthly close # failure = 30 days of stale billing before anyone notices. raise RuntimeError( f"software_billing.auto_close failed for {target_period}: {exc}" ) from exc # Audit — mirrors what the manual close endpoint writes, but # with actor_type="system" so the trail makes it clear this # was the scheduler and not a user. from app.audit.service import AuditService audit = AuditService(session) await audit.log( trace_id=str(uuid.uuid4()), actor_type="system", actor_user_id=None, action="software_billing.close", resource_type="software_monthly_bill", resource_id=target_period, status="success", detail={ "period": target_period, "line_count": result["line_count"], # None on a mixed-currency period — mirror the manual # close endpoint: keep the None (not the string "None") # and let by_currency carry the authoritative breakdown. "total_amount": ( str(result["total_amount"]) if result.get("total_amount") is not None else None ), "by_currency": result.get("by_currency", {}), "currency_warning": result.get("currency_warning", False), "trigger": "auto_close", }, ) await session.commit() return { "status": "closed", "period": target_period, "line_count": result["line_count"], "total_amount": ( str(result["total_amount"]) if result.get("total_amount") is not None else None ), "by_currency": result.get("by_currency", {}), "currency_warning": result.get("currency_warning", False), } # ── Personal access token expiry warning ────────────────────────── @register_scheduled_job( key="auth.pat_expiry_warning", display_name="API Token 即将到期提醒", description=( "每日扫 auth_personal_access_tokens,对距离过期 ≤ 7 天且未撤销、" "且尚未发过提醒(expiry_notified_at IS NULL)的 token 推送一条" "inbox 通知给持有人,并把 expiry_notified_at 置为当前时间避免重复发。" "运行时间建议每日 09:00 — 与 hire_day_it_notify 同班次,不冲突。" ), ) async def run_pat_expiry_warning( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from datetime import timedelta from app.auth.tokens.models import PersonalAccessToken from app.core.database import async_session_factory from app.identity.person_models import IdentityPerson from app.notifications.service import ( NotificationEnvelope, NotificationService, ) warn_window = timedelta(days=int(config.get("warn_days", 7))) now = datetime.now(UTC) cutoff = now + warn_window notified = 0 skipped_inactive = 0 errors: list[str] = [] async with async_session_factory() as session: rows = ( await session.execute( select(PersonalAccessToken).where( PersonalAccessToken.revoked_at.is_(None), PersonalAccessToken.expiry_notified_at.is_(None), PersonalAccessToken.expires_at.is_not(None), PersonalAccessToken.expires_at <= cutoff, PersonalAccessToken.expires_at > now, ) ) ).scalars().all() notif = NotificationService(session) for row in rows: person = await session.get(IdentityPerson, row.user_id) if person is None or person.employment_status != "active": # Departed / soft-deleted owner — flag the row as "notified" # so we don't keep scanning it forever; the token itself # will fail to authenticate anyway. row.expiry_notified_at = now skipped_inactive += 1 continue try: # MySQL/TiDB DATETIME columns return naive datetimes; # coerce to UTC-aware before subtracting from `now` (which # IS aware) to avoid TypeError. Helix always writes UTC, # so this is correct, not a fallback. from app.auth.tokens.service import _as_aware_utc expires_aware = _as_aware_utc(row.expires_at) days_left = max(0, (expires_aware - now).days) if expires_aware else 0 env = NotificationEnvelope( recipient=person, title=f"[Helix] API Token 「{row.name}」即将在 {days_left} 天后过期", body_text=( f"你的个人访问令牌 「{row.name}」" f"({row.token_prefix}…)将在 " f"{row.expires_at.isoformat()} 过期。\n\n" f"如需继续使用,请在 Helix /me/api-tokens 创建新 token " f"并替换 AI 客户端中的旧值;不再使用请直接 revoke 当前 token。" ), # Round-42 fix: ``inapp`` is documented as a placeholder # only — the dispatch loop only knows ``feishu`` and # ``email``. The previous channel list left ``inapp`` # un-delivered forever, every attempt mis-classified # the envelope as failed, and after ``max_attempts`` # the row went DEAD even though the email actually # delivered. Use the same channels as the existing # PAT-expiry sibling at line 2652. channels=["feishu", "email"], template_key=None, related_resource_type="personal_access_token", related_resource_id=row.id, ) await notif.dispatch(env, trace_id=str(uuid.uuid4())) row.expiry_notified_at = now notified += 1 except Exception as exc: # noqa: BLE001 # Mark the row notified anyway so a transient SMTP / Feishu # blip doesn't make us re-scan + re-fail every single day # for the rest of the token's life. Missing the warning by # ≤24h is preferable to a perpetual error-row storm that # fills ``errors`` and the audit table. The row's status # is captured in ``errors`` for the operator to review. # (Pre-launch hardening 2026-05-06.) row.expiry_notified_at = now errors.append( f"{row.id}: {format_exception_for_log(exc)}" ) await session.commit() return { "notified": notified, "skipped_inactive_owner": skipped_inactive, "scanned": len(rows), "errors": errors[:10], # keep summary short } # ── Relationship graph sync ────────────────────────────────────── @register_scheduled_job( key="relationships.full_sync", display_name="关系图谱全量同步", description=( "每日 04:30 全量重建 rel_relationships:经理链 / 资产持有人 / " "seat / API key / 软件 owner+IT contact / 邮件组 owners / " "云 server+scope+grant+domain owners,各源独立,只清理本源未触达的边。" ), ) async def run_relationships_full_sync( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.core.database import async_session_factory from app.domains.relationships.sync_service import RelationshipSyncService async with async_session_factory() as session: summary = await RelationshipSyncService(session).run_full_sync() await session.commit() return summary @register_scheduled_job( key="relationships.incremental_sync", display_name="关系图谱增量同步", description=( "每 5 分钟扫最近 60 分钟变更的资源,重新发射对应边,确保图谱与现实只差几分钟。" "不做删除清理(留给每日 full_sync)。Cadence bumped from 30 min to 5 min in " "migration 094 so the admin RelationshipGraphPage doesn't need a manual " "sync button." ), ) async def run_relationships_incremental_sync( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.core.database import async_session_factory from app.domains.relationships.sync_service import RelationshipSyncService lookback = int(config.get("lookback_minutes", 60)) async with async_session_factory() as session: summary = await RelationshipSyncService(session).run_incremental_sync( lookback_minutes=lookback, ) await session.commit() return summary # ── Leaver Blocked SLA escalation tick ──────────────────────────── @register_scheduled_job( key="leaver.escalation_tick", slo=SLO_LEAVER_ESCALATION_TICK, display_name="Leaver Blocked SLA 升级 tick", description=( "每 30 分钟扫所有过 leaver_date 仍卡在 blocked 的离职流程," "按 first_blocked_at 计算 SLA: 24h 通知 manager 链, 48h 通知 IT, " "72h 创 critical compliance_incident. 同时清理过期的 dual-sign override." ), ) async def run_leaver_escalation_tick( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.domains.leaver_blocked.sla_service import LeaverBlockedSlaService async def _work(session: AsyncSession) -> dict[str, Any]: service = LeaverBlockedSlaService(session) summary = await service.run_tick() # The lock-retry helper commits the fresh session after _work # returns — no explicit commit here (matches sibling ticks). return summary # Wrap in the fresh-session lock-retry helper, matching the sibling # leaver/offboarding escalation ticks. run_tick only levels SLA tiers # up (never down) and every notification is OffboardingReminderLog- # deduped, so restarting the whole tick on a fresh session after a # transient TiDB error (deadlock 1213 / lost-connection 2013) is # idempotent. Previously this used a bare session with no retry, so a # transient error dropped the whole SLA tick until the next cadence. return await _run_in_fresh_session_with_lock_retry( _work, label="leaver.escalation_tick" ) # ── Software seat — auto-revoke pending self-cancellations ──────── @register_scheduled_job( key="software_seat.auto_revoke_pending_cancels", display_name="软件坐席 — 次月自动取消", description=( "每月 1 号 00:05 UTC 扫描所有 status='active' 且 " "cancel_at_period_end <= NOW() 的软件坐席,运行标准 revoke 流程。" "每条 revoke 写一条审计日志(actor 是当初申请的员工,而非系统)、" "给员工发『取消已生效』通知、给软件 owner 发『一个付费坐席已释放』" "通知。幂等:已 revoked 的行被 status='active' 过滤掉,重跑同一 tick " "不会重复出账或重复发通知。" ), ) async def run_software_seat_auto_revoke_pending_cancels( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: """Sweep due self-cancellations and run the standard revoke path. The cron line in the seed migration fires this once per month at 00:05 UTC on the 1st. The handler also accepts an explicit ``cutoff`` ISO datetime in ``config`` so an admin can re-run a delayed sweep manually from /admin/scheduled-jobs without waiting for the next monthly tick. """ from app.audit.service import AuditService from app.core.coordinators.provisioning_coordinator import ( enqueue_revoke_for_seat, ) from app.core.database import async_session_factory from app.domains.asset.models import ( SoftwareLicensePlan, SoftwareSeatAllocation, ) from app.domains.provisioning.models import SOURCE_MANUAL from app.identity.person_models import IdentityPerson from app.notifications.service import ( NotificationEnvelope, NotificationService, ) explicit_cutoff = str(config.get("cutoff") or "").strip() if explicit_cutoff: try: cutoff = datetime.fromisoformat(explicit_cutoff) except ValueError as exc: # A malformed manual-override cutoff (admin typo on a backfill) # must NOT silently fall back to ``now()`` — that would run the # revoke sweep against the wrong (current) window instead of the # intended historical one. Raise so the failure surfaces via the # dedup-aware scheduled-job failure notifier rather than quietly # acting on a different cutoff than the operator asked for. raise RuntimeError( f"invalid 'cutoff' config {explicit_cutoff!r}: {exc}" ) from exc else: cutoff = datetime.now(UTC) revoked_ids: list[str] = [] notify_skipped: list[str] = [] async with async_session_factory() as session: # Order by ``allocated_at ASC`` so the cascade source-vs-sibling # pick is deterministic across runs (oldest seat wins as # source). Round 2 review caught the non-determinism: without # ORDER BY, the user's "已取消订阅" notification could be # labelled with the add-on plan ("Jira Tempo Timesheets") # whenever the DB happened to return it first — confusing if # the user originally clicked Cancel on Jira Software. With # ORDER BY allocated_at, the base plan (allocated first in # PingCAP's seed history) consistently lands as source. rows = (await session.execute( select(SoftwareSeatAllocation, SoftwareLicensePlan) .join( SoftwareLicensePlan, SoftwareSeatAllocation.plan_id == SoftwareLicensePlan.id, ) .where( SoftwareSeatAllocation.status == "active", SoftwareSeatAllocation.cancel_at_period_end.is_not(None), SoftwareSeatAllocation.cancel_at_period_end <= cutoff, ) .order_by( SoftwareSeatAllocation.allocated_at.asc(), SoftwareSeatAllocation.id.asc(), # tiebreaker ) )).all() if not rows: return {"status": "ok", "revoked": 0} audit = AuditService(session) notif = NotificationService(session) # Resolve every employee + software-owner in one batch so the # per-row notification dispatch doesn't issue N+1 queries. person_ids: set[uuid.UUID] = set() for alloc, _plan in rows: person_ids.add(alloc.user_id) if alloc.cancel_requested_by: person_ids.add(alloc.cancel_requested_by) people_by_id: dict[uuid.UUID, IdentityPerson] = {} if person_ids: people_rows = (await session.execute( select(IdentityPerson).where(IdentityPerson.id.in_(person_ids)) )).scalars().all() people_by_id = {p.id: p for p in people_rows} enqueue_failures: list[dict[str, str]] = [] cascade_skipped = 0 for alloc, plan in rows: # Product-access cascade dedup. The first iteration that # hits a Jira-style allocation will call # ``enqueue_revoke_for_seat`` → coordinator cascade-flips # sibling allocations to ``revoked`` IN THE SAME session # (same ORM instances we're iterating). By the time we # reach a sibling in a later iteration of this cron tick, # its in-memory ``status`` is already ``revoked``. Without # this skip we'd: (a) call enqueue again (harmless, dedups # at the (user, integration, product_key) granularity but # spends a query), (b) re-stamp ``status='revoked'`` (no- # op), (c) write a SECOND audit row for the same vendor- # side cancel event, and (d) DM the user a SECOND "your X # was cancelled" notification — once per cascaded plan. # The source seat's audit + notify covers the whole # cascade, so skip the duplicates here. if alloc.status != "active": cascade_skipped += 1 continue # Carry the original requester forward as the audit actor — # the row's audit trail should say "Employee X cancelled # back on date Y; system applied at month boundary", not # "system spontaneously revoked". requester_id = alloc.cancel_requested_by or alloc.user_id # Pre-launch P0 fix (Round 45, CR-302): enqueue FIRST, flip # status only on success. Previously the cron flipped # alloc.status='revoked' BEFORE enqueue_revoke_for_seat, # and on transient enqueue failure (DB blip, connector lookup # error) only logged a warning + still committed → seat # marked revoked in Helix while user retains vendor-side # access. Now: if enqueue raises, leave the row untouched # so next month's tick retries cleanly, and surface a # failure status from the handler so the failure notifier # pages ops. task_id: uuid.UUID | None = None try: # Per-row SAVEPOINT — enqueue_revoke_for_seat does db.add + # flush inside the coordinator (provisioning_coordinator). # A single bad enqueue (FK violation, 1205 lock-wait, etc.) # would otherwise leave the shared session in rolled-back- # pending state, poisoning every later row's writes AND the # terminal session.commit() — so even the rows already in # ``revoked_ids`` never persist. The savepoint isolates the # failure to this row; the broad except below records it and # the loop continues on a usable session. async with session.begin_nested(): task_id = await enqueue_revoke_for_seat( session, seat=alloc, trace_id=f"auto_revoke_self_cancel:{alloc.id}", actor_user_id=requester_id, source=SOURCE_MANUAL, enforce_offboarding_policy=False, # Round-55 P0: for the product_access cascade path # (Jira / Confluence), the coordinator flips every # sibling allocation to ``status='revoked'`` BEFORE # the dispatcher runs. If the vendor call later # fails (Atlassian 5xx, network timeout) and # ``revert_seat_on_failure`` is NOT set, the # dispatcher's failure side-effect short-circuits # — siblings stay revoked in Helix while the # vendor still holds the product. The direct admin # revoke (asset/router.py:5067) correctly opts into # the revert path; the cron-driven self-cancel was # missing it, leaving silent multi-row inconsistency # on every transient dispatcher hiccup. Mirror the # admin path here so a cron-driven cancel is just # as self-healing as a manual one. revert_seat_on_failure=True, ) except OperationalError: # Deadlock (1213) rolls back the whole server-side # transaction — no savepoint saves it. Propagate so the # fresh-session lock-retry wrapper restarts the idempotent # run cleanly. raise except Exception as exc: # noqa: BLE001 logger.warning( "enqueue_revoke_for_seat failed for auto self-cancel %s: %s", alloc.id, exc, ) enqueue_failures.append({ "alloc_id": str(alloc.id), "error": str(exc) or type(exc).__name__, }) # Skip status flip + audit + notify for this row — # leaves it eligible for next month's retry. continue # Pre-launch P0 fix (Round 45, CR-307 follow-up): snapshot # the cancel_* metadata BEFORE clearing so the audit row # carries the full forensic context (who requested, # when, why) — without this, every monthly auto-revoke # audit row loses the three fields a billing-dispute # investigator needs. prior_cancel_at_period_end = alloc.cancel_at_period_end prior_cancel_requested_at = alloc.cancel_requested_at prior_cancel_requested_by = alloc.cancel_requested_by prior_cancel_reason = alloc.cancel_reason alloc.status = "revoked" # Stamp the EXACT effective boundary the user chose # (``cancel_at_period_end`` = first instant of the next billing # month), NOT ``cutoff`` (= now ≈ 00:05 UTC on the 1st, i.e. a # few minutes INTO the new period). With ``cutoff`` the seat's # ``revoked_at`` lands inside period M+1, so the billing overlap # predicate charges a full extra un-prorated month — the exact # next-cycle charge this feature promises to avoid. Pairs with # the strict lower bound in ``billing_service.period_overlap_clauses`` # (revoked_at > period_start) so a seat whose access ends exactly # at a period start has no billable window in that period. alloc.revoked_at = prior_cancel_at_period_end or cutoff # Clear cancel_* metadata on transition so the row's history # doesn't claim "pending cancel" forever and the # has_pending_cancel filter doesn't surface revoked rows # (Round 45, CR-307). alloc.cancel_at_period_end = None alloc.cancel_requested_at = None alloc.cancel_requested_by = None alloc.cancel_reason = None await audit.log( trace_id=f"auto_revoke_self_cancel:{alloc.id}", actor_type="user", actor_user_id=requester_id, action="software_seat.auto_revoke_self_service", resource_type="software_seat_allocation", resource_id=str(alloc.id), status="success", detail={ "plan_id": str(alloc.plan_id), "target_user_id": str(alloc.user_id), "cancel_requested_at": ( prior_cancel_requested_at.isoformat() if prior_cancel_requested_at else None ), "cancel_at_period_end": ( prior_cancel_at_period_end.isoformat() if prior_cancel_at_period_end else None ), "cancel_requested_by": ( str(prior_cancel_requested_by) if prior_cancel_requested_by else None ), "cancel_reason": prior_cancel_reason, "provisioning_task_id": str(task_id) if task_id else None, "trigger": "auto_revoke_pending_cancels", }, ) holder = people_by_id.get(alloc.user_id) sw_name = plan.software.name if plan.software else "—" if holder is not None and holder.employment_status == "active": try: await notif.dispatch(NotificationEnvelope( recipient=holder, title=f"已取消订阅: {sw_name}", body_text=( f"你之前申请的『下月取消』已生效:\n" f" · 软件: {sw_name}\n" f" · 套餐: {plan.name or '—'}\n" f" · 取消时间: {cutoff:%Y-%m-%d %H:%M UTC}\n" ), channels=["feishu", "email"], related_resource_type="software_seat_allocation", related_resource_id=alloc.id, )) except Exception as exc: # noqa: BLE001 logger.warning( "notify holder for self-cancel %s failed: %s", alloc.id, exc, ) notify_skipped.append(str(alloc.id)) else: notify_skipped.append(str(alloc.id)) revoked_ids.append(str(alloc.id)) await session.commit() if enqueue_failures: # Raise so the failure-notifier pages ops. The successful rows # in this tick have already been committed and audited; the # remaining failures will retry next tick. raise RuntimeError( f"auto_revoke_pending_cancels: {len(enqueue_failures)} " f"enqueue failure(s); first sample: " f"{enqueue_failures[0]} ({len(revoked_ids)} succeeded)" ) return { "status": "ok", "revoked": len(revoked_ids), "revoked_ids": revoked_ids[:50], # cap to keep run-log row sane "notify_skipped_count": len(notify_skipped), # Visibility into how many rows got handled by the cascade # path (Jira-style product-access revokes) — surfaced so the # run-log shows e.g. "scanned 4 due rows, revoked 1 source + # skipped 3 cascade-handled siblings". "cascade_skipped_count": cascade_skipped, } # ── DNS owner recertification (90-day "still in use?" nag) ───────── @register_scheduled_job( key="cmdb.dns_owner_recertify", display_name="DNS Owner 季度保鲜", description=( "扫描所有 DNS 记录,若 owner 距上次确认已超过配置的天数(默认 90," "可在 platform_settings.cmdb.dns.recertify_period_days 修改),按" " owner 聚合发送一条飞书提醒,首次运行时 NULL 视作 due。仅写日志" "和飞书消息,不修改任何 DNS 记录或 owner。" ), ) async def run_dns_owner_recertify( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from datetime import UTC as _UTC from datetime import datetime as _dt from datetime import timedelta as _td from sqlalchemy import and_ as _and from sqlalchemy import or_ as _or from sqlalchemy import select as _select from app.core.database import async_session_factory from app.domains.cmdb.models import CmdbDnsRecord from app.domains.cmdb.network_notifications import ( network_user_notifications_enabled, ) from app.identity.person_models import IdentityPerson from app.integrations.compat import get_feishu_config from app.integrations.service import get_platform_setting # Kill switch (migration 174) — keep the cron silent until the # operator flips ``network.user_notifications_enabled`` to true. if not await network_user_notifications_enabled(db): return { "status": "skipped", "reason": "network.user_notifications_enabled is false", } period_days = int( config.get("recertify_period_days") or await get_platform_setting( "cmdb.dns.recertify_period_days", default=90, ) or 90, ) threshold = _dt.now(_UTC) - _td(days=period_days) # Per-record nag dedup window — minimum gap between two nags for # the same record. Without this every cron tick re-DM'd the # owner with the same record list. Default 30 days: owner gets # at most one DM per record per month even if they ignore it. nag_dedup_days = int( config.get("recertify_nag_dedup_days") or await get_platform_setting( "cmdb.dns.recertify_nag_dedup_days", default=30, ) or 30, ) nag_dedup_threshold = _dt.now(_UTC) - _td(days=nag_dedup_days) # Round-30: first-observation suppression. The very first time # this cron runs after a deploy (or after operator enables the # ``network.user_notifications_enabled`` kill switch), every record # with ``last_owner_review_at IS NULL AND last_recertify_nag_at IS # NULL`` is "due" — that can be hundreds of records per owner and # carpet-bombs every employee with their full inventory. Suppress # the first observation: stamp ``last_recertify_nag_at = now()`` so # the next cron tick (after ``nag_dedup_days``) emits the actual # nag. Operators get a chance to fix any data issues (mis-mapped # owners) before owners get the notice. # # The toggle defaults true; an operator can flip it false via # ``cmdb.dns.recertify_suppress_first_obs`` to force a full sweep # on the first run (e.g. after a backfill). first_obs_suppress = bool( config.get("recertify_suppress_first_obs") if "recertify_suppress_first_obs" in config else await get_platform_setting( "cmdb.dns.recertify_suppress_first_obs", default=True, ) ) async with async_session_factory() as session: # Round-28: also exclude owners whose termination_date is within # the pre-leaver warn window (default 7 days). Pre-leaver # handover already DMs those owners with the full inventory; # the recertify cron would double-up the same records. from datetime import date as _date_t from datetime import timedelta as _td_t upcoming_leaver_ids = list((await session.execute( _select(IdentityPerson.id).where(_and( IdentityPerson.termination_date.is_not(None), IdentityPerson.termination_date <= _date_t.today() + _td_t(days=7), )) )).scalars().all()) recertify_query = _select(CmdbDnsRecord).where(_and( CmdbDnsRecord.deleted_at.is_(None), CmdbDnsRecord.owner_person_id.is_not(None), _or( CmdbDnsRecord.last_owner_review_at.is_(None), CmdbDnsRecord.last_owner_review_at < threshold, ), # Skip records nagged within the dedup window. _or( CmdbDnsRecord.last_recertify_nag_at.is_(None), CmdbDnsRecord.last_recertify_nag_at < nag_dedup_threshold, ), )) if upcoming_leaver_ids: recertify_query = recertify_query.where( CmdbDnsRecord.owner_person_id.notin_( [str(pid) for pid in upcoming_leaver_ids], ), ) rows = list((await session.execute(recertify_query)).scalars().all()) if not rows: return {"due_records": 0, "owners_nagged": 0} # Round-30: first-observation carpet-bomb suppression. Split # rows into two groups: "first observation" (both timestamps # NULL) and "real recertify backlog" (review_at is set OR the # record has been nagged before but the dedup window elapsed). # First-observation rows get their ``last_recertify_nag_at`` # stamped without a DM — the next cron tick will fire them # against the new owner via the normal path. Real backlog rows # are routed into ``by_owner`` for DM dispatch as before. first_obs_rows: list[CmdbDnsRecord] = [] backlog_rows: list[CmdbDnsRecord] = [] for r in rows: if ( first_obs_suppress and r.last_owner_review_at is None and r.last_recertify_nag_at is None ): first_obs_rows.append(r) else: backlog_rows.append(r) if first_obs_rows: await session.execute( CmdbDnsRecord.__table__.update() .where(CmdbDnsRecord.id.in_([r.id for r in first_obs_rows])) .values(last_recertify_nag_at=_dt.now(_UTC)), ) await session.commit() if not backlog_rows: return { "due_records": len(rows), "owners_nagged": 0, "first_observation_suppressed": len(first_obs_rows), } # Group by owner — only backlog rows hit Feishu. by_owner: dict[str, list[CmdbDnsRecord]] = {} for r in backlog_rows: by_owner.setdefault(str(r.owner_person_id), []).append(r) # Resolve owner emails in one round-trip. owner_persons = list((await session.execute( _select(IdentityPerson).where( IdentityPerson.id.in_(list(by_owner.keys())), IdentityPerson.employment_status == "active", ), )).scalars().all()) person_by_id = {str(p.id): p for p in owner_persons} cfg = await get_feishu_config() app_id = str(cfg.get("app_id") or "") app_secret = str(cfg.get("app_secret") or "") if not app_id or not app_secret: # Round-28: raise instead of silently returning ``status=ok``. # The kill switch (network.user_notifications_enabled) is the # intentional silence knob; missing Feishu credentials is a # misconfiguration that should page operators. Returning ``ok`` # here landed scheduled_job_runs.status='ok' and the failure # notifier stayed quiet — exactly the silent-misconfig family # the user's memory.feedback_handler_status_must_raise rule # codifies. The kill switch already returned ``skipped`` above, # so an operator who wants silence has a non-error path. raise RuntimeError( "cmdb.dns_owner_recertify: Feishu app_id/app_secret missing " f"— cannot send nags to {len(by_owner)} owners. Configure " "Feishu integration or disable the kill switch." ) from app.approval.feishu_client import FeishuApprovalClient client = FeishuApprovalClient(app_id=app_id, app_secret=app_secret) nagged = 0 send_failures = 0 # Reopen a session to stamp ``last_recertify_nag_at`` after each # successful DM. The fetch-session above was already closed; this # one batches the stamp UPDATEs across all owners with one commit. async with async_session_factory() as stamp_session: for owner_id, recs in by_owner.items(): person = person_by_id.get(owner_id) if not person or not person.work_email: continue try: open_id = await client.resolve_open_id_by_email(person.work_email) except Exception as e: # noqa: BLE001 logger.warning( "dns_owner_recertify: skip %s (no open_id): %s", person.work_email, e, ) send_failures += 1 continue # Plain text — keep dependency on Feishu card schema minimal. # Owner can act via the deep link. lines = [ f"DNS recertify reminder ({len(recs)} record(s) overdue " f">{period_days} days since last confirm)", "", "Open Helix → 我的网络资源 → 我的 DNS to confirm or", "request modification / deletion via self-service.", "", "First 10 records:", ] # Round-30: redact DNS record ``value``. Values frequently # encode internal infra topology (private IPs, VPN gateways, # staging endpoints, MX routing) and end up cached in the # owner's Feishu DM history forever — including if the owner # later leaves and their Feishu account is archived. Owner # has the deep link below to view full detail inside Helix # where access is gated. (Pre-launch security audit R1-C2.) for r in recs[:10]: lines.append( f" - {r.name}.{r.zone_name} {r.record_type}" ) if len(recs) > 10: lines.append(f" ... and {len(recs) - 10} more") text = "\n".join(lines) try: await client.send_text_to_user(open_id=open_id, text=text) except Exception as e: # noqa: BLE001 logger.warning( "dns_owner_recertify: send to %s failed: %s", person.work_email, e, ) send_failures += 1 continue nagged += 1 # Stamp every nagged record so the next cron tick skips # them until the dedup window elapses. Without this we # re-DM the same list every cron tick — the exact bug # the pre-launch audit flagged as BLOCKER B5. await stamp_session.execute( CmdbDnsRecord.__table__.update() .where( CmdbDnsRecord.id.in_([r.id for r in recs]), ) .values(last_recertify_nag_at=_dt.now(_UTC)), ) await stamp_session.commit() # If every owner failed to receive a DM AND there were owners to # nag, raise so failure_notifier surfaces the broken pipeline. if nagged == 0 and send_failures > 0: raise RuntimeError( "cmdb.dns_owner_recertify: every Feishu DM failed " f"(failures={send_failures}). Check Feishu credentials + " "open_id resolution." ) return { "due_records": len(rows), "owners_nagged": nagged, "send_failures": send_failures, "period_days": period_days, "nag_dedup_days": nag_dedup_days, "first_observation_suppressed": len(first_obs_rows), } # ── DNS pre-leaver handover reminder (T-7 .. T-1) ────────────────── @register_scheduled_job( key="cmdb.dns_pre_leaver_handover", display_name="DNS 离职前转移提醒", description=( "扫描所有 termination_date 在未来 N 天内的人员(默认 7)," "若其名下持有 DNS 记录或域名,飞书通知本人尽快走自助流程" " delete 或申请修改。未在离职日处理的将由 hr_lifecycle" " 的 T+0 force-transfer 兜底转给 manager 链。" ), ) async def run_dns_pre_leaver_handover( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from datetime import timedelta as _td from sqlalchemy import and_ as _and from sqlalchemy import select as _select from app.core.database import async_session_factory from app.core.working_hours import resolve_timezone from app.domains.cmdb.models import ( CmdbCdnDomain, CmdbDnsRecord, CmdbDomain, CmdbLoadBalancer, ) from app.domains.cmdb.network_notifications import ( network_user_notifications_enabled, ) from app.domains.offboarding.models import ( OffboardingProcess, OffboardingReminderLog, ) from app.identity.person_models import IdentityPerson from app.integrations.compat import get_feishu_config # Kill switch (migration 174) — keep the cron silent until the # operator flips ``network.user_notifications_enabled`` to true. if not await network_user_notifications_enabled(db): return { "status": "skipped", "reason": "network.user_notifications_enabled is false", } warn_days = int(config.get("warn_days") or 7) # Anchor the coarse scan window to the UTC calendar date. The old # ``date.today()`` read the WORKER-LOCAL date, which is a third clock # on top of the two this handler already reasons about (UTC and each # recipient's own zone) — and the comment on the bucket math below # claimed the window was UTC when it was not. UTC is the right anchor # because it is the one clock that does not move with wherever the # worker happens to be deployed; the per-person bucket arithmetic # further down re-derives the recipient's local date on its own. today = datetime.now(UTC).date() horizon = today + _td(days=warn_days) # Bucket today's nags into one of three reminder kinds so a leaver # gets at most three DMs across their last 7 days. The bucket is # used as part of ``OffboardingReminderLog.reminder_kind`` so the # cron's "at most once per bucket per process" semantics carries # idempotency across multiple ticks per day. def _bucket(days_left: int) -> str: if days_left <= 1: return "dns_handover_T-1" if days_left <= 3: return "dns_handover_T-3" return "dns_handover_T-7" summary: dict[str, Any] = { "scanned_leavers": 0, "leavers_with_assets": 0, "notified": 0, "deduped_already_nagged": 0, "skipped_no_email": 0, "send_failures": 0, "warn_days": warn_days, } async with async_session_factory() as session: # Round-28: include ``leaver_pending`` in the scan. HR confirms # offboarding on day T-N and flips employment_status from # ``active`` → ``leaver_pending``; the prior ``employment_status # == "active"`` filter silently excluded the exact window where # reminders matter most. Both states get the same pre-leaver # DM; the T+0 force-transfer in advance_service.py handles the # actual handover regardless of state. leavers = list((await session.execute( _select(IdentityPerson).where(_and( IdentityPerson.termination_date.is_not(None), IdentityPerson.termination_date >= today, IdentityPerson.termination_date <= horizon, IdentityPerson.employment_status.in_( ("active", "leaver_pending"), ), )), )).scalars().all()) summary["scanned_leavers"] = len(leavers) if not leavers: return summary leaver_ids = [str(p.id) for p in leavers] dns_rows = list((await session.execute( _select(CmdbDnsRecord).where(_and( CmdbDnsRecord.deleted_at.is_(None), CmdbDnsRecord.owner_person_id.in_(leaver_ids), )), )).scalars().all()) domain_rows = list((await session.execute( _select(CmdbDomain).where(_and( CmdbDomain.deleted_at.is_(None), CmdbDomain.owner_person_id.in_(leaver_ids), )), )).scalars().all()) # Round-28: extend pre-leaver scan to LB + CDN. Previously the # handler promised "DNS 离职前转移提醒" but the T+0 force- # transfer covers ALL four network resource types — a leaver # who only owns load balancers or CDN domains got ZERO advance # notice and discovered after-the-fact that everything had # been auto-transferred to their manager. lb_rows = list((await session.execute( _select(CmdbLoadBalancer).where(_and( CmdbLoadBalancer.deleted_at.is_(None), CmdbLoadBalancer.owner_person_id.in_(leaver_ids), )), )).scalars().all()) cdn_rows = list((await session.execute( _select(CmdbCdnDomain).where(_and( CmdbCdnDomain.deleted_at.is_(None), CmdbCdnDomain.owner_person_id.in_(leaver_ids), )), )).scalars().all()) by_owner_dns: dict[str, list[CmdbDnsRecord]] = {} for r in dns_rows: by_owner_dns.setdefault(str(r.owner_person_id), []).append(r) by_owner_domain: dict[str, list[CmdbDomain]] = {} for d in domain_rows: by_owner_domain.setdefault(str(d.owner_person_id), []).append(d) by_owner_lb: dict[str, list[CmdbLoadBalancer]] = {} for lb in lb_rows: by_owner_lb.setdefault(str(lb.owner_person_id), []).append(lb) by_owner_cdn: dict[str, list[CmdbCdnDomain]] = {} for cdn in cdn_rows: by_owner_cdn.setdefault(str(cdn.owner_person_id), []).append(cdn) # Map each leaver to their offboarding process + dedup-check # against existing reminder log rows. proc_by_person: dict[str, OffboardingProcess] = {} if leaver_ids: procs = list((await session.execute( _select(OffboardingProcess).where( OffboardingProcess.person_id.in_(leaver_ids), ), )).scalars().all()) proc_by_person = {str(p.person_id): p for p in procs} cfg = await get_feishu_config(session=session) app_id = str(cfg.get("app_id") or "") app_secret = str(cfg.get("app_secret") or "") if not app_id or not app_secret: # Round-28: raise instead of silently returning ``status=ok``. # Same reasoning as ``dns_owner_recertify`` — Feishu missing # is misconfiguration. The kill switch is the intentional # silence path; misconfig must surface via failure notifier. raise RuntimeError( "cmdb.dns_pre_leaver_handover: Feishu app_id/app_secret " f"missing — cannot warn {len(leavers)} upcoming leavers. " "Configure Feishu integration or disable the kill switch." ) from app.approval.feishu_client import FeishuApprovalClient client = FeishuApprovalClient(app_id=app_id, app_secret=app_secret) for person in leavers: recs = by_owner_dns.get(str(person.id), []) domains = by_owner_domain.get(str(person.id), []) lbs = by_owner_lb.get(str(person.id), []) cdns = by_owner_cdn.get(str(person.id), []) if not recs and not domains and not lbs and not cdns: continue summary["leavers_with_assets"] += 1 if not person.work_email: summary["skipped_no_email"] += 1 continue # Compute the bucket against the leaver's LOCAL calendar date, # not the global UTC ``today``. For a US/EU leaver near the UTC # day boundary, a UTC-based ``days_left`` is off by one — the # last-day handover DM (and its dedup ``reminder_kind``) would # fire a day early/late. Mirrors advance_service's per-person # ``local_today``. The coarse SQL window above is UTC-anchored # and so lands within one day of every recipient's local date — # in BOTH directions, not only over-scan. For a zone behind UTC # the lower bound drops a leaver on their local last day (they # got the T-1 DM on the prior tick, so nothing is lost); for a # zone ahead of UTC the upper bound admits them a day early and # this bucket math re-files them correctly. Either drift is # tolerable — the bucket is what must be per-person local. local_today = ( datetime.now(UTC).astimezone(resolve_timezone(person)).date() ) days_left = ( (person.termination_date - local_today).days if person.termination_date else 0 ) kind = _bucket(days_left) proc = proc_by_person.get(str(person.id)) # Dedup via OffboardingReminderLog — at most one fire per # (process, kind). Without this, every cron tick during # the T-7..T-1 window re-DM'd the same leaver. Mirrors the # offboarding T-7/T-3/T-1 reminder pattern. if proc is not None: existing = await session.scalar( _select(OffboardingReminderLog).where(_and( OffboardingReminderLog.process_id == proc.id, OffboardingReminderLog.reminder_kind == kind, )), ) if existing is not None: summary["deduped_already_nagged"] += 1 continue else: # Round-27 pre-launch (2026-05-25): no OffboardingProcess # row yet for this future leaver → the dedup log key is # unavailable. Previously the cron skipped the dedup # branch entirely and re-DM'd the leaver on EVERY tick. # Fix: skip the leaver until the offboarding flow # creates their process row (typically within hours of # HRIS flagging the termination_date). Count separately # so the gap is visible in the cron summary. summary.setdefault("skipped_no_process", 0) summary["skipped_no_process"] += 1 continue try: open_id = await client.resolve_open_id_by_email( person.work_email, ) except Exception as e: # noqa: BLE001 logger.warning( "dns_pre_leaver_handover: skip %s (no open_id): %s", person.work_email, e, ) summary["send_failures"] += 1 continue lines = [ f"You leave in {days_left} day(s). The following " "network assets still belong to you:", "", ] if recs: lines.append(f"DNS records ({len(recs)}):") for r in recs[:8]: lines.append( f" - {r.name}.{r.zone_name} {r.record_type}" ) if len(recs) > 8: lines.append(f" ... and {len(recs) - 8} more") if domains: lines.append("") lines.append(f"Domains ({len(domains)}):") for d in domains[:8]: lines.append(f" - {d.domain_name}") # Round-28: parity with DNS — surface the overflow count # so the leaver knows they have more than what the DM # shows. Previously truncated silently after 8 entries. if len(domains) > 8: lines.append(f" ... and {len(domains) - 8} more") if lbs: lines.append("") lines.append(f"Load balancers ({len(lbs)}):") for lb in lbs[:8]: lines.append( f" - {lb.name or lb.lb_id} ({lb.region or '-'})" ) if len(lbs) > 8: lines.append(f" ... and {len(lbs) - 8} more") if cdns: lines.append("") lines.append(f"CDN domains ({len(cdns)}):") for cdn in cdns[:8]: lines.append(f" - {cdn.domain_name}") if len(cdns) > 8: lines.append(f" ... and {len(cdns) - 8} more") lines += [ "", "Open Helix self-service:", " 1. Not needed -> request delete", " 2. Still needed -> ask IT to transfer to a successor", "", "Anything left on your last day will be force-", "transferred to your manager.", ] text = "\n".join(lines) try: await client.send_text_to_user(open_id=open_id, text=text) except Exception as e: # noqa: BLE001 logger.warning( "dns_pre_leaver_handover: send to %s failed: %s", person.work_email, e, ) summary["send_failures"] += 1 continue summary["notified"] += 1 # Record the reminder so subsequent cron ticks within the # same T-N bucket skip this leaver. if proc is not None: session.add(OffboardingReminderLog( process_id=proc.id, reminder_kind=kind, channels=["feishu"], detail={ "dns_records": len(recs), "domains": len(domains), "load_balancers": len(lbs), "cdn_domains": len(cdns), "days_left": days_left, }, )) await session.commit() # Round-27 pre-launch (2026-05-25): raise on partial failure too. # Previously the handler only raised when EVERY DM failed — if # half succeeded (notified=5, send_failures=5) the cron exited # ``ok`` and the successes' dedup log rows blocked re-attempt for # the failed half. Failure_notifier dedup keeps the page-storm # bounded; better to surface partial failures than to permanently # lose them. sent = summary["notified"] failed = summary["send_failures"] if failed > 0 and failed >= max(1, sent): raise RuntimeError( "cmdb.dns_pre_leaver_handover: too many DM failures " f"(notified={sent}, failures={failed}). Investigate " "Feishu open-id resolution or webhook health." ) return summary # Mark module imported so the decorator side-effects fire when the # worker / API process boots and lists handlers. _ = ( run_hris_incremental_sync, run_hris_full_sync, run_hris_absent_person_check, run_overdue_leavers_check, run_revoke_leaver_platform_roles, run_domain_expiry_alert, run_hire_day_it_notify, run_hire_day_feishu_welcome, run_day90_survey_notify, run_process_advance, run_post_leaver_verify, run_daily_digest, run_outbox_retry, run_resource_sync, run_billing_sync, run_provisioning_dispatch, run_scopes_sync, run_roles_sync, run_grants_sync, run_billing_daily_sync, run_pat_expiry_warning, run_access_expiry_sweep, run_scope_idle_sweep, run_leaver_grant_sweep, run_leaver_iam_residue_sweep, run_manual_revoke_retry_sweep, run_software_billing_auto_close, run_software_seat_auto_revoke_pending_cancels, run_leaver_escalation_tick, run_relationships_full_sync, run_relationships_incremental_sync, run_dns_owner_recertify, run_dns_pre_leaver_handover, ) @register_scheduled_job( key="cron.cleanup_stale_running", display_name="清理僵死的运行中行", description=( "扫描 cmdb_sync_jobs 与 scheduled_job_runs 中 status='running' " "但 started_at > 6h 的行,标记为 failed 并写入 'auto-fail: stale " "running row' 错误信息。无此清理时,任何在事务中半路死掉的 " "scheduled job(DB 抖动 / 进程被杀)都会留下 'running' 行,后续 " "的 sync 因 JSON_OVERLAPS 锁判而被跳过 —— 永久阻塞,直到管理员手动 " "执行 SQL 修复(见 cmdb-ops.md §5.4)。" ), ) async def run_cleanup_stale_running( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from sqlalchemy import update as _update from app.core.distributed_lock import is_lock_held from app.domains.cmdb.models import CmdbSyncJob from app.scheduled_jobs.models import ( RUN_STATUS_INTERRUPTED, ScheduledJob, ScheduledJobRun, ) from app.scheduled_jobs.service import single_flight_lock_key cutoff_hours = int(config.get("stale_after_hours") or 6) cutoff = datetime.now(UTC) - timedelta(hours=cutoff_hours) sync_swept = ( await db.execute( _update(CmdbSyncJob) .where( CmdbSyncJob.status == "running", CmdbSyncJob.started_at < cutoff, ) .values( status="failed", finished_at=datetime.now(UTC), # ``error_message`` field doesn't exist on CmdbSyncJob; # capability_results is the JSON sidecar. capability_results={ "_helix_auto_fail": ( f"stale running row swept after {cutoff_hours}h" ), }, ) ) ).rowcount or 0 # ── Orphaned scheduled-job runs ────────────────────────────── # Two ways to know a ``running`` row has no live owner, and the fast # one is the interesting one. # # (a) **Nobody holds the job's single-flight lock.** ``execute_job`` # takes that lock with ``auto_renew=True`` and only releases it # after the terminal write-back, so while a run is genuinely in # flight the Redis key is alive. A ``running`` row whose key is # gone therefore has no owner — a worker was killed hard (OOM, # SIGKILL, container reaped past its stop grace) somewhere the # graceful drain could not reach. This resolves within one tick # of this hourly job instead of six hours. # (b) **The six-hour absolute backstop.** Kept because (a) depends on # Redis: if Redis was unreachable the reaper must still converge, # just slowly. # # Both land on ``interrupted``, not ``failed``. In neither case did # the handler do anything wrong — the process ended. Writing # ``failed`` is what produced 386 of 452 scheduler "failures" in a # 30-day production window and made the red queue unusable. # # Redis trouble fails **closed**: an error resolving liveness leaves # the row alone for the backstop to handle. Wrongly reaping a live # run would let a second copy start against the same vendor API. now_sweep = datetime.now(UTC) grace = timedelta(minutes=int(config.get("orphan_grace_minutes") or 5)) candidates = ( await db.execute( select(ScheduledJobRun.id, ScheduledJobRun.job_id, ScheduledJob.config) .join(ScheduledJob, ScheduledJob.id == ScheduledJobRun.job_id) .where( ScheduledJobRun.status == "running", ScheduledJobRun.started_at < now_sweep - grace, ScheduledJobRun.started_at >= cutoff, ) .limit(500) ) ).all() orphan_ids: list[uuid.UUID] = [] for run_id, job_id, job_config in candidates: lock_key = single_flight_lock_key( job_id, ((job_config or {}) if isinstance(job_config, dict) else {}) .get("lock_epoch"), ) try: if not await is_lock_held(lock_key): orphan_ids.append(run_id) except Exception: # noqa: BLE001 - fail closed, backstop covers it logger.warning( "cleanup_stale_running: could not resolve lock liveness for " "run %s; leaving it to the %dh backstop", run_id, cutoff_hours, exc_info=True, ) orphan_swept = 0 if orphan_ids: orphan_result = await db.execute( _update(ScheduledJobRun) .where(ScheduledJobRun.id.in_(orphan_ids)) .values( status=RUN_STATUS_INTERRUPTED, finished_at=now_sweep, error_message=( "interrupted: no live single-flight lock holder — " "the worker died mid-run (hard kill / OOM)" ), ) ) orphan_swept = int(getattr(orphan_result, "rowcount", 0) or 0) sched_swept = ( await db.execute( _update(ScheduledJobRun) .where( ScheduledJobRun.status == "running", ScheduledJobRun.started_at < cutoff, ) .values( status=RUN_STATUS_INTERRUPTED, finished_at=datetime.now(UTC), error_message=( f"interrupted: still running after {cutoff_hours}h with no " "terminal write-back (worker died mid-run)" ), ) ) ).rowcount or 0 # ProvisioningTask: a worker that picked a task, set status= # ``in_progress``, then crashed (OOM, container kill, SDK hang past # worker timeout) leaves the row stuck forever — ``execute_pending_batch`` # only filters ``status='pending'``. A leaver's seat-revoke task or a # hire's seat-grant task silently drops on the floor: user keeps # access (security incident) or never gets access (operational # incident). Reset to ``pending`` (NOT ``failed``) so the dispatcher # picks them back up and the dead-letter logic still applies via # attempt_count. from app.domains.provisioning.models import ( STATUS_IN_PROGRESS, STATUS_PENDING, ProvisioningTask, ) prov_swept = ( await db.execute( _update(ProvisioningTask) .where( ProvisioningTask.status == STATUS_IN_PROGRESS, ProvisioningTask.started_at.isnot(None), ProvisioningTask.started_at < cutoff, ) .values( status=STATUS_PENDING, attempt_count=ProvisioningTask.attempt_count + 1, error_message=( f"auto-reclaim: stale in_progress task swept after " f"{cutoff_hours}h (worker likely died mid-run)" ), ) ) ).rowcount or 0 await db.commit() if sync_swept or sched_swept or orphan_swept or prov_swept: logger.warning( "cleanup_stale_running: swept cmdb_sync_jobs=%d " "scheduled_job_runs=%d (orphaned=%d) provisioning_tasks=%d", sync_swept, sched_swept, orphan_swept, prov_swept, ) return { "cmdb_sync_jobs_swept": sync_swept, "scheduled_job_runs_swept": sched_swept, "scheduled_job_runs_orphaned": orphan_swept, "orphan_candidates_checked": len(candidates), "provisioning_tasks_reclaimed": prov_swept, "cutoff_hours": cutoff_hours, } @register_scheduled_job( key="cmdb.resource_creator_sweep", display_name="云资源创建者归属(CloudTrail / Audit Log)", description=( "Twice-daily walk over cmdb_cost_facts.resource_id for the " "last 90 days' newly-seen resources without a " "cmdb_resource_creators row + existing error rows still " "under attempt cap. Calls each cloud's audit-log API " "(CloudTrail / Cloud Audit Logs / ActionTrail / Activity " "Log) to resolve 'who created this resource' and caches the " "result. Throttled to 1 req/s per resource to stay under " "provider rate limits (AWS CloudTrail caps at 2 req/s). " "config={\"limit\":N, \"integration_id\":\"...\"} for " "narrowing; defaults to 200 resources per tick. Seeded " "enabled by migration 201 once the four cloud integrations' " "audit-log scopes were confirmed." ), ) async def run_resource_creator_sweep( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: """Resolve newly-discovered resources' creators in batches. Strategy: 1. ``list_unresolved_resources`` returns up to ``limit`` (default 200) (integration_id, resource_id) pairs that need a fresh attribution call. 2. Walk one at a time with a 1 sec inter-call sleep so we stay well under AWS CloudTrail's 2 req/s cap and Aliyun ActionTrail's similar limit. GCP / Azure are higher but the uniform throttle keeps the worker's CPU+network steady. 3. Per-resource failures are captured as ``status='error'`` rows (the service layer does this); the handler counts but does NOT raise on a single failure. The handler raises only when every attempt in the batch failed — that signals a systemic outage (auth issue / driver bug) the failure-notifier should surface. Defaults match the production cron tick cadence: twice-daily (UTC 02:00 + 14:00 = Asia/Shanghai 10:00 + 22:00), 200 resources/tick = 400/day. Pair with the 90-day candidate- freshness filter in :func:`app.domains.cmdb.resource_creator_service.list_unresolved_resources` — the candidate pool is bounded by recently-active resources (~38k initial, ~150/day steady-state), so 400/day amortises the initial sweep in ~95 days and trivially keeps up with daily churn afterwards. An operator can run with ``limit=500`` for a one-shot catch-up via the ``/admin/scheduled-jobs/{id}/run-now`` endpoint. """ import asyncio as _asyncio from app.domains.cmdb.resource_creator_service import ( list_unresolved_resources, resolve_resource_creator, ) limit = int(config.get("limit") or 200) if limit < 1 or limit > 1000: raise ValueError( f"limit must be 1..1000 (got {limit}); use the admin run-" "now endpoint for one-shot catch-up.", ) integration_id_filter = ( str(config.get("integration_id")).strip() if config.get("integration_id") else None ) or None candidates = await list_unresolved_resources( db, integration_id=integration_id_filter, limit=limit, ) if not candidates: return { "status": "ok", "resolved": 0, "not_found": 0, "errors": 0, "window_exceeded": 0, "unsupported": 0, "candidates": 0, } counts = { "resolved": 0, "not_found": 0, "errors": 0, "window_exceeded": 0, "unsupported": 0, } per_resource_throttle_s = float( config.get("throttle_seconds") or 1.0, ) for integ_id, rid in candidates: try: row = await resolve_resource_creator( db, integration_id=integ_id, resource_id=rid, ) except LookupError: # Integration row vanished between candidate-list and # resolve — skip and continue (the resource will be # picked up on a future tick when the integration is # recreated, OR will keep being skipped if it stays # gone). continue except Exception: # noqa: BLE001 # The service layer catches per-resource exceptions and # records them; any leak here is a programming bug, not # a transient failure — let it propagate. raise # **Commit per resolve** (2026-05-25 bugfix). The outer # ``_execute_job_locked`` wraps the entire handler call in # one ``async with session`` and only commits on a clean # return. If THIS handler raises the systemic-failure # ``RuntimeError`` below (e.g. 116/200 aliyun DNS errors), # the session exits WITHOUT commit → every ``_upsert`` # already done in ``resolve_resource_creator`` for the # successful AND error rows rolls back. The first prod # manual-trigger left ``cmdb_resource_creators`` at zero # rows even though 84 resolves succeeded and 116 wrote # ``status='error'`` rows — every one lost. Per-resource # commit makes each row durable the instant it's written, # so the systemic-failure raise no longer destroys the # batch's evidence; the next operator can see WHICH 116 # failed and pivot from per-cloud last_error. # # On commit failure itself (DB connection drop / deadlock / # lock-wait timeout / pool exhaustion): log the precise # context — which integration + resource was being # processed, how many had already committed — then re-raise # so the outer wrapper marks the run ``status='failed'`` # and the failure-notifier fires. Already-committed rows # from prior iterations are durable server-side and # unaffected by the subsequent failure. try: await db.commit() except Exception: # noqa: BLE001 logger.exception( "cmdb.resource_creator_sweep: commit failed after " "resolving integration=%s resource=%s; " "%d/%d candidates already durably committed, " "aborting batch", integ_id, rid, ( counts["resolved"] + counts["not_found"] + counts["errors"] + counts["window_exceeded"] + counts["unsupported"] ), len(candidates), ) raise # Map status → counter. if row.status == "resolved": counts["resolved"] += 1 elif row.status == "not_found": counts["not_found"] += 1 elif row.status == "error": counts["errors"] += 1 elif row.status == "window_exceeded": counts["window_exceeded"] += 1 elif row.status == "unsupported": counts["unsupported"] += 1 # Throttle between calls — see handler docstring. await _asyncio.sleep(per_resource_throttle_s) # Raise when the batch's error rate signals systemic failure so # the cron-failure notifier fires (matches the handler-status- # must-raise convention). Threshold: >50% errors AND at least 5 # errors absolute — small batches (e.g. 3 candidates / 2 errors) # might just be transient network blips per-resource. The # absolute floor of 5 prevents a single bad driver auth from # firing on a 3-resource batch. total = len(candidates) error_threshold_pct = float( config.get("error_threshold_pct") or 0.5, ) error_threshold_abs = int( config.get("error_threshold_abs") or 5, ) if ( total > 0 and counts["errors"] >= error_threshold_abs and counts["errors"] / total >= error_threshold_pct ): raise RuntimeError( f"cmdb.resource_creator_sweep: {counts['errors']}/{total} " f"resolves failed (≥ {error_threshold_pct:.0%} threshold) — " "check driver auth / audit-API scopes / network", ) return { "status": "ok", "candidates": total, **counts, } @register_scheduled_job( key="core.idempotency_cleanup", display_name="幂等记录清理", description=( "每日删除已过 TTL 的 core_idempotency_records 行。该表对每个带 " "Idempotency-Key 的请求写一行(资产转移 / 离职 / Leaver-Blocked 等)," "key 唯一,故 lookup 的惰性回收几乎从不触发 —— 必须靠 cron 定期清理," "否则表无界增长,拖慢此后每次 SELECT ... FOR UPDATE 幂等预留。分批删" "(config.batch_size,默认 500)避免 TiDB 单条超大事务('transaction " "too large')整批失败。" ), ) async def run_core_idempotency_cleanup( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.core.idempotency import cleanup_expired try: batch_size = int(config.get("batch_size") or 500) except (TypeError, ValueError): batch_size = 500 deleted = await cleanup_expired(db, batch_size=batch_size) return {"deleted": deleted, "batch_size": batch_size} __all__ = [ "run_hris_incremental_sync", "run_hris_absent_person_check", "run_provisioning_dispatch", "run_overdue_leavers_check", "run_revoke_leaver_platform_roles", "run_domain_expiry_alert", "run_day90_survey_notify", "run_hire_day_feishu_welcome", "run_hire_day_it_notify", "run_process_advance", "run_post_leaver_verify", "run_daily_digest", "run_outbox_retry", "run_resource_sync", "run_billing_sync", "run_scopes_sync", "run_roles_sync", "run_grants_sync", "run_billing_daily_sync", "run_access_expiry_sweep", "run_scope_idle_sweep", "run_leaver_grant_sweep", "run_leaver_iam_residue_sweep", "run_software_billing_auto_close", "run_software_seat_auto_revoke_pending_cancels", "run_pat_expiry_warning", "run_dns_owner_recertify", "run_dns_pre_leaver_handover", "run_cleanup_stale_running", "run_resource_creator_sweep", "run_core_idempotency_cleanup", ] @register_scheduled_job( key="cmdb.cost_recon.aliyun_bq", display_name="阿里云成本 · 财务数仓月度对账", description=( "每月 14 号 09:00(Asia/Shanghai)把【上一个完整自然月】的阿里云" "「业务折前成本」与财务数仓(BigQuery " "gcp-digital-bi.alibaba_cloud.monthly_en_*)按财务口径逐账号对账。" "财务口径 = 保留 item ∈ {Pay-As-You-Go, Purchase, Renewal, " "Upgrade/Downgrade} 的 pretax_gross_amount 按 owner_account_id 汇总 " "—— 与 Helix 的行白名单(cost_bucket='usage' 读 amount_usd_list)" "逐条对应。\n" "⚠️ 只读:不改任何数字。两条链路各自独立(一个走 BSS API 逐日拉," "一个是数仓月表),口径设计一致但会悄悄分叉且没有任何东西会报错 —— " "这个任务把「一致」变成每月一次的可证事实。\n" "排在 cmdb.cost_facts_monthclose(5 号 / 12 号 07:00)的第二遍之后两天 —— " "2026-08 实测阿里云在 5 号还没把上月最后一天结算完(BSS 对 7/31 当时" "只返回 133 行,11 天后是 536 行),对一份没写完的账只会把厂商的滞后" "报成我们的漂移。只对已结月(当月两边都在动,差异恒为噪声)。\n" "config={\"month\":\"YYYY-MM\"} 覆写目标月;" "{\"threshold_pct\":N} 覆写告警阈值。" ), ) async def run_aliyun_bq_cost_recon( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: """阿里云折前口径 × 财务数仓的月度对账(只读)。 **发现实质漂移时 raise** —— 这个仓库里 cron 的告警出口就是失败通知 (``maybe_notify_failure``);``return {"status": "failed"}`` 会被记成 ok,等于把告警吞掉(见 CLAUDE.md 的 handler 约定)。完整报告在 raise 之前先落日志,所以即便告警只带摘要,排查时也拿得到逐账号明细。 """ from zoneinfo import ZoneInfo from app.domains.cmdb.cost_recon_bq import reconcile_month month_cfg = str(config.get("month") or "").strip() if month_cfg: try: year, mon = (int(x) for x in month_cfg.split("-", 1)) month_start = date(year, mon, 1) except (ValueError, TypeError) as exc: raise ValueError( f"bad config.month {month_cfg!r}; expected YYYY-MM", ) from exc else: # 上一个完整自然月(以 Asia/Shanghai 的今天为准 —— cron 行本身 # 也是这个时区,两者不一致会在月初边界上错一个月)。 today = datetime.now(ZoneInfo("Asia/Shanghai")).date() first_of_this = today.replace(day=1) month_start = (first_of_this - timedelta(days=1)).replace(day=1) report = await reconcile_month(db, month_start=month_start) threshold = config.get("threshold_pct") if threshold is not None: # 覆写只影响本次判定,不改模块常量。 material = [ a for a in report["accounts"] if a["delta_pct"] is not None and abs(a["delta_pct"]) > float(threshold) and max(abs(a["bq_usd"]), abs(a["helix_usd"])) >= report["min_material_usd"] ] report["material_drift_count"] = len(material) report["threshold_pct"] = float(threshold) logger.info("cmdb.cost_recon.aliyun_bq report=%s", report) if report["material_drift_count"] > 0: worst = [a for a in report["accounts"] if a["material"]][:5] raise RuntimeError( f"阿里云成本与财务数仓在 {report['month']} 出现 " f"{report['material_drift_count']} 个账号的实质漂移" f"(阈值 {report['threshold_pct']}%):" + "; ".join( f"{a['account_name'] or a['account_id']} " f"BQ ${a['bq_usd']:,.2f} vs Helix ${a['helix_usd']:,.2f} " f"({a['delta_pct']:+.1f}%)" for a in worst ) ) return report @register_scheduled_job( key="cmdb.cost_facts_chain_check", display_name="云成本明细 · 两条链路日级互查", description=( "每天 08:30(Asia/Shanghai)把每个云账号每天的**日总额**" "(cmdb_cloud_billing_daily)与**逐行明细之和**(cmdb_cost_facts 的" "折前口径)对一下 —— 两张表来自**两个不同的厂商调用**,页面上的每个" "折前数字都来自后者。\n" "抓的是「明细被静默截断」:明细是逐日翻页拉的,厂商提前返回一个空 " "NextToken 就等于「这天到此为止」,而**少拉的部分没有任何痕迹** —— " "任务写 status=ok,run summary 里的行数与实际入库行数完全相等" "(拉到多少写多少)。2026 全年实测 3 起(阿里云国际站):" "5/06 日总 12,424 vs 明细 200.83、5/05 12,210 vs 8,521、" "7/31 473.88 vs 91.93 —— 前两起是 5 月对账剩下的 $15,913," "第三起是 7 月对账报出的 $381.95。\n" "⚠️ 只读:发现了只报告 + 给出重拉窗口,绝不自动重拉 —— 重拉是 " "(scope, date) 先删后插的破坏性操作,而「两条链路不一致」不等于" "「明细一定是错的那条」。\n" "阈值 比值<0.85 且缺口≥$200 是在 2026 全年 11,716 个账号×天上量出来的:" "恰好命中那 3 起、零误报。放宽到 5% 会在 AWS 上刷出 224 条噪声 —— " "AWS 两条链路来自不同数据源(Cost Explorer vs BigQuery 里的 CUR)," "天然差几个百分点。\n" "config={\"lookback_days\":N} 覆写回看窗口(默认 45);" "{\"max_ratio\":R,\"min_gap_usd\":G} 覆写阈值。" ), weight="light", ) async def run_cost_facts_chain_check( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: """两条云账单链路的日级互查(只读)。 **发现不一致时 raise** —— 同 ``cmdb.cost_recon.aliyun_bq``:这个仓库里 cron 的告警出口是失败通知,``return {"status": "failed"}`` 会被记成 ok。 """ from decimal import Decimal from zoneinfo import ZoneInfo from app.domains.cmdb.cost_chain_crosscheck import ( DEFAULT_LOOKBACK_DAYS, MAX_HEALTHY_RATIO, MIN_MATERIAL_USD, crosscheck_recent, ) def _num(key: str, fallback: float) -> float: raw = config.get(key) if raw is None: return fallback try: return float(raw) except (TypeError, ValueError): return fallback lookback = int(_num("lookback_days", DEFAULT_LOOKBACK_DAYS)) report = await crosscheck_recent( db, today=datetime.now(ZoneInfo("Asia/Shanghai")).date(), lookback_days=max(1, min(lookback, 400)), max_ratio=_num("max_ratio", MAX_HEALTHY_RATIO), min_gap_usd=Decimal(str(_num("min_gap_usd", float(MIN_MATERIAL_USD)))), ) logger.info("cmdb.cost_facts_chain_check report=%s", report) if report["divergence_count"] > 0: worst = report["divergences"][:5] raise RuntimeError( f"{report['divergence_count']} 个「账号×天」的成本明细显著低于" f"当日总额,合计缺 ${report['total_gap_usd']:,.2f}" f"(疑似翻页被静默截断):" + "; ".join( f"{d['scope_name'] or d['scope']} {d['date']} " f"日总 ${d['daily_total_usd']:,.2f} vs 明细 " f"${d['facts_total_usd']:,.2f}" for d in worst ) + " | 重拉窗口: " + "; ".join( f"{w['integration']} {w['start_date']}..{w['end_date']}" for w in report["repair_windows"][:5] ) ) return report # ── Feishu write-scope preflight ───────────────────────────────── @register_scheduled_job( key="identity_sync.feishu_write_scope_preflight", display_name="飞书 · 建号权限预检", description=( "每天探一次飞书应用是否仍持有 Directory 建号写权限" "(对非法 employee_id 发 PATCH,只验权限不改数据)。缺权限时任务失败," "由定时任务失败通知 page IT —— 避免建号面静默失效到下一个新人入职" "才被发现(#3084 空转 12 天)。" ), ) async def run_feishu_write_scope_preflight( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: """Assert the Feishu app can still create employees. Deliberately independent of whether anyone is being hired — that coupling is the whole bug this job exists for. A runner-side check only fires when a create is attempted, and between 2026-08-19 and 08-30 no create was attempted, so a completely dead create surface looked green for 12 days and was found by a new hire with no account on day one. Raises on a missing scope so ``maybe_notify_failure`` pages IT through the existing scheduled-job failure channel — no new alert plumbing, and the message carries Feishu's own self-service grant link. """ from sqlalchemy import select as _select from app.approval.feishu_client import FeishuApprovalClient from app.domains.identity_sync.models import IdentitySyncTarget from app.domains.identity_sync.runners.feishu_contact import ( FeishuContactRunner, ) from app.integrations.models import Integration from app.integrations.service import _resolve as _resolve_integration target = await db.scalar( _select(IdentitySyncTarget).where( IdentitySyncTarget.kind == "feishu_contact", IdentitySyncTarget.enabled.is_(True), IdentitySyncTarget.deleted_at.is_(None), ) ) if target is None or target.integration_id is None: return {"status": "skipped", "reason": "no enabled feishu_contact target"} integ = await db.get(Integration, target.integration_id) if integ is None: return {"status": "skipped", "reason": "integration row missing"} resolved = await _resolve_integration(db, integ) app_id = resolved.config.get("app_id") app_secret = resolved.config.get("app_secret") if not app_id or not app_secret: return {"status": "skipped", "reason": "integration missing credentials"} runner = FeishuContactRunner( client=FeishuApprovalClient(app_id=app_id, app_secret=app_secret), ) missing = await runner._directory_write_scope_missing() if missing: raise RuntimeError( "飞书应用缺少建号所需的 Directory 写权限,当前任何新员工都建不出" f"账号(更新不受影响)。开通后必须发布新版本才生效。飞书原话:{missing}" ) return {"status": "ok", "target": target.slug, "scope": "directory_write_ok"} @register_scheduled_job( key="apikey.idle_check", display_name="API Key · 闲置检查", description=( "每月 3 日 09:30(Asia/Shanghai,等账单入库)。扫描 status='active' 且发放满 " "min_age_days 的 API Key,取上自然月花费:$0 且数据可用 → 闲置。IT 收一张汇总卡" "(始终);员工本人那张「保留 / 释放 / 找 IT」卡由 apikey.idle_alert." "enabled_for_employees 控制,默认关。Bailian 等拿不到按 Key 花费的提供方跳过并" "计入 skipped_no_data。dedup 表按 (assignment, month) 保证每月最多问一次。" ), ) async def run_apikey_idle_check( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: """Monthly "is this key still used" sweep (migration 603). Same two-transaction posture as ``apikey.expiry_check``: the evaluation (ledger rows + ``detail.idle``) commits before the IT digest goes out, so a notification-provider failure cannot roll back a month's evaluation and make next month's streak arithmetic lie. Raises when every evaluated key errored (systemic — cost facts unreadable), never on individual keys. """ from zoneinfo import ZoneInfo from app.core.database import async_session_factory from app.domains.apikey.idle_alerts import ( load_idle_settings, send_it_digest, sweep_idle, ) trace_id = uuid.uuid4().hex today = datetime.now(ZoneInfo("Asia/Shanghai")).date() async with async_session_factory() as session: settings = await load_idle_settings(session) summary = await sweep_idle(session, today=today, settings=settings, trace_id=trace_id) await session.commit() items = summary.pop("idle_items", []) async with async_session_factory() as digest_session: summary["it_digest_recipients"] = await send_it_digest( digest_session, period=summary["period"], items=items, settings=settings, trace_id=trace_id, ) await digest_session.commit() evaluated = int(summary.get("evaluated") or 0) errors = summary.get("errors") or [] if evaluated > 0 and len(errors) == evaluated: preview = " | ".join( f"{e.get('assignment_id')}: {e.get('error')}" for e in errors[:3] ) raise RuntimeError( f"apikey.idle_check systemic failure: all {evaluated} keys errored. " f"First: {preview}" ) return summary @register_scheduled_job( key="notifications.card_settlement_reconcile", display_name="卡片回执对账", description=( "每 15 分钟扫一遍「还没结案、而它说的那件事已经有结局」的通知行," "把卡片改成回执。回执的快路径是各域在终态迁移处内联调用 settle();" "这个任务是**保证**那一条腿 —— 一个 46 处调用点的平台里「记得加那一行」" "注定会漏,有了对账,漏掉一处只是最多 15 分钟的延迟,不再是一张永久" "骗人的卡片。两边都幂等,抢到同一件事也只有第一次做实事。" ), ) async def run_card_settlement_reconcile( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.core.coordinators.notification_settlement_coordinator import ( RECONCILE_SCAN_LIMIT, reconcile, stale_actionable_count, ) from app.core.database import async_session_factory limit = int(config.get("scan_limit") or RECONCILE_SCAN_LIMIT) async with async_session_factory() as session: summary = await reconcile(session, limit=limit) # 健康指标:超过 30 天仍未结案的 actionable 通知条数。它长期涨意味着有 # 一类卡片既没被内联结案、也不被对账覆盖(它的 resource_type 在 OPT_OUT # 里而域侧忘了内联)—— 那是「回执整体还灵不灵」的那一个数,所以每轮都 # 记一次,让它在运行记录里可回溯。 summary["stale_actionable_over_30d"] = await stale_actionable_count(session) await session.commit() return summary