"""Same-session encoding proof; execute before every data SELECT and fail closed."""
import json
SENTINEL='中文😀 café'
HEX=SENTINEL.encode('utf8').hex().upper()
PROBE_SQL="SELECT JSON_OBJECT('__encoding_probe',1,'client',@@character_set_client,'connection',@@character_set_connection,'results',@@character_set_results,'text','"+SENTINEL+"','hex',HEX('"+SENTINEL+"'),'server_text',CONVERT(0x"+HEX+" USING utf8mb4));"
def validate_encoding(row):
 if not isinstance(row,dict) or row.get('__encoding_probe')!=1:raise RuntimeError('Missing same-session encoding probe')
 if any(row.get(k)!='utf8mb4' for k in ('client','connection','results')):raise RuntimeError('MySQL session is not utf8mb4 for client/connection/results')
 if row.get('text')!=SENTINEL or row.get('server_text')!=SENTINEL or row.get('hex')!=HEX:raise RuntimeError('MySQL non-ASCII roundtrip failed')
 return row

def consume_probe(process):
 try:
  line=process.stdout.readline()
  if not line:raise RuntimeError('MySQL returned no encoding probe')
  return validate_encoding(json.loads(line.decode('utf8','strict')))
 except BaseException:
  process.terminate();process.wait()
  raise