File indexing completed on 2026-07-21 08:46:27
0001 """epicprod action logging — the filterable action stream in AppLog.
0002
0003 Convention: ``app_name='epicprod'`` is the epicprod ACTION stream. Every
0004 state-changing or operationally significant action — a catalog button press,
0005 an MCP action tool, an ops-agent handler, a sweep, a submission, a report
0006 generation — records one row here, regardless of which process performed it.
0007 Process and infrastructure logs stay under their own app names; this stream
0008 answers "what happened, and how did it go", and is the primary corpus
0009 for LLM assessment and reporting.
0010
0011 ``instance_name`` names the component performing the action: 'web',
0012 'ops-agent', 'mcp', 'catalog-sync', 'submit', 'report'.
0013
0014 Structured fields live in ``extra_data``; reserved keys are ``action``,
0015 ``subject_type``, ``subject_key``, ``username``, ``outcome``, and
0016 ``duration_ms``. Additional numeric counts (rows_added=..., tasks_matched=...)
0017 are stored alongside them. Timed actions (sweeps in particular) must pass
0018 ``duration_ms``.
0019
0020 This helper writes via the ORM and serves in-process Django contexts (web
0021 tier, MCP ASGI worker, management commands). Out-of-process callers such as
0022 the ops agent record actions through the existing REST log endpoint with the
0023 same app_name and field conventions.
0024
0025 Retrieval: the ``epicprod_list_actions`` MCP tool (filtered and summarized),
0026 ``swf_list_logs(app_name='epicprod')`` (raw), and the Logs UI filtered on
0027 app_name.
0028
0029 Publication axes (log level is separate and keeps its universal meaning):
0030
0031 ``sublevel`` — the event's declared importance (high | normal | low),
0032 set at the call site, AUTHORITATIVE: changing it means changing the event.
0033 It says which humans an event reaches — high reaches everyone including
0034 email/digest audiences, normal reaches live-page watchers, low reaches only
0035 the deliberately verbose viewer.
0036
0037 ``live_default`` — the event's declared RECOMMENDATION for the live stream,
0038 a special category: "interesting to some humans, now". The effective live
0039 decision is the ``epicprod_live_policy`` override registry in SysConfig
0040 (runtime attention knob, flipped on the live-policy page) over the default.
0041
0042 A channel = its importance threshold applied to live events:
0043 ``live_stream_q(min_sublevel=...)`` is the one filter every channel uses —
0044 the deep live view takes all live events; an email digest takes live events
0045 at importance high.
0046 """
0047
0048 import logging
0049 import os
0050 import threading
0051
0052 from django.utils import timezone
0053
0054 logger = logging.getLogger(__name__)
0055
0056 EPICPROD_APP_NAME = 'epicprod'
0057
0058 LIVE_POLICY_STATE_KEY = 'epicprod_live_policy'
0059
0060 SUBLEVEL_VALUES = ('high', 'normal', 'low')
0061 SUBLEVEL_ORDER = {'high': 2, 'normal': 1, 'low': 0}
0062
0063
0064
0065
0066
0067
0068
0069
0070
0071
0072
0073 ACTION_DEFAULTS = {
0074
0075 'task_submit': {
0076 'sublevel': 'high', 'live': True,
0077 'description': "Submit a campaign task to PanDA through the prun doer "
0078 "under the production credential and record the "
0079 "returned JEDI task id.",
0080 },
0081 'evgen_task_submit': {
0082 'sublevel': 'high', 'live': True,
0083 'description': "Submit an external-EVGEN campaign task to PanDA/JEDI "
0084 "via the client API: build the task parameters, "
0085 "assemble and upload the sandbox, record the JEDI "
0086 "task id.",
0087 },
0088 'panda_task_operation': {
0089 'sublevel': 'high', 'live': True,
0090 'description': "Run a native PanDA operation on an existing JEDI "
0091 "task: raise the allowed attempt count or retry "
0092 "failed work.",
0093 },
0094 'rucio_sweep': {
0095 'sublevel': 'high', 'live': True,
0096 'description': "Refresh the JLab Rucio output snapshot for the "
0097 "current (and last) campaign and rematch produced "
0098 "RECO/FULL datasets onto each task's recorded outputs.",
0099 },
0100 'campaign_instancing': {
0101 'sublevel': 'high', 'live': True,
0102 'description': "Operator-fired campaign instancing: populate the "
0103 "target campaign's working catalog from the source "
0104 "campaign's continuing physics configurations — "
0105 "adopting ingested editions and minting planned "
0106 "ones, dispositions consumed, unresolved "
0107 "identities left to curation.",
0108 },
0109 'campaign_promoted': {
0110 'sublevel': 'high', 'live': True,
0111 'description': "Operator-clicked campaign lifecycle rotation: the "
0112 "named (producing) campaign becomes current, the "
0113 "incumbent current becomes last, the incumbent last "
0114 "becomes past.",
0115 },
0116 'past_import': {
0117 'sublevel': 'high', 'live': True,
0118 'description': "Pull the eic/epic-prod bookkeeping clone and re-run "
0119 "the idempotent FULL/RECO past-campaign output "
0120 "ingest, keeping every campaign's recorded "
0121 "production content current with what the "
0122 "production team publishes.",
0123 },
0124 'rucio_reconcile': {
0125 'sublevel': 'normal', 'live': True,
0126 'description': "Firsthand reconciliation of a producing "
0127 "campaign's catalog records against its fetched "
0128 "Rucio snapshot: known DIDs updated, unknown DIDs "
0129 "resolved to physics configurations and attached "
0130 "to existing editions or created, unresolved left "
0131 "to curation.",
0132 },
0133 'rucio_arrivals': {
0134 'sublevel': 'normal', 'live': True,
0135 'description': "New files landed in JLab Rucio since the last "
0136 "arrivals sweep, counted by campaign and location "
0137 "across all versions — the signal behind the "
0138 "derived 'producing' campaign status. Emitted only "
0139 "when something arrived.",
0140 },
0141 'rucio_arrivals_sweep': {
0142 'sublevel': 'low', 'live': False,
0143 'description': "The clockwork arrivals-sweep step itself (outcome "
0144 "and duration); the arrivals, when any, are the "
0145 "rucio_arrivals event.",
0146 },
0147 'evgen_sweep': {
0148 'sublevel': 'high', 'live': True,
0149 'description': "Assimilate the JLab Rucio EVGEN input inventory "
0150 "(epic:/EVGEN/*) and resolve each catalog request to "
0151 "the registered input dataset(s) that realize it.",
0152 },
0153 'catalog_import': {
0154 'sublevel': 'high', 'live': True,
0155 'description': "Import the production CSV manifest catalog: create "
0156 "or update campaign task records from the source rows.",
0157 },
0158 'association_sweep': {
0159 'sublevel': 'high', 'live': True,
0160 'description': "Associate recent PanDA tasks with campaign tasks; "
0161 "unmatched direct group.EIC submissions are "
0162 "auto-intaken as adopted tasks.",
0163 },
0164 'catalog_sync': {
0165 'sublevel': 'high', 'live': True,
0166 'description': "Nightly composite chain (cron 02:15): csv import, "
0167 "epic-prod past ingest, questionnaire import, "
0168 "association sweep, Rucio output snapshot, Rucio "
0169 "arrivals sweep, EVGEN assimilation, questionnaire "
0170 "automatch, match cache, progress refresh. This "
0171 "record is the catalog-freshness timestamp.",
0172 },
0173 'agent_shutdown': {
0174 'sublevel': 'high', 'live': True,
0175 'description': "Deliberate stop of the production ops agent via the "
0176 "message-bus back door; systemd leaves it stopped.",
0177 },
0178 'payload_log_fetch': {
0179 'sublevel': 'normal', 'live': False,
0180 'description': "Fetch one PanDA job's payload log tarball from Rucio "
0181 "over xrootd and extract it into the shared cache for "
0182 "the job page.",
0183 },
0184 'inventory_sync': {
0185 'sublevel': 'low', 'live': False,
0186 'description': "Refresh the monitor's ePIC production job/file "
0187 "inventory and parsed failure diagnosis for a PanDA "
0188 "job.",
0189 },
0190 'system_status_refresh': {
0191 'sublevel': 'low', 'live': False,
0192 'description': "Periodic refresh of the cached System page status "
0193 "rows for services, agents, and external monitor "
0194 "endpoints.",
0195 },
0196 'questionnaire_import': {
0197 'sublevel': 'normal', 'live': True,
0198 'description': "Import the PWG/DSC production-request questionnaire "
0199 "CSV: create or update request records.",
0200 },
0201 'questionnaire_automatch': {
0202 'sublevel': 'normal', 'live': False,
0203 'description': "LLM matching of production requests to catalog tasks "
0204 "using the full tag map; each new match logs its own "
0205 "questionnaire_match_found record, and a run with new "
0206 "matches posts one live questionnaire_new_matches line.",
0207 },
0208 'questionnaire_match_found': {
0209 'sublevel': 'normal', 'live': False,
0210 'description': "One new request-to-task match from the automatch, "
0211 "with confidence and reason; high/medium confidence "
0212 "lands accepted, low lands suggested.",
0213 },
0214 'questionnaire_new_matches': {
0215 'sublevel': 'normal', 'live': True,
0216 'description': "One-line summary of an automatch run's new "
0217 "request-to-task matches; the record lists every new "
0218 "match with confidence and reason.",
0219 },
0220 'questionnaire_match': {
0221 'sublevel': 'low', 'live': False,
0222 'description': "Rebuild the questionnaire-to-task match cache read "
0223 "by request and task pages.",
0224 },
0225 'progress_refresh': {
0226 'sublevel': 'low', 'live': False,
0227 'description': "Rebuild current-campaign progress data and its "
0228 "rendered progress table cache.",
0229 },
0230
0231 'sysconfig_edit': {
0232 'sublevel': 'high', 'live': True,
0233 'description': "Operator edit of the SysConfig system-configuration "
0234 "document on the System page.",
0235 },
0236 'live_policy_edit': {
0237 'sublevel': 'high', 'live': True,
0238 'description': "Operator change to an action's live-stream override "
0239 "on the live-policy page.",
0240 },
0241 'assessment_register': {
0242 'sublevel': 'normal', 'live': True,
0243 'description': "Register an AI assessment of a production object "
0244 "(campaign task, PanDA task, job, or queue).",
0245 },
0246 'assessment_link': {
0247 'sublevel': 'low', 'live': False,
0248 'description': "Link an existing AI assessment to a production "
0249 "object's record.",
0250 },
0251 'task_set_status': {
0252 'sublevel': 'normal', 'live': True,
0253 'description': "Campaign task lifecycle transition (draft, ready, "
0254 "submitted, completed, failed) with rule enforcement.",
0255 },
0256 'task_intake': {
0257 'sublevel': 'normal', 'live': True,
0258 'description': "Idempotent intake of a production request into a "
0259 "draft campaign task.",
0260 },
0261 'dataset_intake': {
0262 'sublevel': 'normal', 'live': True,
0263 'description': "Idempotent find-or-create of a dataset from a source "
0264 "location such as a CSV manifest.",
0265 },
0266 'task_link_input': {
0267 'sublevel': 'low', 'live': False,
0268 'description': "Link an existing dataset as a campaign task's input "
0269 "by DID.",
0270 },
0271 'dataset_propagation_set': {
0272 'sublevel': 'normal', 'live': True,
0273 'description': "Operator change of dataset propagation disposition "
0274 "(continue, hold, final) with required comment; one "
0275 "event per single or bulk action, carrying the "
0276 "changed count and the selecting filter. Carries an "
0277 "origin stamp when executing an approved AI proposal.",
0278 },
0279 'proposal_created': {
0280 'sublevel': 'normal', 'live': True,
0281 'description': "AI proposal of a dataset propagation change, pending "
0282 "human review; one event per propose call with the "
0283 "proposed count and batch.",
0284 },
0285 'proposal_denied': {
0286 'sublevel': 'normal', 'live': True,
0287 'description': "Human denial of pending AI proposals; denial memory "
0288 "prevents re-proposal until the proposer's inputs "
0289 "change.",
0290 },
0291 'proposal_expired': {
0292 'sublevel': 'normal', 'live': True,
0293 'description': "Withdrawal of pending AI proposals (recurring-scan "
0294 "heartbeat refresh or operator clear), with count.",
0295 },
0296 'proposal_deleted': {
0297 'sublevel': 'normal', 'live': False,
0298 'description': "Operator deletion of AI proposal list rows — "
0299 "housekeeping for test or noise rows; removes audit "
0300 "rows, logged with count.",
0301 },
0302 'narrative_edited': {
0303 'sublevel': 'normal', 'live': False,
0304 'description': "Expert revision of a campaign narrative document — "
0305 "a new corun-ai version of the entry.",
0306 },
0307 'narrative_commented': {
0308 'sublevel': 'normal', 'live': True,
0309 'description': "Comment posted on a campaign narrative — the "
0310 "non-intrusive contribution path beside editing.",
0311 },
0312
0313 'campaign_set_current': {
0314 'sublevel': 'high', 'live': True,
0315 'description': "Operator designation of the current campaign.",
0316 },
0317 'campaign_set_last': {
0318 'sublevel': 'high', 'live': True,
0319 'description': "Operator designation of the last (previous) campaign.",
0320 },
0321 'questionnaire_match_add': {
0322 'sublevel': 'normal', 'live': True,
0323 'description': "Operator addition of a request-to-task match on the "
0324 "request page.",
0325 },
0326 'questionnaire_match_remove': {
0327 'sublevel': 'normal', 'live': True,
0328 'description': "Operator removal of a request-to-task match.",
0329 },
0330 'category_create': {
0331 'sublevel': 'normal', 'live': True,
0332 'description': "Create a physics tag category.",
0333 },
0334 'tag_create': {
0335 'sublevel': 'low', 'live': False,
0336 'description': "Create a configuration tag (physics, evgen, simu, "
0337 "reco, or background).",
0338 },
0339 'tag_edit': {
0340 'sublevel': 'low', 'live': False,
0341 'description': "Edit a draft configuration tag.",
0342 },
0343 'tag_lock': {
0344 'sublevel': 'normal', 'live': True,
0345 'description': "Lock a configuration tag — a permanent, one-way "
0346 "transition to immutability.",
0347 },
0348 'tag_delete': {
0349 'sublevel': 'normal', 'live': True,
0350 'description': "Delete a draft configuration tag.",
0351 },
0352 'dataset_create': {
0353 'sublevel': 'normal', 'live': True,
0354 'description': "Create a dataset: a composed tag identity plus any "
0355 "sample-variant discriminator.",
0356 },
0357 'dataset_block_add': {
0358 'sublevel': 'low', 'live': False,
0359 'description': "Add the next Rucio block (.bN) to a dataset.",
0360 },
0361 'config_create': {
0362 'sublevel': 'low', 'live': False,
0363 'description': "Create a production config template.",
0364 },
0365 'config_edit': {
0366 'sublevel': 'low', 'live': False,
0367 'description': "Edit a production config template.",
0368 },
0369 'task_delete': {
0370 'sublevel': 'normal', 'live': True,
0371 'description': "Delete a campaign task.",
0372 },
0373 'assessment_quality_set': {
0374 'sublevel': 'normal', 'live': True,
0375 'description': "Operator quality review (good, poor, wrong) recorded "
0376 "on an AI assessment.",
0377 },
0378 'queues_update': {
0379 'sublevel': 'normal', 'live': True,
0380 'description': "Update PanDA queue records from the GitHub source.",
0381 },
0382 'endpoints_update': {
0383 'sublevel': 'normal', 'live': True,
0384 'description': "Update Rucio storage endpoint records from the "
0385 "GitHub source.",
0386 },
0387 }
0388
0389
0390 def action_description(action):
0391 """Plain-English one-liner for a known action id, or ''."""
0392 return (ACTION_DEFAULTS.get(str(action)) or {}).get('description', '')
0393
0394 RESERVED_KEYS = ('action', 'subject_type', 'subject_key', 'username',
0395 'outcome', 'duration_ms', 'sublevel', 'live_default')
0396
0397
0398 def log_epicprod_action(instance, action, *, subject_type='', subject_key='',
0399 username='', outcome='ok', duration_ms=None,
0400 sublevel='low', live_default=False, message='',
0401 level=logging.INFO, **counts):
0402 """Record one epicprod action in the AppLog action stream.
0403
0404 Never raises: a failure to record is logged to the module logger and the
0405 calling action proceeds — the action log must not break the action.
0406
0407 Args:
0408 instance: component performing the action ('web', 'ops-agent', 'mcp',
0409 'catalog-sync', 'submit', 'report').
0410 action: short action identifier, e.g. 'rucio_sweep', 'task_submit',
0411 'assessment_register'.
0412 subject_type: acted-on object type when there is one (canonical
0413 assessment subject types where applicable).
0414 subject_key: acted-on object key (composed name, JEDI id, queue, ...).
0415 username: human or service account driving the action.
0416 outcome: 'ok' or 'error' (conventional; free-form refinements allowed).
0417 duration_ms: measured execution time; required in spirit for sweeps
0418 and other timed operations.
0419 sublevel: the event's declared importance — 'high' (reaches
0420 everyone: news, digests, email), 'normal' (live-page watchers),
0421 'low' (verbose viewers only). AUTHORITATIVE: change it by
0422 changing the event, not at runtime.
0423 live_default: the event's declared RECOMMENDATION for the live
0424 stream ("interesting to some humans, now"). The effective live
0425 decision is the epicprod_live_policy override (runtime attention
0426 knob) over this default.
0427 message: optional human-readable one-liner; composed if omitted.
0428 level: python logging level; use logging.ERROR for failed actions.
0429 **counts: numeric counts worth recording (rows_added=..., etc.);
0430 reserved keys are ignored if passed here.
0431
0432 Returns:
0433 The created AppLog row id, or None if the write failed.
0434 """
0435 from .models import AppLog
0436
0437 if sublevel not in SUBLEVEL_VALUES:
0438 logger.warning('epicprod action %s: unknown sublevel %r, using low',
0439 action, sublevel)
0440 sublevel = 'low'
0441 extra = {
0442 'action': str(action),
0443 'outcome': str(outcome),
0444 'sublevel': sublevel,
0445 'live_default': bool(live_default),
0446 }
0447 if subject_type:
0448 extra['subject_type'] = str(subject_type)
0449 if subject_key:
0450 extra['subject_key'] = str(subject_key)
0451 if username:
0452 extra['username'] = str(username)
0453 if duration_ms is not None:
0454 try:
0455 extra['duration_ms'] = int(duration_ms)
0456 except (TypeError, ValueError):
0457 logger.warning('epicprod action %s: non-numeric duration_ms %r',
0458 action, duration_ms)
0459 for key, value in counts.items():
0460 if key not in RESERVED_KEYS:
0461 extra[key] = value
0462
0463 if not message:
0464 subject = f"{subject_type}:{subject_key}" if subject_key else ''
0465 message = ' '.join(part for part in (str(action), subject, str(outcome)) if part)
0466
0467 try:
0468 row = AppLog.objects.create(
0469 app_name=EPICPROD_APP_NAME,
0470 instance_name=str(instance),
0471 timestamp=timezone.now(),
0472 level=int(level),
0473 levelname=logging.getLevelName(int(level)),
0474 message=message,
0475 module='epicprod_logging',
0476 funcname=str(action),
0477 lineno=0,
0478 process=os.getpid(),
0479 thread=threading.get_ident(),
0480 extra_data=extra,
0481 )
0482 return row.id
0483 except Exception:
0484 logger.exception('epicprod action log write failed: %s %s',
0485 instance, action)
0486 return None
0487
0488
0489 def get_live_policy():
0490 """Return the live override registry: {action_id: bool}.
0491
0492 Stored under LIVE_POLICY_STATE_KEY in SysConfig (system configuration
0493 lives in the DB, adjustable via the live-policy page). The runtime
0494 attention knob: an action absent from the registry follows its records'
0495 live_default recommendation. Reads defensively: a missing table
0496 (pre-migration) yields an empty policy with a logged warning rather than
0497 breaking the caller.
0498 """
0499 from .models import SysConfig
0500
0501 try:
0502 policy = SysConfig.get_setting(LIVE_POLICY_STATE_KEY, {}) or {}
0503 except Exception as exc:
0504 logger.warning('live policy read failed (empty policy used): %s', exc)
0505 return {}
0506 if not isinstance(policy, dict):
0507 return {}
0508 return {str(k): bool(v) for k, v in policy.items()}
0509
0510
0511 def set_live_policy_entry(action, value, username=''):
0512 """Set (True/False) or clear (None) an action's live override."""
0513 from .models import SysConfig
0514
0515 policy = get_live_policy()
0516 key = str(action)
0517 if value is None:
0518 policy.pop(key, None)
0519 else:
0520 policy[key] = bool(value)
0521 SysConfig.update_config({LIVE_POLICY_STATE_KEY: policy}, username=username)
0522 return policy
0523
0524
0525 def live_policy_rows():
0526 """Rows for the live-policy page: every known action with its declared
0527 sublevel and live default, any live override, and the effective state.
0528
0529 Known = the ACTION_DEFAULTS catalog plus any action observed in the
0530 stream (the newest record supplies declarations for actions not yet in
0531 the catalog).
0532 """
0533 from .models import AppLog
0534
0535 policy = get_live_policy()
0536 declarations = {k: dict(v) for k, v in ACTION_DEFAULTS.items()}
0537 observed = (AppLog.objects.filter(app_name=EPICPROD_APP_NAME)
0538 .exclude(extra_data__action__isnull=True)
0539 .values_list('extra_data__action', flat=True)
0540 .distinct())
0541 for action in observed:
0542 if action and action not in declarations:
0543 latest = (AppLog.objects.filter(app_name=EPICPROD_APP_NAME,
0544 extra_data__action=action)
0545 .order_by('-id').first())
0546 extra = latest.extra_data if latest and isinstance(latest.extra_data, dict) else {}
0547 stamped = extra.get('sublevel')
0548 declarations[action] = {
0549 'sublevel': stamped if stamped in SUBLEVEL_VALUES else 'low',
0550 'live': bool(extra.get('live_default')),
0551 }
0552
0553 rows = []
0554 for action in sorted(declarations):
0555 decl = declarations[action]
0556 override = policy.get(action)
0557 effective = override if override is not None else decl['live']
0558 rows.append({
0559 'action': action,
0560 'description': decl.get('description', ''),
0561 'sublevel': decl['sublevel'],
0562 'live_default': decl['live'],
0563 'state': 'default' if override is None else ('live' if override else 'quiet'),
0564 'effective': bool(effective),
0565 })
0566 return rows
0567
0568
0569 def live_stream_q(min_sublevel=None):
0570 """Q filter selecting live action records, optionally importance-gated.
0571
0572 Live = the runtime override (epicprod_live_policy) over the record's
0573 declared live_default. A channel applies its importance threshold through
0574 min_sublevel: the deep live view passes None (all live events); an email
0575 digest passes 'high'. The single source of stream semantics for every
0576 channel.
0577 """
0578 from django.db.models import Q
0579
0580 policy = get_live_policy()
0581 force_on = [a for a, v in policy.items() if v]
0582 force_off = [a for a, v in policy.items() if not v]
0583
0584 q = Q(extra_data__live_default=True)
0585 if force_off:
0586 q &= ~Q(extra_data__action__in=force_off)
0587 if force_on:
0588 q |= Q(extra_data__action__in=force_on)
0589 if min_sublevel in SUBLEVEL_VALUES:
0590 min_rank = SUBLEVEL_ORDER[min_sublevel]
0591 ge_values = [v for v in SUBLEVEL_VALUES if SUBLEVEL_ORDER[v] >= min_rank]
0592 q &= Q(extra_data__sublevel__in=ge_values)
0593 return Q(app_name=EPICPROD_APP_NAME) & q
