File indexing completed on 2026-07-21 08:46:25
0001 """AI proposal services (AI_PROPOSALS.md).
0002
0003 The propose / decide / withdraw / delete machinery behind the AI proposal
0004 list. Proposals are frozen executable payloads; everything past authoring
0005 is deterministic. Executors stay in their domain apps — approval here
0006 dispatches to them (``pcs.services.dataset_propagation_set`` for the
0007 campaign-propagation pilot) with the approving human as ``changed_by`` and
0008 the origin stamp on the event.
0009 """
0010 import hashlib as _hashlib
0011 import json as _json
0012 import logging as _logging
0013
0014 from django.db import transaction
0015 from django.db.models import Q
0016 from django.utils import timezone as _timezone
0017
0018 _log = _logging.getLogger(__name__)
0019
0020 from pcs.models import Dataset
0021 from pcs.services import (
0022 PROPAGATION_STATES, ServiceError, dataset_propagation_set,
0023 )
0024
0025 from .models import ACTION_REF_PREFIXES, Proposal
0026
0027
0028 def parse_proposal_ref(ref):
0029 """Resolve a proposal ref ('cp-123') to its Proposal row.
0030
0031 The prefix is corroboration, not decoration: it must match the row's
0032 category, so a garbled or mis-relayed reference is refused loudly
0033 instead of deciding the wrong proposal. Raises ServiceError on any
0034 mismatch; the message names the actual row when one exists.
0035 """
0036 text = (ref or '').strip().lower()
0037 prefix, sep, num = text.partition('-')
0038 known = set(ACTION_REF_PREFIXES.values())
0039 if not sep or not num.isdigit() or prefix not in known:
0040 raise ServiceError(
0041 f'unrecognized proposal ref {ref!r} — expected '
0042 f'<prefix>-<number> with prefix in {sorted(known)}')
0043 row = Proposal.objects.filter(pk=int(num)).first()
0044 if row is None:
0045 raise ServiceError(f'no proposal {text} exists')
0046 if row.ref != text:
0047 raise ServiceError(
0048 f'ref {text} does not match proposal #{row.pk}, which is a '
0049 f'{row.action} proposal with ref {row.ref} — refusing')
0050 return row
0051
0052
0053 def _proposal_input_hash(payload, comment):
0054 blob = _json.dumps({'payload': payload, 'comment': comment}, sort_keys=True)
0055 return _hashlib.sha1(blob.encode()).hexdigest()
0056
0057
0058 def _clear_proposal_projection(name):
0059 """Remove the render projection from the record a proposal targeted."""
0060 head = (Dataset.objects
0061 .filter(composed_name=name).order_by('block_num', 'pk').first())
0062 if head is None:
0063 return
0064 metadata = dict(head.metadata or {})
0065 if 'proposal' in metadata:
0066 metadata.pop('proposal', None)
0067 head.metadata = metadata
0068 head.save(update_fields=['metadata'])
0069
0070
0071 def _refresh_catalog_table_cache():
0072 """Rebuild the current-campaign catalog table fragment after proposal
0073 activity, so the page a human reloads shows the state they just
0074 changed — a stale cached table is otherwise served indefinitely
0075 (page-load rebuild is suppressed). Failure is logged and reported in
0076 the caller's result, never raised: the decision stands regardless."""
0077 try:
0078 from pcs.models import Campaign
0079 from pcs.views import (_campaigns_with_inflow,
0080 rebuild_current_task_list_html_cache)
0081 campaigns = list(Campaign.objects.filter(lifecycle='current')
0082 .order_by('name')[:1])
0083
0084
0085 campaigns += [camp for camp, _ in _campaigns_with_inflow()]
0086 for campaign in campaigns:
0087 rebuild_current_task_list_html_cache(campaign, 'catalog')
0088 return ''
0089 except Exception as e:
0090 _log.warning('catalog table cache refresh failed: %s', e)
0091 return f'catalog table cache refresh failed: {e}'
0092
0093
0094 def _write_proposal_projection(row, head):
0095 """Write the pending proposal's render projection on its target's
0096 head row — at propose time and again when an undo returns the row
0097 to pending."""
0098 pre = row.precondition or {}
0099 metadata = dict(head.metadata or {})
0100 metadata['proposal'] = {
0101 'id': row.id,
0102 'action': row.action,
0103 'payload': row.payload,
0104 'comment': row.comment,
0105 'proposer': row.proposer,
0106 'scan_version': row.scan_version,
0107 'batch_id': row.batch_id,
0108 'prev_state': pre.get('prev_state'),
0109 'prev_replaced_by': pre.get('prev_replaced_by', ''),
0110 'proposed_at': row.created_at.isoformat(),
0111 }
0112 head.metadata = metadata
0113 head.save(update_fields=['metadata'])
0114
0115
0116 def propose_propagation(composed_names, state, comment, *, replaced_by='',
0117 proposer='', scan_version=1, batch_id='',
0118 created_by=''):
0119 """Create AI propagation proposals on dataset editions.
0120
0121 Validates exactly as ``dataset_propagation_set`` does — an unexecutable
0122 proposal is refused at birth. The canonical record is a Proposal row
0123 (frozen payload, required comment, proposer identity,
0124 ``precondition.prev_state`` staleness anchor); the target's head row
0125 (first by block then pk, the same deterministic head every propagation
0126 writer uses) carries a render projection in ``metadata['proposal']``,
0127 written here
0128 and cleared by decision or withdrawal. Skips, all counted and returned:
0129 unknown names, no-ops (already in the target state), and identities
0130 with a denied proposal-list row matching this proposal's input hash (a
0131 denied proposal never returns until its inputs change). An existing
0132 pending proposal is superseded (withdrawn) by the fresh one — the
0133 heartbeat refresh. One ``proposal_created`` action-stream event
0134 per call.
0135 """
0136 from monitor_app.epicprod_logging import log_epicprod_action
0137
0138 state = (state or '').strip()
0139 comment = (comment or '').strip()
0140 replaced_by = (replaced_by or '').strip()
0141 names = [n.strip() for n in (composed_names or []) if n and n.strip()]
0142 if state not in PROPAGATION_STATES:
0143 raise ServiceError(
0144 f'propagation state must be one of '
0145 f'{", ".join(PROPAGATION_STATES)}; got {state!r}')
0146 if not comment:
0147 raise ServiceError('comment is required on every proposal')
0148 if not names:
0149 raise ServiceError('no dataset names supplied')
0150
0151 payload = {'state': state, 'replaced_by': replaced_by}
0152 input_hash = _proposal_input_hash(payload, comment)
0153 now = _timezone.now()
0154 proposed, noop, denied, unknown = [], [], [], []
0155 with transaction.atomic():
0156 for name in names:
0157 head = (Dataset.objects
0158 .filter(composed_name=name).order_by('block_num', 'pk').first())
0159 if head is None:
0160 unknown.append(name)
0161 continue
0162 if head.propagation == state and (
0163 not replaced_by or head.replaced_by == replaced_by):
0164 noop.append(name)
0165 continue
0166 if Proposal.objects.filter(
0167 action='propagation', subject_key=name,
0168 status='denied', input_hash=input_hash).exists():
0169 denied.append(name)
0170 continue
0171
0172 Proposal.objects.filter(
0173 action='propagation', subject_key=name,
0174 status='proposed').update(status='withdrawn', decided_at=now)
0175 row = Proposal.objects.create(
0176 action='propagation',
0177 subject_type='dataset',
0178 subject_key=name,
0179 payload=payload,
0180 comment=comment,
0181 proposer=proposer or '',
0182 scan_version=scan_version,
0183 batch_id=batch_id or '',
0184 executor='service',
0185 precondition={'prev_state': head.propagation,
0186 'prev_replaced_by': head.replaced_by},
0187 input_hash=input_hash,
0188 created_by=created_by or '',
0189 )
0190 _write_proposal_projection(row, head)
0191 proposed.append(name)
0192
0193 log_epicprod_action(
0194 'web', 'proposal_created',
0195 username=created_by,
0196 sublevel='normal', live_default=True,
0197 message=(f'AI proposal: propagation -> {state} on {len(proposed)} '
0198 f'dataset(s) [{batch_id or "no batch"}]: {comment}'),
0199 proposed=len(proposed), noop=len(noop), denied=len(denied),
0200 unknown=len(unknown), state=state, comment=comment,
0201 proposer=proposer or '', batch_id=batch_id or '',
0202 scan_version=scan_version,
0203 )
0204 result = {'proposed': proposed, 'noop': noop, 'denied': denied,
0205 'unknown': unknown, 'state': state}
0206 if proposed:
0207 cache_error = _refresh_catalog_table_cache()
0208 if cache_error:
0209 result['cache_refresh_error'] = cache_error
0210 return result
0211
0212
0213 def proposal_decide(composed_names, decision, *, decided_by='',
0214 quality='', filter_state='', proposal_ids=None):
0215 """Approve or deny pending AI proposals.
0216
0217 Selection by dataset composed names (the catalog and compose surfaces)
0218 and/or by proposal-list row ids (the AI proposal list page). Approval
0219 revalidates each proposal against current state (the
0220 ``precondition.prev_state`` anchor): a record that moved since the
0221 proposal saw it is marked stale and withdrawn from the record, never
0222 re-interpreted. Valid approvals execute through
0223 ``dataset_propagation_set`` — the identical call an operator makes by
0224 hand — grouped by identical (state, replaced_by, comment) so a family
0225 batch is one call and one origin-stamped event; the approving human is
0226 ``changed_by`` and the executed proposal rows record the event's log
0227 id. Denial marks the proposal row (denial memory is the proposal
0228 list); one ``proposal_denied`` event per call. ``quality``
0229 optionally tags the decision with the shared review vocabulary
0230 (wrong | poor | ok | good) — 'wrong' is the one-tap miscalibration
0231 signal that weighs against the proposer's track record.
0232 """
0233 from monitor_app.epicprod_logging import log_epicprod_action
0234
0235 decision = (decision or '').strip()
0236 quality = (quality or '').strip()
0237 names = [n.strip() for n in (composed_names or []) if n and n.strip()]
0238 ids = [int(i) for i in (proposal_ids or [])]
0239 if decision not in ('approve', 'deny'):
0240 raise ServiceError(f"decision must be 'approve' or 'deny'; "
0241 f"got {decision!r}")
0242 if quality and quality not in dict(Proposal.QUALITY_CHOICES):
0243 raise ServiceError(
0244 f"quality must be one of "
0245 f"{', '.join(dict(Proposal.QUALITY_CHOICES))}; got {quality!r}")
0246 if not names and not ids:
0247 raise ServiceError('no dataset names or proposal ids supplied')
0248 if not decided_by:
0249 raise ServiceError('an authenticated decider is required')
0250
0251 pending = Proposal.objects.filter(action='propagation', status='proposed')
0252 selector = Q()
0253 if names:
0254 selector |= Q(subject_key__in=names)
0255 if ids:
0256 selector |= Q(pk__in=ids)
0257 rows = list(pending.filter(selector))
0258 found_names = {r.subject_key for r in rows}
0259 no_proposal = [n for n in names if n not in found_names]
0260
0261 now = _timezone.now()
0262 stale, denied, approved = [], [], []
0263 groups = {}
0264 with transaction.atomic():
0265 for row in rows:
0266 head = (Dataset.objects
0267 .filter(composed_name=row.subject_key)
0268 .order_by('block_num', 'pk').first())
0269 pre = row.precondition or {}
0270 current = head.propagation if head else None
0271 record_moved = current != pre.get('prev_state')
0272 if not record_moved and head is not None and 'prev_replaced_by' in pre:
0273 record_moved = head.replaced_by != pre['prev_replaced_by']
0274 if record_moved:
0275 row.status = 'stale'
0276 row.decided_by = decided_by
0277 row.decided_at = now
0278 row.save(update_fields=['status', 'decided_by', 'decided_at'])
0279 _clear_proposal_projection(row.subject_key)
0280 stale.append(row.subject_key)
0281 continue
0282 if decision == 'deny':
0283 row.status = 'denied'
0284 row.quality = quality
0285 row.decided_by = decided_by
0286 row.decided_at = now
0287 row.save(update_fields=['status', 'quality', 'decided_by',
0288 'decided_at'])
0289 _clear_proposal_projection(row.subject_key)
0290 denied.append(row.subject_key)
0291 continue
0292 payload = row.payload or {}
0293 key = (payload.get('state', ''), payload.get('replaced_by', ''),
0294 row.comment)
0295 groups.setdefault(key, {'rows': [], 'origin': {
0296 'proposer': row.proposer,
0297 'scan_version': row.scan_version,
0298 'batch_id': row.batch_id,
0299 'proposed_at': row.created_at.isoformat(),
0300 }})['rows'].append(row)
0301
0302 for (state, replaced_by, comment), group in groups.items():
0303 group_names = [r.subject_key for r in group['rows']]
0304 result = dataset_propagation_set(
0305 group_names, state, comment, replaced_by=replaced_by,
0306 changed_by=decided_by, filter_state=filter_state,
0307 origin=group['origin'])
0308 executed = set(result['changed']) | set(result['unchanged'])
0309 with transaction.atomic():
0310 for row in group['rows']:
0311 if row.subject_key not in executed:
0312 continue
0313 row.status = 'executed'
0314 row.quality = quality
0315 row.decided_by = decided_by
0316 row.decided_at = now
0317 row.executed_log_id = result.get('log_id')
0318 row.save(update_fields=['status', 'quality', 'decided_by',
0319 'decided_at', 'executed_log_id'])
0320 _clear_proposal_projection(row.subject_key)
0321 approved.append(row.subject_key)
0322
0323 if decision == 'deny':
0324 log_epicprod_action(
0325 'web', 'proposal_denied',
0326 username=decided_by,
0327 sublevel='normal', live_default=True,
0328 message=(f'AI proposal denied on {len(denied)} dataset(s)'
0329 + (f' [{quality}]' if quality else '')),
0330 denied=len(denied), stale=len(stale),
0331 no_proposal=len(no_proposal),
0332 **({'quality': quality} if quality else {}),
0333 )
0334 result = {'approved': approved, 'denied': denied, 'stale': stale,
0335 'no_proposal': no_proposal}
0336 if approved or denied or stale:
0337 cache_error = _refresh_catalog_table_cache()
0338 if cache_error:
0339 result['cache_refresh_error'] = cache_error
0340 return result
0341
0342
0343 def proposal_undo(composed_names, *, undone_by='', proposal_ids=None):
0344 """Undo executed AI proposals — the computed compensating action
0345 (AI_PROPOSALS.md).
0346
0347 Selection mirrors decide: by dataset composed names (the catalog and
0348 compose surfaces) and/or by proposal-list row ids. Each selected
0349 executed proposal is compensated through the identical executor: the
0350 prior state (and prior ``replaced_by``) captured in the precondition
0351 at propose time is restored, with a templated comment naming the
0352 proposal, ``origin: undo`` provenance carrying the proposal id, and
0353 the undoing human as ``changed_by`` — a new history entry, never
0354 erasure. Guarded like decide: if the record has moved past the
0355 executed payload, the undo offer has expired and the row is skipped
0356 (counted, never silent). The undone proposal returns to ``proposed``
0357 — pending again, decision fields cleared, render projection restored
0358 — while the execution and undo events carry the record; the row keeps
0359 a trace of its most recent undo.
0360 """
0361 names = [n.strip() for n in (composed_names or []) if n and n.strip()]
0362 ids = [int(i) for i in (proposal_ids or [])]
0363 if not names and not ids:
0364 raise ServiceError('no dataset names or proposal ids supplied')
0365 if not undone_by:
0366 raise ServiceError('an authenticated undoer is required')
0367
0368 selector = Q()
0369 if names:
0370 selector |= Q(subject_key__in=names)
0371 if ids:
0372 selector |= Q(pk__in=ids)
0373 rows = list(Proposal.objects.filter(status='executed').filter(selector))
0374 found_names = {r.subject_key for r in rows}
0375 found_ids = {r.pk for r in rows}
0376 no_proposal = [n for n in names if n not in found_names]
0377 not_executed = [i for i in ids if i not in found_ids]
0378
0379 now = _timezone.now()
0380 undone, moved = [], []
0381 for row in rows:
0382 head = (Dataset.objects
0383 .filter(composed_name=row.subject_key)
0384 .order_by('block_num', 'pk').first())
0385 payload = row.payload or {}
0386 pre = row.precondition or {}
0387
0388 if head is None or head.propagation != payload.get('state') or (
0389 payload.get('replaced_by')
0390 and head.replaced_by != payload.get('replaced_by')):
0391 moved.append(row.subject_key)
0392 continue
0393 prev_replaced_by = pre.get('prev_replaced_by', '')
0394 touched_replaced_by = bool(payload.get('replaced_by'))
0395 result = dataset_propagation_set(
0396 [row.subject_key], pre.get('prev_state'),
0397 f'undo of AI proposal #{row.pk} '
0398 f'(approved by {row.decided_by or "unknown"})',
0399 replaced_by=prev_replaced_by if touched_replaced_by else '',
0400 clear_replaced_by=touched_replaced_by and not prev_replaced_by,
0401 changed_by=undone_by,
0402 origin={'kind': 'undo', 'undo_of': row.pk},
0403 )
0404 row.status = 'proposed'
0405 row.decided_by = ''
0406 row.decided_at = None
0407 row.quality = ''
0408 row.executed_log_id = None
0409 row.undone_by = undone_by
0410 row.undone_at = now
0411 row.undone_log_id = result.get('log_id')
0412 row.save(update_fields=['status', 'decided_by', 'decided_at',
0413 'quality', 'executed_log_id', 'undone_by',
0414 'undone_at', 'undone_log_id'])
0415 restored_head = (Dataset.objects
0416 .filter(composed_name=row.subject_key)
0417 .order_by('block_num', 'pk').first())
0418 if restored_head is not None:
0419 _write_proposal_projection(row, restored_head)
0420 undone.append(row.subject_key)
0421 result = {'undone': undone, 'moved': moved, 'not_executed': not_executed,
0422 'no_proposal': no_proposal}
0423 if undone:
0424 cache_error = _refresh_catalog_table_cache()
0425 if cache_error:
0426 result['cache_refresh_error'] = cache_error
0427 return result
0428
0429
0430 def proposal_delete(proposal_ids, *, deleted_by=''):
0431 """Operator deletion of AI proposal list rows — housekeeping for test
0432 or noise entries that would confuse readers. Human-only and logged; a
0433 pending row also clears its render projection. This removes decision
0434 history, so it is a cleanup verb, never a decision verb."""
0435 from monitor_app.epicprod_logging import log_epicprod_action
0436
0437 ids = [int(i) for i in (proposal_ids or [])]
0438 if not ids:
0439 raise ServiceError('no proposal ids supplied')
0440 if not deleted_by:
0441 raise ServiceError('an authenticated deleter is required')
0442 deleted = 0
0443 with transaction.atomic():
0444 for row in Proposal.objects.filter(pk__in=ids):
0445 if row.status == 'proposed':
0446 _clear_proposal_projection(row.subject_key)
0447 row.delete()
0448 deleted += 1
0449 log_epicprod_action(
0450 'web', 'proposal_deleted',
0451 username=deleted_by,
0452 sublevel='normal', live_default=False,
0453 message=f'{deleted} AI proposal list row(s) deleted',
0454 deleted=deleted,
0455 )
0456 return {'deleted': deleted}
0457
0458
0459 def proposal_withdraw(*, batch_id=None, created_by=''):
0460 """Withdraw pending proposals — the recurring proposer's heartbeat
0461 (withdraw, then re-derive and re-propose from current inputs) or an
0462 operator clear. Counted and logged (``proposal_expired``),
0463 never silent."""
0464 from monitor_app.epicprod_logging import log_epicprod_action
0465
0466 now = _timezone.now()
0467 withdrawn = 0
0468 with transaction.atomic():
0469 qs = Proposal.objects.filter(action='propagation', status='proposed')
0470 if batch_id:
0471 qs = qs.filter(batch_id=batch_id)
0472 for row in qs:
0473 row.status = 'withdrawn'
0474 row.decided_at = now
0475 row.save(update_fields=['status', 'decided_at'])
0476 _clear_proposal_projection(row.subject_key)
0477 withdrawn += 1
0478 log_epicprod_action(
0479 'web', 'proposal_expired',
0480 username=created_by,
0481 sublevel='normal', live_default=True,
0482 message=f'{withdrawn} pending AI proposal(s) withdrawn'
0483 + (f' [batch {batch_id}]' if batch_id else ''),
0484 withdrawn=withdrawn, batch_id=batch_id or '',
0485 )
0486 result = {'withdrawn': withdrawn}
0487 if withdrawn:
0488 cache_error = _refresh_catalog_table_cache()
0489 if cache_error:
0490 result['cache_refresh_error'] = cache_error
0491 return result
