src/monitor_app/notice_plugins.py

0001 """Push-plugin delivery for the notice router (docs/NOTICE_ROUTING.md).
0002
0003 A push plugin is a named in-process delivery mode: a subscription whose
0004 ``delivery`` value is a plugin name has each matched event handed to that
0005 plugin during the routing pass. The router drives the pass protocol:
0006 ``start_pass()`` before a plugin's first delivery of the pass,
0007 ``deliver(row, extra)`` per matched event, ``end_pass()`` after the scan.
0008
0009 Push delivery is at-most-once: a delivery failure is logged and the pass
0010 continues — the routing mark advances regardless, so a push outage never
0011 stalls buffered-pull delivery. The event remains on the log record page.
0012
0013 The first plugin is ``mattermost-live``, the #epicprod-live channel
0014 publisher moved here from the publish_epicprod_live command with its
0015 formatting unchanged. Its channel is the SysConfig
0016 ``epicprod_live_channel`` knob, re-read every pass; its event selection
0017 is the ``epicprod-live`` subscription, not code.
0018 """
0019 import logging
0020 import os
0021 import re
0022
0023 import requests
0024 from django.utils import timezone
0025
0026 logger = logging.getLogger(__name__)
0027
0028 MM_URL = os.environ.get('MATTERMOST_URL', 'chat.epic-eic.org')
0029 # Post as the dedicated 'epicprod' identity; fall back to the DISpatcher bot
0030 # token (posts then carry the DISpatcher face) until EPICPROD_LIVE_TOKEN is set.
0031 MM_TOKEN = (os.environ.get('EPICPROD_LIVE_TOKEN')
0032             or os.environ.get('MATTERMOST_TOKEN', ''))
0033 MM_TEAM = os.environ.get('MATTERMOST_TEAM', 'main')
0034
0035 DEFAULT_CHANNEL = 'epicprod-live'
0036 PASS_POST_MAX = 20      # per-pass post cap; overflow is summarized, not dropped
0037 HTTP_TIMEOUT = 15
0038
0039
0040 def _link_base():
0041     """Open-face link base so event links work for the whole
0042     collaboration: env override, else the external-face configuration
0043     point plus the production path."""
0044     env = os.environ.get('EPICPROD_LIVE_LINK_BASE')
0045     if env:
0046         return env.rstrip('/')
0047     from monitor_app.models import external_face_base_url
0048     return f"{external_face_base_url()}/prod"
0049
0050
0051 class MattermostLivePlugin:
0052     """The #epicprod-live channel as a router delivery mode."""
0053
0054     UUID_RE = re.compile(
0055         r'[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}',
0056         re.IGNORECASE)
0057
0058     def __init__(self):
0059         self.session = None
0060         self.user_id = ''
0061         self.team_id = ''
0062         self.channel_name = ''
0063         self.channel_id = ''
0064         self._posted = 0
0065         self._skipped = 0
0066
0067     # -- pass protocol -------------------------------------------------------
0068
0069     def start_pass(self):
0070         """Connect on first use, honor a channel rename, reset the cap."""
0071         self._posted = 0
0072         self._skipped = 0
0073         if self.session is None:
0074             if not MM_TOKEN:
0075                 raise RuntimeError(
0076                     "EPICPROD_LIVE_TOKEN/MATTERMOST_TOKEN not set")
0077             self.session = requests.Session()
0078             self.session.headers['Authorization'] = f'Bearer {MM_TOKEN}'
0079             self.base = f'https://{MM_URL}/api/v4'
0080             me = self._get('/users/me')
0081             self.user_id = me['id']
0082             self.team_id = self._get(f'/teams/name/{MM_TEAM}')['id']
0083             logger.info("mattermost-live plugin posting as @%s",
0084                         me.get('username'))
0085         from monitor_app.models import SysConfig
0086         channel = str(SysConfig.get_setting(
0087             'epicprod_live_channel', DEFAULT_CHANNEL) or '')
0088         if not channel:
0089             logger.warning("epicprod_live_channel is blank in SysConfig; "
0090                            "using default %r", DEFAULT_CHANNEL)
0091             channel = DEFAULT_CHANNEL
0092         if channel != self.channel_name:    # first pass or painless rename
0093             self.channel_id = self._get(
0094                 f'/teams/{self.team_id}/channels/name/{channel}')['id']
0095             self.channel_name = channel
0096             try:                # self-join (idempotent, public channel)
0097                 self.session.post(
0098                     f'{self.base}/channels/{self.channel_id}/members',
0099                     json={'user_id': self.user_id},
0100                     timeout=HTTP_TIMEOUT).raise_for_status()
0101             except Exception:
0102                 logger.warning("could not self-join #%s", channel)
0103             logger.info("publishing to #%s (%s)", channel, self.channel_id)
0104
0105     def deliver(self, row, extra):
0106         if self._posted >= PASS_POST_MAX:
0107             self._skipped += 1
0108             return
0109         self._post(self._format(row, extra))
0110         self._posted += 1
0111
0112     def end_pass(self):
0113         if self._skipped:
0114             self._post(
0115                 f"… and {self._skipped} more events this pass — see the "
0116                 f"[live view]({_link_base()}/logs/?app_name=epicprod&live=1)")
0117
0118     # -- formatting ----------------------------------------------------------
0119
0120     def _format(self, row, extra):
0121         """One readable line per event: what happened, to what, by whom,
0122         and the one explanation that matters. Machine tokens stay on the
0123         record page — UUID subjects, the emitting component, and
0124         sub-10-second timings carry nothing for a channel reader."""
0125         action = extra.get('action') or row.funcname or 'action'
0126         outcome = str(extra.get('outcome') or '')
0127         if action == 'assessment_register' and outcome == 'ok':
0128             notice = self._format_assessment(row, extra)
0129             if notice:
0130                 return notice
0131         subject_type = str(extra.get('subject_type') or '').replace('_', ' ')
0132         subject_key = str(extra.get('subject_key') or '')
0133         subject_label = str(extra.get('subject_label') or '')
0134         if subject_label:
0135             subject = subject_label
0136         else:
0137             if self.UUID_RE.search(subject_key):
0138                 subject_key = ''
0139             subject = f'{subject_type} {subject_key}'.strip()
0140         username = str(extra.get('username') or '')
0141         reason = str(extra.get('reason') or '')
0142         summary = str(extra.get('summary') or '')
0143         dur = extra.get('duration_ms')
0144         stamp = timezone.localtime(row.timestamp).strftime('%H:%M')
0145
0146         # The line's title is the most specific thing the event states
0147         # it did: its operation when it carries one, else the action.
0148         title = str(extra.get('operation') or action).replace('_', ' ')
0149         parts = [f"`{stamp}`", f"**{title}**"]
0150         if subject:
0151             parts.append(subject)
0152         if username:
0153             parts.append(f"by {username}")
0154         if outcome and outcome != 'ok':
0155             parts.append(f"⚠️ **{outcome.upper()}**")
0156         explanation = reason if (outcome and outcome != 'ok' and reason) \
0157             else summary
0158         if explanation:
0159             parts.append(explanation)
0160         if isinstance(dur, (int, float)) and dur >= 10000:
0161             parts.append(f"{dur / 1000:.1f} s")
0162         parts.append(f"[record]({_link_base()}/logs/{row.id}/)")
0163         return ' · '.join(parts)
0164
0165     def _format_assessment(self, row, extra):
0166         """Linked publication notice; never duplicate the report body."""
0167         title = ' '.join(str(extra.get('report_title') or '').split())
0168         path = str(extra.get('report_path') or '').strip()
0169         if not title or not path.startswith('/ai/assessments/'):
0170             return ''
0171         url = f"{_link_base()}/{path.lstrip('/')}"
0172         stamp = timezone.localtime(row.timestamp).strftime('%H:%M %Z')
0173         subject_type = str(extra.get('subject_type') or '').strip()
0174         subject_key = str(extra.get('subject_key') or '').strip()
0175         subject = ''
0176         if subject_type and subject_key:
0177             subject = f'{subject_type.replace("_", " ").title()} {subject_key}'
0178         elif subject_key:
0179             subject = subject_key
0180         kind = str(extra.get('assessment_kind') or '').strip().lower()
0181         if kind == 'nightly':
0182             kind = 'daily'
0183         kind_label = kind.replace('_', ' ').title()
0184         verdict = str(extra.get('verdict') or '').strip()
0185         publication = (f'{kind_label} AI assessment published'
0186                        if kind_label else 'AI assessment published')
0187         parts = [f'`{stamp}`', f'**{publication}**']
0188         if subject:
0189             parts.append(subject)
0190         if verdict:
0191             verdict_text = f'Verdict: **{verdict.capitalize()}**'
0192             standing = extra.get('verdict_standing')
0193             if isinstance(standing, dict):
0194                 prior = int(standing.get('prior_consecutive') or 0)
0195                 if prior >= 1:
0196                     verdict_text += f' (standing, {prior + 1} consecutive)'
0197             parts.append(verdict_text)
0198         parts.append(f'[record]({_link_base()}/logs/{row.id}/)')
0199         notice = f'### [{title}]({url})\n' + ' · '.join(parts)
0200         narration = ' '.join(str(extra.get('narration') or '').split())
0201         if narration:
0202             notice += f'\n{narration}'
0203         return notice
0204
0205     # -- plumbing ------------------------------------------------------------
0206
0207     def _get(self, path):
0208         r = self.session.get(self.base + path, timeout=HTTP_TIMEOUT)
0209         r.raise_for_status()
0210         return r.json()
0211
0212     def _post(self, message):
0213         r = self.session.post(self.base + '/posts', timeout=HTTP_TIMEOUT,
0214                               json={'channel_id': self.channel_id,
0215                                     'message': message})
0216         r.raise_for_status()
0217
0218
0219 PLUGINS = {'mattermost-live': MattermostLivePlugin()}