scripts/validate_debug_bundle.py

#!/usr/bin/env python3
"""Fail closed when a NotePilot v2 debug bundle is not portable by itself."""
import base64
import json
import sys
from pathlib import Path


def fail(message):
    print(f'FAIL: {message}', file=sys.stderr)
    raise SystemExit(1)


def require(mapping, keys, label):
    for key in keys:
        if key not in mapping:
            fail(f'missing {label}.{key}')


def main() -> int:
    if len(sys.argv) != 2:
        print('usage: validate_debug_bundle.py bundle.json', file=sys.stderr)
        return 2
    try:
        bundle = json.loads(Path(sys.argv[1]).read_text())
    except Exception as exc:
        fail(f'invalid JSON: {exc}')
    if bundle.get('schema') != 'notepilot.debug-bundle/v2': fail('unexpected schema')
    require(bundle, ('meta', 'events', 'audio', 'clocks', 'diagnostics', 'analysis', 'schemaDocumentation'), 'bundle')
    meta = bundle['meta']
    require(meta, ('score', 'inputMode', 'userAgent', 'scoreEvents', 'latency'), 'meta')
    if not meta['scoreEvents'].get('solo') or 'accompaniment' not in meta['scoreEvents']:
        fail('incomplete score events')
    if meta.get('micEngine') == 'follower' and not meta.get('follower', {}).get('options'):
        fail('missing follower options')
    if not any(e.get('type') == 'mic.gateCalibration' for e in bundle.get('events', [])): fail('missing calibration history')
    events = bundle['events']
    if not events or any(not isinstance(e.get('sessionTimeMs'), (int, float)) for e in events): fail('events lack numeric canonical sessionTimeMs')
    if any(events[i]['sessionTimeMs'] > events[i + 1]['sessionTimeMs'] for i in range(len(events) - 1)):
        fail('event timestamps are not monotonic')
    for kind, description in (('blind.pitch', 'raw blind pitch'), ('mic.gate', 'gate history'), ('follow.frame', 'full follower frames')):
        if not any(e.get('type') == kind for e in events): fail(f'missing {description}')
    if not any(e.get('type') in ('tracker', 'follow.advance', 'skip', 'follow.resync') for e in events): fail('no tracker decisions')
    if not isinstance(bundle['schemaDocumentation'], dict) or not bundle['schemaDocumentation'].get('sal'): fail('missing salience documentation')
    if bundle['clocks'].get('master') != 'sessionTimeMs': fail('missing master-clock documentation')
    timeline = bundle['diagnostics'].get('timeline')
    if not isinstance(timeline, list) or not timeline: fail('missing graph-ready timeline')
    required_row = ('tMs', 'db', 'gateOpen', 'onsetValue', 'blindMidi', 'expectedMidi', 'posterior', 'wrongProbability')
    if any(any(k not in row for k in required_row) for row in timeline): fail('timeline rows are incomplete')
    if any(timeline[i]['tMs'] > timeline[i + 1]['tMs'] for i in range(len(timeline) - 1)): fail('timeline timestamps are not monotonic')
    if not isinstance(bundle['analysis'].get('attempts'), list): fail('missing attempt summary')
    for key in ('falseAdvances', 'autoSkips', 'resyncs', 'hostRejections', 'missedAttempts'):
        if not isinstance(bundle['analysis'].get(key), list): fail(f'missing analysis.{key}')
    mic = bundle['audio'].get('microphone', {})
    require(mic, ('codec', 'sampleRate', 'channels', 'durationMs', 'captureStartSessionMs', 'captureEndSessionMs', 'base64', 'bytes'), 'audio.microphone')
    if not mic.get('base64'): fail('missing embedded microphone audio')
    try:
        raw = base64.b64decode(mic['base64'], validate=True)
    except Exception as exc:
        fail(f'invalid microphone base64: {exc}')
    if len(raw) != mic.get('bytes'): fail('microphone byte count does not match base64')
    if not raw.startswith(b'\x1aE\xdf\xa3'): fail('microphone payload is not WebM/EBML')
    output = bundle['audio'].get('digitalOutput', {})
    if output.get('base64'):
        try: base64.b64decode(output['base64'], validate=True)
        except Exception as exc: fail(f'invalid output base64: {exc}')
    elif output.get('kind') != 'EXACT_EVENT_MANIFEST':
        fail('digital output must be embedded audio or EXACT_EVENT_MANIFEST')
    print(f'PASS: {len(events)} events; {len(timeline)} frame rows; mic WebM {len(raw)} bytes; {len(bundle["analysis"]["attempts"])} attempts')
    return 0


if __name__ == '__main__':
    raise SystemExit(main())