#!/usr/bin/env python3
"""Continuous, failure-isolated supervisor for the AOV fresh reroll worker."""

from __future__ import annotations

import argparse
import concurrent.futures
import fcntl
import html
import json
import os
import re
import selectors
import signal
import subprocess
import sys
import threading
import time
import traceback
from collections import Counter
from datetime import datetime
from pathlib import Path
from typing import Any

import aov_runner
import jackpot_detector


SCRIPT_DIR = Path(__file__).resolve().parent
DEFAULT_RUNNER = SCRIPT_DIR / "aov_runner.py"
DEFAULT_WORKFLOW = SCRIPT_DIR.parent / "references" / "workflow.json"
ENGINE_LINE = re.compile(r"created fresh engine: ([A-Za-z0-9_-]+) \((.+)\)")
ADB_READY_LINE = re.compile(r"ADB ready: ([A-Za-z0-9_-]+) on ([0-9.:]+)")
ACTIVE_WORKERS: dict[int, subprocess.Popen[str]] = {}
ACTIVE_WORKERS_LOCK = threading.RLock()
STOP_EVENT = threading.Event()


def now_iso() -> str:
    return datetime.now().astimezone().isoformat(timespec="seconds")


def timestamp() -> str:
    return datetime.now().strftime("%Y%m%d_%H%M%S")


def atomic_json(path: Path, value: dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    temporary = path.with_suffix(path.suffix + ".tmp")
    temporary.write_text(
        json.dumps(value, ensure_ascii=False, indent=2) + "\n",
        encoding="utf-8",
    )
    os.replace(temporary, path)


def load_json(path: Path) -> dict[str, Any] | None:
    try:
        value = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        return None
    return value if isinstance(value, dict) else None


def load_jsonl(path: Path) -> list[dict[str, Any]]:
    values: list[dict[str, Any]] = []
    if not path.is_file():
        return values
    for line in path.read_text(encoding="utf-8", errors="replace").splitlines():
        try:
            item = json.loads(line)
        except json.JSONDecodeError:
            continue
        if isinstance(item, dict):
            values.append(item)
    return values


def append_jsonl(path: Path, value: dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("a", encoding="utf-8") as handle:
        handle.write(json.dumps(value, ensure_ascii=False) + "\n")
        handle.flush()
        os.fsync(handle.fileno())


def relative_href(path: Path, dashboard_root: Path) -> str:
    return os.path.relpath(path.resolve(), dashboard_root.resolve())


def format_duration(seconds: Any) -> str:
    try:
        total = max(0, round(float(seconds)))
    except (TypeError, ValueError):
        return "—"
    hours, remainder = divmod(total, 3600)
    minutes, secs = divmod(remainder, 60)
    if hours:
        return f"{hours} 小時 {minutes} 分"
    if minutes:
        return f"{minutes} 分 {secs} 秒"
    return f"{secs} 秒"


def format_dashboard_time(value: Any) -> str:
    try:
        parsed = datetime.fromisoformat(str(value))
    except ValueError:
        return str(value or "—")
    return parsed.strftime("%m/%d %H:%M:%S")


def format_elapsed_since(value: Any) -> str:
    try:
        parsed = datetime.fromisoformat(str(value))
    except ValueError:
        return "—"
    if parsed.tzinfo is None:
        parsed = parsed.astimezone()
    return format_duration(
        (datetime.now().astimezone() - parsed).total_seconds()
    )


def stage_display_name(value: Any) -> str:
    stage = str(value or "等待狀態")
    labels = {
        "game_workflow": "遊戲流程",
        "onboarding": "登入／教學對局",
        "victory_exit": "離開勝負畫面",
        "post_battle_results": "對局結算",
        "post_lobby": "新手福利入口",
        "benefit_wait_skip": "新手福利引導",
        "activity_claim": "活動獎勵",
        "backpack_rewards": "背包開包結果",
        "dimension_free_draw": "免費單抽",
        "ten_draw_1": "第一次十連抽",
        "ten_draw_2": "第二次十連抽",
        "ten_draw_3": "第三次十連抽",
        "final_result": "最終抽獎結果",
        "delete_retry": "刪除引擎重試",
    }
    return labels.get(stage, stage.replace("_", " "))


def record_manifest(record: dict[str, Any]) -> dict[str, Any]:
    run_dir = record.get("run_dir")
    if not run_dir:
        return {}
    return load_json(Path(str(run_dir)) / "run.json") or {}


def failure_summary(record: dict[str, Any]) -> tuple[str, str]:
    manifest = record_manifest(record)
    raw = str(
        record.get("watchdog_reason")
        or manifest.get("error")
        or record.get("analysis_error")
        or record.get("worker_status")
        or "未知失敗"
    )
    stage_match = re.search(r"Stage ['\"]([^'\"]+)", raw)
    stage = stage_match.group(1) if stage_match else ""
    known = {
        "activity_claim": "活動分類辨識失敗",
        "backpack_rewards": "背包獎勵結果辨識失敗",
        "ten_draw_3_result": "第三次十連結果辨識失敗",
        "post_battle_results": "對局結算辨識失敗",
        "scoreboard": "舊版對局結算辨識失敗",
        "benefit_exit_after_signin": "新手福利退出辨識失敗",
    }
    if stage in known:
        return known[stage], raw
    if record.get("watchdog_reason"):
        return "單輪程序逾時／無輸出", raw
    if "Expected frame" in raw:
        return "遊戲未進入／畫面未轉橫", raw
    if "Game launch did not reach package" in raw:
        return "遊戲啟動驗證失敗", raw
    if "Game launch did not reach a landscape" in raw:
        return "遊戲啟動後未轉橫", raw
    if (
        "timeoutexpired" in raw.casefold()
        and "'am'" in raw.casefold()
        and "'start'" in raw.casefold()
    ):
        return "遊戲啟動指令逾時", raw
    if "Global timeout" in raw:
        return "整體流程逾時", raw
    if "close" in raw.casefold() or "stop" in raw.casefold():
        return "引擎關閉失敗", raw
    return "其他單輪失敗", raw


def is_startup_launch_failure(record: dict[str, Any]) -> bool:
    """Return true only for failures before the game reached a stable landscape frame."""
    manifest = record_manifest(record)
    raw = " ".join(
        str(value or "")
        for value in (
            record.get("watchdog_reason"),
            manifest.get("error"),
            record.get("analysis_error"),
        )
    ).casefold()
    landscape_startup_failure = any(
        marker in raw
        for marker in (
            "game launch did not reach a landscape",
            "game launch did not reach package",
            "expected frame [2560, 1440], got [1440, 2560]",
        )
    ) and any(
        marker in raw
        for marker in (
            "timed out",
            "did not reach",
            "expected frame",
        )
    )
    delayed_am_start = (
        "timeoutexpired" in raw and "'am'" in raw and "'start'" in raw
    )
    return landscape_startup_failure or delayed_am_start


class BatchContext:
    def __init__(self, batch_root: Path, output_root: Path) -> None:
        self.batch_root = batch_root
        self.output_root = output_root
        self.logs_dir = batch_root / "logs"
        self.dashboard_path = batch_root / "dashboard.html"
        self.state_path = batch_root / "state.json"
        self.rounds_path = batch_root / "rounds.jsonl"
        self.events_path = batch_root / "events.jsonl"
        self.batch_log_path = batch_root / "batch.log"
        self.legacy_active_path = batch_root / "active_round.json"
        self.active_path = batch_root / "active_rounds.json"
        self.lock = threading.RLock()
        self.cleanup_blocked = threading.Event()
        self.cleanup_count = 0
        self.batch_root.mkdir(parents=True, exist_ok=True)
        self.logs_dir.mkdir(parents=True, exist_ok=True)
        self.rounds = sorted(
            load_jsonl(self.rounds_path),
            key=lambda item: int(item.get("round_number", 0)),
        )
        active_document = load_json(self.active_path) or {}
        raw_active = active_document.get("rounds") if active_document else []
        self.active_rounds: dict[int, dict[str, Any]] = {
            int(item["round_number"]): dict(item)
            for item in (raw_active if isinstance(raw_active, list) else [])
            if isinstance(item, dict) and str(item.get("round_number", "")).isdigit()
        }
        self.protected_engines: dict[str, str] = {}
        self.state: dict[str, Any] = {
            "schema_version": 2,
            "supervisor_started_at": now_iso(),
            "updated_at": now_iso(),
            "status": "starting",
            "active": [],
            "resources": {},
            "settings": {},
            "dashboard": str(self.dashboard_path),
        }

    def event(self, kind: str, message: str, **fields: Any) -> None:
        with self.lock:
            value = {"at": now_iso(), "kind": kind, "message": message, **fields}
            append_jsonl(self.events_path, value)
            with self.batch_log_path.open("a", encoding="utf-8") as handle:
                handle.write(f"{value['at']} [{kind}] {message}")
                if fields:
                    handle.write(" " + json.dumps(fields, ensure_ascii=False))
                handle.write("\n")
        print(f"[{kind}] {message}", flush=True)

    def persist(self) -> None:
        with self.lock:
            self.state["updated_at"] = now_iso()
            completed = sum(
                1 for item in self.rounds if item.get("worker_status") == "completed"
            )
            durations = [
                float(item.get("duration_seconds", 0))
                for item in self.rounds
                if item.get("worker_status") == "completed"
            ]
            totals = {
                "rounds": len(self.rounds),
                "completed": completed,
                "completion_rate": completed / len(self.rounds) if self.rounds else 0.0,
                "kept": sum(1 for item in self.rounds if item.get("decision") == "keep"),
                "deleted": sum(1 for item in self.rounds if item.get("engine_action") == "deleted"),
                "failed": sum(1 for item in self.rounds if item.get("worker_status") != "completed"),
                "average_completed_seconds": sum(durations) / len(durations) if durations else 0.0,
            }
            self.state["totals"] = totals
            self.state["active"] = [
                self.active_summary(item)
                for _, item in sorted(self.active_rounds.items())
            ]
            resources = self.state.setdefault("resources", {})
            if isinstance(resources, dict):
                resources["actual_parallel"] = len(self.active_rounds)
            self.state["current"] = self.state["active"][0] if len(self.state["active"]) == 1 else None
            atomic_json(self.state_path, self.state)
            self.render_dashboard()

    def add_round(self, record: dict[str, Any]) -> None:
        with self.lock:
            if any(
                int(item.get("round_number", -1)) == int(record.get("round_number", -2))
                for item in self.rounds
            ):
                return
            append_jsonl(self.rounds_path, record)
            self.rounds.append(record)
            self.rounds.sort(key=lambda item: int(item.get("round_number", 0)))
            self.persist()

    @staticmethod
    def active_summary(item: dict[str, Any]) -> dict[str, Any]:
        return {
            key: item.get(key)
            for key in (
                "round_number",
                "engine",
                "display_name",
                "status",
                "stage",
                "started_at",
                "last_output_at",
                "last_error",
                "run_dir",
                "pid",
                "heartbeat_at",
                "adb_endpoint",
                "lifecycle_ready",
            )
        }

    def write_active(self) -> None:
        with self.lock:
            atomic_json(
                self.active_path,
                {
                    "schema_version": 2,
                    "updated_at": now_iso(),
                    "rounds": [
                        item for _, item in sorted(self.active_rounds.items())
                    ],
                },
            )

    def set_active(self, record: dict[str, Any]) -> None:
        with self.lock:
            round_number = int(record["round_number"])
            self.active_rounds[round_number] = dict(record)
            self.write_active()

    def update_active(self, round_number: int, **fields: Any) -> None:
        with self.lock:
            if round_number not in self.active_rounds:
                return
            self.active_rounds[round_number].update(fields)
            self.write_active()

    def remove_active(self, round_number: int) -> None:
        with self.lock:
            self.active_rounds.pop(round_number, None)
            self.write_active()

    def owned_engines(self, *, excluding_round: int | None = None) -> dict[str, int]:
        with self.lock:
            return {
                str(item["engine"]): number
                for number, item in self.active_rounds.items()
                if number != excluding_round and item.get("engine")
            }

    def begin_cleanup(self) -> None:
        with self.lock:
            self.cleanup_count += 1
            self.cleanup_blocked.set()

    def end_cleanup(self) -> None:
        with self.lock:
            self.cleanup_count = max(0, self.cleanup_count - 1)
            if self.cleanup_count == 0:
                self.cleanup_blocked.clear()

    def render_dashboard(self) -> None:
        active_items = [
            item for item in (self.state.get("active") or []) if isinstance(item, dict)
        ]
        resources = self.state.get("resources") or {}
        settings = self.state.get("settings") or {}
        total_rounds = len(self.rounds)
        completed_rounds = [
            item for item in self.rounds if item.get("worker_status") == "completed"
        ]
        failed_rounds = [
            item for item in self.rounds if item.get("worker_status") != "completed"
        ]
        recent_ten = self.rounds[-10:]
        recent_completed = sum(
            1 for item in recent_ten if item.get("worker_status") == "completed"
        )
        completed_durations = [
            float(item.get("duration_seconds", 0)) for item in completed_rounds
        ]
        draw_completed = sum(
            1
            for record in self.rounds
            if any(
                isinstance(condition, dict)
                and condition.get("name") == "rare_reward_top_left"
                and condition.get("status") != "missing"
                for condition in record.get("conditions") or []
            )
        )
        popup_count = sum(
            1
            for record in self.rounds
            for image in record.get("images") or []
            if isinstance(image, dict) and "draw_gift_popup" in str(image.get("label", ""))
        )
        kept = sum(1 for item in self.rounds if item.get("decision") == "keep")
        deleted = sum(1 for item in self.rounds if item.get("engine_action") == "deleted")
        completion_rate = (
            len(completed_rounds) / total_rounds * 100 if total_rounds else 0.0
        )
        recent_rate = recent_completed / len(recent_ten) * 100 if recent_ten else 0.0
        ordered_durations = sorted(completed_durations)
        midpoint = len(ordered_durations) // 2
        if not ordered_durations:
            median_duration = 0.0
        elif len(ordered_durations) % 2:
            median_duration = ordered_durations[midpoint]
        else:
            median_duration = (
                ordered_durations[midpoint - 1] + ordered_durations[midpoint]
            ) / 2

        failure_counts = Counter(failure_summary(item)[0] for item in failed_rounds)
        failure_max = max(failure_counts.values(), default=1)
        failure_rows = "".join(
            '<div class="failure-row"><div><span>{}</span><b>{}</b></div>'
            '<div class="failure-track"><i style="width:{:.1f}%"></i></div></div>'.format(
                html.escape(label),
                count,
                count / failure_max * 100,
            )
            for label, count in failure_counts.most_common()
        ) or '<p class="empty">目前沒有失敗紀錄。</p>'

        history_items: list[str] = []
        ordinary_completed = sum(
            1
            for record in self.rounds
            if record.get("worker_status") == "completed"
            and record.get("decision") != "keep"
        )
        for record in sorted(
            self.rounds,
            key=lambda item: int(item.get("round_number", 0)),
        ):
            is_complete = record.get("worker_status") == "completed"
            is_keep = record.get("decision") == "keep"
            outcome = "keep" if is_keep else ("ok" if is_complete else "fail")
            symbol = "★" if is_keep else ("✓" if is_complete else "×")
            outcome_label = "保留" if is_keep else ("完成" if is_complete else "失敗")
            title = "第 {} 輪 · {} · {} · {}".format(
                record.get("round_number", "?"),
                record.get("engine") or "未建立引擎",
                outcome_label,
                format_duration(record.get("duration_seconds")),
            )
            history_items.append(
                '<button class="result-cell {}" data-jump="round-{}" title="{}" '
                'aria-label="{}"><span aria-hidden="true">{}</span></button>'.format(
                    outcome,
                    record.get("round_number", "?"),
                    html.escape(title, quote=True),
                    html.escape(title, quote=True),
                    symbol,
                )
            )

        history_summary = (
            '<div class="result-summary">'
            f'<span><i class="summary-symbol ok">✓</i>完成 {ordinary_completed}</span>'
            f'<span><i class="summary-symbol fail">×</i>失敗 {len(failed_rounds)}</span>'
            f'<span><i class="summary-symbol keep">★</i>保留 {kept}</span>'
            f'<span>成功率 {completion_rate:.1f}% · 共 {total_rounds} 輪</span>'
            '</div>'
        )

        cards: list[str] = []
        for card_index, record in enumerate(reversed(self.rounds)):
            decision = str(record.get("decision", "unknown"))
            is_complete = record.get("worker_status") == "completed"
            css_class = "keep" if decision == "keep" else ("complete" if is_complete else "failed")
            filter_value = "keep" if decision == "keep" else ("complete" if is_complete else "failed")
            worker = html.escape(str(record.get("worker_status", "unknown")))
            engine = html.escape(str(record.get("engine") or "未建立"))
            conditions = record.get("conditions") or []
            condition_html = "".join(
                '<li class="condition {}"><strong>{}</strong><span>{}</span><p>{}</p></li>'.format(
                    html.escape(str(item.get("status", "unknown"))),
                    html.escape(
                        {
                            "activity_rewards": "活動獎勵",
                            "rare_reward_top_left": "稀有獎勵左上角",
                        }.get(str(item.get("name", "")), str(item.get("name", "")))
                    ),
                    html.escape(str(item.get("status", ""))),
                    html.escape(str(item.get("reason", ""))),
                )
                for item in conditions
                if isinstance(item, dict)
            )
            images: list[str] = []
            for item in record.get("images", []):
                if not isinstance(item, dict):
                    continue
                href = html.escape(str(item.get("href", "")), quote=True)
                label = html.escape(str(item.get("label", "截圖")))
                images.append(
                    f'<a class="shot" href="{href}" target="_blank">'
                    f'<img loading="lazy" src="{href}" alt="{label}"><span title="{label}">{label}</span></a>'
                )
            log_href = html.escape(str(record.get("log_href", "")), quote=True)
            run_href = html.escape(str(record.get("run_href", "")), quote=True)
            failure_label, failure_raw = failure_summary(record)
            error_html = ""
            if not is_complete:
                error_html = (
                    '<div class="error-box"><strong>{}</strong><code>{}</code></div>'.format(
                        html.escape(failure_label),
                        html.escape(failure_raw),
                    )
                )
            finished = format_dashboard_time(record.get("finished_at"))
            outcome_label = "保留" if decision == "keep" else ("完成" if is_complete else "失敗")
            searchable = " ".join(
                [
                    str(record.get("round_number", "")),
                    str(record.get("engine", "")),
                    str(record.get("worker_status", "")),
                    failure_label if not is_complete else "",
                ]
            ).casefold()
            cards.append(
                f'''<details id="round-{record.get("round_number", "?")}" class="round {css_class}" data-outcome="{filter_value}" data-index="{card_index}" data-search="{html.escape(searchable, quote=True)}">
<summary><span class="round-identity"><b>#{record.get("round_number", "?")}</b><span>{engine}</span></span>
<span class="badge">{outcome_label}</span>
<span class="duration">{format_duration(record.get("duration_seconds"))}</span>
<time>{finished}</time></summary>
<div class="round-body">
  <div class="round-meta"><span>程序：{worker}</span><span>引擎：{html.escape(str(record.get("engine_action", "")))}</span><span>完整耗時：{format_duration(record.get("duration_seconds"))}</span></div>
  {error_html}
  <ul class="conditions">{condition_html or '<li class="condition missing"><strong>結果</strong><span>missing</span><p>沒有可比對的結果截圖</p></li>'}</ul>
  <p class="links"><a href="{log_href}" target="_blank">查看單輪 log</a><a href="{run_href}" target="_blank">開啟執行資料夾</a></p>
  <div class="shots">{''.join(images) or '<em>本輪沒有截圖</em>'}</div>
</div></details>'''
            )

        updated_at = str(self.state.get("updated_at", ""))
        update_age: float | None = None
        try:
            update_age = (datetime.now().astimezone() - datetime.fromisoformat(updated_at)).total_seconds()
        except ValueError:
            pass
        stale = bool(active_items) and update_age is not None and update_age > 90
        active_cards: list[str] = []
        for item in active_items:
            error = str(item.get("last_error") or "")
            if not error and item.get("run_dir"):
                manifest = load_json(Path(str(item["run_dir"])) / "run.json") or {}
                error = str(manifest.get("error") or "")
            detail = " · ".join(
                [
                    str(item.get("engine") or "正在建立引擎"),
                    stage_display_name(item.get("stage") or item.get("status")),
                    f"已執行 {format_elapsed_since(item.get('started_at'))}",
                    f"ADB {item.get('adb_endpoint') or '等待中'}",
                    f"PID {item.get('pid') or '等待中'}",
                ]
            )
            active_cards.append(
                '<article class="current{}"><i class="pulse"></i><div><h2>第 {} 輪</h2>'
                '<p>{}</p></div><div class="freshness">開始：{}</div>{}</article>'.format(
                    " stale" if stale else "",
                    html.escape(str(item.get("round_number", "?"))),
                    html.escape(detail),
                    html.escape(
                        "心跳 "
                        + format_dashboard_time(
                            item.get("heartbeat_at") or item.get("last_output_at")
                        )
                    ),
                    f'<p class="active-error">{html.escape(error)}</p>' if error else "",
                )
            )
        if not active_cards:
            active_cards.append(
                '<article class="current idle"><i class="pulse"></i><div>'
                '<h2>目前沒有執行中的輪次</h2><p>等待下一次啟動</p></div>'
                f'<div class="freshness">資料產生：{html.escape(format_dashboard_time(updated_at))}</div></article>'
            )

        completed_last_hour = 0
        for item in self.rounds:
            if item.get("worker_status") != "completed" or not item.get("finished_at"):
                continue
            try:
                finished = datetime.fromisoformat(str(item["finished_at"]))
                if finished.tzinfo is None:
                    finished = finished.astimezone()
                age = (
                    datetime.now().astimezone()
                    - finished
                ).total_seconds()
            except (TypeError, ValueError):
                continue
            if 0 <= age <= 3600:
                completed_last_hour += 1

        load_note = (
            "僅監測，不限制雙開"
            if settings.get("ignore_load_gate")
            else "參與自適應雙開判斷"
        )
        stats_html = "".join(
            '<div class="stat"><span>{}</span><b>{}</b><small>{}</small></div>'.format(
                html.escape(label), html.escape(value), html.escape(note)
            )
            for label, value, note in [
                ("總輪數", str(total_rounds), "已入帳輪次"),
                ("完整跑完", str(len(completed_rounds)), f"總成功率 {completion_rate:.0f}%"),
                ("失敗輪次", str(len(failed_rounds)), "含逾時與辨識失敗"),
                ("完成 41 抽", str(draw_completed), "已取得第三張截圖"),
                ("中獎候選", str(kept), "保留的引擎"),
                ("已清理引擎", str(deleted), "釋放儲存空間"),
                ("完整耗時中位數", format_duration(median_duration), "較不受清理重試影響"),
                ("近十輪成功率", f"{recent_rate:.0f}%", f"{recent_completed}/{len(recent_ten)} 輪完成"),
                (
                    "目前並行",
                    f"{len(active_items)}/{settings.get('parallel', 1)}",
                    ("自適應 · " if settings.get("adaptive_parallel") else "固定 · ")
                    + str(resources.get("admission_reason") or "正常調度"),
                ),
                ("近一小時產量", f"{completed_last_hour} 輪", "依完成時間滾動統計"),
                ("記憶體可用", f"{resources.get('memory_free_percent', '—')}%", f"Swap {resources.get('swap_used_mb', '—')} MB"),
                (
                    "系統負載",
                    str(resources.get("load1", "—")),
                    f"{load_note} · 磁碟 {resources.get('free_gib', '—')} GiB · "
                    + ("溫度警告" if resources.get("thermal_warning") else "溫度正常"),
                ),
            ]
        )

        styles = r'''
:root{--bg:#07101f;--panel:#101c31;--panel2:#15243d;--line:#263a5d;--text:#f4f7ff;--muted:#93a7c7;--blue:#66a8ff;--cyan:#58e0df;--ok:#35d487;--keep:#ffd166;--bad:#ff6b7d;--shadow:0 18px 50px rgba(0,0,0,.25)}
*{box-sizing:border-box}html{scroll-behavior:smooth}body{margin:0;background:radial-gradient(circle at 12% -10%,#163461 0,transparent 34%),radial-gradient(circle at 90% 0,#25225e 0,transparent 28%),var(--bg);color:var(--text);font:15px/1.55 -apple-system,BlinkMacSystemFont,"Segoe UI",sans-serif;min-height:100vh}
button,input{font:inherit}a{color:#8fc4ff}.shell{max-width:1440px;margin:auto;padding:30px 26px 70px}.topbar{display:flex;align-items:flex-start;justify-content:space-between;gap:24px;margin-bottom:22px}.eyebrow{margin:0 0 5px;color:var(--cyan);font-size:12px;font-weight:800;letter-spacing:.16em;text-transform:uppercase}.topbar h1{font-size:clamp(26px,4vw,42px);line-height:1.08;margin:0}.subtitle{color:var(--muted);margin:8px 0 0}.refresh-controls{display:flex;align-items:center;flex-wrap:wrap;justify-content:flex-end;gap:8px}.control{border:1px solid var(--line);background:rgba(16,28,49,.9);color:var(--text);border-radius:999px;padding:9px 14px;cursor:pointer}.control:hover{border-color:var(--blue)}.refresh-note{color:var(--muted);font-size:13px;flex-basis:100%;text-align:right}
.active-grid{display:grid;grid-template-columns:repeat(auto-fit,minmax(420px,1fr));gap:12px}.current{display:grid;grid-template-columns:auto 1fr auto;gap:16px;align-items:center;background:linear-gradient(135deg,rgba(42,84,143,.42),rgba(18,34,59,.88));border:1px solid #3c67a2;border-radius:18px;padding:18px 20px;box-shadow:var(--shadow)}.current.idle{grid-column:1/-1}.current.stale{border-color:#a16a37;background:linear-gradient(135deg,rgba(120,70,27,.32),rgba(18,34,59,.9))}.pulse{width:13px;height:13px;background:var(--ok);border-radius:50%;box-shadow:0 0 0 7px rgba(53,212,135,.12)}.stale .pulse{background:var(--keep);box-shadow:0 0 0 7px rgba(255,209,102,.12)}.current h2{font-size:16px;margin:0}.current p{margin:3px 0 0;color:#cad7ed}.freshness{color:var(--muted);font-size:13px;text-align:right}.active-error{grid-column:2/-1;margin:0;color:#ffd7dc;font-family:ui-monospace,SFMono-Regular,Menlo,monospace;font-size:12px;overflow-wrap:anywhere}
.summary{display:grid;grid-template-columns:repeat(4,minmax(170px,1fr));gap:12px;margin:20px 0}.stat{background:linear-gradient(150deg,rgba(24,42,70,.95),rgba(13,26,46,.95));border:1px solid var(--line);border-radius:16px;padding:16px 17px;min-height:116px}.stat span{color:var(--muted);font-size:13px}.stat b{display:block;font-size:28px;line-height:1.2;margin:8px 0 5px}.stat small{color:#8196b7}
.analysis-grid{display:grid;grid-template-columns:1fr;gap:14px;margin-bottom:26px}.panel{background:rgba(16,28,49,.9);border:1px solid var(--line);border-radius:18px;padding:18px;box-shadow:var(--shadow)}.panel-head{display:flex;align-items:end;justify-content:space-between;gap:12px;margin-bottom:16px}.panel h2{font-size:17px;margin:0}.panel-head span,.empty{color:var(--muted);font-size:13px}.result-map{display:grid;grid-template-columns:repeat(auto-fill,30px);gap:8px;align-items:center;min-height:30px;padding:4px 0}.result-cell{display:grid;place-items:center;width:30px;height:30px;border:0;border-radius:7px;color:#06121f;background:var(--ok);font-size:19px;font-weight:900;line-height:1;cursor:pointer;box-shadow:inset 0 -2px 0 rgba(0,0,0,.14);transition:transform .12s,filter .12s}.result-cell.fail{background:var(--bad)}.result-cell.keep{background:var(--keep)}.result-cell:hover,.result-cell:focus-visible{transform:translateY(-2px) scale(1.06);filter:brightness(1.12);outline:2px solid #fff;outline-offset:2px}.result-summary{display:flex;flex-wrap:wrap;gap:10px 22px;color:var(--muted);font-size:13px;margin-top:24px}.result-summary span{display:inline-flex;align-items:center;gap:7px}.summary-symbol{display:inline-grid;place-items:center;width:22px;height:22px;border-radius:6px;background:var(--ok);color:#06121f;font-style:normal;font-weight:900}.summary-symbol.fail{background:var(--bad)}.summary-symbol.keep{background:var(--keep)}
.failure-row{margin:12px 0}.failure-row>div:first-child{display:flex;justify-content:space-between;gap:12px}.failure-row span{color:#c9d5e9}.failure-track{height:7px;background:#0a1528;border-radius:10px;overflow:hidden;margin-top:6px}.failure-track i{display:block;height:100%;background:linear-gradient(90deg,#ff6b7d,#ff9a76);border-radius:10px}
.history-header{display:flex;align-items:end;justify-content:space-between;gap:18px;margin:25px 0 12px}.history-header h2{margin:0;font-size:23px}.history-header p{margin:4px 0 0;color:var(--muted)}.history-tools{display:flex;flex-wrap:wrap;justify-content:flex-end;gap:8px}.search{min-width:210px;border:1px solid var(--line);background:#0d192d;color:var(--text);border-radius:10px;padding:9px 12px;outline:none}.search:focus{border-color:var(--blue)}.filter{border:1px solid var(--line);background:#0d192d;color:var(--muted);border-radius:10px;padding:9px 12px;cursor:pointer}.filter.active{color:#fff;background:#244b7f;border-color:#4d82c7}
.round{background:rgba(16,28,49,.94);border:1px solid var(--line);border-left:5px solid var(--ok);border-radius:14px;margin:10px 0;overflow:hidden;box-shadow:0 8px 22px rgba(0,0,0,.16)}.round.failed{border-left-color:var(--bad)}.round.keep{border-left-color:var(--keep)}.round[hidden]{display:none}.round summary{cursor:pointer;display:grid;grid-template-columns:minmax(220px,1fr) auto auto auto;gap:18px;align-items:center;padding:15px 17px;list-style:none}.round summary::-webkit-details-marker{display:none}.round summary:after{content:"＋";color:var(--muted);font-size:20px}.round[open] summary:after{content:"−"}.round-identity{display:flex;align-items:center;gap:12px}.round-identity b{font-size:17px}.round-identity span{color:#c6d4e9}.badge{font-size:12px;font-weight:800;letter-spacing:.08em;border-radius:999px;padding:5px 9px;color:var(--ok);background:rgba(53,212,135,.12)}.failed .badge{color:var(--bad);background:rgba(255,107,125,.12)}.keep .badge{color:var(--keep);background:rgba(255,209,102,.12)}.duration,time{color:var(--muted);font-size:13px}.round-body{border-top:1px solid var(--line);padding:17px}.round-meta{display:flex;flex-wrap:wrap;gap:8px 18px;color:var(--muted);font-size:13px}.error-box{margin:14px 0;padding:12px 14px;background:rgba(255,107,125,.09);border:1px solid rgba(255,107,125,.26);border-radius:10px}.error-box strong{display:block;color:#ff9aa7;margin-bottom:5px}.error-box code{display:block;color:#f2cbd1;white-space:normal;overflow-wrap:anywhere;font-size:12px}.conditions{display:grid;grid-template-columns:repeat(auto-fit,minmax(260px,1fr));gap:10px;padding:0;margin:14px 0;list-style:none}.condition{position:relative;background:#0b172a;border:1px solid var(--line);border-radius:11px;padding:12px}.condition strong{display:block}.condition span{position:absolute;top:10px;right:10px;color:var(--ok);font-size:11px;text-transform:uppercase}.condition.missing span{color:var(--bad)}.condition p{color:var(--muted);font-size:12px;margin:5px 0 0}.links{display:flex;gap:14px}.shots{display:grid;grid-template-columns:repeat(auto-fit,minmax(210px,1fr));gap:11px;margin-top:14px}.shot{display:block;text-decoration:none;background:#07101f;border:1px solid var(--line);border-radius:11px;overflow:hidden;transition:transform .16s,border-color .16s}.shot:hover{transform:translateY(-2px);border-color:var(--blue)}.shot img{display:block;width:100%;height:165px;object-fit:contain;background:#040a13}.shot span{display:block;padding:8px 10px;color:#abc0df;font-size:12px;overflow:hidden;text-overflow:ellipsis;white-space:nowrap}.load-more{display:block;margin:18px auto 0}.history-count{text-align:center;color:var(--muted);font-size:13px;margin-top:10px}
@media(max-width:1050px){.summary{grid-template-columns:repeat(2,1fr)}.analysis-grid{grid-template-columns:1fr}.round summary{grid-template-columns:1fr auto auto}.round summary time{display:none}}
@media(max-width:700px){.shell{padding:20px 14px 50px}.topbar,.history-header{display:block}.refresh-controls,.history-tools{justify-content:flex-start;margin-top:14px}.refresh-note{text-align:left}.active-grid{grid-template-columns:1fr}.current{grid-template-columns:auto 1fr}.freshness{grid-column:2;text-align:left}.summary{grid-template-columns:1fr 1fr}.stat{min-height:102px}.stat b{font-size:22px}.round summary{grid-template-columns:1fr auto}.round summary .duration{display:none}.search{width:100%}}
'''
        script = r'''
(() => {
  const cards = [...document.querySelectorAll('.round')];
  const filters = [...document.querySelectorAll('[data-filter]')];
  const search = document.querySelector('#history-search');
  const more = document.querySelector('#load-more');
  const count = document.querySelector('#history-count');
  const refreshButton = document.querySelector('#refresh-now');
  const autoButton = document.querySelector('#auto-refresh');
  const refreshNote = document.querySelector('#refresh-note');
  let limit = 15;
  let filter = localStorage.getItem('aov-dashboard-filter') || 'all';
  let autoRefresh = localStorage.getItem('aov-dashboard-auto') !== 'off';
  let remaining = 30;

  const applyHistory = () => {
    const query = (search.value || '').trim().toLocaleLowerCase();
    let matched = 0;
    let shown = 0;
    cards.forEach(card => {
      const filterMatch = filter === 'all' || card.dataset.outcome === filter;
      const queryMatch = !query || (card.dataset.search || '').includes(query);
      const eligible = filterMatch && queryMatch;
      if (eligible) matched += 1;
      const visible = eligible && shown < limit;
      card.hidden = !visible;
      if (visible) shown += 1;
    });
    more.hidden = shown >= matched;
    count.textContent = `顯示 ${shown} / ${matched} 筆符合紀錄`;
    filters.forEach(button => button.classList.toggle('active', button.dataset.filter === filter));
  };
  filters.forEach(button => button.addEventListener('click', () => {
    filter = button.dataset.filter;
    localStorage.setItem('aov-dashboard-filter', filter);
    limit = 15;
    applyHistory();
  }));
  search.addEventListener('input', () => { limit = 15; applyHistory(); });
  more.addEventListener('click', () => { limit += 15; applyHistory(); });

  const saveView = () => {
    sessionStorage.setItem('aov-dashboard-view', JSON.stringify({
      scrollY: window.scrollY,
      open: cards.filter(card => card.open).map(card => card.id),
      query: search.value,
      filter,
      limit
    }));
  };
  const reload = () => { saveView(); location.reload(); };
  refreshButton.addEventListener('click', reload);
  autoButton.addEventListener('click', () => {
    autoRefresh = !autoRefresh;
    localStorage.setItem('aov-dashboard-auto', autoRefresh ? 'on' : 'off');
    remaining = 30;
  });
  const browsingHistory = () => window.scrollY > 120 || cards.some(card => card.open) || document.activeElement === search;
  const updateRefreshLabel = () => {
    autoButton.textContent = autoRefresh ? '自動更新：開' : '自動更新：關';
    if (!autoRefresh) refreshNote.textContent = '自動更新已關閉；可隨時手動更新。';
    else if (browsingHistory()) refreshNote.textContent = '正在查看歷史，自動更新已暫停，不會打斷目前位置。';
    else refreshNote.textContent = `${remaining} 秒後更新；向下捲動或展開紀錄會自動暫停。`;
  };
  setInterval(() => {
    if (!autoRefresh || document.hidden || browsingHistory()) {
      remaining = 30;
      updateRefreshLabel();
      return;
    }
    remaining -= 1;
    if (remaining <= 0) reload();
    updateRefreshLabel();
  }, 1000);
  window.addEventListener('scroll', updateRefreshLabel, {passive:true});
  cards.forEach(card => card.addEventListener('toggle', updateRefreshLabel));
  document.querySelectorAll('[data-jump]').forEach(button => button.addEventListener('click', () => {
    const card = document.getElementById(button.dataset.jump);
    if (!card) return;
    limit = Math.max(limit, Number(card.dataset.index || 0) + 1);
    applyHistory();
    card.hidden = false;
    card.open = true;
    card.scrollIntoView({behavior:'smooth', block:'start'});
  }));

  try {
    const saved = JSON.parse(sessionStorage.getItem('aov-dashboard-view') || 'null');
    if (saved) {
      filter = saved.filter || filter;
      limit = saved.limit || limit;
      search.value = saved.query || '';
      applyHistory();
      (saved.open || []).forEach(id => { const card = document.getElementById(id); if (card) card.open = true; });
      requestAnimationFrame(() => window.scrollTo(0, saved.scrollY || 0));
      sessionStorage.removeItem('aov-dashboard-view');
    }
  } catch (_) {}
  applyHistory();
  updateRefreshLabel();
})();
'''
        page = f'''<!doctype html>
<html lang="zh-Hant"><head><meta charset="utf-8">
<meta name="viewport" content="width=device-width,initial-scale=1"><title>AOV 首抽監測</title>
<style>{styles}</style></head><body><main class="shell">
<header class="topbar"><div><p class="eyebrow">Arena of Valor · Reroll Monitor</p><h1>首抽連續執行儀表板</h1><p class="subtitle">追蹤每輪完成率、失敗位置、清理狀態與所有結果截圖。</p></div>
<div class="refresh-controls"><button id="refresh-now" class="control">立即更新</button><button id="auto-refresh" class="control">自動更新：開</button><span id="refresh-note" class="refresh-note"></span></div></header>
<section class="active-grid">{''.join(active_cards)}</section>
<section class="summary">{stats_html}</section>
<section class="analysis-grid"><article class="panel"><div class="panel-head"><div><h2>全部歷史結果</h2><span>每格代表一輪，依輪號由左至右、由上而下；點擊可跳到詳細紀錄</span></div></div><div class="result-map">{''.join(history_items) or '<p class="empty">尚無輪次。</p>'}</div>{history_summary}</article>
<article class="panel"><div class="panel-head"><div><h2>失敗原因</h2><span>依每輪 run.json 的終止原因分類</span></div></div>{failure_rows}</article></section>
<section><div class="history-header"><div><h2>逐輪紀錄</h2><p>預設顯示最新 15 輪，可搜尋、篩選與展開截圖。</p></div><div class="history-tools"><input id="history-search" class="search" type="search" placeholder="搜尋輪次或引擎"><button class="filter active" data-filter="all">全部</button><button class="filter" data-filter="complete">完成</button><button class="filter" data-filter="failed">失敗</button><button class="filter" data-filter="keep">保留</button></div></div>
{''.join(cards) or '<p class="empty">尚無完成紀錄。</p>'}<button id="load-more" class="control load-more">顯示更早紀錄</button><p id="history-count" class="history-count"></p></section>
</main><script>{script}</script></body></html>'''
        temporary = self.dashboard_path.with_suffix(".tmp")
        temporary.write_text(page, encoding="utf-8")
        os.replace(temporary, self.dashboard_path)


def acquire_lock(path: Path) -> Any:
    path.parent.mkdir(parents=True, exist_ok=True)
    handle = path.open("a+")
    try:
        fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
    except BlockingIOError as exc:
        raise SystemExit("另一個 AOV 連續批次已在執行；本次不重複啟動。") from exc
    handle.seek(0)
    handle.truncate()
    handle.write(f"pid={os.getpid()} started={now_iso()}\n")
    handle.flush()
    return handle


def instances_map() -> dict[str, str]:
    return {item.engine: item.display_name for item in aov_runner.read_instances()}


def system_resources(output_root: Path) -> dict[str, Any]:
    memory_free: float | None = None
    swap_used: float | None = None
    thermal_warning = False
    try:
        completed = subprocess.run(
            ["memory_pressure", "-Q"],
            capture_output=True,
            text=True,
            timeout=10,
            check=False,
        )
        match = re.search(
            r"System-wide memory free percentage:\s*([0-9.]+)%?",
            completed.stdout + completed.stderr,
        )
        if match:
            memory_free = float(match.group(1))
    except (OSError, subprocess.SubprocessError, ValueError):
        pass
    try:
        completed = subprocess.run(
            ["sysctl", "vm.swapusage"],
            capture_output=True,
            text=True,
            timeout=10,
            check=False,
        )
        match = re.search(r"used\s*=\s*([0-9.]+)M", completed.stdout)
        if match:
            swap_used = float(match.group(1))
    except (OSError, subprocess.SubprocessError, ValueError):
        pass
    try:
        completed = subprocess.run(
            ["pmset", "-g", "therm"],
            capture_output=True,
            text=True,
            timeout=10,
            check=False,
        )
        output = completed.stdout + completed.stderr
        thermal_warning = not (
            "No thermal warning level has been recorded" in output
            and "No performance warning level has been recorded" in output
        )
    except (OSError, subprocess.SubprocessError):
        pass
    return {
        "at": now_iso(),
        "free_gib": round(aov_runner.available_gib(output_root), 1),
        "memory_free_percent": round(memory_free, 1) if memory_free is not None else None,
        "swap_used_mb": round(swap_used, 1) if swap_used is not None else None,
        "load1": round(os.getloadavg()[0], 2),
        "thermal_warning": thermal_warning,
    }


class AdaptiveGate:
    def __init__(self, *, ignore_load: bool = False) -> None:
        self.degraded = False
        self.healthy_since: float | None = None
        self.reason = "正常調度"
        self.ignore_load = ignore_load

    def observe(self, resources: dict[str, Any], active_count: int) -> None:
        memory_free = resources.get("memory_free_percent")
        load1 = float(resources.get("load1") or 0.0)
        thermal = bool(resources.get("thermal_warning"))
        critical = (
            (memory_free is not None and float(memory_free) < 15.0)
            or (not self.ignore_load and load1 >= 7.0)
            or thermal
        )
        if active_count >= 2 and critical:
            self.degraded = True
            self.healthy_since = None
        if not self.degraded:
            self.reason = "正常調度"
            return
        if thermal:
            self.reason = "偵測到溫度／效能警告，暫停補第二個 worker"
        elif memory_free is not None and float(memory_free) < 30.0:
            self.reason = f"記憶體可用 {float(memory_free):.1f}%，等待恢復至 30%"
        elif not self.ignore_load and load1 >= 5.5:
            self.reason = f"系統負載 {load1:.2f}，等待降至 5.5 以下"
        else:
            if memory_free is None:
                self.healthy_since = None
                self.reason = "無法取得 memory_pressure，維持單開"
                return
            if self.healthy_since is None:
                self.healthy_since = time.monotonic()
            remaining = max(0.0, 300.0 - (time.monotonic() - self.healthy_since))
            if remaining <= 0:
                self.degraded = False
                self.healthy_since = None
                self.reason = "正常調度"
            else:
                self.reason = f"資源已恢復，穩定觀察剩餘 {remaining:.0f} 秒"

    def allow_slot(
        self,
        *,
        active_count: int,
        resources: dict[str, Any],
        args: argparse.Namespace,
        cleanup_blocked: bool,
    ) -> tuple[bool, str]:
        if cleanup_blocked:
            return False, "正在清理或重試刪除引擎"
        free_gib = float(resources.get("free_gib") or 0.0)
        if active_count == 0:
            if free_gib < args.min_free_gb:
                return False, f"磁碟 {free_gib:.1f} GiB < {args.min_free_gb:.1f} GiB"
            return True, "第一個 worker 可啟動"
        if active_count >= args.parallel:
            return False, "已達並行上限"
        if free_gib < args.second_slot_min_free_gb:
            return False, (
                f"第二槽磁碟 {free_gib:.1f} GiB < "
                f"{args.second_slot_min_free_gb:.1f} GiB"
            )
        if not args.adaptive_parallel:
            return True, "固定雙開"
        if self.degraded:
            return False, self.reason
        memory_free = resources.get("memory_free_percent")
        if memory_free is None:
            return False, "無法取得 memory_pressure，保守維持單開"
        if float(memory_free) < 25.0:
            return False, f"記憶體可用 {float(memory_free):.1f}% < 25%"
        load1 = float(resources.get("load1") or 0.0)
        if not self.ignore_load and load1 >= 6.0:
            return False, f"系統負載 {load1:.2f} >= 6"
        if resources.get("thermal_warning"):
            return False, "偵測到溫度／效能警告"
        return True, "自適應門檻通過"


def find_run_dir(output_root: Path, engine: str, known_dirs: set[str]) -> Path | None:
    candidates: list[Path] = []
    for directory in output_root.iterdir() if output_root.is_dir() else []:
        if not directory.is_dir() or directory.name in known_dirs:
            continue
        manifest = load_json(directory / "run.json")
        if manifest and manifest.get("engine") == engine:
            candidates.append(directory)
    return max(candidates, key=lambda path: path.stat().st_mtime) if candidates else None


def terminate_process_group(process: subprocess.Popen[str], context: BatchContext, reason: str) -> None:
    if process.poll() is not None:
        return
    context.event("watchdog", f"終止無回應的單輪程序：{reason}", pid=process.pid)
    try:
        os.killpg(process.pid, signal.SIGTERM)
    except ProcessLookupError:
        return
    try:
        process.wait(timeout=12)
        return
    except subprocess.TimeoutExpired:
        pass
    try:
        os.killpg(process.pid, signal.SIGKILL)
    except ProcessLookupError:
        pass
    try:
        process.wait(timeout=5)
    except subprocess.TimeoutExpired:
        pass


def close_owned_engine(engine: str, context: BatchContext) -> None:
    if engine not in instances_map():
        return
    try:
        aov_runner.close_instance(engine, timeout=90)
    except Exception as exc:
        context.event("cleanup_error", f"關閉 {engine} 失敗，改用精確程序清理", error=str(exc))
        remaining = aov_runner.engine_pids(engine)
        for pid in remaining:
            try:
                os.kill(pid, signal.SIGTERM)
            except ProcessLookupError:
                pass
        deadline = time.monotonic() + 15
        while time.monotonic() < deadline and aov_runner.engine_pids(engine):
            time.sleep(1)
        remaining = aov_runner.engine_pids(engine)
        if remaining:
            context.event(
                "cleanup_forced",
                f"強制停止批次專屬引擎 {engine}",
                pids=remaining,
            )
            for pid in remaining:
                try:
                    os.kill(pid, signal.SIGKILL)
                except ProcessLookupError:
                    pass
            deadline = time.monotonic() + 20
            while time.monotonic() < deadline and aov_runner.engine_pids(engine):
                time.sleep(0.5)
        if aov_runner.engine_pids(engine):
            raise RuntimeError(f"Cannot stop exact engine {engine} after SIGKILL")


def delete_owned_engine(
    engine: str,
    display_name: str,
    protected: dict[str, str],
    context: BatchContext,
    args: argparse.Namespace,
    round_number: int,
) -> str:
    """Retry one exact deletion forever; do not create more engines meanwhile."""
    if engine in protected:
        raise RuntimeError(f"Safety gate: {engine} existed before this round")
    attempt = 0
    context.begin_cleanup()
    try:
        while engine in instances_map():
            if STOP_EVENT.is_set():
                raise InterruptedError(
                    f"supervisor stopped while deletion of {engine} was pending"
                )
            attempt += 1
            try:
                with aov_runner.lifecycle_lock(
                    Path(args.lifecycle_lock),
                    args.lifecycle_lock_timeout,
                ):
                    close_owned_engine(engine, context)
                    method = aov_runner.delete_instance_with_manager(
                        engine, display_name, timeout=240
                    )
                with context.lock:
                    protected_now = dict(context.protected_engines or protected)
                missing_protected = sorted(set(protected_now) - set(instances_map()))
                if missing_protected:
                    raise RuntimeError(
                        "Protected engines disappeared during cleanup: "
                        + ", ".join(missing_protected)
                    )
                settle_seconds = max(0.0, args.post_delete_settle_seconds)
                if settle_seconds > 0:
                    context.event(
                        "post_delete_settle",
                        (
                            f"{engine} 已刪除；等待 {settle_seconds:.0f} 秒讓 "
                            "BlueStacks 釋放資源後再建立新引擎"
                        ),
                        round_number=round_number,
                        engine=engine,
                        settle_seconds=settle_seconds,
                    )
                    context.update_active(
                        round_number,
                        engine=engine,
                        status="post_delete_settle",
                        stage="post_delete_settle",
                        last_error=None,
                        last_output_at=now_iso(),
                    )
                    context.persist()
                    if STOP_EVENT.wait(settle_seconds):
                        raise InterruptedError(
                            f"supervisor stopped during post-delete settle for {engine}"
                        )
                return method
            except (KeyboardInterrupt, InterruptedError):
                raise
            except Exception as exc:
                context.event(
                    "delete_retry",
                    f"{engine} 刪除失敗；不新增引擎，"
                    f"{args.delete_retry_seconds:.0f} 秒後重試",
                    round_number=round_number,
                    attempt=attempt,
                    error=str(exc),
                )
                context.update_active(
                    round_number,
                    engine=engine,
                    status="delete_retry",
                    stage="delete_retry",
                    last_error=str(exc),
                    last_output_at=now_iso(),
                )
                context.persist()
                if STOP_EVENT.wait(max(1.0, args.delete_retry_seconds)):
                    raise InterruptedError(
                        f"supervisor stopped while deletion of {engine} was pending"
                    )
        return "already_deleted"
    finally:
        context.end_cleanup()


def update_run_manifest(run_dir: Path | None, fields: dict[str, Any]) -> None:
    if run_dir is None:
        return
    path = run_dir / "run.json"
    manifest = load_json(path) or {}
    manifest.update(fields)
    atomic_json(path, manifest)


def collect_images(run_dir: Path | None, context: BatchContext) -> list[dict[str, str]]:
    if run_dir is None:
        return []
    paths: list[Path] = []
    for pattern in (
        "*_01_activity_rewards.png",
        "*_02_lsr_box_rewards.png",
        "*_03_41_draw_rewards.png",
        "*_draw_gift_popup*.png",
        "*_failure.png",
        "*_timeout.png",
        "jackpot-evidence/*.png",
    ):
        paths.extend(sorted(run_dir.glob(pattern)))
    unique: list[Path] = []
    seen: set[Path] = set()
    for path in paths:
        if path not in seen:
            seen.add(path)
            unique.append(path)
    return [
        {
            "label": path.name,
            "href": relative_href(path, context.batch_root),
        }
        for path in unique
    ]


def finalize_round(
    *,
    context: BatchContext,
    round_number: int,
    started_wall: float,
    started_at: str,
    worker_status: str,
    worker_exit_code: int | None,
    watchdog_reason: str | None,
    engine: str | None,
    display_name: str | None,
    protected: dict[str, str],
    run_dir: Path | None,
    log_path: Path,
    args: argparse.Namespace,
) -> dict[str, Any]:
    decision: dict[str, Any] = {
        "decision": "delete",
        "keep_engine": False,
        "conditions": [],
    }
    if run_dir is not None:
        try:
            decision = jackpot_detector.analyze_run(run_dir, write=True)
        except Exception as exc:
            context.event("decision_error", "截圖判定失敗；本輪依規則刪除", error=str(exc))
            decision = {
                "decision": "delete",
                "keep_engine": False,
                "conditions": [],
                "analysis_error": str(exc),
            }

    engine_action = "no_engine"
    delete_method: str | None = None
    if engine is not None:
        if engine in protected:
            raise RuntimeError(f"Safety gate: worker engine {engine} was not newly created")
        if bool(decision.get("keep_engine")):
            context.begin_cleanup()
            try:
                with aov_runner.lifecycle_lock(
                    Path(args.lifecycle_lock),
                    args.lifecycle_lock_timeout,
                ):
                    close_owned_engine(engine, context)
                engine_action = "preserved"
                with context.lock:
                    context.protected_engines[engine] = display_name or engine
                    for other_number, other in context.active_rounds.items():
                        if other_number == round_number:
                            continue
                        other_protected = dict(other.get("protected_engines") or {})
                        other_protected[engine] = display_name or engine
                        other["protected_engines"] = other_protected
                    context.write_active()
            finally:
                context.end_cleanup()
            context.event("jackpot_candidate", f"保留中獎候選引擎 {engine}")
        else:
            delete_method = delete_owned_engine(
                engine,
                display_name or engine,
                protected,
                context,
                args,
                round_number,
            )
            engine_action = "deleted"
            context.event("cleanup", f"普通／失敗引擎 {engine} 已刪除")

    finished_at = now_iso()
    record: dict[str, Any] = {
        "schema_version": 1,
        "round_number": round_number,
        "started_at": started_at,
        "finished_at": finished_at,
        "duration_seconds": max(0.0, time.time() - started_wall),
        "worker_status": worker_status,
        "worker_exit_code": worker_exit_code,
        "watchdog_reason": watchdog_reason,
        "engine": engine,
        "display_name": display_name,
        "decision": str(decision.get("decision", "delete")),
        "engine_action": engine_action,
        "delete_method": delete_method,
        "conditions": decision.get("conditions") or [],
        "analysis_error": decision.get("analysis_error"),
        "run_dir": str(run_dir) if run_dir else None,
        "run_href": relative_href(run_dir, context.batch_root) if run_dir else "",
        "log": str(log_path),
        "log_href": relative_href(log_path, context.batch_root),
    }
    record["images"] = collect_images(run_dir, context)
    update_run_manifest(
        run_dir,
        {
            "batch_decision": record["decision"],
            "batch_engine_action": engine_action,
            "batch_delete_method": delete_method,
            "engine_deleted": engine_action == "deleted",
            "batch_finished_at": finished_at,
        },
    )
    return record


def migrate_legacy_active(context: BatchContext) -> None:
    legacy = load_json(context.legacy_active_path)
    if not legacy:
        return
    round_number = int(legacy.get("round_number", len(context.rounds) + 1))
    if round_number not in context.active_rounds:
        legacy["schema_version"] = 2
        legacy.setdefault("status", "interrupted")
        context.set_active(legacy)
        context.event("migration", f"舊版 active_round 已轉入第 {round_number} 輪")
    context.legacy_active_path.unlink(missing_ok=True)


def infer_engine_from_log(active: dict[str, Any]) -> tuple[str, str] | None:
    log_value = active.get("log")
    if not log_value:
        return None
    try:
        text = Path(str(log_value)).read_text(encoding="utf-8", errors="replace")
    except OSError:
        return None
    matches = list(ENGINE_LINE.finditer(text))
    if not matches:
        return None
    match = matches[-1]
    return match.group(1), match.group(2)


def recover_active_round(
    context: BatchContext,
    args: argparse.Namespace,
    round_number: int,
    *,
    reason: str,
) -> None:
    with context.lock:
        active = dict(context.active_rounds.get(round_number) or {})
    if not active:
        return
    if any(
        int(item.get("round_number", -1)) == round_number
        for item in context.rounds
    ):
        context.event("recovery", f"第 {round_number} 輪早已入帳；清除重複旗標")
        context.remove_active(round_number)
        return

    current = instances_map()
    protected = {
        str(key): str(value)
        for key, value in (active.get("protected_engines") or {}).items()
    }
    engine = str(active.get("engine") or "") or None
    display_name = str(active.get("display_name") or "") or None
    if engine is None:
        inferred = infer_engine_from_log(active)
        if inferred and inferred[0] in current:
            engine, display_name = inferred
        else:
            other_claims = {
                str(item["engine"])
                for number, item in context.active_rounds.items()
                if number != round_number and item.get("engine")
            }
            candidates = sorted(set(current) - set(protected) - other_claims)
            if len(candidates) == 1:
                engine = candidates[0]
                display_name = current[engine]
            elif len(candidates) > 1:
                raise RuntimeError(
                    f"Interrupted round {round_number} has ambiguous unowned engines: "
                    + ", ".join(candidates)
                )
        if engine:
            context.update_active(
                round_number,
                engine=engine,
                display_name=display_name or current.get(engine, engine),
            )

    run_dir = Path(str(active["run_dir"])) if active.get("run_dir") else None
    if run_dir is None and engine:
        run_dir = find_run_dir(
            context.output_root,
            engine,
            set(active.get("known_run_dirs") or []),
        )
    log_path = Path(
        str(active.get("log") or (context.logs_dir / f"recovered_{round_number}.log"))
    )
    record = finalize_round(
        context=context,
        round_number=round_number,
        started_wall=float(active.get("started_wall", time.time())),
        started_at=str(active.get("started_at", now_iso())),
        worker_status="interrupted_recovered",
        worker_exit_code=None,
        watchdog_reason=reason,
        engine=engine,
        display_name=display_name or (current.get(engine, engine) if engine else None),
        protected=protected,
        run_dir=run_dir,
        log_path=log_path,
        args=args,
    )
    context.add_round(record)
    context.remove_active(round_number)


def recover_interrupted_rounds(context: BatchContext, args: argparse.Namespace) -> None:
    if not context.active_rounds:
        return
    context.event(
        "recovery",
        f"發現 {len(context.active_rounds)} 個未完成輪次，依序判定並清理",
    )
    for round_number in list(sorted(context.active_rounds)):
        recover_active_round(
            context,
            args,
            round_number,
            reason="supervisor previously stopped",
        )
    context.protected_engines = instances_map()
    context.persist()


def _supervise_one_round_inner(
    context: BatchContext,
    args: argparse.Namespace,
    round_number: int,
) -> dict[str, Any]:
    with context.lock:
        protected = dict(context.protected_engines)
    known_run_dirs = {
        path.name for path in context.output_root.iterdir() if path.is_dir()
    } if context.output_root.is_dir() else set()
    started_at = now_iso()
    started_wall = time.time()
    log_path = context.logs_dir / f"{timestamp()}_round_{round_number:06d}.log"
    active: dict[str, Any] = {
        "schema_version": 2,
        "round_number": round_number,
        "started_at": started_at,
        "started_wall": started_wall,
        "protected_engines": protected,
        "known_run_dirs": sorted(known_run_dirs),
        "engine": None,
        "display_name": None,
        "run_dir": None,
        "log": str(log_path),
        "status": "starting",
        "stage": "create",
        "last_output_at": now_iso(),
        "heartbeat_at": now_iso(),
        "lifecycle_ready": False,
    }
    context.set_active(active)
    command = [
        sys.executable,
        str(args.runner),
        "fresh-run",
        "--xapk",
        str(args.xapk),
        "--workflow",
        str(args.workflow),
        "--output-root",
        str(context.output_root),
        "--min-free-gb",
        str(args.min_free_gb),
        "--engine-cpu",
        str(args.engine_cpu),
        "--engine-ram-mb",
        str(args.engine_ram_mb),
        "--lifecycle-lock",
        str(args.lifecycle_lock),
        "--lifecycle-lock-timeout",
        str(args.lifecycle_lock_timeout),
    ]
    for peer in sorted(context.owned_engines(excluding_round=round_number)):
        command.extend(["--allow-running-engine", peer])
    context.event("round_start", f"開始第 {round_number} 輪", log=str(log_path))
    process = subprocess.Popen(
        command,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        text=True,
        bufsize=1,
        start_new_session=True,
    )
    with ACTIVE_WORKERS_LOCK:
        ACTIVE_WORKERS[round_number] = process
    context.update_active(round_number, pid=process.pid, status="running")
    assert process.stdout is not None
    selector = selectors.DefaultSelector()
    selector.register(process.stdout, selectors.EVENT_READ)
    last_output = time.monotonic()
    last_output_wall = now_iso()
    last_dashboard = 0.0
    watchdog_reason: str | None = None
    engine: str | None = None
    display_name: str | None = None
    run_dir: Path | None = None
    with log_path.open("a", encoding="utf-8") as log:
        log.write(f"command: {json.dumps(command, ensure_ascii=False)}\n")
        log.flush()
        while True:
            for key, _ in selector.select(timeout=1.0):
                line = key.fileobj.readline()
                if line:
                    print(f"[round {round_number}] {line}", end="", flush=True)
                    log.write(line)
                    log.flush()
                    last_output = time.monotonic()
                    last_output_wall = now_iso()
                    match = ENGINE_LINE.search(line)
                    if match:
                        candidate = match.group(1)
                        with context.lock:
                            protected_now = dict(context.protected_engines)
                        other_owners = context.owned_engines(excluding_round=round_number)
                        if candidate in protected_now:
                            raise RuntimeError(
                                f"Worker reported a pre-existing engine as newly created: {candidate}"
                            )
                        if candidate in other_owners:
                            raise RuntimeError(
                                f"Engine {candidate} is already owned by round "
                                f"{other_owners[candidate]}"
                            )
                        engine, display_name = candidate, match.group(2)
                        context.update_active(
                            round_number,
                            engine=engine,
                            display_name=display_name,
                            last_output_at=now_iso(),
                        )
                    ready_match = ADB_READY_LINE.search(line)
                    if ready_match:
                        ready_engine = ready_match.group(1)
                        if engine is not None and ready_engine != engine:
                            raise RuntimeError(
                                f"ADB-ready engine {ready_engine} does not match owner {engine}"
                            )
                        context.update_active(
                            round_number,
                            engine=ready_engine,
                            lifecycle_ready=True,
                            adb_endpoint=ready_match.group(2),
                            last_output_at=now_iso(),
                        )
            if engine is None:
                current_instances = instances_map()
                with context.lock:
                    protected_now = set(context.protected_engines)
                peer_engines = set(context.owned_engines(excluding_round=round_number))
                added = sorted(set(current_instances) - protected_now - peer_engines)
                if len(added) == 1:
                    engine = added[0]
                    display_name = current_instances[engine]
                    context.update_active(
                        round_number,
                        engine=engine,
                        display_name=display_name,
                        last_output_at=now_iso(),
                    )
            if engine is not None and run_dir is None:
                run_dir = find_run_dir(context.output_root, engine, known_run_dirs)
                if run_dir is not None:
                    context.update_active(round_number, run_dir=str(run_dir))

            now = time.monotonic()
            if now - last_dashboard >= 15:
                progress = load_json(run_dir / "progress.json") if run_dir else None
                context.state["status"] = "running"
                context.update_active(
                    round_number,
                    engine=engine,
                    display_name=display_name,
                    run_dir=str(run_dir) if run_dir else None,
                    status="running",
                    stage=progress.get("stage") if progress else "starting",
                    heartbeat_at=now_iso(),
                    last_output_at=last_output_wall,
                    last_error=None,
                )
                context.persist()
                last_dashboard = now
            if process.poll() is not None:
                # Drain any line that arrived with EOF.
                remainder = process.stdout.read()
                if remainder:
                    print(remainder, end="", flush=True)
                    log.write(remainder)
                    log.flush()
                break
            if STOP_EVENT.is_set():
                terminate_process_group(process, context, "batch supervisor stopping")
                break
            if now - last_output > args.silence_timeout:
                watchdog_reason = f"no worker output for {args.silence_timeout:.0f}s"
                terminate_process_group(process, context, watchdog_reason)
                break
            if time.time() - started_wall > args.round_timeout:
                watchdog_reason = f"round exceeded {args.round_timeout:.0f}s"
                terminate_process_group(process, context, watchdog_reason)
                break
    selector.close()
    if STOP_EVENT.is_set():
        raise InterruptedError(f"round {round_number} stopped with supervisor")
    if engine is not None and run_dir is None:
        run_dir = find_run_dir(context.output_root, engine, known_run_dirs)
    exit_code = process.poll()
    manifest = load_json(run_dir / "run.json") if run_dir else None
    worker_status = str(manifest.get("status")) if manifest else (
        "completed" if exit_code == 0 and watchdog_reason is None else "failed"
    )
    record = finalize_round(
        context=context,
        round_number=round_number,
        started_wall=started_wall,
        started_at=started_at,
        worker_status=worker_status,
        worker_exit_code=exit_code,
        watchdog_reason=watchdog_reason,
        engine=engine,
        display_name=display_name,
        protected=dict(context.protected_engines),
        run_dir=run_dir,
        log_path=log_path,
        args=args,
    )
    return record


def supervise_one_round(
    context: BatchContext,
    args: argparse.Namespace,
    round_number: int,
) -> dict[str, Any]:
    try:
        return _supervise_one_round_inner(context, args, round_number)
    except BaseException:
        with ACTIVE_WORKERS_LOCK:
            process = ACTIVE_WORKERS.get(round_number)
        if process is not None:
            terminate_process_group(
                process,
                context,
                "supervisor exception or user interruption",
            )
        raise
    finally:
        with ACTIVE_WORKERS_LOCK:
            ACTIVE_WORKERS.pop(round_number, None)


def round_job(
    context: BatchContext,
    args: argparse.Namespace,
    round_number: int,
) -> dict[str, Any] | None:
    try:
        return supervise_one_round(context, args, round_number)
    except InterruptedError:
        raise
    except BaseException as exc:
        context.event(
            "supervisor_error",
            f"第 {round_number} 輪監督發生例外；只復原這一輪",
            round_number=round_number,
            error=f"{type(exc).__name__}: {exc}",
            traceback=traceback.format_exc(),
        )
        recover_active_round(
            context,
            args,
            round_number,
            reason=f"round supervisor exception: {type(exc).__name__}: {exc}",
        )
        return None


def parser() -> argparse.ArgumentParser:
    result = argparse.ArgumentParser(description=__doc__)
    result.add_argument("--xapk", type=Path, required=True)
    result.add_argument("--runner", type=Path, default=DEFAULT_RUNNER)
    result.add_argument("--workflow", type=Path, default=DEFAULT_WORKFLOW)
    result.add_argument("--output-root", type=Path, required=True)
    result.add_argument("--batch-root", type=Path, required=True)
    result.add_argument("--max-rounds", type=int, default=0, help="0 means continue until Ctrl-C")
    result.add_argument("--min-free-gb", type=float, default=20.0)
    result.add_argument("--round-timeout", type=float, default=3600.0)
    result.add_argument("--silence-timeout", type=float, default=420.0)
    result.add_argument("--delete-retry-seconds", type=float, default=30.0)
    result.add_argument(
        "--post-delete-settle-seconds",
        type=float,
        default=10.0,
        help="keep engine creation blocked briefly after a successful Manager deletion",
    )
    result.add_argument("--between-rounds", type=float, default=3.0)
    result.add_argument("--parallel", type=int, default=1, choices=(1, 2))
    result.add_argument("--engine-cpu", type=int, default=4)
    result.add_argument("--engine-ram-mb", type=int, default=4096)
    result.add_argument("--start-stagger-seconds", type=float, default=120.0)
    result.add_argument("--adaptive-parallel", action="store_true")
    result.add_argument(
        "--ignore-load-gate",
        action="store_true",
        help="show load average in the dashboard but do not use it to block slot two",
    )
    result.add_argument(
        "--startup-failure-backoff-seconds",
        type=float,
        default=90.0,
        help="delay refilling slot two after a fresh-engine game launch failure",
    )
    result.add_argument(
        "--startup-failure-backoff-max-seconds",
        type=float,
        default=300.0,
        help="maximum slot-two cooldown after consecutive startup failures",
    )
    result.add_argument("--second-slot-min-free-gb", type=float, default=26.0)
    result.add_argument(
        "--lifecycle-lock",
        type=Path,
        default=aov_runner.DEFAULT_LIFECYCLE_LOCK,
    )
    result.add_argument("--lifecycle-lock-timeout", type=float, default=600.0)
    return result


def validate_args(args: argparse.Namespace) -> None:
    for name in ("xapk", "runner", "workflow"):
        value = Path(getattr(args, name)).expanduser().resolve()
        if not value.is_file():
            raise SystemExit(f"{name} not found: {value}")
        setattr(args, name, value)
    args.output_root = args.output_root.expanduser().resolve()
    args.batch_root = args.batch_root.expanduser().resolve()
    args.lifecycle_lock = args.lifecycle_lock.expanduser().resolve()
    args.output_root.mkdir(parents=True, exist_ok=True)
    args.batch_root.mkdir(parents=True, exist_ok=True)
    if args.max_rounds < 0:
        raise SystemExit("--max-rounds must be 0 or positive")
    if args.silence_timeout < 60 or args.round_timeout < args.silence_timeout:
        raise SystemExit("timeouts are invalid: round timeout must exceed silence timeout >= 60s")
    if not 1 <= args.engine_cpu <= 8:
        raise SystemExit("--engine-cpu must be between 1 and 8")
    if not 2048 <= args.engine_ram_mb <= 8192:
        raise SystemExit("--engine-ram-mb must be between 2048 and 8192")
    if args.start_stagger_seconds < 0 or args.between_rounds < 0:
        raise SystemExit("launch delays must not be negative")
    if args.post_delete_settle_seconds < 0:
        raise SystemExit("--post-delete-settle-seconds must not be negative")
    if args.startup_failure_backoff_seconds < 0:
        raise SystemExit("--startup-failure-backoff-seconds must not be negative")
    if args.startup_failure_backoff_max_seconds < args.startup_failure_backoff_seconds:
        raise SystemExit(
            "--startup-failure-backoff-max-seconds must be at least "
            "--startup-failure-backoff-seconds"
        )
    if args.parallel == 2 and args.second_slot_min_free_gb < args.min_free_gb:
        raise SystemExit("--second-slot-min-free-gb must be at least --min-free-gb")
    if args.lifecycle_lock_timeout < 30:
        raise SystemExit("--lifecycle-lock-timeout must be at least 30 seconds")


def terminate_all_workers(context: BatchContext, reason: str) -> None:
    STOP_EVENT.set()
    with ACTIVE_WORKERS_LOCK:
        workers = list(ACTIVE_WORKERS.items())
    for round_number, process in workers:
        terminate_process_group(process, context, f"{reason}; round {round_number}")


def next_round_number(context: BatchContext) -> int:
    numbers = [
        int(item.get("round_number", 0))
        for item in context.rounds
        if str(item.get("round_number", "")).isdigit()
    ]
    numbers.extend(context.active_rounds)
    return max(numbers, default=0) + 1


def lifecycle_ready_for_second_slot(
    context: BatchContext,
    expected_active_count: int,
) -> tuple[bool, str]:
    with context.lock:
        active = list(context.active_rounds.values())
    if len(active) < expected_active_count:
        return False, "等待既有 worker 登記輪次歸屬"
    pending = [
        int(item.get("round_number", 0))
        for item in active
        if not item.get("lifecycle_ready")
    ]
    if pending:
        return False, "等待第 {} 輪完成建立、設定與 ADB 啟動".format(
            ", ".join(str(number) for number in pending)
        )
    return True, "既有 worker 的 ADB 已就緒"


def main() -> int:
    args = parser().parse_args()
    validate_args(args)
    lock_handle = acquire_lock(args.batch_root / "batch.lock")
    # Keep the descriptor referenced for the full function lifetime; closing it
    # would release flock and allow a second supervisor to start.
    _ = lock_handle
    context = BatchContext(args.batch_root, args.output_root)
    STOP_EVENT.clear()
    executor: concurrent.futures.ThreadPoolExecutor | None = None
    try:
        migrate_legacy_active(context)
        if context.active_rounds:
            baseline: dict[str, str] = {}
            for active in context.active_rounds.values():
                baseline.update(
                    {
                        str(key): str(value)
                        for key, value in (active.get("protected_engines") or {}).items()
                    }
                )
            context.protected_engines = baseline
            recover_interrupted_rounds(context, args)
        else:
            context.protected_engines = instances_map()

        context.state["settings"] = {
            "parallel": args.parallel,
            "adaptive_parallel": args.adaptive_parallel,
            "ignore_load_gate": args.ignore_load_gate,
            "engine_cpu": args.engine_cpu,
            "engine_ram_mb": args.engine_ram_mb,
            "start_stagger_seconds": args.start_stagger_seconds,
            "post_delete_settle_seconds": args.post_delete_settle_seconds,
            "startup_failure_backoff_seconds": args.startup_failure_backoff_seconds,
            "startup_failure_backoff_max_seconds": args.startup_failure_backoff_max_seconds,
            "existing_engine_snapshot": dict(context.protected_engines),
        }
        context.state["status"] = "preparing"
        context.persist()
        context.event("prepare", "預先建立 OCR helper，避免 worker 同時編譯")
        aov_runner.ensure_vision_ocr()
        aov_runner.ensure_bluestacks_overlay_helper()

        gate = AdaptiveGate(ignore_load=args.ignore_load_gate)
        executor = concurrent.futures.ThreadPoolExecutor(
            max_workers=args.parallel,
            thread_name_prefix="aov-round",
        )
        futures: dict[concurrent.futures.Future[dict[str, Any] | None], int] = {}
        launched_this_session = 0
        next_round = next_round_number(context)
        last_launch = 0.0
        last_resource_sample = 0.0
        last_scheduler_persist = 0.0
        resources: dict[str, Any] = {}
        last_admission_log = ""
        startup_failure_streak = 0
        startup_backoff_until = 0.0

        while True:
            for future, round_number in list(futures.items()):
                if not future.done():
                    continue
                del futures[future]
                try:
                    record = future.result()
                    if record is not None:
                        context.add_round(record)
                        context.remove_active(round_number)
                        if is_startup_launch_failure(record):
                            startup_failure_streak += 1
                            cooldown = min(
                                args.startup_failure_backoff_max_seconds,
                                args.startup_failure_backoff_seconds
                                * startup_failure_streak,
                            )
                            startup_backoff_until = max(
                                startup_backoff_until,
                                time.monotonic() + cooldown,
                            )
                            context.event(
                                "startup_backoff",
                                (
                                    f"第 {round_number} 輪未進入穩定橫向遊戲畫面；"
                                    f"第二槽冷卻 {cooldown:.0f} 秒後再補"
                                ),
                                round_number=round_number,
                                consecutive_failures=startup_failure_streak,
                                cooldown_seconds=cooldown,
                            )
                        elif record.get("worker_status") == "completed":
                            startup_failure_streak = 0
                except InterruptedError:
                    if not STOP_EVENT.is_set():
                        raise
                except BaseException as exc:
                    context.event(
                        "fatal_safety_error",
                        f"第 {round_number} 輪無法安全復原；停止建立新引擎",
                        round_number=round_number,
                        error=f"{type(exc).__name__}: {exc}",
                        traceback=traceback.format_exc(),
                    )
                    terminate_all_workers(context, "safety recovery failure")
                    raise

            if STOP_EVENT.is_set():
                break
            requested_done = (
                args.max_rounds > 0 and launched_this_session >= args.max_rounds
            )
            if requested_done and not futures:
                break

            now = time.monotonic()
            if now - last_resource_sample >= 15 or not resources:
                resources = system_resources(args.output_root)
                gate.observe(resources, len(futures))
                last_resource_sample = now

            allowed, admission_reason = gate.allow_slot(
                active_count=len(futures),
                resources=resources,
                args=args,
                cleanup_blocked=context.cleanup_blocked.is_set(),
            )
            if requested_done:
                allowed = False
                admission_reason = "已啟動指定輪數，等待執行完成"
            if (
                allowed
                and futures
                and now < startup_backoff_until
            ):
                allowed = False
                admission_reason = (
                    "遊戲啟動失敗冷卻，第二槽暫緩 "
                    f"{startup_backoff_until - now:.0f} 秒"
                )
            if allowed and futures:
                ready, ready_reason = lifecycle_ready_for_second_slot(
                    context,
                    len(futures),
                )
                if not ready:
                    allowed = False
                    admission_reason = ready_reason
            launch_delay = (
                args.start_stagger_seconds if futures else args.between_rounds
            )
            if allowed and last_launch and now - last_launch < launch_delay:
                allowed = False
                admission_reason = (
                    f"錯開啟動，剩餘 {launch_delay - (now - last_launch):.0f} 秒"
                )

            resources["admission_reason"] = admission_reason
            resources["configured_parallel"] = args.parallel
            resources["actual_parallel"] = len(futures)
            resources["degraded"] = gate.degraded
            context.state["resources"] = dict(resources)
            context.state["status"] = (
                "running" if futures else ("waiting_resources" if not allowed else "starting")
            )
            if now - last_scheduler_persist >= 15:
                context.persist()
                last_scheduler_persist = now

            if allowed:
                round_number = next_round
                next_round += 1
                launched_this_session += 1
                last_launch = now
                future = executor.submit(round_job, context, args, round_number)
                futures[future] = round_number
                context.event(
                    "worker_launch",
                    f"已派發第 {round_number} 輪（{len(futures)}/{args.parallel}）",
                    round_number=round_number,
                    active_workers=len(futures),
                )
                last_admission_log = ""
            elif admission_reason != last_admission_log:
                context.event(
                    "admission",
                    admission_reason,
                    active_workers=len(futures),
                )
                last_admission_log = admission_reason
            STOP_EVENT.wait(1.0)

        context.state["status"] = "completed_requested_rounds"
        context.persist()
        executor.shutdown(wait=True, cancel_futures=False)
        return 0
    except KeyboardInterrupt:
        terminate_all_workers(context, "received Ctrl-C")
        context.event("stopped", "收到 Ctrl-C；保留 active_rounds 供下次逐輪清理")
        context.state["status"] = "stopped_by_user"
        context.persist()
        return 130
    except BaseException as exc:
        terminate_all_workers(context, "unexpected supervisor failure")
        context.event(
            "stopped_error",
            "批次主管因安全性錯誤停止；保留 active_rounds 供下次復原",
            error=f"{type(exc).__name__}: {exc}",
            traceback=traceback.format_exc(),
        )
        context.state["status"] = "stopped_error"
        context.persist()
        raise
    finally:
        if executor is not None:
            executor.shutdown(wait=False, cancel_futures=True)


if __name__ == "__main__":
    raise SystemExit(main())
