swf-monitor/scripts/live-watch-enforce.py

0001 #!/usr/bin/env python3
0002 """live-watch-enforce.py — accept a live watch run's result.
0003
0004 The enforcement end of the live watch (swf-epicprod
0005 docs/EPICPROD_ASSESSMENTS.md, The live watch), on the segfault diagnosis's
0006 pattern: on the corun completion callback the ops agent runs this with the
0007 job, the prompt group and the result page. The artifact is extracted and
0008 validated against the schema and the floor; on failure one bounded repair
0009 run receives the exact problems and the prior output, and a second failure
0010 quarantines it.
0011
0012 A valid run is kept as the cached product ``live_watch`` (the latest
0013 report) and recorded as a ``live_watch`` action. It is registered as an
0014 assessment, and its action marked ``notify`` for the Capcom feed, only
0015 when it is news (``spec.news``): a new verdict, the return to clean
0016 included, or a real problem or noise action not named since the verdict
0017 began. A newsworthy alarm reaches the channel; nothing else the watch
0018 does posts there.
0019
0020     cd /data/wenauseic/github/swf-monitor/src
0021     ../../swf-testbed/.venv/bin/python ../scripts/live-watch-enforce.py \\
0022         --job-id ID --prompt-group-id GID --page-group-id PGID --status completed
0023 """
0024 import argparse
0025 import json
0026 import logging
0027 import os
0028 import sys
0029 import time
0030
0031 THIS_DIR = os.path.dirname(os.path.abspath(__file__))
0032 sys.path.insert(0, os.path.join(THIS_DIR, '..', 'src'))
0033 os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'swf_monitor_project.settings')
0034
0035 import django  # noqa: E402
0036 django.setup()
0037
0038 from django.utils import timezone  # noqa: E402
0039
0040 from monitor_app.cached_product import get_product  # noqa: E402
0041 from monitor_app.epicprod_logging import log_epicprod_action  # noqa: E402
0042 from monitor_app.mcp.ai_content import _register_ai_assessment_sync  # noqa: E402
0043 from monitor_app.models import PersistentState  # noqa: E402
0044 from swf_epicprod.assessment.bundle import _get  # noqa: E402
0045 from swf_epicprod.assessment.trigger import (CORUN_API_TOKEN, CORUN_API_URL,  # noqa: E402
0046                                              _request)
0047 from swf_epicprod.livewatch import spec  # noqa: E402
0048
0049 SECTION = os.environ.get('CORUN_LIVEWATCH_SECTION', spec.DEFAULT_SECTION)
0050 DEFINITION = os.environ.get('CORUN_LIVEWATCH_DEFINITION', '')
0051 CHANNEL_URL = 'https://chat.epic-eic.org/main/channels/epicprod-live'
0052 SEVERITY = {'ok': 'info', 'attention': 'warning', 'alarm': 'alarm'}
0053 PRODUCT_KEY = 'live_watch'
0054 PRODUCT_TTL_S = 7 * 24 * 3600
0055 log = logging.getLogger('live-watch-enforce')
0056
0057
0058 def _state():
0059     return dict(PersistentState.get_state().get('live_watch') or {})
0060
0061
0062 def _save(state):
0063     PersistentState.update_state({'live_watch': state})
0064
0065
0066 def _mark(state, run_state, **fields):
0067     current = dict(state.get('current') or {})
0068     current['state'] = run_state
0069     current.update(fields)
0070     state['current'] = current
0071     _save(state)
0072
0073
0074 def _log(outcome, *, notify=False, live=False, username='', **counts):
0075     log_epicprod_action('live-watch', 'live_watch', outcome=outcome,
0076                         subject_type='live_channel', subject_key='epicprod-live',
0077                         username=username, sublevel='high' if live else ('normal' if notify else 'low'),
0078                         live_default=live, notify=notify,
0079                         level=logging.ERROR if outcome == 'error' else logging.INFO, **counts)
0080
0081
0082 def main():
0083     ap = argparse.ArgumentParser()
0084     ap.add_argument('--job-id', required=True)
0085     ap.add_argument('--prompt-group-id', required=True)
0086     ap.add_argument('--page-group-id', default='')
0087     ap.add_argument('--status', required=True)
0088     args = ap.parse_args()
0089     logging.basicConfig(level=logging.INFO, stream=sys.stderr,
0090                         format='%(asctime)s %(name)s %(levelname)s %(message)s')
0091     t0 = time.monotonic()
0092     state = _state()
0093     requested_by = (state.get('current') or {}).get('requested_by') or ''
0094
0095     prompt = _get(f'{CORUN_API_URL}/prompts/{args.prompt_group_id}/', token=CORUN_API_TOKEN)
0096     try:
0097         submitted = json.loads(prompt.get('content') or '{}')
0098     except json.JSONDecodeError:
0099         submitted = {}
0100     bundle = submitted.get('bundle') or {}
0101
0102     if args.status != 'completed':
0103         _mark(state, 'failed', job_id=args.job_id, reason=f'run ended {args.status}')
0104         _log('error', username=requested_by, reason=f'corun run {args.job_id} ended {args.status}')
0105         print(json.dumps({'outcome': 'failed', 'status': args.status}))
0106         return 1
0107
0108     page = _get(f'{CORUN_API_URL}/pages/{args.page_group_id}/', token=CORUN_API_TOKEN)
0109     content = page.get('content') or ''
0110     artifact, remainder, problems = spec.extract_artifact(content)
0111     if artifact is not None:
0112         problems += spec.validate_artifact(artifact, bundle)
0113         problems += spec.validate_remainder(remainder)
0114
0115     if problems:
0116         if not submitted.get('repair'):
0117             repair = dict(submitted)
0118             repair['repair'] = {'validation_problems': problems, 'previous_output': content,
0119                                 'instruction': 'Produce a complete replacement JSON artifact that '
0120                                                'corrects every listed validation problem while '
0121                                                'preserving the supported findings.'}
0122             retry_prompt = _request(f'{CORUN_API_URL}/prompts/',
0123                                     payload={'section': SECTION, 'content': json.dumps(repair, default=str),
0124                                              'definition_id': DEFINITION},
0125                                     token=CORUN_API_TOKEN)
0126             retry_job = _request(f'{CORUN_API_URL}/jobs/',
0127                                  payload={'prompt_group_id': str(retry_prompt.get('group_id') or ''),
0128                                           'definition_id': DEFINITION},
0129                                  token=CORUN_API_TOKEN)
0130             _mark(state, 'repairing', repair_job_id=str(retry_job.get('id') or retry_job.get('job_id') or ''),
0131                   problems=problems[:20])
0132             _log('repair', username=requested_by, reason='; '.join(problems)[:300],
0133                  job_id=args.job_id, problems_count=len(problems))
0134             print(json.dumps({'outcome': 'repair', 'problems': problems}))
0135             return 0
0136         _mark(state, 'quarantined', job_id=args.job_id, problems=problems[:20],
0137               raw_page_group_id=args.page_group_id)
0138         _log('error', username=requested_by, reason='quarantined: ' + '; '.join(problems)[:280],
0139              job_id=args.job_id, problems_count=len(problems))
0140         print(json.dumps({'outcome': 'quarantined', 'problems': problems}))
0141         return 1
0142
0143     report = spec.render_report(bundle, artifact)
0144     verdict = artifact.get('verdict', 'ok')
0145     found = spec.issue_set(artifact)
0146     # Before the episode was kept, the last registered run stood for it.
0147     changed, episode = spec.news(found, state.get('episode') or state.get('last_registered'))
0148     now = timezone.now().isoformat(timespec='seconds')
0149     latest = {'at': now, 'verdict': verdict, 'floor': (bundle.get('floor') or {}).get('verdict'),
0150               'narration': artifact.get('narration', ''), 'issue_set': found,
0151               'report': report, 'structured': artifact, 'job_id': args.job_id,
0152               'bundle': bundle.get('artifact'), 'registered': changed}
0153     get_product(PRODUCT_KEY, lambda: latest, ttl_seconds=PRODUCT_TTL_S, refresh=True)
0154
0155     page_group_id = ''
0156     if changed:
0157         result = _register_ai_assessment_sync(
0158             subject_type='live_channel', subject_key='epicprod-live', assessment=report,
0159             username='live-watch', ai='corun-job', subject_label='epicprod-live channel',
0160             subject_url=CHANNEL_URL,
0161             title=f"Live watch: {verdict}"
0162                   f"{' — ' + ', '.join(found['real_problems'][:3]) if found['real_problems'] else ''}",
0163             data={'assessment_kind': 'live_watch', 'origin': 'harness',
0164                   'schema_version': spec.SCHEMA_VERSION, 'verdict': verdict,
0165                   'narration': artifact.get('narration', '')[:1000], 'structured': artifact,
0166                   'prompt_group_id': args.prompt_group_id, 'job_id': args.job_id,
0167                   'bundle': bundle.get('artifact')})
0168         if not result.get('success'):
0169             _mark(state, 'registration_failed', job_id=args.job_id, reason=str(result.get('error'))[:300])
0170             _log('error', username=requested_by, reason=f"registration failed: {result.get('error')}")
0171             print(json.dumps({'outcome': 'registration_failed', 'error': result.get('error')}))
0172             return 1
0173         page_group_id = result.get('corun_page_group_id') or ''
0174         state['last_registered'] = found
0175         state['last_registered_at'] = now
0176     state['episode'] = episode
0177     state['last'] = {'at': now, 'verdict': verdict, 'issue_set': found, 'registered': changed}
0178     _mark(state, 'accepted', job_id=args.job_id, accepted_at=now, page_group_id=page_group_id)
0179     _log('ok', notify=changed, live=(verdict == 'alarm' and changed), username=requested_by,
0180          duration_ms=int((time.monotonic() - t0) * 1000), severity=SEVERITY.get(verdict, 'info'),
0181          verdict=verdict, narration=artifact.get('narration', '')[:600],
0182          summary=artifact.get('narration', '')[:600], registered=changed,
0183          real_problems=found['real_problems'], noise=found['noise'],
0184          corun_page_group_id=page_group_id)
0185     print(json.dumps({'outcome': 'accepted', 'verdict': verdict, 'registered': changed,
0186                       'issue_set': found}))
0187     return 0
0188
0189
0190 if __name__ == '__main__':
0191     sys.exit(main())