"""Scheduled-job handlers for the security auditor domain. Seven keys, each registered against the ``scheduled_jobs`` rows seeded by migration 092: * auditor.mfa_sync_daily * auditor.public_exposure_scan_daily * auditor.secret_rotation_check_daily * auditor.signin_anomaly_score_hourly * auditor.pat_audit_daily * auditor.cert_expiry_check_daily * auditor.risk_score_recompute_hourly Plus an opt-in ``auditor.evidence_pack_worker`` that drains pending ``auditor_evidence_packs`` rows on each tick. """ from __future__ import annotations import logging import uuid from typing import Any from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.domains.auditor.models import AuditorEvidencePack from app.domains.auditor.services import ( EvidencePackService, RiskScoreService, ) from app.domains.auditor.sync_services import ( CertExpirySyncService, MfaSyncService, PatAuditService, PublicExposureScanService, SecretRotationSyncService, SigninAnomalyService, ) from app.scheduled_jobs.registry import register_scheduled_job from app.scheduled_jobs.slo import SLO_DAILY_DIGEST logger = logging.getLogger(__name__) def _raise_if_systemic_failure( key: str, result: dict[str, Any], *, progress_keys: tuple[str, ...], ) -> dict[str, Any]: """Turn "every upstream failed" into a FAILED run. The scheduler marks a run failed — and fires the failure notifier — **only if the handler raises**; returning normally lands ``status='ok'``. These sync services collect provider errors into ``result["errors"]`` and always return, so a total outage was invisible: 30 days of "ok" runs while nothing synced (#2136). Worse than the missing alert: ``last_run_at`` keeps advancing, so the continuous-monitoring freshness gate — which only looks at ``job.last_run_at`` — never flips to ``stale`` either. MFA coverage, key-rotation overdue and cert-expiry then keep scoring pass/fail against month-old data. #2122 (three adapters pointing at a non-existent driver kind, swallowed as INFO "not configured" for the module's entire lifetime) is the live proof this channel exists. "Systemic" = there were errors AND nothing at all got through, which is why ``progress_keys`` is per-handler: a partial failure is a real and normal state for a multi-provider sync and must NOT alarm. Mirrors the explicit checks ``evidence_pack_worker`` and ``revoke_fulfil`` already carry. """ errors = result.get("errors") or [] if not errors: return result def _count(value: Any) -> int: # Payload shapes vary: a plain count, a list of scanned sources, # or a per-provider dict whose values are themselves count dicts # (``{"feishu": {"total": 120, ...}}``). Walk one level in so a # nested shape is not read as zero progress — that would turn # "one provider failed" into a paging alarm. if isinstance(value, bool): return 0 if isinstance(value, (int, float)): return int(value) if isinstance(value, list): return len(value) if isinstance(value, dict): return sum(_count(v) for v in value.values()) return 0 progress = sum(_count(result.get(pkey)) for pkey in progress_keys) if progress > 0: return result raise RuntimeError( f"{key}: every upstream failed and nothing was synced " f"({len(errors)} error(s)); refusing to report a healthy run — " f"a silent 'ok' also keeps the CCM freshness gate green. " f"errors={errors[:5]}" ) @register_scheduled_job( key="auditor.mfa_sync_daily", display_name="安全员 · MFA 合规每日同步", description=( "每日同步 Google / Feishu / GitHub 用户的 MFA 启用状态到 " "auditor_mfa_status 表;为安全员仪表板和风险评分模块提供 " "合规数据基线。" ), ) async def run_mfa_sync( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: providers = config.get("providers") return _raise_if_systemic_failure( "auditor.mfa_sync_daily", await MfaSyncService.sync( db, providers=providers, actor_user_id=None, trace_id=uuid.uuid4(), ), progress_keys=("per_provider",), ) @register_scheduled_job( key="auditor.public_exposure_scan_daily", display_name="安全员 · 公网暴露每日扫描", description=( "每日扫描 AWS Security Group 0.0.0.0/0 入站规则与 S3 公开桶 " "(ACL + 桶策略),将 finding 落入 auditor_public_exposure_findings;" "本次扫描中未观察到的开放项自动标记为 resolved。" ), ) async def run_public_exposure_scan( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: return _raise_if_systemic_failure( "auditor.public_exposure_scan_daily", await PublicExposureScanService.scan( db, actor_user_id=None, trace_id=uuid.uuid4(), ), progress_keys=("sources_scanned", "findings_count"), ) @register_scheduled_job( key="auditor.secret_rotation_check_daily", display_name="安全员 · 密钥轮换每日检查", description=( "刷新 Helix API key / AWS IAM access key / AWS ACM 证书的 " "最近轮换时间与下一次到期日,标记超期 (overdue) 与即将到期 " "(due_soon) 的密钥;auditor_secret_rotation_records 表是台账。" ), ) async def run_secret_rotation_check( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: return _raise_if_systemic_failure( "auditor.secret_rotation_check_daily", await SecretRotationSyncService.sync( db, actor_user_id=None, trace_id=uuid.uuid4(), ), progress_keys=("scanned_sources", "record_count"), ) @register_scheduled_job( key="auditor.signin_anomaly_score_hourly", display_name="安全员 · 登录事件每小时拉取与异常评分", description=( "每小时从 Google Reports Audit API 拉取登录事件,计算异常评分 " "(离职后登录 / 异常时段 / Headless UA / 失败重试) 并落入 " "auditor_signin_events;高/极高 anomaly_level 在仪表板显眼。" ), ) async def run_signin_anomaly_score( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: lookback = int(config.get("lookback_minutes", 70)) return _raise_if_systemic_failure( "auditor.signin_anomaly_score_hourly", await SigninAnomalyService.pull_and_score( db, lookback_minutes=lookback, actor_user_id=None, trace_id=uuid.uuid4(), ), progress_keys=("pulled_providers", "new_event_count"), ) @register_scheduled_job( key="auditor.pat_audit_daily", display_name="安全员 · PAT / MCP 每日审计", description=( "刷新 GitHub 组织 PAT、Helix API key 分配、OAuth grants 的状态;" "为休眠 (dormant)、广 admin 范围 (broad_admin_scope)、 " "持有人已离职 (owner_left)、无到期日 (no_expiry) 等风险打标。" ), ) async def run_pat_audit( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: return _raise_if_systemic_failure( "auditor.pat_audit_daily", await PatAuditService.sync( db, actor_user_id=None, trace_id=uuid.uuid4(), ), progress_keys=("pulled_sources", "record_count"), ) @register_scheduled_job( key="auditor.cert_expiry_check_daily", display_name="安全员 · 证书 / ICP 每日到期检查", description=( "刷新 AWS ACM 证书与手工台账 (域名 / ICP 备案 / 自管 SSL) 的 " "到期日,在 90 / 60 / 30 天阈值上推升 alert_level 让仪表板预警。" ), ) async def run_cert_expiry_check( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: return _raise_if_systemic_failure( "auditor.cert_expiry_check_daily", await CertExpirySyncService.sync( db, actor_user_id=None, trace_id=uuid.uuid4(), ), progress_keys=("pulled_sources", "record_count"), ) @register_scheduled_job( key="auditor.risk_score_recompute_hourly", display_name="安全员 · 风险评分每小时重算", description=( "基于 MFA / 登录异常 / 超期密钥 / 风险 PAT / 过期证书 等上游信号 " "重算每个未软删 person(含离职者 —— 离职后登录检测需要)的综合" "风险评分(0-100),写入 " "auditor_risk_scores;高/极高分数会出现在仪表板和员工 /me 页。" ), ) async def run_risk_score_recompute( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: return await RiskScoreService.recompute_all( db, actor_user_id=None, trace_id=uuid.uuid4(), ) @register_scheduled_job( key="auditor.privileged_review_auto_open_quarterly", display_name="安全员 · 特权角色季度复核自动开盘", description=( "每个日历季度的第一天 09:00 UTC (1/4/7/10 月 1 号) 自动开一个 " "特权角色访问认证活动(campaign_type=privileged_role),从 Helix / " "Google / Feishu / AWS / GitHub / Atlassian(若已配)抓取所有特权" "角色持有人作为 baseline,挂在 /compliance/certifications 等待审计员" "逐人 keep / revoke。label 形如 '2026-Q2 特权访问复核',该季度已有" "则幂等跳过。(2026-06-04 起收编进统一访问认证引擎。)" ), ) async def run_privileged_review_auto_open( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.domains.auditor import access_certification_service as acs due_after_days = int(config.get("due_after_days", 30)) result = await acs.auto_open_quarterly_privileged( db, due_after_days=due_after_days, trace_id=uuid.uuid4(), ) await db.commit() return result @register_scheduled_job( key="auditor.access_self_attestation_auto_open_semiannual", display_name="安全员 · 全员访问自查半年度自动开盘", description=( "每日 08:00 UTC 运行,但仅在边界后 7 天窗口内开盘:6 月 1 日开 " "H1、12 月 1 日开 H2(campaign_type=self_attestation)。一次性快照" "每位在职员工的完整访问足迹(坐席/云授权/软件权限/SSO/API Key)," "并给每个持有人发一张批量确认卡(去 /me/access-review 一屏勾选" "「还需要 / 我不需要了」)。label 形如 '2026-H2 全员访问自查自证'," "该期已有则幂等跳过。刻意不追溯已过的历史边界(防中途上线全员推送" "风暴);错过的当期由管理员手动开盘。完成的活动 = SOC2 CC6.3 / " "ISO A.5.15 用户访问复核证据。" ), # #2243: snapshots every active employee's full access footprint and # then fans out ~500 × 2 channels through enqueue-then-drain, all on # the handler's pooled session. Minutes of held connection = the # registry's ``heavy`` criterion; it was defaulting to ``light``. weight="heavy", ) async def run_access_self_attestation_auto_open( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.domains.auditor import access_certification_service as acs due_after_days = int(config.get("due_after_days", 21)) window_days = int(config.get("window_days", 7)) result = await acs.auto_open_semiannual_self_attestation( db, due_after_days=due_after_days, window_days=window_days, trace_id=uuid.uuid4(), ) await db.commit() return result @register_scheduled_job( key="auditor.identity_consumer_recert_auto_open_quarterly", display_name="安全员 · 身份消费方季度复核自动开盘", description=( "每季度首日 09:00 UTC 自动开一个身份消费方访问复核活动" "(campaign_type=identity_consumer),把所有 active 消费方(对外数据" "拉取面 §10.6)挂到 /compliance/certifications 等审计员逐个 " "keep/revoke;item 带近 90 天调用量信号便于识别可下线的消费方。" "label 形如 '2026-Q2 身份消费方复核',该季度已有则幂等跳过。" ), ) async def run_identity_consumer_recert_auto_open( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.domains.auditor import access_certification_service as acs due_after_days = int(config.get("due_after_days", 30)) result = await acs.auto_open_quarterly_identity_consumer( db, due_after_days=due_after_days, trace_id=uuid.uuid4(), ) await db.commit() return result @register_scheduled_job( key="auditor.identity_consumer_recert_lapse_sweep", display_name="安全员 · 身份消费方复核逾期自动停用", description=( "每日 10:00 UTC 扫描身份消费方复核项:决定仍 pending 且已超过" "截止 + 14 天宽限期的,自动把对应消费方置 disabled(停止对外 PII " "出口)并标记 recert_lapsed_at;审计员事后判 keep 会自动重新启用。" "幂等:已停用 / 已标记的跳过。" ), ) async def run_identity_consumer_recert_lapse_sweep( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.domains.auditor import access_certification_service as acs result = await acs.lapse_disable_identity_consumers( db, trace_id=str(uuid.uuid4()), ) await db.commit() return result @register_scheduled_job( key="auditor.self_check_opener_monthly", display_name="安全员 · 周期自检台账每月自动开单", description=( "每月 1 号 01:00 UTC 为每个启用的周期自检类目开当前周期的待检查台账" "(季度/半年/年的边界都落在月初,单月节律即可覆盖全部周期)。" "幂等:(类目, 年, 周期标签) 已有记录则跳过。新开记录默认指派给类目的" "默认检查人,逾期由 periodic_self_check 控制项在合规总览体现。" ), ) async def run_self_check_opener( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from sqlalchemy.exc import IntegrityError from app.domains.auditor.self_check_service import open_due_records try: result = await open_due_records(db) await db.commit() return result except IntegrityError: # A concurrent manual /open-due (or a duplicate tick) won the # unique-key race on (category, period) — the records exist, so # this is an idempotent no-op, not a failure. await db.rollback() return {"opened": [], "opened_count": 0, "skipped": 0, "skipped_reason": "raced (records already opened)"} @register_scheduled_job( key="auditor.assessment_cycle_opener_semiannual", display_name="安全员 · 信息安全自评估半年开盘", description=( "每年 1/1 与 7/1 02:00 UTC 为刚结束的半年(1 月开 → 上一年下半年;" "7 月开 → 本年上半年)开一份信息安全自评估报告草稿:按 ISMS 模板实例化" "各章节并从资产/RBAC/MFA/离职/复核/密钥/变更/事件/云/周期自检等真值" "自动填表,等审计员补叙述与评估结论后定稿。该周期已有则幂等跳过。" ), ) async def run_assessment_cycle_opener( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from sqlalchemy.exc import IntegrityError from app.core.exceptions import ValidationError from app.domains.auditor.assessment_service import create_cycle try: cycle = await create_cycle(db) await db.commit() return {"opened": cycle.period_key} except (ValidationError, IntegrityError) as exc: # Period already has a cycle (pre-checked) OR a concurrent writer # won the unique-key race → idempotent no-op, not a failure. await db.rollback() return {"opened": None, "skipped_reason": str(exc)} @register_scheduled_job( key="auditor.evidence_pack_worker", display_name="安全员 · 证据包后台生成", description=( "扫描 auditor_evidence_packs 中的 pending 行,合成 PDF + ZIP 证据包" "并落盘;写入 file_path / sha256 / file_size。每次 tick 处理至 " "多 N 个 (默认 3) 行,避免长 tick。" ), ) async def run_evidence_pack_worker( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.core.database import async_session_factory # noqa: PLC0415 max_per_tick = int(config.get("max_per_tick", 3)) pending = (await db.execute( select(AuditorEvidencePack) .where(AuditorEvidencePack.status == "pending") .order_by(AuditorEvidencePack.created_at) .limit(max_per_tick) )).scalars().all() built: list[str] = [] failed: list[str] = [] for pack in pending: # Use an independent session per pack so a failure on pack N # is fully isolated: flushed-but-not-committed partial state # from pack N is rolled back on __aexit__ without affecting # pack N+1 or the caller's shared session. try: async with async_session_factory() as pack_session: result = await EvidencePackService.generate( pack_session, pack_id=pack.id, ) await pack_session.commit() if result.status == "completed": built.append(str(pack.id)) else: failed.append(str(pack.id)) except Exception: # noqa: BLE001 logger.exception( "evidence pack worker: %s build crashed", pack.id, ) failed.append(str(pack.id)) # Systemic-failure: every queued pack failed. PDF/ZIP generation # is fully broken (missing renderer binary, FS permissions, etc.) # — auditors will only notice on the next download attempt. if pending and not built and failed: raise RuntimeError( f"auditor.evidence_pack_worker: 0 of {len(pending)} packs built; " f"all failed: {failed[:3]}" ) return { "scanned": len(pending), "built": built, "failed": failed, } @register_scheduled_job( key="auditor.privileged_review_escalate", display_name="特权审阅升级扫描", description=( "每天 03:30 (Asia/Shanghai) 扫描所有 decision='pending' 且 " "decided_at_deadline 已经过去 14 天的 AccessCertificationItem, " "设置 escalated_at + 写 compliance_incident " "(type=access_review_overdue, severity=warning),记录为审计证据。" "Idempotent — 一次写入 escalated_at 之后不再重复扫描该行。" "GRC PR3 起特权复核已并入访问认证引擎(原 privileged-review 表已删)。" ), ) async def run_privileged_review_escalate( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.core.database import async_session_factory from app.domains.auditor import access_certification_service as acs trace_id = ( f"system:cron:auditor.privileged_review_escalate:{uuid.uuid4()}" ) async with async_session_factory() as session: result = await acs.escalate_overdue(session, trace_id=trace_id) await session.commit() result["trace_id"] = trace_id return result @register_scheduled_job( key="compliance.control_status_recompute", display_name="合规 · 控制项状态重算(CCM)", description=( "每小时(:15)持续控制监控(CCM):先从 compliance_catalog 幂等同步" "框架/条款/控制项/crosswalk,再按每个控制项的支撑信号(检测类定时任务" "新鲜度 + 它产出的指标)评估 pass/fail/stale/not_configured,upsert " "进 compliance_control_status。喂给「合规总览」看板与各框架就绪度。" ), ) async def run_compliance_control_status_recompute( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.core.database import async_session_factory from app.domains.auditor.compliance_service import ( recompute_control_status, ) async with async_session_factory() as session: # recompute_control_status commits internally. return await recompute_control_status(session) @register_scheduled_job( key="auditor.access_certification_revoke_fulfil", display_name="访问认证 · 撤权执行与对账", description=( "每 5 分钟对账 access_certification_items 中 decision='revoke' 且 " "revocation_status='pending' 的行:① 之前 enqueue 未落地的重试入队;" "② 关联的 provisioning 撤权任务已完成/已通知 → 标记 completed;" "失败/取消 → 标记 failed;仍在途 → 保持 pending。复用 provisioning " "队列(幂等 + 重试 + 审计),不另造机制,把复核员的 revoke 决策真正" "落到下游 SaaS。" ), ) async def run_access_certification_revoke_fulfil( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.core.coordinators.access_certification_coordinator import ( fulfil_pending_revocations, ) from app.core.database import async_session_factory trace_id = ( f"system:cron:auditor.access_certification_revoke_fulfil:" f"{uuid.uuid4()}" ) batch_size = int((config or {}).get("batch_size", 100)) async with async_session_factory() as session: result = await fulfil_pending_revocations( session, trace_id=trace_id, batch_size=batch_size, ) await session.commit() result["trace_id"] = trace_id # ``fulfil_pending_revocations`` increments ``failed`` when an auditor's # revoke decision's downstream provisioning task landed FAILED/CANCELLED — # i.e. the revoke did NOT reach the SaaS. A run is only marked failed (and # the failure-notifier only fires) if the handler RAISES; returning # normally lands status='ok', so an unenforced revoke would sit silently — # exactly the governance gap this 5-minute cron exists to close. Commit # first (so REVOCATION_FAILED state persists), then raise so the run alarms # and an operator is paged. Mirrors the "handlers raise on failure" rule. failed = int(result.get("failed", 0) or 0) if failed > 0: raise RuntimeError( f"access_certification revoke fulfil: {failed} item(s) failed " f"to revoke downstream; {result}" ) return result @register_scheduled_job( key="auditor.vendor_assessment_review_reminder", display_name="安全问卷 · 供应商复评到期提醒", description=( "每天 04:00 UTC 扫描已完成的供应商安全评估,按 next_review_date 分阶段" "(提前 30 天 / 14 天 / 到期日 / 逾期)提醒评估负责人复评。每个阶段仅提醒" "一次(review_reminder_stage 记录已发阶段,幂等且不重复打扰),仅推送给" "内部负责人(auditor/IT),绝不触达员工或外部供应商。" ), ) async def run_vendor_assessment_review_reminder( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.core.database import async_session_factory from app.domains.auditor import vendor_assessment_service as vsvc from app.identity.person_models import IdentityPerson from app.notifications.service import NotificationEnvelope, NotificationService stage_labels = { "30": "30 天后到期", "14": "14 天后到期", "due": "今日到期", "overdue": "已逾期", } reminded = 0 skipped_no_owner = 0 async with async_session_factory() as session: due = await vsvc.find_due_review_reminders(session) notifier = NotificationService(session) for assessment, stage, days_left in due: owner_id = assessment.owner_person_id or assessment.reviewer_person_id if owner_id is None: skipped_no_owner += 1 # Still advance the stage so a null-owner row doesn't jam the # scan every day — the posture dashboard surfaces it instead. vsvc.mark_reminder_sent(assessment, stage) continue person = await session.get(IdentityPerson, owner_id) if person is None or person.deleted_at is not None: skipped_no_owner += 1 vsvc.mark_reminder_sent(assessment, stage) continue label = stage_labels.get(stage, stage) title = f"供应商安全复评提醒 · {assessment.vendor_name}" body = ( f"供应商「{assessment.vendor_name}」的安全评估复评{label}" f"(下次复评日期 {assessment.next_review_date})。" f"当前风险评级:{assessment.risk_rating or '未评级'}。" f"请安排复评并在 Helix 安全问卷中新建/更新评估。" ) vsvc.mark_reminder_sent(assessment, stage) # Commit THIS reminder's stage advance before sending the # next one. Previously the whole scan ran in one transaction # with the sends inline, so (a) the dirty ``reminder_stage`` # rows and their outbox INSERTs held locks across every # subsequent Feishu round-trip, and (b) an exception at # reminder k rolled back the stage advances for reminders # 1..k-1 whose messages had ALREADY been sent — so the next # day re-sent every one of them (#2140). Same rule the # notification retry worker states for itself: commit the # row's terminal state BEFORE the next external send. await session.commit() try: await notifier.push_inbox_and_feishu(NotificationEnvelope( recipient=person, title=title, body_text=body, channels=["feishu", "inbox"], related_resource_type="auditor_vendor_assessment", related_resource_id=assessment.id, url="/auditor/security-questionnaire", )) await session.commit() except Exception: # noqa: BLE001 # The stage advance is ALREADY COMMITTED, so this # reminder will not be retried tomorrow — deliberate: a # Feishu outage must not turn into a repeat blast. # # Be precise about what the rollback discards: # ``push_inbox_and_feishu`` writes ONE outbox row with # ``status=DELIVERED`` and has no retry queue (that is # its documented bell-first contract), so rolling back # drops that bell row too — this recipient simply does # not get this stage's reminder. Losing one nudge is the # accepted cost of not re-blasting everyone; the # ``logger.exception`` below is the only record, and is # why it is an ``exception`` and not a ``warning``. await session.rollback() logger.exception( "vendor review reminder send failed for %s", assessment.id, ) continue reminded += 1 await session.commit() return {"reminded": reminded, "skipped_no_owner": skipped_no_owner} @register_scheduled_job( key="auditor.inbound_questionnaire_due_reminder", display_name="安全问卷 · 客户问卷截止提醒", description=( "每天 05:00 UTC 扫描未完成的客户安全问卷(乙方侧),按 due_date 分阶段" "(提前 7 天 / 3 天 / 到期日 / 逾期)提醒问卷负责人。每个阶段仅提醒一次" "(reminder_stage 记录已发阶段,幂等且不重复打扰),仅推送给内部负责人" "(auditor/IT),绝不触达员工或外部客户。" ), ) async def run_inbound_questionnaire_due_reminder( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.core.database import async_session_factory from app.domains.auditor import inbound_service as inb from app.identity.person_models import IdentityPerson from app.notifications.service import NotificationEnvelope, NotificationService stage_labels = { "7": "7 天后到期", "3": "3 天后到期", "due": "今日到期", "overdue": "已逾期", } reminded = 0 skipped_no_owner = 0 async with async_session_factory() as session: due = await inb.find_due_reminders(session) notifier = NotificationService(session) for q, stage, _days in due: owner_id = q.owner_person_id if owner_id is None: skipped_no_owner += 1 inb.mark_reminder_sent(q, stage) continue person = await session.get(IdentityPerson, owner_id) if person is None or person.deleted_at is not None: skipped_no_owner += 1 inb.mark_reminder_sent(q, stage) continue label = stage_labels.get(stage, stage) inb.mark_reminder_sent(q, stage) # See the vendor-reminder twin above for why the commit goes # before the send (#2140). await session.commit() try: await notifier.push_inbox_and_feishu(NotificationEnvelope( recipient=person, title=f"客户安全问卷提醒 · {q.customer_name}", body_text=( f"客户「{q.customer_name}」的安全问卷《{q.title}》{label}" f"(截止日期 {q.due_date})。请在 Helix 安全问卷 → 客户问卷中" f"完成作答并导出。" ), channels=["feishu", "inbox"], related_resource_type="auditor_inbound_questionnaire", related_resource_id=q.id, url="/auditor/security-questionnaire", )) await session.commit() except Exception: # noqa: BLE001 # See the vendor twin: the stage advance is already # committed, and the rollback drops this recipient's # bell row with it (``push_inbox_and_feishu`` has no # retry queue). One lost nudge, no repeat blast. await session.rollback() logger.exception( "inbound questionnaire reminder send failed for %s", q.id, ) continue reminded += 1 await session.commit() return {"reminded": reminded, "skipped_no_owner": skipped_no_owner} @register_scheduled_job( key="auditor.security_faq_embed_daily", display_name="安全 FAQ · 语义索引回填", description=( "每天 05:40(Asia/Shanghai)把还没有向量、或向量是用旧嵌入模型算的" "员工可见答案条目补齐,让 安全 FAQ 的语义检索覆盖整个库。\n" "为什么需要一个 cron 而不是只在写入时嵌:条目会被**种子迁移**批量落库" "(381/528/562 三次共 100+ 条),那条路径上没有任何应用代码可以挂钩;" "而且写入时的后台嵌入是 fire-and-forget —— 进程重启、模型限流、Key 当时" "没配好,都会留下一条永远不在语义索引里的条目,而它的表现是「搜不到」," "不是报错。\n" "逐条提交:一批里的每个向量算完就落库,中途厂商挂掉不会把已经付过费的" "向量一起回滚;整批零进展就停,不空转。\n" "没配 AI/LLM Key 时是干净的空跑(embedded=0),不报错 —— 语义检索不可用" "时 安全 FAQ 仍按词法检索工作。" ), slo=SLO_DAILY_DIGEST, ) async def run_security_faq_embed( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: """Backfill the 安全 FAQ semantic index. Never rewrites an answer.""" from app.domains.auditor import security_faq_rag as rag batch = int((config or {}).get("batch") or 50) max_total = int((config or {}).get("max_total") or 1000) result = await rag.drain(db, batch=batch, max_total=max_total) stats = await rag.index_stats(db) return {**result, "model": stats["model"], "indexed": stats["indexed"]}