#!/usr/bin/env python3
"""U85 公开入口：真实 xxl-job 3.5.0 admin 与 frameless 执行器，跑有界调度场景。

前置（不满足则明确受阻退出，不伪造通过）：
- 本机可连接的 MySQL 8.x（默认 127.0.0.1:3306 root 无密码；DBngin 路径自动探测，
  可用 MYSQL_HOST/PORT/USER/PASSWORD/MYSQL_BIN 覆盖）
- JDK 17+（优先 JAVA_HOME 或 /usr/libexec/java_home -v 21）与 Maven 3.9.x
- 首次运行需要网络：拉取固定 tag 源码与官方建表 SQL（之后走 ~/.cache 缓存）

每次运行创建独有数据库 study_u85_xxl_<runid>、独有执行器 appname、随机端口与
缓存目录下独立日志/标记目录，结束时停进程、删数据库；不触碰用户既有库与配置。
无论成功失败都写 evidence/u85-2026-10-27.json（含 exitCode、场景观察与清理结果）。
"""
import glob
import hashlib
import http.cookiejar
import json
import os
import pathlib
import platform
import re
import shutil
import signal
import socket
import subprocess
import sys
import time
import traceback
import urllib.parse
import urllib.request
from datetime import datetime

LAB = pathlib.Path(__file__).resolve().parent
CACHE = pathlib.Path.home() / '.cache/software-engineering-study/java-depth-impl/u85'
UPSTREAM = CACHE / 'upstream' / 'xxl-job-3.5.0'
ADMIN_JAR = UPSTREAM / 'xxl-job-admin' / 'target' / 'xxl-job-admin-3.5.0.jar'
UPSTREAM_TAG = '3.5.0'
UPSTREAM_COMMIT = '3bebfc1609b95413581b11e270e5670dce679813'
SCHEMA_URL = ('https://raw.githubusercontent.com/xuxueli/xxl-job/'
              f'{UPSTREAM_TAG}/doc/db/tables_xxl_job.sql')
SCHEMA_CACHE = CACHE / 'schema' / 'tables_xxl_job.sql'
EXECUTOR_JAR = LAB / 'executor' / 'target' / 'data-scheduler-executor-1.0.0.jar'
EXECUTOR_CP = LAB / 'executor' / 'target' / 'classpath.txt'
EVIDENCE = LAB / 'evidence' / 'u85-2026-10-27.json'

run_id = datetime.now().strftime('%H%M%S') + hashlib.sha1(
    str(time.time_ns()).encode()).hexdigest()[:4]
DB = f'study_u85_xxl_{run_id.lower()}'
APPNAME = f'study-u85-{run_id.lower()}'
TOKEN = 'study_u85_token'
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', '')}
JDK = None
MVN = None
GROUP_ID = 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'U85 BLOCKED/FAILED: {message}')


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


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',
                                   '/usr/local/mysql/bin/mysql']):
        if pathlib.Path(candidate).is_file():
            return candidate
    die('未找到 mysql 客户端：设置 MYSQL_BIN 指向 MySQL 8.x 的 mysql 可执行文件')


def mysql(sql, db=None):
    # 本机 LANG 为空时客户端默认 latin1，中文备注会变问号；显式走 utf8mb4
    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=120)
    if done.returncode != 0:
        raise RuntimeError(f'mysql failed: {sql[:140]} → {done.stderr.strip()[:400]}')
    return done.stdout


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


def free_port():
    with socket.socket() as s:
        s.bind(('127.0.0.1', 0))
        return s.getsockname()[1]


def sha256(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 java_home():
    candidate = os.environ.get('JAVA_HOME')
    if candidate and pathlib.Path(candidate, 'bin', 'java').is_file():
        return candidate
    try:
        return subprocess.check_output(['/usr/libexec/java_home', '-v', '21'],
                                       text=True).strip()
    except subprocess.CalledProcessError:
        return ''


def run_cmd(args, **kwargs):
    return subprocess.run(args, capture_output=True, text=True, **kwargs)


def ensure_schema():
    SCHEMA_CACHE.parent.mkdir(parents=True, exist_ok=True)
    if not SCHEMA_CACHE.is_file():
        with urllib.request.urlopen(SCHEMA_URL, timeout=60) as response:
            SCHEMA_CACHE.write_bytes(response.read())
    return SCHEMA_CACHE.read_text()


def apply_schema(text):
    patched = re.sub(r'CREATE DATABASE IF NOT EXISTS `xxl_job`[^;]*;',
                     f'CREATE DATABASE IF NOT EXISTS `{DB}` DEFAULT CHARACTER SET '
                     'utf8mb4 COLLATE utf8mb4_unicode_ci;', text)
    patched = patched.replace('USE `xxl_job`;', f'USE `{DB}`;')
    sql_file = RUNTIME / 'schema.sql'
    sql_file.write_text(patched)
    mysql(f'SOURCE {sql_file}')
    tables = sql_rows('SHOW TABLES')
    if len(tables) < 8:
        die(f'建表不完整: {tables}')
    return tables


def ensure_admin_jar():
    CACHE.mkdir(parents=True, exist_ok=True)
    if not UPSTREAM.is_dir():
        done = run_cmd(['git', 'clone', '--depth', '1', '--branch', UPSTREAM_TAG,
                        'https://github.com/xuxueli/xxl-job', str(UPSTREAM)])
        if done.returncode != 0:
            die(f'克隆 xxl-job {UPSTREAM_TAG} 失败: {done.stderr[-300:]}')
    commit = run_cmd(['git', '-C', str(UPSTREAM), 'rev-parse', 'HEAD']).stdout.strip()
    if commit != UPSTREAM_COMMIT:
        die(f'上游提交不符: {commit} != {UPSTREAM_COMMIT}')
    if not ADMIN_JAR.is_file():
        env = dict(os.environ, JAVA_HOME=JDK)
        done = run_cmd([MVN, '-q', '-DskipTests', '-pl', 'xxl-job-admin', '-am', 'package'],
                       cwd=str(UPSTREAM), env=env)
        if done.returncode != 0 or not ADMIN_JAR.is_file():
            die(f'构建 xxl-job-admin 失败: {done.stderr[-500:]}')
    return commit


def build_executor():
    env = dict(os.environ, JAVA_HOME=JDK)
    done = run_cmd([MVN, '-q', '-DskipTests', 'package'], cwd=str(LAB / 'executor'), env=env)
    if done.returncode != 0 or not EXECUTOR_JAR.is_file() or not EXECUTOR_CP.is_file():
        die(f'构建执行器失败: {done.stderr[-500:]}')


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)


class Admin:
    def __init__(self, name):
        self.port = free_port()
        self.name = name
        self.base = f'http://127.0.0.1:{self.port}'
        self.proc = None
        self.log = None
        self.logdir = RUNTIME / f'admin-{name}'
        self.opener = None

    def start(self):
        self.logdir.mkdir(parents=True, exist_ok=True)
        (self.logdir / 'applogs').mkdir(exist_ok=True)
        env = dict(os.environ, JAVA_HOME=JDK, LOG_HOME=str(self.logdir / 'applogs'))
        cmd = [str(pathlib.Path(JDK, 'bin', 'java')), '-jar', str(ADMIN_JAR),
               f'--server.port={self.port}',
               '--spring.datasource.url=jdbc:mysql://127.0.0.1:3306/'
               f'{DB}?useUnicode=true&characterEncoding=UTF-8&serverTimezone=Asia/Shanghai',
               f'--spring.datasource.username={MYSQL["user"]}',
               f'--spring.datasource.password={MYSQL["password"]}',
               '--spring.mail.host=127.0.0.1', '--spring.mail.port=11025',
               '--spring.mail.username=lab@example.invalid',
               '--spring.mail.password=lab-placeholder-not-a-secret',
               '--xxl.job.triggerpool.fast.max=20', '--xxl.job.triggerpool.slow.max=20']
        self.log = open(self.logdir / 'admin.log', 'w')
        self.proc = subprocess.Popen(cmd, stdout=self.log, stderr=subprocess.STDOUT,
                                     env=env, start_new_session=True)
        procs.append(self.proc)
        deadline = time.time() + 120
        while time.time() < deadline:
            if self.proc.poll() is not None:
                tail = (self.logdir / 'admin.log').read_text()[-1500:]
                die(f'admin {self.name} 提前退出: {tail}')
            try:
                with urllib.request.urlopen(self.base + '/auth/login', timeout=2) as resp:
                    if resp.status == 200:
                        self._login()
                        return self
            except Exception:
                time.sleep(0.5)
        die(f'admin {self.name} 120s 未就绪')

    def _login(self):
        jar = http.cookiejar.CookieJar()
        self.opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(jar))
        self.post('/auth/doLogin', {'userName': 'admin', 'password': '123456'})

    def post(self, path, data):
        body = urllib.parse.urlencode(data).encode()
        try:
            with self.opener.open(self.base + path, body, timeout=15) as resp:
                payload = json.loads(resp.read())
        except urllib.error.HTTPError as err:
            raise
        if payload.get('code') == 500 and 'not login' in str(payload.get('msg', '')):
            # xxl-sso 同一账号只保留最新 token；另一 admin 登录会顶掉本会话，重登后重试一次
            self._login()
            with self.opener.open(self.base + path, body, timeout=15) as resp:
                payload = json.loads(resp.read())
        return payload

    def get(self, path):
        try:
            with self.opener.open(self.base + path, timeout=15) as resp:
                payload = json.loads(resp.read())
        except urllib.error.HTTPError:
            raise
        if payload.get('code') == 500 and 'not login' in str(payload.get('msg', '')):
            self._login()
            with self.opener.open(self.base + path, timeout=15) as resp:
                payload = json.loads(resp.read())
        return payload

    def stop(self, wait=30):
        kill_group(self.proc, signal.SIGTERM, wait)
        try:
            self.log.close()
        except Exception:
            pass


class Executor:
    def __init__(self, jdk, name):
        self.port = free_port()
        self.jdk = jdk
        self.name = name
        self.logdir = RUNTIME / f'executor-{name}'
        self.markers = self.logdir / 'markers'
        self.proc = None
        self.log = None

    def start(self, admin_ports):
        ports = [admin_ports] if isinstance(admin_ports, int) else list(admin_ports)
        self.logdir.mkdir(parents=True, exist_ok=True)
        self.markers.mkdir(parents=True, exist_ok=True)
        cp = f'{EXECUTOR_JAR}:{EXECUTOR_CP.read_text().strip()}'
        addresses = ','.join(f'http://127.0.0.1:{port}' for port in ports)
        cmd = [str(pathlib.Path(self.jdk, 'bin', 'java')), '-cp', cp,
               f'-Dlab.admin.addresses={addresses}',
               f'-Dlab.appname={APPNAME}', f'-Dlab.accessToken={TOKEN}',
               f'-Dlab.port={self.port}',
               f'-Dlab.logpath={self.logdir / "joblogs"}',
               f'-Dlab.marker.dir={self.markers}',
               'study.datasched.ExecutorMain']
        self.log = open(self.logdir / 'executor.log', 'w')
        self.proc = subprocess.Popen(cmd, stdout=self.log, stderr=subprocess.STDOUT,
                                     start_new_session=True)
        procs.append(self.proc)
        deadline = time.time() + 60
        while time.time() < deadline:
            if self.proc.poll() is not None:
                tail = (self.logdir / 'executor.log').read_text()[-1200:]
                die(f'执行器 {self.name} 提前退出: {tail}')
            if (self.markers / f'executor-ready-{self.port}.txt').is_file():
                return self
            time.sleep(0.3)
        die(f'执行器 {self.name} 60s 未就绪')

    def stop(self, sig=signal.SIGTERM, wait=30):
        kill_group(self.proc, sig, wait)
        try:
            self.log.close()
        except Exception:
            pass

    def marker_lines(self, name):
        path = self.markers / name
        return path.read_text().splitlines() if path.is_file() else []


def poll(description, fn, timeout, interval=1.0):
    deadline = time.time() + timeout
    last = None
    while time.time() < deadline:
        last = fn()
        if last:
            return last
        time.sleep(interval)
    die(f'超时等待 {description}（{timeout}s），最后观察: {last}')


def registry_rows():
    return sql_rows(f"SELECT registry_value, update_time FROM xxl_job_registry "
                    f"WHERE registry_group='EXECUTOR' AND registry_key='{APPNAME}'")


def registry_values():
    return {row.split('\t')[0] for row in registry_rows() if row}


def group_address():
    rows = sql_rows(f"SELECT IFNULL(address_list,'') FROM xxl_job_group "
                    f"WHERE app_name='{APPNAME}'")
    return rows[0] if rows else ''


def logs_for(job_id):
    # 备注含真实换行/制表符；用 HEX(JSON) 单行输出，避免批处理按行切分错位
    raw = mysql(
        "SELECT HEX(JSON_OBJECT('id', id, 'triggerTime', "
        "DATE_FORMAT(trigger_time, '%H:%i:%s'), 'address', IFNULL(executor_address, ''), "
        "'triggerCode', trigger_code, 'handleCode', IFNULL(handle_code, 0), "
        "'retryLeft', executor_fail_retry_count, 'handleMsg', IFNULL(handle_msg, ''), "
        "'triggerMsg', IFNULL(trigger_msg, ''))) "
        f"FROM xxl_job_log WHERE job_id={job_id} ORDER BY id", db=DB)
    items = [json.loads(bytes.fromhex(line).decode('utf-8'))
             for line in raw.strip().splitlines()[1:] if line]
    return [{'id': item['id'], 'trigger_time': item['triggerTime'],
             'address': item['address'], 'trigger_code': item['triggerCode'],
             'handle_code': item['handleCode'], 'retry_left': item['retryLeft'],
             'handle_msg': item['handleMsg'], 'trigger_msg': item['triggerMsg']}
            for item in items]


def wait_logs(job_id, predicate, description, timeout=90):
    def check():
        rows = logs_for(job_id)
        return rows if predicate(rows) else None
    return poll(description, check, timeout)


def add_job(admin, name, handler, param='', schedule_type='FIX_RATE', conf='3600',
            retry=0, block='SERIAL_EXECUTION', misfire='DO_NOTHING', route='FIRST',
            child=''):
    response = admin.post('/jobinfo/insert', {
        'jobGroup': GROUP_ID, 'name': name, 'author': 'u85-lab',
        'alarmEmail': '', 'scheduleType': schedule_type, 'scheduleConf': conf,
        'misfireStrategy': misfire, 'executorRouteStrategy': route,
        'executorHandler': handler, 'executorParam': param,
        'executorBlockStrategy': block, 'executorTimeout': '0',
        'executorFailRetryCount': retry, 'glueType': 'BEAN', 'glueSource': '',
        'glueRemark': 'u85-lab', 'childJobId': child})
    if response.get('code') != 200:
        die(f'创建任务 {name} 失败: {response}')
    return int(response['data'])


def update_job(admin, job_id, name, handler, param, schedule_type='FIX_RATE',
               conf='3600', retry=0, block='SERIAL_EXECUTION',
               misfire='DO_NOTHING', route='FIRST', child=''):
    response = admin.post('/jobinfo/update', {
        'id': job_id, 'jobGroup': GROUP_ID, 'name': name, 'author': 'u85-lab',
        'alarmEmail': '', 'scheduleType': schedule_type, 'scheduleConf': conf,
        'misfireStrategy': misfire, 'executorRouteStrategy': route,
        'executorHandler': handler, 'executorParam': param,
        'executorBlockStrategy': block, 'executorTimeout': '0',
        'executorFailRetryCount': retry, 'glueType': 'BEAN', 'glueSource': '',
        'glueRemark': 'u85-lab', 'childJobId': child})
    if response.get('code') != 200:
        die(f'更新任务 {job_id} 失败: {response}')


def job_action(admin, action, job_id):
    response = admin.post(f'/jobinfo/{action}', {f'ids[]': job_id})
    if response.get('code') != 200:
        die(f'{action} 任务 {job_id} 失败: {response}')


def trigger(admin, job_id, param=None, address=''):
    data = {'id': job_id, 'addressList': address}
    if param is not None:
        data['executorParam'] = param
    response = admin.post('/jobinfo/trigger', data)
    if response.get('code') != 200:
        die(f'触发任务 {job_id} 失败: {response}')
    return response


def delete_job(admin, job_id):
    job_action(admin, 'delete', job_id)


def scenario_dag(admin, executor):
    phase('DAG：child_jobid 链与成功才触发')
    started = datetime.now().astimezone().isoformat()
    tag = run_id
    a = add_job(admin, f'u85-dag-a-{tag}', 'u85Record', 'dag=a')
    b = add_job(admin, f'u85-dag-b-{tag}', 'u85Record', 'dag=b')
    c = add_job(admin, f'u85-dag-c-{tag}', 'u85Record', 'dag=c')
    update_job(admin, a, f'u85-dag-a-{tag}', 'u85Record', 'dag=a', child=str(b))
    update_job(admin, b, f'u85-dag-b-{tag}', 'u85Record', 'dag=b', child=str(c))
    trigger(admin, a, 'dag=a')
    rows_a = wait_logs(a, lambda rows: rows and rows[-1]['handle_code'] == 200,
                       'DAG A 完成', 30)
    rows_c = wait_logs(c, lambda rows: rows and rows[-1]['handle_code'] == 200,
                       'DAG C 完成', 40)
    rows_b = wait_logs(b, lambda rows: rows and rows[-1]['handle_code'] == 200,
                       'DAG B 完成', 40)
    assert_true(len(rows_a) == len(rows_b) == len(rows_c) == 1, 'DAG 每节点应只执行一次')
    observed = [
        f'U85 dagChain logs A={len(rows_a)} B={len(rows_b)} C={len(rows_c)}',
        f'U85 dagChain triggerType A={rows_a[-1]["trigger_msg"].split("<br>")[0]}',
        f'U85 dagChain parentMsgHasChild={"触发子任务" in rows_a[-1]["handle_msg"]}',
        f'U85 dagChain order={rows_a[0]["trigger_time"]} -> {rows_b[0]["trigger_time"]} '
        f'-> {rows_c[0]["trigger_time"]}',
    ]
    assert_true('触发子任务' in rows_a[-1]['handle_msg'], '父任务日志应记录触发子任务')
    assert_true(rows_a[0]['trigger_time'] <= rows_b[0]['trigger_time'] <= rows_c[0]['trigger_time'],
                'DAG 触发顺序应为 A<=B<=C')

    d = add_job(admin, f'u85-dag-d-{tag}', 'u85Record', 'dag=d')
    e = add_job(admin, f'u85-dag-e-{tag}', 'u85Record', 'dag=e')
    update_job(admin, d, f'u85-dag-d-{tag}', 'u85Record', 'dag=d', child=str(e))
    delete_job(admin, e)
    trigger(admin, d, 'dag=d')
    rows_d = wait_logs(d, lambda rows: rows and rows[-1]['handle_code'] == 200,
                       'DAG D 完成', 30)
    observed.append(f'U85 dagMissingChild parentHandle=200 childLogs={len(logs_for(e))} '
                    f'parentClaimsChild={"触发子任务" in rows_d[-1]["handle_msg"]}')
    assert_true(logs_for(e) == [], '已删除子任务不应有日志')
    record('dag-child-chain', observed, started)


def scenario_blocking(admin, executor):
    phase('执行器阻塞策略：DISCARD_LATER 丢弃、SERIAL 排队')
    started = datetime.now().astimezone().isoformat()
    tag = run_id
    discard = add_job(admin, f'u85-discard-{tag}', 'u85Sleep',
                      'millis=6000', block='DISCARD_LATER')
    trigger(admin, discard, 'millis=6000')
    time.sleep(1.5)
    trigger(admin, discard, 'millis=6000')
    rows = wait_logs(discard, lambda rows: len(rows) >= 2 and
                     rows[0]['handle_code'] == 200, 'DISCARD_LATER 首条完成', 40)
    rejected = rows[1]
    executor_rejected = 'block strategy effect' in rejected['trigger_msg']
    observed = [
        f'U85 blockDiscard firstHandle={rows[0]["handle_code"]} '
        f'secondTriggerCode={rejected["trigger_code"]} secondHandle={rejected["handle_code"]}',
        f'U85 blockDiscard executorRejected={executor_rejected} '
        f'policyField={"丢弃后续调度" in rejected["trigger_msg"]}',
    ]
    assert_true(rejected['trigger_code'] == 500, '被丢弃的触发应在调度结果中失败')
    assert_true(rejected['handle_code'] == 0, '被丢弃的触发不应有执行回调')
    # 每条触发日志都带"阻塞处理策略：丢弃后续调度"字段；拒绝原因要看执行器返回的英文文案
    assert_true(executor_rejected, '调度备注应含执行器拒绝文案 block strategy effect')

    serial = add_job(admin, f'u85-serial-{tag}', 'u85Sleep',
                     'millis=4000', block='SERIAL_EXECUTION')
    trigger(admin, serial, 'millis=4000')
    time.sleep(1.0)
    trigger(admin, serial, 'millis=4000')
    rows = wait_logs(serial, lambda rows: len(rows) >= 2 and
                     all(row['handle_code'] == 200 for row in rows[:2]),
                     'SERIAL 两次触发都完成', 60)
    # marker 行格式 kind|time|jobId=..|logId=..|detail；用 logId 对齐两条触发
    first_id, second_id = str(rows[0]['id']), str(rows[1]['id'])
    def marker_times(kind):
        out = {}
        for line in executor.marker_lines(kind):
            parts = line.split('|')
            if len(parts) > 3:
                out[parts[3].replace('logId=', '')] = parts[1]
        return out
    m_start, m_finish = marker_times('start.log'), marker_times('finish.log')
    serialized = (str(second_id) in m_start and str(first_id) in m_finish and
                  m_start[str(second_id)] >= m_finish[str(first_id)])
    observed += [
        f'U85 blockSerial codes={[r["handle_code"] for r in rows[:2]]} '
        f'addressReused={rows[0]["address"] == rows[1]["address"]}',
        f'U85 blockSerial serializedByMarker={serialized} '
        f'secondStart={m_start.get(str(second_id))} firstFinish={m_finish.get(str(first_id))}',
    ]
    assert_true(serialized, '第二个 4000ms 任务应在第一个结束后才开始（串行队列）')
    record('blocking-strategies', observed, started)


def scenario_retry(admin, executor):
    phase('失败重试：executor_fail_retry_count 递减与 RETRY 触发')
    started = datetime.now().astimezone().isoformat()
    job = add_job(admin, f'u85-retry-{run_id}', 'u85FailThenSucceed',
                  f'key={run_id}-retry&failTimes=2', retry=2)
    trigger(admin, job, f'key={run_id}-retry&failTimes=2')
    rows = wait_logs(job, lambda rows: len(rows) >= 3 and rows[-1]['handle_code'] == 200,
                     '两次失败后重试成功', 150)
    observed = [f'U85 retryCodes={[r["handle_code"] for r in rows[:3]]}',
                f'U85 retryLeft={[r["retry_left"] for r in rows[:3]]}',
                f'U85 retryTypes={["失败重试触发" in r["trigger_msg"] for r in rows[:3]]}',
                f'U85 retryHandleMsgs={[r["handle_msg"][:40] for r in rows[:3]]}']
    assert_true([r['handle_code'] for r in rows[:3]] == [500, 500, 200],
                '重试轨迹应为 500,500,200')
    assert_true([r['retry_left'] for r in rows[:3]] == [2, 1, 0],
                '失败重试次数应逐次递减')
    record('fail-retry', observed, started)


def scenario_misfire(admin, executor):
    phase('调度过期（misfire）：DO_NOTHING 跳过与 FIRE_ONCE_NOW 补偿一次')
    started = datetime.now().astimezone().isoformat()
    tag = run_id
    skip = add_job(admin, f'u85-misfire-skip-{tag}', 'u85Record', 'misfire=skip',
                   conf='20', misfire='DO_NOTHING')
    once = add_job(admin, f'u85-misfire-once-{tag}', 'u85Record', 'misfire=once',
                   conf='20', misfire='FIRE_ONCE_NOW')
    backdate = int(time.time() * 1000) - 65 * 1000
    mysql(f'UPDATE xxl_job_info SET trigger_status=1, trigger_last_time={backdate - 20000}, '
          f'trigger_next_time={backdate} WHERE id IN ({skip},{once})', db=DB)
    rows_once = wait_logs(once, lambda rows: len(rows) >= 1 and
                          rows[0]['handle_code'] == 200, 'FIRE_ONCE_NOW 补偿一次', 60)
    time.sleep(6)
    rows_skip = logs_for(skip)
    texts = ' '.join(row['trigger_msg'] for row in rows_once)
    observed = [
        f'U85 misfireSkip logs={len(rows_skip)} (错过约3个20s周期，无补偿)',
        f'U85 misfireOnce logs={len(rows_once)} compensation={"调度过期补偿" in texts}',
    ]
    assert_true(rows_skip == [], 'DO_NOTHING 不应补偿停机期间错过的周期')
    assert_true(len(rows_once) == 1, 'FIRE_ONCE_NOW 只补一次，不逐周期回填')
    assert_true('调度过期补偿' in texts, '补偿触发应标记 MISFIRE 类型')
    job_action(admin, 'stop', skip)
    job_action(admin, 'stop', once)
    record('misfire-strategies', observed, started)


def scenario_duplicate_and_failover(admin_a, admin_b, jdk):
    phase('重复调度：双 admin 行锁互斥；主节点切换与重启恢复')
    started = datetime.now().astimezone().isoformat()
    tag = run_id
    dup_executor = Executor(jdk, 'dup').start(admin_a.port)
    poll('重复调度执行器注册', lambda: registry_rows() != [], 60)
    poll('组地址就绪', lambda: group_address() != '', 60)
    cron = add_job(admin_a, f'u85-cron-{tag}', 'u85Record', 'cron=1',
                   schedule_type='CRON', conf='0/2 * * * * ?')
    job_action(admin_a, 'start', cron)
    time.sleep(14)
    base = len(logs_for(cron))
    time.sleep(22)
    rows = logs_for(cron)
    added = len(rows) - base
    repeat_triggers = sum(1 for line in (dup_executor.logdir / 'executor.log').read_text().splitlines()
                          if 'repeat trigger job' in line)
    same_second = len(rows) - len({row['trigger_time'] for row in rows})
    executions = [line.split('|') for line in dup_executor.marker_lines('record.log')]
    cron_log_ids = [fields[3] for fields in executions
                    if len(fields) > 3 and 'cron=1' in fields[4]]
    observed = [
        f'U85 duplicateFirst14sTwoAdmins={base} next22sAdded={added} total={len(rows)}',
        f'U85 duplicateSameSecond={same_second} executorRepeatTriggers={repeat_triggers}',
        f'U85 duplicateExecutions={len(cron_log_ids)} distinctLogIds={len(set(cron_log_ids))}',
    ]
    assert_true(4 <= base <= 8, f'单 admin 14s@2s 期望 5-7 次，实际 {base}')
    assert_true(7 <= added <= 15, f'两 admin 22s 期望约 11 次新增（失效时约 22），实际 {added}')
    assert_true(same_second == 0 and repeat_triggers == 0, '出现同一秒重复调度或执行器 logId 去重拦截')
    assert_true(len(cron_log_ids) == len(set(cron_log_ids)) == len(rows),
                '每次调度应恰好执行一次（marker 与 log row 一一对应）')

    admin_a.stop(wait=20)
    before = len(logs_for(cron))
    time.sleep(7)
    after = len(logs_for(cron))
    observed.append(f'U85 schedulerFailover killedA at={before} plus7s={after}')
    assert_true(after - before >= 2, 'A 退出后 B 应继续调度')

    admin_b.stop(wait=30)
    down_start = len(logs_for(cron))
    time.sleep(8)
    admin_c = Admin('c').start()
    restart_mark = len(logs_for(cron))
    time.sleep(15)
    resumed = len(logs_for(cron))
    observed.append(f'U85 restartResume downAt={down_start} restartMark={restart_mark} '
                    f'after15s={resumed} newRows={resumed - restart_mark}')
    assert_true(restart_mark == down_start, '全部 admin 停机期间不应有调度')
    assert_true(resumed - restart_mark <= 9,
                '重启后 DO_NOTHING 不应补跑停机期间的错过槽位（仅恢复常规节奏）')
    assert_true(resumed - restart_mark >= 3, '重启后应恢复常规节奏')
    job_action(admin_c, 'stop', cron)
    dup_executor.stop(signal.SIGTERM, wait=45)
    record('duplicate-and-failover', observed, started)


def scenario_graceful(admin, executor):
    phase('执行器优雅退出：在途任务被中断、排队任务补失败回调、注册即时移除')
    started = datetime.now().astimezone().isoformat()
    job = add_job(admin, f'u85-term-{run_id}', 'u85Sleep', 'millis=15000',
                  block='SERIAL_EXECUTION')
    trigger(admin, job, 'millis=15000')
    wait_logs(job, lambda rows: rows and rows[-1]['handle_code'] == 0,
              '在途任务进入运行中', 20)
    time.sleep(1.0)
    trigger(admin, job, 'millis=15000')
    time.sleep(1.0)
    executor.stop(signal.SIGTERM, wait=45)
    rows = wait_logs(job, lambda rows: len(rows) >= 2 and
                     all(row['handle_code'] == 500 for row in rows[:2]),
                     'SIGTERM 后两条日志均为失败', 60)
    messages = ' | '.join(row['handle_msg'] for row in rows[:2])
    poll('SIGTERM 后注册行移除', lambda: registry_values() == set(), 30)
    observed = [
        f'U85 gracefulKill codes={[r["handle_code"] for r in rows[:2]]}',
        f'U85 gracefulKill runningKilled={"killed" in messages} '
        f'queuedSkipped={"job not executed" in messages}',
        f'U85 gracefulKill registryRemoved={registry_values() == set()}',
    ]
    assert_true('[job running, killed]' in messages, '在途任务应记录被杀')
    assert_true('job not executed' in messages, '排队任务应以未执行失败回调')
    record('graceful-shutdown', observed, started)


def scenario_crash(admin, jdk):
    phase('执行器崩溃：无回调、注册过期、丢失任务判失败、地址为空、重启恢复')
    started = datetime.now().astimezone().isoformat()
    crash_exec = Executor(jdk, 'crash').start(admin.port)
    crash_address = f'http://127.0.0.1:{crash_exec.port}'
    poll('崩溃执行器注册', lambda: crash_address in registry_values(), 60)
    job = add_job(admin, f'u85-crash-{run_id}', 'u85Crash', 'millis=30000',
                  block='SERIAL_EXECUTION')
    # 明确把这次触发路由到崩溃执行器（组地址可能仍含已退出实例的陈旧地址）
    trigger(admin, job, 'millis=30000', address=crash_address)
    rows = wait_logs(job, lambda rows: rows and rows[-1]['trigger_code'] == 200 and
                     rows[-1]['handle_code'] == 0, '崩溃任务被接收进入运行中', 20)
    log_id = rows[-1]['id']
    deadline = time.time() + 20
    while time.time() < deadline and crash_exec.proc.poll() is None:
        time.sleep(0.3)
    assert_true(crash_exec.marker_lines('crash.log'), '崩溃 handler 应已开始执行')
    crash_exec.stop(signal.SIGKILL, wait=5)
    observed = [f'U85 crashHandler halted pidGone={crash_exec.proc.poll() is not None} '
                f'registryRowsStill={len(registry_rows())}']

    # 崩溃后的陈旧地址窗口：崩溃执行器的注册行还在（心跳 90s 内），地址仍可路由但连不上
    stale = trigger(admin, job, 'millis=1000', address=crash_address)
    stale_row = poll('崩溃后陈旧地址触发', lambda: (lambda rows: rows[-1]
                     if rows and rows[-1]['id'] != log_id else None)(logs_for(job)), 20)
    observed.append(f'U85 crashStaleAddress triggerCode={stale_row["trigger_code"]} '
                    f'msg={stale_row["trigger_msg"][-80:].replace(chr(10), " ")}')

    mysql(f'UPDATE xxl_job_log SET trigger_time = DATE_SUB(NOW(), INTERVAL 11 MINUTE) '
          f'WHERE id={log_id}', db=DB)
    # 完成监控每 60s 一轮；崩溃后注册行在 90s 内仍算在线。为跨过至少一整轮监控而注册仍新鲜，
    # 用每 20s 刷新 update_time 模拟心跳（受控构造"执行器仍在线"的真值条件，证据 limits 已标注）。
    for _ in range(4):
        mysql(f'UPDATE xxl_job_registry SET update_time = NOW() '
              f"WHERE registry_key='{APPNAME}'", db=DB)
        time.sleep(20)
    alive_handle = [row for row in logs_for(job) if row['id'] == log_id][0]['handle_code']
    observed.append(f'U85 crash11minButRegistryAliveAfter70s handleCode={alive_handle} '
                    '(跨过完整监控周期仍不判失败，须注册不在线)')
    assert_true(alive_handle == 0,
                '执行器注册仍在线时，仅超过 10 分钟不应判丢失失败')

    mysql(f'UPDATE xxl_job_registry SET update_time = DATE_SUB(NOW(), INTERVAL 100 SECOND) '
          f"WHERE registry_key='{APPNAME}'", db=DB)
    poll('崩溃执行器注册过期移除', lambda: registry_rows() == [], 120)
    poll('组地址清空', lambda: group_address() == '', 60)
    observed.append('U85 crashRegistryExpired addressCleared=true')

    trigger(admin, job, 'millis=1000')
    failed = poll('执行器离线时触发失败', lambda: (lambda rows: rows[-1] if rows and
                  rows[-1]['trigger_code'] != 200 else None)(logs_for(job)), 20)
    observed.append(f'U85 crashAddressEmpty triggerCode={failed["trigger_code"]} '
                    f'msg={failed["trigger_msg"][-90:].replace(chr(10), " ")}')

    lost = poll('丢失任务被标记失败', lambda: (lambda rows: rows if any(
        row['id'] == log_id and row['handle_code'] == 500 and
        '任务结果丢失' in row['handle_msg'] for row in rows) else None)(logs_for(job)), 150)
    lost_row = [row for row in lost if row['id'] == log_id][0]
    observed.append(f'U85 crashLostJob handleCode=500 msg={lost_row["handle_msg"]}')

    # 验证 xxl-sso 单 token：第二个 admin 用同一账号登录后，原会话的下一次请求被拒
    admin_q = Admin('q').start()
    with admin.opener.open(admin.base + '/jobgroup/pageList?offset=0&pagesize=1',
                           timeout=10) as resp:
        kicked = json.loads(resp.read())
    observed.append(f'U85 ssoSingleToken afterOtherLoginCode={kicked.get("code")} '
                    f'msg={str(kicked.get("msg"))[:40]}')
    admin._login()  # 原会话重新登录，继续后续场景
    admin_q.stop(wait=20)

    recover = Executor(jdk, 'recover').start(admin.port)
    recover_address = f'http://127.0.0.1:{recover.port}'
    poll('恢复执行器注册', lambda: recover_address in registry_values(), 60)
    poll('组地址回填', lambda: group_address() != '', 60)
    # 恢复验证用普通记录任务：u85Crash 每次执行都会 halt，不能作为恢复探针
    recover_job = add_job(admin, f'u85-recover-{run_id}', 'u85Record', 'recover=1')
    trigger(admin, recover_job, 'recover=1', address=recover_address)
    wait_logs(recover_job, lambda rows: rows and rows[-1]['trigger_code'] == 200 and
              rows[-1]['handle_code'] == 200, '恢复后触发成功', 40)
    observed.append('U85 crashRecovery recovered=true')
    recover.stop(signal.SIGTERM, wait=45)
    record('crash-and-lost-job', 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 = [proc.pid for proc in procs if proc.poll() is None]
    dropped = False
    drop_error = None
    try:
        mysql(f'DROP DATABASE IF EXISTS `{DB}`')
        dropped = True
    except Exception as exc:  # noqa: BLE001
        drop_error = str(exc)
    leftovers = [row for row in
                 mysql("SHOW DATABASES LIKE 'study\\_u85\\_xxl\\_%'").strip().splitlines()[1:]
                 if row]
    result = {'processesSurviving': survivors, 'databaseDropped': dropped,
              'dropError': drop_error, 'labDatabasesRemaining': leftovers,
              'runtimeDir': str(RUNTIME)}
    if survivors or not dropped or leftovers:
        errors.append(f'清理不完整: {result}')
    return result


def run():
    global JDK, MVN, GROUP_ID
    exit_code = 0
    failure = None
    environment = {}
    cleanup_result = None
    try:
        JDK = java_home()
        MVN = os.environ.get('MAVEN_BIN') or shutil.which('mvn')
        if not JDK or not pathlib.Path(JDK, 'bin', 'java').is_file():
            die('需要 JDK 17+：设置 JAVA_HOME')
        if not MVN:
            die('需要 Maven 3.9.x：设置 MAVEN_BIN 或 PATH')
        MYSQL['bin'] = mysql_bin()
        RUNTIME.mkdir(parents=True, exist_ok=True)
        say(f'U85 preflight mysql={MYSQL["bin"]} db={DB} appname={APPNAME} '
            f'runtime={RUNTIME}')
        mysql_version = mysql('SELECT VERSION()').splitlines()[1]
        ensure_schema()
        schema_sha = sha256(SCHEMA_CACHE)
        tables = apply_schema(SCHEMA_CACHE.read_text())
        commit = ensure_admin_jar()
        build_executor()
        environment = {
            'os': platform.platform(), 'arch': platform.machine(),
            'jdk': subprocess.run([str(pathlib.Path(JDK, 'bin', 'java')), '-version'],
                                  capture_output=True, text=True).stderr.splitlines()[0],
            'maven': subprocess.run([MVN, '-version'], capture_output=True, text=True
                                    ).stdout.splitlines()[0],
            'mysql': mysql_version, 'mysqlBin': MYSQL['bin'],
            'xxlJobVersion': UPSTREAM_TAG, 'upstreamCommit': commit,
            'adminJarSha256': sha256(ADMIN_JAR),
            'executorJarSha256': sha256(EXECUTOR_JAR),
            'schemaSqlSha256': schema_sha, 'tableCount': len(tables),
            'externalServices': f'MySQL {MYSQL["host"]}:{MYSQL["port"]} 独有库 {DB}；'
                                'admin/executor 均在 127.0.0.1 随机端口',
        }
        say(f'U85 versions mysql={mysql_version} upstreamCommit={commit} '
            f'adminJarSha={environment["adminJarSha256"][:16]} '
            f'executorJarSha={environment["executorJarSha256"][:16]} '
            f'schemaSha={schema_sha[:16]} tables={len(tables)}')

        admin_a = Admin('a').start()
        response = admin_a.post('/jobgroup/insert', {
            'appname': APPNAME, 'name': f'U85 study executor {run_id}',
            'addressType': 0, 'accessToken': TOKEN})
        if response.get('code') != 200:
            die(f'创建执行器组失败: {response}')
        page = admin_a.get('/jobgroup/pageList?offset=0&pagesize=50&appname=' + APPNAME)
        GROUP_ID = page['data']['data'][0]['id']
        executor = Executor(JDK, 'main').start(admin_a.port)
        poll('执行器注册与组地址', lambda: registry_rows() != [] and group_address() != '', 90)
        say(f'U85 registry groupId={GROUP_ID} address={group_address()}')

        scenario_dag(admin_a, executor)
        scenario_blocking(admin_a, executor)
        scenario_retry(admin_a, executor)
        scenario_misfire(admin_a, executor)
        scenario_graceful(admin_a, executor)
        scenario_crash(admin_a, JDK)
        scenario_duplicate_and_failover(admin_a, Admin('b').start(), JDK)
        say('U85 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:
        cleanup_result = cleanup()
        if cleanup_result and (cleanup_result['processesSurviving'] or
                               not cleanup_result['databaseDropped'] or
                               cleanup_result['labDatabasesRemaining']):
            exit_code = exit_code or 1

    evidence = {
        'ticket': 'U85 #122',
        'generatedAt': datetime.now().astimezone().isoformat(),
        'startedAt': started_all,
        'runId': run_id,
        'input': 'python3 labs/data-scheduler/run.py',
        'environment': environment,
        'scenarios': scenarios,
        'observed': observations,
        'exitCode': exit_code,
        'failure': failure,
        'errors': errors,
        'rawLogs': str(RUNTIME),
        'cleanup': cleanup_result,
        'runnerSha256': sha256(pathlib.Path(__file__).resolve()),
        'limits': [
            'MySQL 为本机 DBngin 8.4.x 的独有库，结束即 DROP；不使用 Docker、不写用户既有库。',
            'admin 与执行器都是固定 tag 3.5.0 的公开源码构建产物；admin 由本机 JDK21/Maven 从上游提交 '
            f'{UPSTREAM_COMMIT[:12]} 构建，jar/建表 SQL 的 SHA-256 见 environment。',
            '双 admin 只验证数据库行锁（schedule_lock FOR UPDATE）与 trigger_next_time 的互斥；'
            '不覆盖网络分区、主备切换与租约代际，这些归 U87。',
            '丢失任务依赖两个真值条件：注册行已过期（执行器不在线）且日志运行中超过 10 分钟。'
            '实验把该行 trigger_time 回拨 11 分钟以获得时间窗，回拨是实验操作而非生产事实；'
            '注册过期通过回拨 registry.update_time 加速同一判据；为验证\"注册在线时不判丢失\"，'
            '实验中每 20 秒 UPDATE 一次 registry.update_time 模拟心跳并持续 70 秒，跨过完整监控周期。',
            'misfire 通过直接改库构造停机错过场景；补偿只发生一次，不等于按周期逐段回填。',
            '所有任务与参数为合成标签，不含用户业务数据与凭据；失败告警邮件地址留空，不访问真实 SMTP。',
        ],
    }
    EVIDENCE.parent.mkdir(parents=True, exist_ok=True)
    EVIDENCE.write_text(json.dumps(evidence, ensure_ascii=False, indent=2) + '\n')
    print(f'\nU85 exit={exit_code} evidence={EVIDENCE}', flush=True)
    return exit_code


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