swf-monitor/scripts/segfault-diagnosis-enforce.py

0001 #!/usr/bin/env python3
0002 """segfault-diagnosis-enforce.py — accept a segfault diagnosis run's result.
0003
0004 Stage 5 of swf-epicprod docs/SEGFAULT_DIAGNOSIS.md, the enforcement end
0005 of the harness (EPICPROD_ASSESSMENTS_V1.md): on the corun completion
0006 callback the ops agent runs this with the job, the prompt group and the
0007 result page. The artifact is extracted and validated against the schema
0008 and the verdict floor; on failure one bounded repair run receives the
0009 exact problems and the prior output, and a second failure quarantines
0010 the artifact (the raw output retained, nothing registered as a
0011 diagnosis). A valid artifact is rendered with the bundle's facts and
0012 registered as an assessment on the signature (subject type
0013 ``crash_signature``), the signature's status becomes ``diagnosed``
0014 (``handed_off`` when the operator action is hand_off, with the handoff
0015 text stored) and its verdict the classification and action; the
0016 ``segfault_diagnosis`` action carries the classification as its
0017 severity for notice routing.
0018
0019     cd /data/wenauseic/github/swf-monitor/src
0020     ../../swf-testbed/.venv/bin/python ../scripts/segfault-diagnosis-enforce.py \\
0021         --job-id ID --prompt-group-id GID --page-group-id PGID --status completed
0022 """
0023 import argparse
0024 import json
0025 import logging
0026 import os
0027 import sys
0028 import time
0029
0030 THIS_DIR = os.path.dirname(os.path.abspath(__file__))
0031 sys.path.insert(0, os.path.join(THIS_DIR, '..', 'src'))
0032 os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'swf_monitor_project.settings')
0033
0034 import django  # noqa: E402
0035 django.setup()
0036
0037 from django.utils import timezone  # noqa: E402
0038
0039 from monitor_app.epicprod_logging import log_epicprod_action  # noqa: E402
0040 from monitor_app.mcp.ai_content import _register_ai_assessment_sync  # noqa: E402
0041 from monitor_app.models import CrashSignature  # noqa: E402
0042 from swf_epicprod.assessment.bundle import _get  # noqa: E402
0043 from swf_epicprod.assessment.trigger import (CORUN_API_TOKEN, CORUN_API_URL,  # noqa: E402
0044                                              _request)
0045 from swf_epicprod.segfault import spec  # noqa: E402
0046
0047 SECTION = os.environ.get('CORUN_SEGFAULT_SECTION', spec.DEFAULT_SECTION)
0048 DEFINITION = os.environ.get('CORUN_SEGFAULT_DEFINITION', '')
0049 SEVERITY = {'software_defect': 'warning', 'configuration': 'warning',
0050             'event_shaped': 'info', 'platform': 'info', 'unresolved': 'info'}
0051 log = logging.getLogger('segfault-diagnosis-enforce')
0052
0053
0054 def _log(key, outcome, username='', **counts):
0055     log_epicprod_action('segfault-diagnosis', 'segfault_diagnosis', outcome=outcome,
0056                         subject_type='crash_signature', subject_key=key, username=username,
0057                         sublevel='normal', live_default=True,
0058                         level=logging.ERROR if outcome == 'error' else logging.INFO,
0059                         url=f'/panda/segfaults/{key}/', **counts)
0060
0061
0062 def _mark(sig, state, **fields):
0063     data = dict(sig.data or {})
0064     diag = dict(data.get('diagnosis') or {})
0065     diag['state'] = state
0066     diag.update(fields)
0067     data['diagnosis'] = diag
0068     sig.data = data
0069     sig.save(update_fields=['data', 'updated_at'])
0070
0071
0072 def main():
0073     ap = argparse.ArgumentParser()
0074     ap.add_argument('--job-id', required=True)
0075     ap.add_argument('--prompt-group-id', required=True)
0076     ap.add_argument('--page-group-id', default='')
0077     ap.add_argument('--status', required=True)
0078     args = ap.parse_args()
0079     logging.basicConfig(level=logging.INFO, stream=sys.stderr,
0080                         format='%(asctime)s %(name)s %(levelname)s %(message)s')
0081     t0 = time.monotonic()
0082
0083     prompt = _get(f'{CORUN_API_URL}/prompts/{args.prompt_group_id}/', token=CORUN_API_TOKEN)
0084     try:
0085         submitted = json.loads(prompt.get('content') or '{}')
0086     except json.JSONDecodeError:
0087         submitted = {}
0088     bundle = submitted.get('bundle') or {}
0089     key = (bundle.get('params') or {}).get('key') or ''
0090     sig = CrashSignature.objects.filter(key=key).first() if key else None
0091     if sig is None:
0092         _log(key or 'unknown', 'error', reason=f'no signature for prompt {args.prompt_group_id}')
0093         print(json.dumps({'error': f'no signature for prompt {args.prompt_group_id}'}))
0094         return 1
0095     requested_by = ((sig.data or {}).get('diagnosis') or {}).get('requested_by') or ''
0096
0097     if args.status != 'completed':
0098         _mark(sig, 'failed', job_id=args.job_id, reason=f'run ended {args.status}')
0099         _log(key, 'error', requested_by, reason=f'corun run {args.job_id} ended {args.status}')
0100         print(json.dumps({'key': key, 'outcome': 'failed', 'status': args.status}))
0101         return 1
0102
0103     page = _get(f'{CORUN_API_URL}/pages/{args.page_group_id}/', token=CORUN_API_TOKEN)
0104     content = page.get('content') or ''
0105     artifact, remainder, problems = spec.extract_artifact(content)
0106     if artifact is not None:
0107         problems += spec.validate_artifact(artifact, bundle)
0108         problems += spec.validate_remainder(remainder)
0109
0110     if problems:
0111         if not submitted.get('repair'):
0112             repair = dict(submitted)
0113             repair['repair'] = {'validation_problems': problems, 'previous_output': content,
0114                                 'instruction': 'Produce a complete replacement JSON artifact that '
0115                                                'corrects every listed validation problem while '
0116                                                'preserving the supported findings.'}
0117             retry_prompt = _request(f'{CORUN_API_URL}/prompts/',
0118                                     payload={'section': SECTION, 'content': json.dumps(repair, default=str),
0119                                              'definition_id': DEFINITION},
0120                                     token=CORUN_API_TOKEN)
0121             retry_job = _request(f'{CORUN_API_URL}/jobs/',
0122                                  payload={'prompt_group_id': str(retry_prompt.get('group_id') or ''),
0123                                           'definition_id': DEFINITION},
0124                                  token=CORUN_API_TOKEN)
0125             _mark(sig, 'repairing', repair_job_id=str(retry_job.get('id') or retry_job.get('job_id') or ''),
0126                   problems=problems[:20])
0127             _log(key, 'repair', requested_by, reason='; '.join(problems)[:300],
0128                  job_id=args.job_id, problems_count=len(problems))
0129             print(json.dumps({'key': key, 'outcome': 'repair', 'problems': problems}))
0130             return 0
0131         _mark(sig, 'quarantined', job_id=args.job_id, problems=problems[:20],
0132               raw_page_group_id=args.page_group_id)
0133         _log(key, 'error', requested_by, reason='quarantined: ' + '; '.join(problems)[:280],
0134              job_id=args.job_id, problems_count=len(problems))
0135         print(json.dumps({'key': key, 'outcome': 'quarantined', 'problems': problems}))
0136         return 1
0137
0138     report = spec.render_report(bundle, artifact)
0139     classification = artifact.get('classification', 'unresolved')
0140     action = artifact.get('operator_action', 'none')
0141     verdict = (f"{classification} ({artifact.get('confidence', '')}): "
0142                f"{action} — {artifact.get('action_detail', '')}").strip()
0143     result = _register_ai_assessment_sync(
0144         subject_type='crash_signature', subject_key=key, assessment=report,
0145         username='segfault-diagnosis', ai='corun-job', subject_label='', subject_url='',
0146         title=f'Segfault diagnosis {key}: {classification}, {action}',
0147         data={'assessment_kind': 'segfault_diagnosis', 'origin': 'harness',
0148               'schema_version': spec.SCHEMA_VERSION, 'verdict': classification,
0149               'narration': ' '.join(artifact.get('diagnosis') or [])[:1000],
0150               'structured': artifact, 'prompt_group_id': args.prompt_group_id,
0151               'job_id': args.job_id, 'bundle': bundle.get('artifact')})
0152     if not result.get('success'):
0153         _mark(sig, 'registration_failed', job_id=args.job_id, reason=str(result.get('error'))[:300])
0154         _log(key, 'error', requested_by, reason=f"registration failed: {result.get('error')}")
0155         print(json.dumps({'key': key, 'outcome': 'registration_failed', 'error': result.get('error')}))
0156         return 1
0157
0158     sig = CrashSignature.objects.get(key=key)
0159     sig.verdict = verdict
0160     ids = list((sig.data or {}).get('corun_page_group_ids') or [])
0161     if result.get('corun_page_group_id') and result['corun_page_group_id'] not in ids:
0162         ids.append(result['corun_page_group_id'])
0163     sig.assessment_ids = ids
0164     if action == 'hand_off':
0165         sig.status = 'handed_off'
0166         package = dict(sig.package or {})
0167         package['handoff_text'] = artifact.get('handoff_text', '')
0168         sig.package = package
0169     elif sig.status not in ('fixed', 'accepted'):
0170         sig.status = 'diagnosed'
0171     sig.save(update_fields=['verdict', 'assessment_ids', 'status', 'package', 'updated_at'])
0172     _mark(sig, 'diagnosed', job_id=args.job_id, classification=classification,
0173           operator_action=action, page_group_id=result.get('corun_page_group_id'),
0174           diagnosed_at=timezone.now().isoformat(timespec='seconds'))
0175     _log(key, 'ok', requested_by, duration_ms=int((time.monotonic() - t0) * 1000),
0176          severity=SEVERITY.get(classification, 'info'), classification=classification,
0177          operator_action=action, narration=' '.join(artifact.get('diagnosis') or [])[:600],
0178          corun_page_group_id=result.get('corun_page_group_id') or '')
0179     print(json.dumps({'key': key, 'outcome': 'diagnosed', 'classification': classification,
0180                       'operator_action': action, 'page_group_id': result.get('corun_page_group_id')}))
0181     return 0
0182
0183
0184 if __name__ == '__main__':
0185     sys.exit(main())