#!/usr/bin/env python3 """Offline 77-cell/effect/time/typed-byte verification; preserve actual source counterexamples.""" import copy import hashlib import json import re import sys from decimal import Decimal from pathlib import Path ROOT = Path(__file__).parent PRIVATE = ROOT / 'probe2b-private-evidence' PLAN_SHA = '89343d0caf38bb47b863e6b70e731bbfeb8a10f3a4116ddf1019d5375c96e0fc' OVERLAY_SHA = '4c10852620ee6b19dd3ef27050f4f8fc51379d9c2d56879da785e27fe6e161b9' def events(name): return [json.loads(line) for line in (PRIVATE / name).read_text().splitlines()] def single(es, kind): found = [e['data'] for e in es if e['kind'] == kind] assert len(found) == 1, (kind, len(found)) return found[0] def php_type(value): return 'NULL' if value is None else 'integer' if type(value) is int else 'double' if type(value) is float else 'string' if type(value) is str else None def verify(plan, original_c, lanes, preflight, deps, user_deps): source = json.loads((ROOT / 'private-evidence/B-fresh.json').read_text()) types = {c['column_name'].upper(): c['column_type'] for c in source['columns']} columns = plan['columns'] assert len(columns) == 77 and set(columns) == set(types) resolve = lambda v: plan['payloads'][v['payload_ref']] if isinstance(v, dict) else v summary = {} all_semantics = {} mutation_results = {} assert not any(e['kind'].startswith('DML-') for e in original_c) assert [e['data']['SQL'] for e in original_c if e['kind'] == 'DDL-success'] == [plan['lanes']['C']['create_table']] stopped = single(original_c, 'stopped') assert stopped['DML_attempts'] == 0 and stopped['DDL_attempts'] == 1 and stopped['SELECT'] == 23 assert not stopped['DDL_unknown'] and 'USER_ERRORS' in stopped['message'] and '-2106' in stopped['message'] assert len(single(original_c, 'read-journal')) == 23 assert len(preflight['columns']) == 77 and len(preflight['Staff_columns']) == 70 assert preflight['own_rows'] == [] and preflight['objects'] == [{'object_name': 'CDX_SS7_2B261011', 'object_type': 'TABLE', 'status': 'VALID'}] assert {i['index_type'] for i in preflight['indexes']} == {'CLUSTER', 'NORMAL'} assert len(preflight['indexes']) == 2 # Actual recorded set, not a generic DM index-count assumption. assert deps['independent_row_count'] == [{'cnt': 0}] and deps['foreign_keys'] == [] assert '-5504' in deps['native_dependencies']['read_error'] and user_deps['own_user_dependencies'] == [] assert preflight['SELECT'] + deps['SELECT'] + user_deps['SELECT'] == 15 def row_semantics(row, lane): assert set(row) == {c.lower() for c in columns} | { ('__hex_' if re.match(r'^(char|varchar)', types[c]) else '__canon_' if re.match(r'^(date|datetime)', types[c]) else '__num_') + c.lower() for c in columns if c != 'SPORT_EVENT_UUID' } out = {} for field in columns: name, type_name = field.lower(), types[field] value = row[name] if re.match(r'^(char|varchar|mediumtext)', type_name): assert value is None or type(value) is str out[field] = value if field != 'SPORT_EVENT_UUID': hex_wanted = None if value is None or (lane == 'C' and value == '') else value.encode().hex().upper() assert row['__hex_' + name] == hex_wanted elif re.match(r'^(date|datetime)', type_name): assert value is None or type(value) is str canon = row['__canon_' + name] if value is None: assert canon is None elif type_name.startswith('datetime'): assert re.fullmatch(r'\d{4}-\d\d-\d\d \d\d:\d\d:\d\d\.000000', canon) assert value == (canon if lane == 'C' and field in ('CREATE_TIME', 'PAL_MTIME') else canon[:19]) else: assert re.fullmatch(r'\d{4}-\d\d-\d\d', canon) and value == canon out[field] = canon else: text = row['__num_' + name] if value is None: assert text is None out[field] = None else: assert type(text) is str and type(value) in (int, float, str) if field == 'DINGTALK_LAST_SEEN_FULL_BATCH_ID': assert type(value) is (float if lane == 'C' else str if Decimal(text) > 9223372036854775807 else int) elif type_name.startswith('decimal'): assert type(value) is (str if lane == 'B' else float) elif type_name.startswith('float'): assert type(value) is float else: assert type(value) is int out[field] = Decimal(text) return out def expected_semantics(values): out = {} for field in columns: value, type_name = resolve(values[field]), types[field] if value is None: out[field] = None elif type_name.startswith('char'): out[field] = value.rstrip(' ') elif re.match(r'^(varchar|mediumtext)', type_name): out[field] = value elif type_name.startswith('datetime'): out[field] = value + '.000000' elif type_name.startswith('date'): out[field] = value else: out[field] = Decimal(str(value)) return out def check_detail(detail, row): assert detail['id'] == row['id'] assert detail['native_types'] == {c.lower(): php_type(row[c.lower()]) for c in columns} assert detail['string_bytes'] == {c.lower(): {'bytes': len(row[c.lower()].encode()), 'sha256': hashlib.sha256(row[c.lower()].encode()).hexdigest()} for c in columns if type(row[c.lower()]) is str} value = row['sport_event_uuid'] assert detail['CLOB_full'] == {'native_type': php_type(value), 'bytes': len(value.encode()) if value is not None else None, 'sha256': hashlib.sha256(value.encode()).hexdigest() if value is not None else None, 'unicode_codepoints': len(value) if value is not None else None} def check_time(entry, mode, explicit=None): t = entry['time'] row = entry['after'][0] value = row['update_time'] assert t['native'] == value and t['canonical'] == row['__canon_update_time'] == value + '.000000' assert type(value) is str and len(value) == 19 and t['fsp0_native19'] window = t['window'] assert len(window['before']) == len(window['after']) == 1 start, finish = window['before'][0]['db_time'], window['after'][0]['db_time'] assert start <= finish if mode == 'advance_to_own_DB_window': assert start <= value <= finish and value > '2001-01-01 00:00:00' elif mode == 'preserve_anchor': assert value == '2001-01-01 00:00:00' elif mode == 'explicit_value': assert value == explicit else: assert mode == 'invalid_source_observe_only' return value for lane, es in lanes.items(): assert not any(e['kind'] in ('skipped', 'stopped', 'DDL-error') for e in es) spec = plan['lanes'][lane] starts = [e['data'] for e in es if e['kind'] == 'DML-start'] results = [e['data'] for e in es if e['kind'] == 'DML-result'] expected_ops = [('SEED', spec['seed'], None, None)] for case in spec['cases']: expected_ops += [(case['label'] + '.reset', case['reset'], case, 'reset'), (case['label'] + '.mutation', case['mutation'], case, 'mutation')] assert len(starts) == len(results) == len(expected_ops) == 217 previous = [] semantic_steps = [] mutation_results[lane] = {} success = 0 for start, end, (operation, statement, case, phase) in zip(starts, results, expected_ops): assert start == {k: end[k] for k in start} assert start['operation'] == operation and start['SQL'] == statement['SQL'] assert start['bindings'] == [{'value': resolve(v), 'type': t} for v, t in zip(statement['binds'], statement['PDO_types'])] assert all(type(b['value']) in (str, int, type(None)) and b['type'] == (0 if b['value'] is None else 1 if type(b['value']) is int else 2) for b in start['bindings']) assert start['before'] == previous assert len(end['after']) == len(end['details']) == 1 row = end['after'][0] assert row['id'] in (1, 2) and row['uuid'].startswith('CDXSS2B') check_detail(end['details'][0], row) actual = row_semantics(row, lane) is_invalid = case is not None and case.get('invalid_source') and phase == 'mutation' assert end['ok'] == (not (lane == 'C' and is_invalid)) success += int(end['ok']) if operation == 'SEED': values = plan['baseline_values'].copy() check_time(end, 'advance_to_own_DB_window') else: values = dict(plan['baseline_values'], **case['reset_overrides']) if phase == 'mutation' and not is_invalid: values.update(case['set_values']) if phase == 'reset': assert actual['UPDATE_TIME'] == '2001-01-01 00:00:00.000000' else: mode = case['expected_time'] if case['label'] == 'ALL76_SELF': mode = 'advance_to_own_DB_window' if lane == 'B' else 'preserve_anchor' assert end['time']['matched'] == (lane == 'C') # Original preserve-anchor hypothesis is source-false. if lane == 'C': assert end['source_observed_time']['expected'] == 'advance_to_own_DB_window' and end['source_observed_time']['matched'] is False elif not is_invalid: assert end['time']['matched'] is True check_time(end, mode, case.get('explicit_time')) mutation_results[lane][case['label']] = end wanted = expected_semantics(values) for field in columns: if field == 'UPDATE_TIME' and phase != 'reset': continue assert actual[field] == wanted[field], (lane, operation, field, actual[field], wanted[field]) if is_invalid: assert end['after'] == start['before'] and end['value_mismatches'] is None if lane == 'B': assert end['rowCount'] == 0 and [w['code'] for w in end['warnings']] == [1048] else: assert '-6609' in end['error']['message'] else: assert end['value_mismatches'] == [] if operation.startswith('ALL76_SELF.') and phase == 'mutation': a, b = end['after'][0], start['before'][0] diff = {k for k in a if a[k] != b[k]} assert diff == ({'update_time', '__canon_update_time'} if lane == 'B' else set()) assert end['rowCount'] == 1 previous = end['after'] semantic_steps.append({k: v for k, v in actual.items() if k != 'UPDATE_TIME'}) assert success == (217 if lane == 'B' else 216) assert single(es, 'own-final') == previous and previous[0]['id'] == 2 all_semantics[lane] = semantic_steps ddl_es = original_c + es if lane == 'C' else es ddl = [x for x in (spec['create_table'], spec['create_unique_index'], spec['create_trigger']) if x] + spec['cleanup'] assert [e['data']['SQL'] for e in ddl_es if e['kind'] == 'DDL-start'] == ddl assert [e['data']['SQL'] for e in ddl_es if e['kind'] == 'DDL-success'] == ddl for event in ddl_es: if event['kind'] == 'DDL-start': assert event['data']['sha256'] == hashlib.sha256(event['data']['SQL'].encode()).hexdigest() assert single(es, 'schema-fingerprint')['structural_exact'] assert all(value == [] for value in single(es, 'collision-after').values()) before, after = single(es, 'business-before'), single(es, 'business-after') drift = {} for group in before: if before[group] != after[group]: assert lane == 'B' and group == 'aggregate' for key in before[group][0]: if before[group][0][key] != after[group][0][key]: assert key == 'update_time_max' assert [before[group][0][key], after[group][0][key]] == ['2026-10-11 05:11:01.000000', '2026-10-11 05:11:46.000000'] drift[key] = [before[group][0][key], after[group][0][key]] assert len(drift) == (1 if lane == 'B' else 0) s = single(es, 'summary') journal = single(es, 'read-journal') assert len(journal) == (470 if lane == 'B' else 504) assert all(q['ok'] and (q['SQL'].startswith('SELECT ') or q['SQL'] == 'SHOW WARNINGS') for q in journal) assert s['SELECT'] == (470 if lane == 'B' else 23 + 15 + 504) <= 650 assert s['DDL_attempts'] == len(ddl) and s['DML_attempts'] == 217 assert s['HTTP'] == s['business_table_writes'] == s['objects_remaining'] == 0 and s['completed'] assert s['Staff_metadata_aggregate_exact'] == (lane == 'C') if lane == 'C': gate = single(es, 'trigger-definition-gate') assert gate['body_byte_exact'] and gate['body_token_exact'] and gate['actual_body_bytes'] == gate['approved_body_bytes'] == 21268 assert gate['actual_body_sha256'] == gate['approved_body_sha256'] == '38e782714f1f7661e86ee40571bc598b71611a12637bf87bcbcf3ad3d4038afe' meta = single(es, 'metadata-created-final') assert len(meta['trigger_objects']) == 3 and all(o['status'] == 'VALID' for o in meta['trigger_objects']) assert len(meta['owned_trigger']) == 1 and meta['owned_trigger'][0]['status'] == 'Y' assert {i['index_name'] for i in meta['indexes']} == {'INDEX33556691', 'INDEX33556692', 'UX_CDX_SS7_2B_USERID'} summary[lane] = {'DML_attempts': 217, 'DML_success': success, 'DDL': len(ddl), 'SELECT': s['SELECT'], 'objects_remaining': 0, 'Staff_exact': lane == 'C', 'unattributed_external_drift': drift} assert all_semantics['B'] == all_semantics['C'] # Every legitimate case has exact 76 non-update-time cells; timestamps use per-lane clocks, not cross-host equality. for label in mutation_results['B']: b, c = mutation_results['B'][label], mutation_results['C'][label] if label not in ('ALL76_SELF', 'CLOB_NULL_INVALID'): assert b['time']['matched'] and c['time']['matched'] return {'evidence_verified': True, 'full_Staff_compatibility_passed': False, 'columns_each': 77, 'FIELD_actual_coverage': 76, 'cases_each': 108, 'legitimate_cases': 107, 'legitimate_time_equivalent': 106, 'all_76_other_stored_value_semantics_equal_every_operation': True, 'native_types_preserved_and_checked': True, 'native_legacy_format_differences_not_hidden': ['HEIGHT/MONEY B string C double', 'CREATE_TIME/PAL_MTIME B fsp0 native19 C fsp6 native26', 'UINT64 C native double loss versus exact server text'], 'lanes': summary, 'preserved_source_hypothesis_failure': 'ALL76_SELF B advances, C freezes with all other76 fields unchanged', 'invalid_CLOB_NULL': 'B warning1048/coerces empty/noop; C rejects -6609; physical rows unchanged', 'tool_errors_preserved': ['USER_ERRORS -2106', 'SYSDEPENDENCIES -5504'], 'Staff_business_DDL_DML_HTTP': 0} def main(): raw = (ROOT / 'probe2b-full77-time-proposal.json').read_bytes() assert hashlib.sha256(raw).hexdigest() == PLAN_SHA assert hashlib.sha256((ROOT / 'probe2b-C-recovery-overlay.json').read_bytes()).hexdigest() == OVERLAY_SHA plan = json.loads(raw) original_c = events('C-probe2b.jsonl') lanes = {'B': events('B-probe2b.jsonl'), 'C': events('C-continuation.jsonl')} preflight = json.loads((PRIVATE / 'C-recovery-preflight.json').read_text()) deps = json.loads((PRIVATE / 'C-recovery-dependencies.json').read_text()) user_deps = json.loads((PRIVATE / 'C-recovery-user-dependencies.json').read_text()) assert all(p.read_bytes() == b'' for p in PRIVATE.glob('*.stderr')) result = verify(plan, original_c, lanes, preflight, deps, user_deps) if '--test-mutations' in sys.argv: def operation(d, name, lane='C'): return next(e['data'] for e in d[lane] if e['kind'] == 'DML-result' and e['data']['operation'] == name) mutations = [ lambda d: operation(d, 'ALL76_SELF.mutation', 'B')['after'][0].update(update_time='2001-01-01 00:00:00'), lambda d: operation(d, 'ALL76_SELF.mutation')['after'][0].update(update_time='2026-10-11 05:22:48'), lambda d: operation(d, 'FIELD_DINGTALK_AVATAR.mutation')['after'][0].update(dingtalk_avatar='😀' * 1023), lambda d: operation(d, 'CLOB_LONG_LAST_CASE.mutation')['after'][0].update(sport_event_uuid=plan['payloads']['LOB_LONG_A']), lambda d: operation(d, 'CLOB_UTF8_CHUNK_LAST_CASE.mutation')['after'][0].update(sport_event_uuid=plan['payloads']['LOB_UTF8_A']), lambda d: operation(d, 'FIELD_DINGTALK_LAST_SEEN_FULL_BATCH_ID.mutation')['bindings'][0].update(type=1), lambda d: operation(d, 'FIELD_DINGTALK_LAST_SEEN_FULL_BATCH_ID.mutation')['after'][0].update(__num_dingtalk_last_seen_full_batch_id='18446744073709552000'), lambda d: operation(d, 'CHAR_TAIL_ONLY_NOOP.mutation')['after'][0].update(primary_department_uuid='P '), lambda d: operation(d, 'EXPLICIT_TIME_AND_CHANGE.mutation')['after'][0].update(update_time='2026-10-11 05:22:48'), lambda d: operation(d, 'CLOB_NULL_INVALID.mutation').update(ok=True), lambda d: single(d['C'], 'collision-after').update(indexes=[{'index_name': 'INDEX33556691'}]), lambda d: single(d['B'], 'business-after')['aggregate'][0].update(update_time_max='2026-10-11 05:11:01.000000'), lambda d: single(d['C'], 'metadata-created-final')['trigger_objects'][0].update(status='INVALID'), lambda d: operation(d, 'FIELD_NAME.mutation')['after'][0].update(station_uuid='UNRELATED_DRIFT'), ] for mutate in mutations: changed = copy.deepcopy(lanes) mutate(changed) try: verify(plan, original_c, changed, preflight, deps, user_deps) except AssertionError: continue raise AssertionError('Mutation not detected') result['mutations_rejected'] = len(mutations) manifest = PRIVATE / 'manifest.json' if manifest.exists(): entries = json.loads(manifest.read_text())['files'] assert {e['path'] for e in entries} == {'B-probe2b.jsonl', 'B-probe2b.stderr', 'C-probe2b.jsonl', 'C-probe2b.stderr', 'C-continuation.jsonl', 'C-continuation.stderr', 'C-recovery-preflight.json', 'C-recovery-preflight.stderr', 'C-recovery-dependencies.json', 'C-recovery-dependencies.stderr', 'C-recovery-user-dependencies.json', 'C-recovery-user-dependencies.stderr', 'gate.json'} for entry in entries: content = (PRIVATE / entry['path']).read_bytes() assert len(content) == entry['bytes'] and hashlib.sha256(content).hexdigest() == entry['sha256'] assert len(entries) == 13 print(json.dumps(result, ensure_ascii=False, indent=2)) if __name__ == '__main__': main()