새 검토와 보류 후속을 나누는 합성 모듈

"""Independent synthetic workset; declared metadata, no I/O or authorization."""
from copy import deepcopy
import re

ALIAS = re.compile(r'[A-Za-z0-9_][A-Za-z0-9_.-]{0,47}\Z')


def _index(rows, recorded):
    if not isinstance(rows, list) or len(rows) > 64:
        raise ValueError('Expected at most 64 rows')
    required = {'id', 'revision', 'sha256'} | ({'full_frame_checked', 'status'} if recorded else {'holds'})
    indexed = {}
    for source in rows:
        if not isinstance(source, dict) or set(source) != required:
            raise ValueError('Unexpected or missing fields')
        row = deepcopy(source)
        if not isinstance(row['id'], str) or not ALIAS.fullmatch(row['id']):
            raise ValueError('Synthetic alias required')
        if row['id'] in indexed:
            raise ValueError('Duplicate id; preserve rejected evidence separately')
        if type(row['revision']) is not int or row['revision'] < 1:
            raise ValueError('Positive revision required')
        if not isinstance(row['sha256'], str) or not re.fullmatch(r'[0-9a-f]{64}', row['sha256']):
            raise ValueError('Lowercase SHA-256 declaration required')
        if recorded:
            if type(row['full_frame_checked']) is not bool or row['status'] not in ('complete', 'held', 'follow_up'):
                raise ValueError('Explicit inspection evidence and known status required')
        else:
            holds = row['holds']
            if not isinstance(holds, list) or len(holds) > 8:
                raise ValueError('At most eight hold aliases required')
            if any(not isinstance(x, str) or not ALIAS.fullmatch(x) for x in holds) or len(holds) != len(set(holds)):
                raise ValueError('Unique hold aliases required')
        indexed[row['id']] = row
    return indexed


def plan(current, records):
    current, records = _index(current, False), _index(records, True)
    inspections, followups, complete = [], [], []
    for key, item in sorted(current.items()):
        previous = records.get(key)
        if previous is None:
            reason = 'no_record'
        elif previous['revision'] != item['revision']:
            reason = 'revision_changed'
        elif previous['sha256'] != item['sha256']:
            reason = 'fingerprint_changed'
        elif not previous['full_frame_checked']:
            reason = 'inspection_evidence_missing'
        else:
            reason = None
        if reason:
            inspections.append({'id': key, 'reason': reason, 'holds': item['holds']})
        elif previous['status'] == 'complete' and not item['holds']:
            complete.append(key)
        else:
            followups.append({'id': key, 'recorded_status': previous['status'], 'holds': item['holds']})
    return {'new_inspections': inspections, 'next_inspection': inspections[0] if inspections else None,
            'existing_followups': followups, 'complete_ids': complete,
            'historical_only_record_ids': sorted(set(records) - set(current)),
            'records_authenticated': False, 'performs_execution': False,
            'changes_holds': False, 'goal_complete': False}