"""Notification service — unified cross-channel dispatch. Domain code composes a :class:`NotificationEnvelope` (who, title, body, optional card) and calls :meth:`NotificationService.dispatch`. The service writes one ``notification_outbox`` row FIRST, then tries every requested channel independently: * ``feishu`` — resolve person.work_email → Feishu open_id, push text (or card) via the primary chat_approval integration. * ``email`` — render the template, send via the primary SMTP integration. * ``inapp`` — (placeholder; the outbox row itself is the inbox entry; a per-user /me inbox can read directly from outbox rows.) The outbox ensures "no silent drops": **only** when *every* requested channel fails on the same attempt is the row marked retryable and picked up by the :func:`notification.outbox_retry` scheduled handler with exponential backoff. As soon as *one* channel delivers (this attempt or any prior one) the row freezes at ``partial`` — the recipient has been notified at least once, so re-firing the failed channel would only duplicate the same message into a second inbox. The failed-channel error is preserved on the row for admin visibility and the admin "重试" button can still force a per-channel retry. After ``max_attempts`` exhausted (or a permanent failure with no channel ever delivered) the row is marked ``dead`` and surfaces in the admin outbox page for manual re-send. Never raises — per-channel failures land in the outbox row AND return in the :class:`NotificationResult` so the caller can still see what happened synchronously. Template rendering reads :class:`PlatformSetting` rows keyed ``email_template.``. Missing rows fall back to the built-in Chinese defaults so dev / greenfield installs just work. """ from __future__ import annotations import logging import re import zlib from collections.abc import Mapping, Sequence from dataclasses import asdict, dataclass, field, fields from datetime import UTC, datetime, timedelta from typing import Any, cast from urllib.parse import urlsplit from uuid import UUID, uuid4 from sqlalchemy.exc import OperationalError from sqlalchemy.ext.asyncio import AsyncSession from app.audit.service import AuditService from app.core.exceptions import AppException from app.core.locale import ( DEFAULT_LANG, normalize_lang, pick_localized, pick_variant, resolve_lang, ) from app.core.locale import is_en as locale_is_en from app.core.txn import best_effort from app.identity.person_models import IdentityPerson from app.integrations.compat import get_feishu_config from app.integrations.driver_base import EmailDriver, EmailMessage from app.integrations.service import ( build_primary_driver, get_platform_setting, ) from app.notifications import card_design, card_icon, categories from app.notifications.email_layout import render_email_shell from app.notifications.models import ( ACK_NOTE_PREFIX, STATUS_DEAD, STATUS_DELIVERED, STATUS_DELIVERING, STATUS_DELIVERY_UNCERTAIN, STATUS_FAILED, STATUS_PARTIAL, STATUS_PENDING, NotificationOutbox, ) from app.notifications.preferences_service import ( GOVERNABLE_CHANNELS, NotificationPreferenceService, ) from app.notifications.templates import BUILTIN_TEMPLATES as _DOMAIN_TEMPLATES logger = logging.getLogger(__name__) # Exponential backoff schedule (seconds) — index = attempts made so far. # Attempts past the end fall back to the last value until max_attempts # is reached, at which point the row is marked ``dead``. _BACKOFF_SCHEDULE: tuple[int, ...] = ( 5 * 60, # first retry 5 min later 30 * 60, # 30 min 2 * 60 * 60, # 2 h 8 * 60 * 60, # 8 h 24 * 60 * 60, # 24 h ) # The claim is committed before provider I/O. A healthy send should finish # in seconds; 15 minutes leaves room for configured transport timeouts while # still surfacing a dead worker promptly. Feishu only guarantees duplicate # suppression for one hour, so stale claims are auto-recoverable only while a # conservative 55-minute window remains and only when every claimed channel # is Feishu. DELIVERY_LEASE = timedelta(minutes=15) FEISHU_SAFE_RETRY_WINDOW = timedelta(minutes=55) def _stable_provider_delivery_keys(outbox_id: UUID) -> dict[str, str]: return { "feishu": str(outbox_id), "email": f"", } # Reserved outbox context key for content whose authorization must still be # true when an asynchronous worker performs the real external send. Domain # producers persist both the required roles and a non-confidential fallback; # ``_deliver`` owns the final locking-current-read and fail-closed downgrade. DELIVERY_ROLE_GUARD_CONTEXT_KEY = "_helix_delivery_role_guard" #: How many delivery passes a PARTIALLY-delivered row gets before its #: missing channel is abandoned. The recipient already has the message #: on the other channel, so the second copy is only worth sending while #: it still reads as part of the same notification — a card that turns #: up a day after the email is worse than one that never turns up. #: Three passes walk the backoff's 5-minute and 30-minute steps, which #: is where a provider blip actually resolves, and stop before the #: 2h/8h/24h steps that produced the original "days apart" complaint. #: #: Expressed in attempts rather than wall-clock age on purpose: #: ``created_at`` is a server default and is EXPIRED on the freshly #: flushed row, so reading it inside the delivery path triggers a lazy #: load from sync flush context and raises ``MissingGreenlet``. #: ``attempts`` is already in memory — it was just incremented above. #: #: Rows that reached NOBODY are unaffected: they keep the full #: ``max_attempts`` ladder, because there the alternative is silence, #: not a late duplicate. _PARTIAL_MAX_ATTEMPTS = 3 def _next_attempt_at(attempts: int) -> datetime: idx = min(max(attempts - 1, 0), len(_BACKOFF_SCHEDULE) - 1) return datetime.now(UTC) + timedelta(seconds=_BACKOFF_SCHEDULE[idx]) def refused_redispatch_terminal(row: NotificationOutbox) -> str: """Terminal status for a row the retry worker refuses to redispatch. ``dead`` for a row that never reached anybody. But since ``retry_due_clause`` began admitting armed ``partial`` rows, this loop can also see a row whose OTHER channel already landed — a recipient who offboarded between the first attempt and the retry, or whose person row was hard-deleted. Marking that ``dead`` would overwrite a delivery that really happened: the email is in their mailbox and the row would claim nothing was ever sent. It also matters downstream — ``_DELIVERED_STATUSES`` treats ``partial`` as reach, so flipping it to ``dead`` silently rewrites campaign coverage and awareness compliance history for somebody who WAS notified. So: anything delivered ⇒ stay ``partial`` (terminal, since the caller clears ``next_attempt_at``); nothing delivered ⇒ ``dead``. """ return STATUS_PARTIAL if (row.delivered_channels or []) else STATUS_DEAD def retry_due_clause(now: datetime): # type: ignore[no-untyped-def] """Rows the retry worker may claim, as one reusable WHERE clause. ``pending`` / ``failed`` are due on the usual terms: a NULL ``next_attempt_at`` means "never scheduled, take it now". ``partial`` is the subtle one. A partial row reached SOMEBODY — one channel landed — so it is never dead and it must never be re-fired just because it is not fully delivered: that would push the same card into a second inbox forever. But "one channel landed" and "the other channel can never land" are different claims, and until this clause existed the code conflated them. A dual-channel security broadcast where email succeeded and Feishu returned a one-off ``2200 Internal Error`` froze at ``partial`` on attempt 1 and the card was gone for good — no backoff, no second try, and the console counted the person as reached. So a partial row is claimable **only when ``_deliver`` deliberately armed it** by writing a ``next_attempt_at`` — which it does exactly when the still-missing channel failed transiently and attempts remain. NULL stays terminal, which also leaves every partial row that pre-dates this change untouched: they are historical, their transient window closed months ago, and reviving them would deliver stale notices. ``_deliver`` only attempts channels absent from ``delivered_channels``, so a claimed partial re-fires the missing channel alone — the one that already landed is not re-sent. """ from sqlalchemy import and_, or_ return or_( and_( NotificationOutbox.status.in_([STATUS_PENDING, STATUS_FAILED]), or_( NotificationOutbox.next_attempt_at.is_(None), NotificationOutbox.next_attempt_at <= now, ), ), and_( NotificationOutbox.status == STATUS_PARTIAL, NotificationOutbox.next_attempt_at.is_not(None), NotificationOutbox.next_attempt_at <= now, ), ) # Pre-launch hardening 2026-05-17 (Round-19 audit, finding D): error # fragments that signal the failure cannot heal between retries. A # recipient without a work_email never grows one mid-backoff; a Feishu # integration that isn't configured won't self-configure 5 minutes # later. Retrying these wastes 5 backoff cycles per envelope and # clutters the audit log with five identical failure rows. The match # is on a stable fragment (not the full message) so localised / # wrapped variants still classify. _PERMANENT_ERROR_FRAGMENTS: tuple[str, ...] = ( "recipient has no work_email", "feishu integration not configured", # Round-51 finding R2-P1-1: a Feishu open_id miss for the recipient # email (contractor / external / just-onboarded person not yet in # the Feishu directory) is structurally permanent — the next 5 # retries over 24h won't conjure them into Feishu. Stop wasting # backoff cycles + audit-row noise on these. "feishu open_id not found", ) def _is_permanent_channel_error(err: str | None) -> bool: if not err: return False lowered = err.lower() return any(frag in lowered for frag in _PERMANENT_ERROR_FRAGMENTS) # Channel honesty 2026-06-10: which integration ``kind`` backs each push # channel. A channel whose backing integration is disabled (or absent) # is SKIPPED at delivery time instead of being attempted-and-failed — # prod ran 90 days with the ``email`` integration row at ``enabled=0`` # and every dual-channel envelope burned 5 retries into ``dead`` (Apr, # May: 4 839 rows) or froze at ``partial`` forever (Jun: 414 rows). # Skipping is honest: the admin explicitly turned the channel off, so # not sending on it is the *configured* outcome, not a failure. _CHANNEL_INTEGRATION_KIND: dict[str, str] = { "feishu": "chat_approval", "email": "email", } # Guaranteed-delivery templates (2026-06-10, execution-plane PR4). # Ops-critical alert classes whose silent loss leaves the operator # blind — prod ran 90 days with 445 dead ``scheduled_job.failure`` # rows (the alerting system's own alerts), 72 dead leaver-SLA # urgent/breach rows and 1 409 dead billing alerts. When a row whose # ``template_key`` matches one of these prefixes lands terminal DEAD # (every push channel failed/exhausted), ``_deliver`` fires a # last-resort Feishu **group webhook** (``alert_webhook_url`` on the # chat_approval integration — a different transport from the DM API # that just failed) describing the lost notification, and stamps the # outcome into the row's ``context['critical_fallback']``. Prefix # match so ``leaver_blocked_sla_urgent`` / ``…_breach`` and future # per-severity variants are covered without enumerating every key. _CRITICAL_TEMPLATE_PREFIXES: tuple[str, ...] = ( "scheduled_job.failure", "leaver_blocked_sla_urgent", "leaver_blocked_sla_breach", "cloud.billing_alert", "cloud.grant.manual_revoke_escalated", "provisioning.revoke_failure", "identity_sync.run_failure", ) def is_critical_template(template_key: str | None) -> bool: if not template_key: return False return any( template_key.startswith(prefix) for prefix in _CRITICAL_TEMPLATE_PREFIXES ) # Template keys whose ``body_text`` carries a live secret (the Google # first-login temporary password). Recorded for IT hand-delivery fallback, # these rows get special handling everywhere they surface: # * NEVER resent/retried — both clone/re-fire the body through the normal # work_email + Feishu pipeline, the exact channels the out-of-band send # avoided (see ``_resend_one`` + the ``/retry`` router guard). # * body MASKED in the admin ``/outbox`` for non-IT roles (router). # * hidden from the recipient's own ``/inbox`` (inbox_service). # Canonical single source — imported by router + inbox_service. SECRET_TEMPLATE_KEYS = frozenset({ "google_account_password", "tencent_cam_initial_password", }) # Per-process rate limit for the critical-fallback webhook, keyed by # template_key. The backstop's design scenario is a CHANNEL-level # outage — every critical notification × every recipient dying at # once. One webhook line per template per window tells the operator # everything; hundreds of identical lines would bury the real alert. # Per-process (worker + API are separate processes) → worst case one # extra post per window per process, which is fine for a backstop. # # Review round-2 P2: the window is consumed only by the OUTCOME, not # by the check — a failed/unconfigured attempt takes a short cooldown # (so a row-lock-holding batch doesn't burn 5 s per row against a # down webhook) instead of the full hour (which would silently # swallow every later real fallback for that template). _CRITICAL_FALLBACK_WINDOW_SECONDS = 3600 _CRITICAL_FALLBACK_FAIL_COOLDOWN_SECONDS = 300 _critical_fallback_last_fired: dict[str, float] = {} def _critical_fallback_rate_limited(template_key: str | None) -> bool: """Pure check — does NOT consume the window.""" import time key = template_key or "(none)" last = _critical_fallback_last_fired.get(key) return last is not None and time.monotonic() < last def _critical_fallback_stamp(template_key: str | None, *, sent: bool) -> None: """Record the attempt outcome: success holds the full window, failure holds only a short cooldown (the value stored is the timestamp the NEXT attempt becomes allowed).""" import time key = template_key or "(none)" hold = ( _CRITICAL_FALLBACK_WINDOW_SECONDS if sent else _CRITICAL_FALLBACK_FAIL_COOLDOWN_SECONDS ) _critical_fallback_last_fired[key] = time.monotonic() + hold def _all_failures_permanent( env: NotificationEnvelope, result: NotificationResult, *, delivered: list[str], skipped: list[str] | None = None, ) -> bool: """True iff every channel that was ATTEMPTED on this delivery pass AND failed returned a permanent error. Channels already delivered on a prior attempt are excluded — the helper inspects only the work this attempt actually did, so a prior-attempt feishu success + this-attempt permanent-email-failure cleanly evaluates to "permanent" without relying on the upstream ``any_delivered`` guard accidentally preserving the row in PARTIAL. Channels skipped because their integration is disabled were never attempted — they are neither failures nor successes and are excluded the same way. """ has_failure = False for channel in env.channels: if channel in delivered: # Already succeeded on a prior attempt — skip; the helper # iterates per-attempt work only. continue if skipped and channel in skipped: # Integration disabled — never attempted on this pass. continue if channel == "feishu": if result.feishu_ok: continue has_failure = True if not _is_permanent_channel_error(result.feishu_error): return False elif channel == "email": if result.email_ok: continue has_failure = True if not _is_permanent_channel_error(result.email_error): return False else: # Unknown channel — be safe and treat as retryable so a # future channel type doesn't accidentally get marked # DEAD on its first failure. return False return has_failure @dataclass(slots=True) class CardStat: """A single cell in the 2x2 (or 2xN) Feishu card stat grid. ``label`` is the small grey title above the value (e.g. "近 30 天花费"). ``value`` is the big bold number with its unit ("$1,234.56" / "23 人"). ``trend`` (optional) is the environmental change shown next to the value — pass the raw token, e.g. ``"▼ 58%"`` / ``"▲ 12%"`` — the builder colours green for ``▼`` and red for ``▲`` automatically. Some metrics invert (cost-down=good, count-down=bad); pass ``trend_is_good=False`` to swap the colour for the upcoming token. ``hint`` (optional) is the second-line caption — typically a "who" sentence like "崔佳宁 操作最多, 共 16 次" or a context line. """ label: CardText value: CardText trend: CardText | None = None trend_is_good: bool | None = None hint: CardText | None = None @dataclass(slots=True, frozen=True) class LocalizedText: """One card string in every language Helix ships. Card CHROME — stat labels, button labels, section titles, highlight labels, subtitles — never went through the template layer. ``_select_template_field`` covers ``subject`` / ``body_text`` / ``body_html`` / ``card_button_label`` and nothing else, so ~175 structured field values across 22 producer modules were hardcoded in whichever language their author happened to write in. The monthly attendance card goes to every employee and every intern with no language branching at all; the offboarding cards are pure Chinese and their reader is the LEAVER; the Day-1 hire card branches on ``is_china_employee``, so a Japanese hire's first card from their new employer is in English. Resolution is PER RECIPIENT (see ``resolve_card_text``), which is why this is a value and not a string chosen by the producer: one envelope fans out to many people who do not share a language. A plain ``str`` remains valid everywhere a ``CardText`` is accepted and renders verbatim — deliberately, so a producer that genuinely has one language (a user-typed message, an external vendor's error string) is not forced to invent translations of it. """ zh: str en: str ja: str CardText = str | LocalizedText """A card string: localized, or a bare ``str`` that renders as-is.""" def resolve_card_text(value: Any, lang: str | None) -> Any: """Resolve a :class:`LocalizedText` (or its serialized dict) for ``lang``. Accepts the dict form because card fields round-trip through the outbox as JSON: the retry worker rebuilds the envelope from the stored row, and a ``LocalizedText`` comes back as ``{"zh": …, "en": …, "ja": …}``. Without this branch a retried card would render that dict's ``repr`` at the reader. Anything else — ``str``, ``None``, a number — passes through untouched. """ target = normalize_lang(lang) or DEFAULT_LANG if isinstance(value, LocalizedText): return pick_variant( {"zh": value.zh, "en": value.en, "ja": value.ja}, target, ) or value.zh if isinstance(value, Mapping) and {"zh", "en", "ja"} <= set(value): # The serialized form, restored from the outbox on a worker retry. return pick_variant(value, target) or str(value.get("zh") or "") return value #: The signature line appended to every card that does not supply its own. #: Was a two-way ``zh if not is_en else en``, so a Japanese reader got the #: CHINESE footer stapled under a Japanese body. DEFAULT_CARD_FOOTER: dict[str, str] = { "zh": "— Helix IT 平台", "en": "— Helix IT Platform", "ja": "— Helix IT プラットフォーム", } #: 「回复这张卡片就能问」这件事**不再写在每张卡片上**(2026-08-31)。 #: #: 它原本是为发现性加的:卡片落在助手监听的同一个 1:1 会话里,569 之后回复 #: 还能解析回它下面那张卡,但没人知道可以在那儿打字 —— 30 天推了约 3,785 张 #: 卡,而助手历史上只收到过 3 个人的 98 条消息。 #: #: 摘掉的理由是它出现在**全公司每一张卡**的最后一行:一句功能性提示重复到 #: 第二十张时就是版面噪音,而卡片本身要承载的是那件事本身。发现性改由助手的 #: 使用说明承担(``assistant/help_card.py`` 那张帮助卡,以及它发不出去时兜底的 #: ``assistant/messages.py`` 的 ``_HELP``)。 #: #: 这是一个**权衡,不是纯改进**:只收卡片、从不打开助手的人,现在没有任何 #: 地方会告诉他可以回复。如果将来要把发现性补回来,更好的形状是「只在某人 #: 收到的第一张卡上出现一次」,而不是恢复成每张都有。 #: Appended after the default footer's platform signature, so the last line of #: a card is a person talking rather than a system stamping its name. Rotated #: per template (see :func:`compose_default_card_footer`) — one fixed sentence #: on every push in the company reads as boilerplate within a week. #: #: Every line has to survive the WORST card it can land under: an SLA breach, #: an offboarding gate, a security finding. So they offer company and help #: ("we're here", "pull us in"), never celebration — 「今天也辛苦啦」 under a #: compliance-incident card reads as the system not having read its own alert. CARD_FOOTER_WARMTH: tuple[dict[str, str], ...] = ( { "zh": "有需要随时找 IT,我们都在", "en": "IT is here whenever you need us", "ja": "何かあれば、いつでも IT にご相談ください", }, { "zh": "卡住了就叫我们一起看,别绕远路", "en": "Stuck on this? Pull us in — that is what we are here for", "ja": "困ったときは、遠慮なく IT を巻き込んでください", }, { "zh": "别一个人扛,IT 一直在你这边", "en": "You are not on your own here — IT has your back", "ja": "ひとりで抱え込まないでください。IT がついています", }, { "zh": "谢谢你把这件事放在心上", "en": "Thanks for taking care of this", "ja": "ご対応いただき、ありがとうございます", }, ) def compose_default_card_footer(lang: str | None, *, seed: str | None = None) -> str: """The default footer note: platform signature + one warm line. ``seed`` (the template key) picks the warm line, so the SAME notification always signs off the same way — a reader who gets the expiry reminder monthly sees a consistent voice — while the platform as a whole does not repeat one sentence on every card. It must be a STABLE digest: builtin ``hash()`` is salted per process (``PYTHONHASHSEED``), so an outbox row retried by the worker would render a different sign-off than its first attempt. Kept OUT of :data:`DEFAULT_CARD_FOOTER` on purpose: that table is also the strip list (``_CANONICAL_FOOTERS``) matched against template bodies, which sign off with the bare signature and know nothing about the warm line. """ key = normalize_lang(lang) or DEFAULT_LANG base = DEFAULT_CARD_FOOTER[key] parts = [base] if CARD_FOOTER_WARMTH: index = zlib.crc32((seed or "").encode("utf-8")) % len(CARD_FOOTER_WARMTH) parts.append(CARD_FOOTER_WARMTH[index][key]) # 页脚到此为止 —— 「回复这条消息」那一行 2026-08-31 摘掉了,理由见上面 # 那段常量注释。 return " · ".join(parts) #: Fallback action-button label for envelopes whose template defines none. #: Also was a two-way zh/en split, so the only button on a Japanese reader's #: card read 「去 Helix 查看」. DEFAULT_CARD_BUTTON_LABEL: dict[str, str] = { "zh": "去 Helix 查看", "en": "Open in Helix", "ja": "Helix で確認する", } #: The ``{cta_line}`` label, and the label of the HTML fallback CTA anchor. #: ONE table for both: they render into the same email, and two independent #: if/else copies is exactly how the text body ended up saying #: 「Helix で確認する」 while the HTML body said "View in Helix". CTA_LINE_LABELS: dict[str, str] = { "zh": "请到 Helix 查看", "en": "View in Helix", "ja": "Helix で確認する", } @dataclass(slots=True) class CardHighlight: """A list-style highlight section (categorical risk, top people, etc.). ``label`` is the section row label ("数据安全" / "高频访问"). ``items`` is one of: * a list[str] of pre-formatted entries — rendered verbatim * a single str — rendered as the only line Names that should pop blue use ``[b]name[/b]`` inline. """ label: CardText items: list[CardText] | CardText @dataclass(slots=True) class CardSection: """A free-form text block, optionally divider-separated. ``title`` (optional) is a bold inline title above the body. ``body`` is markdown content (``[r]`` / ``[b]`` markers supported). ``tone`` ∈ {``info`` | ``warn`` | ``danger`` | ``muted``} sets a leading emoji prefix on the title so the card still reads at a glance even for users who skim past colour. """ body: CardText title: CardText | None = None tone: str = "info" @dataclass(slots=True) class CardAction: """One button in the card's bottom action row. ``label`` is the button text. ``url`` is the destination (absolute or relative — relative resolves via ``app.frontend_base_url``). ``style`` ∈ {``primary`` | ``default`` | ``danger``}. """ label: CardText url: str style: str = "default" @dataclass(slots=True) class CardBadge: """One coloured chip in the card header, right of the title. A badge is for the ONE word that says what state this card is in — 待处理 / 已批准 / 已过期 — plus at most a couple of classifying words (the domain, the size of the thing). :data:`card_design.TONE_BY_STATUS` already supplies the state badge for every status, so producers only pass badges here to ADD classification, never to restate the state. ``color`` is a Feishu text-tag colour (blue / green / orange / red / grey / turquoise / violet / carmine / yellow / indigo / wathet). """ text: CardText color: str = "blue" @dataclass(slots=True) class CardFactBox: """The one shaded box that carries what the card is *about*. Left side is the subject (``subject`` bold, ``detail`` small grey under it); right side is the single number or date that decides the reader's answer (``metric`` bold, ``metric_hint`` small grey under it, both right- aligned). Deliberately singular: a card gets ONE of these. Two shaded boxes stop being emphasis and become wallpaper, and the reader is back to scanning. Everything else that is merely *true* belongs in :class:`CardFact` rows. """ subject: CardText detail: CardText | None = None metric: CardText | None = None metric_hint: CardText | None = None @dataclass(slots=True) class CardFact: """One ``icon **label** value`` row: the details, in scannable form. This is what replaces prose. 「施宏伟 申请续期 PingCAP SCC 上的 1 项权限, 时长 90 天。申请原因:need billing details」 is a paragraph a reader has to parse; the same content as 申请人 / 原因 / 时长 rows is read at a glance. ``icon`` is an emoji — pass a key from :data:`card_design.FACT_ICONS` or a literal emoji. Empty means no icon, which is the right answer whenever no icon genuinely carries meaning. """ label: CardText value: CardText icon: str = "" @dataclass(slots=True) class CardCollapse: """A list folded behind its own summary line. For the long tail a card must CONTAIN but should not SHOW: 12 role names, 9 affected servers, a full diff. The header states the count so the reader can decide whether to open it — a fold with no count is just a hidden thing. """ title: CardText body: CardText expanded: bool = False @dataclass(slots=True) class NotificationEnvelope: """Input to ``dispatch`` — the bare semantic payload. ``template_key`` selects the ``email_template.`` override; if the key is missing the built-in default is used. ``context`` is the variable dict passed into the template (str.format). ``related_resource_type`` / ``related_resource_id`` let the outbox trace back to the entity that triggered the send — useful for admin drill-in and for deduping retries per source. Rich-card fields (Feishu-only, ignored on email): * ``card_subtitle`` — small grey one-line description under the title, before the body. Set this to the "为什么发这张卡" hint. * ``card_stats`` — list of :class:`CardStat`. Renders as a 2-column grid; 2 / 4 / 6 stats produce 1x2 / 2x2 / 3x2 layouts. * ``card_highlights`` — list of :class:`CardHighlight`. Renders as a key-value block under a "高亮" subtitle (subtitle from ``card_highlights_title``). * ``card_sections`` — list of :class:`CardSection`. Rendered after the body, each separated by a horizontal divider. * ``card_actions`` — list of :class:`CardAction`. When present, supersedes the single ``url`` button — useful for multi-CTA cards like "确认 / 转移 / 详情" or "保留 / 清理 / 联系IT". * ``card_footer`` — small grey note line at the bottom. Defaults to the standard "— Helix IT 平台" footer when None; pass an empty string to omit. """ recipient: IdentityPerson title: str body_text: str channels: list[str] = field(default_factory=lambda: ["feishu", "email"]) template_key: str | None = None context: dict[str, Any] = field(default_factory=dict) body_html: str | None = None url: str | None = None cc_emails: list[str] = field(default_factory=list) related_resource_type: str | None = None related_resource_id: UUID | None = None card_subtitle: CardText | None = None card_stats: list[CardStat] = field(default_factory=list) card_highlights: list[CardHighlight] = field(default_factory=list) card_highlights_title: CardText | None = None card_sections: list[CardSection] = field(default_factory=list) card_actions: list[CardAction] = field(default_factory=list) card_footer: CardText | None = None # Semantic state of the THING this card is about — one of # ``card_design.STATUS_*``. Drives header colour, header badge and header # icon together, so those three can never disagree. Leave it None and the # renderer derives one: an action-bucket template is ``pending``, anything # else is ``info`` (see ``_card_status``). card_status: str | None = None # Extra header chips, on top of the status badge the tone supplies. card_badges: list[CardBadge] = field(default_factory=list) # The single shaded "what this is about" box. card_fact_box: CardFactBox | None = None # Scannable detail rows, in place of prose. card_facts: list[CardFact] = field(default_factory=list) # The long tail, folded behind a counted summary. card_collapse: CardCollapse | None = None # Feishu markdown elements don't expose ``text_size``, but the # markdown subset DOES accept ``…`` # inline tags as a size override. Set this to e.g. ``"heading_3"`` # on celebratory cards where you want the body text to read at # ~2 sizes above the default. None → default markdown size. card_body_size: str | None = None # Push-only dispatch: the recipient already holds this content as # another inbox row, so hide THIS row from the personal bell/inbox # (admin outbox surfaces still show it). Use for follow-up pushes # of an already-inboxed notification — e.g. the Day-1 Feishu # welcome card that trails the Day-1 welcome email. hidden_from_inbox: bool = False # Stable provider keys are derived from the durable outbox id. They are # populated by ``_deliver`` for first attempts and restored from the # committed delivery claim for worker retries. delivery_idempotency_key: str | None = None email_message_id: str | None = None def _resolve_card_chrome( env: NotificationEnvelope, lang: str | None, *, tmpl: dict[str, str] | None = None, absolute_url: str | None = None, ) -> dict[str, Any]: """Per-recipient copies of every card-chrome field. Returns copies; the envelope is NEVER mutated, because ``dispatch`` reuses one envelope across a fan-out and mutating it would freeze the first recipient's language onto everybody else's card. ``tmpl`` lets ``card_subtitle`` / ``card_footer`` / ``card_highlights_title`` be overridden per template the same way ``subject`` and ``body_text`` already are — they are editorial copy, so an admin rewording a card in the template UI must be able to reach them. A template value wins over the producer's, and comes with the ``ja → en → zh`` ladder for free. """ def _t(v: Any) -> Any: return resolve_card_text(v, lang) # The SAME context the body and subject get, auto-vars included. # ``template_catalog._AUTO_VARS`` is ``{"cta_line", "portal_url"}`` and the # editor labels both as auto-supplied ON THESE FIELDS, so leaving them out # shipped a literal ``{portal_url}`` — the identical bug this expansion was # added to fix, one variable name over. chrome_context = _localized_context(env.context, lang) if absolute_url and "portal_url" not in chrome_context: chrome_context["portal_url"] = absolute_url chrome_context.setdefault( "cta_line", _resolve_cta_line( channel="feishu_card", portal_url=absolute_url, lang=lang, ), ) def _chrome(field: str, fallback: Any) -> Any: if tmpl is not None: override = _select_template_field(tmpl, field, lang) if override is not None: # Formatted, because ``_extract_vars`` now scans these fields # and the template editor therefore ADVERTISES their # placeholders as available variables. Returning the raw # string shipped a literal ``{person_name}`` to every reader # of a subtitle an admin had every reason to think would # interpolate. # # The context is the EXPANDED one, identical to what the body # and subject receive — raw ``env.context`` left # ``{portal_url}`` / ``{cta_line}`` and every renamed legacy # alias unresolved, i.e. the same literal-placeholder bug one # variable name over. # # Guarded, like every other ``_safe_format`` call site in this # module: it tolerates a MISSING key but still raises on a # positional field (``{0}``) or a bad spec (``{amount:d}``), # and these fields are admin-editable. Unguarded, one stray # character in a subtitle took down the whole dispatch and # poisoned the outbox drain. try: return _safe_format(override, chrome_context) except (KeyError, IndexError, ValueError): return override return _t(fallback) return { "subtitle": _chrome("card_subtitle", env.card_subtitle), "highlights_title": _chrome( "card_highlights_title", env.card_highlights_title, ), "footer": _chrome("card_footer", env.card_footer), "stats": [ CardStat( label=_t(s.label), value=_t(s.value), hint=_t(s.hint), trend=_t(s.trend), trend_is_good=s.trend_is_good, ) for s in env.card_stats ], "highlights": [ CardHighlight( label=_t(h.label), items=( [_t(i) for i in h.items] if isinstance(h.items, list) else _t(h.items) ), ) for h in env.card_highlights ], "sections": [ CardSection(body=_t(s.body), title=_t(s.title), tone=s.tone) for s in env.card_sections ], "actions": [ CardAction(label=_t(a.label), url=a.url, style=a.style) for a in env.card_actions ], "badges": [ CardBadge(text=_t(b.text), color=b.color) for b in env.card_badges ], "fact_box": ( CardFactBox( subject=_t(env.card_fact_box.subject), detail=_t(env.card_fact_box.detail), metric=_t(env.card_fact_box.metric), metric_hint=_t(env.card_fact_box.metric_hint), ) if env.card_fact_box else None ), "facts": [ CardFact(label=_t(f.label), value=_t(f.value), icon=f.icon) for f in env.card_facts ], "collapse": ( CardCollapse( title=_t(env.card_collapse.title), body=_t(env.card_collapse.body), expanded=env.card_collapse.expanded, ) if env.card_collapse else None ), } @dataclass(slots=True) class NotificationResult: feishu_ok: bool = False feishu_error: str | None = None email_ok: bool = False email_error: str | None = None delivered_channels: list[str] = field(default_factory=list) def _serialize_card_payload(env: NotificationEnvelope) -> dict[str, Any] | None: """Freeze an envelope's Feishu rich-card fields into a JSON-safe dict for the outbox row. The retry worker rebuilds the envelope FROM the persisted row; a plain row would re-render the card without its buttons / stats / sections, degrading a "一键续期" multi-CTA card into the generic single "Open in Helix" button. Returns ``None`` when the envelope carries no rich-card content (the common case) so plain notifications don't grow a payload. """ if not ( env.card_actions or env.card_stats or env.card_highlights or env.card_sections or env.card_badges or env.card_facts or env.card_fact_box is not None or env.card_collapse is not None or env.card_status is not None or env.card_subtitle is not None or env.card_highlights_title is not None or env.card_footer is not None or env.card_body_size is not None ): return None return { # ``asdict`` recurses into nested dataclasses, so a LocalizedText # inside a CardStat / CardSection / … becomes a plain dict on its own. # The three BARE chrome fields have no dataclass around them, so they # need the same flattening explicitly — passed raw, a LocalizedText # made the whole outbox INSERT fail with "not JSON serializable", # which takes the notification down rather than degrading it. "subtitle": _jsonable_card_text(env.card_subtitle), "stats": [asdict(s) for s in env.card_stats], "highlights": [asdict(h) for h in env.card_highlights], "highlights_title": _jsonable_card_text(env.card_highlights_title), "sections": [asdict(s) for s in env.card_sections], "actions": [asdict(a) for a in env.card_actions], "footer": _jsonable_card_text(env.card_footer), "body_size": env.card_body_size, # 2.0 chrome. Frozen for the same reason as everything above: the retry # worker and the admin resend rebuild the envelope FROM this row, so a # field that is not here degrades a settled-state receipt card back # into a 「去审批」 card on its second attempt. "status": env.card_status, "badges": [asdict(b) for b in env.card_badges], "fact_box": asdict(env.card_fact_box) if env.card_fact_box else None, "facts": [asdict(f) for f in env.card_facts], "collapse": asdict(env.card_collapse) if env.card_collapse else None, } def _jsonable_card_text(value: Any) -> Any: """A card-chrome value in a form ``json.dumps`` accepts.""" return asdict(value) if isinstance(value, LocalizedText) else value def _rebuild_dataclass(cls: type, data: dict[str, Any]): """Instantiate ``cls`` from a persisted dict, ignoring keys that are no longer declared fields — so a payload frozen before a field was added/removed still reconstructs across a deploy.""" valid = {f.name for f in fields(cls)} return cls(**{k: v for k, v in data.items() if k in valid}) def _deserialize_card_payload(payload: dict[str, Any] | None) -> dict[str, Any]: """Rebuild the ``NotificationEnvelope`` rich-card kwargs from a stored ``card_payload`` dict. Tolerant of a missing / empty payload (plain notification) and of unknown keys (cross-deploy schema drift).""" if not payload: return {} return { "card_subtitle": payload.get("subtitle"), "card_stats": [ _rebuild_dataclass(CardStat, s) for s in (payload.get("stats") or []) ], "card_highlights": [ _rebuild_dataclass(CardHighlight, h) for h in (payload.get("highlights") or []) ], "card_highlights_title": payload.get("highlights_title"), "card_sections": [ _rebuild_dataclass(CardSection, s) for s in (payload.get("sections") or []) ], "card_actions": [ _rebuild_dataclass(CardAction, a) for a in (payload.get("actions") or []) ], "card_footer": payload.get("footer"), "card_body_size": payload.get("body_size"), "card_status": payload.get("status"), "card_badges": [ _rebuild_dataclass(CardBadge, b) for b in (payload.get("badges") or []) ], "card_fact_box": ( _rebuild_dataclass(CardFactBox, payload["fact_box"]) if isinstance(payload.get("fact_box"), dict) else None ), "card_facts": [ _rebuild_dataclass(CardFact, f) for f in (payload.get("facts") or []) ], "card_collapse": ( _rebuild_dataclass(CardCollapse, payload["collapse"]) if isinstance(payload.get("collapse"), dict) else None ), } class _SafeFormatContext(dict): """Format context that renders MISSING keys as their literal ``{placeholder}`` instead of raising KeyError. Why: templates evolve. An outbox row enqueued BEFORE a template gained a new placeholder (e.g. ``{pending_resources}``) carries a frozen ``context`` without that key, so on retry ``body.format(**ctx)`` raised KeyError and the ``except: pass`` shipped the WHOLE body unformatted — every placeholder literal, including the ones the context did have. With ``format_map`` over this dict the known placeholders still render and only the genuinely-missing one stays literal (a visible, greppable ``{key}`` beats a fully-broken body).""" def __missing__(self, key: str) -> str: return "{" + key + "}" def _safe_format(text: str, context: dict[str, Any]) -> str: """``str.format`` with missing-key tolerance (see _SafeFormatContext). A missing key accessed as an object — ``{missing.attr}`` / ``{missing[0]}`` — is degraded to the literal ``{...}`` too: the ``__missing__`` sentinel is a plain str, so attribute/index access on it would raise ``AttributeError``/``TypeError`` and escape the caller's ``except (KeyError, IndexError, ValueError)`` (KeyError is what a bare missing placeholder raised before). We render those field names literally rather than re-raising, so one evolved placeholder can't blank an otherwise-renderable body. Positional fields (``{0}``/``{}``) and bad format specs (``{amount:d}`` on a str) still raise — callers keep their broad except for those admin-template errors.""" try: return text.format_map(_SafeFormatContext(context)) except (AttributeError, TypeError): # Missing-key attribute/index access. Re-render leaving those # whole field expressions literal (best-effort — known # placeholders that DID resolve stay resolved is not achievable # once one field expression is malformed, so fall back to the # raw text, matching the caller's last-resort degrade). return text # ── Built-in default templates. Override per-install via # ``platform_settings`` rows (key=email_template., # value={subject, body_html, body_text, subject_en, body_text_en, # body_html_en, card_button_label, card_button_label_en}). # str.format substitutes {person_name}, {person_email}, {leaver_date} # etc. from ``context``. The "_en" variants are used when the recipient's # ``preferred_language`` starts with "en"; missing variants fall back # to the non-suffixed key (Chinese baseline) — that way half-translated # templates degrade to "always Chinese" instead of breaking. _DEFAULT_TEMPLATES: dict[str, dict[str, str]] = { # ── SSO 撤销落地(2026-09-05,迁移 603)────────────────────────── # # 四条卡片,一条链路:撤销那一刻给系统负责人开任务(provisioning.sso_removal, # 完成必须带凭证)、IT 的一句话知会(sso.revoked_admin_headsup)、阶梯催办 # (sso.removal_reminder)、以及 Google 登录审计发现撤销后仍有登录 # (sso.removal_contradicted)。占位符与 ``app.domains.sso.removal`` 及 # ``provisioning.service._notify_manual_assignees`` 的 context 一一对应。 "provisioning.sso_removal": { "subject": "[Helix] 请在 {sso_provider} 停用 SSO 访问:{system_name} · {actor}", "subject_en": "[Helix] Disable SSO access in {sso_provider}: {system_name} · {actor}", "subject_ja": "[Helix] {sso_provider} で SSO アクセスを無効化してください:{system_name} · {actor}", "body_text": ( "Helix 已撤销 {actor}({actor_email})对 {system_name} 的 SSO 授权" "(原因:{revoke_reason})。\n\n" "这一步只发生在 Helix 里 —— 目标系统 / {sso_provider} 侧的访问还在。" "请到控制台停用该账号的 SSO 访问,然后回到 Helix 填写凭证" "(工单号、控制台链接或一句说明)完成任务。没有凭证的任务不能关闭;" "Helix 之后会用 IdP 的登录审计复核这次停用。\n\n" "任务编号:{task_id}" ), "body_text_en": ( "Helix has revoked {actor}'s ({actor_email}) SSO grant for " "{system_name} (reason: {revoke_reason}).\n\n" "That happened in Helix only — the access in the target system / " "{sso_provider} is still live. Disable this account's SSO access in " "the console, then come back and complete the task with evidence " "(ticket id, console URL or a short note). A task without evidence " "cannot be closed; Helix later verifies the removal against the " "IdP's login audit.\n\n" "Task: {task_id}" ), "body_text_ja": ( "Helix は {actor}({actor_email})の {system_name} に対する SSO 権限を" "取り消しました(理由:{revoke_reason})。\n\n" "これは Helix 側の処理にとどまり、対象システム / {sso_provider} 側の" "アクセスはまだ有効です。コンソールでこのアカウントの SSO アクセスを" "無効化し、Helix に戻って根拠(チケット番号・コンソール URL・短いメモ)を" "添えてタスクを完了してください。根拠のないタスクは閉じられません。" "Helix はその後 IdP のログイン監査で無効化を検証します。\n\n" "タスク:{task_id}" ), }, "sso.revoked_admin_headsup": { "subject": "SSO 授权已撤销,移除任务已开:{system_name} · {holder}", "subject_en": "SSO grant revoked — removal task opened: {system_name} · {holder}", "subject_ja": "SSO 権限を取り消し、解除タスクを作成:{system_name} · {holder}", "body_text": ( "Helix 已撤销 {holder} 对 {system_name} 的 SSO 授权(原因:{reason})。" "IdP 侧的停用任务已开给系统负责人;每日复核会用登录审计核对," "拖延会逐级催办并在 14 天开合规事件。任务:{task_id}" ), "body_text_en": ( "Helix revoked {holder}'s SSO grant for {system_name} (reason: {reason}). " "A removal task is open for the system owner; the daily reconcile " "verifies it against the login audit, reminds on a widening ladder " "and opens a compliance incident at day 14. Task: {task_id}" ), "body_text_ja": ( "Helix は {holder} の {system_name} に対する SSO 権限を取り消しました" "(理由:{reason})。IdP 側の無効化タスクはシステム責任者に割り当て済みです。" "日次の照合でログイン監査と突き合わせ、遅延時は段階的にリマインドし、" "14 日でコンプライアンス事案を起票します。タスク:{task_id}" ), }, "sso.removal_reminder": { "subject": "⏳ SSO 撤销待落地(第 {rung} 次提醒):{system_name} · {holder}", "subject_en": "⏳ SSO removal still outstanding (reminder #{rung}): {system_name} · {holder}", "subject_ja": "⏳ SSO 解除が未完了(リマインド {rung} 回目):{system_name} · {holder}", "body_text": ( "Helix 已于 {age_days} 天前撤销 {holder} 对 {system_name} 的 SSO 授权," "但目标系统里还没有人确认停用。请到 IdP / 目标系统控制台停用," "然后回到 Helix 填写凭证完成任务。任务:{task_id}" ), "body_text_en": ( "Helix revoked {holder}'s SSO grant for {system_name} {age_days} days " "ago, and nobody has confirmed it was disabled in the target system. " "Disable it in the IdP / target console, then complete the task in " "Helix with evidence. Task: {task_id}" ), "body_text_ja": ( "Helix は {age_days} 日前に {holder} の {system_name} に対する SSO 権限を" "取り消しましたが、対象システム側での無効化はまだ確認されていません。" "IdP / 対象システムのコンソールで無効化し、Helix で根拠を添えてタスクを" "完了してください。タスク:{task_id}" ), }, "sso.removal_contradicted": { "subject": "🚨 撤销后仍在登录:{system_name} · {holder}", "subject_en": "🚨 Login observed after revoke: {system_name} · {holder}", "subject_ja": "🚨 取り消し後もログインあり:{system_name} · {holder}", "body_text": ( "Google 登录审计显示 {holder} 在 Helix 撤销之后,仍于 {last_login_at} " "通过 SAML 登录了 {system_name}。之前的「已停用」确认已作废。" "请立即在 IdP / 目标系统停用,并重新完成任务。任务:{task_id}" ), "body_text_en": ( "Google's login audit shows {holder} logged in to {system_name} via " "SAML at {last_login_at}, after the Helix revoke. The earlier " "'disabled' confirmation has been voided. Disable the access in the " "IdP / target system now and complete the task again. Task: {task_id}" ), "body_text_ja": ( "Google のログイン監査では、Helix の取り消し後 {last_login_at} に " "{holder} が {system_name} へ SAML ログインした記録があります。" "以前の「無効化済み」確認は無効になりました。至急 IdP / 対象システムで" "無効化し、タスクを再度完了してください。タスク:{task_id}" ), }, # ── API Key:转手轮换 / 闲置(2026-09-05,迁移 603)────────────── "apikey.transfer_rotation_manual": { "subject": "API Key 转手后需手动轮换:{request_number}({provider_name})", "subject_en": "API key needs a manual rotation after hand-over: {request_number} ({provider_name})", "subject_ja": "API キーの引き継ぎ後に手動ローテーションが必要:{request_number}({provider_name})", "body_text": ( "API Key {request_number}({provider_name})已转交给新持有人,但 Helix 无法" "自动轮换这把 Key(原因:{reason})。原持有人仍然知道旧凭证 —— 请在厂商" "控制台生成新凭证,并由管理员在 Helix 更新 Key 值。" ), "body_text_en": ( "API key {request_number} ({provider_name}) has been handed over to a " "new owner, but Helix cannot rotate it automatically (reason: {reason}). " "The previous owner still knows the old credential — generate a new one " "in the vendor console and have an admin update the key in Helix." ), "body_text_ja": ( "API キー {request_number}({provider_name})は新しい保持者に引き継がれ" "ましたが、Helix では自動ローテーションできません(理由:{reason})。" "以前の保持者は旧クレデンシャルを知っています — ベンダーコンソールで新しい" "クレデンシャルを発行し、管理者が Helix 上のキーを更新してください。" ), }, "apikey.idle_alert": { "subject": "💤 这把 API Key 上月没有花费:{request_number}({provider_name})", "subject_en": "💤 This API key had no spend last month: {request_number} ({provider_name})", "subject_ja": "💤 この API キーは先月使われていません:{request_number}({provider_name})", "body_text": ( "你名下的 API Key {request_number}({provider_name},项目「{project_name}」)" "在 {period} 的花费为 $0,已连续 {streak} 个月。\n\n" "如果还在用或近期会用,点「保留」并写一句理由(之后 {keep_months} 个月不再问);" "如果不需要了,点「释放」发起回收(需你的主管确认);拿不准就联系 IT。" ), "body_text_en": ( "Your API key {request_number} ({provider_name}, project \"{project_name}\") " "had $0 spend in {period} — {streak} month(s) in a row.\n\n" "Still using it, or about to? Press Keep with a one-line reason (no more " "questions for {keep_months} months). Done with it? Press Release to start " "the reclaim (your manager confirms). Not sure — contact IT." ), "body_text_ja": ( "あなたの API キー {request_number}({provider_name}、プロジェクト" "「{project_name}」)は {period} の費用が $0 で、{streak} か月連続です。\n\n" "まだ使う / 近く使う場合は「保持」を押して理由を一言(以後 {keep_months} か月は" "確認しません)。不要なら「解放」で回収を開始(上長が確認)。判断できない場合は " "IT へ連絡してください。" ), }, # ── DAX grant lifecycle (migration 606) ───────────────────────── "data_access.expiry_reminder": { "subject": "⏳ 数据集访问权限 {days} 天后到期:{locator}", "subject_en": "⏳ Dataset access expires in {days} day(s): {locator}", "subject_ja": "⏳ データセットのアクセス権が {days} 日後に期限切れ:{locator}", "body_text": ( "你对 BigQuery 数据集 {locator} 的个人读权限将于 {expires_at} 到期。\n\n" "还需要就点「续期」写一句理由(还可自助续期 {renewals_remaining} 次,之后需重新申请);" "不需要了就「释放」。" ), "body_text_en": ( "Your personal read access to BigQuery dataset {locator} expires on {expires_at}.\n\n" "Still need it? Renew with a one-line reason ({renewals_remaining} self-renewal(s) " "left before a new request is required). Done with it? Release it." ), "body_text_ja": ( "BigQuery データセット {locator} への個人の読み取り権限は {expires_at} に期限切れになります。\n\n" "まだ必要なら「更新」で理由を一言(自己更新は残り {renewals_remaining} 回、その後は再申請)、" "不要なら「解放」してください。" ), }, "data_access.expired_revoked": { "subject": "数据集访问权限已到期并回收:{locator}", "subject_en": "Dataset access expired and was revoked: {locator}", "subject_ja": "データセットのアクセス権が期限切れで回収されました:{locator}", "body_text": ( "你对 {locator} 的个人读权限已到期,Helix 已发起回收。如仍需要,请重新提交数据访问申请。" ), "body_text_en": ( "Your personal read access to {locator} has expired and Helix has started the " "revoke. Submit a new data-access request if you still need it." ), "body_text_ja": ( "{locator} への個人の読み取り権限は期限切れとなり、Helix が回収を開始しました。" "引き続き必要な場合は再申請してください。" ), }, "data_access.owner_confirmation": { "subject": "请确认服务账号仍需要数据集访问:{service_account} → {locator}", "subject_en": "Confirm this service account still needs dataset access: {service_account} → {locator}", "subject_ja": "サービスアカウントのデータセットアクセスを確認してください:{service_account} → {locator}", "body_text": ( "你是服务账号 {service_account} 的技术负责人。它对 {locator} 的读权限到了半年度确认" "(到期日 {due_at},已逾期 {overdue_days} 天,宽限 {grace_days} 天)。\n\n" "还在用就点「确认仍需要」;不用了就「回收」。" ), "body_text_en": ( "You are the technical owner of service account {service_account}. Its read access to " "{locator} is due for its semiannual confirmation (due {due_at}, overdue {overdue_days} " "day(s), grace {grace_days} days).\n\nPress Confirm if the workload still needs it, or " "Revoke if not." ), "body_text_ja": ( "あなたはサービスアカウント {service_account} の技術責任者です。{locator} への読み取り権限の" "半期確認の時期です(期限 {due_at}、超過 {overdue_days} 日、猶予 {grace_days} 日)。\n\n" "まだ必要なら「確認」、不要なら「回収」を押してください。" ), }, "data_access.owner_confirmation_lapsed": { "subject": "服务账号数据集权限因未确认已回收:{service_account} → {locator}", "subject_en": "Service-account dataset access revoked — confirmation lapsed: {service_account} → {locator}", "subject_ja": "確認がなかったためサービスアカウントのアクセス権を回収しました:{service_account} → {locator}", "body_text": ( "{service_account} 对 {locator} 的读权限半年度确认逾期未答复,Helix 已按策略发起回收。" "如工作负载仍需要,请重新提交服务账号授权申请。" ), "body_text_en": ( "The semiannual confirmation for {service_account}'s read access to {locator} was not " "answered in time; Helix has started the revoke. Submit a new service-account grant " "request if the workload still needs it." ), "body_text_ja": ( "{service_account} の {locator} への読み取り権限の半期確認が期限内に回答されなかったため、" "Helix は回収を開始しました。必要であれば再申請してください。" ), }, "data_access.lifecycle_it_digest": { "subject": "DAX 授权日报:{expiring_count} 条即将到期 · {expired_count} 条已过期 · {confirmation_lapsed_count} 条负责人确认逾期", "subject_en": "DAX grants daily — {expiring_count} expiring · {expired_count} expired · {confirmation_lapsed_count} owner confirmations lapsed", "subject_ja": "DAX 権限日報:期限間近 {expiring_count} · 期限切れ {expired_count} · 責任者確認超過 {confirmation_lapsed_count}", "body_text": ( "个人授权:{expiring_count} 条即将到期,{expired_count} 条已过期待处理,今日回收 " "{expired_revoked_count} 条。服务账号:{confirmation_due_count} 条待确认," "{confirmation_lapsed_count} 条逾期,今日回收 {confirmation_revoked_count} 条。" "员工提醒{employee_delivery};到期自动回收{auto_revoke}。" ), "body_text_en": ( "Personal grants: {expiring_count} expiring, {expired_count} expired awaiting action, " "{expired_revoked_count} revoked today. Service accounts: {confirmation_due_count} " "confirmations due, {confirmation_lapsed_count} lapsed, {confirmation_revoked_count} " "revoked today. Employee reminders {employee_delivery}; auto-revoke on expiry {auto_revoke}." ), "body_text_ja": ( "個人権限:期限間近 {expiring_count} 件、期限切れ未処理 {expired_count} 件、本日回収 " "{expired_revoked_count} 件。サービスアカウント:確認待ち {confirmation_due_count} 件、" "超過 {confirmation_lapsed_count} 件、本日回収 {confirmation_revoked_count} 件。" "従業員リマインドは{employee_delivery}、期限切れ自動回収は{auto_revoke}。" ), }, # ── Permission catalog in-app approval (migration 606) ───────── "permission.approval_requested": { "subject": "待你审批:{requester} 申请 {software} · {permission}", "subject_en": "Approval needed: {requester} requests {software} · {permission}", "subject_ja": "承認待ち:{requester} が {software} · {permission} を申請", "body_text": ( "{requester} 申请软件权限「{software} · {permission}」,理由:{reason}\n\n" "你是该软件的负责人 / IT 联系人,请在 Helix 待办里批准或拒绝(不走飞书)。" ), "body_text_en": ( "{requester} requests the software permission \"{software} · {permission}\". " "Reason: {reason}\n\nAs the software owner / IT contact, approve or reject it in your " "Helix to-dos (no Feishu step)." ), "body_text_ja": ( "{requester} がソフトウェア権限「{software} · {permission}」を申請しました。理由:{reason}\n\n" "ソフトウェアの責任者 / IT 担当として、Helix の To-Do で承認または却下してください(飛書は経由しません)。" ), }, "permission.approval_reminder": { "subject": "⏳ 权限申请待审批已 {days} 天:{requester} · {software} · {permission}", "subject_en": "⏳ Permission request waiting {days} day(s): {requester} · {software} · {permission}", "subject_ja": "⏳ 権限申請が {days} 日間承認待ち:{requester} · {software} · {permission}", "body_text": ( "{requester} 对「{software} · {permission}」的申请已等待 {days} 天。请在 Helix 待办里处理。" ), "body_text_en": ( "{requester}'s request for \"{software} · {permission}\" has been waiting {days} day(s). " "Please decide it in your Helix to-dos." ), "body_text_ja": ( "{requester} の「{software} · {permission}」申請は {days} 日間承認待ちです。Helix の To-Do で処理してください。" ), }, "permission.approval_escalated": { "subject": "权限申请超过 {days} 天未批,请 IT 接手:{requester} · {software} · {permission}", "subject_en": "Permission request unanswered for {days} days — IT please take over: {requester} · {software} · {permission}", "subject_ja": "権限申請が {days} 日間未処理 — IT が引き継いでください:{requester} · {software} · {permission}", "body_text": ( "{requester} 对「{software} · {permission}」的申请已等待 {days} 天,负责人未答复。" "IT 可直接在权限管理台批准或拒绝。" ), "body_text_en": ( "{requester}'s request for \"{software} · {permission}\" has waited {days} days without a " "decision from the owner. IT can approve or reject it in the permission console." ), "body_text_ja": ( "{requester} の「{software} · {permission}」申請は {days} 日間責任者から回答がありません。" "IT は権限管理台で直接承認または却下できます。" ), }, "permission.approval_decided": { "subject": "你的权限申请{decision_label}:{software} · {permission}", "subject_en": "Your permission request was {decision_label}: {software} · {permission}", "subject_ja": "権限申請が{decision_label}:{software} · {permission}", "body_text": ( "{approver} 已{decision_label}你对「{software} · {permission}」的申请。{note}{validity}" ), "body_text_en": ( "{approver} {decision_label} your request for \"{software} · {permission}\". {note}{validity}" ), "body_text_ja": ( "{approver} があなたの「{software} · {permission}」申請を{decision_label}。{note}{validity}" ), }, "permission.expired": { "subject": "软件权限已到期并回收:{software} · {permission}", "subject_en": "Software permission expired and revoked: {software} · {permission}", "subject_ja": "ソフトウェア権限が期限切れで回収されました:{software} · {permission}", "body_text": ( "你持有的「{software} · {permission}」有效期已到({expires_at}),Helix 已发起回收。" "如仍需要,请重新申请。" ), "body_text_en": ( "Your \"{software} · {permission}\" permission reached its validity end ({expires_at}); " "Helix has started the revoke. Request it again if you still need it." ), "body_text_ja": ( "保持していた「{software} · {permission}」の有効期限({expires_at})が終了し、Helix が回収を開始しました。" "必要であれば再申請してください。" ), }, "apikey.idle_it_digest": { "subject": "API Key 闲置月报:{idle_count} 把 Key 在 {period} 花费为 $0", "subject_en": "Idle API keys — {idle_count} key(s) had $0 spend in {period}", "subject_ja": "アイドル API キー月報:{period} に費用 $0 のキーが {idle_count} 件", "body_text": ( "{period} 有 {idle_count} 把 active 的 API Key 花费为 $0" "(其中 {streak2_count} 把已连续 ≥2 个月)。员工提醒当前{employee_delivery}。" "到管理台按「闲置」筛选处理。" ), "body_text_en": ( "{idle_count} active API key(s) had $0 spend in {period} " "({streak2_count} idle for 2+ months). Employee reminders are currently " "{employee_delivery}. Filter the admin console by \"idle\" to act." ), "body_text_ja": ( "{period} に費用 $0 だった有効な API キーは {idle_count} 件" "(うち {streak2_count} 件は 2 か月以上連続)。従業員へのリマインドは現在" "{employee_delivery}。管理コンソールで「アイドル」フィルタから対応してください。" ), }, # ── GitHub 组织(2026-08-29 补注册)───────────────────────────── # # 这三条通知一直在发,但 template_key 从未登记 —— 于是 # ``_select_template_field`` 找不到条目,收件人一律读到生产者 # 硬编码的那门语言,无论他的 ``preferred_language`` 是什么。这正是 # 平台做过 250 处清理要消灭的那类缺陷,守门 # ``test_every_sent_template_key_is_registered`` 抓到了它。 # # 文案取自调用点原有的中文,一字未改语义;en / ja 是补的。占位符与 # 调用点 ``context`` 一一对应(org_slug / github_login / ttl_days / # orgs)。 # ── 自动化异常闭环(2026-09-06)─────────────────────────────── # # 注册在这里,而不是只在 ``template_catalog`` 里分组:后者只决定 # 管理端编辑器的归类,**语言是这里选的**。没有条目的 key 会让 # ``_select_template_field`` 找不到东西,于是收件人一律读到生产者 # 硬编码的那门语言 —— 而本公司多数员工读英文。 # # 🔴 这两条曾躲过 ``test_every_sent_template_key_is_registered``: # 常量名是 ``TPL_*``(不含 TEMPLATE),调用点传的是名字而不是字面量, # 于是两条正则都不匹配。守门已按此加宽 —— 见该测试。 "automation_exception.subject_recovered": { "subject": "你在等的那一步已经完成", "subject_en": "The step you were waiting on is done", "subject_ja": "お待ちいただいていた処理が完了しました", "body_text": ( "{display_name},你好。\n\n" "此前有一步自动化没有走通,我们已经修好并重新执行完成 —— " "现在它已经生效,你不需要再做任何事。\n\n" "如果你发现它仍然不可用,请联系 IT。" ), "body_text_en": ( "Hi {display_name},\n\n" "Something we automate for you had failed. It is fixed and has " "now gone through — it is live, and there is nothing for you to " "do.\n\n" "If it still does not work for you, please contact IT." ), "body_text_ja": ( "{display_name} さん\n\n" "自動処理の一部が失敗していましたが、修正のうえ再実行が完了しました。" "現在は有効になっており、お客様の操作は不要です。\n\n" "まだ利用できない場合は IT までご連絡ください。" ), }, "automation_exception.owner_recurred": { "subject": "这个问题又回来了", "subject_en": "This problem has come back", "subject_ja": "この問題が再発しました", "body_text": ( "「{signature}」在被处理之后又出现了 {recurrence_count} 次。\n\n" "一个被解决之后又回来的问题,说明上次的处置没有解决根因。\n" "当前判定的根因:{root_cause}\n" "建议的下一步:{remedy}" ), "body_text_en": ( "\"{signature}\" has returned {recurrence_count} time(s) after " "being handled.\n\n" "A problem that comes back is a problem whose remedy was wrong.\n" "Current root cause: {root_cause}\n" "Suggested next step: {remedy}" ), "body_text_ja": ( "「{signature}」は対処後に {recurrence_count} 回再発しています。\n\n" "再発する問題は、前回の対処が根本原因に届いていなかったことを意味します。\n" "現在の根本原因:{root_cause}\n" "推奨される次の一手:{remedy}" ), }, "github_org_invited": { "subject": "GitHub 组织邀请已发出:{org_slug}", "subject_en": "Your GitHub org invite is out: {org_slug}", "subject_ja": "GitHub 組織への招待を送信しました:{org_slug}", "body_text": ( "你申请加入 {org_slug} 已通过,邀请已发到 GitHub 账号 " "{github_login}。\n\n" "请打开 https://github.com/{org_slug} 接受邀请。" "GitHub 的组织邀请 {ttl_days} 天后过期;过期了也不用重新申请," "Helix 会自动重发。" ), "body_text_en": ( "Your request to join {org_slug} was approved, and the invite " "has been sent to the GitHub account {github_login}.\n\n" "Open https://github.com/{org_slug} to accept it. GitHub org " "invites expire after {ttl_days} days — if yours does, there is " "nothing to re-request: Helix resends it automatically." ), "body_text_ja": ( "{org_slug} への参加申請が承認され、GitHub アカウント " "{github_login} 宛に招待を送信しました。\n\n" "https://github.com/{org_slug} を開いて承諾してください。" "GitHub の組織招待は {ttl_days} 日で期限切れになりますが、" "申請し直す必要はありません — Helix が自動で再送します。" ), }, "github_org_invite_resent": { "subject": "GitHub 组织邀请已重发:{org_slug}", "subject_en": "GitHub org invite resent: {org_slug}", "subject_ja": "GitHub 組織への招待を再送しました:{org_slug}", "body_text": ( "你加入 {org_slug} 的邀请上次没有被及时接受,已自动重发到 " "GitHub 账号 {github_login}。\n\n" "请打开 https://github.com/{org_slug} 接受邀请。" ), "body_text_en": ( "Your invite to {org_slug} was not accepted in time, so it has " "been resent automatically to the GitHub account " "{github_login}.\n\n" "Open https://github.com/{org_slug} to accept it." ), "body_text_ja": ( "{org_slug} への招待が期限内に承諾されなかったため、GitHub " "アカウント {github_login} 宛に自動で再送しました。\n\n" "https://github.com/{org_slug} を開いて承諾してください。" ), }, "github_org_missing_handle": { "subject": "补一下 GitHub 用户名,就能自动加入这些组织", "subject_en": "Add your GitHub username to be added automatically", "subject_ja": "GitHub ユーザー名の登録で自動参加できます", "body_text": ( "按你所在部门的规则,你本该被自动加入以下 GitHub 组织:\n" "{orgs}\n\n" "但我们这边没有你的 GitHub 用户名,所以这一步没能完成。到 " "Helix 的「个人设置 → 基本信息」登记一下 GitHub ID(提交后由你的 " "HR Ops 确认),之后同类规则会自动生效。" ), "body_text_en": ( "Under your department's rules you should have been added to " "these GitHub organizations automatically:\n" "{orgs}\n\n" "We do not have your GitHub username on file, so that step " "could not complete. Add it under Personal settings → Basic " "info in Helix (your HR Ops confirms it), and rules like this " "will apply automatically from then on." ), "body_text_ja": ( "所属部門のルールにより、以下の GitHub 組織へ自動で追加される" "はずでした:\n" "{orgs}\n\n" "ただ、こちらにあなたの GitHub ユーザー名が登録されていないため、" "この手順を完了できませんでした。Helix の「個人設定 → 基本情報」" "から GitHub ID を登録してください(HR Ops が確認します)。" "以降は同種のルールが自動で適用されます。" ), }, # ── GitHub ID 自助修改(migration 579)──────────────────────────── # 三条:推给审批人(HR Ops / HRBP)、通过回执、驳回回执。渠道只走 # feishu(DM + 站内信),不发邮件 —— 改一个 handle 不值得占一封工作 # 邮件。``verify_mark`` 是语言中立的记号(``✓`` / ``✗ (HTTP 404)`` / # ``— (n/a)``),因为 context 的值在发送时就定死了,一句中文会被英文/ # 日文模板原样印出来。 "github_id_change_requested": { "subject": "{requester_name} 申请修改 GitHub ID", "subject_en": "{requester_name} requested a GitHub ID change", "subject_ja": "{requester_name} さんが GitHub ID の変更を申請しました", "body_text": ( "Hi {person_name},\n\n" "{requester_name}({requester_email})申请把自己的 GitHub ID " "从 {old_github_id} 改为 {new_github_id},等待你确认。\n\n" " · GitHub 账号是否存在:{verify_mark}(— 表示我们没能问到 " "GitHub,不代表账号不存在)\n" " · 理由:{reason}\n\n" "确认后 Helix 会同步更新 HRIS(SuccessFactors)与飞书通讯录。" "请在 Helix「我的待办」中确认或驳回。\n\n" "— Helix IT 平台" ), "body_text_en": ( "Hi {person_name},\n\n" "{requester_name} ({requester_email}) asked to change their " "GitHub ID from {old_github_id} to {new_github_id} and is " "waiting for your confirmation.\n\n" " · GitHub account exists: {verify_mark} (— means we could not " "reach GitHub, not that the account is missing)\n" " · Reason: {reason}\n\n" "Once you confirm, Helix updates the HRIS (SuccessFactors) and " "the Feishu directory. Confirm or reject it under \"My to-dos\" " "in Helix.\n\n" "— Helix IT Platform" ), "body_text_ja": ( "{person_name} 様\n\n" "{requester_name}({requester_email})さんが GitHub ID を " "{old_github_id} から {new_github_id} へ変更することを申請し、" "あなたの確認をお待ちしています。\n\n" " · GitHub アカウントの存在:{verify_mark}(— は GitHub に" "問い合わせできなかったことを示し、アカウントが無いという" "意味ではありません)\n" " · 理由:{reason}\n\n" "確認すると、Helix が HRIS(SuccessFactors)と Feishu の" "連絡先を更新します。Helix の「マイタスク」から承認または" "却下してください。\n\n" "— Helix IT プラットフォーム" ), "card_button_label": "去确认", "card_button_label_en": "Review request", "card_button_label_ja": "確認画面を開く", }, "github_id_change_approved": { "subject": "你的 GitHub ID 已更新为 {new_github_id}", "subject_en": "Your GitHub ID is now {new_github_id}", "subject_ja": "GitHub ID が {new_github_id} に更新されました", "body_text": ( "Hi {person_name},\n\n" "{decider_name} 已确认你的修改,你的 GitHub ID 已从 " "{old_github_id} 更新为 {new_github_id}。Helix 与飞书通讯录会在" "几分钟内显示新值。\n\n" " · 备注:{note}\n" "{hris_note}\n" "— Helix IT 平台" ), "body_text_en": ( "Hi {person_name},\n\n" "{decider_name} confirmed your change — your GitHub ID went from " "{old_github_id} to {new_github_id}. Helix and the Feishu " "directory will show the new value within a few minutes.\n\n" " · Note: {note}\n" "{hris_note}\n" "— Helix IT Platform" ), "body_text_ja": ( "{person_name} 様\n\n" "{decider_name} さんが変更を承認しました。GitHub ID は " "{old_github_id} から {new_github_id} に更新されました。Helix と " "Feishu の連絡先には数分以内に反映されます。\n\n" " · 備考:{note}\n" "{hris_note}\n" "— Helix IT プラットフォーム" ), }, "github_id_change_rejected": { "subject": "你的 GitHub ID 修改申请被驳回", "subject_en": "Your GitHub ID change was rejected", "subject_ja": "GitHub ID の変更申請が却下されました", "body_text": ( "Hi {person_name},\n\n" "{decider_name} 驳回了你把 GitHub ID 改为 {new_github_id} 的" "申请,当前值保持不变。\n\n" " · 备注:{note}\n\n" "如果这是误判,直接找 {decider_name} 说明后重新提交即可。\n\n" "— Helix IT 平台" ), "body_text_en": ( "Hi {person_name},\n\n" "{decider_name} rejected your request to change your GitHub ID " "to {new_github_id}; the current value is unchanged.\n\n" " · Note: {note}\n\n" "If that was a mistake, talk to {decider_name} and submit " "again.\n\n" "— Helix IT Platform" ), "body_text_ja": ( "{person_name} 様\n\n" "{decider_name} さんが GitHub ID を {new_github_id} に変更する" "申請を却下しました。現在の値は変更されていません。\n\n" " · 備考:{note}\n\n" "誤りであれば、{decider_name} さんに説明のうえ再度申請して" "ください。\n\n" "— Helix IT プラットフォーム" ), }, # The SINGLE new-hire onboarding email to the employee's work mailbox # (consolidated 2026-06-29 — previously a thin "welcome" email here PLUS # a separate code-rendered "hire-day IT info" email both landed in the # work inbox the same day). Sent once per hire, deduped by # ``identity_persons.it_info_sent_at``, fired from BOTH the confirm path # and the hire_day cron (see hire_day_service.notify_person / # hr_sync_coordinator.send_onboarding_it_info). Channels: feishu + email. # The dynamic {software_list}/{hardware_list} arrive pre-formatted in the # context. The first-login PASSWORD is NOT here — it ships separately to # the personal email from account_provisioner (the work mailbox is empty # until the employee logs in). "hire_day_it_info": { "subject": "🎉 欢迎加入 {company_name}!你的 IT 账号与第一天指南", "subject_en": "🎉 Welcome to {company_name} — your IT account & day-one guide", "subject_ja": "🎉 {company_name} へようこそ — IT アカウントと初日ガイド", "body_text": ( "{person_name},你好,欢迎加入 {company_name}!🎉\n\n" "下面是入职第一天要用到的关键信息,按 基础信息 / IT / " "HR / 财务 / 法务·合规·安全 分好——花几分钟读完,第一天会顺很多。\n\n" "━━━━━ ① 基础信息 ━━━━━\n" " · 公司邮箱:{person_email}\n" " · 部门:{department}\n" " · 工作地:{location}\n" " · 直属经理:{manager_name}\n" " · HRBP:{hrbp_name}\n\n" "━━━━━ ② IT · 账号与日常工具 ━━━━━\n" "为了帮助你顺利上手,我们整理了一些与日常工作相关的必备 IT 指南," "请你抽空仔细阅读:\n" " · 入职首日必读 —— 入职首日 - PingCAP IT 指南(中国区):\n" " https://pingcap.feishu.cn/wiki/wikcnKhelzg1rdNVKFUmMXzfvoW\n" " · 电脑设备领取:如尚未领或收到笔记本电脑," "请联系当地 HR Ops 联系人领取\n" " · 一站式 IT 服务入口:IT Help Desk 随时为你提供支持" "({it_contact})\n" " 🔹 Lark/飞书搜索:直接搜索「IT Help Desk」即可快速进入\n" " 🔹 Lark/飞书路径:工作台 → 服务台 → IT Help Desk\n" " 🔹 快捷链接:https://applink.feishu.cn/T8TQpyK5mVGj\n\n" "━━━━━ ③ HR 相关 ━━━━━\n" " · HR 相关问题都可以优先咨询 HR 服务台:\n" " https://applink.feishu.cn/T97IDmdiMsag\n" " · 假期 / 调休 在 HRIS (SuccessFactors) 查看与申请({hris_manual_title}):\n" " {hris_manual_url}\n" " · 新员工培训课程入口,请直接从 飞书 → 工作台 → TiLearning 进入\n\n" "━━━━━ ④ 财务 · 报销 ━━━━━\n" " · 差旅 / 日常 / 采购报销在【汇联易】提交" "(飞书工作台 - Featured Apps - Helios)\n" " · 有疑问请至飞书财务工作台查询\n\n" "━━━━━ ⑤ 法务 · 合规 · 安全 ━━━━━\n" " · 🔐 入职必修:在学习平台完成「安全学习与考试」,请务必按时通过:\n" " https://learning.pingcap.com/o2o/#/project/detail/1935586295680002050\n" " · 请遵守已签署的保密协议(NDA)与信息安全规范\n" " · 需要 SOC2 合规材料(客户审计等)?在【飞书工作台】发起" "「SOC2 申请」审批流程即可,无需邮件\n\n" "━━━━━ 📝 新员工入职体验问卷 ━━━━━\n" " · 今天抽几分钟填写,你的真实感受会让入职流程越来越好:\n" " https://pingcap.feishu.cn/share/base/form/shrcnx5vNqWfJ6yUgng9A1C444f\n\n" "有任何不清楚的地方,随时找我们——IT 团队很乐意帮你。\n\n" "{blessing_zh}" ), "body_text_en": ( "Hi {person_name}, welcome to {company_name}! 🎉\n\n" "We're so glad you're here! Here's the key info for your first day, grouped" " as Basics / IT / HR / Finance / Legal·Compliance·Security" " — a few minutes now makes day one much smoother.\n\n" "━━━━━ 1. The basics ━━━━━\n" " · Work email: {person_email}\n" " · Team: {department}\n" " · Location: {location}\n" " · Manager: {manager_name}\n" " · HRBP: {hrbp_name}\n\n" "━━━━━ 2. IT — accounts & everyday tools ━━━━━\n" "To help you get started smoothly, we've put together some" " essential IT guides related to your daily work. Please take a" " moment to review them carefully:\n" " · Must-Read on Your First Day — Day 1 - IT Guide at PingCAP (Global):\n" " https://pingcap.feishu.cn/wiki/Tu9AwCr2ui4fRtkQCdScG99Unaf\n" " · Laptop Pickup: no laptop yet? Contact your local HR Ops to pick one up\n" " · Your One-Stop IT Service Portal: IT Help Desk is always here for you ({it_contact})\n" " 🔹 Lark/Feishu Search: simply type \"IT Help Desk\" for instant access\n" " 🔹 Lark/Feishu Path: Workbench → Service Desk → IT Help Desk\n" " 🔹 Quick Link: https://applink.feishu.cn/T8UNQO6Y5TAl\n\n" "━━━━━ 3. HR ━━━━━\n" " · For anything HR, the HR helpdesk is your first stop:\n" " https://applink.feishu.cn/T97IDmdiMsag\n" " · View and request leave / time-off in HRIS (SuccessFactors)" " ({hris_manual_title}):\n" " {hris_manual_url}\n" " · To access the New Employee Orientation Online Course, please go" " directly to Lark > Workplace > TiLearning\n\n" "━━━━━ 4. Finance — reimbursement ━━━━━\n" " · Submit travel / daily / procurement reimbursements via Navan" " (Lark Workplace > Featured Apps > Navan)\n" " · Questions? Check the Lark Finance Desk\n\n" "━━━━━ 5. Legal · Compliance · Security ━━━━━\n" " · 🔐 Required for onboarding: complete the security training & exam on" " the learning platform, please pass it on time:\n" " https://learning.pingcap.com/o2o/#/project/detail/1935586295680002050\n" " · Please honour the NDA you signed and our information-security policies\n" " · Need SOC2 materials (customer audits, etc.)? Start a \"SOC2 request\"" " approval on the Feishu workspace — no email needed\n\n" "━━━━━ 📝 New Hire Onboarding Survey ━━━━━\n" " · Take a few minutes today — your honest feedback keeps making" " onboarding better:\n" " https://pingcap.feishu.cn/share/base/form/shrcnx5vNqWfJ6yUgng9A1C444f\n\n" "Anything unclear? Reach out anytime — IT is happy to help.\n\n" "{blessing_en}" ), "body_text_ja": ( "{person_name} さん、{company_name} へようこそ。🎉\n\n" "入社初日に必要となる主な情報を、基本情報 / IT / " "HR / 財務 / 法務・コンプライアンス・セキュリティ に分けてご案内します。" "数分で読み終わりますので、初日を迎える前にご確認ください。\n\n" "━━━━━ ① 基本情報 ━━━━━\n" " · 会社メールアドレス:{person_email}\n" " · 部門:{department}\n" " · 勤務地:{location}\n" " · 直属の上長:{manager_name}\n" " · HRBP:{hrbp_name}\n\n" "━━━━━ ② IT・アカウントと日常ツール ━━━━━\n" "スムーズにお仕事を始めていただけるよう、日常業務に必要な IT ガイドを" "まとめました。お時間のあるときにご一読ください:\n" " · 入社初日の必読 —— Day 1 - IT Guide at PingCAP (Global):\n" " https://pingcap.feishu.cn/wiki/Tu9AwCr2ui4fRtkQCdScG99Unaf\n" " · ノート PC の受け取り:まだ受け取っていない場合は、" "各拠点の HR Ops 担当者にご連絡ください\n" " · IT サービスの総合窓口:IT Help Desk がいつでもサポートします" "({it_contact})\n" " 🔹 Feishu/Lark で検索:「IT Help Desk」と入力してください\n" " 🔹 Feishu/Lark の経路:ワークプレイス → ヘルプデスク → " "IT Help Desk\n" " 🔹 クイックリンク:https://applink.feishu.cn/T8UNQO6Y5TAl\n\n" "━━━━━ ③ HR 関連 ━━━━━\n" " · HR に関するご質問は、まず HR ヘルプデスクにお問い合わせください:\n" " https://applink.feishu.cn/T97IDmdiMsag\n" " · 休暇・振替休日は HRIS (SuccessFactors) で確認・申請できます({hris_manual_title}):\n" " {hris_manual_url}\n" " · 新入社員向けオンライン研修は、Feishu → ワークプレイス → TiLearning から" "直接ご受講ください\n\n" "━━━━━ ④ 財務・経費精算 ━━━━━\n" " · 出張 / 日常 / 購買の経費精算は Navan から申請してください" "(Feishu ワークプレイス > Featured Apps > Navan)\n" " · ご不明な点は Feishu の財務ワークプレイスでご確認ください\n\n" "━━━━━ ⑤ 法務・コンプライアンス・セキュリティ ━━━━━\n" " · 🔐 入社時の必須項目:学習プラットフォームで「セキュリティ研修と試験」を" "期限内に修了してください:\n" " https://learning.pingcap.com/o2o/#/project/detail/1935586295680002050\n" " · 署名済みの秘密保持契約(NDA)および情報セキュリティ規程を遵守してください\n" " · SOC2 のコンプライアンス資料(顧客監査など)が必要な場合は、【Feishu ワークプレイス】で" "「SOC2 申請」の承認フローを起票してください。メールでのご依頼は不要です\n\n" "━━━━━ 📝 新入社員オンボーディングアンケート ━━━━━\n" " · 本日中に数分お時間をいただければ幸いです。率直なご意見が" "オンボーディングの改善につながります:\n" " https://pingcap.feishu.cn/share/base/form/shrcnx5vNqWfJ6yUgng9A1C444f\n\n" "ご不明な点があれば、いつでも IT チームまでお気軽にお問い合わせください。\n\n" "{blessing_ja}" ), }, "transfer_received": { "subject": "[待处理] {initiator_name} 转让了一项资产给你", "subject_en": "[Action required] {initiator_name} transferred an asset to you", "subject_ja": "[要対応] {initiator_name} さんから資産が引き継がれました", "body_text": ( "{initiator_name} 发起了一项资产转让,请登录 Helix 平台确认接收:\n\n" " · 资产: {asset_label}\n" " · 类型: {asset_type}\n" " · 理由: {reason}\n\n" "{cta_line}\n\n" "— Helix IT 平台" ), "body_text_en": ( "{initiator_name} initiated an asset transfer to you. " "Please review and accept in Helix:\n\n" " · Asset: {asset_label}\n" " · Type: {asset_type}\n" " · Reason: {reason}\n\n" "{cta_line}\n\n" "— Helix IT Platform" ), "body_text_ja": ( "{initiator_name} さんが資産の引き継ぎを開始しました。" "Helix にログインして受け取りをご確認ください。\n\n" " · 資産: {asset_label}\n" " · 種別: {asset_type}\n" " · 理由: {reason}\n\n" "{cta_line}\n\n" "— Helix IT プラットフォーム" ), "card_button_label": "去 Helix 接收", "card_button_label_en": "Accept in Helix", "card_button_label_ja": "Helix で受け取る", }, # Sent to the leaver themselves the moment an IT admin clicks # "Mark as Leaver" (or HRIS sync confirms an offboard event). # Pairs with ``offboarding.process_started_manager`` (next entry) # so both leaver and manager learn of the flow at the same moment. # Added 2026-05-08 alongside the offboarding-flow hardening PR — the # platform previously assumed the leaver would learn via the admin # panel UX, which never happens for distributed teams. "offboarding.process_started_leaver": { "subject": "[离职流程] 你的离职流程已启动 — 请完成资产/权限交接", "subject_en": ( "[Offboarding] Your offboarding has started — " "please complete handover" ), "subject_ja": ( "[退職手続き] 退職プロセスが開始されました — " "資産・権限の引き継ぎをお願いします" ), "body_text": ( "Hi {person_name},\n\n" "你的离职流程已启动,预计离职日 [r]{leaver_date}[/r]。\n\n" "请尽快登录 Helix 自助门户,逐项完成:\n" " · 资产返还(电脑、配件)\n" " · 共享权限交接(Drive、邮件组、SSO 应用)\n" " · 文档与会议数据交接\n\n" "未在离职日前处理的项,会按系统兜底策略自动转给你的直属经理。\n\n" "{cta_line}\n\n" "— Helix IT 平台" ), "body_text_en": ( "Hi {person_name},\n\n" "Your offboarding has been initiated. Last working day: " "[r]{leaver_date}[/r].\n\n" "Please sign in to Helix self-service and complete:\n" " · Asset returns (laptop, accessories)\n" " · Shared-resource handover (Drive, email groups, SSO apps)\n" " · Document and meeting data hand-offs\n\n" "Anything still pending on your leaver date will be auto-" "transferred to your manager as a fallback.\n\n" "{cta_line}\n\n" "— Helix IT Platform" ), "body_text_ja": ( "{person_name} さん\n\n" "退職プロセスが開始されました。最終出社日は [r]{leaver_date}[/r] です。\n\n" "Helix セルフサービスポータルにログインし、以下の項目を順にご対応ください。\n" " · 資産の返却(PC、周辺機器)\n" " · 共有権限の引き継ぎ(Drive、メーリングリスト、SSO アプリ)\n" " · ドキュメントおよび会議データの引き継ぎ\n\n" "最終出社日までに処理されなかった項目は、" "システムの既定動作により直属の上長へ自動的に引き継がれます。\n\n" "{cta_line}\n\n" "— Helix IT プラットフォーム" ), "card_button_label": "去 Helix 处理", "card_button_label_en": "Go to Helix", "card_button_label_ja": "Helix で対応する", }, # Sent to the leaver's direct manager the moment offboarding starts. # Mirrors ``offboarding.process_started_leaver`` so both ends of # the handover learn at the same time. Distinct template so the # leaver-facing one says "your" + handover steps while the # manager-facing one says "your team member" + "watch for transfer # requests". "offboarding.process_started_manager": { "subject": ( "[团队成员离职] {leaver_name} 离职流程已启动 — 关注资产转交" ), "subject_en": ( "[Team member offboarding] {leaver_name} — watch for transfer " "requests" ), "subject_ja": ( "[メンバーの退職] {leaver_name} さんの退職プロセスが開始 — " "資産の引き継ぎにご注意ください" ), "body_text": ( "Hi {manager_name},\n\n" "你的下属 [b]{leaver_name}[/b] ({leaver_email}) 的离职流程已启动," "预计离职日 [r]{leaver_date}[/r]。\n\n" "请关注 Helix 上的资产转交请求 —— 未在离职日前处理完毕的" "资产会按兜底策略自动转给你接收。\n\n" "{cta_line}\n\n" "— Helix IT 平台" ), "body_text_en": ( "Hi {manager_name},\n\n" "Your team member [b]{leaver_name}[/b] ({leaver_email}) is " "leaving on [r]{leaver_date}[/r].\n\n" "Please watch Helix for incoming transfer requests — " "anything still pending on the leaver date will auto-transfer " "to you as a fallback.\n\n" "{cta_line}\n\n" "— Helix IT Platform" ), "body_text_ja": ( "{manager_name} さん\n\n" "部下の [b]{leaver_name}[/b] さん({leaver_email})の退職プロセスが開始されました。" "最終出社日は [r]{leaver_date}[/r] です。\n\n" "Helix に届く資産の引き継ぎ依頼をご確認ください。" "最終出社日までに処理されなかった資産は、既定動作により" "あなたへ自動的に引き継がれます。\n\n" "{cta_line}\n\n" "— Helix IT プラットフォーム" ), "card_button_label": "查看 Helix", "card_button_label_en": "Open Helix", "card_button_label_ja": "Helix を開く", }, # Sent when a process transitions to READY (all blockers resolved). # Audience: IT admin pool — operator's signal that suspend_sso is # safe to click. "offboarding.ready": { "subject": ( "[离职就绪] {leaver_name} 所有阻塞项已结清,可执行 SSO suspend" ), "subject_en": ( "[Offboarding ready] {leaver_name} — all blockers cleared, " "ready for SSO suspend" ), "subject_ja": ( "[退職準備完了] {leaver_name} さん — 未処理項目がすべて解消、" "SSO の停止が可能です" ), "body_text": ( "员工 {leaver_name} ({leaver_email}) 的离职流程已就绪:所有" "阻塞项均已 resolved。可在 Helix 上执行 [b]Suspend SSO[/b] " "完成 vendor 端账号停用。\n\n" "离职日: [r]{leaver_date}[/r]\n\n" "{cta_line}\n\n" "— Helix IT 平台" ), "body_text_en": ( "{leaver_name} ({leaver_email}) offboarding is ready: every " "blocker resolved. Run [b]Suspend SSO[/b] in Helix to " "execute the vendor-side suspends.\n\n" "Leaver date: [r]{leaver_date}[/r]\n\n" "{cta_line}\n\n" "— Helix IT Platform" ), "body_text_ja": ( "{leaver_name} さん({leaver_email})の退職プロセスは準備完了です。" "未処理項目はすべて解消されました。Helix で [b]Suspend SSO[/b] を実行し、" "ベンダー側アカウントの停止を完了してください。\n\n" "最終出社日: [r]{leaver_date}[/r]\n\n" "{cta_line}\n\n" "— Helix IT プラットフォーム" ), "card_button_label": "去 Helix 执行", "card_button_label_en": "Open in Helix", "card_button_label_ja": "Helix で実行する", }, # Sent when a process transitions to COMPLETED (operator clicked # "Complete"). Audience: leaver's manager — final acknowledgement. "offboarding.completed": { "subject": "[离职已完成] {leaver_name} 的离职流程已闭环", "subject_en": ( "[Offboarding complete] {leaver_name} has been fully offboarded" ), "subject_ja": "[退職完了] {leaver_name} さんの退職プロセスが完了しました", "body_text": ( "员工 [b]{leaver_name}[/b] ({leaver_email}) 的离职流程已" "完成,所有资产 / 权限 / 账号均已处理完毕。\n\n" "如有遗漏问题请联系 IT 团队。\n\n" "— Helix IT 平台" ), "body_text_en": ( "{leaver_name} ({leaver_email}) has been fully offboarded — " "every asset / permission / account has been handled.\n\n" "Contact IT if any item appears to be still outstanding.\n\n" "— Helix IT Platform" ), "body_text_ja": ( "[b]{leaver_name}[/b] さん({leaver_email})の退職プロセスが" "完了しました。資産・権限・アカウントはすべて処理済みです。\n\n" "未処理の項目がございましたら IT チームまでご連絡ください。\n\n" "— Helix IT プラットフォーム" ), }, "leaver_reminder": { "subject": "[提醒] {person_name} 距离离职还有 [r]{days_left}[/r] 天", "subject_en": "[Reminder] {person_name} is leaving in [r]{days_left}[/r] days", "subject_ja": "[リマインド] {person_name} さんの退職まであと [r]{days_left}[/r] 日です", "body_text": ( "员工 {person_name} ({person_email}) 的预计离职日为 [r]{leaver_date}[/r]。\n\n" "仍需处理的资产:\n{pending_blockers}\n\n" "请尽快在系统中完成转让或归还。到离职日当天,未处理的项目将自动转给其经理兜底。\n\n" "{cta_line}\n\n" "— Helix IT 平台" ), "body_text_en": ( "{person_name} ({person_email}) is scheduled to leave on " "[r]{leaver_date}[/r].\n\n" "Outstanding items:\n{pending_blockers}\n\n" "Please complete transfers / returns. On the leaver date, " "anything unfinished is auto-transferred to their manager.\n\n" "{cta_line}\n\n" "— Helix IT Platform" ), "body_text_ja": ( "{person_name} さん({person_email})の最終出社日は [r]{leaver_date}[/r] です。\n\n" "未処理の資産:\n{pending_blockers}\n\n" "システム上で引き継ぎまたは返却を早めにご完了ください。" "最終出社日当日に未処理の項目は、上長へ自動的に引き継がれます。\n\n" "{cta_line}\n\n" "— Helix IT プラットフォーム" ), "card_button_label": "处理离职项", "card_button_label_en": "Handle leaver items", "card_button_label_ja": "退職項目を処理する", }, "alias_created": { "subject": "[Helix] 邮箱别名 {alias_email} 已创建", "subject_en": "[Helix] Email alias {alias_email} created", "subject_ja": "[Helix] メールエイリアス {alias_email} を作成しました", "body_text": ( "Hi {person_name},\n\n" "邮箱别名 {alias_email} 已添加到 {target_email}。\n" "发往这个别名的邮件会自动投递到你的主邮箱。\n\n" "操作人:{actor_name}\n\n" "— Helix IT 平台" ), "body_text_en": ( "Hi {person_name},\n\n" "Email alias {alias_email} was added to {target_email}. " "Mail to this alias will be delivered to your primary inbox.\n\n" "Performed by: {actor_name}\n\n" "— Helix IT Platform" ), "body_text_ja": ( "{person_name} さん\n\n" "メールエイリアス {alias_email} を {target_email} に追加しました。\n" "このエイリアス宛のメールは、主メールボックスへ自動的に配信されます。\n\n" "実施者:{actor_name}\n\n" "— Helix IT プラットフォーム" ), }, "alias_deleted": { "subject": "[Helix] 邮箱别名 {alias_email} 已删除", "subject_en": "[Helix] Email alias {alias_email} removed", "subject_ja": "[Helix] メールエイリアス {alias_email} を削除しました", "body_text": ( "Hi {person_name},\n\n" "邮箱别名 {alias_email} 已从 {target_email} 中删除。\n" "之后发往这个地址的邮件将不再投递。\n\n" "操作人:{actor_name}\n\n" "如非本人操作,请立即联系 IT 团队。\n\n" "— Helix IT 平台" ), "body_text_en": ( "Hi {person_name},\n\n" "Email alias {alias_email} was removed from {target_email}. " "Mail sent to this address will no longer be delivered.\n\n" "Performed by: {actor_name}\n\n" "If you didn't expect this, contact IT immediately.\n\n" "— Helix IT Platform" ), "body_text_ja": ( "{person_name} さん\n\n" "メールエイリアス {alias_email} を {target_email} から削除しました。\n" "今後このアドレス宛のメールは配信されません。\n\n" "実施者:{actor_name}\n\n" "お心当たりがない場合は、直ちに IT チームまでご連絡ください。\n\n" "— Helix IT プラットフォーム" ), }, "leaver_escalation": { "subject": "[兜底转让] {person_name} 的 [r]{count}[/r] 项资产已自动转给你", "subject_en": "[Auto-transfer] [r]{count}[/r] items from {person_name} were transferred to you", "subject_ja": "[自動引き継ぎ] {person_name} さんの資産 [r]{count}[/r] 件があなたへ引き継がれました", "body_text": ( "{person_name} ({person_email}) 今日离职,仍有 [r]{count}[/r] 项未处理的资产已按系统" "策略自动转让给你(其直属经理)。请登录 Helix 接收:\n\n" "{asset_list}\n\n" "{cta_line}\n\n" "— Helix IT 平台" ), "body_text_en": ( "{person_name} ({person_email}) left today. [r]{count}[/r] " "outstanding items were auto-transferred to you (their " "manager). Please review and confirm in Helix:\n\n" "{asset_list}\n\n" "{cta_line}\n\n" "— Helix IT Platform" ), "body_text_ja": ( "{person_name} さん({person_email})が本日退職しました。未処理の資産 " "[r]{count}[/r] 件が、システムの既定動作により直属の上長である" "あなたへ自動的に引き継がれました。Helix で受け取りをご確認ください。\n\n" "{asset_list}\n\n" "{cta_line}\n\n" "— Helix IT プラットフォーム" ), "card_button_label": "去 Helix 接收", "card_button_label_en": "Accept in Helix", "card_button_label_ja": "Helix で受け取る", }, # Round-14 (2026-05-08): when a manager / colleague accepts an # AssetTransferRequest (hardware, email_group, api_key, ...) IT gets # a fire-and-forget heads-up. Without this, hardware silently changes # hands during offboarding and the IT inventory diverges from reality # until someone runs the reconcile job. Body intentionally has no # inline portal URL - the action button (rendered from ``url=``) is # the only CTA so dev / staging do not leak ``localhost`` into the # message and prod stays clean even if portal.base_url is empty. "transfer_accepted_it": { "subject": "[资产转让] {recipient_name} 已接收 {initiator_name} 的 {asset_label}", "subject_en": "[Asset transfer] {recipient_name} accepted {asset_label} from {initiator_name}", "subject_ja": "[資産の引き継ぎ] {recipient_name} さんが {initiator_name} さんの {asset_label} を受け取りました", "body_text": ( "资产转让已完成,IT 知会:\n\n" " · 资产: {asset_label}\n" " · 类型: {asset_type}\n" " · 原持有人: {initiator_name} ({initiator_email})\n" " · 接收人: {recipient_name} ({recipient_email})\n" " · 流转背景: {context_label}\n" " · 接收时间: {accepted_at}\n\n" "{cta_line}\n\n" "— Helix IT 平台" ), "body_text_en": ( "Asset transfer completed - heads-up to IT:\n\n" " · Asset: {asset_label}\n" " · Type: {asset_type}\n" " · Previous holder: {initiator_name} ({initiator_email})\n" " · New holder: {recipient_name} ({recipient_email})\n" " · Context: {context_label}\n" " · Accepted at: {accepted_at}\n\n" "{cta_line}\n\n" "— Helix IT Platform" ), "body_text_ja": ( "資産の引き継ぎが完了しました。IT へのお知らせです。\n\n" " · 資産: {asset_label}\n" " · 種別: {asset_type}\n" " · 旧保有者: {initiator_name} ({initiator_email})\n" " · 新保有者: {recipient_name} ({recipient_email})\n" " · 引き継ぎの経緯: {context_label}\n" " · 受領日時: {accepted_at}\n\n" "{cta_line}\n\n" "— Helix IT プラットフォーム" ), "card_button_label": "查看转让记录", "card_button_label_en": "View transfer record", "card_button_label_ja": "引き継ぎ履歴を見る", }, # T+0 grace-window IT push (round-13 + round-14). Replaces the # inline body that used to leak ``localhost:3000`` URLs in dev when # ``portal.base_url`` was unset. The button at the bottom is the # only CTA - body lists the pending resources and what the system # will auto-do at the 24h boundary. "leaver_it_grace": { "subject": ( "[离职宽限期] {person_name} ({employee_no}) 待人工接收 [r]{pending_count}[/r] 项 — 24h 后系统将兜底" ), "subject_en": ( "[Leaver grace] {person_name} ({employee_no}) has [r]{pending_count}[/r] item(s) awaiting human handover - system will auto-handle in 24h" ), "subject_ja": ( "[退職猶予期間] {person_name} さん({employee_no})に手動対応待ちの項目が [r]{pending_count}[/r] 件あります — 24 時間後にシステムが自動処理します" ), "body_text": ( "员工 {person_name} ({person_email}) 离职流程已进入宽限期。\n" "离职日期: [r]{leaver_date}[/r]\n" "发现日期: {discovery_date}\n\n" "待人工决定接收人 ({pending_count} 项):\n{blocker_lines}\n\n" "{immediate_revoked_block_zh}" "如需手动转让 / 标记完成,请在 24 小时内操作。\n" "{cta_line}\n\n" "超过 24 小时仍未处理,系统将自动:\n" " · 硬件 → 归属不会自动转给经理。已在离职控制台设置处置方式的" "按设置执行(本人留存 / 退回办公室 / 转给指定接收人);未设置的" "保持在离职人名下并标记「离职留存」,同时通知当地资产管理员" "线下跟进回收\n" " · API key / 域名 / 云资源 / IAM / 邮件组 → 转给经理\n\n" "(软件席位由 auto_revoke 后台任务在离职日 +1 当地零点自动 revoke,不在此列表。)\n\n" "— Helix IT 平台" ), "body_text_en": ( "Offboarding for {person_name} ({person_email}) has entered " "the grace window.\n" "Leaver date: [r]{leaver_date}[/r]\n" "Discovered: {discovery_date}\n\n" "Awaiting human-decided successor ({pending_count}):\n{blocker_lines}\n\n" "{immediate_revoked_block_en}" "If you want to transfer / mark resolved manually, please " "act within the next 24 hours.\n" "{cta_line}\n\n" "After 24h the system will auto:\n" " · Hardware -> ownership is NOT auto-moved to the manager. " "If a disposition was set in the offboarding console it is " "applied (leaver keeps / recycle to office / transfer to a " "named person); otherwise the device stays under the leaver " "and is flagged 'held by leaver', and the local asset admin " "is notified to chase the physical return\n" " · API key / domain / cloud resources / IAM / mail groups -> manager\n\n" "(Software seats are auto-revoked by a background task at " "leaver_date+1 local midnight and are not part of this list.)\n\n" "— Helix IT Platform" ), "body_text_ja": ( "{person_name} さん({person_email})の退職プロセスが猶予期間に入りました。\n" "最終出社日: [r]{leaver_date}[/r]\n" "検出日: {discovery_date}\n\n" "引き継ぎ先の指定が必要な項目({pending_count} 件):\n{blocker_lines}\n\n" "{immediate_revoked_block_ja}" "手動で引き継ぐ、または完了として処理する場合は、24 時間以内にご対応ください。\n" "{cta_line}\n\n" "24 時間を過ぎても未処理の場合、システムが自動的に以下を実行します。\n" " · ハードウェア → 所有者が自動的に上長へ移ることはありません。" "退職コンソールで処置方法が設定されていればそのとおりに実行します" "(本人保持 / オフィスへ返却 / 指定の引き継ぎ先へ移管)。" "未設定の場合は退職者の名義のまま「退職者保持」として記録し、" "拠点の資産管理者に現物回収の対応を依頼します\n" " · API キー / ドメイン / クラウドリソース / IAM / メーリングリスト → 上長へ引き継ぎ\n\n" "(ソフトウェアライセンスは auto_revoke バックグラウンドジョブが" "最終出社日の翌日、現地時間 0 時に自動失効させるため、この一覧には含まれません。)\n\n" "— Helix IT プラットフォーム" ), "card_button_label": "去 Helix 查看", "card_button_label_en": "Open in Helix", "card_button_label_ja": "Helix で確認する", }, # MDM device cleanup (2026-08-26). The ONE thing an MDM admin is # asked to do, and it is not triggered by anybody leaving: an asset # left company ownership while a device carrying its serial was still # enrolled. Says the serial out loud, because that is what the admin # searches for in the console, and says what happened to the machine, # because "remove this" without a reason is a request nobody trusts. "mdm.device_cleanup": { "subject": "[Helix] 请从 MDM 移除已处置设备: {device_name}", "subject_en": "[Helix] Please remove a disposed device from MDM: {device_name}", "subject_ja": "[Helix] 処分済みデバイスを MDM から削除してください: {device_name}", "body_text": ( "Hi,\n\n" "以下设备已完成资产处置,不再属于公司资产,但仍在 " "{provider} 中处于纳管状态,请在 {provider} 控制台移除该设备记录:\n\n" " · 设备: {device_name}({device_model})\n" " · 序列号: {serial_number}\n" " · 资产编号: {asset_tag}\n" " · 处置方式: {disposal_label}\n" " · 处置时间: {disposed_on}\n" " · {holder_line}\n\n" "说明: 员工离职本身不需要清理 MDM —— 设备回收后重装再分配," "全程保持纳管即可。只有设备离开公司资产(回购 / 外售 / 报废 / 报损)" "才需要从 MDM 移除。\n\n" "完成后: 飞书卡片可一键确认(无需登录)。下一次设备同步若在 " "{provider} 中查不到该设备,也会自动结案。\n\n" "— Helix IT 平台" ), "body_text_en": ( "Hi,\n\n" "The device below has been disposed of and is no longer a " "company asset, but {provider} still has it enrolled. Please " "remove the record in the {provider} console:\n\n" " · Device: {device_name} ({device_model})\n" " · Serial: {serial_number}\n" " · Asset tag: {asset_tag}\n" " · Disposal: {disposal_label}\n" " · Disposed on: {disposed_on}\n" " · {holder_line}\n\n" "Note: an employee leaving does not require MDM cleanup — a " "returned machine is re-imaged and reassigned while staying " "enrolled. Only a device leaving company ownership (buyback, " "sale, scrap, write-off) needs removing.\n\n" "When done: confirm with one click on the Feishu card (no " "login needed). The next device sync also closes this " "automatically if {provider} no longer reports the device.\n\n" "— Helix IT Platform" ), "body_text_ja": ( "お世話になっております。\n\n" "以下のデバイスは資産処分が完了し、すでに会社資産ではありませんが、" "{provider} では引き続き管理対象になっています。{provider} の管理" "コンソールから該当レコードを削除してください。\n\n" " · デバイス: {device_name}({device_model})\n" " · シリアル番号: {serial_number}\n" " · 資産番号: {asset_tag}\n" " · 処分区分: {disposal_label}\n" " · 処分日: {disposed_on}\n" " · {holder_line}\n\n" "補足: 従業員の退職自体では MDM の整理は不要です。返却された端末は" "再セットアップして次の担当者へ引き継ぐため、管理対象のままで問題" "ありません。会社資産から外れた場合(買い取り / 売却 / 廃棄 / 紛失)" "のみ削除が必要です。\n\n" "完了後: Feishu カードからワンクリックで確認できます(ログイン不要)。" "次回のデバイス同期で {provider} に該当デバイスが見つからない場合も、" "自動的にクローズされます。\n\n" "— Helix IT プラットフォーム" ), "card_button_label": "✅ 我已移除,标记完成", "card_button_label_en": "✅ Removed — mark done", "card_button_label_ja": "✅ 削除しました。完了にする", }, # The nudge for a cleanup nobody has actioned. Separate template # rather than a flag on the one above, because the sentence that # matters changes: the first card explains a policy, this one reports # that the tenant is still returning a device the company sold. "mdm.device_cleanup_reminder": { "subject": ( "[Helix] 清理待办已开 {days_open} 天,设备仍在 MDM 中: {device_name}" ), "subject_en": ( "[Helix] Cleanup open {days_open} days, still enrolled in MDM: " "{device_name}" ), "subject_ja": ( "[Helix] 依頼から {days_open} 日経過、MDM に残っています: " "{device_name}" ), "body_text": ( "Hi,\n\n" "这条清理待办已经开了 {days_open} 天,最近一次同步仍能在 " "{provider} 中查到该设备,说明尚未从控制台移除:\n\n" " · 设备: {device_name}({device_model})\n" " · 序列号: {serial_number}\n" " · 资产编号: {asset_tag}\n" " · 处置方式: {disposal_label}\n" " · 处置时间: {disposed_on}\n" " · {holder_line}\n\n" "设备已不属于公司资产,继续纳管会让 {provider} 的设备数和许可" "用量都偏高。处理完成后卡片可一键确认;下一次同步查不到该设备" "也会自动结案。\n\n" "— Helix IT 平台" ), "body_text_en": ( "Hi,\n\n" "This cleanup has been open for {days_open} days and the most " "recent sync still finds the device in {provider}, so it has " "not been removed from the console yet:\n\n" " · Device: {device_name} ({device_model})\n" " · Serial: {serial_number}\n" " · Asset tag: {asset_tag}\n" " · Disposal: {disposal_label}\n" " · Disposed on: {disposed_on}\n" " · {holder_line}\n\n" "The company no longer owns it, and leaving it enrolled " "overstates both the device count and the licence usage. " "Confirm on the card once it is done; the next sync closes it " "automatically if the device is gone.\n\n" "— Helix IT Platform" ), "body_text_ja": ( "お世話になっております。\n\n" "この整理依頼は {days_open} 日が経過していますが、直近の同期でも " "{provider} 上に該当デバイスが存在しており、コンソールからまだ" "削除されていません。\n\n" " · デバイス: {device_name}({device_model})\n" " · シリアル番号: {serial_number}\n" " · 資産番号: {asset_tag}\n" " · 処分区分: {disposal_label}\n" " · 処分日: {disposed_on}\n" " · {holder_line}\n\n" "すでに会社資産ではないため、管理対象のままではデバイス数も" "ライセンス消費も過大に計上されます。対応後はカードからワンクリック" "で確認できます。次回同期で見つからなくなれば自動的にクローズ" "されます。\n\n" "— Helix IT プラットフォーム" ), "card_button_label": "✅ 我已移除,标记完成", "card_button_label_en": "✅ Removed — mark done", "card_button_label_ja": "✅ 削除しました。完了にする", }, # Provisioning manual-handle DM (round-14). Replaces the inline # body that hardcoded ``{portal_url}/admin/provisioning`` into the # message text - same localhost-leak class as leaver_it_grace. "provisioning.manual_notify": { "subject": "[Helix] 请手工{action_label_zh} {software_name} 坐席: {actor_name}", "subject_en": ( "[Helix] Please manually {action_label_en} {software_name} " "seat for {actor_name}" ), "subject_ja": ( "[Helix] {actor_name} さんの {software_name} ライセンスを" "手動で{action_label_ja}してください" ), "body_text": ( "Hi,\n\n" "员工 {actor_name} ({actor_email}) 的 {software_name}" "{plan_suffix} 坐席需要您手动{action_label_zh}:\n\n" " · 操作: 在外部系统中{action_label_zh}该用户的账号\n" " · 任务号: {task_id}\n" " · 完成后: 飞书卡片可一键确认(无需登录)," "或在 Helix 控制台标记完成;坐席台账会随之关闭。\n\n" "{cta_line}\n\n" "— Helix IT 平台" ), "body_text_en": ( "Hi,\n\n" "{actor_name} ({actor_email}) has a {software_name}" "{plan_suffix} seat that needs you to manually " "{action_label_en}:\n\n" " · Action: {action_label_en} the user account in the vendor " "system\n" " · Task ID: {task_id}\n" " · After: one-click confirm on the Feishu card (no " "login), or mark done in the Helix console; the seat " "ledger closes automatically.\n\n" "{cta_line}\n\n" "— Helix IT Platform" ), "body_text_ja": ( "お世話になっております。\n\n" "{actor_name} さん({actor_email})の {software_name}" "{plan_suffix} ライセンスについて、手動での{action_label_ja}を" "お願いします。\n\n" " · 作業内容: 外部システム上で該当ユーザーのアカウントを" "{action_label_ja}する\n" " · タスク番号: {task_id}\n" " · 完了後: Feishu カードからワンクリックで確認できます(ログイン不要)。" "または Helix コンソールで完了としてマークしてください。" "ライセンス台帳は自動的にクローズされます。\n\n" "{cta_line}\n\n" "— Helix IT プラットフォーム" ), "card_button_label": "✅ 我已处理,标记完成", "card_button_label_en": "✅ Mark done", "card_button_label_ja": "✅ 対応済みとしてマーク", }, # Offboarding auto-release reminder (2026-06-13). Fires when an # offboarding-sourced seat / permission revoke resolves to # ``manual_notify``: Helix has ALREADY reclaimed the local # allocation and resolved the blocker (so the leaver is never held # in the red "需要你处理" queue), and this DM only reminds IT / the # software owner to finish the vendor-side cleanup. NO one-click # confirm button — the ledger is already closed; confirming would be # a misleading no-op. Distinct from ``provisioning.manual_notify`` # (which DOES gate on a human confirm for ad-hoc/mid-employment # revokes). "provisioning.offboarding_auto_release_reminder": { "subject": ( "[Helix] 离职回收提醒: 请在外部系统{action_label_zh} " "{software_name} 账号 — {actor_name}" ), "subject_en": ( "[Helix] Offboarding reminder: please {action_label_en} the " "{software_name} account in the vendor system — {actor_name}" ), "subject_ja": ( "[Helix] 退職に伴う対応のお願い: 外部システムで {software_name} " "アカウントを{action_label_ja}してください — {actor_name}" ), "body_text": ( "Hi,\n\n" "离职员工 {actor_name} ({actor_email}) 的 {software_name}" "{plan_suffix} 坐席,Helix 已自动回收(台账已关闭," "无需在 Helix 确认)。\n\n" "请完成外部系统侧的收尾:\n" " · 操作: 在外部系统中{action_label_zh}该用户的账号\n" " · 任务号: {task_id}\n" " · 说明: 本提醒不阻塞离职流程,仅用于触发供应商侧清理。\n\n" "{cta_line}\n\n" "— Helix IT 平台" ), "body_text_en": ( "Hi,\n\n" "Helix has already reclaimed {actor_name} " "({actor_email})'s {software_name}{plan_suffix} seat (the " "ledger is closed — no confirmation needed in Helix).\n\n" "Please finish the vendor-side cleanup:\n" " · Action: {action_label_en} the user's account in the vendor " "system\n" " · Task ID: {task_id}\n" " · Note: this reminder does NOT block the offboarding; it " "only prompts the vendor-side teardown.\n\n" "{cta_line}\n\n" "— Helix IT Platform" ), "body_text_ja": ( "お世話になっております。\n\n" "退職者 {actor_name} さん({actor_email})の {software_name}" "{plan_suffix} ライセンスは、Helix 側で自動的に回収済みです" "(台帳はクローズ済みのため、Helix での確認操作は不要です)。\n\n" "外部システム側の後処理をお願いします。\n" " · 作業内容: 外部システム上で該当ユーザーのアカウントを" "{action_label_ja}する\n" " · タスク番号: {task_id}\n" " · 補足: 本通知は退職プロセスをブロックするものではなく、" "ベンダー側の後処理を促すためのご案内です。\n\n" "{cta_line}\n\n" "— Helix IT プラットフォーム" ), "card_button_label": "打开任务详情", "card_button_label_en": "Open task", "card_button_label_ja": "タスク詳細を開く", }, # Revoke-task failure DM (round 2026-05-15). Fires from # ``ProvisioningService._dispatch_revoke_failure_dms`` after a seat # revoke task transitions to ``failed`` (dead-letter or vendor # rejection). Bilingual subject + body + button so the bell card # uses "前往取消中心" / "Open cancellations" instead of the generic # "Open in Helix" fallback that fires when no template registers. "provisioning.revoke_failure": { "subject": "[Helix] 软件坐席回收失败: {actor_name} · {software_name}", "subject_en": ( "[Helix] Software seat revoke failed: " "{actor_name} · {software_name}" ), "subject_ja": ( "[Helix] ソフトウェアライセンスの回収に失敗しました: " "{actor_name} · {software_name}" ), "body_text": ( "Hi,\n\n" "{actor_name} ({actor_email}) 的 {software_name} 坐席" "自动回收任务执行失败。\n\n" " · 错误: {error_brief}\n" " · 任务号: {task_id}\n" " · 当前状态: 坐席已自动回滚为 active,等待管理员介入\n\n" "{cta_line}\n\n" "— Helix IT 平台" ), "body_text_en": ( "Hi,\n\n" "Auto-revoke task for {actor_name} ({actor_email})'s " "{software_name} seat failed.\n\n" " · Error: {error_brief}\n" " · Task ID: {task_id}\n" " · Current state: seat reverted to active; needs admin " "intervention\n\n" "{cta_line}\n\n" "— Helix IT Platform" ), "body_text_ja": ( "お世話になっております。\n\n" "{actor_name} さん({actor_email})の {software_name} ライセンスの" "自動回収タスクが失敗しました。\n\n" " · エラー: {error_brief}\n" " · タスク番号: {task_id}\n" " · 現在の状態: ライセンスは active に自動ロールバック済みで、" "管理者による対応をお待ちしています\n\n" "{cta_line}\n\n" "— Helix IT プラットフォーム" ), "card_button_label": "前往取消中心", "card_button_label_en": "Open cancellations", "card_button_label_ja": "解約センターを開く", }, # Zoom paid-seat eviction (round-14). Same fix - replaces the # inline ``{portal_url}/my/requests`` URL in the body with a # template + button. "zoom.seat_evicted": { "subject": "[Helix] 您的 Zoom 付费坐席已被回收", "subject_en": "[Helix] Your Zoom paid seat has been reclaimed", "subject_ja": "[Helix] Zoom の有償ライセンスを回収しました", "body_text": ( "Hi {person_name},\n\n" "由于团队 Zoom 付费坐席(License)已达上限,而您是当前所有" "付费用户中最久未登录的一位(最近登录: {last_login_time})," "系统已将您的 {plan_name} 坐席回收以分配给新申请的同事。\n\n" " · 您仍然可以使用 Zoom Basic (免费档) 进行最长 40 分钟的会议;\n" " · 如确有常规工作需要 Licensed 坐席,请重新提交申请,系统会在下次有空位或再次触发回收时为您分配;\n" " · 会议历史、云录制、个人设置不会被删除。\n\n" "{cta_line}\n\n" "— Helix IT 平台" ), "body_text_en": ( "Hi {person_name},\n\n" "Your team has reached the Zoom paid-seat (License) cap. As " "the longest-idle paid user (last sign-in: {last_login_time}), " "the system reclaimed your {plan_name} seat to allocate it " "to a colleague who just requested one.\n\n" " · You can still use Zoom Basic (free, 40-minute meeting " "cap);\n" " · If you actually need a Licensed seat for regular work, " "please re-submit your request - the system will allocate " "as soon as a slot opens up;\n" " · Meeting history, cloud recordings, and personal " "settings are kept.\n\n" "{cta_line}\n\n" "— Helix IT Platform" ), "body_text_ja": ( "{person_name} さん\n\n" "チームの Zoom 有償ライセンス(License)が上限に達したため、" "現在の有償ユーザーの中で最も長くログインのなかった方(最終ログイン: " "{last_login_time})として、あなたの {plan_name} ライセンスを回収し、" "新たに申請した同僚へ割り当てました。\n\n" " · Zoom Basic(無償プラン)は引き続きご利用いただけます(会議は最長 40 分)。\n" " · 業務上 Licensed ライセンスが必要な場合は、あらためて申請してください。" "空きが出た時点、または次回の回収時に自動で割り当てられます。\n" " · 会議履歴、クラウド録画、個人設定は削除されません。\n\n" "{cta_line}\n\n" "— Helix IT プラットフォーム" ), "card_button_label": "重新申请坐席", "card_button_label_en": "Re-request seat", "card_button_label_ja": "ライセンスを再申請する", }, # Round-15 (2026-05-09) — IT-admin push when a scheduled job lands # status='failed'. Dedup-aware caller in # ``app.scheduled_jobs.failure_notifier`` decides whether to fire; # this template just renders bilingual title + body. ``url=`` is the # action button to /admin/scheduled-jobs/runs filtered by handler_key. "scheduled_job.failure": { "subject": "[定时任务失败] {job_display_name}", "subject_en": "[Scheduled job failed] {job_display_name}", "subject_ja": "[定期ジョブ失敗] {job_display_name}", "body_text": ( "{job_display_name}({handler_key})执行失败。\n\n" "错误:{error_brief}\n\n" "{cta_line}\n\n" "— Helix IT 平台" ), "body_text_en": ( "Scheduled job {job_display_name} ({handler_key}) failed.\n\n" "Error: {error_brief}\n\n" "{cta_line}\n\n" "— Helix IT Platform" ), "body_text_ja": ( "定期ジョブ {job_display_name}({handler_key})の実行に失敗しました。\n\n" "エラー:{error_brief}\n\n" "{cta_line}\n\n" "— Helix IT プラットフォーム" ), "card_button_label": "查看详情并确认", "card_button_label_en": "View and acknowledge", "card_button_label_ja": "詳細を確認して承認する", }, "identity_sync.run_failure": { "subject": "[数据同步失败] {target_name}", "subject_en": "[Identity sync failed] {target_name}", "subject_ja": "[データ同期失敗] {target_name}", # ``{status_label_*}`` is a locale-fragment triple (alert_notifier # builds zh / en / ja): "失败" vs "部分失败" is the difference between a # dead target and a partial run, so every language carries it. The # English body used to drop the slot entirely. "body_text": ( "数据同步目标『{target_name}』({kind})一次同步{status_label_zh}。\n\n" "失败 {failed_count} 人 / 范围 {in_scope} 人,触发源 {trigger_source}。\n" "错误:{error_brief}\n\n" "{cta_line}\n\n" "— Helix IT 平台" ), "body_text_en": ( "Identity-sync target '{target_name}' ({kind}) " "{status_label_en} on its latest run.\n\n" "{failed_count} failed / {in_scope} in scope, " "trigger {trigger_source}.\n" "Error: {error_brief}\n\n" "{cta_line}\n\n" "— Helix IT Platform" ), "body_text_ja": ( "データ同期ターゲット「{target_name}」({kind})の直近の実行が" "{status_label_ja}しました。\n\n" "失敗 {failed_count} 名 / 対象 {in_scope} 名、実行契機 {trigger_source}。\n" "エラー:{error_brief}\n\n" "{cta_line}\n\n" "— Helix IT プラットフォーム" ), "card_button_label": "查看运行历史", "card_button_label_en": "View run history", "card_button_label_ja": "実行履歴を見る", }, "asset_inventory_notify": { "subject": "[资产盘点] 『{cycle_name}』需要你确认", "subject_en": "[Asset inventory] {cycle_name} needs your confirmation", "subject_ja": "[資産棚卸]「{cycle_name}」のご確認をお願いします", "body_text": ( "Hi {person_name},\n\n" "资产盘点周期『{cycle_name}』已开始,需要你在 Helix 上" "逐项确认在用状态或反馈问题:\n\n" " · 个人持有: [b]{personal_count}[/b] 件\n" " · 办公室待盘: [b]{office_count}[/b] 件\n" " · 截止日期: [r]{due_date}[/r]\n\n" "{cta_line}\n\n" "如有疑问,请联系 IT 团队。\n\n" "— Helix IT 平台" ), "body_text_en": ( "Hi {person_name},\n\n" "Asset inventory round '{cycle_name}' has started. " "Please confirm each item or report any issue in Helix:\n\n" " · Personally held: [b]{personal_count}[/b]\n" " · Office stock to check: [b]{office_count}[/b]\n" " · Due date: [r]{due_date}[/r]\n\n" "{cta_line}\n\n" "Questions? Contact IT.\n\n" "— Helix IT Platform" ), "body_text_ja": ( "{person_name} さん\n\n" "資産棚卸「{cycle_name}」が開始されました。Helix 上で各資産の" "利用状況をご確認いただくか、問題があればご報告をお願いします。\n\n" " · 個人保有: [b]{personal_count}[/b] 点\n" " · オフィス棚卸対象: [b]{office_count}[/b] 点\n" " · 期限: [r]{due_date}[/r]\n\n" "{cta_line}\n\n" "ご不明な点は IT チームまでお問い合わせください。\n\n" "— Helix IT プラットフォーム" ), "card_button_label": "去 Helix 确认", "card_button_label_en": "Confirm in Helix", "card_button_label_ja": "Helix で確認する", }, # Cloud billing alert — daily / monthly threshold breach. Money # fields wrapped in ``[r]`` so they render red+bold on the Feishu # card and bold in email. ``{cta_line}`` collapses on Feishu (the # action button covers the CTA) and prints the URL inline for # email. Added 2026-05-02 — previously this dispatch was inline # in lifecycle_service.py with no template_key, no button, and # no money emphasis. "cloud.billing_alert": { # 2026-05-03: include the cloud provider so recipients can # tell which cloud the alert is for at a glance — the prior # version showed only ``Scope: tidb-o11y`` which is ambiguous # (multiple clouds may have a similarly-named scope). The # provider is rendered into the subject prefix and as its own # line in the body. Money fields use ``[r]…[/r]``; the title # path strips the markers (Feishu header is plain_text-only) # while the cloud.billing_alert card is rendered with a red # header colour to convey urgency. "subject": ( "[云账单·{provider_label_zh}] {scope_label} 超出{period_zh}阈值 " "[r]${overage_str}[/r]" ), "subject_en": ( "[Cloud bill · {provider_label_en}] {scope_label} exceeded " "{period_en} threshold by [r]${overage_str}[/r]" ), "subject_ja": ( "[クラウド請求・{provider_label_ja}] {scope_label} が" "{period_ja}のしきい値を [r]${overage_str}[/r] 超過しました" ), "body_text": ( "💰 云账单告警 ({rule_type})\n" "云厂商: {provider_label_zh}\n" "Scope: {scope_label}\n" "{period_label_zh}: [r]${current_str}[/r]\n" "阈值: ${threshold_str}\n" "超出: [r]${overage_str}[/r]\n\n" "{cta_line}\n\n" "— Helix IT 平台" ), "body_text_en": ( "💰 Cloud bill alert ({rule_type})\n" "Cloud provider: {provider_label_en}\n" "Scope: {scope_label}\n" "{period_label_en}: [r]${current_str}[/r]\n" "Threshold: ${threshold_str}\n" "Over by: [r]${overage_str}[/r]\n\n" "{cta_line}\n\n" "— Helix IT Platform" ), "body_text_ja": ( "💰 クラウド請求アラート ({rule_type})\n" "クラウド事業者: {provider_label_ja}\n" "Scope: {scope_label}\n" "{period_label_ja}: [r]${current_str}[/r]\n" "しきい値: ${threshold_str}\n" "超過額: [r]${overage_str}[/r]\n\n" "{cta_line}\n\n" "— Helix IT プラットフォーム" ), "card_button_label": "查看账单详情", "card_button_label_en": "View billing detail", "card_button_label_ja": "請求の詳細を見る", }, # Sent by ``apikey.budget_check`` cron when an active # ApiKeyAssignment crosses 80% / 100% of its self-declared # ``monthly_budget_usd``. Routed to BOTH the applicant AND their # direct manager (when present + active). Money fields use # ``[r]…[/r]`` and the card renders in the alert state — red header, # 告警 badge (see ``card_design.STATUS_BY_TEMPLATE``). "apikey.budget_alert": { "subject": ( "[Helix] API Key 月度预算 {threshold_pct}% — {provider_name} · " "{applicant_name} 已用 [r]${mtd_usd}[/r]" ), "subject_en": ( "[Helix] API key monthly budget {threshold_pct}% — " "{provider_name} · {applicant_name} spent [r]${mtd_usd}[/r]" ), "subject_ja": ( "[Helix] API キー月次予算 {threshold_pct}% — {provider_name} · " "{applicant_name} さんの利用額 [r]${mtd_usd}[/r]" ), "body_text": ( "💰 API Key 月度预算告警\n" "申请人: {applicant_name}\n" "申请号: {request_number}\n" "提供方: {provider_name}\n" "项目: {project_name}\n" "月度预算: ${monthly_budget_usd} USD\n" "本月已花费: [r]${mtd_usd}[/r] USD ({pct_used}%)\n" "触发阈值: {threshold_pct}%\n\n" "{cta_line}\n\n" "— Helix IT 平台" ), "body_text_en": ( "💰 API key monthly-budget alert\n" "Applicant: {applicant_name}\n" "Request: {request_number}\n" "Provider: {provider_name}\n" "Project: {project_name}\n" "Monthly budget: ${monthly_budget_usd} USD\n" "Spent so far: [r]${mtd_usd}[/r] USD ({pct_used}%)\n" "Threshold crossed: {threshold_pct}%\n\n" "{cta_line}\n\n" "— Helix IT Platform" ), "body_text_ja": ( "💰 API キー月次予算アラート\n" "申請者: {applicant_name}\n" "申請番号: {request_number}\n" "プロバイダー: {provider_name}\n" "プロジェクト: {project_name}\n" "月次予算: ${monthly_budget_usd} USD\n" "今月の利用額: [r]${mtd_usd}[/r] USD ({pct_used}%)\n" "到達したしきい値: {threshold_pct}%\n\n" "{cta_line}\n\n" "— Helix IT プラットフォーム" ), "card_button_label": "查看 / 调整预算", "card_button_label_en": "Review / adjust budget", "card_button_label_ja": "予算を確認 / 調整する", }, # Sent to global_admin holders by the SSO grant approval executor # when an employee files a request for a NEW SAML app that isn't # in the catalog yet. Helix creates the local catalog row but # cannot create the app in Google admin (no public API), so this # template walks IT through the manual steps with all the SAML # metadata pre-filled. Added 2026-05-07 with the SSO Apps catalog # pivot — see [`server/app/domains/sso/app_catalog_service.py`]. "sso.app_create_pending": { "subject": "[SSO] 请在 Google admin 创建 SAML 应用: {app_name}", "subject_en": "[SSO] Please create SAML app '{app_name}' in Google admin", "subject_ja": "[SSO] Google 管理コンソールで SAML アプリ「{app_name}」の作成をお願いします", "body_text": ( "🆔 新 SAML 应用待创建\n" "应用名: {app_name}\n" "申请人: {requester_label}\n" "归属负责人: {owner_label}\n" "申请原因: {reason}\n\n" "SAML 配置:\n" " ACS URL: {acs_url}\n" " Entity ID: {entity_id}\n\n" "请在 Google admin 控制台 → Apps → Web and mobile apps → " "Add custom SAML app 中创建该应用,使用以上 ACS URL 与 " "Entity ID。完成后回到 Helix 该应用行,把 Google 那边的 " "external_id 回填进去(可选)。\n\n" "{cta_line}\n\n" "— Helix IT 平台" ), "body_text_en": ( "🆔 New SAML app pending creation\n" "App: {app_name}\n" "Requester: {requester_label}\n" "Owner: {owner_label}\n" "Reason: {reason}\n\n" "SAML config:\n" " ACS URL: {acs_url}\n" " Entity ID: {entity_id}\n\n" "Please go to Google admin → Apps → Web and mobile apps → " "Add custom SAML app and create the app using the ACS URL " "and Entity ID above. Once done, return to the Helix row " "and back-fill the Google external_id (optional).\n\n" "{cta_line}\n\n" "— Helix IT Platform" ), "body_text_ja": ( "🆔 新規 SAML アプリの作成依頼\n" "アプリ名: {app_name}\n" "申請者: {requester_label}\n" "管理責任者: {owner_label}\n" "申請理由: {reason}\n\n" "SAML 設定:\n" " ACS URL: {acs_url}\n" " Entity ID: {entity_id}\n\n" "Google 管理コンソール → Apps → Web and mobile apps → " "Add custom SAML app から、上記の ACS URL と Entity ID を使用して" "アプリを作成してください。作成後、Helix の当該アプリ行に Google 側の " "external_id を登録してください(任意)。\n\n" "{cta_line}\n\n" "— Helix IT プラットフォーム" ), "card_button_label": "查看应用详情", "card_button_label_en": "View app detail", "card_button_label_ja": "アプリ詳細を見る", }, # The return leg of the SSO grant flow: Google is the IdP, so IT has # to hand ITS generated values back to the requester before the SP # side can be configured at all. The certificate is inlined verbatim # on purpose — it is a public signing certificate and the requester # must paste it somewhere, so a fingerprint would only force another # round trip. # # Recipient is an EMPLOYEE, which the "new pushes are IT-only" rule # normally forbids — allowed here because the user asked for exactly # this handoff, and because nothing fires it automatically: there is # no cron and no state machine behind it, only an IT admin pressing # "deliver" on one app. Existing rows cannot batch-trigger on deploy. "sso.idp_config_ready": { "subject": "[SSO] {system_name} 的 Google SSO 配置信息已就绪", "subject_en": "[SSO] Google SSO config for {system_name} is ready", "subject_ja": "[SSO] {system_name} の Google SSO 設定情報が準備できました", "body_text": ( "你申请的 {system_name} 已在 Google Workspace 完成 SAML 应用创建。\n\n" "请把下面三项配置到 {system_name} 那一侧(或转交给厂商):\n\n" " SSO URL / 登录 URL:\n {idp_sso_url}\n\n" " IdP Entity ID:\n {idp_entity_id}\n\n" " X.509 证书:\n{idp_certificate}\n\n" "配置完成后即可用公司 Google 账号登录该系统。如果对方要求的字段名" "不一样,常见别名:SSO URL = Login URL / IdP SSO URL;" "Entity ID = Issuer / IdP Issuer URL。\n\n" "配置过程中有问题,直接回复本通知或找 IT。\n\n" "{cta_line}\n\n" "— Helix IT 平台" ), "body_text_en": ( "The SAML app for {system_name} has been created in Google " "Workspace.\n\n" "Configure these three values on the {system_name} side (or " "pass them to the vendor):\n\n" " SSO URL / Login URL:\n {idp_sso_url}\n\n" " IdP Entity ID:\n {idp_entity_id}\n\n" " X.509 certificate:\n{idp_certificate}\n\n" "Once configured you can sign in with your company Google " "account. If their field names differ, common aliases: " "SSO URL = Login URL / IdP SSO URL; Entity ID = Issuer / " "IdP Issuer URL.\n\n" "Reply to this notification or ping IT if you hit trouble.\n\n" "{cta_line}\n\n" "— Helix IT Platform" ), "body_text_ja": ( "申請いただいた {system_name} の SAML アプリを Google Workspace 上に" "作成しました。\n\n" "以下の 3 点を {system_name} 側(またはベンダー)に設定してください:\n\n" " SSO URL / ログイン URL:\n {idp_sso_url}\n\n" " IdP Entity ID:\n {idp_entity_id}\n\n" " X.509 証明書:\n{idp_certificate}\n\n" "設定が完了すると、会社の Google アカウントでログインできます。" "項目名が異なる場合の一般的な別名: SSO URL = Login URL / " "IdP SSO URL、Entity ID = Issuer / IdP Issuer URL。\n\n" "ご不明な点は本通知に返信するか IT までご連絡ください。\n\n" "{cta_line}\n\n" "— Helix IT プラットフォーム" ), "card_button_label": "查看我的申请", "card_button_label_en": "View my requests", "card_button_label_ja": "申請一覧を見る", }, "asset_inventory_reminder": { "subject": "[提醒] 资产盘点『{cycle_name}』仍有待处理项", "subject_en": "[Reminder] Asset inventory '{cycle_name}' still has pending items", "subject_ja": "[リマインド] 資産棚卸「{cycle_name}」に未対応の項目が残っています", "body_text": ( "Hi {person_name},\n\n" "资产盘点周期『{cycle_name}』已开始 7 天,你仍有" "[r]{personal_count}[/r] 件个人持有 + [r]{office_count}[/r] 件办公室" "待盘资产未确认。\n\n" "{cta_line}\n\n" "— Helix IT 平台" ), "body_text_en": ( "Hi {person_name},\n\n" "Asset inventory round '{cycle_name}' has been open for " "7 days. [r]{personal_count}[/r] personal item(s) and " "[r]{office_count}[/r] office item(s) still need your response.\n\n" "{cta_line}\n\n" "— Helix IT Platform" ), "body_text_ja": ( "{person_name} さん\n\n" "資産棚卸「{cycle_name}」の開始から 7 日が経過しました。" "個人保有 [r]{personal_count}[/r] 点、オフィス棚卸対象 " "[r]{office_count}[/r] 点が未確認のままです。\n\n" "{cta_line}\n\n" "— Helix IT プラットフォーム" ), "card_button_label": "去 Helix 确认", "card_button_label_en": "Confirm in Helix", "card_button_label_ja": "Helix で確認する", }, # Manager / HRBP nudge digest, sent only when an admin opts in on a # manual inventory remind ("也通知直线 Lead / HRBP"). ONE aggregated # card per Lead / HRBP listing their unresponsive people — the # ``{pending_names}`` bullet block is pre-formatted by the caller. # ``[r]…[/r]`` reddens the count on the Feishu card / bolds it in email. "asset_inventory_escalation": { "subject": "[协助提醒] 资产盘点『{cycle_name}』有 [r]{pending_count}[/r] 位同事待反馈", "subject_en": "[Please nudge] {pending_count} colleague(s) pending in asset inventory '{cycle_name}'", "subject_ja": "[ご協力のお願い] 資産棚卸「{cycle_name}」で [r]{pending_count}[/r] 名の未回答があります", "body_text": ( "你好 {person_name},\n\n" "资产盘点周期『{cycle_name}』进行中,以下与你相关的 " "[r]{pending_count}[/r] 位同事仍未反馈其资产情况:\n\n" "{pending_names}\n\n" "烦请协助提醒他们尽快在 Helix 完成确认。\n\n" "{cta_line}\n\n" "— Helix IT 平台" ), "body_text_en": ( "Hi {person_name},\n\n" "Asset inventory round '{cycle_name}' is in progress. " "[r]{pending_count}[/r] colleague(s) related to you have not " "responded yet:\n\n" "{pending_names}\n\n" "Please remind them to confirm their assets in Helix.\n\n" "{cta_line}\n\n" "— Helix IT Platform" ), "body_text_ja": ( "{person_name} さん\n\n" "資産棚卸「{cycle_name}」が進行中です。あなたに関係する以下の " "[r]{pending_count}[/r] 名の方から、資産に関する回答がまだ届いていません。\n\n" "{pending_names}\n\n" "Helix での確認を完了するよう、お声がけのご協力をお願いします。\n\n" "{cta_line}\n\n" "— Helix IT プラットフォーム" ), "card_button_label": "打开 Helix", "card_button_label_en": "Open Helix", "card_button_label_ja": "Helix を開く", }, # Leaver-Blocked SLA escalation templates (PR 2026-05-02). Three # severity tiers fire on T+24h / T+48h / T+72h after first_blocked_at; # at T+48h dispatch is targeted at the leaver's HRBP (resolved from # Feishu custom_attrs) plus the it_admin pool — not the entire HR # role. ``[r]…[/r]`` wraps the time-elapsed marker so it renders red # on the Feishu card and bold in email; the rest of the body is # plain prose so admins can scan a flight of escalations quickly. "leaver_blocked_sla_warn": { "subject": ( "[警告] {leaver_name} 离职流程已卡住 [r]{level_label_zh}[/r]" ), "subject_en": ( "[Warning] {leaver_name} offboarding blocked for [r]{level_label_en}[/r]" ), "subject_ja": ( "[警告] {leaver_name} さんの退職プロセスが [r]{level_label_ja}[/r] 停滞しています" ), "body_text": ( "员工 [b]{leaver_name}[/b] ({leaver_email}) 离职流程已卡住超过 " "[r]{level_label_zh}[/r]。请尽快进入 Helix 处理未清资源,避免" "升级到紧急 / 严重档位。\n\n" "仍需处理的资源:\n{pending_resources}\n\n" "{cta_line}\n\n" "未清资源会持续阻断离职完成,影响合规。\n\n" "— Helix IT 平台" ), "body_text_en": ( "{leaver_name} ({leaver_email}) offboarding has been blocked " "for [r]{level_label_en}[/r]. Please clear outstanding " "resources in Helix before this escalates to URGENT / BREACH.\n\n" "Outstanding resources:\n{pending_resources}\n\n" "{cta_line}\n\n" "Unresolved resources continue to block leaver completion " "and create a compliance gap.\n\n" "— Helix IT Platform" ), "body_text_ja": ( "[b]{leaver_name}[/b] さん({leaver_email})の退職プロセスが " "[r]{level_label_ja}[/r] にわたり停滞しています。緊急 / 重大の段階に" "上がる前に、Helix で未処理のリソースをご対応ください。\n\n" "未処理のリソース:\n{pending_resources}\n\n" "{cta_line}\n\n" "未処理のリソースは退職完了を継続的に阻害し、" "コンプライアンス上のリスクとなります。\n\n" "— Helix IT プラットフォーム" ), "card_button_label": "去 Helix 处理", "card_button_label_en": "Handle in Helix", "card_button_label_ja": "Helix で対応する", }, "leaver_blocked_sla_urgent": { # The elapsed-time marker rides in the subject of all three tiers # (warn / urgent / breach) and in all three languages — the zh # subject used to hardcode "SLA urgent" and drop the marker, so a # Chinese reader could not tell 48h from 72h without opening the body. "subject": ( "[紧急] {leaver_name} 离职流程 SLA 已卡住 " "([r]{level_label_zh}[/r]) — 仍有未清资源" ), "subject_en": ( "[URGENT] {leaver_name} offboarding SLA breached " "([r]{level_label_en}[/r]) — resources still pending" ), "subject_ja": ( "[緊急] {leaver_name} さんの退職プロセスが SLA 緊急段階 " "([r]{level_label_ja}[/r]) — 未処理のリソースが残っています" ), "body_text": ( "员工 [b]{leaver_name}[/b] ({leaver_email}) 离职流程已卡住超过 " "[r]{level_label_zh}[/r],处于 [r]紧急[/r] SLA 档位。\n\n" "你收到这条通知是因为你是该员工的 HRBP 或 IT 管理员。请立即" "进入 Helix 处理未清资源 —— 再过 24 小时将升级为严重档位" "(自动开 critical incident)。\n\n" "仍需处理的资源:\n{pending_resources}\n\n" "{cta_line}\n\n" "— Helix IT 平台" ), "body_text_en": ( "{leaver_name} ({leaver_email}) offboarding has been blocked " "for [r]{level_label_en}[/r] and is now at [r]URGENT[/r] SLA. " "You're receiving this because you are the leaver's HRBP or " "an IT admin. Please clear the outstanding resources in Helix " "now — another 24h escalates to BREACH (auto-creates a critical " "incident).\n\n" "Outstanding resources:\n{pending_resources}\n\n" "{cta_line}\n\n" "— Helix IT Platform" ), "body_text_ja": ( "[b]{leaver_name}[/b] さん({leaver_email})の退職プロセスが " "[r]{level_label_ja}[/r] にわたり停滞しており、[r]緊急[/r] SLA 段階です。\n\n" "本通知は、あなたが当該社員の HRBP または IT 管理者であるため送信されています。" "直ちに Helix で未処理のリソースをご対応ください。" "さらに 24 時間経過すると重大段階へ移行し、" "critical incident が自動起票されます。\n\n" "未処理のリソース:\n{pending_resources}\n\n" "{cta_line}\n\n" "— Helix IT プラットフォーム" ), "card_button_label": "立即处理", "card_button_label_en": "Handle now", "card_button_label_ja": "今すぐ対応する", }, "leaver_blocked_sla_breach": { "subject": ( "[严重] {leaver_name} 离职流程 SLA breach ([r]{level_label_zh}[/r])" ), "subject_en": ( "[BREACH] {leaver_name} offboarding SLA breach ([r]{level_label_en}[/r])" ), "subject_ja": ( "[重大] {leaver_name} さんの退職プロセスが SLA 違反 ([r]{level_label_ja}[/r])" ), "body_text": ( "员工 [b]{leaver_name}[/b] ({leaver_email}) 离职流程已卡住超过 " "[r]{level_label_zh}[/r],SLA 已 [r]突破[/r]。系统已自动开启" "一条 critical compliance incident。\n\n" "请立即处理 —— 这条通知发给该员工的 HRBP 与全部 IT 管理员。\n\n" "仍需处理的资源:\n{pending_resources}\n\n" "{cta_line}\n\n" "— Helix IT 平台" ), "body_text_en": ( "{leaver_name} ({leaver_email}) offboarding has been blocked " "for [r]{level_label_en}[/r] — SLA [r]breached[/r]. Helix has " "auto-opened a critical compliance incident.\n\n" "Act now — this notification was sent to the leaver's HRBP " "and every IT admin.\n\n" "Outstanding resources:\n{pending_resources}\n\n" "{cta_line}\n\n" "— Helix IT Platform" ), "body_text_ja": ( "[b]{leaver_name}[/b] さん({leaver_email})の退職プロセスが " "[r]{level_label_ja}[/r] にわたり停滞し、SLA を [r]違反[/r] しました。" "Helix は critical compliance incident を自動起票しました。\n\n" "直ちにご対応ください。本通知は当該社員の HRBP および" "全 IT 管理者に送信されています。\n\n" "未処理のリソース:\n{pending_resources}\n\n" "{cta_line}\n\n" "— Helix IT プラットフォーム" ), "card_button_label": "立即处理", "card_button_label_en": "Handle now", "card_button_label_ja": "今すぐ対応する", }, # Sent from ``apikey/{bedrock,bailian}_autogrant.py`` after a # successful auto-grant lands. Tells the assignee that the # credential is live, points them at the self-serve Reveal page, # and embeds a deep-link to the provider-specific usage manual so # the employee doesn't have to hunt for setup docs themselves. "apikey.ready": { "subject": "[Helix] {provider_name} API Key 已就绪", "subject_en": "[Helix] {provider_name} API key ready", "subject_ja": "[Helix] {provider_name} の API キーが利用可能になりました", "body_text": ( "Hi {person_name},\n\n" "你申请的 [b]{provider_name}[/b] API Key(申请号 " "[r]{request_number}[/r],项目「{project_name}」)已自动开通。\n\n" "请到 Helix 的「我的 API Keys」页面查看凭证 + 使用方法:\n" " {portal_url}\n\n" "详细使用手册见:{help_anchor}\n\n" "{cta_line}\n\n" "— Helix IT 平台" ), "body_text_en": ( "Hi {person_name},\n\n" "Your [b]{provider_name}[/b] API key (request " "[r]{request_number}[/r], project \"{project_name}\") is " "now active.\n\n" "View the credential and the SDK setup notes on your Helix " "\"My API Keys\" page:\n {portal_url}\n\n" "Provider-specific manual: {help_anchor}\n\n" "{cta_line}\n\n" "— Helix IT Platform" ), "body_text_ja": ( "{person_name} さん\n\n" "申請いただいた [b]{provider_name}[/b] の API キー(申請番号 " "[r]{request_number}[/r]、プロジェクト「{project_name}」)が自動で開通しました。\n\n" "認証情報と利用方法は、Helix の「マイ API キー」ページでご確認ください:\n" " {portal_url}\n\n" "詳細な利用マニュアルはこちら:{help_anchor}\n\n" "{cta_line}\n\n" "— Helix IT プラットフォーム" ), "card_button_label": "查看 API Key", "card_button_label_en": "View API key", "card_button_label_ja": "API キーを確認する", }, } # Domain-owned catalogs (``app.notifications.templates``) carry the rest of the # platform's template keys — declared there rather than inline here so each # domain's copy sits with its domain and this module stays navigable. Merged in # rather than replacing: a key defined both places keeps the literal above, and # the parity test asserts the two sets stay disjoint so nobody has to guess # which definition is live. _DEFAULT_TEMPLATES.update({ key: tmpl for key, tmpl in _DOMAIN_TEMPLATES.items() if key not in _DEFAULT_TEMPLATES }) # Context keys renamed by the 2026-08-09 localisation sweep, old → new. # # Outbox rows freeze their ``context`` at send time, and two surfaces re-render # from that frozen dict long afterwards: the retry worker, and the notification # centre (which re-renders any row whose stored body is the ``(see template)`` # sentinel). A rename therefore strands every historical row — the template now # asks for ``{action_label_zh}`` while the row only carries ``action_label``, # and the reader gets a literal brace-wrapped token. # # Rather than backfill the column, expand the aliases at render time: cheap, # reversible, and it keeps working for rows that outlive any one backfill. _LEGACY_CONTEXT_ALIASES: dict[str, str] = { "action_label": "action_label_zh", "action_en": "action_label_en", "status_label": "status_label_zh", "blessing": "blessing_zh", "blessing_en": "blessing_en", # 2026-08-17: the offboarding delete-gate alerts turned two single-language # context values into locale-fragment triples when their templates were # registered. Rows already sitting in the retry queue carry only the old # spelling, so without these the zh body of a retried row would render a # literal ``{reason_zh}``. Only the Chinese variant can be healed — the old # row never carried English or Japanese copy to begin with. "reason": "reason_zh", "account_list": "account_list_zh", } def _expand_legacy_context(context: dict[str, Any]) -> dict[str, Any]: """Fill renamed placeholder keys from their pre-rename spellings. Only fills keys the context does NOT already carry, so a current row (which has the new names) is untouched. """ expanded = dict(context) for old, new in _LEGACY_CONTEXT_ALIASES.items(): if new not in expanded and old in expanded: expanded[new] = expanded[old] return expanded def _localized_context( context: dict[str, Any] | None, lang: str | None, ) -> dict[str, Any]: """The format context a recipient's copy is rendered against. Expands the legacy aliases, then resolves every VALUE for ``lang`` the way card chrome already is. A context value may be a :class:`LocalizedText` (or its serialized ``{"zh": …, "en": …, "ja": …}`` form, which is what comes back off the outbox row on a retry / bell re-render); anything else passes through untouched. Why this exists: the template catalog is fully localised and the recipient's language is resolved per send, but the placeholder VALUES the producer interpolates were single-language strings by construction. A production render of the 45-day outbox against an ``en`` reader showed the result — e.g. ``offboarding.process_started_leaver`` delivered an English card whose last line was the producer's literal ``请尽快登录 Helix 完成交接。``. Values that are genuinely one language (a person's name, a user-typed reply, a vendor's error string) stay plain ``str`` and are unaffected. """ expanded = _expand_legacy_context(context or {}) return {k: resolve_card_text(v, lang) for k, v in expanded.items()} def _jsonable_context(context: dict[str, Any] | None) -> dict[str, Any] | None: """``context`` in a form the outbox's JSON column accepts. Mirrors :func:`_jsonable_card_text` for the localizable values :func:`_localized_context` resolves: a bare ``LocalizedText`` passed straight to the INSERT fails with "not JSON serializable", which takes the whole notification down rather than degrading it. """ if not context: return None return {k: _jsonable_card_text(v) for k, v in context.items()} def _is_en(lang: str | None) -> bool: """Whether ``lang`` reads the English chrome (button labels, footers). Thin wrapper over :func:`app.core.locale.is_en` — kept because a dozen call sites in this module thread a language *string* rather than a person. Japanese maps to True: the surrounding chrome has no Japanese copy, so English is the correct degradation (:data:`~app.core.locale.FALLBACK_CHAIN`). Template *bodies* do get real Japanese when a ``_ja`` variant exists — that selection happens in :func:`_select_template_field`, not here. """ return locale_is_en(lang) def _select_template_field( tmpl: dict[str, str], field: str, lang: str | None, ) -> str | None: """Pick ``field``'s copy in the recipient's language. Walks the ``ja → en → zh`` degradation ladder, so a template that only ships Chinese + English still renders something sensible for a Japanese recipient instead of falling back to Chinese. Returns None if NO variant exists in any language — caller decides how to degrade (fall through to the envelope's own title/body, log + skip channel). Never raises. """ return pick_localized( tmpl, field, normalize_lang(lang) or DEFAULT_LANG, ) # 头色/徽章/图标现在由 ``card_design`` 的**状态**表统一给出(见 ``_card_status``)。 # 原来这里是一张按模板配颜色的表 —— 颜色和「这张卡是什么状态」是两个各自维护的 # 概念,于是徽章和图标无处挂靠。偏离状态默认色的那几个模板搬到了 # ``card_design.COLOR_BY_TEMPLATE``。 def _card_status(env: NotificationEnvelope) -> str: """The semantic state this card is in — see :mod:`.card_design`. Resolution order, most specific first: 1. What the producer said (``env.card_status``). A producer that knows the card is a receipt, or an alert, is authoritative. 2. The per-template map, for templates whose state is a property of the template itself (an SLA breach is always critical). 3. **The notification category's bucket.** An ACTION-bucket template is by definition "somebody is waiting for you to do something", which is exactly ``pending``. Deriving it here is what gives every existing actionable card its 待处理 badge without 46 producers each remembering to ask for one — and it keeps the badge honest, because the bucket is the same fact the red dot on the bell is already counting. 4. ``info`` — a card that merely tells you something. """ if env.card_status: return env.card_status key = env.template_key if key: mapped = card_design.STATUS_BY_TEMPLATE.get(key) if mapped: return mapped if categories.bucket_for(key) == categories.BUCKET_ACTION: return card_design.STATUS_PENDING return card_design.STATUS_INFO def _card_header_color(template_key: str | None, status: str | None = None) -> str: """The Feishu header colour: the status' own, unless the template overrides. The override list is short and every entry is deliberate (the feed's warm palette, the Day-1 card matching its email masthead) — see ``card_design.COLOR_BY_TEMPLATE``. """ if template_key: override = card_design.COLOR_BY_TEMPLATE.get(template_key) if override: return override return card_design.tone_for(status).color # Sentinel callers store in ``outbox.body_text`` when the real copy is # produced from a template at send time (Feishu / email). The in-app # bell reads the persisted ``body_text``, so it sees this sentinel until # ``NotificationService.render_inbox_body`` renders the template for the # read surface. Named so the producers (10+ call sites) and the inbox # renderer agree on one literal. _SEE_TEMPLATE_SENTINEL = "(see template)" #: A ``{placeholder}`` that survived rendering — i.e. the context could not #: satisfy it. Deliberately not matching ``{{`` escapes or bare braces in #: prose. _UNRESOLVED_PLACEHOLDER = re.compile(r"\{[a-zA-Z_][\w.\[\]]*\}") #: Does a string carry any inline-markup marker? Used by the stat grid to #: decide whether the value still needs an outer ``**…**`` — see the nesting #: note there. Mirrors the marker vocabulary of ``_apply_inline_markup``. _MARKUP_MARKER_RE = re.compile(r"\[/?[rbh]\]", re.IGNORECASE) def _strip_inline_markup(text: str) -> str: """Remove ``[r]``/``[b]``/``[h]`` wrappers without renaming the content. Used by surfaces that can't render rich text (Feishu card header title is plain_text-only; email subject lines render asterisks literally in most clients) — keep the meaningful tokens, drop the decoration tags. Without this strip, alert templates leak literal ``[r]$140.76[/r]`` into Feishu titles (regression flagged 2026-05-03). """ return ( text .replace("[r]", "").replace("[/r]", "") .replace("[b]", "").replace("[/b]", "") .replace("[h]", "").replace("[/h]", "") ) # Tone → leading emoji prefix on section titles. Lets a user who # skims past colour still see urgency at a glance. _TONE_PREFIX: dict[str, str] = { "info": "", "warn": "⚠️ ", "danger": "🚨 ", "muted": "", } # Tone → markdown body wrapper. ``warn`` / ``danger`` bodies pick up a # coloured prefix so the section actually stands out from the muted # body text above; ``info`` and ``muted`` stay plain. _TONE_BODY_PREFIX: dict[str, str] = { "info": "", "warn": "", "danger": "", "muted": "", } _TONE_BODY_SUFFIX: dict[str, str] = { "info": "", "warn": "", "danger": "", "muted": "", } def _format_trend(token: str | None, is_good: bool | None) -> str: """Colour a trend token (e.g. ``▲ 12%``) based on direction. ``is_good=None`` infers: ``▼`` = good (green), ``▲`` = bad (red). Callers whose semantics invert (member count dropping is bad, cost dropping is good) pass ``is_good`` explicitly. """ if not token: return "" if is_good is None: is_good = token.lstrip().startswith("▼") colour = "green" if is_good else "red" return f" {_escape_card_markdown(token)}" def _rendered(value: Any) -> str: """A card-chrome value AFTER per-recipient resolution. Everything the card BUILDER touches has already been through ``_resolve_card_chrome``, so it is a plain string by then — but the field types stay ``CardText``, because they describe what a PRODUCER may supply. The builder cannot prove the resolution happened, and should not assume it. A ``LocalizedText`` arriving here means some send path reached the builder without resolving. Degrade to the baseline language and log loudly, rather than let ``str()`` print ``LocalizedText(zh='…', en='…', ja='…')`` into somebody's Feishu card — silent, and extremely visible to the reader. """ if isinstance(value, LocalizedText): # Log that it happened, NOT what it said. This module's ``logger`` is # a plain ``logging.getLogger`` — it does not pass through # ``app.core.safe_logging``, so an ``extra`` here goes to the cloud log # collector unredacted. Card content routinely carries invoice-net # cloud amounts, employee names and leaver details, and the collector's # audience is far wider than any of those. The zh prefix below is # bounded and only exists to make the offending template findable. # Location only — never the value, not even a prefix. This module's # ``logger`` is a plain ``logging.getLogger``: it does not pass through # ``app.core.safe_logging``, so anything here reaches the cloud log # collector unredacted, and card chrome routinely carries invoice-net # cloud amounts, employee names and leaver details. A 16-character # prefix was tried and dropped: it still fit a full Chinese name. logger.error( "card chrome reached the renderer unresolved — a send path is " "bypassing _resolve_card_chrome (value withheld: unredacted sink)", ) return value.zh return "" if value is None else str(value) def _render_stat_grid(stats: list[CardStat]) -> dict[str, Any] | None: """Compose a 2-column ``column_set`` element from card stats. 2/4/6 stats → 1x2 / 2x2 / 3x2; odd counts pad the last row with a spacer. Each cell renders as: label (small grey) **VALUE** ▲% (coloured) hint (small grey) """ if not stats: return None columns: list[dict[str, Any]] = [] for stat in stats: label_md = _escape_card_markdown(_rendered(stat.label)) raw_value = _rendered(stat.value) or "—" value_md = _apply_inline_markup( _escape_card_markdown(raw_value), channel="feishu_card", ) trend_md = _format_trend(_rendered(stat.trend) or None, stat.trend_is_good) hint_md = "" if stat.hint: hint_md = "\n" + _apply_inline_markup( _escape_card_markdown(_rendered(stat.hint)), channel="feishu_card", ) # Bold the value ONLY when it carries no markup of its own. A value like # ``[h]4[/h] 天`` expands to ``**4** 天``; # wrapping that in another ``**…**`` produces NESTED bold, which Feishu's # markdown resolves by dropping BOTH — observed in production 2026-08-18: # the leave-balance card rendered "4 天" at body size, neither bold nor # enlarged, and the same latent bug had been silently killing every # ``[r]`` red-bold stat value since the stat grid shipped. value_block = ( value_md if _MARKUP_MARKER_RE.search(raw_value) else f"**{value_md}**" ) content = ( f"{label_md}\n" f"{value_block}{trend_md}{hint_md}" ) columns.append({ "tag": "column", "width": "weighted", "weight": 1, "vertical_align": "top", "elements": [{"tag": "markdown", "content": content}], }) # Pad to even columns so the grid stays balanced. if len(columns) % 2 == 1: columns.append({ "tag": "column", "width": "weighted", "weight": 1, "vertical_align": "top", "elements": [{"tag": "markdown", "content": " "}], }) # Split into rows of 2. rows: list[dict[str, Any]] = [] for i in range(0, len(columns), 2): rows.append({ "tag": "column_set", "flex_mode": "none", "background_style": "grey", "columns": columns[i : i + 2], }) if len(rows) == 1: return rows[0] # Wrap multi-row grid as a div-of-column_sets via a single # column_set with nested rows. Feishu supports stacked column_sets # at top level too — return as a "div" elements list marker via # a special structure; the renderer will splice. return {"_multi_column_set": rows} def _render_highlight_block( title: str | None, highlights: list[CardHighlight], ) -> str: """Build the markdown body for a highlights section. Lines look like: **数据安全**: 范时军、吴礼富 等 4 人命中数据安全规则 **高频操作**: 柳冬冬 同时命中高频操作模型下的 1 个规则 """ if not highlights: return "" lines: list[str] = [] if title: lines.append(f"**{_escape_card_markdown(_rendered(title))}**") for h in highlights: label = _escape_card_markdown(_rendered(h.label)) if isinstance(h.items, list): if not h.items: continue body = "、".join( _apply_inline_markup( _escape_card_markdown(_rendered(it)), channel="feishu_card", ) for it in h.items ) else: body = _apply_inline_markup( _escape_card_markdown(_rendered(h.items)), channel="feishu_card", ) lines.append(f"{label}:{body}") return "\n".join(lines) def _render_sections(sections: list[CardSection]) -> list[dict[str, Any]]: """Build per-section elements separated by dividers. Tones ``info`` / ``warn`` / ``danger`` / ``muted`` render as full- weight markdown elements (same font size as the body). Tone ``hint`` renders as a Feishu ``note`` element — the same small- grey treatment as the card footer. Use ``hint`` for "this is a side note / not the main point" copy (e.g. "what to expect today" under a celebration card) so the section visually demotes itself relative to the body markdown above. """ out: list[dict[str, Any]] = [] for sec in sections: out.append({"tag": "hr"}) if sec.tone == "hint": # Small-grey treatment, the same one the footer gets. Card 2.0 # DELETED the ``note`` element this used to use (the API answers # 200861 "cards of schema V2 no longer support this capability"), # so the demotion is now a ``notation``-size grey markdown block. # That is strictly better here: a note was plain_text-only, which # is why the title had to be squashed into the body as a bare line # and inline markup had to be stripped rather than rendered. title_part = ( f"**{_escape_card_markdown(_rendered(sec.title))}**\n" if sec.title else "" ) body_part = _apply_inline_markup( _escape_card_markdown(_rendered(sec.body)), channel="feishu_card", ) out.append({ "tag": "markdown", "text_size": card_design.FOOTER_TEXT_SIZE, "content": f"{title_part}{body_part}", }) continue title_md = "" if sec.title: prefix = _TONE_PREFIX.get(sec.tone, "") title_md = ( f"**{prefix}{_escape_card_markdown(_rendered(sec.title))}**\n" ) body_md = _apply_inline_markup( _escape_card_markdown(_rendered(sec.body)), channel="feishu_card", ) body_md = _collapse_blank_lines(body_md) wrap_pre = _TONE_BODY_PREFIX.get(sec.tone, "") wrap_suf = _TONE_BODY_SUFFIX.get(sec.tone, "") out.append({ "tag": "markdown", "content": f"{title_md}{wrap_pre}{body_md}{wrap_suf}", }) return out #: Helix button style → the card-2.0 ``type`` that actually looks like it. #: #: 2.0 splits each colour into an outlined and a FILLED variant, and the #: unsuffixed name is the outlined one. That is the whole reason the old cards' #: 「去审批」 read as a footnote: the code asked for ``primary``, got a narrow #: white button with a blue border, and the single most important thing on the #: card looked less important than the body text above it. ``primary_filled`` #: is the solid blue. _ACTION_STYLE_TO_FEISHU: dict[str, str] = { "primary": "primary_filled", "danger": "danger_filled", "default": "default", } def _link_button( *, label: str, style: str, url: str, full_width: bool, ) -> dict[str, Any]: """One card-2.0 button that opens a URL. 2.0 moved the destination out of a bare ``url`` field and into ``behaviors``, and added ``width`` — which is what finally makes a primary action look like one. In v1 the only "primary" available rendered as a narrow outlined button, so the single most important thing on a card (「去审批」) read as a footnote next to it. """ button: dict[str, Any] = { "tag": "button", "text": {"tag": "plain_text", "content": label}, "type": _ACTION_STYLE_TO_FEISHU.get(style, "default"), "behaviors": [{"type": "open_url", "default_url": url}], } if full_width: button["width"] = "fill" return button def _render_actions( actions: list[CardAction], absolute_urls: list[str], callback_buttons: list[dict[str, Any]] | None = None, ) -> list[dict[str, Any]]: """Build the card's action row(s) from a CardAction list. ``absolute_urls`` is the matched-by-index resolved URL list (the caller resolves them via :py:meth:`NotificationService._resolve_absolute_url` before passing in, since URL resolution needs DB access). Card 2.0 deleted the ``action`` container that used to hold buttons, so the shape depends on the count: a lone button is a top-level element and goes full width; several share one ``column_set`` row and split it evenly. Both were verified against the live API — an ``action`` element is rejected outright (200861), which would have taken down every card with a button. Returns a LIST (possibly empty) rather than an optional element, because the single-button case is one element and the multi-button case is one row — the caller just extends. ``callback_buttons`` are pre-built buttons that post back to us instead of opening a browser (the assistant's 「确认提交」). They lead the row, because acting in place is the primary path and the link is the fallback. Passing them THROUGH the renderer — rather than splicing them into the finished card afterwards, which is what the assistant used to do — is what keeps the row layout in one place; the old splice hunted for an ``action`` element by tag, and card 2.0 does not have one. """ if not actions and not callback_buttons: return [] buttons: list[dict[str, Any]] = list(callback_buttons or []) for action, abs_url in zip(actions, absolute_urls, strict=False): if not abs_url: continue buttons.append(_link_button( label=_rendered(action.label), style=action.style, url=abs_url, # A single button owns the full width; in a row, each column is # already an equal share and "fill" makes the button fill ITS # column, which is what keeps a 2-up row visually even. full_width=True, )) if not buttons: return [] if len(buttons) == 1: return buttons return [{ "tag": "column_set", "flex_mode": "bisect", "horizontal_spacing": card_design.COLUMN_GAP, "columns": [ { "tag": "column", "width": "weighted", "weight": 1, "elements": [button], } for button in buttons ], }] def _render_fact_box(box: CardFactBox) -> dict[str, Any] | None: """The one shaded box: subject on the left, the deciding number on the right. ``column.padding`` rather than ``column_set.padding`` — the set-level property is rejected (200621), and without padding the shaded block sits flush against its own text, which reads like a rendering bug rather than a panel. """ subject = _rendered(box.subject) if not subject.strip(): return None left = f"**{_apply_inline_markup(_escape_card_markdown(subject), channel='feishu_card')}**" detail = _rendered(box.detail) if detail.strip(): detail_md = _apply_inline_markup( _escape_card_markdown(detail), channel="feishu_card", ) left = f"{left}\n{detail_md}" columns: list[dict[str, Any]] = [{ "tag": "column", "width": "weighted", "weight": card_design.FACT_BOX_WEIGHTS[0], "vertical_align": "center", "padding": card_design.FACT_BOX_PADDING, "elements": [{"tag": "markdown", "content": left}], }] metric = _rendered(box.metric) if metric.strip(): right = f"**{_apply_inline_markup(_escape_card_markdown(metric), channel='feishu_card')}**" hint = _rendered(box.metric_hint) if hint.strip(): hint_md = _apply_inline_markup( _escape_card_markdown(hint), channel="feishu_card", ) right = f"{right}\n{hint_md}" columns.append({ "tag": "column", "width": "weighted", "weight": card_design.FACT_BOX_WEIGHTS[1], "vertical_align": "center", "padding": card_design.FACT_BOX_PADDING, "elements": [{ "tag": "markdown", "text_align": "right", "content": right, }], }) return { "tag": "column_set", "flex_mode": "none", "background_style": "grey", "horizontal_spacing": card_design.COLUMN_GAP, "columns": columns, } def _render_facts(facts: list[CardFact]) -> dict[str, Any] | None: """The detail rows, as one markdown block. One element rather than one per fact: separate elements get the card's inter-element spacing between them, which turns four short rows into a column of floating fragments. Inside one block they read as a table. """ lines: list[str] = [] for fact in facts: label = _rendered(fact.label) value = _rendered(fact.value) if not label.strip() and not value.strip(): continue icon = card_design.FACT_ICONS.get(fact.icon, fact.icon) prefix = f"{icon} " if icon else "" label_md = _escape_card_markdown(label) value_md = _apply_inline_markup( _escape_card_markdown(value), channel="feishu_card", ) lines.append( f"{prefix}**{label_md}**{card_design.FACT_SEPARATOR}{value_md}" ) if not lines: return None return {"tag": "markdown", "content": "\n".join(lines)} def _render_collapse(collapse: CardCollapse) -> dict[str, Any] | None: """The long tail, folded behind its own counted summary line. A SHORT list is not folded at all. A producer asks for a fold because the list is unbounded — 12 roles, 9 servers — but the same call site often renders one item, and hiding a single line behind a click costs the reader an interaction to learn nothing. Under :data:`card_design.COLLAPSE_LINE_THRESHOLD` lines the content is simply shown, with its title as a bold lead-in. """ title = _rendered(collapse.title) body = _rendered(collapse.body) if not body.strip(): return None title_md = _apply_inline_markup( _escape_card_markdown(title or ""), channel="feishu_card", ) body_md = _collapse_blank_lines(_apply_inline_markup( _escape_card_markdown(body), channel="feishu_card", )) lines = [line for line in body_md.split("\n") if line.strip()] if len(lines) <= card_design.COLLAPSE_LINE_THRESHOLD: lead = f"**{title_md}**\n" if title_md else "" return {"tag": "markdown", "content": f"{lead}{body_md}"} return { "tag": "collapsible_panel", "expanded": bool(collapse.expanded), "header": { "title": {"tag": "markdown", "content": f"**{title_md}**"}, "icon": { "tag": "standard_icon", "token": card_design.COLLAPSE_ICON, "color": "grey", }, "icon_position": "right", "icon_expanded_angle": card_design.COLLAPSE_EXPANDED_ANGLE, }, # The only corner radius card 2.0 will give you: ``column_set`` rejects # ``corner_radius`` outright, so a bordered panel is also the only way # to get a rounded container. "border": { "color": "grey", "corner_radius": card_design.PANEL_CORNER_RADIUS, }, "elements": [{"tag": "markdown", "content": body_md}], } def _build_interactive_card( *, title: str, body_text: str, url: str | None, button_label: str, is_en: bool, # None = take the colour the status implies. A caller only passes this to # deliberately override the status' colour (the feed's warm palette). header_color: str | None = None, subtitle: str | None = None, stats: list[CardStat] | None = None, highlights: list[CardHighlight] | None = None, highlights_title: str | None = None, sections: list[CardSection] | None = None, extra_actions: list[CardAction] | None = None, extra_action_urls: list[str] | None = None, footer: str | None = None, body_size: str | None = None, lang: str | None = None, status: str | None = None, badges: list[CardBadge] | None = None, fact_box: CardFactBox | None = None, facts: list[CardFact] | None = None, collapse: CardCollapse | None = None, icon_img_key: str | None = None, callback_actions: list[dict[str, Any]] | None = None, ) -> dict[str, Any]: """Compose a Feishu **card 2.0** payload in the Helix design language. :: ┌──────────────────────────────────────────────┐ │ ⬡ Title [待处理] │ ← icon, badges │ subtitle │ ├──────────────────────────────────────────────┤ │ body markdown │ │ ┌────────────────────────────────────────┐ │ │ │ **Subject** **90 天** │ │ ← the one fact box │ │ detail (grey) 到 12-06 │ │ │ └────────────────────────────────────────┘ │ │ 👤 **申请人** 施宏伟 │ ← fact rows │ 📝 **原因** need billing details │ │ ┌ **涉及角色 · 12 项** ──────────────── ▾ ┐│ ← folded tail │ ┌───────────┬───────────┐ │ │ │ stat │ stat │ │ ← stat grid │ └───────────┴───────────┘ │ │ highlights / sections │ │ ── divider ── │ │ footer (grey, notation size) │ │ [ primary action, full width ] │ └──────────────────────────────────────────────┘ Everything below the header is optional; a card with none of it is a title, a paragraph and a button, which is what most notifications are. Two orderings are load-bearing and easy to get wrong: * **Escape before markup.** User-supplied display names and scope ids are escaped for ``<`` / ``>`` FIRST, so they cannot inject inline HTML; the trusted ``[r]`` / ``[b]`` markers are expanded to ```` tags AFTER. The markers are ASCII brackets and survive the escape unchanged. * **Body before the fact box.** The prose says why the card arrived; the box says what it is about. Reading the box first and the reason second is how a card ends up feeling like a database row. The header title is plain_text — Feishu never renders markup there — so ``[r]`` / ``[b]`` wrappers are stripped rather than shipped as literals. """ # Drop a trailing "— Helix IT 平台" / "— Helix 安全团队" line from # body_text before rendering. Every default body_text template # carries a signature inline so the email channel (plaintext, no # dedicated footer element) signs off, but the Feishu card renders # a signature AGAIN via the dedicated ``footer`` note below — # the recipient sees two. Strip here (card-only) so the email body # keeps its signature intact. body_text = _strip_trailing_footer(body_text or "") # …and the bare link line above the CTA, when a button already goes # there. Which URLs are actually BUTTONS mirrors the # ``if extra_actions: … elif url:`` choice the action row makes at the # bottom of this function — reading ``url`` unconditionally would strip # the body's only path to a destination on multi-CTA cards, where the # primary ``url`` is superseded and never rendered. body_text = _strip_button_duplicate_links( body_text, [u for u in (extra_action_urls or []) if u] if extra_actions else ([url] if url else []), ) body_md = _escape_card_markdown(body_text or "") body_md = _apply_inline_markup(body_md, channel="feishu_card") body_md = _collapse_blank_lines(body_md) elements: list[dict[str, Any]] = [] # The subtitle is header furniture in 2.0, not a body element. It used to # be the first grey line of the body, which put the card's "why am I # getting this" one line BELOW the fold of the header it belongs to and # pushed the actual content down. ``plain_text`` only up there, so inline # markup is stripped rather than left to render as literal brackets. subtitle_text = _strip_inline_markup(_rendered(subtitle)).strip() if subtitle else "" if url and not (url.startswith("http://") or url.startswith("https://")): notice = ( "**[Config missing — ask IT to set app.frontend_base_url]** " if is_en else "**[配置缺失 — 请 IT 配置 app.frontend_base_url]** " ) body_md = f"{notice}\n\n{body_md}" if body_md.strip(): if body_size: # Wrap the body markdown in an inline ````. # Feishu's markdown subset honours this even though the # ``markdown`` element itself has no ``text_size`` field. # The wrap is appended AFTER escape so the surrounding tag # itself is never escaped — only the body content was. body_md = f"{body_md}" elements.append({ "tag": "markdown", "content": body_md, }) if fact_box is not None: box = _render_fact_box(fact_box) if box is not None: elements.append(box) if facts: rows = _render_facts(facts) if rows is not None: elements.append(rows) if collapse is not None: panel = _render_collapse(collapse) if panel is not None: elements.append(panel) if stats: grid = _render_stat_grid(stats) if grid is not None: elements.append({"tag": "hr"}) multi = grid.get("_multi_column_set") if multi is not None: for row in multi: elements.append(row) else: elements.append(grid) if highlights: hl_md = _render_highlight_block(highlights_title, highlights) if hl_md: elements.append({"tag": "hr"}) elements.append({"tag": "markdown", "content": hl_md}) if sections: for elem in _render_sections(sections): elements.append(elem) if footer is not None and footer != "": # Skip the footer's own leading divider when the previous element is # already small-grey (a hint-tone section) — back-to-back demoted # blocks read cleaner without a rule between them. Body markdown / # sections / stat grids still get their separating divider so the # footer stays visually decoupled from the main content. prev = elements[-1] if elements else {} already_demoted = ( prev.get("tag") == "markdown" and prev.get("text_size") == card_design.FOOTER_TEXT_SIZE ) if not already_demoted: elements.append({"tag": "hr"}) elements.append({ "tag": "markdown", "text_size": card_design.FOOTER_TEXT_SIZE, "content": ( f"" f"{_escape_card_markdown(_rendered(footer))}" ), }) # Action buttons: extra_actions (multi-CTA) supersede single url. if extra_actions or callback_actions: elements.extend(_render_actions( extra_actions or [], extra_action_urls or [], callback_actions, )) elif url: elements.append(_link_button( label=button_label, style="primary", url=url, full_width=True, )) tone = card_design.tone_for(status) colour = header_color or tone.color header: dict[str, Any] = { "title": {"tag": "plain_text", "content": _strip_inline_markup(title)}, "template": colour, } if subtitle_text: header["subtitle"] = {"tag": "plain_text", "content": subtitle_text} # A semantic icon beats the brand mark whenever the state IS the message: # a green 已批准 card whose icon is still the neutral logo throws away the # signal its colour and badge just established. Brand identity is already # carried by the sender avatar next to every card in the chat. if tone.icon: # Tinted to the HEADER's colour, not the status' own default: a # template that overrides the colour (a 已转人工 receipt kept blue # rather than green) would otherwise get an icon in one colour on a # header in another. header["icon"] = { "tag": "standard_icon", "token": tone.icon, "color": colour, } elif icon_img_key: header["icon"] = {"tag": "custom_icon", "img_key": icon_img_key} tags: list[dict[str, Any]] = [] if tone.badge: tags.append({ "tag": "text_tag", "text": { "tag": "plain_text", "content": pick_variant( tone.badge, normalize_lang(lang) or DEFAULT_LANG, ) or tone.badge["zh"], }, "color": tone.badge_color or "blue", }) for badge in badges or []: text = _rendered(badge.text) if not text.strip(): continue tags.append({ "tag": "text_tag", "text": {"tag": "plain_text", "content": text}, "color": badge.color, }) if tags: header["text_tag_list"] = tags return { "schema": card_design.SCHEMA_VERSION, # ``update_multi`` is what makes a card patchable in place for every # viewer at once — the mechanism the whole receipt story rests on. "config": {"update_multi": True, "width_mode": "fill"}, "header": header, "body": {"elements": elements}, } def _escape_card_markdown(text: str) -> str: """Feishu's markdown subset accepts most prose plus a handful of inline tags. Two characters need escaping: ``<`` and ``>`` (Feishu interprets them as inline-HTML start). Newlines render as line breaks already. We deliberately preserve the input's other formatting (asterisks, dashes, colons) since templates may use them intentionally (e.g. bolded labels in scope-request DMs). """ return text.replace("<", "<").replace(">", ">") # Signature strings that every default template inlines at the bottom # of its ``body_text``. For the Feishu card, the renderer re-attaches # an identical line via the dedicated ``footer`` note element, so the # inline line is a duplicate and gets stripped here at card-build time. # Other channels (email plaintext / HTML) DON'T render a dedicated # footer — they keep the inline signature so the message still signs # off. Keep this list narrow + exact-match so we never strip a legit # substring inside body content. # # DERIVED from DEFAULT_CARD_FOOTER rather than re-listed, because the two lists # drifted: this one carried only zh + en while 78 templates already signed off # in Japanese, so a Japanese reader got their own footer AND the appended # default one — doubled, and in two different languages. _CANONICAL_FOOTERS: tuple[str, ...] = tuple(DEFAULT_CARD_FOOTER.values()) #: A sign-off line that is NOT the platform's own: domain catalogs sign their #: bodies with the desk that sent them —「— Helix 安全团队」,「— Helix 流程台账」, #: "— Helix Approval Registry", 「— Helix システム台帳」. The exact-match list #: above could never cover those, so the card showed the domain's sign-off AND #: the appended platform footer note: two signatures, one card. #: #: Anchored on a leading dash so it only ever matches a sign-off line, never #: prose that happens to mention Helix. _SIGNOFF_LINE_RE = re.compile(r"^[—–]{1,2}\s*Helix\b.*$") #: A line that is nothing but a link — the " {portal_url}" every template puts #: above its CTA. Absolute (what ``_resolve_absolute_url`` composes) or the #: relative path a producer passed through ``context`` verbatim. _BARE_LINK_LINE_RE = re.compile(r"^(?:https?://\S+|/\S*)$") #: ASCII and fullwidth colon, named rather than inlined as ``"::"`` — the #: two glyphs are hard to tell apart in a diff, and a paste that silently #: dropped the fullwidth one would look correct while doing half the job. _TRAILING_COLONS = "::" def _strip_trailing_footer(text: str) -> str: """Drop the body's own trailing sign-off line, if it has one. Card-only: the email channel has no dedicated footer element, so its body keeps the signature and still signs off. """ if not text: return text stripped = text.rstrip() for sig in _CANONICAL_FOOTERS: if stripped.endswith(sig): without = stripped[: -len(sig)] return without.rstrip() head, sep, last = stripped.rpartition("\n") if sep and _SIGNOFF_LINE_RE.match(last.strip()): return head.rstrip() return text def _link_identity(url: str) -> tuple[str, str, str] | None: """``(path, query, fragment)`` of ``url``, or ``None`` if it has no path. The HOST is deliberately dropped from the comparison: the body renders whatever the producer put in ``context`` — often the relative ``/me/api-keys`` — while the button carries the absolute URL ``_resolve_absolute_url`` composed from ``app.frontend_base_url``. Same destination, two spellings, and a plain string compare sees two links. """ try: parts = urlsplit(url.strip()) except ValueError: return None path = parts.path.rstrip("/") if not path: return None return (path, parts.query, parts.fragment) def _strip_button_duplicate_links(text: str, urls: Sequence[str]) -> str: """Drop body lines that are just a link the card already has a button for. Templates were written for email, where the body is the only clickable surface, so they park the destination on its own line above the CTA. On a card that line renders as a wrapped 60-character URL sitting directly above a button that goes to the SAME place — the reader is asked to choose between two identical options, one of which looks like debug output. Conservative on purpose: a link only goes away when the card actually offers it as a button. A body that links somewhere the card has no button for (a manual, a vendor console) keeps its link, because dropping that one would strand the reader. The colon that introduced the link goes with it — 「请到 Helix 的『我的 API Keys』页面查看凭证:」 was written to hand off to the next line, and a sentence ending in a dangling colon reads as truncated. Both colons: the catalog is 183 ASCII to 7 fullwidth, and the rare one is exactly the kind of character an editor pastes in without anybody noticing. """ targets = {key for key in (_link_identity(u) for u in urls if u) if key} if not targets or not text: return text out: list[str] = [] for line in text.split("\n"): token = line.strip() if _BARE_LINK_LINE_RE.match(token) and _link_identity(token) in targets: for i in range(len(out) - 1, -1, -1): if out[i].strip(): out[i] = out[i].rstrip().rstrip(_TRAILING_COLONS).rstrip() break continue out.append(line) return "\n".join(out) # Inline-markup convention shared by every notification template # (PR 2026-05-02). Templates use ASCII bracket tags so the ``<>`` escape # in ``_escape_card_markdown`` doesn't touch them; per-channel rewrite # happens AFTER escape so the resulting Feishu HTML survives. # # ``[r]…[/r]`` — red + bold (used for money / overage / SLA breach) # ``[b]…[/b]`` — bold (general emphasis) # ``[h]…[/h]`` — bold + one size up (the ONE number a card is about — # a leave balance, a countdown — read at a glance) # # All three render acceptably on every channel: # * Feishu interactive card → ``**…**`` / ``**…**`` # / ``**…**`` # (``heading_3`` is the size token already proven on the feed cards — # the markdown element itself has no ``text_size`` field) # * Plain-text email → ``**…**`` (most clients render asterisks # as bold; worst case they # survive as visible decoration # so the amount still pops) # * HTML email → ``…`` # / ``…`` # / ``…`` def _apply_inline_markup(text: str, *, channel: str) -> str: """Channel-aware rewrite of ``[r]`` / ``[b]`` / ``[h]`` markers. Called AFTER format-substitution AND (for Feishu) AFTER markdown escape, so the inserted ```` / ```` tags reach the renderer intact. ``channel`` ∈ {``feishu_card``, ``email_text``, ``email_html``}. """ if channel == "feishu_card": return ( text .replace("[r]", "**").replace("[/r]", "**") .replace("[b]", "**").replace("[/b]", "**") .replace("[h]", "**") .replace("[/h]", "**") ) if channel == "email_html": return ( text .replace("[r]", "").replace("[/r]", "") .replace("[b]", "").replace("[/b]", "") .replace("[h]", "") .replace("[/h]", "") ) # Default = email_text (and any other plain-text consumer). return ( text .replace("[r]", "**").replace("[/r]", "**") .replace("[b]", "**").replace("[/b]", "**") .replace("[h]", "**").replace("[/h]", "**") ) def _collapse_blank_lines(text: str) -> str: """Drop runs of 2+ consecutive blank lines down to one. Templates that pre-format with ``{cta_line}`` for the email path end up with ``"…body\n\n{cta_line}\n\n— Footer"``. When the placeholder evaluates to empty for Feishu, that becomes ``"…body\n\n\n\n— Footer"`` and Feishu renders a visible double gap. Normalise so the card looks the same regardless of which channel-specific values collapsed away. """ lines = text.split("\n") out: list[str] = [] blank_run = 0 for line in lines: if line.strip() == "": blank_run += 1 if blank_run <= 1: out.append("") else: blank_run = 0 out.append(line) return "\n".join(out) def _resolve_cta_line( *, channel: str, portal_url: str | None, lang: str | None, ) -> str: """Compute the ``cta_line`` placeholder value for a template. Templates put ``{cta_line}`` on its own line where they used to inline ``"请到 Helix 审批: {portal_url}"``. The Feishu card already carries an action button, so duplicating the URL there is noise — we render an empty string and the line collapses. Email has no button concept, so the URL stays in the body verbatim. Empty ``portal_url`` → empty string for both channels (no CTA available, don't pretend there is one). """ if not portal_url: return "" if channel == "feishu_card": return "" # Three-way, like every other localized string on this path. It used to be # a two-way ``is_en`` split — and because ``_is_en`` deliberately folds ja # to True (``core.locale.is_en``), a Japanese reader got the ENGLISH label # on the one clickable line of their email. Same two-way-split shape as the # bug already fixed in ``DEFAULT_CARD_FOOTER`` and # ``DEFAULT_CARD_BUTTON_LABEL`` above, one string over. label = CTA_LINE_LABELS[normalize_lang(lang) or DEFAULT_LANG] return f"{label}: {portal_url}" class NotificationService: def __init__(self, db: AsyncSession) -> None: self.db = db self.audit = AuditService(db) # Per-instance cache of channel→state. A service instance # lives for one request / one job tick; integration toggles are # rare, so one lookup per instance is fresh enough while a # 400-recipient broadcast loop pays the SELECT only once. self._channel_available_cache: dict[str, str] | None = None async def channel_availability(self) -> dict[str, str]: """Map each known push channel to one of three states: * ``available`` — an enabled integration backs it; send normally. * ``disabled`` — integration row(s) exist and EVERY one of them is disabled. The admin explicitly turned the channel off, so delivery SKIPS it (configured outcome, not a failure). * ``unconfigured`` — no row of the kind at all, OR enabled rows exist but ``get_primary`` still can't resolve one (multiple enabled, none primary). Both are misconfigurations, not choices — delivery still attempts the channel so it fails loudly into the retry/dead path ("no silent drops") instead of vanishing into a skip. Post-review fix (2026-06-10): the multiple-enabled-no-primary shape was previously classified "disabled", which silently skipped the channel AND marked rows DELIVERED — worse than the loud-DEAD behaviour this feature replaced, and it bypassed the critical-alert backstop. Unknown channels are always attempted — a future channel type must fail at send time, not disappear. """ if self._channel_available_cache is not None: return self._channel_available_cache from sqlalchemy import func as sa_func from sqlalchemy import select from app.integrations.models import Integration from app.integrations.service import get_primary states: dict[str, str] = {} for channel, kind in _CHANNEL_INTEGRATION_KIND.items(): try: await get_primary(kind, session=self.db) states[channel] = "available" continue except AppException: pass row_count = ( await self.db.execute( select(sa_func.count(Integration.id)).where( Integration.kind == kind, ) ) ).scalar() or 0 enabled_count = ( await self.db.execute( select(sa_func.count(Integration.id)).where( Integration.kind == kind, Integration.enabled.is_(True), ) ) ).scalar() or 0 states[channel] = ( "disabled" if (row_count and not enabled_count) else "unconfigured" ) self._channel_available_cache = states return states async def push_inbox_and_feishu( self, env: NotificationEnvelope, *, feishu: bool = True, ) -> None: """Bell-first dispatch for transient social notifications. Unlike :meth:`dispatch`, this: * Writes a single ``notification_outbox`` row with ``status=DELIVERED`` immediately — so the recipient's bell icon lights up regardless of external-channel success. * Optionally fires a best-effort Feishu push (no retry queue, no audit log). A Feishu failure leaves the row visible in the inbox; that's the right trade for low-stakes social pings. Use for feed comments / likes / mentions, where missing a single Feishu retry is acceptable but the bell icon is the user-visible contract. ``dispatch`` is still the right method for approvals / provisioning / leaver flows where the retry + dead-letter machinery genuinely matters. """ now = datetime.now(UTC) outbox = NotificationOutbox( recipient_person_id=env.recipient.id, recipient_email=env.recipient.work_email, template_key=env.template_key, title=env.title, body_text=env.body_text, body_html=env.body_html, url=env.url, # ``_jsonable_context`` (not ``dict``): a LocalizedText value # is not JSON-serialisable and would fail the INSERT outright. context=_jsonable_context(env.context), channels=list(env.channels), delivered_channels=[], status=STATUS_DELIVERED, delivered_at=now, attempts=1, max_attempts=1, hidden_from_inbox=env.hidden_from_inbox, related_resource_type=env.related_resource_type, related_resource_id=env.related_resource_id, ) # SAVEPOINT for the same reason as ``dispatch`` — a transient # DB error here shouldn't poison the caller's transaction. async with self.db.begin_nested(): self.db.add(outbox) await self.db.flush() provider_keys = _stable_provider_delivery_keys(outbox.id) outbox.provider_delivery_keys = provider_keys env.delivery_idempotency_key = provider_keys["feishu"] # Honour recipient preferences on this lightweight (social) path # too. The inbox row is already written DELIVERED above, so we only # gate the Feishu PUSH: if the recipient muted this category's # feishu channel, or is inside their quiet-hours window, suppress # the ping but keep the inbox record (reduce noise, never lose # data). Critical templates never reach this path (feed only). if feishu and "feishu" in env.channels and env.recipient is not None \ and not is_critical_template(env.template_key): muted, park = await NotificationPreferenceService( self.db, ).resolve_gating( recipient=env.recipient, template_key=env.template_key, requested_push_channels=["feishu"], ) if "feishu" in muted or park is not None: feishu = False if feishu and "feishu" in env.channels: async with best_effort( self.db, label="service best-effort step", log=logger, ) as be_2325_step: ok, err, feishu_message_id = await self._send_feishu(env) if ok: outbox.delivered_channels = ["feishu"] if feishu_message_id: outbox.feishu_message_id = feishu_message_id else: outbox.failed_channels = ["feishu"] if err: outbox.last_error = err[:4000] await self.db.flush() if not be_2325_step.ok: logger.error( "push_inbox_and_feishu Feishu push failed " "outbox_id=%s recipient=%s", outbox.id, env.recipient.id, ) async def record_external_email( self, *, recipient: IdentityPerson, recipient_email: str, subject: str, body_text: str, ok: bool, template_key: str, error: str | None = None, related_resource_type: str | None = None, related_resource_id: UUID | None = None, ) -> None: """Record an email sent OUTSIDE the ``dispatch`` pipeline as a terminal outbox row, so it appears in the notification log for admin fallback. The Google first-login password mail must go to the person's PERSONAL email (``work_email`` doesn't exist to them yet), so it's sent directly by ``account_provisioner`` rather than through ``dispatch`` (which hardcodes ``work_email``). This records that send — INCLUDING the temporary password in ``body_text`` — so an admin can read it back and hand-deliver it when the personal-email send bounced (the whole point of the record). The temp password forces a change on first login, so the exposure is bounded. Terminal (``attempts == max_attempts``, no ``next_attempt_at``) so the retry worker never re-sends it — delivery already happened out-of-band and a retry would wrongly target ``work_email``. """ row = NotificationOutbox( recipient_person_id=recipient.id, recipient_email=recipient_email, template_key=template_key, title=subject, body_text=body_text, channels=["email"], delivered_channels=["email"] if ok else [], failed_channels=[] if ok else ["email"], # A failed out-of-band send is DEAD, not FAILED: it must NOT be # auto-retried (the retry worker selects PENDING/FAILED and would # re-fire the password to work_email). DEAD = dead-letter for # manual IT fallback (reset in the Google console), which is # exactly the intent. ``_deliver`` is never involved here. status=STATUS_DELIVERED if ok else STATUS_DEAD, attempts=1, max_attempts=1, last_error=None if ok else error, delivered_at=datetime.now(UTC) if ok else None, related_resource_type=related_resource_type, related_resource_id=related_resource_id, ) self.db.add(row) await self.db.flush() async def enqueue( self, env: NotificationEnvelope, ) -> NotificationOutbox: """Persist a PENDING ``notification_outbox`` row WITHOUT attempting delivery. The durable half of :meth:`dispatch`, split out so large fan-outs can write every recipient's row first (guaranteed-delivery record) and drain delivery separately — so a crash mid-drain still leaves every un-delivered recipient recoverable by the ``notification.outbox_retry`` worker. Delivery is then done by :meth:`deliver` (immediate) or the retry worker (backstop). Returns the flushed outbox row (id populated); the caller owns the commit. """ outbox = NotificationOutbox( recipient_person_id=env.recipient.id, recipient_email=env.recipient.work_email, template_key=env.template_key, title=env.title, body_text=env.body_text, body_html=env.body_html, url=env.url, # ``_jsonable_context`` (not ``dict``): a LocalizedText value # is not JSON-serialisable and would fail the INSERT outright. context=_jsonable_context(env.context), card_payload=_serialize_card_payload(env), # Freeze the CC list (Day-1 welcome HRBP/HR Ops CC &c.) so the # retry worker / admin resend — which rebuild the envelope FROM # this row — don't silently drop it. Migration 409. cc_emails=list(env.cc_emails) if env.cc_emails else None, channels=list(env.channels), status=STATUS_PENDING, attempts=0, max_attempts=5, hidden_from_inbox=env.hidden_from_inbox, related_resource_type=env.related_resource_type, related_resource_id=env.related_resource_id, ) # Isolate the outbox INSERT in a SAVEPOINT. A transient DB error # on this flush — notably a (1205) lock-wait-timeout when a long- # running caller transaction contends on the outbox indexes — # would otherwise leave the caller's session in rolled-back- # pending state. A caller that catches the dispatch error and # keeps using the session (the documented best-effort pattern in # the cloud-bill alert paths) then hits an opaque # PendingRollbackError on its NEXT statement, masking the real # 1205 and aborting the whole job (prod cmdb.billing_daily_sync, # 2026-05-26). The savepoint rolls back only this INSERT and # expunges the pending row, keeping the caller's session usable # while re-raising the real error. Same idiom as # cmdb.lifecycle_service._try_claim_cloud_scope_alert_slot. async with self.db.begin_nested(): self.db.add(outbox) await self.db.flush() return outbox async def _apply_delivery_role_guard( self, outbox: NotificationOutbox, env: NotificationEnvelope, ) -> bool: """Re-authorize delayed confidential content immediately before I/O. The qualifying role-assignment rows are locked through the caller's send + status commit. A concurrent revocation therefore either lands before this current-read (and selects the safe fallback) or waits until this already-authorized delivery is durable. Malformed policy data and non-database lookup failures fail closed to the fallback; database operational failures propagate before any external send. """ guard = (outbox.context or {}).get(DELIVERY_ROLE_GUARD_CONTEXT_KEY) if not isinstance(guard, dict): return False required_roles = guard.get("any_roles") fallback = guard.get("fallback") policy_invalid = ( not isinstance(required_roles, list) or not required_roles or not all(isinstance(role, str) and role for role in required_roles) or not isinstance(fallback, dict) or not isinstance(fallback.get("body_text"), str) ) if policy_invalid: authorized = False else: from sqlalchemy import select from app.auth.models import AuthRole, AuthUserRole from app.core.rbac import normalize_role_names required_role_names = cast(list[str], required_roles) # savepoint-exempt: OperationalError re-raised below, read-only otherwise try: role_names = list((await self.db.execute( select(AuthRole.name) .select_from(AuthUserRole) .join(AuthRole, AuthRole.id == AuthUserRole.role_id) .where(AuthUserRole.user_id == env.recipient.id) .with_for_update() )).scalars().all()) authorized = bool( set(normalize_role_names(role_names)).intersection( normalize_role_names(required_role_names) ) ) # savepoint-exempt: OperationalError (the transaction-aborting # family) is already re-raised above rather than swallowed, which # is the property the savepoint would provide; the remaining # branch only reads and then fails closed. except OperationalError: # The transaction may be invalid and cannot durably record a # send. Bubble before network I/O; the worker will roll back. raise except Exception: # noqa: BLE001 logger.exception( "notification delivery role guard lookup failed (outbox=%s)", getattr(outbox, "id", None), ) authorized = False if authorized: return False self._replace_with_delivery_role_fallback(outbox, env) return True @staticmethod def _replace_with_delivery_role_fallback( outbox: NotificationOutbox, env: NotificationEnvelope | None = None, ) -> bool: """Permanently replace guarded content with its amount-free form. This helper deliberately works without an envelope so crash recovery can redact a stale/uncertain claimed row in a fresh transaction. A durable delivery claim is committed before role lookup; therefore a worker can die before the lookup or after provider I/O. In either case an uncertain personal-inbox row must never retain the frozen confidential representation. """ guard = (outbox.context or {}).get(DELIVERY_ROLE_GUARD_CONTEXT_KEY) if not isinstance(guard, dict): return False # Permanently replace the frozen confidential representation with the # fallback. Once a role check fails, a later retry must not resurrect # the old body from this row, and the personal inbox must only expose # the same safe content after delivery. fallback = guard.get("fallback") safe_fallback = fallback if isinstance(fallback, dict) else {} title = str(safe_fallback.get("title") or "受限通知") body_text = str( safe_fallback.get("body_text") or "该通知包含受限信息,投递时无法确认当前权限,内容已隐藏。" ) body_html = ( safe_fallback.get("body_html") if isinstance(safe_fallback.get("body_html"), str) else None ) fallback_url = safe_fallback.get("url") or outbox.url url = str(fallback_url) if fallback_url else None card_payload = ( safe_fallback.get("card_payload") if isinstance(safe_fallback.get("card_payload"), dict) else None ) if env is not None: card_fields = _deserialize_card_payload(card_payload) env.title = title env.body_text = body_text env.body_html = body_html env.url = url env.card_subtitle = card_fields.get("card_subtitle") env.card_stats = card_fields.get("card_stats", []) env.card_highlights = card_fields.get("card_highlights", []) env.card_highlights_title = card_fields.get("card_highlights_title") env.card_sections = card_fields.get("card_sections", []) env.card_actions = card_fields.get("card_actions", []) env.card_footer = card_fields.get("card_footer") env.card_body_size = card_fields.get("card_body_size") outbox.title = title outbox.body_text = body_text outbox.body_html = body_html outbox.url = url outbox.card_payload = card_payload return True async def deliver( self, outbox: NotificationOutbox, env: NotificationEnvelope, *, trace_id: str = "", actor_user_id: UUID | None = None, ) -> NotificationResult: """Durably deliver an already-committed :meth:`enqueue` row. This method owns the claim/result commits. It is for phase-two drains only; :meth:`dispatch` retains its legacy caller-transaction semantics and uses the internal attempt path directly. """ fresh = await self.db.get( NotificationOutbox, outbox.id, with_for_update=True, populate_existing=True, ) if fresh is None or fresh.status not in { STATUS_PENDING, STATUS_FAILED, STATUS_PARTIAL, }: return NotificationResult() return await self._deliver_with_durable_claim( fresh, env, trace_id=trace_id, actor_user_id=actor_user_id, ) async def dispatch( self, env: NotificationEnvelope, *, trace_id: str = "", actor_user_id: UUID | None = None, ) -> NotificationResult: """Outbox-first dispatch. Creates a ``notification_outbox`` row, tries each requested channel, updates the row with delivery results + retry state. Partial failures stay pending for retry; total failures sit until the retry worker picks them up. The returned ``NotificationResult`` reflects the outcome of THIS attempt only — subsequent retries update the outbox row asynchronously. Convenience over :meth:`enqueue` + :meth:`deliver` for the common single-recipient case. High-volume fan-outs should enqueue every row first, then drain delivery, so the caller's transaction never holds locks across the (slow, external) channel sends — see ``AssetInventoryService._dispatch_initial_notifications``. """ outbox = await self.enqueue(env) return await self._deliver( outbox, env, trace_id=trace_id, actor_user_id=actor_user_id, ) async def _deliver( self, outbox: NotificationOutbox, env: NotificationEnvelope, *, trace_id: str, actor_user_id: UUID | None, ) -> NotificationResult: """Run one delivery attempt against the channels in ``env``, updating the outbox row. Shared between first-try dispatch and the retry worker. """ await self._apply_delivery_role_guard(outbox, env) provider_keys = dict( outbox.provider_delivery_keys or _stable_provider_delivery_keys(outbox.id) ) outbox.provider_delivery_keys = provider_keys env.delivery_idempotency_key = provider_keys["feishu"] env.email_message_id = provider_keys["email"] result = NotificationResult() delivered: list[str] = list(outbox.delivered_channels or []) failed: list[str] = [] errors: list[str] = [] # Only try channels that haven't already been delivered — a # partial success on an earlier attempt stays delivered. todo = [c for c in env.channels if c not in delivered] # Channel honesty: channels whose backing integration is # explicitly DISABLED are skipped, not attempted-and-failed. # They don't count toward failure/permanence/status math — the # admin turned them off, so "didn't send there" is the # configured outcome. A skip note lands in last_error for # forensics but never in failed_channels. Channels that were # never configured at all still go through the attempt path so # the failure stays loud ("no silent drops"). availability = await self.channel_availability() skipped = [ c for c in todo if availability.get(c) == "disabled" ] todo = [c for c in todo if c not in skipped] skip_notes = [ f"{c}: skipped — integration disabled" for c in skipped ] # Per-recipient preference gating (notification governance). The # in-app inbox row is already persisted (this outbox row IS it), so # the gate governs only the PUSH channels. Critical templates bypass # it entirely — an ops-critical alert always pushes regardless of any # mute / quiet-hours setting. Recipients with no preference row # resolve to "no muting, no quiet hours" (the sparse-table default). if env.recipient is not None and not is_critical_template( env.template_key, ): push_todo = [c for c in todo if c in GOVERNABLE_CHANNELS] muted_push, park_until = await NotificationPreferenceService( self.db, ).resolve_gating( recipient=env.recipient, template_key=env.template_key, requested_push_channels=push_todo, ) if muted_push: # A muted channel is a configured outcome, not a failure — # treat it exactly like a disabled-integration skip so the # status math stays honest (the row completes inbox-only when # every push channel is muted). todo = [c for c in todo if c not in muted_push] skipped.extend(muted_push) skip_notes.extend( f"{c}: skipped — muted by recipient preference" for c in muted_push ) remaining_push = [c for c in todo if c in GOVERNABLE_CHANNELS] if park_until is not None and remaining_push: # Quiet hours: park the still-active push channels. Leave the # row PENDING with next_attempt_at at the window end; the # existing outbox retry-drain delivers it then (iOS # scheduled-summary semantics — held during the window, then # delivered once it lifts). attempts is NOT incremented: # nothing was attempted. A synthetic ``deferred`` marker on # the result tells re-fan-out callers the dispatch was handled # so they don't re-enqueue a duplicate every tick (mirrors the # ``inbox`` synthetic marker used by the all-skipped path). outbox.next_attempt_at = park_until outbox.status = STATUS_PENDING note = "; ".join( skip_notes + ["deferred — recipient quiet hours"] ) outbox.last_error = note[:4000] or None # Persist the deferral inside a SAVEPOINT so a transient # flush failure (e.g. 1205 lock-wait on the outbox indexes # while a long-running caller transaction contends) rolls # back ONLY this status update and leaves the CALLER's # shared session usable. Without this, a flush failure here # leaves the session in rolled-back-pending state and every # subsequent statement in a per-item cron loop hits an # opaque PendingRollbackError, aborting the whole job # (root-cause fix, 2026-06-20). Deadlocks (1213) still # bubble so the fresh-session lock-retry wrapper restarts # the idempotent run cleanly. try: async with self.db.begin_nested(): await self.db.flush() except OperationalError: raise except Exception as e: # noqa: BLE001 logger.warning( "notification deferral flush failed (outbox=%s): %s", getattr(outbox, "id", None), e, ) result.delivered_channels.append("deferred") return result result.delivered_channels.append("deferred") return result if "feishu" in todo: ok, err, feishu_message_id = await self._send_feishu(env) result.feishu_ok = ok result.feishu_error = err if ok: delivered.append("feishu") result.delivered_channels.append("feishu") if feishu_message_id: outbox.feishu_message_id = feishu_message_id else: failed.append("feishu") if err: errors.append(f"feishu: {err}") if "email" in todo: ok, err = await self._send_email(env) result.email_ok = ok result.email_error = err if ok: delivered.append("email") result.delivered_channels.append("email") else: failed.append("email") if err: errors.append(f"email: {err}") outbox.attempts = (outbox.attempts or 0) + 1 outbox.delivered_channels = delivered outbox.failed_channels = failed or None outbox.last_error = ( "; ".join(errors + skip_notes)[:4000] if (errors or skip_notes) else None ) # Status math runs over the EFFECTIVE channel set — requested # minus disabled-integration skips. With every push channel # skipped the in-app inbox row (which this outbox row IS) is # the delivery surface, so the row completes as DELIVERED # rather than rotting in the retry queue. effective_channels = [c for c in env.channels if c not in skipped] all_requested_delivered = all( c in delivered for c in effective_channels ) any_delivered = bool(delivered) # Pre-launch hardening 2026-05-17 (Round-19 audit, finding D): # classify permanent errors so we don't re-attempt them five # times with exponential backoff. A recipient with no # ``work_email``, a person never re-onboarded to Feishu, a # Feishu integration that's still unconfigured — these don't # heal between attempts and the retry storm only wastes worker # cycles + audit rows. Treat them like ``attempts >= max``: # transition to DEAD on the first try, no next_attempt_at. # Network / 5xx / rate-limit errors continue to retry normally. permanent_error = _all_failures_permanent( env, result, delivered=delivered, skipped=skipped, ) if all_requested_delivered: outbox.status = STATUS_DELIVERED outbox.delivered_at = datetime.now(UTC) outbox.next_attempt_at = None # Post-review fix (2026-06-10): when the row converged via # the inbox-only path (every push channel skipped, nothing # actually pushed), tell the CALLER so too. Several callers # gate their own bookkeeping on "did anything deliver" — # offboarding escalation skips its ReminderLog write when # ``delivered_channels`` is empty and re-fans-out every # tick, flooding the in-app inbox with duplicate rows that # all claim DELIVERED. ``inbox`` is a synthetic channel # marker for the result only; the outbox row's own # ``delivered_channels`` stays push-channel-truthful. if not delivered and skipped: result.delivered_channels.append("inbox") elif any_delivered: # At least one channel landed — the recipient WAS notified, # so this row is PARTIAL and can never become DEAD. What is # still open is whether the channel that MISSED deserves # another go, and that turns entirely on why it failed: # # * permanent ("Feishu open_id not found" — the person is # not in the directory) → nothing to retry. Freeze. # * ``_PARTIAL_MAX_ATTEMPTS`` passes used → freeze. The # other channel already delivered, so a very late second # copy is a surprise, not a completion; the short cap is # what keeps this from becoming the "days apart" # duplicate the 2026-06-02 freeze was written for. # * attempts exhausted → freeze; the ladder is spent. # * anything else (Feishu ``2200 Internal Error``, ``9499 # too many request``, a socket timeout) → arm the # backoff. ``_deliver`` re-attempts only channels absent # from ``delivered_channels``, so the retry sends the # missing card and does NOT re-send the mail that landed. # Completing the second channel is not duplication: the # envelope asked for BOTH, and every fully-delivered # recipient of the same fan-out got both. # # Freezing unconditionally (the behaviour until now) meant # one transient Feishu blip permanently dropped that card # while the row still read as reached. Found on the 2026-09-07 # CEO-fraud awareness broadcast: 1 of 416 recipients lost her # Feishu card to a single 2200 and nothing ever retried it. outbox.status = STATUS_PARTIAL if ( permanent_error or outbox.attempts >= _PARTIAL_MAX_ATTEMPTS or outbox.attempts >= outbox.max_attempts ): outbox.next_attempt_at = None else: outbox.next_attempt_at = _next_attempt_at(outbox.attempts) elif outbox.attempts >= outbox.max_attempts or permanent_error: outbox.status = STATUS_DEAD outbox.next_attempt_at = None # Guaranteed-delivery backstop: an ops-critical alert just # died on every push channel — fire the last-resort group # webhook (independent transport) so a human still hears # about it. Best-effort by design; the outcome lands in # ``context['critical_fallback']`` + an audit row either way. if is_critical_template(outbox.template_key): await self._fire_critical_fallback( outbox, errors=errors, trace_id=trace_id, ) else: outbox.status = STATUS_FAILED outbox.next_attempt_at = _next_attempt_at(outbox.attempts) # Persist the delivery-status update inside a SAVEPOINT. The actual # channel sends (feishu/email above) have already happened and are # reflected in ``result``; this flush only writes the outbox row's # status bookkeeping. A transient flush failure (notably 1205 # lock-wait when a long-running caller transaction contends on the # outbox indexes) must NOT leave the caller's shared session in # rolled-back-pending state — otherwise every subsequent statement # in a per-item cron loop that catches the dispatch error and keeps # using the session hits an opaque PendingRollbackError and the # whole job aborts (root-cause fix for cron mid-batch session # poisoning, 2026-06-20). The savepoint rolls back only this status # update and keeps the session usable; we return the # already-computed ``result`` (which reflects the true per-channel # delivery outcome). Deadlocks (1213) still bubble so the # fresh-session lock-retry wrapper restarts the idempotent run. try: async with self.db.begin_nested(): await self.db.flush() except OperationalError: raise except Exception as e: # noqa: BLE001 logger.warning( "notification status flush failed (outbox=%s): %s", getattr(outbox, "id", None), e, ) return result # 🔴 SAVEPOINT around the audit write too, for the same reason as the # status flush above — and this one was the LAST unisolated mutation # on the ``dispatch`` path. # # ``AuditService.log`` is a bare ``add()`` + ``flush()``; it does not # self-isolate (the savepoint guard's own docstring says so). So # until this block existed, "``dispatch`` cannot poison the caller's # transaction" was true of the outbox INSERT and the status UPDATE # and false of the audit row — and ~52 call sites across the codebase # rely on that sentence to justify a bare ``except`` around a # dispatch. A ``detail`` value that overflows its column, or a lock # wait on ``audit_logs``, would have taken the caller down with it. # # Losing an audit row is the lesser harm and it is loud (this # warning); losing the caller's whole transaction is silent until # its next statement raises ``PendingRollbackError`` somewhere that # names neither notifications nor auditing. try: async with self.db.begin_nested(): await self.audit.log( trace_id=trace_id, actor_type="system" if not actor_user_id else "user", actor_user_id=actor_user_id, action="notification.dispatched", status="success" if all_requested_delivered else "failure", resource_type="identity_person", resource_id=str(env.recipient.id), detail={ "outbox_id": str(outbox.id), "title": env.title, "template_key": env.template_key, "channels_requested": list(env.channels), "channels_delivered": list(delivered), "channels_failed": list(failed), "channels_skipped": list(skipped), "attempt": outbox.attempts, "status": outbox.status, "errors": errors, }, ) except OperationalError: # Deadlock / lock-wait aborts the transaction server-side; a # savepoint rollback cannot rescue it, so let the fresh-session # retry wrapper restart the run. Same posture as the status # flush above. raise except Exception as audit_exc: # noqa: BLE001 logger.warning( "notification dispatch audit write failed (outbox=%s): %s", getattr(outbox, "id", None), audit_exc, ) return result async def _fire_critical_fallback( self, outbox: NotificationOutbox, *, errors: list[str], trace_id: str, ) -> None: """Last-resort delivery for a DEAD ops-critical notification. Posts a short text to the Feishu **group webhook** (``alert_webhook_url`` on the chat_approval integration) — a transport independent of the per-user DM API that just failed. Never raises; the outcome (sent / rate_limited / no-webhook / post-failed) is stamped into ``context['critical_fallback']``; attempted sends are additionally audited (the rate-limited early-return is context-stamp-only — the earlier send within the window carries the audit trail). The notification BODY is deliberately not forwarded — only title + recipient + error — so a webhook with a broader audience than the original recipient can't leak payload detail (credentials, per-person figures). Post-review hardening (2026-06-10): rate-limited per template_key (1 webhook line per hour per process) — the design scenario is a channel-level outage where hundreds of rows die at once, and a per-row webhook would bury the signal; the config read catches ALL exceptions (a transient DB error must not breach the never-raises contract); HTTP timeout is 5 s because the caller may hold row locks in the retry batch transaction. """ from sqlalchemy.orm.attributes import flag_modified outcome: dict[str, Any] = { "at": datetime.now(UTC).isoformat(), } if _critical_fallback_rate_limited(outbox.template_key): outcome["status"] = "rate_limited" ctx = dict(outbox.context or {}) ctx["critical_fallback"] = outcome outbox.context = ctx flag_modified(outbox, "context") return webhook_url = "" try: cfg = await get_feishu_config(session=self.db) webhook_url = str(cfg.get("alert_webhook_url") or "") except Exception: # noqa: BLE001 — backstop never raises pass if not webhook_url: outcome["status"] = "no_webhook_configured" else: msg = ( "[必达兜底] 一条关键通知在所有渠道投递失败,已死信:\n" f" · 模板: {outbox.template_key}\n" f" · 标题: {outbox.title[:120]}\n" f" · 收件人: {outbox.recipient_email or outbox.recipient_person_id}\n" f" · 错误: {('; '.join(errors) or outbox.last_error or '')[:300]}\n" f" · outbox_id: {outbox.id}\n" "请到 Helix 后台「通知记录」页排查渠道并重投。" ) try: import httpx from app.integrations._url_safety import ( UnsafeURLError, safe_async_http_transport, ) try: transport = await safe_async_http_transport( webhook_url, allowed_suffixes=("feishu.cn", "larksuite.com"), ) except UnsafeURLError as exc: outcome["status"] = "webhook_rejected" # ``exc.message`` = host + reason only — str(exc) # is the args tuple incl. the FULL webhook URL, # whose path IS the bot token. outcome["error"] = exc.message[:200] else: async with httpx.AsyncClient( timeout=5, trust_env=False, transport=transport, ) as client: resp = await client.post( webhook_url, json={ "msg_type": "text", "content": {"text": msg}, }, ) body = resp.json() ok = ( resp.status_code == 200 and body.get("StatusCode") in (0, None) and body.get("code") in (0, None) ) outcome["status"] = "sent" if ok else "post_failed" if not ok: outcome["error"] = str(body)[:200] except Exception as exc: # noqa: BLE001 — backstop never raises outcome["status"] = "post_failed" outcome["error"] = str(exc)[:200] _critical_fallback_stamp( outbox.template_key, sent=outcome.get("status") == "sent", ) ctx = dict(outbox.context or {}) ctx["critical_fallback"] = outcome outbox.context = ctx flag_modified(outbox, "context") async with best_effort( self.db, label="notification: critical-fallback audit", log=logger, ) as be_3020_step: await self.audit.log( trace_id=trace_id, actor_type="system", actor_user_id=None, action="notification.critical_fallback", status=( "success" if outcome.get("status") == "sent" else "failure" ), resource_type="notification_outbox", resource_id=str(outbox.id), detail={ "template_key": outbox.template_key, "outcome": outcome, }, ) if not be_3020_step.ok: logger.error( "critical-fallback audit write failed for outbox=%s", outbox.id, ) # ── Feishu ─────────────────────────────────────────────── async def _send_feishu( self, env: NotificationEnvelope, ) -> tuple[bool, str | None, str | None]: """Push the notification to the recipient's Feishu DM. We always send an **interactive card** (msg_type=interactive) rather than plain text: * The card has a header (recipient locale-aware), a markdown body, and an action button when ``env.url`` is set. * URLs are rendered absolute by ``_resolve_absolute_url`` so users can tap the button straight to the real Helix page — previously the body contained raw paths like ``/my/cloud/scopes/`` which Feishu rendered as plain text and users couldn't click. * The Chinese baseline / English variant is picked by the recipient's ``preferred_language`` (the same selection used for the email body), so a user who set en in /me/settings gets an English Feishu card. Plain-text fallback is dropped on purpose — the previous code path produced unclickable URLs. If a future driver/channel needs plain text, it should pull ``env.body_text`` directly. """ email = env.recipient.work_email if not email: return False, "recipient has no work_email", None try: cfg = await get_feishu_config(session=self.db) except AppException as e: return False, f"feishu integration not configured: {e.message}", None from app.approval.feishu_client import FeishuApprovalClient client = FeishuApprovalClient( app_id=str(cfg.get("app_id") or ""), app_secret=str(cfg.get("app_secret") or ""), ) absolute_url = await self._resolve_absolute_url(env.url) # One resolver for every channel (``app.core.locale``): explicit # preference → country → labour-contract entity → baseline. Reading # ``preferred_language`` raw here is what let free-text SF values # ("chinese") pick the wrong template on some surfaces. lang = resolve_lang(env.recipient) # Loaded once here and reused by ``_resolve_button_label``, saving one # of the three ``platform_settings`` reads this path used to do per # recipient (``_render_for_recipient`` still loads its own). # # NOTE the saving only applies when the template EXISTS: a missing key # yields ``None``, the reuse check falls through, and all three loads # (and all three "no template" warnings) happen as before. The Day-1 # welcome card is exactly that case — ``hire_day_it_info.feishu_card`` # is a classification-only key with no registered template. chrome_tmpl = ( await self._load_template(env.template_key) if env.template_key else None ) button_label = await self._resolve_button_label(env, lang, tmpl=chrome_tmpl) title, body_text = await self._render_for_recipient( env, lang, channel="feishu_card", absolute_url=absolute_url, ) # Resolve every CardAction URL up-front (sync DB lookup, can't # be done inside the pure builder). extra_action_urls: list[str] = [] for action in env.card_actions: extra_action_urls.append( await self._resolve_absolute_url(action.url) or "" ) # Card chrome is resolved HERE, per recipient — one envelope fans out # to people who do not share a language, so the producer cannot have # picked. Non-localized values pass through untouched. chrome = _resolve_card_chrome( env, lang, tmpl=chrome_tmpl, absolute_url=absolute_url, ) footer = chrome["footer"] if footer is None: # Only the DEFAULT footer carries the warm sign-off. A template # that supplied its own ``card_footer`` (the leave reminder's HR # 寄语) already chose its closing voice deliberately. footer = compose_default_card_footer( lang, seed=env.template_key or env.title, ) status = _card_status(env) # Best-effort: a card with no custom icon is a card; a card that # failed to send because the icon lookup raised is a person who never # heard about the thing. icon_img_key = await card_icon.resolve_icon_img_key( self.db, client, app_id=str(cfg.get("app_id") or ""), ) card = _build_interactive_card( title=title, body_text=body_text, url=absolute_url, button_label=button_label, is_en=_is_en(lang), lang=lang, status=status, header_color=_card_header_color(env.template_key, status), subtitle=chrome["subtitle"], stats=chrome["stats"] or None, highlights=chrome["highlights"] or None, highlights_title=chrome["highlights_title"], sections=chrome["sections"] or None, extra_actions=chrome["actions"] or None, extra_action_urls=extra_action_urls or None, footer=footer, body_size=env.card_body_size, badges=chrome["badges"] or None, fact_box=chrome["fact_box"], facts=chrome["facts"] or None, collapse=chrome["collapse"], icon_img_key=icon_img_key, ) try: open_id = await client.resolve_open_id_by_email(email) sent = await client.send_card_to_user( open_id=open_id, card=card, idempotency_key=env.delivery_idempotency_key, ) # The id Feishu assigned to this card. Kept because an # employee replying under it sends ``parent_id`` pointing # here, and that is the ONLY thing that can turn a bare # 「这台我早就还了」 into a question about a known asset. # Read defensively: a benign-code path can return ``{}``, # and losing a message id must never fail a delivery that # actually happened. message_id = None if isinstance(sent, dict): raw = sent.get("message_id") if isinstance(raw, str) and raw: message_id = raw[:64] return True, None, message_id except AppException as e: return False, e.message, None except Exception as e: # noqa: BLE001 — network flakiness logger.warning("feishu push failed for %s: %s", email, e) return False, str(e), None async def _resolve_absolute_url(self, url: str | None) -> str | None: """Compose an absolute URL from a NotificationEnvelope.url. ``url`` may be: * ``None`` — no link rendered. * Already absolute (``http://`` / ``https://``) — pass through. * A relative path (``/my/cloud/scopes/...``) — prefix with the configured ``app.frontend_base_url`` setting. If the platform setting is unset / empty, we return the relative path AS-IS rather than dropping the link, and the admin will see the missing config in the Feishu card body (``[配置缺失]`` wrapper added in `_build_interactive_card`). That's strictly better than silently swallowing the navigation hint — the user can still copy/paste the path. """ if not url: return None if url.startswith("http://") or url.startswith("https://"): return url base = await get_platform_setting( "app.frontend_base_url", default=None, session=self.db, ) if not base or not isinstance(base, str): # Surface the misconfig but don't drop the path entirely; # the user can still copy it into the address bar. return url # Strip trailing slashes from base + leading slash from path # to form a clean concatenation regardless of either side. base = base.rstrip("/") path = url if url.startswith("/") else f"/{url}" return f"{base}{path}" async def _resolve_button_label( self, env: NotificationEnvelope, lang: str | None, *, tmpl: dict[str, str] | None = None, ) -> str: """Pick the action-button label (Chinese / English) for the Feishu interactive card. Defaults to a generic 'Helix' label if the template doesn't define one — common for envelopes constructed without a template_key. """ if env.template_key: # Reuse the caller's copy when it has one — the Feishu path already # loaded this exact template to resolve the card chrome. if tmpl is None: tmpl = await self._load_template(env.template_key) if tmpl is not None: label = _select_template_field(tmpl, "card_button_label", lang) if label: # ``_safe_format`` + the expanded context + ValueError, # matching the chrome path. A bare ``.format`` on an # admin-editable field raised ValueError on ``{amount:d}``, # and that escaped ``_send_feishu`` into # ``_deliver_with_durable_claim``'s ``except BaseException`` # — which flips the row to ``delivery_uncertain``, i.e. # "automatic resend refused". One bad character in a button # label stranded the notification. try: return _safe_format( label, _localized_context(env.context, lang), ) except (KeyError, IndexError, ValueError): return label return DEFAULT_CARD_BUTTON_LABEL[normalize_lang(lang) or DEFAULT_LANG] async def _render_for_recipient( self, env: NotificationEnvelope, lang: str | None, *, channel: str, absolute_url: str | None = None, ) -> tuple[str, str]: """Render (title, body_text) for the recipient's preferred language. Falls back to ``env.title`` / ``env.body_text`` when no template is set (envelopes built directly with rendered strings — common for one-off user-message helpers). ``absolute_url``, when provided, is injected into the format context as ``portal_url`` (unless the caller already supplied their own ``portal_url`` in ``env.context``). This mirrors ``_send_email`` so a template referencing ``{portal_url}`` renders identically on both channels — Feishu cards and emails — without each caller having to remember to pass the URL twice. Empty / None ``absolute_url`` skips the injection so a missing URL surfaces as a KeyError-recovered raw ``{portal_url}`` (caught below; the body still ships). ``channel`` ∈ {``feishu_card``, ``email_text``, ``email_html``} controls two channel-aware substitutions: * ``{cta_line}`` — empty for Feishu (the action button covers the call-to-action), ``"请到 Helix 查看: "`` for email (the only place users can click). * Inline ``[r]…[/r]`` / ``[b]…[/b]`` markers — rewritten to per-channel emphasis (red+bold or bold) AFTER format substitution. Templates use the brackets uniformly; the renderer adapts. Bodies that don't use the markers / placeholder keep working unchanged — both rewrites are no-ops on absent input. """ if not env.template_key: # Direct envelope (no template). Markup conversion happens # in the channel renderer (Feishu must escape first), so # we just hand back the raw body verbatim. return env.title, env.body_text tmpl = await self._load_template(env.template_key) if tmpl is None: # No template registered for this key — fall through to the # envelope. Without this guard ``_load_template`` used to # synthesize a placeholder dict whose truthy "[Helix] 通知" / # "(no template configured)" strings overrode the envelope's # real title and body, shipping a useless card to production. # The fallback now lives entirely on the envelope path so the # caller's rendered content reaches the recipient verbatim. return env.title, env.body_text subject = _select_template_field(tmpl, "subject", lang) or env.title body = _select_template_field(tmpl, "body_text", lang) or env.body_text format_context = _localized_context(env.context, lang) if absolute_url and "portal_url" not in format_context: format_context["portal_url"] = absolute_url # Channel-aware CTA line: Feishu card hides it (button covers # the CTA), email keeps the inline URL since email has no # interactive surface. Caller-supplied cta_line wins so a # specialty template can override the wording. if "cta_line" not in format_context: format_context["cta_line"] = _resolve_cta_line( channel=channel, portal_url=format_context.get("portal_url"), lang=lang, ) # ``_safe_format`` renders known placeholders and leaves a # genuinely-missing key literal (template gained a placeholder # AFTER this row's context was frozen — the retry-KeyError bug); # the broad except stays as last resort for positional fields / # bad format specs in admin-authored overrides. try: subject = _safe_format(subject, format_context) except (KeyError, IndexError, ValueError): pass try: body = _safe_format(body, format_context) except (KeyError, IndexError, ValueError): pass # NB: ``[r]`` / ``[b]`` markup conversion is NOT done here — # it happens in the channel renderer (``_build_interactive_card`` # for Feishu, ``_send_email`` for email) so the Feishu path can # escape ``<>`` BEFORE the markers expand to ```` tags. return subject, body # ── Email ──────────────────────────────────────────────── async def _send_email(self, env: NotificationEnvelope) -> tuple[bool, str | None]: if not env.recipient.work_email: return False, "recipient has no work_email" try: driver = await build_primary_driver("email", session=self.db) except AppException as e: return False, f"email integration not configured: {e.message}" if not isinstance(driver, EmailDriver): return False, f"primary email integration is not an EmailDriver ({driver.provider})" # Same resolver as the Feishu path — see ``_send_feishu``. lang = resolve_lang(env.recipient) absolute_url = await self._resolve_absolute_url(env.url) # Inject the absolute URL into the format context so templates # can reference {portal_url} regardless of whether the caller # already passed it. Caller-supplied portal_url wins (don't # overwrite an explicit per-call value). # Same alias expansion as the Feishu / bell renderers. Without it a # row frozen before the 2026-08-09 context renames, retried through # email, renders a literal ``{blessing_zh}`` in the day-one welcome # mail while the very same row's card and bell entry render fine. format_context = _localized_context(env.context, lang) if absolute_url and "portal_url" not in format_context: format_context["portal_url"] = absolute_url # ``cta_line`` mirrors the Feishu render path — see # ``_render_for_recipient``. For email, the line stays inline # since email has no clickable button. if "cta_line" not in format_context: format_context["cta_line"] = _resolve_cta_line( channel="email_text", portal_url=format_context.get("portal_url"), lang=lang, ) subject = env.title body_text = env.body_text body_html = env.body_html template_applied = False if env.template_key: tmpl = await self._load_template(env.template_key) # ``tmpl is None`` → no override, no builtin: the envelope's # own title/body/html are the sole source of truth. Guard # against the (no template configured) regression: previously # ``_load_template`` returned a dummy dict whose truthy # subject/body fields silently displaced the envelope. if tmpl is not None: subject_t = _select_template_field(tmpl, "subject", lang) body_t = _select_template_field(tmpl, "body_text", lang) html_t = _select_template_field(tmpl, "body_html", lang) # Only count as ``template_applied`` when the template # actually contributed at least one non-empty body / # subject field. An empty-tmpl-dict (badly-seeded # platform_settings override) otherwise suppresses the # CTA fallback below and the recipient gets prose-only. if subject_t or body_t or html_t: template_applied = True try: # _safe_format: missing keys stay literal instead of # KeyError-ing the whole render (same fix as # _render_for_recipient — a frozen pre-template-change # context must not blank every placeholder on retry). if subject_t: subject = _safe_format(subject_t, format_context) if body_t: body_text = _safe_format(body_t, format_context) if html_t: body_html = _safe_format(html_t, format_context) except (KeyError, IndexError, ValueError) as e: # Match the Feishu (_render_for_recipient) and inbox # render paths, which already catch (KeyError, IndexError, # ValueError). A bad admin-authored template override — # a positional/empty placeholder ({0}/{}) → IndexError, or # a bad format spec ({amount:d} on a str) → ValueError — # must NOT escape _send_email: retry_pending_outbox loops # _deliver over up to 100 rows with a single post-loop # commit and no per-row guard, so an uncaught raise here # rolls back the whole batch (re-sending rows already # physically delivered → duplicate pings) and 500s the # manual single-row retry endpoint. logger.warning( "email template %s render failed (%s): %s", env.template_key, type(e).__name__, e, ) # Fallback CTA append: when no template handled the body but the # envelope carries a URL (typical for rich-card envelopes whose # template_key isn't in _DEFAULT_TEMPLATES, e.g. cloud.owner_*), # append the resolved CTA line so email recipients can actually # click through. Without this, email gets prose-only and the # recipient has nowhere to act — the owner-confirmation card was # a 14-day silent gap before this fix. if not template_applied and absolute_url: cta_line = format_context.get("cta_line", "") if ( cta_line and body_text and cta_line not in body_text and absolute_url not in body_text ): body_text = f"{body_text}\n\n{cta_line}" # Mirror the append into the HTML body so html-only readers # also get a clickable CTA. HTML-escape both the URL (which # carries query-string ``&`` that breaks unescaped HTML # entity parsing in Outlook / Gmail's strict MUAs) and the # label (defense-in-depth — locale strings are constants # today but future-proof against translation injection). if ( body_html and absolute_url not in body_html ): import html # 与 ``_resolve_cta_line`` **同一张表**:两处各写一份 if/else 的 # 结果就是同一封信的纯文本正文说「Helix で確認する」、HTML 正文说 # 「View in Helix」。日语被 ``_is_en`` 刻意折成 True,所以两分支 # 版本给 ja 读者的是英文标签(不是中文 —— 这一点此前记错了)。 cta_label = CTA_LINE_LABELS[normalize_lang(lang) or DEFAULT_LANG] safe_url = html.escape(absolute_url, quote=True) safe_label = html.escape(cta_label) body_html = ( f'{body_html}

' f'{safe_label}

' ) # Snapshot the resolved-but-still-raw body (placeholders # substituted, inline ``[r]/[b]`` markup intact) BEFORE the # text-channel mutation below. The branded HTML shell does its # own escaping + markup conversion, so it needs the raw text. html_source_text = body_text # Inline-markup conversion runs AFTER format substitution so # context-supplied ``[r]…[/r]`` values get rewritten too. # Email plain-text → ``**bold**`` (most MUAs render asterisks as # bold; worst case the asterisks survive as visible decoration); # email HTML → ``…``. if body_text: body_text = _apply_inline_markup(body_text, channel="email_text") if body_html: body_html = _apply_inline_markup(body_html, channel="email_html") # Brand-consistent HTML for every email. When a caller already # provided ``body_html`` (the rich Day-1 onboarding mail, or an # admin-authored ``body_html`` template override) we leave it # untouched. Otherwise wrap the resolved plain text in the shared # Helix shell so the mail reads like the onboarding welcome # rather than shipping as bare text. The plain-text alternative # (``body_text``) is still sent for text-only readers — this only # *adds* the ``text/html`` part, so the send path is unchanged. if not body_html and html_source_text: try: body_html = render_email_shell( subject=subject or env.title or "", body_text=html_source_text, cta_url=absolute_url, lang=lang, ) except Exception: # noqa: BLE001 — never let chrome break a send logger.warning( "render_email_shell failed for template_key=%r; " "falling back to plain-text only", env.template_key, exc_info=True, ) body_html = None # Email subject lines render asterisks literally in most clients # and get truncated by inbox previews — keep the meaningful # tokens, drop the ``[r]/[b]`` decoration. Fix mirrors the Feishu # header sanitisation in ``_build_interactive_card``. if subject: subject = _strip_inline_markup(subject) msg = EmailMessage( to=[env.recipient.work_email], cc=list(env.cc_emails or []), subject=subject, body_text=body_text, body_html=body_html, message_id=env.email_message_id, ) send_result = await driver.send(msg) return send_result.ok, send_result.error_message # ── Retry ──────────────────────────────────────────────── async def _commit_delivery_claim( self, row: NotificationOutbox, env: NotificationEnvelope, ) -> UUID: """Commit the non-transactional provider-I/O intent. The caller must hold the outbox row lock and have re-checked its status. This commit deliberately happens before any Feishu/SMTP call; it releases the lock, but the persisted ``delivering`` state excludes every other worker and admin retry path. """ token = uuid4() now = datetime.now(UTC) delivered = set(row.delivered_channels or []) claimed_channels = [c for c in env.channels if c not in delivered] keys = dict( row.provider_delivery_keys or _stable_provider_delivery_keys(row.id) ) row.status = STATUS_DELIVERING row.delivery_claim_token = token row.delivery_claimed_at = now row.delivery_lease_expires_at = now + DELIVERY_LEASE row.delivery_claimed_channels = claimed_channels row.delivery_claim_count = (row.delivery_claim_count or 0) + 1 row.provider_delivery_keys = keys row.next_attempt_at = None env.delivery_idempotency_key = keys["feishu"] env.email_message_id = keys["email"] await self.db.commit() return token async def _mark_delivery_uncertain( self, row_id: UUID, token: UUID, *, reason: str, ) -> None: """Fence a claimed attempt whose provider outcome is unknowable.""" from sqlalchemy import select row = ( await self.db.execute( select(NotificationOutbox) .where( NotificationOutbox.id == row_id, NotificationOutbox.status == STATUS_DELIVERING, NotificationOutbox.delivery_claim_token == token, ) .with_for_update() ) ).scalars().first() if row is None: await self.db.rollback() return row.status = STATUS_DELIVERY_UNCERTAIN self._replace_with_delivery_role_fallback(row) row.next_attempt_at = None row.last_error = reason[:4000] await self.db.commit() async def _deliver_with_durable_claim( self, row: NotificationOutbox, env: NotificationEnvelope, *, trace_id: str, actor_user_id: UUID | None, ) -> NotificationResult: """Commit claim, perform I/O, then commit the provider outcome. Any exception or final-commit failure is converted to ``delivery_uncertain`` in a fresh transaction. If the process dies too abruptly to execute that handler, stale-lease reconciliation in :meth:`retry_pending_outbox` performs the same transition. """ # ⚠️ This method COMMITS. A commit inside an open SAVEPOINT block # (``best_effort`` / ``session.begin_nested()``) closes the # transaction the block still owns, and every statement after it — # including the ``delivery_uncertain`` handler below — dies with # "Can't operate on closed transaction inside context manager". # The row is then stranded in ``delivering`` with attempts=0 until # the stale-lease reconciler flips it to ``delivery_uncertain``, # which by design refuses to auto-resend: the notification is never # sent and an operator has to reconcile it by hand. That is # exactly how EVERY ``offboarding.handover_assigned`` push between # 2026-08-13 and 2026-08-28 was lost (#2672 wrapped delivery in # ``best_effort``; #2414 had made delivery commit). Refuse BEFORE # the claim so the row stays PENDING and the hourly # ``notification.outbox_retry`` worker delivers it on its own # session — a loud, self-healing failure instead of a silent, # permanent one. Enqueue inside the savepoint, deliver outside it; # :func:`deliver_enqueued` is the ready-made shape. if self.db.in_nested_transaction(): raise RuntimeError( "NotificationService.deliver() commits and cannot run " "inside a SAVEPOINT (best_effort / begin_nested). Enqueue " "inside the savepoint, deliver after it — see " "app.notifications.deliver_enqueued." ) row_id = row.id token = await self._commit_delivery_claim(row, env) try: # If authorization fails, commit the amount-free replacement # before provider I/O. This preserves #2407's inbox guarantee # across #2408's send-success/local-commit crash boundary. An # authorized path does not commit here: its role-row lock remains # held through provider send and the final outcome commit, so a # concurrent revocation cannot slip into the checked interval. if await self._apply_delivery_role_guard(row, env): await self.db.commit() result = await self._deliver( row, env, trace_id=trace_id, actor_user_id=actor_user_id, ) row.delivery_claim_token = None row.delivery_claimed_at = None row.delivery_lease_expires_at = None row.delivery_claimed_channels = None await self.db.commit() return result except BaseException: await self.db.rollback() try: await self._mark_delivery_uncertain( row_id, token, reason=( "provider delivery outcome uncertain — worker failed " "after durable claim; automatic resend refused" ), ) except Exception: # noqa: BLE001 await self.db.rollback() logger.exception( "failed to persist uncertain delivery outcome outbox=%s", row_id, ) raise async def _reconcile_stale_delivery_claims( self, *, now: datetime, limit: int, ) -> dict[str, int]: """Recover expired claims without blindly repeating side effects. A Feishu-only claim can safely re-enter the queue while its stable UUID is still within the documented provider de-duplication window. Email, mixed-channel, and older Feishu claims become operator-visible ``delivery_uncertain`` rows and never auto-retry. """ from sqlalchemy import or_, select ids = list((await self.db.execute( select(NotificationOutbox.id) .where( NotificationOutbox.status == STATUS_DELIVERING, or_( NotificationOutbox.delivery_lease_expires_at.is_(None), NotificationOutbox.delivery_lease_expires_at <= now, ), ) .order_by(NotificationOutbox.delivery_lease_expires_at.asc()) .limit(limit) )).scalars().all()) result = {"recovered_feishu": 0, "uncertain": 0} for row_id in ids: row = (await self.db.execute( select(NotificationOutbox) .where( NotificationOutbox.id == row_id, NotificationOutbox.status == STATUS_DELIVERING, or_( NotificationOutbox.delivery_lease_expires_at.is_(None), NotificationOutbox.delivery_lease_expires_at <= now, ), ) .with_for_update(skip_locked=True) )).scalars().first() if row is None: continue # A crash may happen immediately after the durable claim, before # delivery-time authorization, or after an authorized provider # send but before the outcome commit. Both are uncertain states; # retain the amount-free fallback in the personal inbox. self._replace_with_delivery_role_fallback(row) channels = list(row.delivery_claimed_channels or []) claimed_at = row.delivery_claimed_at if claimed_at is not None and claimed_at.tzinfo is None: # MySQL/TiDB DATETIME may round-trip without tzinfo even # though the SQLAlchemy column is declared timezone-aware. claimed_at = claimed_at.replace(tzinfo=UTC) age = ( now - claimed_at if claimed_at is not None else FEISHU_SAFE_RETRY_WINDOW ) if channels and set(channels) == {"feishu"} and ( age < FEISHU_SAFE_RETRY_WINDOW ): row.status = STATUS_FAILED row.next_attempt_at = None row.last_error = ( "stale Feishu claim recovered inside provider UUID " "de-duplication window" ) row.delivery_claim_token = None row.delivery_claimed_at = None row.delivery_lease_expires_at = None row.delivery_claimed_channels = None result["recovered_feishu"] += 1 else: row.status = STATUS_DELIVERY_UNCERTAIN row.next_attempt_at = None row.last_error = ( "provider delivery outcome uncertain after stale claim; " "automatic resend refused — reconcile by stable provider " "key, then explicitly resend or cancel" ) result["uncertain"] += 1 await self.db.commit() return result async def retry_pending_outbox( self, *, limit: int = 100, ) -> dict[str, int]: """Process outbox rows whose ``next_attempt_at`` is due. Called by the hourly ``notification.outbox_retry`` scheduled job. Returns a summary dict so the job's ``result_summary`` shows "retried: 12, delivered: 8, dead: 1". """ from sqlalchemy import select now = datetime.now(UTC) stale = await self._reconcile_stale_delivery_claims( now=now, limit=limit, ) # NOTE: ``NULLS FIRST`` is Postgres-only. On MySQL we fake it # with a compound order_by — NULLs sort first naturally here # because ``is_null_order`` == 0 on NULL vs 1 on NOT NULL. is_null_order = NotificationOutbox.next_attempt_at.is_(None).desc() # Which rows are due lives in ``retry_due_clause`` — one clause, # used by both the id scan and the per-row re-check below, so the # two can never drift apart. A ``partial`` row qualifies ONLY # when ``_deliver`` armed its ``next_attempt_at`` (transient # failure on the still-missing channel, attempts remaining); a # partial row with NULL is terminal and stays out. See that # function for why the distinction matters. # # Claim CANDIDATE IDS first (no lock held across the batch), then # process each row in its OWN locked transaction below — see the # loop. Rationale (double-send hardening): # * ``_deliver`` performs a NON-idempotent external send (Feishu # DM / email). The old code selected the whole batch # ``FOR UPDATE`` and deferred the single ``commit`` to the # caller, so a mid-batch crash (OOM / redeploy) rolled back # the terminal status of every already-sent row → the next # tick re-selected and RE-DELIVERED all of them. # * A naive per-row ``commit`` on a batch-wide ``FOR UPDATE`` # select is ALSO wrong: the first commit releases the locks on # every still-unprocessed row, exposing them to a concurrent # drain (the hourly cron and the endpoint-spawned # ``drain_due_outbox_in_background`` use independent sessions # and can overlap). # Re-locking + re-checking status per row (below) makes each send # durable before the next one and lets a concurrent worker's # ``SKIP LOCKED`` + status re-check cleanly skip rows we're mid- # flight on, closing both windows. candidate_ids = ( await self.db.execute( select(NotificationOutbox.id) .where(retry_due_clause(now)) .order_by(is_null_order, NotificationOutbox.next_attempt_at.asc()) .limit(limit) ) ).scalars().all() # Round-51 R2-P0-1: an outbox row claimed when a recipient was # active will resurface 5×–10× over a 24h+ backoff. Between the # original claim and any retry, the recipient may have # offboarded — the Feishu open_id resolver typically still # returns a valid open_id (personal Feishu account not # auto-revoked) and email keeps flowing. Refuse to redispatch # to departed / deleted recipients; mark ``dead`` so the # admin outbox surfaces *why*. from app.auth.service import is_effectively_departed summary = { # Rows claimed by this batch — the drain loop's exit signal. # ``retried`` alone undercounts: a batch that goes 100% # down the direct-DEAD branches below has retried=0 but # did real work, and more due rows may remain. "picked": 0, "retried": 0, "delivered": 0, "still_pending": 0, "dead": 0, # An armed ``partial`` row the worker refused to redispatch # (recipient departed / person row gone) and settled back to # terminal ``partial``. Counted apart from ``dead`` because # the recipient WAS notified on the other channel — folding # it into ``dead`` would make the job summary claim a # delivery was lost. "frozen_partial": 0, "skipped_left": 0, "recovered_feishu": stale["recovered_feishu"], "uncertain": stale["uncertain"], } for row_id in candidate_ids: # Re-fetch this row under a per-row lock and RE-CHECK it is # still pending/failed and due. ``SKIP LOCKED`` means a row a # concurrent drain is mid-flight on is skipped here; the # status re-check means a row that concurrent worker already # delivered (between the id-scan and now) is skipped too. row = ( await self.db.execute( select(NotificationOutbox) .where( NotificationOutbox.id == row_id, retry_due_clause(now), ) .with_for_update(skip_locked=True) ) ).scalars().first() if row is None: continue summary["picked"] += 1 # A password-bearing row (Google first-login temp password, # recorded out-of-band for IT fallback) must NEVER be # re-delivered — ``_deliver`` sends to recipient.work_email + # Feishu, the exact channels the original send avoided. These are # recorded terminal (DEAD) already, but guard here too: mark dead # + skip so a FAILED one (personal-email bounce) can't be # auto-re-fired by this cron / the batch-retry drain. if row.template_key in SECRET_TEMPLATE_KEYS: if row.status != STATUS_DEAD: row.status = STATUS_DEAD row.last_error = ( "password-bearing notification — auto-retry refused " "(reset in the Google admin console instead)" ) await self.db.commit() summary["dead"] += 1 continue recipient: IdentityPerson | None = None if row.recipient_person_id: recipient = await self.db.get(IdentityPerson, row.recipient_person_id) # These two branches refuse to redispatch and settle the row # here. ``refused_redispatch_terminal`` decides between # ``dead`` and ``partial`` — a row whose other channel # already landed must not be relabelled as never-sent — and # the critical-alert backstop only fires for a row that # reached nobody, since a critical alert that arrived by # email is not a lost notification. if recipient is None: row.status = refused_redispatch_terminal(row) row.next_attempt_at = None row.last_error = "recipient person not found" # Post-review fix (2026-06-10): the direct-DEAD branches # bypass ``_deliver`` — the critical-alert backstop must # still fire here, or a critical alert whose recipient # left silently violates the guaranteed-delivery # contract. if row.status == STATUS_DEAD: if is_critical_template(row.template_key): await self._fire_critical_fallback( row, errors=[row.last_error], trace_id=str(row.id), ) summary["dead"] += 1 else: summary["frozen_partial"] += 1 await self.db.commit() continue if recipient.deleted_at is not None or is_effectively_departed(recipient): row.status = refused_redispatch_terminal(row) row.next_attempt_at = None row.last_error = "recipient departed / deleted — redispatch refused" if row.status == STATUS_DEAD: if is_critical_template(row.template_key): await self._fire_critical_fallback( row, errors=[row.last_error], trace_id=str(row.id), ) summary["dead"] += 1 else: summary["frozen_partial"] += 1 await self.db.commit() summary["skipped_left"] += 1 continue env = NotificationEnvelope( recipient=recipient, title=row.title, body_text=row.body_text, channels=list(row.channels or []), template_key=row.template_key, context=dict(row.context or {}), body_html=row.body_html, url=row.url, related_resource_type=row.related_resource_type, related_resource_id=row.related_resource_id, # Restore the frozen CC list — without it a retried Day-1 # welcome email re-sends without its HRBP/HR Ops CC. cc_emails=list(row.cc_emails or []), # Restore the frozen rich-card payload so a retried Feishu # card re-renders with its buttons / stats / sections # intact instead of falling back to the plain title+url # "Open in Helix" button (the vanished "一键续期" button). **_deserialize_card_payload(row.card_payload), ) # audit.log requires a UUID-shape trace_id; use the row id # directly so admin audit trails can link back to this retry. await self._deliver_with_durable_claim( row, env, trace_id=str(row.id), actor_user_id=None, ) summary["retried"] += 1 if row.status == STATUS_DELIVERED: summary["delivered"] += 1 elif row.status == STATUS_DEAD: summary["dead"] += 1 else: summary["still_pending"] += 1 return summary async def cancel_outbox( self, outbox_id: UUID, *, actor_user_id: UUID | None, reason: str | None = None, ) -> NotificationOutbox: """Admin-initiated stop: flip the row to ``dead`` so the retry handler ignores it on the next tick. Idempotent — already-delivered rows return unchanged, as do already-ACKNOWLEDGED dead rows. Anything retryable (pending / partial / failed) gets ``status='dead'`` + a "cancelled by {actor}" prefix on ``last_error``; an un-acknowledged dead row keeps ``status='dead'`` and only gains the prefix (the acknowledgement marker the critical-dead counters exclude). Audit trail is important here — someone silencing a flight of alerts is a sensitive operation, so we log the before/after status + the cancel reason. """ # Post-review fix (2026-06-10): row lock so a cancel racing an # in-flight delivery can't overwrite a just-committed DELIVERED # with dead — the terminal-status check below must read the # settled value, not a pre-delivery snapshot. # ``populate_existing`` (review round-2 P1): when the object is # already in the session's identity map (the batch-cancel route # pre-loads it without a lock), a bare locked get blocks on the # concurrent transaction but returns the CACHED attributes — the # terminal check would still read the pre-delivery snapshot. # populate_existing refreshes the object from the locked read. row = await self.db.get( NotificationOutbox, outbox_id, with_for_update=True, populate_existing=True, ) if row is None: from app.core.exceptions import NotFoundError raise NotFoundError("NotificationOutbox", str(outbox_id)) previous_status = row.status if row.status in (STATUS_DELIVERED, STATUS_DELIVERING): # Nothing to do — report as noop so the UI can de-dupe # duplicate button presses without a confusing error. return row if row.status == STATUS_DEAD and ( row.last_error or "" ).startswith(ACK_NOTE_PREFIX): # Already acknowledged — idempotent noop. return row # DEAD rows are no longer a hard noop (2026-06-10, dead-letter # ack): cancelling a dead row prepends the standard # ``cancelled by admin`` note — the explicit "a human looked # at this and consciously closed it" marker that the # critical-dead banner / counters exclude. Without it, a dead # ops-critical alert kept the banner red forever with no UI # path to acknowledge it (retry was the only exit, and a # structurally-failed send never delivers). Status stays # ``dead`` — the row WAS never delivered; only the bookkeeping # changes. cancel_note = f"{ACK_NOTE_PREFIX} ({actor_user_id or 'system'})" if reason: cancel_note = f"{cancel_note}: {reason}" # Prepend so the original last error (if any) is still # recoverable for forensics. combined = ( f"{cancel_note} | prev: {row.last_error}" if row.last_error else cancel_note ) row.last_error = combined[:4000] row.status = STATUS_DEAD row.next_attempt_at = None await self.db.flush() await self.audit.log( trace_id=str(row.id), actor_type="user" if actor_user_id else "system", actor_user_id=actor_user_id, action="notification.cancelled", status="success", resource_type="notification_outbox", resource_id=str(row.id), detail={ "previous_status": previous_status, "attempts": row.attempts, "max_attempts": row.max_attempts, "reason": reason, "title": row.title, "recipient_person_id": str(row.recipient_person_id) if row.recipient_person_id else None, }, ) return row async def reconcile_uncertain_outbox( self, outbox_id: UUID, *, delivered: bool, actor_user_id: UUID, note: str | None = None, ) -> NotificationOutbox: """Record an operator's provider-side reconciliation decision. ``delivered=True`` closes the claimed channels as delivered. ``delivered=False`` means the provider was checked and confirmed no side effect; only then does the same row safely re-enter automatic retry. This is deliberately unavailable for ordinary failed rows. """ row = await self.db.get( NotificationOutbox, outbox_id, with_for_update=True, populate_existing=True, ) if row is None: from app.core.exceptions import NotFoundError raise NotFoundError("NotificationOutbox", str(outbox_id)) if row.status != STATUS_DELIVERY_UNCERTAIN: return row previous_error = row.last_error claimed = list(row.delivery_claimed_channels or row.channels or []) if delivered: delivered_channels = list(row.delivered_channels or []) for channel in claimed: if channel not in delivered_channels: delivered_channels.append(channel) row.delivered_channels = delivered_channels row.failed_channels = [ channel for channel in (row.failed_channels or []) if channel not in claimed ] or None row.status = ( STATUS_DELIVERED if all(c in delivered_channels for c in (row.channels or [])) else STATUS_PARTIAL ) row.delivered_at = datetime.now(UTC) row.next_attempt_at = None decision = "provider_confirmed_delivered" else: row.status = STATUS_FAILED row.next_attempt_at = None decision = "provider_confirmed_not_delivered" row.last_error = ( f"reconciled by {actor_user_id}: {decision}" f"{f' — {note}' if note else ''}" f" | previous: {previous_error or ''}" )[:4000] row.delivery_claim_token = None row.delivery_claimed_at = None row.delivery_lease_expires_at = None row.delivery_claimed_channels = None await self.db.flush() await self.audit.log( trace_id=str(row.id), actor_type="user", actor_user_id=actor_user_id, action="notification.delivery_reconciled", status="success", resource_type="notification_outbox", resource_id=str(row.id), detail={ "decision": decision, "claimed_channels": claimed, "provider_delivery_keys": dict(row.provider_delivery_keys or {}), "note": note, }, ) return row async def revive_outbox_rows( self, ids: list[UUID], *, actor_user_id: UUID | None, ) -> dict[str, Any]: """Bulk re-queue: put the given rows back into the retry pool. Used by the admin "批量重试" button — the dead-letter triage path. Each non-terminal-success row gets: * ``status=FAILED`` (the retry worker's pickup state), * ``next_attempt_at=now`` (due immediately), * ``max_attempts`` bumped to ``attempts + 1`` when exhausted — one fresh attempt per revive, not an infinite license. Rows are NOT delivered inline — delivery happens in the retry worker (kicked immediately by the router via a background drain), which applies the departed-recipient and channel-skip guards uniformly. One summary audit row records the batch; the per-row ``notification.dispatched`` audit comes from the actual delivery attempt. Returns ``{"revived": n, "skipped_terminal": n, "skipped_locked_or_missing": n}``. """ from sqlalchemy import select # De-dup while preserving order — a repeated id must not count # twice anywhere (review round-2 P2: the old arithmetic counted # duplicates as "terminal"). ids = list(dict.fromkeys(ids)) if not ids: return { "revived": 0, "skipped_terminal": 0, "skipped_locked_or_missing": 0, } # Post-review fix (2026-06-10): take row locks, skipping rows a # concurrent retry worker is delivering RIGHT NOW. Without the # lock, revive's snapshot read could observe a PENDING row, # block on the worker's lock at flush, and then overwrite the # worker's freshly-committed DELIVERED back to FAILED — the # next drain would re-deliver a duplicate. Locked / unknown ids # are reported as their own bucket (review round-2: they were # lumped into "terminal", which made the admin toast lie); the # admin can re-click once the in-flight delivery settles. rows = ( await self.db.execute( select(NotificationOutbox) .where(NotificationOutbox.id.in_(ids)) .with_for_update(skip_locked=True) ) ).scalars().all() revived = 0 skipped_terminal = 0 skipped_locked_or_missing = len(ids) - len(rows) for row in rows: if row.status in ( STATUS_DELIVERED, STATUS_DELIVERING, STATUS_DELIVERY_UNCERTAIN, ): skipped_terminal += 1 continue if row.template_key in SECRET_TEMPLATE_KEYS: # Password rows are intentionally terminal — never revive # into the retry pool (the drain would try to re-send the # body to work_email/Feishu; the loop guard would re-kill it, # a pointless DEAD→FAILED→DEAD flap). Reset in the Google # console for fallback instead. skipped_terminal += 1 continue row.status = STATUS_FAILED # NULL = "due immediately" per the model contract. Setting # ``now`` instead is subtly racy: MySQL DATETIME rounds # fractional seconds UP, so a value written at T+0.7s # stores as T+1s and a drain pass within the same second # sees the row as not-yet-due. row.next_attempt_at = None if (row.attempts or 0) >= (row.max_attempts or 0): row.max_attempts = (row.attempts or 0) + 1 revived += 1 await self.db.flush() await self.audit.log( trace_id="", actor_type="user" if actor_user_id else "system", actor_user_id=actor_user_id, action="notification.batch_revived", status="success", resource_type="notification_outbox", resource_id=None, detail={ "requested": len(ids), "revived": revived, "skipped_terminal": skipped_terminal, "skipped_locked_or_missing": skipped_locked_or_missing, "ids_sample": [str(i) for i in ids[:50]], }, ) return { "revived": revived, "skipped_terminal": skipped_terminal, "skipped_locked_or_missing": skipped_locked_or_missing, } async def _resend_one( self, row: NotificationOutbox, *, actor_user_id: UUID | None, trace_id: str, ) -> dict[str, Any]: """Clone one source row into a BRAND-NEW dispatch and deliver it. This is the "再次发送" primitive and is deliberately distinct from :meth:`_deliver`-on-the-same-row (the ``retry`` path): * ``retry`` re-fires the SAME outbox row and REFUSES terminal rows (delivered / dead) — re-firing a delivered row would be a silent double-send and a spam vector if a token leaks. * ``resend`` is an EXPLICIT operator action — it leaves the original delivery record untouched and inserts a fresh outbox row (a new "已送达" record with its own audit trail), so it works on ANY source status, ``delivered`` included. The clone copies the persisted envelope verbatim (title, body_text/html, url, channels, template_key, context, provenance). When ``template_key`` resolves to a registered template the renderer re-renders from the CURRENT template + stored context (desired for "send again"); otherwise the stored title/body reproduce the original verbatim. Rich-card structure (stats / highlights / sections) is not persisted on the outbox, so a resent Feishu card renders from title + body_text + url — an accepted fidelity limit of the outbox model. Guards mirror the manual ``retry`` path: a row we cannot resolve to a live, non-departed Person is skipped (``dispatch`` needs a Person to resolve the Feishu open_id + per-recipient preferences) rather than blasting an ex-employee. Returns a per-row outcome dict; never raises for a skip. """ from app.auth.service import is_effectively_departed from app.identity.person_models import IdentityPerson source_id = str(row.id) # Refuse to resend rows whose body carries a live secret (the Google # first-login temporary password). Resend clones the body verbatim # and dispatches through the NORMAL pipeline → recipient.work_email + # Feishu — exactly the channels the original out-of-band send avoided # (the mailbox the hire can't reach yet). Re-broadcasting a (by now # stale) plaintext credential to the wrong surface is never wanted; # to re-deliver, IT resets the password in the Google console. if row.template_key in SECRET_TEMPLATE_KEYS: return { "status": "skipped", "reason": ( "password-bearing notification — resend refused (reset in " "the Google admin console instead)" ), "source_id": source_id, } if row.recipient_person_id is None: # Raw-email-only send (rare vendor case) — dispatch has no # Person to resolve channels/preferences against. return { "status": "skipped", "reason": "recipient not linked to a person — resend unavailable", "source_id": source_id, } recipient = await self.db.get(IdentityPerson, row.recipient_person_id) if recipient is None: return { "status": "skipped", "reason": "recipient not found", "source_id": source_id, } if recipient.deleted_at is not None or is_effectively_departed(recipient): return { "status": "skipped", "reason": "recipient departed / deleted — resend refused", "source_id": source_id, } # Debounce a duplicate resend of the SAME source row. A client # timeout / proxy retry / double-click re-POSTs the same ids, and # resend has no natural idempotency (it clones + sends regardless of # source status) → the recipient gets the card/email twice. A short # Redis SETNX keyed on the source row id collapses the re-fire # without ever false-skipping a DIFFERENT notification (the key is # per-source-row, not per-recipient/template). Fail-OPEN on a Redis # fault — a cache blip must not block a legitimate resend. try: from app.core.redis_client import get_redis # noqa: PLC0415 _r = get_redis() if _r is not None: if not await _r.set( f"notif:resend:{source_id}", "1", nx=True, ex=30, ): return { "status": "skipped", "reason": "duplicate resend within 30s — debounced", "source_id": source_id, } except Exception: # noqa: BLE001 — fail-open on Redis fault logger.warning( "resend debounce cache unavailable for %s — proceeding " "without dedup", source_id, ) env = NotificationEnvelope( recipient=recipient, title=row.title, body_text=row.body_text, body_html=row.body_html, url=row.url, channels=list(row.channels or []), template_key=row.template_key, context=dict(row.context) if row.context else {}, related_resource_type=row.related_resource_type, related_resource_id=row.related_resource_id, # Restore the frozen CC list (same fix as the retry path). cc_emails=list(row.cc_emails or []), # Restore the frozen rich-card payload so a resent Feishu # card keeps its buttons / stats / sections (same fix as the # retry path). **_deserialize_card_payload(row.card_payload), ) # Build the fresh outbox row here (rather than via ``dispatch``) # so we hold a handle on the new id for the audit + response. # Mirrors ``dispatch``'s outbox-INSERT-in-a-SAVEPOINT idiom so a # transient lock-wait on this flush can't poison the caller's # session. new_row = NotificationOutbox( recipient_person_id=recipient.id, recipient_email=recipient.work_email, template_key=row.template_key, title=row.title, body_text=row.body_text, body_html=row.body_html, url=row.url, context=dict(row.context) if row.context else None, card_payload=row.card_payload, cc_emails=row.cc_emails, channels=list(row.channels or []), status=STATUS_PENDING, attempts=0, max_attempts=5, related_resource_type=row.related_resource_type, related_resource_id=row.related_resource_id, ) async with self.db.begin_nested(): self.db.add(new_row) await self.db.flush() new_id = str(new_row.id) # Make the fresh intent durable before creating the provider-I/O # claim. This also releases the recipient FK lock before network # latency, preserving the fan-out/login lock-hygiene invariant. await self.db.commit() claimed_row = await self.db.get( NotificationOutbox, new_row.id, with_for_update=True, populate_existing=True, ) if claimed_row is None: # pragma: no cover - committed row cannot vanish raise RuntimeError("fresh notification outbox row disappeared") result = await self._deliver_with_durable_claim( claimed_row, env, trace_id=trace_id, actor_user_id=actor_user_id, ) await self.audit.log( trace_id=trace_id or source_id, actor_type="user" if actor_user_id else "system", actor_user_id=actor_user_id, action="notification.resent", status="success", resource_type="notification_outbox", resource_id=source_id, detail={ "source_id": source_id, "new_outbox_id": new_id, "recipient_person_id": str(recipient.id), "title": row.title, "channels": list(row.channels or []), "delivered_channels": result.delivered_channels, }, ) new_status = claimed_row.status # Commit THIS row before returning so the INSERT's FK shared-lock on # the recipient's ``identity_persons`` row is released before the # caller moves to the next row's external send. A bulk resend must # never hold a fan-out of recipient FK locks across a minutes-long # sequence of Feishu/email HTTP calls — that is exactly the # #1837/#1940 login-brownout incident class (SSO login writes the # recipient's identity FK children and would 1205 lock-wait). The # per-row commit also collapses the long batch transaction that was # driving the client-timeout → duplicate-POST resend. await self.db.commit() return { "status": "resent", "source_id": source_id, "new_outbox_id": new_id, "new_status": new_status, "delivered_channels": result.delivered_channels, "feishu_error": result.feishu_error, "email_error": result.email_error, } async def resend_outbox_row( self, outbox_id: UUID, *, actor_user_id: UUID | None, trace_id: str = "", ) -> dict[str, Any]: """Resend a single outbox row (see :meth:`_resend_one`).""" row = await self.db.get(NotificationOutbox, outbox_id) if row is None: from app.core.exceptions import NotFoundError raise NotFoundError("NotificationOutbox", str(outbox_id)) return await self._resend_one( row, actor_user_id=actor_user_id, trace_id=trace_id, ) async def resend_outbox_rows( self, ids: list[UUID], *, actor_user_id: UUID | None, trace_id: str = "", ) -> dict[str, Any]: """Bulk "再次发送": clone each row into a fresh dispatch. Unlike batch-retry (which revives the same rows and skips delivered ones), resend clones every resolvable row regardless of source status — the operator explicitly asked to send the message again. Departed / person-less rows are skipped with a per-id reason so the admin toast stays honest. Returns ``{"resent": n, "skipped": n, "new_ids": [...], "skipped_detail": [...]}``. """ from sqlalchemy import select ids = list(dict.fromkeys(ids)) if not ids: return {"resent": 0, "skipped": 0, "new_ids": [], "skipped_detail": []} rows = ( await self.db.execute( select(NotificationOutbox).where(NotificationOutbox.id.in_(ids)) ) ).scalars().all() # Snapshot ids while the instances are still fresh — ``_resend_one`` # now COMMITs per row (FK-lock-hold fix), which expires every loaded # instance. Re-``get`` each row inside the loop so attribute access # never triggers a sync lazy-load on an expired object. found_id_list = [r.id for r in rows] found_ids = set(found_id_list) resent = 0 skipped = 0 new_ids: list[str] = [] skipped_detail: list[dict[str, Any]] = [] for missing in [i for i in ids if i not in found_ids]: skipped += 1 skipped_detail.append({"source_id": str(missing), "reason": "not found"}) for uid in found_id_list: row = await self.db.get(NotificationOutbox, uid) if row is None: # Deleted between the select and now — count as missing. skipped += 1 skipped_detail.append({"source_id": str(uid), "reason": "not found"}) continue outcome = await self._resend_one( row, actor_user_id=actor_user_id, trace_id=trace_id, ) if outcome.get("status") == "resent": resent += 1 if outcome.get("new_outbox_id"): new_ids.append(outcome["new_outbox_id"]) else: skipped += 1 skipped_detail.append({ "source_id": outcome.get("source_id"), "reason": outcome.get("reason"), }) return { "resent": resent, "skipped": skipped, "new_ids": new_ids, "skipped_detail": skipped_detail, } async def _load_template(self, key: str) -> dict[str, str] | None: """Return template fields for a template key, or ``None`` when no template exists (neither platform-setting override nor a built-in default). Precedence: ``platform_settings`` row ``email_template.`` (JSON with subject / body_text / body_html and their ``_en`` variants) > built-in default in :data:`_DEFAULT_TEMPLATES`. Returning ``None`` (rather than a dummy ``{"subject": "[Helix] 通知", "body_text": "(no template configured)"}``) is deliberate — the old dummy was truthy, so :meth:`_render_for_recipient` / :meth:`_send_email` would substitute it for the envelope's own title/body and ship a useless "(no template configured)" card to production. With ``None``, every renderer falls through to the rich values the caller already put on the envelope. Found 2026-05-02 against the SLA URGENT dispatch — the SLA service composed a full body, but Feishu cards rendered the dummy. The returned dict carries every present field — including the ``_en`` variants and ``card_button_label`` / ``card_button_label_en`` — so callers can pick by recipient language without re-loading. Missing fields surface as absent keys (not empty strings) so ``_select_template_field`` can fall through cleanly. """ override = await get_platform_setting( f"email_template.{key}", default=None, session=self.db, ) builtin = _DEFAULT_TEMPLATES.get(key) if isinstance(override, dict) and override.get("subject") and override.get("body_text"): # Carry every string field through verbatim so the en # variants + button labels reach the renderer. Coerce to # str so a misformatted setting (e.g. someone wrote a # number) still renders rather than crashes the channel. cleaned = { k: str(v) for k, v in override.items() if isinstance(v, (str, int, float)) and str(v).strip() } # MERGE onto the builtin rather than replacing it. A whole-dict # replace meant a Chinese-only override (migrations 035 / 043 seed # exactly that shape: ``subject`` + ``body_text``, no ``_en``) # *deleted* the English/Japanese variants the builtin ships — so # every recipient of ``transfer_received`` / ``password_reset`` / # ``alias_created`` got Chinese regardless of their language, with # perfectly good English copy sitting unreachable in this file. # Per-field merge keeps the admin's intent (they edited the fields # they edited) without silently dropping the ones they didn't. return {**(builtin or {}), **cleaned} if builtin is None: logger.warning( "notifications._load_template: no template (override or " "builtin) for key %r — caller must rely on envelope " "title/body_text", key, ) return None return dict(builtin) async def _template_for_row( self, row: NotificationOutbox, template_cache: dict[str, dict[str, str] | None] | None, ) -> dict[str, str] | None: """``_load_template`` for an outbox row, memoised across a page.""" if not row.template_key: return None if template_cache is not None and row.template_key in template_cache: return template_cache[row.template_key] tmpl = await self._load_template(row.template_key) if template_cache is not None: template_cache[row.template_key] = tmpl return tmpl async def render_inbox_title( self, row: NotificationOutbox, *, lang: str | None, template_cache: dict[str, dict[str, str] | None] | None = None, ) -> str: """Title for the in-app bell / notification center, in ``lang``. The outbox persists ``title`` as whatever the *producer* passed on the envelope, which is a hardcoded literal at most call sites. Feishu and email never show that value — they re-render the template subject per recipient (``_render_for_recipient``) — but the bell used to read the stored column directly, so one notification could arrive with a Chinese title and an English body for the same reader. Render the template subject here on the same terms the send path uses. Fail-safe: no template_key, no registered template, no subject in any language, or a context that doesn't satisfy the format string → the stored title is returned unchanged, so the worst case is the pre-fix behaviour rather than a blank or an exception. """ stored = row.title or "" tmpl = await self._template_for_row(row, template_cache) if tmpl is None: return stored subject = _select_template_field(tmpl, "subject", lang) if not subject: return stored try: # ``_safe_format``, matching ``render_inbox_body``. A bare # ``.format`` lets AttributeError / TypeError escape — an admin who # writes ``{person_name.foo}`` or ``{n[0]}`` in a template override # would 500 the ENTIRE notification centre for every recipient of # that template, rather than degrading that one line. rendered = _safe_format(subject, _localized_context(row.context, lang)) except (KeyError, IndexError, ValueError): return stored # ``_safe_format`` leaves an unsatisfied placeholder LITERAL, which is # the right call for a body (the rest of the copy still informs) but # not for a headline: "[待处理] {initiator_name} 转让了一项资产给你" is # worse than the stored title, which at least names the person. Fall # back whenever anything failed to resolve. if _UNRESOLVED_PLACEHOLDER.search(rendered): return stored subject = rendered return _strip_inline_markup(subject).strip() or stored async def render_inbox_body( self, row: NotificationOutbox, *, lang: str | None, template_cache: dict[str, dict[str, str] | None] | None = None, ) -> str: """Plain-text body for the in-app bell / notification center. Outbox rows persist ``body_text`` as the ``(see template)`` sentinel: the real per-recipient copy is produced from the template only at Feishu/email send time and never written back, so the read surface would otherwise show the literal sentinel. Render the template body here in the recipient's language with the ``[r]``/``[b]`` emphasis markers stripped to plain text. Fail-safe: any miss (no template_key, non-sentinel body already stored, template missing, format error) returns the stored ``body_text`` — EXCEPT that the sentinel itself is never returned. A miss on a sentinel row degrades to an empty body (the title still carries the message) rather than showing the reader the literal string ``(see template)``, which is what every miss used to do. ``template_cache`` (optional) memoises ``_load_template`` across a page of rows so listing N notifications doesn't issue N template look-ups. """ stored = (row.body_text or "").strip() def _degrade() -> str: return "" if stored == _SEE_TEMPLATE_SENTINEL else (row.body_text or "") if not row.template_key or (stored and stored != _SEE_TEMPLATE_SENTINEL): return row.body_text or "" tmpl = await self._template_for_row(row, template_cache) if tmpl is None: return _degrade() body = _select_template_field(tmpl, "body_text", lang) if not body: return _degrade() fmt = _localized_context(row.context, lang) # The bell carries its own deep-link (``row.url``), so the inline # CTA line that the email channel needs is suppressed here. fmt.setdefault("cta_line", "") # ``portal_url`` is injected by the SEND path, not stored on the row — # so a template body that references it rendered a literal # "{portal_url}" in the notification centre (before the safe-format # switch it was worse: the strict format raised and the whole body # degraded to the "(see template)" sentinel). Resolve it the same way # the send path does, from the row's own deep link. if "portal_url" not in fmt: fmt["portal_url"] = await self._resolve_absolute_url(row.url) or "" try: # ``_safe_format`` (not bare ``.format``) — the same renderer the # send path uses. A strict format meant one placeholder the row's # frozen context didn't carry aborted the whole render, and the # abort path returned ``row.body_text`` … which on this branch IS # the ``(see template)`` sentinel. Users read internal jargon in # their notification centre. Rendering what we can and leaving the # one unknown field literal is strictly better. body = _safe_format(body, fmt) except (KeyError, IndexError, ValueError): body = "" rendered = _collapse_blank_lines(_strip_inline_markup(body)).strip() if not rendered: # Never leak the sentinel to a reader. An empty body is honest — # the title still carries the message — while "(see template)" # looks like a bug to the person holding the phone. return _degrade() return rendered # ── Dead-letter triage helpers (admin outbox bulk ops) ────────────── # Variable-token scrubbers for error clustering. Order matters: emails # before the generic hex pass (an email local part can look hexish), # UUIDs before bare hex. _ERR_EMAIL_RE = re.compile(r"[\w.+-]+@[\w-]+(?:\.[\w-]+)+") # Full UUIDs AND truncated dashed-hex ids (request ids cut by a column # limit, e.g. "8e1d3357-e30d-4696-b446-39b6a8") — any run of 3+ dashed # hex groups collapses to one token. _ERR_UUID_RE = re.compile( r"\b[0-9a-f]{4,}(?:-[0-9a-f]{2,}){2,}\b", re.IGNORECASE, ) _ERR_HEX_RE = re.compile(r"\b[0-9a-f]{8,}\b", re.IGNORECASE) _ERR_NUM_RE = re.compile(r"\b\d{2,}\b") _ERR_WS_RE = re.compile(r"\s+") def normalize_error_signature(err: str | None) -> str: """Collapse one ``last_error`` into a stable cluster signature. Strips the tokens that vary per-row (emails, UUIDs, request ids, timestamps) so "Feishu open_id not found for email a@x.com" and "... for email b@y.com" land in the same bucket. The signature is admin-facing — it should still read as the error it came from. """ if not err or not err.strip(): return "(no error recorded)" sig = err.strip()[:300] sig = _ERR_EMAIL_RE.sub("", sig) sig = _ERR_UUID_RE.sub("", sig) sig = _ERR_HEX_RE.sub("", sig) sig = _ERR_NUM_RE.sub("", sig) sig = _ERR_WS_RE.sub(" ", sig) return sig[:160] async def compute_error_clusters( db: AsyncSession, *, statuses: list[str], days: int | None = None, scan_cap: int = 20000, max_clusters: int = 50, ids_per_cluster: int = 500, exclude_acked: bool = False, ) -> dict[str, Any]: """Group outbox rows by normalised ``last_error`` signature. Scans the newest ``scan_cap`` rows matching the status filter (and optional created-at window), clusters in Python — the variable-token scrub can't be expressed portably in SQL — and returns the top clusters with enough row ids for the bulk retry / cancel buttons to act on a cluster directly. ``exclude_acked=True`` drops dead rows an admin already acknowledged (``last_error`` starts with :data:`ACK_NOTE_PREFIX`) — they are closed history, and with thousands of them the scan cap fills with handled storms before any live problem makes it into a cluster. """ from sqlalchemy import or_, select stmt = ( select( NotificationOutbox.id, NotificationOutbox.status, NotificationOutbox.template_key, NotificationOutbox.last_error, ) .where(NotificationOutbox.status.in_(statuses)) .order_by(NotificationOutbox.created_at.desc()) .limit(scan_cap) ) if exclude_acked: stmt = stmt.where(or_( NotificationOutbox.status != STATUS_DEAD, NotificationOutbox.last_error.is_(None), ~NotificationOutbox.last_error.like(f"{ACK_NOTE_PREFIX}%"), )) if days is not None: stmt = stmt.where( NotificationOutbox.created_at >= datetime.now(UTC) - timedelta(days=days) ) rows = (await db.execute(stmt)).all() clusters: dict[str, dict[str, Any]] = {} for row_id, status, template_key, last_error in rows: sig = normalize_error_signature(last_error) bucket = clusters.get(sig) if bucket is None: bucket = clusters[sig] = { "signature": sig, "sample_error": (last_error or "")[:300], "count": 0, "by_status": {}, "by_template": {}, "ids": [], } bucket["count"] += 1 bucket["by_status"][status] = bucket["by_status"].get(status, 0) + 1 tkey = template_key or "(none)" bucket["by_template"][tkey] = bucket["by_template"].get(tkey, 0) + 1 if len(bucket["ids"]) < ids_per_cluster: bucket["ids"].append(str(row_id)) ordered = sorted( clusters.values(), key=lambda c: c["count"], reverse=True, )[:max_clusters] # Trim by_template to the top 5 per cluster — the admin panel shows # a tag list, not the full distribution. for c in ordered: c["by_template"] = dict( sorted( c["by_template"].items(), key=lambda kv: kv[1], reverse=True, )[:5] ) return { "clusters": ordered, "scanned": len(rows), "scan_capped": len(rows) >= scan_cap, } async def drain_due_outbox_in_background( *, batch_limit: int = 100, max_batches: int = 20, ) -> None: """Open an owned session and drain due retry rows until none are left (or the batch cap is hit — the hourly cron mops up any rest). Spawned by the batch-retry endpoint so a bulk revive starts delivering immediately instead of waiting for the next cron tick. Each batch commits independently; a mid-drain crash loses at most one uncommitted batch, which the cron re-claims (delivery is guarded by ``FOR UPDATE SKIP LOCKED`` + per-row status checks). """ from app.core.database import async_session_factory for _ in range(max_batches): async with async_session_factory() as session: svc = NotificationService(session) summary = await svc.retry_pending_outbox(limit=batch_limit) await session.commit() # Post-review fix (2026-06-10): exit on "claimed nothing", not # "delivered nothing" — a batch that went 100% down the # direct-DEAD branches (departed recipients) has retried=0 but # processed a full batch, and more due rows may remain. if not summary.get("picked"): break async def deliver_enqueued( db: AsyncSession, queued: Sequence[tuple[NotificationOutbox, NotificationEnvelope]], *, trace_id: str, actor_user_id: UUID | None = None, log: Any = logger, ) -> int: """Commit the caller's enqueue, then deliver each row outside any SAVEPOINT. The shape every "enqueue inside ``best_effort``, push afterwards" call site needs. ``NotificationService.deliver`` commits (durable provider- I/O claim), so it MUST NOT be called from inside a savepoint block — see the guard in ``_deliver_with_durable_claim``. Collect ``(outbox, env)`` pairs inside the savepoint, call this after it: queued: list[tuple[Any, Any]] = [] async with best_effort(self.db, label="…", log=logger) as outcome: ... queued.append((await notify.enqueue(env), env)) if outcome.error is None: await deliver_enqueued(self.db, queued, trace_id=trace_id) The commit is what makes the enqueue (and any idempotency receipt written beside it) durable BEFORE delivery is attempted — the ordering #2672 locked in. A per-row vendor failure is logged and skipped: the row stays PENDING/FAILED and the hourly ``notification.outbox_retry`` worker redelivers it. Returns the number of rows delivered without raising. If an OUTER caller still holds a savepoint (some cron paths wrap a whole notify helper in ``best_effort``), inline delivery is skipped with a warning rather than attempted — the rows are durable and the retry worker owns them. Late beats lost. """ if not queued: return 0 if db.in_nested_transaction(): # A SAVEPOINT is still open — usually because an OUTER caller wrapped # this whole code path in ``best_effort`` (the daily # ``reconcile_default_handovers`` does exactly that around # ``notify_handover_assigned``). Inline delivery is impossible here: # the commit it needs would close the block's transaction. Degrade # instead of raising — every row is already durably PENDING, so the # hourly ``notification.outbox_retry`` worker delivers it on its own # session within the hour. Late, not lost. log.warning( "deliver_enqueued: caller still holds a SAVEPOINT — leaving %d " "row(s) to notification.outbox_retry (templates: %s)", len(queued), sorted({env.template_key or "?" for _, env in queued}), ) return 0 await db.commit() svc = NotificationService(db) delivered = 0 for outbox, env in queued: # Read the id BEFORE delivering: the rollback in the error branch # expires every instance, and touching ``outbox.id`` afterwards # would fire a lazy refresh from a sync context (MissingGreenlet) # — an error INSIDE the error handler, which would then escape the # loop and cost the remaining recipients their message. outbox_id = outbox.id try: await svc.deliver( outbox, env, trace_id=trace_id, actor_user_id=actor_user_id, ) delivered += 1 except Exception: # noqa: BLE001 — one bad recipient must not # cost the rest of the batch theirs; the row keeps its # non-terminal status and the retry worker owns redelivery. log.exception( "failed to deliver notification outbox=%s template=%s", outbox_id, env.template_key, ) # Roll back ONLY a transaction the failure actually # deactivated (``deliver`` lets a 1213 deadlock escape its own # savepoint, which leaves the session pending-rollback and # would fail every remaining recipient with an opaque # PendingRollbackError — #2241). An unconditional rollback # would be worse than useless here: it expires every ORM # instance the CALLER still holds, so the next ``proc.id`` in # their loop fires a lazy refresh from a sync context and dies # with MissingGreenlet. trans = db.get_transaction() if trans is not None and not trans.is_active: await db.rollback() return delivered