"""催办 —— chasing the people who owe a required acknowledgement. The ladder is the whole design. This platform already learned, with real numbers, what happens without one: asset-inventory reminders were averaging **16.7 pushes per person per month** and cloud-owner confirmations 7.6. Somebody who has ignored twenty identical cards is not going to be moved by the twenty-first; the only thing those cards reliably achieve is training the reader to filter the whole source out — including the one card that will matter later. So this reuses ``app.notifications.cadence`` rather than counting days itself, and **lengthens** its ladder rather than shortening it: * First chase **7 days** after the campaign went out, not the next morning. The material takes minutes to read and most people do it in the first week; chasing before that spends the channel on people who were going to comply anyway. * Then the ladder — 7, 7, 14 days, and every 14 after that. Never stops while the ask is open (an open ask stays open), but the second month costs two pushes rather than thirty. * **Managers are told once, in a digest, after 21 days** — one card naming every report who is outstanding, not one card per report. The platform has an explicit finding on this: handing ninety people's todos to five admins "produced a chase queue and moved no decisions". A manager digest is the escalation that names the situation without transferring the work. Three rules the cadence module documents and this caller has to honour: 1. **The first message is not a chase.** The campaign's own send is that message; the ladder only governs repeats. 2. **Settle the account.** ``cadence.clear`` runs the moment somebody acknowledges. Without it their next unrelated ask inherits this counter and starts on the two-week rung — a "noise reduction" that turns a fresh problem into near-silence. 3. **Per-tick caps, oldest first, and say what was held back.** A run that quietly drops half the queue looks identical to a healthy one. """ from __future__ import annotations import logging import uuid from datetime import UTC, datetime from typing import Any from sqlalchemy import select, update from sqlalchemy.ext.asyncio import AsyncSession from app.domains.security_awareness.catalog import ( CAMPAIGN_SENT, COMPLETION_ACK, ) from app.domains.security_awareness.models import ( SecawCampaign, SecawCampaignTarget, SecawMaterial, ) from app.domains.security_awareness.service import pick_block from app.identity.person_models import IdentityPerson from app.notifications import cadence logger = logging.getLogger(__name__) TEMPLATE_REMINDER = "security_awareness.campaign.reminder" TEMPLATE_MANAGER = "security_awareness.campaign.manager_digest" #: Days after the send before the FIRST chase. The material takes #: minutes; chasing on day 1 spends the channel on people who were about #: to comply anyway. FIRST_CHASE_AFTER_DAYS = 7 #: Days after the send before a manager hears about it. One digest, then #: the manager's own ladder rung. ESCALATE_TO_MANAGER_AFTER_DAYS = 21 #: Per-tick ceilings. A cron that tries to page the entire unacknowledged #: population in one pass is one bad campaign away from being the #: platform's loudest sender. MAX_EMPLOYEE_PUSHES_PER_TICK = 120 MAX_MANAGER_PUSHES_PER_TICK = 30 def install_ladder() -> None: """Register this domain's cadence override. Called at import time. The platform default is 1, 1, 1, 2, 3, 5 then weekly, which is right for "your laptop is unaccounted for" and wrong for "please read a comic": daily chasing on required reading is how the immutable security category gets tuned out, and then the flows that genuinely cannot be muted stop being read too. Longer only — the ladder may be stretched, never tightened. """ cadence.LADDERS[TEMPLATE_REMINDER] = (7, 7, 14) cadence.TAIL_OVERRIDES[TEMPLATE_REMINDER] = 14 cadence.LADDERS[TEMPLATE_MANAGER] = (14,) cadence.TAIL_OVERRIDES[TEMPLATE_MANAGER] = 28 install_ladder() def _as_utc(value: datetime | None) -> datetime | None: """Stored DATETIMEs come back naive. Compare in one space rather than crashing on an offset-naive/aware subtraction.""" if value is None: return None return value if value.tzinfo else value.replace(tzinfo=UTC) class ReminderService: def __init__(self, db: AsyncSession): self.db = db async def run( self, *, trace_id: str | uuid.UUID = "", now: datetime | None = None, ) -> dict[str, Any]: """One tick. Returns counts plus what was held back.""" now = now or datetime.now(UTC) campaigns = (await self.db.execute( select(SecawCampaign).where( SecawCampaign.mandatory.is_(True), SecawCampaign.completion_mode == COMPLETION_ACK, # ``sent`` only: a CLOSED campaign has stopped asking, so # chasing its stragglers would contradict the act of # closing it. SecawCampaign.status == CAMPAIGN_SENT, SecawCampaign.sent_at.is_not(None), ) )).scalars().all() due: list[tuple[SecawCampaign, SecawCampaignTarget, IdentityPerson]] = [] escalate: list[tuple[SecawCampaign, SecawCampaignTarget, IdentityPerson]] = [] for campaign in campaigns: sent_at = _as_utc(campaign.sent_at) if sent_at is None: continue age_days = (now - sent_at).days if age_days < FIRST_CHASE_AFTER_DAYS: continue rows = (await self.db.execute( select(SecawCampaignTarget, IdentityPerson) .join( IdentityPerson, IdentityPerson.id == SecawCampaignTarget.person_id, ) .where( SecawCampaignTarget.campaign_id == campaign.id, SecawCampaignTarget.acknowledged_at.is_(None), # Somebody who has left is skipped here rather than # left to the outbox's leaver gate: the gate marks # the row dead, which is correct but shows up in the # console as a delivery failure rather than as a # deliberate skip. IdentityPerson.employment_status != "left", IdentityPerson.deleted_at.is_(None), ) .order_by(SecawCampaignTarget.created_at.asc()) )).all() for target, person in rows: due.append((campaign, target, person)) if age_days >= ESCALATE_TO_MANAGER_AFTER_DAYS: escalate.append((campaign, target, person)) sent, held = await self._chase_employees(due, trace_id=trace_id, now=now) managers, m_held = await self._chase_managers( escalate, trace_id=trace_id, now=now, ) summary = { "campaigns": len(campaigns), "outstanding": len(due), "employee_pushes": sent, # Named explicitly: a run that quietly drops half the queue # looks identical to a healthy one. "employee_held_back": held, "manager_pushes": managers, "manager_held_back": m_held, } logger.info("secaw.reminders %s", summary) return summary async def _chase_employees( self, due: list[tuple[SecawCampaign, SecawCampaignTarget, IdentityPerson]], *, trace_id: str | uuid.UUID, now: datetime, ) -> tuple[int, int]: from app.notifications import fanout envelopes: list[Any] = [] recorded: list[tuple[uuid.UUID, str]] = [] nudged_target_ids: list[uuid.UUID] = [] held = 0 for campaign, target, person in due: subject_key = str(campaign.id) if not await cadence.should_send( self.db, recipient_id=person.id, template_key=TEMPLATE_REMINDER, subject_key=subject_key, now=now, ): continue if len(envelopes) >= MAX_EMPLOYEE_PUSHES_PER_TICK: held += 1 continue env = await self._employee_envelope(campaign, person, now) if env is None: continue envelopes.append(env) recorded.append((person.id, subject_key)) nudged_target_ids.append(target.id) if not envelopes: return 0, held result = await fanout.enqueue_then_drain( self.db, envelopes, trace_id=str(trace_id or ""), label="security_awareness.reminder", ) for person_id, subject_key in recorded: await cadence.record_sent( self.db, recipient_id=person_id, template_key=TEMPLATE_REMINDER, subject_key=subject_key, now=now, ) # Count the automatic chase on the target row, exactly like an # admin-triggered resend. ``resend_count`` is what the evidence # CSV prints as "reminders_sent", and an auditor reading a 0 next # to somebody who was chased four times by the ladder would draw # the wrong conclusion about whether we followed up — which is # one of the things the file exists to answer. ``last_sent_at`` # moves for the same reason: it is "when did we last ask them", # and the ladder asking is asking. if nudged_target_ids: await self.db.execute( update(SecawCampaignTarget) .where(SecawCampaignTarget.id.in_(nudged_target_ids)) .values( resend_count=SecawCampaignTarget.resend_count + 1, last_sent_at=now, ) ) await self.db.commit() return int(result["enqueued"]), held async def _chase_managers( self, escalate: list[tuple[SecawCampaign, SecawCampaignTarget, IdentityPerson]], *, trace_id: str | uuid.UUID, now: datetime, ) -> tuple[int, int]: """One digest per manager, naming every outstanding report. NOT one card per report: handing somebody a queue of other people's todos produces a chase queue and moves no decisions — the platform has an explicit finding on exactly this shape. """ from app.notifications import fanout # Keyed on (manager, campaign), not on the manager alone. One # digest per manager would have to pick a campaign to title # itself with and to link to — and it would then list people who # owe a DIFFERENT piece of reading under that title, while both # campaigns shared one cadence counter so the second one could # never chase on its own schedule. A manager with reports # outstanding on two campaigns is being asked about two things. by_manager: dict[ tuple[uuid.UUID, uuid.UUID], tuple[SecawCampaign, list[IdentityPerson]], ] = {} for campaign, _target, person in escalate: manager_id = person.manager_person_id if not manager_id or manager_id == person.id: continue bucket = by_manager.setdefault( (manager_id, campaign.id), (campaign, []), ) bucket[1].append(person) if not by_manager: return 0, 0 managers = { m.id: m for m in (await self.db.execute( select(IdentityPerson).where( IdentityPerson.id.in_({mid for mid, _cid in by_manager}), IdentityPerson.employment_status != "left", IdentityPerson.deleted_at.is_(None), ) )).scalars().all() } envelopes: list[Any] = [] recorded: list[tuple[uuid.UUID, str]] = [] held = 0 for (manager_id, _campaign_id), (campaign, people) in by_manager.items(): manager = managers.get(manager_id) if manager is None: continue subject_key = str(campaign.id) if not await cadence.should_send( self.db, recipient_id=manager.id, template_key=TEMPLATE_MANAGER, subject_key=subject_key, now=now, ): continue if len(envelopes) >= MAX_MANAGER_PUSHES_PER_TICK: held += 1 continue env = await self._manager_envelope(campaign, manager, people, now) if env is None: continue envelopes.append(env) recorded.append((manager.id, subject_key)) if not envelopes: return 0, held result = await fanout.enqueue_then_drain( self.db, envelopes, trace_id=str(trace_id or ""), label="security_awareness.reminder.manager", ) for person_id, subject_key in recorded: await cadence.record_sent( self.db, recipient_id=person_id, template_key=TEMPLATE_MANAGER, subject_key=subject_key, now=now, ) await self.db.commit() return int(result["enqueued"]), held # ── Envelopes ─────────────────────────────────────────────────── async def _employee_envelope( self, campaign: SecawCampaign, person: IdentityPerson, now: datetime, ) -> Any: from app.core.locale import resolve_lang from app.notifications.service import ( _SEE_TEMPLATE_SENTINEL, LocalizedText, NotificationEnvelope, ) material = await self.db.get(SecawMaterial, campaign.material_id) if material is None: # pragma: no cover — FK guarantees it return None lang = resolve_lang(person) _lang, block, _fb = pick_block(material.localized, lang) due_at = _as_utc(campaign.due_at) overdue = due_at is not None and now > due_at return NotificationEnvelope( recipient=person, title=str(block.get("title") or ""), body_text=_SEE_TEMPLATE_SENTINEL, template_key=TEMPLATE_REMINDER, channels=list(campaign.channels or []), url=f"/me/security-learning/{material.id}", context={ "material_title": str(block.get("title") or ""), "due_label": ( due_at.strftime("%Y-%m-%d") if due_at else "" ), # Pre-rendered per recipient because it is a whole # sentence that differs by state, and a template # placeholder cannot be conditionally absent. "overdue_line": self._overdue_line(lang, overdue, due_at), }, related_resource_type="secaw_campaign", related_resource_id=campaign.id, card_subtitle=LocalizedText( zh="安全宣贯 · 待确认", en="Security awareness · not yet confirmed", ja="セキュリティ啓発 · 未確認", ), ) @staticmethod def _overdue_line( lang: str, overdue: bool, due_at: datetime | None, ) -> str: if overdue and due_at is not None: return { "zh": f"截止日期 {due_at:%Y-%m-%d} 已经过了。", "en": f"The deadline of {due_at:%Y-%m-%d} has passed.", "ja": f"期限({due_at:%Y-%m-%d})を過ぎています。", }.get(lang, f"截止日期 {due_at:%Y-%m-%d} 已经过了。") if due_at is not None: return { "zh": f"请在 {due_at:%Y-%m-%d} 前完成。", "en": f"Please finish before {due_at:%Y-%m-%d}.", "ja": f"{due_at:%Y-%m-%d} までにお願いします。", }.get(lang, f"请在 {due_at:%Y-%m-%d} 前完成。") return "" async def _manager_envelope( self, campaign: SecawCampaign, manager: IdentityPerson, people: list[IdentityPerson], now: datetime, ) -> Any: """One digest: this manager, this campaign, these reports. ``people`` all owe THIS campaign — the caller groups on (manager, campaign) so the title, the link and the listing are about the same piece of reading. """ from app.core.locale import resolve_lang from app.notifications.service import ( _SEE_TEMPLATE_SENTINEL, CardSection, LocalizedText, NotificationEnvelope, ) material = await self.db.get(SecawMaterial, campaign.material_id) if material is None: # pragma: no cover return None lang = resolve_lang(manager) _lang, block, _fb = pick_block(material.localized, lang) # Language-neutral detail lines (names + addresses), so they can # safely sit inside any of the three localized bodies. listing = "\n".join( f"· {p.display_name or ''} <{p.work_email or ''}>" for p in people[:25] ) if len(people) > 25: listing += f"\n· … +{len(people) - 25}" return NotificationEnvelope( recipient=manager, title=str(block.get("title") or ""), body_text=_SEE_TEMPLATE_SENTINEL, template_key=TEMPLATE_MANAGER, channels=list(campaign.channels or []), url=f"/me/security-learning/{material.id}", context={ "material_title": str(block.get("title") or ""), "pending_count": len(people), "campaign_name": campaign.name, }, related_resource_type="secaw_campaign", related_resource_id=campaign.id, card_subtitle=LocalizedText( zh="你的下属还没确认", en="Your reports have not confirmed", ja="あなたの部下が未確認です", ), card_sections=[ CardSection( title=LocalizedText( zh="待确认", en="Outstanding", ja="未確認", ), body=LocalizedText(zh=listing, en=listing, ja=listing), ), ], ) async def clear_nag_state( db: AsyncSession, *, person_id: uuid.UUID, campaign_id: uuid.UUID, ) -> None: """Settle the account when somebody acknowledges. Constraint (3) from ``cadence``'s module docstring: without this the person's NEXT ask of the same kind inherits this counter and starts on the two-week rung — a "noise reduction" that turns a fresh problem into near-silence. """ await cadence.clear( db, recipient_id=person_id, template_key=TEMPLATE_REMINDER, subject_key=str(campaign_id), )