"""Owner-governance service for cloud scopes. Three responsibilities: 1. Build & persist pending governance prompts (semiannual confirm, zero-spend alert) and dispatch the Feishu card to the scope owner. 2. Respond endpoints — owner clicks ✅/🔄/🗑/🆘 → flip the pending row to its leaf-kind, perform the side-effect (transfer ownership / record a keep-reason / etc.), and (for transfers) bump ``last_owner_confirmed_at``. 3. Escalate — daily sweep finds ``pending`` rows older than 14 days, notifies every IT admin, and writes a separate ``escalated_to_it`` timeline row. All snapshot data (spend, member count, top accessors) is captured at prompt-creation time and stored on the confirmation row so the timeline UI can show "what the owner saw when they decided" — independent of later cost-facts re-ingest or membership churn. Side-effects (Feishu push, audit log, scope mutation) happen WITHIN the caller's session — we never start a fresh session here. Caller commits. """ from __future__ import annotations import logging import uuid from collections.abc import Sequence from dataclasses import dataclass from datetime import UTC, date, datetime, timedelta from decimal import Decimal from typing import Any from sqlalchemy import func, or_, select from sqlalchemy.ext.asyncio import AsyncSession from app.audit.service import AuditService from app.core.exceptions import AppException, ForbiddenError, NotFoundError from app.core.txn import best_effort from app.domains.cmdb.cost_view import fact_engineering_amount from app.domains.cmdb.models import ( OWNER_CONFIRMATION_KIND_OWNER_ASSIGNED, OWNER_CONFIRMATION_KIND_OWNER_AUTO_ASSIGNED, OWNER_CONFIRMATION_KIND_OWNER_FORCE_TRANSFERRED, OWNER_CONFIRMATION_KIND_OWNER_TRANSFERRED, OWNER_CONFIRMATION_KIND_SEMIANNUAL_CONFIRMED, OWNER_CONFIRMATION_KIND_SEMIANNUAL_PENDING, OWNER_CONFIRMATION_KIND_SEMIANNUAL_TRANSFERRED, OWNER_CONFIRMATION_KIND_ZERO_SPEND_CLEANUP, OWNER_CONFIRMATION_KIND_ZERO_SPEND_HANDOVER_IT, OWNER_CONFIRMATION_KIND_ZERO_SPEND_KEEP, OWNER_CONFIRMATION_KIND_ZERO_SPEND_PENDING, OWNER_CONFIRMATION_RESPONSE_VOIDED_OWNER_CHANGED, OWNER_CONFIRMATION_RESPONSE_VOIDED_SCOPE_CLOSED, OWNER_CONFIRMATION_STATUS_PENDING, OWNER_CONFIRMATION_STATUS_RESPONDED, PROD_ENVIRONMENTS, SCOPE_STATUS_ACTIVE, SCOPE_STATUS_CLOSED, TENANT_LEVEL_SCOPE_TYPES, CmdbCloudAccessGrant, CmdbCloudScope, CmdbCloudScopeOwnerConfirmation, CmdbCostFact, ) from app.identity.person_models import IdentityPerson from app.notifications import cadence from app.notifications.service import ( NotificationEnvelope, NotificationService, ) logger = logging.getLogger(__name__) # Semiannual cadence — owners are prompted to confirm every 180 days # since their last confirmation (or scope creation if never confirmed). SEMIANNUAL_DAYS = 180 # Default "newborn guard" — skip scopes younger than this so a freshly # synced scope doesn't get a card the same day it lands. Admins can # override per-run via the ``min_age_days`` config knob on the cron # (e.g. set to 0 for a first-time init sweep on a tenant whose scopes # were all bulk-imported within the last 30 days). DEFAULT_NEWBORN_GUARD_DAYS = 30 # Round-51 L3-P0: platform-level "first-observation" stamp written on # the FIRST cron tick. Before this stamp every scope's # ``last_owner_confirmed_at = NULL`` would match the SEMIANNUAL_PENDING # eligibility window, carpet-bombing ~200 owners with cards on launch # day. The cron records ``now`` into ``platform_settings`` on first # run; subsequent ticks only fire for scopes whose ``created_at < # first_observation_stamp - min_age_days``. Backfill-aware (does not # re-stamp on tick 2+); admin-overridable by deleting the row. OWNER_CONFIRMATION_FIRST_OBSERVED_KEY = ( "cmdb.owner_confirmation.first_observed_at" ) # Minimum gap between two reminders for the SAME open prompt. The daily # cron fires at a fixed local time, so a 20-hour floor yields exactly # one reminder per day and makes a same-day manual re-trigger a no-op — # without any timezone arithmetic. See ``remind_unanswered_owners``. # # (This replaced ``ESCALATION_DAYS = 14``: unanswered prompts are no # longer escalated to IT after two weeks, they are re-sent to the owner # every day until answered. 2026-08-24 product decision.) REMINDER_MIN_INTERVAL_HOURS = 20 # Cost-fact aggregation windows (days back from today). SPEND_WINDOW_30 = 30 SPEND_WINDOW_90 = 90 SPEND_WINDOW_180 = 180 # Statuses considered "live access" when counting members for the # owner-card snapshot. We INTENTIONALLY count only ``active`` here # (NOT ``pending``) — Round-22 audit: # # * ``pending`` grants are requests awaiting owner approval; they # do not have cloud-side IAM yet, so counting them inflates the # "成员数" the owner sees on the semiannual card and misleads # the keep-vs-rotate decision. Pending requests are surfaced on # the scope detail page separately, where they are actionable. # * ``approved`` is not a real ``CmdbCloudAccessGrant.status`` — # the constants are ``pending → active → revoked/expired/disabled`` # (see ``GRANT_STATUS_*`` in models.py). It was dead code. LIVE_GRANT_STATUSES: tuple[str, ...] = ("active",) # Frontend deep-link path templates. The notification service resolves # them against ``app.frontend_base_url`` so callers don't need to know # the base URL. _SCOPE_DETAIL_PATH = "/cloud/scopes/{scope_id}" _OWNER_RESPOND_PATH = ( "/cloud/scopes/{scope_id}?ownerAction={action}&confirmationId={confirmation_id}" ) def _scope_detail_url(scope_id: str) -> str: return _SCOPE_DETAIL_PATH.format(scope_id=scope_id) def _owner_action_url(scope_id: str, confirmation_id: str, action: str) -> str: return _OWNER_RESPOND_PATH.format( scope_id=scope_id, confirmation_id=confirmation_id, action=action, ) def _aware(dt: datetime | None) -> datetime | None: """Coerce a possibly-naive datetime to UTC-aware.""" if dt is None: return None return dt if dt.tzinfo else dt.replace(tzinfo=UTC) # Owner-change events that exist ONLY as ``audit_logs`` rows — no # confirmation row is written for either. Mapped to the read-time-only # kinds so the governance timeline can render them alongside stored # rows. See the kind constants in ``models.py`` for why each path never # reached the leaf-row writer. _AUDIT_DERIVED_OWNER_EVENTS: dict[str, str] = { "cloud.scope.auto_owner": ( OWNER_CONFIRMATION_KIND_OWNER_AUTO_ASSIGNED ), "cmdb.cloud_scope.owner_force_transferred": ( OWNER_CONFIRMATION_KIND_OWNER_FORCE_TRANSFERRED ), # Every ``assign_scope_owner`` call. This is the ONLY record that # names the true predecessor: the stored rows carry the pending # card's ADDRESSEE, which is whoever the card was mailed to weeks # earlier and not necessarily the owner at transfer time. "cloud.scope.assign_owner": OWNER_CONFIRMATION_KIND_OWNER_ASSIGNED, } # Stored kinds whose whole content is "ownership moved A → B". Now that # ``cloud.scope.assign_owner`` is merged, each of these has a derived # twin written in the same transaction (verified in production # 2026-08-24: 53 of 53 leaf rows pair with an audit row within 5s), and # the twin is strictly better — correct predecessor, plus the acting # user. Rendering both would show the same handover twice. _SUPERSEDED_BY_DERIVED = {OWNER_CONFIRMATION_KIND_OWNER_TRANSFERRED} _DERIVED_TWIN_WINDOW = timedelta(seconds=5) @dataclass(frozen=True) class OwnerTimelineEntry: """One governance-timeline event, stored OR audit-derived. Field names mirror ``CmdbCloudScopeOwnerConfirmation`` so the API response schema and the MCP payload stay a single shape. ``derived`` tells consumers the entry has no backing row: it can be rendered and read, but never responded to, and its ``id`` is the audit-log id (stable across reloads, and namespaced away from confirmation ids because it comes from a different table's PK space). """ id: str scope_id: str kind: str status: str owner_person_id: str | None response_kind: str | None response_at: datetime | None responded_by_person_id: str | None transferred_to_person_id: str | None reason: str | None spend_recent_30d_usd: Decimal | None spend_recent_180d_usd: Decimal | None spend_last_month_usd: Decimal | None member_count_snapshot: int | None notified_at: datetime | None escalated_at: datetime | None parent_confirmation_id: str | None created_at: datetime derived: bool # Daily-reminder bookkeeping. 0/None on derived entries — an # audit-derived owner-change fact is not an outstanding ask. reminder_count: int last_reminded_at: datetime | None @property def sort_key(self) -> datetime: """The moment this row last developed — its response if it has one, otherwise when it was raised.""" return ( _aware(self.response_at) or _aware(self.created_at) or datetime.min.replace(tzinfo=UTC) ) @classmethod def from_row( cls, row: CmdbCloudScopeOwnerConfirmation, ) -> OwnerTimelineEntry: return cls( id=str(row.id), scope_id=str(row.scope_id), kind=row.kind, status=row.status, owner_person_id=( str(row.owner_person_id) if row.owner_person_id else None ), response_kind=row.response_kind, response_at=row.response_at, responded_by_person_id=( str(row.responded_by_person_id) if row.responded_by_person_id else None ), transferred_to_person_id=( str(row.transferred_to_person_id) if row.transferred_to_person_id else None ), reason=row.reason, spend_recent_30d_usd=row.spend_recent_30d_usd, spend_recent_180d_usd=row.spend_recent_180d_usd, spend_last_month_usd=row.spend_last_month_usd, member_count_snapshot=row.member_count_snapshot, notified_at=row.notified_at, escalated_at=row.escalated_at, parent_confirmation_id=( str(row.parent_confirmation_id) if row.parent_confirmation_id else None ), created_at=row.created_at, derived=False, reminder_count=int(row.reminder_count or 0), last_reminded_at=row.last_reminded_at, ) @classmethod def from_audit(cls, log: Any) -> OwnerTimelineEntry | None: """Synthesize an entry from an owner-change ``audit_logs`` row. Returns ``None`` when the row carries no usable new-owner id — an event that cannot name who took ownership is noise on a timeline whose whole purpose is answering "who owns this and since when". """ kind = _AUDIT_DERIVED_OWNER_EVENTS.get(log.action) if kind is None: return None detail = log.detail if isinstance(log.detail, dict) else {} new_owner = detail.get("new_owner_person_id") if not new_owner: return None previous_owner = detail.get("previous_owner_person_id") # ``prev == new`` happens on a first assignment recorded through # the generic path; there is no handover to show. if previous_owner and str(previous_owner) == str(new_owner): return None # ``cloud.scope.auto_owner`` records HOW the owner was picked # (``active_grant_fallback:first_by_person_id``, an org-unit # match, …). That attribution basis is the single most useful # thing on the row — an owner who was guessed from "first # grant holder by person id" deserves to be read as a guess. reason = detail.get("reason") return cls( id=str(log.id), scope_id=str(log.resource_id), kind=kind, status=OWNER_CONFIRMATION_STATUS_RESPONDED, owner_person_id=str(previous_owner) if previous_owner else None, response_kind=kind, response_at=log.created_at, responded_by_person_id=( str(log.actor_user_id) if log.actor_user_id else None ), transferred_to_person_id=str(new_owner), reason=str(reason) if reason else None, spend_recent_30d_usd=None, spend_recent_180d_usd=None, spend_last_month_usd=None, member_count_snapshot=None, notified_at=None, escalated_at=None, parent_confirmation_id=None, created_at=log.created_at, derived=True, reminder_count=0, last_reminded_at=None, ) class OwnerConfirmationService: """Read/write side of the owner-governance domain. Persists pending prompts, processes owner responses, and runs the 14-day escalation sweep. The cron handlers wrap this — direct API callers (the respond endpoints, MCP tools) also use this service. """ def __init__(self, db: AsyncSession) -> None: self.db = db self.audit = AuditService(db) self.notifications = NotificationService(db) # ── Spend & member snapshot ───────────────────────────────── async def _sum_spend( self, scope_id: uuid.UUID, *, days_back: int, end: date | None = None, ) -> Decimal: """Sum cost-facts in [end-days_back, end). ``end`` defaults to today. Inclusive-exclusive window so two adjacent calls (``recent`` and ``prior``) don't double-count the boundary day. """ end_date = end or date.today() start_date = end_date - timedelta(days=days_back) res = await self.db.execute( select(func.coalesce(func.sum(fact_engineering_amount()), 0)) .where( CmdbCostFact.scope_id == str(scope_id), CmdbCostFact.date >= start_date, CmdbCostFact.date < end_date, ) ) return Decimal(str(res.scalar() or 0)) async def _sum_spend_month( self, scope_id: uuid.UUID, *, year: int, month: int, ) -> Decimal: """Sum cost-facts in a calendar month.""" start_date = date(year, month, 1) # First day of NEXT month — exclusive upper bound. if month == 12: end_date = date(year + 1, 1, 1) else: end_date = date(year, month + 1, 1) res = await self.db.execute( select(func.coalesce(func.sum(fact_engineering_amount()), 0)) .where( CmdbCostFact.scope_id == str(scope_id), CmdbCostFact.date >= start_date, CmdbCostFact.date < end_date, ) ) return Decimal(str(res.scalar() or 0)) async def _count_members(self, scope_id: uuid.UUID) -> int: """Count distinct live-grant persons on a scope.""" res = await self.db.execute( select(func.count(func.distinct(CmdbCloudAccessGrant.person_id))) .where( CmdbCloudAccessGrant.scope_id == str(scope_id), CmdbCloudAccessGrant.status.in_(LIVE_GRANT_STATUSES), CmdbCloudAccessGrant.person_id.isnot(None), ) ) return int(res.scalar() or 0) async def _top_member_names( self, scope_id: uuid.UUID, *, limit: int = 3, ) -> list[str]: """Names of distinct live-grant persons — order by created_at desc. The current ``CmdbCloudAccessGrant`` schema doesn't store per-person usage frequency at write time; absent that, the most recently-added members are a reasonable "fresh accessors" proxy for the card. Future PRs can swap this for an actual usage-frequency join when that signal lands. """ # Two queries instead of one CAST-based join. The join used to be # ``grants.person_id == CAST(identity_persons.id AS CHAR(36))``, # and a CAST result carries the SERVER's default collation # (``utf8mb4_0900_ai_ci`` on MySQL 8) while the column carries the # table's (``utf8mb4_unicode_ci``) — MySQL then refuses the # comparison outright with 1267 "Illegal mix of collations", # regardless of how many rows match. It happens to work on the # production engine, which is exactly what made it dangerous: the # only path that reaches it is the cron, so nothing local ever # executed it and the first test to try turned red on # infrastructure rather than on behaviour. Splitting the join # removes the cast entirely and is engine-neutral. id_res = await self.db.execute( select(CmdbCloudAccessGrant.person_id) .where( CmdbCloudAccessGrant.scope_id == str(scope_id), CmdbCloudAccessGrant.status.in_(LIVE_GRANT_STATUSES), CmdbCloudAccessGrant.person_id.isnot(None), ) .order_by(CmdbCloudAccessGrant.created_at.desc()) .limit(limit * 3) ) # Grant order is the recency proxy the card documents, so keep it # and de-duplicate in Python rather than losing it to a DISTINCT. ordered_pids = [str(r[0]) for r in id_res.all() if r[0]] if not ordered_pids: return [] lookup_ids: list[uuid.UUID] = [] for pid in dict.fromkeys(ordered_pids): try: lookup_ids.append(uuid.UUID(pid)) except ValueError: # A grant row whose person_id isn't a UUID (legacy import # residue) simply has no name to show. continue if not lookup_ids: return [] name_res = await self.db.execute( select( IdentityPerson.id, IdentityPerson.display_name, IdentityPerson.work_email, ) .where(IdentityPerson.id.in_(lookup_ids)) ) name_by_id = { str(pid): (dn or em or "—") for pid, dn, em in name_res.all() } names: list[str] = [] seen: set[str] = set() for pid in ordered_pids: name = name_by_id.get(pid) if not name or name in seen: continue seen.add(name) names.append(name) if len(names) >= limit: break return names # ── Pending-row creation ──────────────────────────────────── async def _has_open_pending( self, scope_id: uuid.UUID, kind: str, ) -> CmdbCloudScopeOwnerConfirmation | None: """Find an outstanding pending row of the given kind on a scope. Used to guarantee at-most-one outstanding prompt per (scope, kind) — re-running the cron the same day MUST NOT produce a duplicate Feishu card. """ res = await self.db.execute( select(CmdbCloudScopeOwnerConfirmation) .where( CmdbCloudScopeOwnerConfirmation.scope_id == str(scope_id), CmdbCloudScopeOwnerConfirmation.kind == kind, CmdbCloudScopeOwnerConfirmation.status == OWNER_CONFIRMATION_STATUS_PENDING, ) .order_by(CmdbCloudScopeOwnerConfirmation.created_at.desc()) .limit(1) ) return res.scalar_one_or_none() async def create_semiannual_pending( self, scope: CmdbCloudScope, owner: IdentityPerson, *, trace_id: str, ) -> tuple[CmdbCloudScopeOwnerConfirmation, dict[str, Any]]: """Compose snapshot data, persist a pending row, dispatch the card. Returns ``(row, snapshot)`` so the caller can include the row id in its summary. The snapshot dict carries the same fields written to the row (for telemetry). """ # Race-safe at-most-one pending guard. The caller's # ``list_scopes_due_semiannual`` filters out scopes with an # existing pending row, but two cron ticks (or a manual retry # racing the cron) can both pass that filter and both INSERT. # Re-check inside the create path so a concurrent run gets a # short-circuit return instead of producing a duplicate Feishu # card. # # NB: this is a SELECT-then-INSERT TOCTOU, not a hard lock — # the window where both callers see ``existing is None`` is # narrow (microseconds between the SELECT and the flush). A # TiDB partial unique index would close it but TiDB doesn't # support partials. Worst-case outcome of the residual race # is ONE duplicate Feishu card per (scope, kind) per cron tick # — recoverable by either owner click (the other card's row # transitions to RESPONDED on the next sweep). Not a security # issue, just a UX nit. existing = await self._has_open_pending( scope.id, OWNER_CONFIRMATION_KIND_SEMIANNUAL_PENDING, ) if existing is not None: return existing, { "confirmation_id": str(existing.id), "spend_recent_30d_usd": str(existing.spend_recent_30d_usd or 0), "spend_recent_180d_usd": str(existing.spend_recent_180d_usd or 0), "member_count": existing.member_count_snapshot, "deduplicated": True, } snapshot = await self._semiannual_snapshot(scope) confirmation = CmdbCloudScopeOwnerConfirmation( id=uuid.uuid4(), scope_id=str(scope.id), owner_person_id=scope.owner_person_id, kind=OWNER_CONFIRMATION_KIND_SEMIANNUAL_PENDING, status=OWNER_CONFIRMATION_STATUS_PENDING, spend_recent_30d_usd=snapshot["spend_30"], spend_recent_180d_usd=snapshot["spend_180"], member_count_snapshot=snapshot["member_count"], notified_at=datetime.now(UTC), ) self.db.add(confirmation) await self.db.flush() confirmation_id = str(confirmation.id) await self._send_semiannual_card( scope, owner, confirmation_id, snapshot=snapshot, trace_id=trace_id, reminder_count=0, ) await self.audit.log( trace_id=trace_id, actor_type="system", actor_user_id=None, action="cloud.scope.owner_semiannual_prompt", status="success", resource_type="cmdb_cloud_scope", resource_id=str(scope.id), detail={ "confirmation_id": confirmation_id, "owner_person_id": scope.owner_person_id, "spend_30d_usd": str(snapshot["spend_30"]), "spend_180d_usd": str(snapshot["spend_180"]), "member_count": snapshot["member_count"], }, ) return confirmation, { "confirmation_id": confirmation_id, "spend_recent_30d_usd": str(snapshot["spend_30"]), "spend_recent_180d_usd": str(snapshot["spend_180"]), "member_count": snapshot["member_count"], } async def _semiannual_snapshot( self, scope: CmdbCloudScope, ) -> dict[str, Any]: """Spend / member figures for a semiannual card. Rolling 30/180 windows plus their immediately-prior windows (for the trend arrows). Recomputed for every send — including a reminder — because a card showing three-month-old numbers is worse than no numbers. The row's stored snapshot is written ONCE at creation and deliberately NOT refreshed by reminders: the timeline's contract is "what the owner saw when first asked". """ today = date.today() spend_30 = await self._sum_spend( scope.id, days_back=SPEND_WINDOW_30, end=today, ) spend_prior_30 = await self._sum_spend( scope.id, days_back=SPEND_WINDOW_30, end=today - timedelta(days=SPEND_WINDOW_30), ) spend_180 = await self._sum_spend( scope.id, days_back=SPEND_WINDOW_180, end=today, ) spend_prior_180 = await self._sum_spend( scope.id, days_back=SPEND_WINDOW_180, end=today - timedelta(days=SPEND_WINDOW_180), ) return { "spend_30": spend_30, "spend_prior_30": spend_prior_30, "spend_180": spend_180, "spend_prior_180": spend_prior_180, "member_count": await self._count_members(scope.id), "top_names": await self._top_member_names(scope.id, limit=3), } async def _send_semiannual_card( self, scope: CmdbCloudScope, owner: IdentityPerson, confirmation_id: str, *, snapshot: dict[str, Any], trace_id: str, reminder_count: int, first_asked_at: datetime | None = None, ) -> None: """Dispatch the semiannual card. Shared by the first ask and by every daily reminder so the two can never drift in content, buttons or deep-links.""" from app.domains.cmdb.owner_confirmation_cards import ( build_semiannual_card_fields, ) scope_id_str = str(scope.id) fields = build_semiannual_card_fields( scope_display_name=scope.display_name or scope.external_id, scope_environment=scope.environment, spend_recent_30d=snapshot["spend_30"], spend_prior_30d=snapshot["spend_prior_30"], spend_recent_180d=snapshot["spend_180"], spend_prior_180d=snapshot["spend_prior_180"], member_count=snapshot["member_count"], member_count_180d_ago=None, # historical headcount not yet tracked last_confirmed_at=_aware(scope.last_owner_confirmed_at), top_active_members=snapshot["top_names"], confirm_url=_owner_action_url( scope_id_str, confirmation_id, "semiannual_confirm", ), transfer_url=_owner_action_url( scope_id_str, confirmation_id, "semiannual_transfer", ), detail_url=_scope_detail_url(scope_id_str), lang=owner.preferred_language or "zh-CN", reminder_count=reminder_count, first_asked_at=first_asked_at, ) envelope = NotificationEnvelope( recipient=owner, related_resource_type="cmdb_cloud_scope", related_resource_id=scope.id, template_key="cloud.owner_semiannual_confirm", # ``url`` powers the email-channel cta_line fallback when # no template body is registered (rich-card envelopes # otherwise leave email recipients without an action URL). url=_owner_action_url( scope_id_str, confirmation_id, "semiannual_confirm", ), **fields, ) await self.notifications.dispatch(envelope, trace_id=trace_id) async def create_zero_spend_pending( self, scope: CmdbCloudScope, owner: IdentityPerson, *, billing_year: int, billing_month: int, trace_id: str, ) -> tuple[CmdbCloudScopeOwnerConfirmation, dict[str, Any]]: """Persist a pending zero-spend row + dispatch the alert card. Caller has already verified spend_last_month == 0 and the scope is older than the month boundary. """ # Same race-safe guard as semiannual: re-check inside the # create path so a manual retry colliding with the monthly # cron tick can't produce a duplicate zero-spend card. existing = await self._has_open_pending( scope.id, OWNER_CONFIRMATION_KIND_ZERO_SPEND_PENDING, ) if existing is not None: period_label = f"{billing_year}-{billing_month:02d}" return existing, { "confirmation_id": str(existing.id), "period": period_label, "member_count": existing.member_count_snapshot, "deduplicated": True, } snapshot = await self._zero_spend_snapshot( scope, billing_year=billing_year, billing_month=billing_month, ) confirmation = CmdbCloudScopeOwnerConfirmation( id=uuid.uuid4(), scope_id=str(scope.id), owner_person_id=scope.owner_person_id, kind=OWNER_CONFIRMATION_KIND_ZERO_SPEND_PENDING, status=OWNER_CONFIRMATION_STATUS_PENDING, spend_last_month_usd=snapshot["spend_last_month"], spend_recent_180d_usd=snapshot["spend_180"], member_count_snapshot=snapshot["member_count"], notified_at=datetime.now(UTC), ) self.db.add(confirmation) await self.db.flush() confirmation_id = str(confirmation.id) await self._send_zero_spend_card( scope, owner, confirmation_id, snapshot=snapshot, trace_id=trace_id, reminder_count=0, ) await self.audit.log( trace_id=trace_id, actor_type="system", actor_user_id=None, action="cloud.scope.owner_zero_spend_prompt", status="success", resource_type="cmdb_cloud_scope", resource_id=str(scope.id), detail={ "confirmation_id": confirmation_id, "owner_person_id": scope.owner_person_id, "period": snapshot["period_label"], "spend_last_month_usd": str(snapshot["spend_last_month"]), "spend_recent_90d_usd": str(snapshot["spend_90"]), "member_count": snapshot["member_count"], }, ) return confirmation, { "confirmation_id": confirmation_id, "period": snapshot["period_label"], "member_count": snapshot["member_count"], } async def _zero_spend_snapshot( self, scope: CmdbCloudScope, *, billing_year: int, billing_month: int, ) -> dict[str, Any]: """Spend / member figures for a zero-spend card. See ``_semiannual_snapshot`` for why a reminder recomputes these.""" today = date.today() return { "period_label": f"{billing_year}-{billing_month:02d}", "spend_last_month": await self._sum_spend_month( scope.id, year=billing_year, month=billing_month, ), "spend_90": await self._sum_spend( scope.id, days_back=SPEND_WINDOW_90, end=today, ), "spend_180": await self._sum_spend( scope.id, days_back=SPEND_WINDOW_180, end=today, ), "member_count": await self._count_members(scope.id), "top_names": await self._top_member_names(scope.id, limit=3), } async def _send_zero_spend_card( self, scope: CmdbCloudScope, owner: IdentityPerson, confirmation_id: str, *, snapshot: dict[str, Any], trace_id: str, reminder_count: int, first_asked_at: datetime | None = None, ) -> None: """Dispatch the zero-spend card. Shared by the first ask and by every daily reminder.""" from app.domains.cmdb.owner_confirmation_cards import ( build_zero_spend_card_fields, ) scope_id_str = str(scope.id) fields = build_zero_spend_card_fields( scope_display_name=scope.display_name or scope.external_id, scope_environment=scope.environment, period_label=snapshot["period_label"], spend_last_month=snapshot["spend_last_month"], spend_recent_90d=snapshot["spend_90"], spend_recent_180d=snapshot["spend_180"], member_count=snapshot["member_count"], top_members=snapshot["top_names"], scope_created_at=_aware(scope.created_at), keep_url=_owner_action_url( scope_id_str, confirmation_id, "zero_spend_keep", ), cleanup_url=_owner_action_url( scope_id_str, confirmation_id, "zero_spend_cleanup", ), handover_url=_owner_action_url( scope_id_str, confirmation_id, "zero_spend_handover", ), detail_url=_scope_detail_url(scope_id_str), lang=owner.preferred_language or "zh-CN", reminder_count=reminder_count, first_asked_at=first_asked_at, ) envelope = NotificationEnvelope( recipient=owner, related_resource_type="cmdb_cloud_scope", related_resource_id=scope.id, template_key="cloud.owner_zero_spend", url=_scope_detail_url(scope_id_str), **fields, ) await self.notifications.dispatch(envelope, trace_id=trace_id) # ── Owner-response endpoints ──────────────────────────────── async def _assert_owner_or_admin( self, scope_id: uuid.UUID, *, responder_person_id: str, actor_is_admin: bool, ) -> None: """Push owner-response authorization DOWN into the service so it travels with the state transition, not only in the callers. Defence-in-depth: today both the HTTP router and the MCP tool gate with ``_require_scope_owner_or_admin`` / the inline owner-or-admin check, but a future third caller must not be able to bypass simply by forgetting to repeat it. The sibling ``server_owner_confirmation_service.respond_confirm_all`` already self-enforces; this brings the cloud-scope twin to parity. """ if actor_is_admin: return scope = await self.db.get(CmdbCloudScope, scope_id) if scope is None or str(scope.owner_person_id or "") != str( responder_person_id, ): raise ForbiddenError( "Only the current scope owner or a cloud admin may respond " "to this confirmation.", ) async def _get_pending( self, scope_id: uuid.UUID, confirmation_id: uuid.UUID, ) -> CmdbCloudScopeOwnerConfirmation: res = await self.db.execute( select(CmdbCloudScopeOwnerConfirmation) .where( CmdbCloudScopeOwnerConfirmation.id == str(confirmation_id), CmdbCloudScopeOwnerConfirmation.scope_id == str(scope_id), ) .with_for_update() ) row = res.scalar_one_or_none() if row is None: raise NotFoundError( "CmdbCloudScopeOwnerConfirmation", str(confirmation_id), ) if row.status != OWNER_CONFIRMATION_STATUS_PENDING: raise AppException( code="OWNER_CONFIRMATION_NOT_PENDING", message=( "This confirmation has already been responded to or " "escalated; it can no longer be modified." ), status_code=409, ) return row async def respond_semiannual_confirm( self, scope_id: uuid.UUID, confirmation_id: uuid.UUID, *, responder_person_id: str, actor_user_id: uuid.UUID | None, trace_id: str, actor_is_admin: bool = False, ) -> CmdbCloudScopeOwnerConfirmation: """Owner clicks ✅ — current owner stays, clock resets.""" row = await self._get_pending(scope_id, confirmation_id) await self._assert_owner_or_admin( scope_id, responder_person_id=responder_person_id, actor_is_admin=actor_is_admin, ) scope = await self.db.get(CmdbCloudScope, scope_id) if scope is None or scope.deleted_at is not None: raise NotFoundError("CmdbCloudScope", str(scope_id)) if row.kind != OWNER_CONFIRMATION_KIND_SEMIANNUAL_PENDING: raise AppException( code="OWNER_CONFIRMATION_KIND_MISMATCH", message="This action is only valid on a semiannual prompt.", status_code=400, ) now = datetime.now(UTC) row.status = OWNER_CONFIRMATION_STATUS_RESPONDED row.response_kind = OWNER_CONFIRMATION_KIND_SEMIANNUAL_CONFIRMED row.response_at = now row.responded_by_person_id = responder_person_id scope.last_owner_confirmed_at = now await self.audit.log( trace_id=trace_id, actor_type="user" if actor_user_id else "system", actor_user_id=actor_user_id, action="cloud.scope.owner_semiannual_confirmed", status="success", resource_type="cmdb_cloud_scope", resource_id=str(scope.id), detail={ "confirmation_id": str(row.id), "owner_person_id": scope.owner_person_id, }, ) return row async def respond_zero_spend( self, scope_id: uuid.UUID, confirmation_id: uuid.UUID, *, action: str, reason: str | None, responder_person_id: str, actor_user_id: uuid.UUID | None, trace_id: str, actor_is_admin: bool = False, ) -> CmdbCloudScopeOwnerConfirmation: """Owner clicks one of keep / cleanup / handover. ``action`` ∈ {``keep``, ``cleanup``, ``handover``}. ``reason`` is required for ``keep`` (owner must justify keeping a zero-spend scope); ignored for the other two. """ row = await self._get_pending(scope_id, confirmation_id) await self._assert_owner_or_admin( scope_id, responder_person_id=responder_person_id, actor_is_admin=actor_is_admin, ) if row.kind != OWNER_CONFIRMATION_KIND_ZERO_SPEND_PENDING: raise AppException( code="OWNER_CONFIRMATION_KIND_MISMATCH", message="This action is only valid on a zero-spend prompt.", status_code=400, ) leaf = { "keep": OWNER_CONFIRMATION_KIND_ZERO_SPEND_KEEP, "cleanup": OWNER_CONFIRMATION_KIND_ZERO_SPEND_CLEANUP, "handover": OWNER_CONFIRMATION_KIND_ZERO_SPEND_HANDOVER_IT, }.get(action) if leaf is None: raise AppException( code="OWNER_CONFIRMATION_INVALID_ACTION", message=( "action must be one of 'keep' / 'cleanup' / 'handover'." ), status_code=400, ) if leaf == OWNER_CONFIRMATION_KIND_ZERO_SPEND_KEEP and not ( reason and reason.strip() ): raise AppException( code="OWNER_CONFIRMATION_REASON_REQUIRED", message=( "A written reason is required when keeping a " "zero-spend subscription." ), status_code=400, ) now = datetime.now(UTC) row.status = OWNER_CONFIRMATION_STATUS_RESPONDED row.response_kind = leaf row.response_at = now row.responded_by_person_id = responder_person_id if reason and reason.strip(): row.reason = reason.strip()[:4000] await self.audit.log( trace_id=trace_id, actor_type="user" if actor_user_id else "system", actor_user_id=actor_user_id, action="cloud.scope.owner_zero_spend_responded", status="success", resource_type="cmdb_cloud_scope", resource_id=str(scope_id), detail={ "confirmation_id": str(row.id), "leaf_kind": leaf, "reason": row.reason, }, ) return row async def mark_transferred( self, confirmation_id: uuid.UUID, *, new_owner_person_id: str, responder_person_id: str, actor_user_id: uuid.UUID | None, trace_id: str, previous_owner_person_id: str | None = None, ) -> CmdbCloudScopeOwnerConfirmation: """Owner clicked 🔄 and picked a new owner via ``assign_scope_owner``. The transfer-of-ownership side-effect already happened in ``assign_scope_owner`` (which also updates ``last_owner_confirmed_at``); here we just stamp the pending row as responded so the timeline + cron-idempotency logic sees a clean state. """ res = await self.db.execute( select(CmdbCloudScopeOwnerConfirmation) .where(CmdbCloudScopeOwnerConfirmation.id == str(confirmation_id)) .with_for_update() ) row = res.scalar_one_or_none() if row is None: raise NotFoundError( "CmdbCloudScopeOwnerConfirmation", str(confirmation_id), ) if row.status != OWNER_CONFIRMATION_STATUS_PENDING: # Idempotent: a parallel responder already marked it; return # the current row state instead of raising. return row row.status = OWNER_CONFIRMATION_STATUS_RESPONDED row.response_kind = OWNER_CONFIRMATION_KIND_SEMIANNUAL_TRANSFERRED row.response_at = datetime.now(UTC) row.responded_by_person_id = responder_person_id row.transferred_to_person_id = new_owner_person_id await self.audit.log( trace_id=trace_id, actor_type="user" if actor_user_id else "system", actor_user_id=actor_user_id, action="cloud.scope.owner_semiannual_transferred", status="success", resource_type="cmdb_cloud_scope", resource_id=row.scope_id, detail={ "confirmation_id": str(row.id), # The scope's owner AT TRANSFER TIME, supplied by # ``assign_scope_owner``. ``row.owner_person_id`` is the # card's ADDRESSEE — whoever it was mailed to, possibly # months earlier — and using it recorded a false # predecessor whenever ownership had moved in between # (one such row in production 2026-08-24: the card named # a leaver while the scope had since been handed to # someone else, so the audit claimed the leaver was the # outgoing owner). Falls back to the addressee only when # a caller cannot supply the real one. "previous_owner_person_id": ( previous_owner_person_id or row.owner_person_id ), "card_addressee_person_id": row.owner_person_id, "new_owner_person_id": new_owner_person_id, }, ) return row async def void_pending_for_owner_change( self, scope_id: str | uuid.UUID, *, new_owner_person_id: str | None, trace_id: str, actor_user_id: uuid.UUID | None = None, via: str, ) -> list[CmdbCloudScopeOwnerConfirmation]: """Close outstanding prompts addressed to a superseded owner. A ``_pending`` row names the person the card was sent to. When the scope changes hands while that card is still outstanding, the row becomes unanswerable AND blocking: * ``list_scopes_due_semiannual`` skips any scope that already has an open pending row, so the NEW owner is never prompted — the semiannual clock is frozen indefinitely. * the daily reminder sweep keeps pushing a card addressed to the previous owner — who may have left the company — every single morning. Voiding it releases both. The new owner is prompted again on the next cadence tick rather than inheriting a card composed from the previous owner's spend/member snapshot. Does NOT commit — callers run inside their own transaction (offboarding's force-transfer tick, the escalation sweep). """ res = await self.db.execute( select(CmdbCloudScopeOwnerConfirmation) .where( CmdbCloudScopeOwnerConfirmation.scope_id == str(scope_id), CmdbCloudScopeOwnerConfirmation.status == OWNER_CONFIRMATION_STATUS_PENDING, CmdbCloudScopeOwnerConfirmation.kind.in_( ( OWNER_CONFIRMATION_KIND_SEMIANNUAL_PENDING, OWNER_CONFIRMATION_KIND_ZERO_SPEND_PENDING, ), ), ) .with_for_update() ) rows = list(res.scalars().all()) voided: list[CmdbCloudScopeOwnerConfirmation] = [] now = datetime.now(UTC) for row in rows: # Defensive: a prompt already addressed to the incoming # owner is still valid — do not void the card the person # is about to answer. if ( new_owner_person_id and row.owner_person_id and str(row.owner_person_id) == str(new_owner_person_id) ): continue row.status = OWNER_CONFIRMATION_STATUS_RESPONDED row.response_kind = ( OWNER_CONFIRMATION_RESPONSE_VOIDED_OWNER_CHANGED ) row.response_at = now row.transferred_to_person_id = ( str(new_owner_person_id) if new_owner_person_id else None ) await self.audit.log( trace_id=trace_id, actor_type="user" if actor_user_id else "system", actor_user_id=actor_user_id, action="cloud.scope.owner_confirmation_voided", status="success", resource_type="cmdb_cloud_scope", resource_id=str(scope_id), detail={ "confirmation_id": str(row.id), "pending_kind": row.kind, "previous_owner_person_id": row.owner_person_id, "new_owner_person_id": ( str(new_owner_person_id) if new_owner_person_id else None ), "via": via, }, ) voided.append(row) return voided # ── Daily reminder sweep ──────────────────────────────────── async def remind_unanswered_owners( self, *, trace_id: str, max_per_run: int | None = None, ) -> dict[str, Any]: """Re-send every unanswered prompt to its OWNER, once a day. Replaces the 14-day escalate-to-IT sweep (retired 2026-08-24, product decision). The ask belongs to the owner: handing ~90 of other people's unanswered asks to 5 admins produced a chase queue and moved no decisions, and — because ``escalated`` was a status the respond endpoints refuse — an escalated prompt could never be answered by anyone at all. So: the same card, to the same person, every day, until they press one of its buttons. Three cases never get a card: * **Scope gone or decommissioned** — the question is moot. The prompt is closed as ``voided_scope_closed``. * **Owner changed while the prompt was open** — the card names somebody who is not the owner. Closed as ``voided_owner_changed``; the cadence cron then asks the new owner with a fresh prompt. (Reminding daily makes this case urgent rather than cosmetic: without it we would push to a departed colleague every single morning.) * **Owner is no longer active** — skipped, not closed. Left open on purpose: offboarding's force-transfer will re-point the scope shortly and the previous case then closes it. Nothing is pushed to a leaver in the meantime. ``max_per_run`` bounds one tick's fan-out. Default (``None``) is unbounded, which is the intended steady state — one card per open prompt per day, and the number of open prompts is bounded by the number of subscriptions. Set it to throttle a large backlog's first pass; the summary always reports ``deferred_over_cap`` so a capped run can never read as a drained queue. """ now = datetime.now(UTC) # At-most-once-per-day guard expressed as a minimum interval # rather than a calendar-day boundary: the cron fires at a fixed # local time, so a 20h floor yields exactly one reminder per day # while making a same-day manual re-trigger a no-op — with no # timezone arithmetic to get wrong. due_before = now - timedelta(hours=REMINDER_MIN_INTERVAL_HOURS) base_predicate = ( CmdbCloudScopeOwnerConfirmation.status == OWNER_CONFIRMATION_STATUS_PENDING, CmdbCloudScopeOwnerConfirmation.kind.in_( ( OWNER_CONFIRMATION_KIND_SEMIANNUAL_PENDING, OWNER_CONFIRMATION_KIND_ZERO_SPEND_PENDING, ), ), or_( CmdbCloudScopeOwnerConfirmation.last_reminded_at.is_(None), CmdbCloudScopeOwnerConfirmation.last_reminded_at < due_before, ), ) res = await self.db.execute( select(CmdbCloudScopeOwnerConfirmation) .where(*base_predicate) # Oldest first: the longest-unanswered ask is the one a cap # must never starve. .order_by(CmdbCloudScopeOwnerConfirmation.notified_at.asc()) .limit(max_per_run) .with_for_update(skip_locked=True) ) pending_rows: list[CmdbCloudScopeOwnerConfirmation] = list( res.scalars().all() ) deferred_over_cap = 0 if max_per_run is not None and len(pending_rows) == max_per_run: total_res = await self.db.execute( select(func.count()) .select_from(CmdbCloudScopeOwnerConfirmation) .where(*base_predicate) ) deferred_over_cap = max( 0, int(total_res.scalar() or 0) - len(pending_rows), ) summary: dict[str, Any] = { "reminded": 0, "voided_owner_changed": 0, "voided_scope_closed": 0, "skipped_owner_inactive": 0, # Open prompts whose owner is simply not due yet on the # cadence ladder. Reported rather than silently folded into # "nothing happened": a capped or tapered run must never be # readable as a drained queue. "deferred_by_cadence": 0, "errors": 0, "deferred_over_cap": deferred_over_cap, } if not pending_rows: return summary for pending in pending_rows: try: outcome = await self._remind_one(pending, trace_id=trace_id, now=now) except Exception: # noqa: BLE001 logger.exception( "remind_unanswered_owners: row %s failed", pending.id, ) summary["errors"] += 1 # Roll back to a clean session so the next row starts # from a known-good state (a failed flush poisons the # session for every subsequent statement). try: await self.db.rollback() except Exception: # noqa: BLE001 pass continue summary[outcome] = summary.get(outcome, 0) + 1 return summary async def _remind_one( self, pending: CmdbCloudScopeOwnerConfirmation, *, trace_id: str, now: datetime, ) -> str: """Handle one open prompt. Returns the summary key to increment. Commits per row: a failure on row N+1 must not roll back the reminders already dispatched for rows 1..N, or every one of those owners gets a second card on the next tick. """ scope = await self.db.get( CmdbCloudScope, uuid.UUID(pending.scope_id), ) if scope is None or scope.status == SCOPE_STATUS_CLOSED: pending.status = OWNER_CONFIRMATION_STATUS_RESPONDED pending.response_kind = ( OWNER_CONFIRMATION_RESPONSE_VOIDED_SCOPE_CLOSED ) pending.response_at = now await self.audit.log( trace_id=trace_id, actor_type="system", actor_user_id=None, action="cloud.scope.owner_confirmation_voided", status="success", resource_type="cmdb_cloud_scope", resource_id=pending.scope_id, detail={ "confirmation_id": str(pending.id), "pending_kind": pending.kind, "via": ( "reminder_sweep_scope_deleted" if scope is None else "reminder_sweep_scope_decommissioned" ), }, ) await self.db.commit() return "voided_scope_closed" # The card names ``pending.owner_person_id``. If the scope has # changed hands, that name is wrong — and under a DAILY cadence # a wrong name means pushing to the previous owner (often a # leaver) every morning until someone notices. if ( pending.owner_person_id and scope.owner_person_id and str(pending.owner_person_id) != str(scope.owner_person_id) ): await self.void_pending_for_owner_change( pending.scope_id, new_owner_person_id=scope.owner_person_id, trace_id=trace_id, via="reminder_sweep_stale_addressee", ) await self.db.commit() return "voided_owner_changed" owner = ( await self.db.get( IdentityPerson, uuid.UUID(pending.owner_person_id), ) if pending.owner_person_id else None ) if ( owner is None or owner.deleted_at is not None or owner.employment_status != "active" or not owner.work_email ): # Left open on purpose — see ``remind_unanswered_owners``. # Stamp ``last_reminded_at`` anyway so a skipped row doesn't # get re-examined on every manual trigger within the day. pending.last_reminded_at = now await self.db.commit() return "skipped_owner_inactive" confirmation_id = str(pending.id) next_count = int(pending.reminder_count or 0) + 1 first_asked_at = _aware(pending.notified_at) or _aware(pending.created_at) # ── Cadence taper (2026-08-30) ────────────────────────────── # # This sweep's own docstring defends daily-forever, and that # defence still holds against what it was actually compared # with in 2026-08-24: handing the ask to somebody else. Nothing # here goes back to that — the ask stays with the owner and # stays open indefinitely. # # What was never compared was "keep asking, more slowly". Over # the last 30 days this template averaged 7.6 cards per owner. # An owner who has ignored the same card twenty times is not # moved by the twenty-first; the only thing those twenty # reliably achieve is teaching them to filter this sender — # taking some future card that does matter with it. The ladder # keeps the first few daily (when nearly all responses arrive) # and stretches the tail to weekly, so the ask survives and the # noise does not. ``reminder_count`` already tracks the rung, so # this needs no extra state. gap_days = cadence.required_gap_days( "cloud.owner_semiannual_confirm", int(pending.reminder_count or 0), ) last_sent = _aware(pending.last_reminded_at) # The taper may only ever LENGTHEN the interval. The daily rungs # are already governed by this module's own # ``REMINDER_MIN_INTERVAL_HOURS`` floor (20h, deliberately under # 24 so a fixed-time cron never skips a day), and layering a # second, slightly stricter "24h minus slack" test on top of it # would quietly drop a daily reminder whenever the two disagree — # a noise change silently becoming a delivery change. So the # ladder only engages once it asks for more than a day. if last_sent is not None and gap_days > 1: # Same 2h slack as the shared helper: crons fire on a fixed # wall clock, so an exact N-day comparison skips a rung # whenever the tick runs a few seconds early. due_at = last_sent + timedelta(days=gap_days) - timedelta(hours=2) if now < due_at: return "deferred_by_cadence" if pending.kind == OWNER_CONFIRMATION_KIND_SEMIANNUAL_PENDING: snapshot = await self._semiannual_snapshot(scope) await self._send_semiannual_card( scope, owner, confirmation_id, snapshot=snapshot, trace_id=trace_id, reminder_count=next_count, first_asked_at=first_asked_at, ) else: # Zero-spend prompts carry no billing period on the row, so # derive the one the prompt was about: the cron creates them # on the 1st for the PREVIOUS calendar month, so it is the # month before ``notified_at``'s month. anchor = first_asked_at or now year, month = anchor.year, anchor.month - 1 if month == 0: year, month = year - 1, 12 snapshot = await self._zero_spend_snapshot( scope, billing_year=year, billing_month=month, ) await self._send_zero_spend_card( scope, owner, confirmation_id, snapshot=snapshot, trace_id=trace_id, reminder_count=next_count, first_asked_at=first_asked_at, ) pending.reminder_count = next_count pending.last_reminded_at = now await self.audit.log( trace_id=trace_id, actor_type="system", actor_user_id=None, action="cloud.scope.owner_confirmation_reminded", status="success", resource_type="cmdb_cloud_scope", resource_id=pending.scope_id, detail={ "confirmation_id": confirmation_id, "pending_kind": pending.kind, "owner_person_id": pending.owner_person_id, "reminder_count": next_count, "first_asked_at": ( first_asked_at.isoformat() if first_asked_at else None ), }, ) # Bookkeeping commits only after a successful dispatch: a card # that never left must not consume today's reminder slot. await self.db.commit() return "reminded" # ── Cron-eligible scope queries ───────────────────────────── async def _resolve_first_observed_at( self, *, now: datetime, ) -> datetime: """Round-51 L3-P0: return the platform-wide first-observation timestamp for the owner-confirmation feature, stamping it atomically on first call. Every scope that pre-existed the feature must wait at least ``min_age_days`` AFTER this stamp before being eligible — without this, the first cron tick would see ~N scopes with ``last_owner_confirmed_at IS NULL`` and ``created_at`` months ago, all of them matching the SEMIANNUAL_PENDING window simultaneously, blasting every owner with cards on day 1. """ from app.integrations.models import PlatformSetting row = await self.db.get( PlatformSetting, OWNER_CONFIRMATION_FIRST_OBSERVED_KEY, ) if row is not None and row.value: raw = row.value if isinstance(row.value, str) else ( row.value.get("at") if isinstance(row.value, dict) else None ) if raw: try: parsed = datetime.fromisoformat(str(raw)) if parsed.tzinfo is None: parsed = parsed.replace(tzinfo=UTC) return parsed except ValueError: pass # First tick on this tenant — stamp ``now``. Best-effort: if the # write fails (transient DB) the next tick will retry. We use # ``merge`` to handle a concurrent first-tick race so two # workers both starting at the same time still converge. # Round-51 regression-review fix: the caller's session # (`scan_session` in cloud_owner_handlers.py:100) exits the # `async with` block WITHOUT calling `commit()` — `flush()` # alone is rolled back when the txn closes, so the stamp # never persists and every subsequent tick re-stamps `now`. # Effect: `eligible_for_null_scopes` is permanently False, the # cooldown becomes a permanent silencer, NULL-confirmed legacy # scopes never escalate to semiannual. Commit here so the # stamp survives independently of the caller's commit # discipline. stamp = PlatformSetting( key=OWNER_CONFIRMATION_FIRST_OBSERVED_KEY, value={"at": now.isoformat()}, ) async with best_effort( self.db, label="owner_confirmation_service best-effort step", log=logger, ) as be_1250_step: await self.db.merge(stamp) await self.db.commit() if not be_1250_step.ok: logger.error( "owner_confirmation: failed to stamp first_observed_at", ) return now async def list_scopes_due_semiannual( self, *, now: datetime | None = None, min_age_days: int = DEFAULT_NEWBORN_GUARD_DAYS, ) -> Sequence[CmdbCloudScope]: """Non-prod active scopes whose owner is due for confirmation. Eligibility: * environment != prod * deleted_at IS NULL * status = active * owner_person_id IS NOT NULL * last_owner_confirmed_at IS NULL OR last_owner_confirmed_at < now - 180d * created_at < now - ``min_age_days`` (newborn guard; default 30 so a freshly-synced scope doesn't get a card the same day. Pass 0 for an init sweep on a tenant whose scopes were all bulk-imported within the last month — see ``cmdb.owner_semiannual_confirmation`` config override.) * Round-51 L3-P0: also created_at < ``first_observed_at - min_age_days`` for scopes whose ``last_owner_confirmed_at`` is NULL — silences the launch-day cold-bomb. Scopes with a non-NULL ``last_owner_confirmed_at`` skip this gate (already part of the existing rhythm). * NO outstanding semiannual_pending row (de-dup against re-runs) """ if min_age_days < 0: raise ValueError(f"min_age_days must be >= 0, got {min_age_days}") now = now or datetime.now(UTC) cutoff_confirmed = now - timedelta(days=SEMIANNUAL_DAYS) cutoff_newborn = now - timedelta(days=min_age_days) first_observed_at = await self._resolve_first_observed_at(now=now) # Scopes whose ``last_owner_confirmed_at`` is NULL (i.e. they # pre-existed the feature) are eligible only after # ``first_observed_at + min_age_days`` of wall clock has # elapsed. This gives the cron a cooling-off period to # suppress the first-tick fanout — every legacy NULL scope # gets at LEAST ``min_age_days`` between Helix going live and # any card being dispatched, so an operator can backfill # ``last_owner_confirmed_at`` for known-good legacy scopes # before the cron starts cold-bombing. eligible_for_null_scopes = ( now - first_observed_at >= timedelta(days=min_age_days) ) existing_pending = ( select(CmdbCloudScopeOwnerConfirmation.scope_id) .where( CmdbCloudScopeOwnerConfirmation.kind == OWNER_CONFIRMATION_KIND_SEMIANNUAL_PENDING, CmdbCloudScopeOwnerConfirmation.status == OWNER_CONFIRMATION_STATUS_PENDING, ) .scalar_subquery() ) # Round-51 L3-P0: branch on whether the "first-observed cooldown" # has elapsed. If still inside the cooldown, ONLY ever-confirmed # scopes whose 180d window expired qualify (legacy never- # confirmed scopes are silenced). After the cooldown, the # original both-clause behaviour returns. if eligible_for_null_scopes: confirmed_clause = or_( CmdbCloudScope.last_owner_confirmed_at.is_(None), CmdbCloudScope.last_owner_confirmed_at < cutoff_confirmed, ) else: confirmed_clause = ( CmdbCloudScope.last_owner_confirmed_at < cutoff_confirmed ) res = await self.db.execute( select(CmdbCloudScope) .where( CmdbCloudScope.deleted_at.is_(None), CmdbCloudScope.status == SCOPE_STATUS_ACTIVE, CmdbCloudScope.environment.notin_(PROD_ENVIRONMENTS), CmdbCloudScope.owner_person_id.isnot(None), CmdbCloudScope.created_at < cutoff_newborn, confirmed_clause, CmdbCloudScope.id.notin_(existing_pending), # Never ask an owner to confirm a cost-allocation row # from a decommissioned tenant — there is no access to # confirm, and 30 of the 37 imported legacy-billing rows # carry an owner (see migration 466). CmdbCloudScope.scope_type.in_(TENANT_LEVEL_SCOPE_TYPES), ) # MySQL/TiDB sort NULLs first under ASC by default — no # explicit ``NULLS FIRST`` clause needed (which would emit # PostgreSQL-only syntax and fail with 1064 on TiDB). See # memory: feedback_tidb_fk_strictness sibling — order-by # nulls semantics also differ across dialects. .order_by(CmdbCloudScope.last_owner_confirmed_at.asc()) ) return list(res.scalars().all()) async def list_scopes_with_zero_spend_last_month( self, *, billing_year: int, billing_month: int, ) -> list[CmdbCloudScope]: """Non-prod active scopes whose last-calendar-month spend is exactly $0. Eligibility: * environment != prod * deleted_at IS NULL * status = active * owner_person_id IS NOT NULL * scope.created_at < first day of the billing month * AT LEAST one cost-facts row in [month_start - 60d, month_start) — Round-51 L3-P0 ingest-presence guard. A scope that genuinely never had any cost data ingested (mis-configured driver, a cloud account whose CUR/billing export hasn't been wired yet) would otherwise look like "zero spend" and trigger the card on every monthly tick, blasting owners for an integration problem. Require evidence that cost data IS flowing for this scope before claiming "they had zero spend last month". * SUM(cost_facts.amount_usd) WHERE date in [month_start, next_month_start) = 0 * NO outstanding zero_spend_pending row created this run * Round-51 L3-P0 first-observation guard: platform-wide ``first_observed_at`` stamp + 30d cooling-off period. Mirrors the semiannual cron. Note: a scope with NO cost-facts rows at all SUMs to zero — that matches the operational definition we want, BUT only if data IS flowing for that scope. The ingest-presence pre-filter ensures we don't conflate "no driver ingest" with "zero spend". """ from app.integrations.models import PlatformSetting # First-observation cooldown — silence the first-tick cold bomb # of every non-prod scope that genuinely had $0 spend last # month before Helix was even alive. now = datetime.now(UTC) observed_row = await self.db.get( PlatformSetting, OWNER_CONFIRMATION_FIRST_OBSERVED_KEY, ) first_observed_at: datetime if observed_row is not None and observed_row.value: raw = observed_row.value if isinstance(observed_row.value, str) else ( observed_row.value.get("at") if isinstance(observed_row.value, dict) else None ) try: first_observed_at = datetime.fromisoformat(str(raw)) if raw else now if first_observed_at.tzinfo is None: first_observed_at = first_observed_at.replace(tzinfo=UTC) except ValueError: first_observed_at = now else: # Stamp on first observation (zero_spend is monthly so this # branch fires once per tenant). Round-51 regression-review # fix: same caller-doesn't-commit issue as # ``_resolve_first_observed_at`` — flush() alone is rolled # back when the scan_session txn closes. Commit here so # the stamp survives the txn boundary. stamp = PlatformSetting( key=OWNER_CONFIRMATION_FIRST_OBSERVED_KEY, value={"at": now.isoformat()}, ) async with best_effort( self.db, label="owner_confirmation_service best-effort step", log=logger, ) as be_1412_step: await self.db.merge(stamp) await self.db.commit() if not be_1412_step.ok: logger.error( "zero_spend: failed to stamp first_observed_at", ) first_observed_at = now if (now - first_observed_at) < timedelta( days=DEFAULT_NEWBORN_GUARD_DAYS, ): logger.info( "zero_spend: still inside first-observation cooldown " "(first_observed_at=%s, age=%s days, threshold=%s); " "returning empty candidate list", first_observed_at.isoformat(), (now - first_observed_at).days, DEFAULT_NEWBORN_GUARD_DAYS, ) return [] month_start = date(billing_year, billing_month, 1) if billing_month == 12: month_end = date(billing_year + 1, 1, 1) else: month_end = date(billing_year, billing_month + 1, 1) ingest_presence_window_start = month_start - timedelta(days=60) ingest_presence_subq = ( select(CmdbCostFact.scope_id) .where( CmdbCostFact.date >= ingest_presence_window_start, CmdbCostFact.date < month_start, CmdbCostFact.scope_id.isnot(None), ) .group_by(CmdbCostFact.scope_id) .scalar_subquery() ) # Per-scope spend sum subquery. spend_subq = ( select( CmdbCostFact.scope_id.label("scope_id"), func.sum(fact_engineering_amount()).label("total"), ) .where( CmdbCostFact.date >= month_start, CmdbCostFact.date < month_end, CmdbCostFact.scope_id.isnot(None), ) .group_by(CmdbCostFact.scope_id) .subquery() ) # Pre-launch hardening 2026-05-17 (Round-19 audit, finding B): # the predicate was previously ``created_at >= month_end`` which # is the START of the NEXT month — i.e. it only matched rows # created in the future. The dedup subquery was therefore always # empty and every scope was returned every tick. ``_has_open_ # pending`` at create time saved us from sending duplicate cards, # but the eligibility query was misleading and one refactor away # from real spam. Anchor to the cron-tick instant so a re-run # the same month (or a manual re-trigger after a partial cron # failure) skips scopes already prompted in the current sweep. billing_month_start_dt = datetime.combine( month_start, datetime.min.time(), tzinfo=UTC, ) existing_pending = ( select(CmdbCloudScopeOwnerConfirmation.scope_id) .where( CmdbCloudScopeOwnerConfirmation.kind == OWNER_CONFIRMATION_KIND_ZERO_SPEND_PENDING, CmdbCloudScopeOwnerConfirmation.status == OWNER_CONFIRMATION_STATUS_PENDING, # Rows created during (or after) this billing month — # the only rows that represent "already prompted for # this period". Pending rows from prior months are # treated as stale and the scope is re-eligible (the # operator-facing semantics of "monthly zero-spend # alert" require a fresh card each new month). CmdbCloudScopeOwnerConfirmation.created_at >= billing_month_start_dt, ) .scalar_subquery() ) res = await self.db.execute( select(CmdbCloudScope) .outerjoin(spend_subq, spend_subq.c.scope_id == CmdbCloudScope.id) .where( CmdbCloudScope.deleted_at.is_(None), CmdbCloudScope.status == SCOPE_STATUS_ACTIVE, CmdbCloudScope.environment.notin_(PROD_ENVIRONMENTS), CmdbCloudScope.owner_person_id.isnot(None), CmdbCloudScope.created_at < datetime.combine( month_start, datetime.min.time(), tzinfo=UTC, ), # ``coalesce(total, 0) = 0`` — covers both "no rows" and # "rows summing to 0". func.coalesce(spend_subq.c.total, 0) == 0, CmdbCloudScope.id.notin_(existing_pending), # Round-51 L3-P0 ingest-presence: cost data MUST be # flowing for this scope (≥1 fact row in the trailing # 60d). Otherwise a misconfigured driver looks identical # to a genuinely-idle scope and the cron pings the # owner for the operator's mistake. CmdbCloudScope.id.in_(ingest_presence_subq), ) .order_by(CmdbCloudScope.created_at.asc()) ) return list(res.scalars().all()) # ── Detail-page timeline query ────────────────────────────── async def list_for_scope( self, scope_id: uuid.UUID, *, limit: int = 100, ) -> list[CmdbCloudScopeOwnerConfirmation]: res = await self.db.execute( select(CmdbCloudScopeOwnerConfirmation) .where(CmdbCloudScopeOwnerConfirmation.scope_id == str(scope_id)) .order_by(CmdbCloudScopeOwnerConfirmation.created_at.desc()) .limit(limit) ) return list(res.scalars().all()) async def list_timeline_for_scope( self, scope_id: uuid.UUID, *, limit: int = 200, ) -> list[OwnerTimelineEntry]: """Full owner-governance timeline for one scope, newest first. The union of two sources: * stored ``cmdb_cloud_scope_owner_confirmations`` rows — every prompt, response, transfer and escalation; * owner-CHANGE events that only ever landed in ``audit_logs`` (see ``_AUDIT_DERIVED_OWNER_EVENTS``), synthesized as read-only entries. Without the second source the timeline silently omits how a scope acquired its owner in the first place and every offboarding-driven handover — which is exactly the history a reader opens this view to see. Both API and MCP go through here so the two surfaces cannot drift. """ stored = await self.list_for_scope(scope_id, limit=limit) from app.audit.models import AuditLog audit_res = await self.db.execute( select(AuditLog) .where( AuditLog.resource_type == "cmdb_cloud_scope", AuditLog.resource_id == str(scope_id), AuditLog.action.in_(tuple(_AUDIT_DERIVED_OWNER_EVENTS)), AuditLog.status == "success", ) .order_by(AuditLog.created_at.desc()) .limit(limit) ) entries: list[OwnerTimelineEntry] = [] derived_at: list[datetime] = [] for log_row in audit_res.scalars().all(): entry = OwnerTimelineEntry.from_audit(log_row) if entry is not None: entries.append(entry) stamp = _aware(entry.created_at) if stamp is not None: derived_at.append(stamp) for row in stored: if row.kind in _SUPERSEDED_BY_DERIVED: stamp = _aware(row.created_at) if stamp is not None and any( abs(stamp - d) <= _DERIVED_TWIN_WINDOW for d in derived_at ): continue entries.append(OwnerTimelineEntry.from_row(row)) # Sort on the moment the row's LATEST development happened, not # on when it was created. A card raised on 05-19 and answered on # 08-24 belongs next to 08-24: sorting it by 05-19 while the UI # shows its 08-24 response is what made the column read # non-monotonically (05-31 → 08-24 → 04-29) and the whole tab # unreadable. ``created_at`` is tz-aware on audit_logs and can be # naive on the confirmations table, so coerce before comparing. entries.sort(key=lambda e: e.sort_key, reverse=True) return entries[:limit] __all__ = [ "REMINDER_MIN_INTERVAL_HOURS", "SEMIANNUAL_DAYS", "OwnerConfirmationService", "OwnerTimelineEntry", ]