"""Telling the person who was waiting that it is fixed. An automation exception usually has a human on the other end of it: the new hire whose account never got created, the employee whose seat request died in a vendor 500. Until now the platform knew that and said nothing. They found out by asking IT, or by not finding out at all. This module closes that half. Three rules shape it, and each one exists because the opposite is worse: **Only when it is actually fixed, and only once.** The echo fires on the transition from open to recovered — not on "we retried it", not on "we noticed". ``AutomationException.subject_notified_at`` (left by PR1 for exactly this) makes it a transition rather than a state, so a re-run of the collector cannot send a second copy. A person told twice that their account is ready stops believing the first message. **No new interruption.** The category is not set on the envelope — this platform derives it from the template key (``categories.classify``), so routing happens by registering the two keys in the template catalogue under an EXISTING category. An employee's mute and do-not-disturb settings therefore govern these automatically, and nobody gains a new class of ping they cannot turn off. Adding a category would have been easier and would have quietly bypassed every preference the notification-governance work established. **In the reader's own language.** Both keys are registered in the platform template registry, not merely grouped in the admin catalogue — the registry is what selects a language. An unregistered key falls through to the producer's hardcoded copy, which is single-language by construction, and most people here read English. **Never tell someone about a failure they cannot act on.** The employee echo says "this is now working". It does NOT say "this failed 47 times first" — that is the admin's problem, and handing it to the person who was merely waiting turns a fixed problem into a new worry. The failure narrative goes to the owner, who can do something about it. Recurrence escalation is the other half and points the other way: when a signature comes BACK after being resolved, the owner hears about it, because a problem that returns is a problem whose remedy was wrong. """ from __future__ import annotations import logging from dataclasses import dataclass from datetime import UTC, datetime from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.core.txn import best_effort from app.domains.automation_exception.models import ( AutomationException, AutomationExceptionOccurrence, ) from app.domains.automation_exception.taxonomy import STATUS_RESOLVED, remedy_for from app.identity.person_models import IdentityPerson from app.notifications.service import NotificationEnvelope, NotificationService logger = logging.getLogger(__name__) #: Template keys. Registered in the template catalogue so the admin #: override surface can edit the copy without a deploy. TPL_SUBJECT_RECOVERED = "automation_exception.subject_recovered" TPL_OWNER_RECURRED = "automation_exception.owner_recurred" #: Cap per run. A collector tick that recovers a thousand rows is a #: good day, not a reason to send a thousand notifications in one #: transaction — the rest go out on the next tick. MAX_ECHO_PER_RUN = 200 #: How many times a signature must come back before the owner is told. #: One recurrence is noise (a vendor blip that re-fired); the second is #: a pattern, and a pattern means the remedy was wrong. RECURRENCE_ESCALATION_THRESHOLD = 2 @dataclass class EchoOutcome: subjects_notified: int = 0 owners_escalated: int = 0 skipped_no_person: int = 0 skipped_already_sent: int = 0 def as_dict(self) -> dict[str, int]: return { "subjects_notified": self.subjects_notified, "owners_escalated": self.owners_escalated, "skipped_no_person": self.skipped_no_person, "skipped_already_sent": self.skipped_already_sent, } def _subject_body(exc: AutomationException, person: IdentityPerson) -> tuple[str, str]: """What the waiting person is told. Deliberately short and free of failure detail. They asked for an account, or a seat, or a permission; the useful sentence is that it now exists. The 47 failed attempts behind it are the owner's problem, and telling the subject about them converts a solved problem into a new worry they cannot act on. """ title = "你在等的那一步已经完成" body = ( f"{person.display_name},你好。\n\n" "此前有一步自动化没有走通,我们已经修好并重新执行完成 —— " "现在它已经生效,你不需要再做任何事。\n\n" "如果你发现它仍然不可用,请直接回复本条或联系 IT。" ) return title, body def _owner_body(exc: AutomationException) -> tuple[str, str]: """What the owner is told when a signature comes back. Carries the remedy, not just the fact. "It came back" without "and here is what to check" is the dead end this epic removes. """ title = "这个问题又回来了" body = ( f"「{exc.signature[:120]}」在被处理之后又出现了 " f"{exc.recurrence_count} 次。\n\n" "一个被解决之后又回来的问题,说明上次的处置没有解决根因。\n" f"当前判定的根因:{exc.root_cause}\n" f"建议的下一步:{remedy_for(exc.root_cause)}" ) return title, body async def send_recovery_echoes( db: AsyncSession, *, trace_id: str = "", now: datetime | None = None, ) -> EchoOutcome: """Tell the people who were waiting that the thing now works. Scoped to the EXCEPTION, not the occurrence, because that is where PR1 put ``subject_notified_at`` — and it is the right grain: one resolved signature is one piece of news, even when it affected six people. Each of those six is told once; the exception is stamped once; a later collector tick re-examines neither. """ moment = now or datetime.now(UTC) outcome = EchoOutcome() rows = ( await db.execute( select(AutomationException) .where( # 🔴 ``status``, not ``resolved_at``. ``close_wont_fix`` # also stamps ``resolved_at`` — it is the column that # says "this cluster is closed", not "this cluster is # fixed". Selecting on it mailed every affected employee # "现在它已经生效,你不需要再做任何事" for a problem an # admin had just explicitly declared unfixed, which is # the precise harm this module's first rule forbids. AutomationException.status == STATUS_RESOLVED, AutomationException.subject_notified_at.is_(None), ) .order_by(AutomationException.resolved_at.asc()) .limit(MAX_ECHO_PER_RUN) # Locked AND refreshed: two ticks running at once would # otherwise both read ``subject_notified_at IS NULL`` from # their own identity map and both send. Being told twice # that your account is ready is what makes people stop # believing the first message — the exact harm this # function's docstring promises not to cause. .with_for_update() .execution_options(populate_existing=True), ) ).scalars().all() service = NotificationService(db) for exc in rows: # The distinct people this signature actually kept waiting. person_ids = ( await db.execute( select(AutomationExceptionOccurrence.person_id) .where( AutomationExceptionOccurrence.exception_id == exc.id, AutomationExceptionOccurrence.person_id.is_not(None), ) .distinct(), ) ).scalars().all() if not person_ids: # Nobody was waiting — plenty of automation has no human on # the other end. Stamped so it is not re-examined forever. exc.subject_notified_at = moment outcome.skipped_no_person += 1 continue had_send_failure = False for pid in person_ids: person = await db.get(IdentityPerson, pid) if person is None or person.employment_status == "left": # A leaver does not need to hear that their old account # was fixed. continue title, body = _subject_body(exc, person) # SAVEPOINT rather than a bare ``except``. To be precise about # what this does and does not fix: ``dispatch`` ALREADY # self-isolates its own writes — ``enqueue`` and both of # ``_deliver``'s flushes each sit inside ``begin_nested`` — # so this loop was not in fact poisoning its own transaction. # The savepoint makes the loop's promise ("one bad recipient # must not stop the rest") independent of that, which is worth # one round trip on a path that runs at most 200 times a tick, # and it is the shape the CI ratchet wants in new code. async with best_effort( db, label="send automation-exception recovery echo", log=logger, log_level=logging.DEBUG, # re-logged with a stack below ) as sent: await service.dispatch( NotificationEnvelope( recipient=person, title=title, body_text=body, template_key=TPL_SUBJECT_RECOVERED, # The registered template renders in the # recipient's own language; these variables are # what it interpolates. Without them the # producer's hardcoded copy would reach # everyone in one language — and most people # here read English. context={"display_name": person.display_name or ""}, related_resource_type="automation_exception", related_resource_id=exc.id, # One echo per (exception, person), ever — belt # to the stamp's braces, because the outbox # dedupe survives a crash between send and stamp. delivery_idempotency_key=f"autoexc-echo:{exc.id}:{pid}", ), trace_id=trace_id, ) if not sent.ok: # One bad recipient must not stop the rest. ``exc_info`` # carries the traceback that the bare ``except`` used to # log — ``best_effort`` records only the message. logger.warning( "automation-exception echo failed (exc=%s person=%s)", exc.id, pid, exc_info=sent.error, ) had_send_failure = True continue outcome.subjects_notified += 1 # Stamp unless a send actually FAILED. Two cases stamp: someone # was told, and nobody was left to tell (every candidate had # gone). Both are finished states, and re-walking them every # tick would be pure waste. A failed send is the one case that # deserves another attempt, so it leaves the row unstamped. if not had_send_failure: exc.subject_notified_at = moment await db.flush() return outcome async def escalate_recurrences( db: AsyncSession, *, trace_id: str = "", now: datetime | None = None, ) -> EchoOutcome: """Tell the owner when a resolved problem comes back. A problem that returns is a problem whose remedy was wrong, and the person who applied that remedy is the one who can pick a different one. Escalates once per crossing of the threshold, not once per recurrence — ``escalated_at`` (PR1) is the marker. """ moment = now or datetime.now(UTC) outcome = EchoOutcome() rows = ( await db.execute( select(AutomationException) .where( AutomationException.recurrence_count >= RECURRENCE_ESCALATION_THRESHOLD, AutomationException.escalated_at.is_(None), AutomationException.resolved_at.is_(None), AutomationException.owner_person_id.is_not(None), ) .order_by(AutomationException.recurrence_count.desc()) .limit(MAX_ECHO_PER_RUN) # Same reason as the recovery echo above. .with_for_update() .execution_options(populate_existing=True), ) ).scalars().all() service = NotificationService(db) for exc in rows: owner = await db.get(IdentityPerson, exc.owner_person_id) if owner is None or owner.employment_status == "left": # An owner who has left cannot act. Left unstamped on # purpose: once ownership is re-resolved to somebody who is # still here, this row should escalate to them. continue title, body = _owner_body(exc) # SAVEPOINT — same reason as the recovery echo above. async with best_effort( db, label="send automation-exception recurrence escalation", log=logger, log_level=logging.DEBUG, # re-logged with a stack below ) as sent: await service.dispatch( NotificationEnvelope( recipient=owner, title=title, body_text=body, template_key=TPL_OWNER_RECURRED, context={ "signature": exc.signature[:120], "recurrence_count": exc.recurrence_count, "root_cause": exc.root_cause, "remedy": remedy_for(exc.root_cause), }, related_resource_type="automation_exception", related_resource_id=exc.id, delivery_idempotency_key=f"autoexc-recur:{exc.id}", ), trace_id=trace_id, ) if not sent.ok: logger.warning( "automation-exception escalation failed (exc=%s)", exc.id, exc_info=sent.error, ) continue exc.escalated_at = moment outcome.owners_escalated += 1 await db.flush() return outcome __all__ = [ "MAX_ECHO_PER_RUN", "RECURRENCE_ESCALATION_THRESHOLD", "TPL_OWNER_RECURRED", "TPL_SUBJECT_RECOVERED", "EchoOutcome", "escalate_recurrences", "send_recovery_echoes", ]