#!/usr/bin/env python3
"""U86 公开入口：真实 Spark local 提交 + Jupyter kernel 生命周期 + 元数据/结果不一致。

前置（不满足则受阻退出）：
- 仓库外缓存有 Spark 4.0.1 发行包（首次运行自动下载并校验 SHA-512）
- python3 venv（脚本自建在缓存目录）安装 jupyter_client/ipykernel 与 PyMySQL
- 本机 MySQL（DBngin，独有库 study_u86_<runid>）

结束时停内核、DROP 数据库、删临时目录；清理不完整会使退出码非 0。
失败也写 evidence JSON（exitCode != 0），不伪装成功。
"""
import glob
import hashlib
import json
import os
import pathlib
import platform
import shutil
import signal
import subprocess
import sys
import tarfile
import time
import traceback
import urllib.request
from datetime import datetime

LAB = pathlib.Path(__file__).resolve().parent
CACHE = pathlib.Path.home() / '.cache/software-engineering-study/java-depth-impl/u86'
DOWNLOADS = CACHE / 'downloads'
VENV = CACHE / 'venv'
SPARK_DIR = CACHE / 'spark-4.0.1-bin-hadoop3'
SPARK_TGZ = DOWNLOADS / 'spark-4.0.1-bin-hadoop3.tgz'
SPARK_URL = 'https://archive.apache.org/dist/spark/spark-4.0.1/spark-4.0.1-bin-hadoop3.tgz'
SPARK_SHA512 = ('9198602c6b931b46686f32a25793b3bb58b522cd98a5b6a94d2484bae32e3e7b520d60'
                'f4bffe72ba29ff5c9ecd862443841ee47dde0f2f9e1bf52539f7baef41')
EVIDENCE = LAB / 'evidence' / 'u86-2026-10-28.json'

run_id = datetime.now().strftime('%H%M%S') + hashlib.sha1(
    str(time.time_ns()).encode()).hexdigest()[:4]
DB = f'study_u86_{run_id.lower()}'
RUNTIME = CACHE / 'runtime' / f'run-{run_id}'
MYSQL = {'bin': None, 'host': os.environ.get('MYSQL_HOST', '127.0.0.1'),
         'port': os.environ.get('MYSQL_PORT', '3306'),
         'user': os.environ.get('MYSQL_USER', 'root'),
         'password': os.environ.get('MYSQL_PASSWORD', '')}
JAVA_HOME = subprocess.check_output(['/usr/libexec/java_home', '-v', '21'],
                                    text=True).strip()
SPARK_BIN = None
procs = []
observations = []
scenarios = []
errors = []
started_all = datetime.now().astimezone().isoformat()


class Blocked(SystemExit):
    pass


def say(line):
    print(line, flush=True)
    observations.append(line)


def phase(name):
    print(f'\n===== {name} =====', flush=True)


def record(name, observed, started):
    scenarios.append({'name': name, 'startedAt': started,
                      'finishedAt': datetime.now().astimezone().isoformat(),
                      'observed': observed})


def die(message):
    raise Blocked(f'U86 BLOCKED/FAILED: {message}')


def assert_true(cond, message):
    if not cond:
        raise AssertionError(message)


def sha256_file(path):
    digest = hashlib.sha256()
    with open(path, 'rb') as handle:
        for chunk in iter(lambda: handle.read(1 << 20), b''):
            digest.update(chunk)
    return digest.hexdigest()


def sha512_file(path):
    digest = hashlib.sha512()
    with open(path, 'rb') as handle:
        for chunk in iter(lambda: handle.read(1 << 20), b''):
            digest.update(chunk)
    return digest.hexdigest()


def mysql_bin():
    for candidate in filter(None, [os.environ.get('MYSQL_BIN'), shutil.which('mysql'),
                                   *sorted(glob.glob('/Users/Shared/DBngin/mysql/*/bin/mysql')),
                                   '/opt/homebrew/bin/mysql', '/usr/local/bin/mysql']):
        if pathlib.Path(candidate).is_file():
            return candidate
    die('未找到 mysql 客户端：设置 MYSQL_BIN')


def mysql(sql, db=None):
    args = [MYSQL['bin'], '-h', MYSQL['host'], '-P', MYSQL['port'], '-u', MYSQL['user'],
            '--batch', '--default-character-set=utf8mb4']
    if MYSQL['password']:
        args.insert(1, f"-p{MYSQL['password']}")
    if db:
        args.append(db)
    args += ['-e', sql]
    done = subprocess.run(args, capture_output=True, text=True, timeout=180)
    if done.returncode != 0:
        raise RuntimeError(f'mysql failed: {sql[:120]} → {done.stderr.strip()[:300]}')
    return done.stdout


def sql_rows(sql):
    out = mysql(sql, db=DB).strip().splitlines()
    return out[1:] if out else []


def ensure_spark():
    if SPARK_DIR.is_dir() and (SPARK_DIR / 'bin' / 'spark-submit').is_file():
        return 'cached'
    DOWNLOADS.mkdir(parents=True, exist_ok=True)
    if not SPARK_TGZ.is_file() or sha512_file(SPARK_TGZ) != SPARK_SHA512:
        say('U86 downloading Spark 4.0.1 (~550MB, one-time; interrupted downloads retry)...')
        tmp = SPARK_TGZ.with_suffix('.part')
        for attempt in range(5):
            try:
                urllib.request.urlretrieve(SPARK_URL, tmp)
                break
            except Exception as exc:  # noqa: BLE001
                say(f'U86 download retry {attempt + 1}: {exc}')
                time.sleep(5)
        if sha512_file(tmp) != SPARK_SHA512:
            die('Spark 发行包 SHA-512 校验失败')
        tmp.rename(SPARK_TGZ)
    with tarfile.open(SPARK_TGZ) as tar:
        tar.extractall(CACHE)
    assert_true((SPARK_DIR / 'bin' / 'spark-submit').is_file(), '解压后应有 spark-submit')
    return 'extracted'


def ensure_venv():
    if (VENV / 'bin' / 'python').is_file():
        return
    subprocess.run([sys.executable, '-m', 'venv', str(VENV)], check=True)
    subprocess.run([str(VENV / 'bin' / 'python'), '-m', 'pip', 'install', '--quiet',
                    'jupyter_client==8.6.3', 'ipykernel==6.31.0', 'pymysql==1.1.2'],
                   check=True)


def kill_group(proc, sig, wait):
    if proc is None or proc.poll() is not None:
        return
    try:
        os.killpg(proc.pid, sig)
    except ProcessLookupError:
        pass
    try:
        proc.wait(timeout=wait)
    except subprocess.TimeoutExpired:
        try:
            os.killpg(proc.pid, signal.SIGKILL)
        except ProcessLookupError:
            pass
        proc.wait(timeout=10)


def wait_meta(job_id, status, timeout=180):
    deadline = time.time() + timeout
    last = None
    while time.time() < deadline:
        rows = sql_rows(f"SELECT status FROM job_meta WHERE job_id='{job_id}'")
        last = rows[0].split('\t')[0] if rows else None
        if last == status:
            return True
        time.sleep(1)
    return False


def spark_env(event_log_dir=None):
    env = dict(os.environ, JAVA_HOME=JAVA_HOME)
    env['SPARK_LOCAL_DIRS'] = str(RUNTIME / 'spark-local')
    # PySpark driver 跑在系统 python3 上；把 venv 的 site-packages（含 pymysql）
    # 通过 PYTHONPATH 提供给任务，不改动用户全局环境。
    site = subprocess.run([str(VENV / 'bin' / 'python'), '-c',
                           'import site; print(site.getsitepackages()[0])'],
                          capture_output=True, text=True, check=True).stdout.strip()
    env['PYTHONPATH'] = site + os.pathsep + env.get('PYTHONPATH', '')
    if event_log_dir:
        env['SPARK_EVENTLOG_ENABLED'] = 'true'
    return env


def write_spark_task(name, payload, meta_status_before_run='RUNNING',
                     fail=False, mark_success_first=False, delete_result_after=False):
    """生成一个 PySpark 任务脚本：跑真实 SparkSession 聚合，经 pymysql 写结果与元表。"""
    task_dir = RUNTIME / 'tasks'
    task_dir.mkdir(parents=True, exist_ok=True)
    script = task_dir / f'{name}.py'
    script.write_text(f'''
import sys
from datetime import datetime
from pyspark.sql import SparkSession
import pymysql

DB = {DB!r}
JOB_ID = {name!r}
PAYLOAD = {payload!r}
FAIL = {fail!r}
MARK_SUCCESS_FIRST = {mark_success_first!r}
DELETE_RESULT_AFTER = {delete_result_after!r}
CONN = dict(host={MYSQL['host']!r}, port={MYSQL['port']}, user={MYSQL['user']!r},
            password={MYSQL['password']!r}, database=DB, charset='utf8mb4')


def meta(status):
    conn = pymysql.connect(**CONN)
    with conn.cursor() as cur:
        cur.execute("UPDATE job_meta SET status=%s, updated_at=NOW() WHERE job_id=%s",
                    (status, JOB_ID))
    conn.commit()
    conn.close()


meta('RUNNING')
if MARK_SUCCESS_FIRST:
    # 反例：在真实执行前就把元表写成 SUCCESS（先标记后执行的错误写法）
    meta('SUCCESS')
spark = (SparkSession.builder.appName('study-u86-' + JOB_ID)
         .master('local[2]')
         .config('spark.sql.shuffle.partitions', '2')
         .getOrCreate())
try:
    if FAIL:
        # 真实失败：聚合能跑，随后抛异常让 driver 以非零退出
        df = spark.createDataFrame([(1, 'a'), (2, 'b')], ['id', 'v'])
        df.count()
        raise RuntimeError('planned spark failure inside executor task')
    df = spark.createDataFrame([(1, 'x'), (2, 'y'), (3, 'x')], ['id', 'v'])
    counted = df.count()
    distinct = df.select('v').distinct().count()
    if MARK_SUCCESS_FIRST:
        meta('SUCCESS')
    conn = pymysql.connect(**CONN)
    with conn.cursor() as cur:
        cur.execute('REPLACE INTO job_results (job_id, payload, written_at) '
                    'VALUES (%s, %s, NOW())',
                    (JOB_ID, PAYLOAD + ' rows=%d distinct=%d' % (counted, distinct)))
        if not MARK_SUCCESS_FIRST:
            cur.execute("UPDATE job_meta SET status='SUCCESS', updated_at=NOW() WHERE job_id=%s",
                        (JOB_ID,))
    conn.commit()
    conn.close()
    if DELETE_RESULT_AFTER:
        conn = pymysql.connect(**CONN)
        with conn.cursor() as cur:
            cur.execute('DELETE FROM job_results WHERE job_id=%s', (JOB_ID,))
        conn.commit()
        conn.close()
finally:
    spark.stop()
''', encoding='utf-8')
    return script


def submit_spark(name, script, timeout=300, confs=()):
    log = open(RUNTIME / f'spark-{name}.log', 'w')
    cmd = [SPARK_BIN]
    for conf in confs:
        cmd += ['--conf', conf]
    cmd.append(str(script))
    proc = subprocess.Popen(cmd, stdout=log, stderr=subprocess.STDOUT,
                            env=spark_env(), start_new_session=True)
    procs.append(proc)
    proc.wait(timeout=timeout)
    log.close()
    return proc.returncode


def scenario_spark_success():
    phase('场景1：调度元数据 + Spark local 真实执行 + 结果入库（成功闭环）')
    started = datetime.now().astimezone().isoformat()
    job = f'ok-{run_id}'
    mysql(f"INSERT INTO job_meta (job_id, engine, status, submitted_at, updated_at) "
          f"VALUES ('{job}', 'spark', 'SUBMITTED', NOW(), NOW())", db=DB)
    script = write_spark_task(job, 'success-payload')
    code = submit_spark('ok', script)
    rows = sql_rows(f"SELECT status FROM job_meta WHERE job_id='{job}'")
    results = sql_rows(f"SELECT payload FROM job_results WHERE job_id='{job}'")
    status = rows[0].split('\t')[0] if rows else None
    payload = results[0].split('\t')[0] if results else None
    observed = [f'U86 sparkOk exit={code} meta={status} result={payload[:60] if payload else None}']
    assert_true(code == 0, 'spark-submit 应成功退出')
    assert_true(status == 'SUCCESS', '元表应为 SUCCESS')
    assert_true(payload and 'rows=3' in payload and 'distinct=2' in payload,
                '结果行应有真实聚合产物')
    record('spark-success-closed-loop', observed, started)


def scenario_meta_success_result_missing():
    phase('场景2：元数据 SUCCESS 但结果行丢失（元数据/结果不一致反例）')
    started = datetime.now().astimezone().isoformat()
    job = f'lost-{run_id}'
    mysql(f"INSERT INTO job_meta (job_id, engine, status, submitted_at, updated_at) "
          f"VALUES ('{job}', 'spark', 'SUBMITTED', NOW(), NOW())", db=DB)
    script = write_spark_task(job, 'lost-payload', delete_result_after=True)
    code = submit_spark('lost', script)
    meta = sql_rows(f"SELECT status FROM job_meta WHERE job_id='{job}'")
    results = sql_rows(f"SELECT payload FROM job_results WHERE job_id='{job}'")
    meta_status = meta[0].split('\t')[0] if meta else None
    observed = [
        f'U86 metaResultMissing exit={code} meta={meta_status} resultRows={len(results)}',
        'U86 metaResultMissing conclusion=调度侧只看元表会误判成功；结果表才是权威产物',
    ]
    assert_true(meta_status == 'SUCCESS', '元表仍标记成功')
    assert_true(len(results) == 0, '结果行应已被删除（不一致状态成立）')
    record('meta-success-result-missing', observed, started)


def scenario_failed_but_meta_success():
    phase('场景3：Spark 任务失败但元表误标 SUCCESS，再演示按退出码收敛')
    started = datetime.now().astimezone().isoformat()
    job = f'bad-{run_id}'
    mysql(f"INSERT INTO job_meta (job_id, engine, status, submitted_at, updated_at) "
          f"VALUES ('{job}', 'spark', 'SUBMITTED', NOW(), NOW())", db=DB)
    script = write_spark_task(job, 'never', mark_success_first=True, fail=True)
    code = submit_spark('bad', script)
    meta = sql_rows(f"SELECT status FROM job_meta WHERE job_id='{job}'")
    results = sql_rows(f"SELECT payload FROM job_results WHERE job_id='{job}'")
    meta_status = meta[0].split('\t')[0] if meta else None
    # 对账：以 spark-submit 退出码为准修正元表
    if code != 0:
        mysql(f"UPDATE job_meta SET status='FAILED', updated_at=NOW() WHERE job_id='{job}'", db=DB)
    meta_fixed = sql_rows(f"SELECT status FROM job_meta WHERE job_id='{job}'")
    observed = [
        f'U86 failedMeta exit={code} metaDuringRun={meta_status} resultRows={len(results)} '
        f'metaAfterReconcile={meta_fixed[0].split(chr(9))[0] if meta_fixed else None}',
        'U86 failedMeta conclusion=先写成功再执行会把失败伪装成成功；对账以进程退出码与结果存在性为准',
    ]
    assert_true(code != 0, 'spark-submit 应失败退出')
    assert_true(meta_status == 'SUCCESS', '失败前元表被误标成功')
    assert_true(len(results) == 0, '失败任务不应有结果行')
    record('failed-but-meta-success', observed, started)


def scenario_kernel_lifecycle():
    phase('场景4：Jupyter kernel 生命周期：start/execute/interrupt/shutdown')
    started = datetime.now().astimezone().isoformat()
    script = RUNTIME / 'kernel_lifecycle.py'
    script.write_text(f'''
import json, time, signal, os
from jupyter_client.manager import start_new_kernel

out = {{}}
km, kc = start_new_kernel(kernel_name='python3')
try:
    proc = km.provisioner.process
    out['kernel_pid_alive'] = proc is not None and proc.poll() is None
    # 正常执行
    msg_id = kc.execute('answer = 41 + 1; answer')
    result = None
    deadline = time.time() + 60
    while time.time() < deadline:
        msg = kc.get_iopub_msg(timeout=10)
        if msg['parent_header'].get('msg_id') != msg_id:
            continue
        if msg['msg_type'] == 'execute_result':
            result = msg['content']['data'].get('text/plain')
        if msg['msg_type'] == 'status' and msg['content']['execution_state'] == 'idle':
            break
    reply = kc.get_shell_msg(timeout=10)
    out['exec_result'] = result
    out['exec_status'] = reply['content']['status']

    # interrupt：运行中的 sleep 应被中断
    slow_id = kc.execute('import time; time.sleep(60)')
    time.sleep(2.0)
    t0 = time.time()
    km.interrupt_kernel()
    interrupt_reply = None
    deadline = time.time() + 30
    while time.time() < deadline:
        msg = kc.get_shell_msg(timeout=10)
        if msg['parent_header'].get('msg_id') == slow_id:
            interrupt_reply = msg['content']
            break
    out['interrupt_status'] = interrupt_reply['status'] if interrupt_reply else None
    out['interrupt_seconds'] = round(time.time() - t0, 1)

    # 变量跨 cell 存活（同一 kernel）：实际校验 print(mark) 的 stdout 内容
    kc.execute('mark = "kept"')
    print_id = kc.execute('print(mark)')
    stdout_text, print_status = '', None
    deadline = time.time() + 30
    while time.time() < deadline:
        msg = kc.get_iopub_msg(timeout=10)
        if msg['parent_header'].get('msg_id') != print_id:
            continue
        if msg['msg_type'] == 'stream':
            stdout_text += msg['content'].get('text', '')
        if msg['msg_type'] == 'status' and msg['content']['execution_state'] == 'idle':
            break
    reply_p = kc.get_shell_msg(timeout=10)
    print_status = reply_p['content']['status']
    out['state_kept'] = ('kept' in stdout_text and print_status == 'ok')
    out['state_stdout'] = stdout_text.strip()

    # shutdown：内核进程应退出
    km.shutdown_kernel(now=False)
    deadline = time.time() + 20
    while time.time() < deadline:
        if proc.poll() is not None:
            break
        time.sleep(0.3)
    out['kernel_exitcode_after_shutdown'] = proc.poll()
finally:
    try:
        kc.stop_channels()
    except Exception:
        pass
print('U86KERNEL ' + json.dumps(out))
''', encoding='utf-8')
    log = open(RUNTIME / 'kernel-lifecycle.log', 'w')
    proc = subprocess.Popen([str(VENV / 'bin' / 'python'), str(script)],
                            stdout=log, stderr=subprocess.STDOUT, start_new_session=True)
    procs.append(proc)
    proc.wait(timeout=240)
    log.close()
    text = (RUNTIME / 'kernel-lifecycle.log').read_text()
    line = next((l for l in text.splitlines() if l.startswith('U86KERNEL ')), None)
    assert_true(line, '应有 U86KERNEL 观察行')
    data = json.loads(line[len('U86KERNEL '):])
    observed = [f'U86 kernelLifecycle {k}={v}' for k, v in data.items()]
    assert_true(data.get('exec_status') == 'ok' and data.get('exec_result') == '42',
                'kernel 应真实执行 41+1=42')
    assert_true(data.get('interrupt_status') == 'error',
                '被中断的 sleep 应返回 error 状态')
    assert_true(data.get('interrupt_seconds', 99) < 15, '中断应迅速返回')
    assert_true(data.get('state_kept') is True, '同一 kernel 内变量应跨 cell 保留')
    assert_true(data.get('kernel_exitcode_after_shutdown') == 0,
                'shutdown 后内核进程应以 0 优雅退出')
    record('kernel-lifecycle', observed, started)


def scenario_kernel_crash():
    phase('场景5：kernel SIGKILL 崩溃：会话记录仍在，状态丢失')
    started = datetime.now().astimezone().isoformat()
    script = RUNTIME / 'kernel_crash.py'
    script.write_text(f'''
import json, time, os, signal
from jupyter_client.manager import start_new_kernel

out = {{}}
km, kc = start_new_kernel(kernel_name='python3')
proc = km.provisioner.process
pid = proc.pid
msg_id = kc.execute('session_state = "filled"; 7 * 6')
deadline = time.time() + 30
while time.time() < deadline:
    msg = kc.get_iopub_msg(timeout=10)
    if msg['parent_header'].get('msg_id') == msg_id and msg['msg_type'] == 'execute_result':
        out['last_result'] = msg['content']['data'].get('text/plain')
        break
# 真实落一份"会话执行记录"到磁盘（模拟 notebook/审计记录），崩溃后读回
record_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'session-record.json')
with open(record_path, 'w') as handle:
    json.dump([{{'cell': 1, 'status': 'ok', 'result': out.get('last_result')}}], handle)
# SIGKILL：模拟 kernel 崩溃（不是优雅 shutdown）
os.kill(pid, signal.SIGKILL)
deadline = time.time() + 20
while time.time() < deadline:
    if proc.poll() is not None:
        break
    time.sleep(0.2)
out['exit_after_kill'] = proc.poll()
out['is_alive_after_kill'] = km.is_alive()
# 从磁盘读回会话记录：它仍记着"已成功执行"这条 cell
with open(record_path) as handle:
    record = json.load(handle)
out['session_record_still_says_success'] = (record and record[0]['status'] == 'ok')
out['session_record_cell_result'] = record[0]['result'] if record else None
try:
    msg_id = kc.execute('len(session_state)')
    reply = kc.get_shell_msg(timeout=15)
    out['new_exec_status'] = reply['content']['status']
except Exception as exc:
    out['new_exec_status'] = f'error: {{type(exc).__name__}}'
try:
    kc.stop_channels()
except Exception:
    pass
print('U86KERNELCRASH ' + json.dumps(out))
''', encoding='utf-8')
    log = open(RUNTIME / 'kernel-crash.log', 'w')
    proc = subprocess.Popen([str(VENV / 'bin' / 'python'), str(script)],
                            stdout=log, stderr=subprocess.STDOUT, start_new_session=True)
    procs.append(proc)
    proc.wait(timeout=240)
    log.close()
    text = (RUNTIME / 'kernel-crash.log').read_text()
    line = next((l for l in text.splitlines() if l.startswith('U86KERNELCRASH ')), None)
    assert_true(line, '应有 U86KERNELCRASH 观察行')
    data = json.loads(line[len('U86KERNELCRASH '):])
    observed = [f'U86 kernelCrash {k}={v}' for k, v in data.items()]
    assert_true(data.get('exit_after_kill') is not None and
                data.get('exit_after_kill') != 0, 'SIGKILL 后内核进程应消失')
    assert_true(data.get('session_record_still_says_success') is True,
                '会话记录仍显示上一次成功执行')
    assert_true(str(data.get('new_exec_status', '')).startswith('error:'),
                '崩溃后同一 channel 上的执行应报错')
    assert_true('Empty' in str(data.get('new_exec_status', '')),
                '崩溃后的错误应是连接 Empty（断连），而不是别的异常')
    record('kernel-crash-state-lost', observed, started)


def cleanup():
    phase('清理')
    for proc in procs:
        kill_group(proc, signal.SIGTERM, 5)
    for proc in procs:
        if proc.poll() is None:
            kill_group(proc, signal.SIGKILL, 10)
    survivors = [p.pid for p in procs if p.poll() is None]
    dropped = False
    drop_err = None
    try:
        mysql(f'DROP DATABASE IF EXISTS `{DB}`')
        dropped = True
    except Exception as exc:
        drop_err = str(exc)
    # 孤儿库（此前失败运行残留）按 U86 命名空间定向清理：这是本实验自己的库，不是用户数据
    orphan_rows = [row for row in
                   mysql("SHOW DATABASES LIKE 'study\\_u86\\_%'").strip().splitlines()[1:]
                   if row]
    orphans_dropped = []
    for row in orphan_rows:
        name = row.strip()
        try:
            mysql(f'DROP DATABASE IF EXISTS `{name}`')
            orphans_dropped.append(name)
        except Exception as exc:  # noqa: BLE001
            errors.append(f'孤儿库清理失败 {name}: {exc}')
    result = {'processesSurviving': survivors, 'databaseDropped': dropped,
              'dropError': drop_err, 'orphanDatabasesDropped': orphans_dropped,
              'labDatabasesRemaining': [r for r in
                                        mysql("SHOW DATABASES LIKE 'study\\_u86\\_%'")
                                        .strip().splitlines()[1:] if r],
              'runtimeDir': str(RUNTIME)}
    if result['processesSurviving'] or not result['databaseDropped'] or result['labDatabasesRemaining']:
        errors.append(f'清理不完整: {result}')
    return result



def scenario_spark_eventlog():
    phase('场景6：Spark 事件日志观测 job/stage/task 状态与 submission 状态')
    started = datetime.now().astimezone().isoformat()
    event_dir = RUNTIME / 'eventlog'
    event_dir.mkdir(parents=True, exist_ok=True)
    script = RUNTIME / 'tasks' / 'eventlog.py'
    script.parent.mkdir(parents=True, exist_ok=True)
    script.write_text("""
from pyspark.sql import SparkSession
spark = (SparkSession.builder.appName('study-u86-eventlog').master('local[2]')
         .config('spark.sql.shuffle.partitions', '2').getOrCreate())
df = spark.createDataFrame([(i % 3, f'v{i}') for i in range(30)], ['k', 'v'])
counts = df.groupBy('k').count().orderBy('k').collect()
print('EVENTLOG_RESULT', counts)
spark.stop()
""", encoding='utf-8')
    code = submit_spark('eventlog', script, timeout=300, confs=(
        f'spark.eventLog.enabled=true',
        f'spark.eventLog.dir=file://{event_dir}',
        'spark.eventLog.logBlockUpdates.enabled=false',
    ))
    assert_true(code == 0, 'eventlog 任务应成功')
    # Spark 4 的 eventLog 是滚动 v2 格式：目录 eventlog_v2_<app>/events_<n>_<app>.zstd
    candidates = sorted(event_dir.rglob('events_*.zstd'))
    assert_true(candidates, '应生成事件日志文件（Spark 4 v2 滚动格式）')
    event_file = candidates[-1]
    decompressed = RUNTIME / 'eventlog.jsonl'
    decompressed.write_bytes(subprocess.run(['zstd', '-dc', str(event_file)],
                                            capture_output=True, check=True).stdout)
    kinds = {}
    jobs, stages, tasks = {}, {}, []
    app_end = False
    for line in decompressed.read_text().splitlines():
        try:
            ev = json.loads(line)
        except Exception:
            continue
        kind = ev.get('Event')
        kinds[kind] = kinds.get(kind, 0) + 1
        if kind == 'SparkListenerJobStart':
            jobs[ev['Job ID']] = 'STARTED'
        elif kind == 'SparkListenerJobEnd':
            jobs[ev['Job ID']] = ev['Job Result'].get('Result')
        elif kind == 'SparkListenerStageSubmitted':
            stages[ev['Stage Info']['Stage ID']] = 'SUBMITTED'
        elif kind == 'SparkListenerStageCompleted':
            stages[ev['Stage Info']['Stage ID']] = 'COMPLETED'
        elif kind == 'SparkListenerTaskEnd':
            info = ev['Task Info']
            tasks.append('FAILED' if info.get('Failed') else
                         ('KILLED' if info.get('Killed') else 'SUCCESS'))
        elif kind == 'SparkListenerApplicationEnd':
            app_end = True
    sql_event_count = sum(v for k, v in kinds.items() if k.endswith('SQLExecutionStart'))
    observed = [
        f'U86 sparkEventLog exit={code} file={event_file.name} events={sum(kinds.values())}',
        f'U86 sparkEventLog jobs={len(jobs)} jobResults={sorted(set(jobs.values()))}',
        f'U86 sparkEventLog stages={len(stages)} stageStates={sorted(set(stages.values()))}',
        f'U86 sparkEventLog tasks={len(tasks)} taskStatuses={sorted(set(tasks))}',
        f'U86 sparkEventLog appEnd={app_end} sqlEvents={sql_event_count}',
    ]
    assert_true(jobs and all(v == 'JobSucceeded' for v in jobs.values()), '事件日志应记录全部 job 成功')
    assert_true(len(stages) >= 2, 'groupBy shuffle 应产生至少两个 stage')
    assert_true(tasks and all(status == 'SUCCESS' for status in tasks), 'task 状态应为 SUCCESS')
    assert_true(app_end, '应记录 ApplicationEnd')
    record('spark-eventlog-states', observed, started)


def run():
    global SPARK_BIN
    exit_code = 0
    failure = None
    environment = {}
    cleanup_result = None
    try:
        MYSQL['bin'] = mysql_bin()
        RUNTIME.mkdir(parents=True, exist_ok=True)
        mysql_version = mysql('SELECT VERSION()').splitlines()[1]
        mysql(f'CREATE DATABASE IF NOT EXISTS `{DB}` DEFAULT CHARACTER SET utf8mb4')
        mysql(f'CREATE TABLE {DB}.job_meta ('
              'job_id VARCHAR(64) PRIMARY KEY, engine VARCHAR(32) NOT NULL, '
              'status VARCHAR(16) NOT NULL, submitted_at DATETIME NOT NULL, '
              'updated_at DATETIME NOT NULL)')
        mysql(f'CREATE TABLE {DB}.job_results ('
              'job_id VARCHAR(64) PRIMARY KEY, payload VARCHAR(255) NOT NULL, '
              'written_at DATETIME NOT NULL)')
        spark_state = ensure_spark()
        ensure_venv()
        jupyter_version = subprocess.run(
            [str(VENV / 'bin' / 'python'), '-c',
             'import jupyter_client, ipykernel; print(jupyter_client.__version__, ipykernel.__version__)'],
            capture_output=True, text=True, check=True).stdout.strip()
        SPARK_BIN = str(SPARK_DIR / 'bin' / 'spark-submit')
        version_run = subprocess.run([SPARK_BIN, '--version'], capture_output=True, text=True,
                                     env=dict(os.environ, JAVA_HOME=JAVA_HOME))
        version_text = version_run.stderr or version_run.stdout
        version_line = next((l for l in version_text.splitlines() if 'version' in l.lower()), '?')
        environment = {
            'os': platform.platform(), 'arch': platform.machine(),
            'jdk': subprocess.run([f'{JAVA_HOME}/bin/java', '-version'],
                                  capture_output=True, text=True).stderr.splitlines()[0],
            'mysql': mysql_version,
            'spark': version_line.strip(),
            'sparkTarSha512Prefix': SPARK_SHA512[:16],
            'sparkTarState': spark_state,
            'jupyter': jupyter_version,
            'externalServices': f'MySQL {MYSQL["host"]}:{MYSQL["port"]} 独有库 {DB}；'
                                'Spark local[2] 本机；无 HDFS/YARN/集群',
        }
        say(f'U86 preflight db={DB} spark={version_line.strip()} jupyter={jupyter_version}')

        scenario_spark_success()
        scenario_meta_success_result_missing()
        scenario_failed_but_meta_success()
        scenario_spark_eventlog()
        scenario_kernel_lifecycle()
        scenario_kernel_crash()
        say('U86 scenarios completed')
    except SystemExit as exc:
        exit_code = 1
        failure = str(exc)
        errors.append(failure)
        print(failure, flush=True)
    except Exception as exc:  # noqa: BLE001
        exit_code = 1
        failure = f'{type(exc).__name__}: {exc}'
        errors.append(failure)
        traceback.print_exc()
    finally:
        try:
            cleanup_result = cleanup()
        except Exception as cleanup_exc:  # noqa: BLE001
            errors.append(f'清理异常: {cleanup_exc}')
            cleanup_result = {'error': str(cleanup_exc)}
            exit_code = exit_code or 1
        if cleanup_result and (cleanup_result['processesSurviving'] or
                               not cleanup_result['databaseDropped'] or
                               cleanup_result['labDatabasesRemaining']):
            exit_code = exit_code or 1

    evidence = {
        'ticket': 'U86 #123',
        'generatedAt': datetime.now().astimezone().isoformat(),
        'startedAt': started_all, 'runId': run_id,
        'input': 'python3 labs/data-engine/run.py',
        'environment': environment, 'scenarios': scenarios, 'observed': observations,
        'exitCode': exit_code, 'failure': failure, 'errors': errors,
        'rawLogs': str(RUNTIME), 'cleanup': cleanup_result,
        'runnerSha256': sha256_file(pathlib.Path(__file__).resolve()),
        'limits': [
            'Spark 以 local[2] 单机模式运行；无 HDFS/YARN/Standalone 集群，集群模式行为不外推。',
            'Flink 与 Hive 未运行：提交/监控差异按固定版本官方文档与源码对照说明，在正文与账本单独标注。',
            '元数据/结果不一致通过本机 MySQL 元表构造：job_meta 模拟调度侧登记，job_results 模拟引擎产物；'
            '不是某个生产调度系统的真实 bug 复现。',
            'kernel 崩溃通过 SIGKILL 注入；只覆盖"进程死→变量丢失"，不代表 JupyterHub/网关的多用户行为。',
            'PySpark driver 进程经 pymysql 写回 MySQL 元表与结果表，不是 Spark JDBC 数据源写入；pymysql 版本见 environment。',
            'MySQL 为本机 DBngin 8.4.x 独有库，结束 DROP；所有任务数据为合成标签，不含用户数据。',
        ],
    }
    # 先写到本次 runtime 目录；只有成功运行才复制为仓库内的公开证据，避免失败覆盖已提交的成功证据
    local = RUNTIME / 'evidence.json'
    local.write_text(json.dumps(evidence, ensure_ascii=False, indent=2) + '\n')
    if exit_code == 0:
        EVIDENCE.parent.mkdir(parents=True, exist_ok=True)
        shutil.copyfile(local, EVIDENCE)
        print(f'\nU86 exit=0 evidence={EVIDENCE}', flush=True)
    else:
        print(f'\nU86 exit={exit_code} evidence(local)={local}', flush=True)
    return exit_code


if __name__ == '__main__':
    sys.exit(run())