#!/usr/bin/env python3 """Offline verification of the first approved probe; divergences are assertions, not greens.""" import copy import hashlib import json import sys from pathlib import Path ROOT = Path(__file__).parent PRIVATE = ROOT / 'probe-private-evidence' PLAN_SHA = 'b38252b664d4e4d401579cf00eaeab3bce4563fc258e96a4ffc5fbd01d9e9bc9' def verify(plan, lanes): result = {} for lane, events in lanes.items(): grouped = {} for event in events: grouped.setdefault(event['kind'], []).append(event['data']) one = lambda kind: grouped[kind][0] if len(grouped[kind]) == 1 else None assert 'stopped' not in grouped and 'skipped' not in grouped starts, ends = grouped['DML-start'], grouped['DML-result'] spec = plan['lanes'][lane] assert len(starts) == len(ends) == len(spec['steps']) == 19 ddl = [spec['create_table']] + ([spec['create_unique_index']] if spec['create_unique_index'] else []) + spec['cleanup'] assert [x['SQL'] for x in grouped['DDL-start']] == ddl assert [x['SQL'] for x in grouped['DDL-success']] == ddl for item in grouped['DDL-start']: assert item['sha256'] == hashlib.sha256(item['SQL'].encode()).hexdigest() assert one('collision-before') == one('collision-after') == {'table': [], 'index': []} previous = [] expected_failures = {'S05-tail-duplicate', 'S07-space-duplicate', 'S08-empty-duplicate'} if lane == 'C': expected_failures |= {'S04-lower', 'S18-md5-char'} for step, start, end in zip(spec['steps'], starts, ends): expected_bindings = [{'value': x, 'type': 0 if x is None else 1 if type(x) is int else 2} for x in step['binds']] assert start == {key: end[key] for key in start} assert start['SQL'] == step['SQL'] and start['label'] == step['label'] assert start['bindings'] == expected_bindings assert start['hypothesis'] == step['hypothesis'] and start['before'] == previous assert end['ok'] == (step['label'] not in expected_failures) rows = end['after'] assert len(rows) <= 6 and len({x['id'] for x in rows}) == len(rows) assert all(type(x['id']) is int and 1 <= x['id'] <= 6 and x['null_v'] == 'CDXSS7' for x in rows) for row in rows: for field, alias in [('dingtalk_avatar', 'avatar_hex'), ('dingtalk_userid', 'userid_hex'), ('primary_department_uuid', 'primary_hex'), ('dingtalk_sync_md5', 'md5_hex')]: native = row[field] # This build returns NULL HEX for the empty string, although native empty remains distinct from NULL. expected_hex = native.encode().hex().upper() if native else None assert row[alias] in ([expected_hex, ''] if native == '' else [expected_hex]) if not end['ok']: assert rows == previous assert ('1062' if lane == 'B' else '-6604' if step['label'] == 'S18-md5-char' else '-6602') in end['error']['message'] if end['ok'] and step['SQL'].startswith('UPDATE'): before_map = {r['id']: r for r in previous} assignments = step['SQL'].split(' SET ')[1].split(' WHERE ')[0].split(',') changed = {a.split('=')[0].lower() for a in assignments} aliases = {'dingtalk_avatar': 'avatar_hex', 'dingtalk_userid': 'userid_hex', 'dingtalk_sync_md5': 'md5_hex', 'dingtalk_last_seen_full_batch_id': 'batch_exact'} allowed = changed | {aliases[k] for k in changed if k in aliases} | {'update_time'} assert set(before_map) == {r['id'] for r in rows} for row in rows: assert {k: v for k, v in row.items() if row['id'] != 6 or k not in allowed} == {k: v for k, v in before_map[row['id']].items() if row['id'] != 6 or k not in allowed} if step['label'] in ('S09-max-capacities', 'S19-avatar-last-byte'): assert all(end['storage_checks'].values()) and len(end['storage_checks']) >= 4 row = next(r for r in rows if r['id'] == 6) avatar = step['binds'][4] if step['label'] == 'S09-max-capacities' else step['binds'][0] userid = step['binds'][3] if step['label'] == 'S09-max-capacities' else step['binds'][1] assert row['dingtalk_avatar'] == avatar and row['dingtalk_userid'] == userid assert len(avatar) == 1024 and len(avatar.encode()) == end['avatar_bytes'] == 4096 assert len(userid) == 128 and len(userid.encode()) == end['userid_bytes'] == 512 assert end['avatar_sha256'] == hashlib.sha256(avatar.encode()).hexdigest() assert end['userid_sha256'] == hashlib.sha256(userid.encode()).hexdigest() if step['label'].endswith('uint64'): assert expected_bindings[0]['type'] in (0, 2) and end['storage_checks']['batch_exact_text'] row = next(r for r in rows if r['id'] == 6) assert row['batch_exact'] == step['binds'][0] assert type(row['dingtalk_last_seen_full_batch_id']) is (type(None) if step['binds'][0] is None else float if lane == 'C' else str if int(step['binds'][0]) > 9223372036854775807 else int) previous = rows by_label = {e['label']: e for e in ends} own6 = lambda label: next(r for r in by_label[label]['after'] if r['id'] == 6) assert own6('S16-md5-char')['dingtalk_sync_md5'] == ('A' if lane == 'B' else 'A' + ' ' * 31) assert own6('S17-md5-char')['dingtalk_sync_md5'] == ('a' if lane == 'B' else 'a' + ' ' * 31) assert by_label['S01-null1']['after'][0]['dingtalk_sync_md5'] == ('' if lane == 'B' else ' ' * 32) assert own6('S18-md5-char')['dingtalk_sync_md5'] == ('?' if lane == 'B' else 'a' + ' ' * 31) if lane == 'B': assert [w['code'] for w in by_label['S18-md5-char']['warnings']] == [1366] predicate = one('predicate-final') assert predicate['A']['plain'] == predicate['A ']['plain'] == [{'id': 3}] assert predicate['a']['plain'] == [{'id': 4 if lane == 'B' else 3}] assert predicate['']['plain'] == predicate[' ']['plain'] == [{'id': 5}] if lane == 'C': assert predicate['a']['binary'] == [] and predicate['A']['binary'] == predicate['A ']['binary'] == [{'id': 3}] assert one('own-final') == previous and len(previous) == (6 if lane == 'B' else 5) before, after = one('business-before'), one('business-after') drift = [] for key in before: if before[key] != after[key]: assert lane == 'B' and key == 'aggregate' assert len(before[key]) == len(after[key]) == 1 for field in before[key][0]: if before[key][0][field] != after[key][0][field]: assert field == 'update_time_max' assert (before[key][0][field], after[key][0][field]) == ('2026-10-11 04:23:16.000000', '2026-10-11 04:23:31.000000') drift.append({'field': field, 'before': before[key][0][field], 'after': after[key][0][field], 'attribution': 'unknown concurrent drift; no probe Staff writes'}) assert len(drift) == (1 if lane == 'B' else 0) summary = one('summary') assert summary['completed'] and summary['DML_attempts'] == 19 and summary['DDL_attempts'] == len(ddl) assert summary['SELECT'] == (45 if lane == 'B' else 56) <= 120 assert summary['HTTP'] == summary['business_table_writes'] == summary['objects_remaining'] == 0 assert summary['staff_metadata_aggregate_exact'] == (lane == 'C') result[lane] = {'DML_attempts': 19, 'success': 19 - len(expected_failures), 'DDL': len(ddl), 'readonly': summary['SELECT'], 'objects_remaining': 0, 'staff_exact': lane == 'C', 'explicit_drift': drift} return {'evidence_verified': True, 'business_compatibility_passed': False, 'lanes': result, 'greens': ['full4096B_avatar_and512B_userid', 'UINT64_exact_server_text_with_STR_NULL_binds', 'two_NULL_and_single_empty_userid', 'exact_owned_cleanup'], 'reds': ['BINARY_unique_rejects_source_legal_case_pair', 'CHAR32_native_padding', 'invalid_ASCII_coercion_vs_rejection', 'C_native_UINT64_double_loss'], 'Staff_business_DDL_DML_HTTP': 0} def main(): plan_bytes = (ROOT / 'probe-character-unique-uint64-proposal.json').read_bytes() assert hashlib.sha256(plan_bytes).hexdigest() == PLAN_SHA plan = json.loads(plan_bytes) lanes = {lane: [json.loads(line) for line in (PRIVATE / (lane + '-probe.jsonl')).read_text().splitlines()] for lane in ('B', 'C')} for lane in lanes: assert (PRIVATE / (lane + '-probe.stderr')).read_bytes() == b'' manifest_path = PRIVATE / 'manifest.json' if manifest_path.exists(): manifest = json.loads(manifest_path.read_text()) assert {f['path'] for f in manifest['files']} == {'B-probe.jsonl', 'B-probe.stderr', 'C-probe.jsonl', 'C-probe.stderr'} for entry in manifest['files']: raw = (PRIVATE / entry['path']).read_bytes() assert len(raw) == entry['bytes'] and hashlib.sha256(raw).hexdigest() == entry['sha256'] result = verify(plan, lanes) if '--test-mutations' in sys.argv: mutations = [ ('C', 'DML-result', lambda d: d.update(ok=True) if d['label'] == 'S04-lower' else None), ('C', 'DML-result', lambda d: d['after'][0].update(dingtalk_sync_md5='') if d['label'] == 'S01-null1' else None), ('B', 'business-after', lambda d: d['aggregate'][0].update(update_time_max='2026-10-11 04:23:16.000000')), ('C', 'collision-after', lambda d: d.update(table=[{'table_name': 'CDX_SS7_261011'}])), ('C', 'DML-result', lambda d: d['bindings'][0].update(type=1) if d['label'] == 'S12-uint64' else None), ('B', 'DML-result', lambda d: d['after'][-1].update(dingtalk_avatar=d['after'][-1]['dingtalk_avatar'][:-1]) if d['label'] == 'S19-avatar-last-byte' else None), ] for lane, kind, alter in mutations: changed = copy.deepcopy(lanes) for event in changed[lane]: if event['kind'] == kind: alter(event['data']) try: verify(plan, changed) except AssertionError: continue raise AssertionError('Undetected mutation: ' + kind) result['mutations_rejected'] = len(mutations) print(json.dumps(result, ensure_ascii=False, indent=2)) if __name__ == '__main__': main()