autool-dispatcher/monitoring.py

import json
import os
import socket
import sqlite3
import threading
import time
from collections import Counter, defaultdict
from contextlib import contextmanager
from datetime import datetime, timedelta
from typing import Any, Dict, Iterable, List, Optional, Tuple
from zoneinfo import ZoneInfo

from config import (
    MANAGED_WORKER_IDS,
    MONITORING_DB_PATH,
    MONITORING_TIMEZONE,
    MONITORING_TIMELINE_BUCKET_MINUTES,
    normalize_worker_id,
)
from log_manager import logger
from result_codes import (
    APP_ERROR_DESC,
    BUSINESS_ERROR_DESC,
    DOWNLOAD_ERROR_DESC,
    INFRA_ERROR_DESC,
    AppError,
    BusinessError,
    DownloadError,
    InfraError,
)


SHANGHAI_TZ = ZoneInfo(MONITORING_TIMEZONE)
RUNNING_STATES = {"running_download", "running_collect"}
ONLINE_STATES = {"idle_waiting_task", "running_download", "running_collect", "disabled"}
DISPLAY_STATE_LABELS = {
    "idle_waiting_task": "idle",
    "running_download": "download",
    "running_collect": "collect",
    "disabled": "disabled",
    "offline": "offline",
}
RUNTIME_STATE_ORDER = [
    "running_download",
    "running_collect",
    "idle_waiting_task",
    "disabled",
    "offline",
]
RUNTIME_STATE_LABELS = {
    "running_download": "下载中",
    "running_collect": "采集中",
    "idle_waiting_task": "空闲等待",
    "disabled": "已禁用",
    "offline": "离线",
}
STAGE_LABELS = {
    "download": "下载",
    "collect": "采集",
}
STAGE_ORDER = ["download", "collect"]
FAILURE_DOMAIN_LABELS = {
    "download_network": "下载网络",
    "download_other": "下载其他",
    "infra": "基础设施",
    "app": "应用问题",
    "business": "业务限制",
    "other": "其他",
}
DOWNLOAD_NETWORK_ERROR_CODES = {
    int(DownloadError.DOWNLOAD_TIMEOUT),
    int(DownloadError.NETWORK_ERROR),
}
DOWNLOAD_NETWORK_MESSAGE_HINTS = (
    "timeout",
    "overtime",
    "jump failed",
    "app page load failed",
    "page load failed",
    "network error",
    "connection error",
    "连接超时",
    "网络错误",
    "超时",
)


def _safe_json_dumps(payload: Dict[str, Any]) -> str:
    return json.dumps(payload, ensure_ascii=False, sort_keys=True)


def _safe_json_loads(payload: Optional[str], fallback: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
    if not payload:
        return fallback or {}
    try:
        value = json.loads(payload)
        return value if isinstance(value, dict) else (fallback or {})
    except (TypeError, ValueError):
        return fallback or {}


def _date_str_from_ts(ts: float) -> str:
    return datetime.fromtimestamp(ts, SHANGHAI_TZ).strftime("%Y-%m-%d")


def _day_bounds(date_str: str, now_ts: Optional[float] = None) -> Tuple[float, float, bool]:
    local_day = datetime.strptime(date_str, "%Y-%m-%d").replace(tzinfo=SHANGHAI_TZ)
    day_start = local_day.timestamp()
    day_end = (local_day + timedelta(days=1)).timestamp()
    now_ts = now_ts or time.time()
    is_today = _date_str_from_ts(now_ts) == date_str
    return day_start, min(day_end, now_ts) if is_today else day_end, is_today


def _interval_overlap(start: float, end: float, bucket_start: float, bucket_end: float) -> bool:
    return max(start, bucket_start) < min(end, bucket_end)


def _build_histogram(values: List[float], edges: List[float]) -> List[Dict[str, Any]]:
    buckets: List[Dict[str, Any]] = []
    previous = 0.0
    for edge in edges:
        buckets.append({"label": _format_bucket_label(previous, edge), "start": previous, "end": edge, "count": 0})
        previous = edge
    buckets.append({"label": _format_bucket_label(previous, None), "start": previous, "end": None, "count": 0})
    for value in values:
        if value is None:
            continue
        placed = False
        for bucket in buckets:
            bucket_end = bucket["end"]
            if bucket_end is None:
                bucket["count"] += 1
                placed = True
                break
            if bucket["start"] <= value < bucket_end:
                bucket["count"] += 1
                placed = True
                break
        if not placed and buckets:
            buckets[-1]["count"] += 1
    return buckets


def _percentile(values: List[float], ratio: float) -> Optional[float]:
    if not values:
        return None
    ordered = sorted(values)
    index = int(round((len(ordered) - 1) * ratio))
    return round(ordered[index], 2)


def _format_minutes(seconds: Optional[float]) -> str:
    seconds = float(seconds or 0.0)
    minutes = seconds / 60.0
    if minutes >= 10:
        return f"{round(minutes):.0f}m"
    if minutes >= 1:
        return f"{minutes:.1f}m"
    if seconds <= 0:
        return "0m"
    return f"{minutes:.1f}m"


def _format_bucket_label(start_seconds: float, end_seconds: Optional[float]) -> str:
    if end_seconds is None:
        return f"{_format_minutes(start_seconds)}+"
    return f"{_format_minutes(start_seconds)}-{_format_minutes(end_seconds)}"


def _humanize_failure_reason(error_type: Optional[str], error_message: Optional[str]) -> str:
    if error_message:
        return str(error_message)
    if not error_type:
        return "未知原因"
    try:
        category_name, code_str = str(error_type).split("/", 1)
        code = int(code_str)
    except (ValueError, TypeError):
        return str(error_type)

    mapping = {
        "INFRA_ERROR": INFRA_ERROR_DESC,
        "DOWNLOAD_ERROR": DOWNLOAD_ERROR_DESC,
        "APP_ERROR": APP_ERROR_DESC,
        "BUSINESS_ERROR": BUSINESS_ERROR_DESC,
    }.get(category_name)
    if not mapping:
        return str(error_type)
    return mapping.get(code, str(error_type))


def _task_row_total_duration(row: sqlite3.Row) -> float:
    return float(row["total_duration_seconds"] or row["failed_total_duration_seconds"] or 0.0)


def _build_failure_task_sample(row: sqlite3.Row) -> Dict[str, Any]:
    task_started_at = float(row["task_started_at"]) if row["task_started_at"] is not None else None
    task_ended_at = float(row["task_ended_at"]) if row["task_ended_at"] is not None else None
    failure_domain = row["failure_domain"]
    failure_subtype = row["failure_subtype"]
    diagnostics = _extract_task_diagnostics(row)
    if (row["status"] or "unknown") != "success" and (not failure_domain or not failure_subtype):
        failure_domain, failure_subtype = _classify_failure(row["error_type"], row["error_message"])
    return {
        "execution_id": row["execution_id"],
        "task_key": row["task_key"],
        "app_name": row["app_name"] or "--",
        "package_name": row["package_name"] or "--",
        "worker_id": row["worker_id"],
        "status": row["status"] or "unknown",
        "failure_domain": failure_domain or "",
        "failure_domain_label": _humanize_failure_domain(failure_domain),
        "failure_subtype": failure_subtype or "",
        "failure_subtype_label": _humanize_failure_subtype(
            failure_domain,
            failure_subtype,
            row["error_type"],
            row["error_message"],
        ),
        "failure_reason": _humanize_failure_reason(row["error_type"], row["error_message"]),
        "error_type": row["error_type"] or "",
        "error_message": row["error_message"] or "",
        "failed_stage": row["failed_stage"] or "--",
        "download_duration_seconds": row["download_duration_seconds"] or 0.0,
        "collect_duration_seconds": row["collect_duration_seconds"] or 0.0,
        "total_duration_seconds": row["total_duration_seconds"] or 0.0,
        "failed_total_duration_seconds": row["failed_total_duration_seconds"] or 0.0,
        "diagnostics": diagnostics,
        "diagnostics_note": _format_diagnostics_note(diagnostics, int(row["retry_count"] or 0)),
        "task_started_at": datetime.fromtimestamp(task_started_at, SHANGHAI_TZ).strftime("%Y-%m-%d %H:%M:%S")
        if task_started_at
        else "--",
        "task_ended_at": datetime.fromtimestamp(task_ended_at, SHANGHAI_TZ).strftime("%Y-%m-%d %H:%M:%S")
        if task_ended_at
        else "--",
    }


def _parse_error_type(error_type: Optional[str]) -> Tuple[Optional[str], Optional[int]]:
    if not error_type:
        return None, None
    try:
        category_name, code_str = str(error_type).split("/", 1)
        return category_name, int(code_str)
    except (TypeError, ValueError):
        return str(error_type), None


def _download_error_message_is_network(message: Optional[str]) -> bool:
    if not message:
        return False
    lowered = str(message).lower()
    return any(token in lowered for token in DOWNLOAD_NETWORK_MESSAGE_HINTS)


def _normalize_download_errors(download_errors: Any) -> Dict[str, Dict[str, Any]]:
    if isinstance(download_errors, dict):
        return download_errors
    return {}


def _all_download_errors_are_network(download_errors: Any) -> bool:
    errors = _normalize_download_errors(download_errors)
    if not errors:
        return False
    for detail in errors.values():
        if not isinstance(detail, dict):
            return False
        code = detail.get("code")
        try:
            if code is not None and int(code) in DOWNLOAD_NETWORK_ERROR_CODES:
                continue
        except (TypeError, ValueError):
            pass
        if _download_error_message_is_network(detail.get("message")):
            continue
        return False
    return True


def _error_code_name(category_name: Optional[str], code: Optional[int]) -> str:
    if category_name == "DOWNLOAD_ERROR" and code is not None:
        try:
            return DownloadError(int(code)).name
        except ValueError:
            return f"DOWNLOAD_{code}"
    if category_name == "INFRA_ERROR" and code is not None:
        try:
            return InfraError(int(code)).name
        except ValueError:
            return f"INFRA_{code}"
    if category_name == "APP_ERROR" and code is not None:
        try:
            return AppError(int(code)).name
        except ValueError:
            return f"APP_{code}"
    if category_name == "BUSINESS_ERROR" and code is not None:
        try:
            return BusinessError(int(code)).name
        except ValueError:
            return f"BUSINESS_{code}"
    if category_name:
        return str(category_name)
    return "OTHER"


def _classify_failure(
    error_type: Optional[str],
    error_message: Optional[str],
    download_errors: Any = None,
) -> Tuple[str, str]:
    category_name, code = _parse_error_type(error_type)
    subtype = _error_code_name(category_name, code)
    if category_name == "DOWNLOAD_ERROR":
        if code in DOWNLOAD_NETWORK_ERROR_CODES:
            return "download_network", subtype
        if code == int(DownloadError.ALL_SOURCES_FAILED) and _all_download_errors_are_network(download_errors):
            return "download_network", "ALL_SOURCES_NETWORK_FAILED"
        if _download_error_message_is_network(error_message):
            return "download_network", subtype
        return "download_other", subtype
    if category_name == "INFRA_ERROR":
        return "infra", subtype
    if category_name == "APP_ERROR":
        return "app", subtype
    if category_name == "BUSINESS_ERROR":
        return "business", subtype
    return "other", subtype


def _humanize_failure_domain(domain: Optional[str]) -> str:
    return FAILURE_DOMAIN_LABELS.get(domain or "", "其他")


def _humanize_failure_subtype(
    domain: Optional[str],
    subtype: Optional[str],
    error_type: Optional[str],
    error_message: Optional[str],
) -> str:
    if domain == "download_network" and subtype == "ALL_SOURCES_NETWORK_FAILED":
        return "所有下载源网络失败"
    reason = _humanize_failure_reason(error_type, None)
    if reason and reason != str(error_type):
        return reason
    if error_message and not subtype:
        return str(error_message)
    if subtype:
        return str(subtype)
    return "未知原因"


def _extract_task_diagnostics(row: sqlite3.Row) -> Dict[str, Any]:
    trace_json = row["trace_json"] if "trace_json" in row.keys() else None
    trace = _safe_json_loads(trace_json)
    diagnostics = trace.get("diagnostics")
    return diagnostics if isinstance(diagnostics, dict) else {}


def _format_diagnostics_note(diagnostics: Dict[str, Any], retry_count: int) -> str:
    labels = []
    if retry_count > 0:
        labels.append(f"重试{retry_count}次")
    if diagnostics.get("emulator_restarted"):
        labels.append("模拟器重启")
    if diagnostics.get("airtest_reinitialized"):
        labels.append("Airtest重连")
    if diagnostics.get("download_recovered"):
        labels.append("下载恢复")
    if diagnostics.get("collect_recovered"):
        labels.append("采集恢复")
    return " / ".join(labels) if labels else "--"


class MonitoringRepository:
    def __init__(self, db_path: str = MONITORING_DB_PATH, initialize_schema: bool = True):
        self.db_path = db_path
        self._write_lock = threading.Lock()
        if initialize_schema:
            self._initialize_schema()

    @contextmanager
    def _connect(self):
        """每次操作使用独立连接，避免跨线程共享连接导致游标状态串扰。"""
        connection = sqlite3.connect(self.db_path, timeout=30)
        connection.row_factory = sqlite3.Row
        connection.execute("PRAGMA busy_timeout=30000")
        connection.execute("PRAGMA synchronous=NORMAL")
        connection.execute("PRAGMA wal_autocheckpoint=1000")
        connection.execute("PRAGMA cache_size=-8192")
        connection.execute("PRAGMA temp_store=MEMORY")
        try:
            yield connection
            connection.commit()
        except Exception:
            try:
                connection.rollback()
            except Exception:
                pass
            raise
        finally:
            connection.close()

    def _table_exists(self, connection: sqlite3.Connection, table_name: str) -> bool:
        row = connection.execute(
            """
            SELECT 1
              FROM sqlite_master
             WHERE type = 'table' AND name = ?
            """,
            (table_name,),
        ).fetchone()
        return bool(row)

    def close(self):
        """保留接口兼容；独立连接模式下无需额外关闭。"""
        return

    def _initialize_schema(self):
        os.makedirs(os.path.dirname(self.db_path) or ".", exist_ok=True)
        with self._connect() as connection:
            connection.execute("PRAGMA journal_mode=WAL")
            connection.executescript(
                """
                CREATE TABLE IF NOT EXISTS controller_session (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    started_at REAL NOT NULL,
                    ended_at REAL,
                    host TEXT,
                    version TEXT,
                    stop_reason TEXT,
                    last_heartbeat_at REAL
                );

                CREATE TABLE IF NOT EXISTS worker_state_event (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    worker_id TEXT NOT NULL,
                    session_id INTEGER,
                    event_time REAL NOT NULL,
                    state TEXT NOT NULL,
                    reason TEXT,
                    task_key TEXT
                );
                CREATE INDEX IF NOT EXISTS idx_worker_state_event_worker_time
                    ON worker_state_event(worker_id, event_time);

                CREATE TABLE IF NOT EXISTS task_stage_event (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    execution_id TEXT NOT NULL,
                    task_key TEXT NOT NULL,
                    worker_id TEXT NOT NULL,
                    session_id INTEGER,
                    event_time REAL NOT NULL,
                    event_type TEXT NOT NULL,
                    stage TEXT,
                    status TEXT,
                    reason TEXT,
                    metrics_json TEXT
                );
                CREATE INDEX IF NOT EXISTS idx_task_stage_event_exec_time
                    ON task_stage_event(execution_id, event_time);

                CREATE TABLE IF NOT EXISTS task_execution (
                    execution_id TEXT PRIMARY KEY,
                    task_key TEXT NOT NULL,
                    app_name TEXT,
                    package_name TEXT,
                    last_updated TEXT,
                    collection_task_type TEXT,
                    worker_id TEXT NOT NULL,
                    session_id INTEGER,
                    stat_date TEXT NOT NULL,
                    task_started_at REAL,
                    task_ended_at REAL,
                    status TEXT,
                    failed_stage TEXT,
                    failure_bucket TEXT,
                    retry_count INTEGER DEFAULT 0,
                    download_duration_seconds REAL,
                    collect_duration_seconds REAL,
                    failed_total_duration_seconds REAL,
                    total_duration_seconds REAL,
                    download_source TEXT,
                    error_type TEXT,
                    error_message TEXT,
                    result_detail TEXT,
                    droidbot_steps INTEGER,
                    guiagent_steps INTEGER,
                    total_steps INTEGER,
                    num_nodes INTEGER,
                    num_reached_activities INTEGER,
                    app_num_total_activities INTEGER,
                    failure_domain TEXT,
                    failure_subtype TEXT,
                    trace_json TEXT,
                    last_updated_at REAL
                );
                CREATE INDEX IF NOT EXISTS idx_task_execution_stat_date
                    ON task_execution(stat_date, worker_id, task_started_at);
                """
            )
            columns = {
                row["name"]
                for row in connection.execute("PRAGMA table_info(task_execution)").fetchall()
            }
            if "failure_bucket" not in columns:
                connection.execute("ALTER TABLE task_execution ADD COLUMN failure_bucket TEXT")
            if "failure_domain" not in columns:
                connection.execute("ALTER TABLE task_execution ADD COLUMN failure_domain TEXT")
            if "failure_subtype" not in columns:
                connection.execute("ALTER TABLE task_execution ADD COLUMN failure_subtype TEXT")
            if "trace_json" not in columns:
                connection.execute("ALTER TABLE task_execution ADD COLUMN trace_json TEXT")
            if "result_detail" not in columns:
                connection.execute("ALTER TABLE task_execution ADD COLUMN result_detail TEXT")
            if "droidbot_steps" not in columns:
                connection.execute("ALTER TABLE task_execution ADD COLUMN droidbot_steps INTEGER")
            if "guiagent_steps" not in columns:
                connection.execute("ALTER TABLE task_execution ADD COLUMN guiagent_steps INTEGER")
            if "total_steps" not in columns:
                connection.execute("ALTER TABLE task_execution ADD COLUMN total_steps INTEGER")
            if "num_nodes" not in columns:
                connection.execute("ALTER TABLE task_execution ADD COLUMN num_nodes INTEGER")
            if "num_reached_activities" not in columns:
                connection.execute("ALTER TABLE task_execution ADD COLUMN num_reached_activities INTEGER")
            if "app_num_total_activities" not in columns:
                connection.execute("ALTER TABLE task_execution ADD COLUMN app_num_total_activities INTEGER")
            if "last_updated" not in columns:
                connection.execute("ALTER TABLE task_execution ADD COLUMN last_updated TEXT")
            if "collection_task_type" not in columns:
                connection.execute("ALTER TABLE task_execution ADD COLUMN collection_task_type TEXT")

            # ── APK 注册表（由 apk_cloud.registry.ApkRegistry 使用）──
            connection.executescript(
                """
                CREATE TABLE IF NOT EXISTS apk_registry (
                    package_name TEXT PRIMARY KEY,
                    download_date TEXT NOT NULL,
                    download_time REAL NOT NULL,
                    local_dir TEXT,
                    smb_dir TEXT,
                    version_name TEXT,
                    source TEXT,
                    apk_files_json TEXT,
                    updated_at REAL NOT NULL
                );

                CREATE TABLE IF NOT EXISTS apk_pending_download (
                    package_name TEXT PRIMARY KEY,
                    created_at REAL NOT NULL
                );

                CREATE TABLE IF NOT EXISTS app_download_record (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    package_name TEXT NOT NULL,
                    country_code TEXT NOT NULL DEFAULT '',
                    app_name TEXT DEFAULT '',
                    last_updated TEXT NOT NULL DEFAULT '',
                    downloads INTEGER DEFAULT 0,

                    worker_play_status TEXT NOT NULL DEFAULT 'pending',
                    worker_play_error TEXT,
                    worker_play_date TEXT,

                    us_play_status TEXT NOT NULL DEFAULT 'pending',
                    us_play_error TEXT,
                    us_play_date TEXT,

                    us_aurora_status TEXT NOT NULL DEFAULT 'pending',
                    us_aurora_error TEXT,
                    us_aurora_date TEXT,

                    overall_status TEXT NOT NULL DEFAULT 'pending',
                    overall_error TEXT,
                    apk_version_name TEXT DEFAULT '',

                    local_apk_dir TEXT DEFAULT '',
                    local_apk_file_count INTEGER DEFAULT 0,
                    local_apk_is_stale INTEGER DEFAULT 1,

                    created_at REAL NOT NULL,
                    updated_at REAL NOT NULL,

                    UNIQUE(package_name, country_code)
                );
                CREATE INDEX IF NOT EXISTS idx_download_record_pkg
                    ON app_download_record(package_name);
                CREATE INDEX IF NOT EXISTS idx_download_record_country
                    ON app_download_record(country_code);
                CREATE INDEX IF NOT EXISTS idx_download_record_overall
                    ON app_download_record(overall_status);
                CREATE INDEX IF NOT EXISTS idx_download_record_stale
                    ON app_download_record(local_apk_is_stale);
                """
            )

    def clear_monitoring_tables(self):
        table_names = [
            "controller_session",
            "worker_state_event",
            "task_stage_event",
            "task_execution",
        ]
        with self._write_lock, self._connect() as connection:
            for table_name in table_names:
                if self._table_exists(connection, table_name):
                    connection.execute(f"DELETE FROM {table_name}")
            if self._table_exists(connection, "sqlite_sequence"):
                for table_name in table_names:
                    connection.execute("DELETE FROM sqlite_sequence WHERE name = ?", (table_name,))

    def create_controller_session(self, host: str, version: str, started_at: float) -> int:
        with self._write_lock, self._connect() as connection:
            cursor = connection.execute(
                """
                INSERT INTO controller_session (
                    started_at, host, version, last_heartbeat_at
                ) VALUES (?, ?, ?, ?)
                """,
                (started_at, host, version, started_at),
            )
            return int(cursor.lastrowid)

    def heartbeat_controller_session(self, session_id: int, heartbeat_at: float):
        with self._write_lock, self._connect() as connection:
            connection.execute(
                """
                UPDATE controller_session
                   SET last_heartbeat_at = ?
                 WHERE id = ?
                """,
                (heartbeat_at, session_id),
            )

    def close_controller_session(self, session_id: int, ended_at: float, stop_reason: str):
        with self._write_lock, self._connect() as connection:
            connection.execute(
                """
                UPDATE controller_session
                   SET ended_at = ?, stop_reason = ?, last_heartbeat_at = COALESCE(last_heartbeat_at, ?)
                 WHERE id = ?
                """,
                (ended_at, stop_reason, ended_at, session_id),
            )

    def repair_open_sessions(self, repaired_at: float):
        with self._write_lock, self._connect() as connection:
            rows = connection.execute(
                """
                SELECT id, last_heartbeat_at
                  FROM controller_session
                 WHERE ended_at IS NULL
                """
            ).fetchall()
            for row in rows:
                ended_at = float(row["last_heartbeat_at"] or repaired_at)
                connection.execute(
                    """
                    UPDATE controller_session
                       SET ended_at = ?, stop_reason = COALESCE(stop_reason, 'recovered_after_restart')
                     WHERE id = ?
                    """,
                    (ended_at, row["id"]),
                )

    def insert_worker_state_event(
        self,
        worker_id: str,
        session_id: Optional[int],
        event_time: float,
        state: str,
        reason: Optional[str],
        task_key: Optional[str],
    ):
        with self._write_lock, self._connect() as connection:
            connection.execute(
                """
                INSERT INTO worker_state_event (
                    worker_id, session_id, event_time, state, reason, task_key
                ) VALUES (?, ?, ?, ?, ?, ?)
                """,
                (worker_id, session_id, event_time, state, reason, task_key),
            )

    def insert_task_stage_event(
        self,
        execution_id: str,
        task_key: str,
        worker_id: str,
        session_id: Optional[int],
        event_time: float,
        event_type: str,
        stage: Optional[str],
        status: Optional[str],
        reason: Optional[str],
        metrics: Optional[Dict[str, Any]],
    ):
        with self._write_lock, self._connect() as connection:
            connection.execute(
                """
                INSERT INTO task_stage_event (
                    execution_id, task_key, worker_id, session_id, event_time,
                    event_type, stage, status, reason, metrics_json
                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                """,
                (
                    execution_id,
                    task_key,
                    worker_id,
                    session_id,
                    event_time,
                    event_type,
                    stage,
                    status,
                    reason,
                    _safe_json_dumps(metrics or {}),
                ),
            )

    def upsert_task_execution(
        self,
        execution_id: str,
        task_key: str,
        worker_id: str,
        session_id: Optional[int],
        stat_date: str,
        task_started_at: float,
        app_name: Optional[str] = None,
        package_name: Optional[str] = None,
        last_updated: Optional[str] = None,
        collection_task_type: Optional[str] = None,
        retry_count: int = 0,
    ):
        with self._write_lock, self._connect() as connection:
            connection.execute(
                """
                INSERT INTO task_execution (
                    execution_id, task_key, app_name, package_name, last_updated, collection_task_type, worker_id, session_id,
                    stat_date, task_started_at, retry_count, last_updated_at
                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                ON CONFLICT(execution_id) DO UPDATE SET
                    app_name = COALESCE(excluded.app_name, task_execution.app_name),
                    package_name = COALESCE(excluded.package_name, task_execution.package_name),
                    last_updated = COALESCE(excluded.last_updated, task_execution.last_updated),
                    collection_task_type = COALESCE(excluded.collection_task_type, task_execution.collection_task_type),
                    retry_count = MAX(task_execution.retry_count, excluded.retry_count),
                    worker_id = excluded.worker_id,
                    session_id = excluded.session_id,
                    last_updated_at = excluded.last_updated_at
                """,
                (
                    execution_id,
                    task_key,
                    app_name,
                    package_name,
                    last_updated,
                    collection_task_type,
                    worker_id,
                    session_id,
                    stat_date,
                    task_started_at,
                    retry_count,
                    time.time(),
                ),
            )

    def update_task_execution_stage(
        self,
        execution_id: str,
        stage: str,
        duration_seconds: Optional[float],
        download_source: Optional[str] = None,
    ):
        if duration_seconds is None:
            return
        with self._write_lock, self._connect() as connection:
            if stage == "download":
                connection.execute(
                    """
                    UPDATE task_execution
                       SET download_duration_seconds = ?,
                           download_source = COALESCE(?, download_source),
                           last_updated_at = ?
                     WHERE execution_id = ?
                    """,
                    (duration_seconds, download_source, time.time(), execution_id),
                )
            elif stage == "collect":
                connection.execute(
                    """
                    UPDATE task_execution
                       SET collect_duration_seconds = ?,
                           last_updated_at = ?
                     WHERE execution_id = ?
                    """,
                    (duration_seconds, time.time(), execution_id),
                )

    def finish_task_execution(
        self,
        execution_id: str,
        task_ended_at: float,
        status: str,
        failed_stage: Optional[str],
        failure_bucket: Optional[str],
        total_duration_seconds: Optional[float],
        failed_total_duration_seconds: Optional[float],
        error_type: Optional[str],
        error_message: Optional[str],
        result_detail: Optional[str],
        droidbot_steps: Optional[int],
        guiagent_steps: Optional[int],
        total_steps: Optional[int],
        num_nodes: Optional[int],
        num_reached_activities: Optional[int],
        app_num_total_activities: Optional[int],
        failure_domain: Optional[str],
        failure_subtype: Optional[str],
        trace_payload: Optional[Dict[str, Any]],
        retry_count: Optional[int] = None,
    ):
        with self._write_lock, self._connect() as connection:
            connection.execute(
                """
                UPDATE task_execution
                   SET task_ended_at = ?,
                       status = ?,
                       failed_stage = ?,
                       failure_bucket = COALESCE(?, failure_bucket),
                       total_duration_seconds = COALESCE(?, total_duration_seconds),
                       failed_total_duration_seconds = COALESCE(?, failed_total_duration_seconds),
                       error_type = COALESCE(?, error_type),
                       error_message = COALESCE(?, error_message),
                       result_detail = COALESCE(?, result_detail),
                       droidbot_steps = COALESCE(?, droidbot_steps),
                       guiagent_steps = COALESCE(?, guiagent_steps),
                       total_steps = COALESCE(?, total_steps),
                       num_nodes = COALESCE(?, num_nodes),
                       num_reached_activities = COALESCE(?, num_reached_activities),
                       app_num_total_activities = COALESCE(?, app_num_total_activities),
                       failure_domain = COALESCE(?, failure_domain),
                       failure_subtype = COALESCE(?, failure_subtype),
                       trace_json = COALESCE(?, trace_json),
                       retry_count = COALESCE(?, retry_count),
                       last_updated_at = ?
                 WHERE execution_id = ?
                """,
                (
                    task_ended_at,
                    status,
                    failed_stage,
                    failure_bucket,
                    total_duration_seconds,
                    failed_total_duration_seconds,
                    error_type,
                    error_message,
                    result_detail,
                    droidbot_steps,
                    guiagent_steps,
                    total_steps,
                    num_nodes,
                    num_reached_activities,
                    app_num_total_activities,
                    failure_domain,
                    failure_subtype,
                    _safe_json_dumps(trace_payload) if trace_payload else None,
                    retry_count,
                    time.time(),
                    execution_id,
                ),
            )

    def list_task_executions(self, stat_date: str, worker_id: Optional[str] = None, limit: int = 200) -> List[sqlite3.Row]:
        with self._connect() as connection:
            if worker_id:
                return connection.execute(
                    """
                    SELECT *
                      FROM task_execution
                     WHERE stat_date = ? AND worker_id = ?
                     ORDER BY COALESCE(task_started_at, last_updated_at, 0) DESC
                     LIMIT ?
                    """,
                    (stat_date, worker_id, limit),
                ).fetchall()
            return connection.execute(
                """
                SELECT *
                  FROM task_execution
                 WHERE stat_date = ?
                 ORDER BY COALESCE(task_started_at, last_updated_at, 0) DESC
                 LIMIT ?
                """,
                (stat_date, limit),
            ).fetchall()

    def list_app_collect_statuses(self, package_names: Iterable[str]) -> Dict[str, Dict[str, str]]:
        """
        查询应用的采集状态分类（新架构）。

        直接从 collection_task 查询最新执行记录，基于 execution_status,
        error_category, error_code, num_nodes 和流量比例动态计算状态。
        """
        normalized = [str(item or "").strip() for item in package_names if str(item or "").strip()]
        if not normalized:
            return {}

        with self._connect() as connection:
            rows = []
            for package_name in normalized:
                row = connection.execute(
                    """
                    SELECT
                        package_name,
                        execution_status,
                        error_category,
                        error_code,
                        num_nodes,
                        CASE WHEN total_traffic_bytes > 0
                             THEN self_traffic_bytes * 100.0 / total_traffic_bytes
                             ELSE 0 END AS self_ratio
                      FROM collection_task
                     WHERE package_name = ?
                     ORDER BY
                          datetime(COALESCE(NULLIF(completed_at, ''), NULLIF(started_at, ''), created_at)) DESC,
                          attempt DESC,
                          batch_tag DESC,
                          run_kind DESC
                     LIMIT 1
                    """,
                    (package_name,),
                ).fetchone()
                if row:
                    rows.append(row)

        result = {}
        # 不可重试的 DOWNLOAD_ERROR 错误码
        NON_RETRYABLE_DOWNLOAD_CODES = {1, 5, 6, 10, 403, 404}

        for row in rows:
            pkg = row["package_name"]
            if not pkg:
                continue

            exec_status = (row["execution_status"] or "").strip()
            error_cat = (row["error_category"] or "").strip()
            error_code = row["error_code"]
            nodes = row["num_nodes"] or 0
            self_ratio = row["self_ratio"] or 0

            # 动态计算 restriction_status
            if exec_status == "success":
                # 成功任务：根据质量指标判断
                if nodes >= 5 and self_ratio >= 30:
                    restriction_status = "success"
                else:
                    restriction_status = "light_restricted"
            else:
                # 失败任务：严重受限
                restriction_status = "severe_restricted"

            # 动态计算 retryability
            if exec_status == "success":
                retryability = "retryable"
            elif error_cat == "DOWNLOAD_ERROR" and error_code in NON_RETRYABLE_DOWNLOAD_CODES:
                retryability = "non_retryable"
            else:
                retryability = "retryable"

            # 动态计算 collection_status（业务层分类）
            if restriction_status == "success":
                collection_status = "qualified"
            else:
                collection_status = "restricted"

            result[pkg] = {
                "collection_status": collection_status,
                "restriction_status": restriction_status,
                "retryability": retryability,
            }

        return result

    def get_task_execution_totals(self, worker_id: Optional[str] = None) -> Dict[str, int]:
        with self._connect() as connection:
            if worker_id:
                row = connection.execute(
                    """
                    SELECT
                        COUNT(*) AS task_total,
                        COALESCE(SUM(CASE WHEN status = 'success' THEN 1 ELSE 0 END), 0) AS task_success,
                        COALESCE(SUM(CASE WHEN status != 'success' THEN 1 ELSE 0 END), 0) AS task_non_success,
                        COALESCE(SUM(CASE WHEN status = 'failed' THEN 1 ELSE 0 END), 0) AS task_failed
                      FROM task_execution
                     WHERE worker_id = ?
                    """,
                    (worker_id,),
                ).fetchone()
            else:
                row = connection.execute(
                    """
                    SELECT
                        COUNT(*) AS task_total,
                        COALESCE(SUM(CASE WHEN status = 'success' THEN 1 ELSE 0 END), 0) AS task_success,
                        COALESCE(SUM(CASE WHEN status != 'success' THEN 1 ELSE 0 END), 0) AS task_non_success,
                        COALESCE(SUM(CASE WHEN status = 'failed' THEN 1 ELSE 0 END), 0) AS task_failed
                      FROM task_execution
                    """
                ).fetchone()
            return {
                "task_total": int(row["task_total"] or 0),
                "task_success": int(row["task_success"] or 0),
                "task_non_success": int(row["task_non_success"] or 0),
                "task_failed": int(row["task_failed"] or 0),
            }

    def get_task_overview_metrics(self, stat_date: str, worker_id: Optional[str] = None) -> Dict[str, Any]:
        with self._connect() as connection:
            where_clause = "WHERE stat_date = ?"
            params: List[Any] = [stat_date]
            if worker_id:
                where_clause += " AND worker_id = ?"
                params.append(worker_id)

            summary_row = connection.execute(
                f"""
                SELECT
                    COUNT(*) AS task_total,
                    COALESCE(SUM(CASE WHEN status = 'success' THEN 1 ELSE 0 END), 0) AS task_success,
                    COALESCE(SUM(CASE WHEN status != 'success' THEN 1 ELSE 0 END), 0) AS task_non_success,
                    COALESCE(SUM(CASE WHEN status = 'failed' THEN 1 ELSE 0 END), 0) AS task_failed,
                    COALESCE(AVG(download_duration_seconds), 0) AS avg_download_duration_seconds,
                    COALESCE(AVG(collect_duration_seconds), 0) AS avg_collect_duration_seconds,
                    COALESCE(SUM(CASE WHEN download_duration_seconds IS NOT NULL THEN 1 ELSE 0 END), 0) AS download_value_count,
                    COALESCE(SUM(CASE WHEN collect_duration_seconds IS NOT NULL THEN 1 ELSE 0 END), 0) AS collect_value_count
                  FROM task_execution
                  {where_clause}
                """,
                params,
            ).fetchone()

            def percentile_for(column: str, count: int, ratio: float) -> float:
                if count <= 0:
                    return 0.0
                offset = int(round((count - 1) * ratio))
                row = connection.execute(
                    f"""
                    SELECT {column} AS value
                      FROM task_execution
                      {where_clause} AND {column} IS NOT NULL
                     ORDER BY {column}
                     LIMIT 1 OFFSET ?
                    """,
                    [*params, offset],
                ).fetchone()
                if not row or row["value"] is None:
                    return 0.0
                return round(float(row["value"]), 2)

            download_count = int(summary_row["download_value_count"] or 0)
            collect_count = int(summary_row["collect_value_count"] or 0)
            return {
                "task_total": int(summary_row["task_total"] or 0),
                "task_success": int(summary_row["task_success"] or 0),
                "task_non_success": int(summary_row["task_non_success"] or 0),
                "task_failed": int(summary_row["task_failed"] or 0),
                "avg_download_duration_seconds": round(float(summary_row["avg_download_duration_seconds"] or 0.0), 2),
                "avg_collect_duration_seconds": round(float(summary_row["avg_collect_duration_seconds"] or 0.0), 2),
                "p50_download_duration_seconds": percentile_for("download_duration_seconds", download_count, 0.5),
                "p90_download_duration_seconds": percentile_for("download_duration_seconds", download_count, 0.9),
                "p50_collect_duration_seconds": percentile_for("collect_duration_seconds", collect_count, 0.5),
                "p90_collect_duration_seconds": percentile_for("collect_duration_seconds", collect_count, 0.9),
            }

    def list_state_events_before(self, day_end: float, worker_ids: Optional[Iterable[str]] = None) -> List[sqlite3.Row]:
        with self._connect() as connection:
            if worker_ids:
                worker_ids = list(worker_ids)
                placeholders = ",".join("?" for _ in worker_ids)
                query = f"""
                    SELECT worker_id, event_time, state, reason, task_key
                      FROM worker_state_event
                     WHERE event_time < ? AND worker_id IN ({placeholders})
                     ORDER BY worker_id, event_time
                """
                return connection.execute(query, [day_end, *worker_ids]).fetchall()
            return connection.execute(
                """
                SELECT worker_id, event_time, state, reason, task_key
                  FROM worker_state_event
                 WHERE event_time < ?
                 ORDER BY worker_id, event_time
                """,
                (day_end,),
            ).fetchall()

    def list_worker_ids_up_to(self, day_end: float) -> List[str]:
        with self._connect() as connection:
            rows = connection.execute(
                """
                SELECT DISTINCT worker_id
                  FROM worker_state_event
                 WHERE event_time < ?
                """,
                (day_end,),
            ).fetchall()
            return [row["worker_id"] for row in rows]

    def list_sessions_overlapping(self, day_start: float, day_end: float) -> List[sqlite3.Row]:
        with self._connect() as connection:
            return connection.execute(
                """
                SELECT *
                  FROM controller_session
                 WHERE started_at < ? AND COALESCE(ended_at, last_heartbeat_at, ?) > ?
                 ORDER BY started_at
                """,
                (day_end, day_end, day_start),
            ).fetchall()

    def update_failure_bucket(self, execution_id: str, failure_bucket: str):
        with self._write_lock, self._connect() as connection:
            connection.execute(
                """
                UPDATE task_execution
                   SET failure_bucket = ?,
                       last_updated_at = ?
                 WHERE execution_id = ?
                """,
                (failure_bucket, time.time(), execution_id),
            )

    def update_failure_bucket_by_task_key(self, task_key: str, failure_bucket: str):
        with self._write_lock, self._connect() as connection:
            connection.execute(
                """
                UPDATE task_execution
                   SET failure_bucket = ?,
                       last_updated_at = ?
                 WHERE execution_id = (
                     SELECT execution_id
                       FROM task_execution
                      WHERE task_key = ?
                      ORDER BY COALESCE(task_ended_at, last_updated_at, task_started_at) DESC
                      LIMIT 1
                 )
                """,
                (failure_bucket, time.time(), task_key),
            )

    def get_latest_execution_id(self, task_key: str) -> Optional[str]:
        with self._connect() as connection:
            row = connection.execute(
                """
                SELECT execution_id
                  FROM task_execution
                 WHERE task_key = ?
                 ORDER BY COALESCE(task_ended_at, last_updated_at, task_started_at) DESC
                 LIMIT 1
                """,
                (task_key,),
            ).fetchone()
            return row["execution_id"] if row else None

    def list_stage_finish_events(self, stat_date: str) -> List[sqlite3.Row]:
        with self._connect() as connection:
            return connection.execute(
                """
                SELECT
                    te.execution_id,
                    te.status,
                    te.failure_bucket,
                    te.total_duration_seconds,
                    te.failed_total_duration_seconds,
                    te.error_type,
                    te.error_message,
                    e.stage,
                    e.metrics_json
                  FROM task_stage_event e
                  JOIN task_execution te ON te.execution_id = e.execution_id
                 WHERE te.stat_date = ?
                   AND e.event_type = 'stage_finished'
                 ORDER BY e.event_time
                """,
                (stat_date,),
            ).fetchall()


class MonitoringService:
    def __init__(self, redis_client, managed_worker_ids: Optional[List[str]] = None):
        self.redis = redis_client
        self.repo = MonitoringRepository()
        self.managed_worker_ids = list(managed_worker_ids or MANAGED_WORKER_IDS or [])
        self._lock = threading.RLock()
        self._session_id: Optional[int] = None
        self._version = "unknown"
        self.repo.repair_open_sessions(time.time())

    def start_controller_session(self, version: str = "unknown") -> int:
        with self._lock:
            if self._session_id is not None:
                return self._session_id
            started_at = time.time()
            self._version = version
            self._session_id = self.repo.create_controller_session(
                host=socket.gethostname(),
                version=version,
                started_at=started_at,
            )
            self.redis.set("controller:current_session", str(self._session_id))
            return self._session_id

    def heartbeat_controller_session(self):
        with self._lock:
            session_id = self.current_session_id
            if session_id is None:
                return
            self.repo.heartbeat_controller_session(session_id, time.time())

    def close_controller_session(self, stop_reason: str = "stopped"):
        with self._lock:
            session_id = self._session_id
            if session_id is None:
                return
            self.repo.close_controller_session(session_id, time.time(), stop_reason)
            self.redis.delete("controller:current_session")
            self._session_id = None

    def _classify_rows_by_collection_quality(
        self,
        rows: List[sqlite3.Row],
    ) -> Tuple[List[sqlite3.Row], List[sqlite3.Row], Dict[str, int]]:
        package_map = self.repo.list_app_collect_statuses(
            row["package_name"] for row in rows if row["package_name"]
        )
        qualified_items: List[sqlite3.Row] = []
        severe_items: List[sqlite3.Row] = []
        counters = {
            "qualified_task_count": 0,
            "qualified_success_task_count": 0,
            "qualified_light_task_count": 0,
            "non_success_task_count": 0,
            "non_retryable_task_count": 0,
        }
        for row in rows:
            package_name = str(row["package_name"] or "").strip()
            row_state = package_map.get(package_name, {})
            restriction_status = row_state.get("restriction_status", "")
            retryability = row_state.get("retryability", "")
            if restriction_status == "severe_restricted":
                severe_items.append(row)
                counters["non_success_task_count"] += 1
                if retryability == "non_retryable":
                    counters["non_retryable_task_count"] += 1
            else:
                qualified_items.append(row)
                counters["qualified_task_count"] += 1
                if restriction_status == "light_restricted":
                    counters["qualified_light_task_count"] += 1
                else:
                    counters["qualified_success_task_count"] += 1
        return qualified_items, severe_items, counters

    def _split_rows_by_quality(self, rows: List[sqlite3.Row]) -> Tuple[List[sqlite3.Row], List[sqlite3.Row]]:
        qualified_items, severe_items, _ = self._classify_rows_by_collection_quality(rows)
        return qualified_items, severe_items

    @property
    def current_session_id(self) -> Optional[int]:
        if self._session_id is not None:
            return self._session_id
        raw = self.redis.get("controller:current_session")
        if raw:
            try:
                self._session_id = int(raw)
            except (TypeError, ValueError):
                self._session_id = None
        return self._session_id

    def _execution_key(self, task_key: str) -> str:
        return f"monitor:execution:{task_key}"

    def _current_state_key(self, worker_id: str) -> str:
        return f"worker:current_state:{worker_id}"

    def _get_current_state(self, worker_id: str) -> Dict[str, Any]:
        return _safe_json_loads(self.redis.get(self._current_state_key(worker_id)))

    def _set_current_state(self, worker_id: str, payload: Dict[str, Any]):
        self.redis.set(self._current_state_key(worker_id), _safe_json_dumps(payload))

    def _delete_monitoring_redis_state(self):
        self.redis.delete("controller:current_session")
        patterns = [
            "worker:current_state:*",
            "worker:last_event:*",
            "monitor:execution:*",
        ]
        keys_to_delete: List[str] = []
        for pattern in patterns:
            keys_to_delete.extend(list(self.redis.scan_iter(match=pattern)))
        if keys_to_delete:
            self.redis.delete(*keys_to_delete)

    def _normalize_reseed_state(self, worker: Dict[str, Any]) -> Optional[str]:
        if not worker.get("online", True):
            return None
        state = str(worker.get("state") or "").strip()
        if state in {"idle_waiting_task", "running_download", "running_collect", "disabled"}:
            return state
        if worker.get("status") == "busy":
            return "running_collect"
        if worker.get("status") == "idle":
            return "idle_waiting_task"
        return "idle_waiting_task"

    def reset_monitoring_data(self, worker_snapshots: Optional[List[Dict[str, Any]]] = None) -> Dict[str, Any]:
        with self._lock:
            reset_time = time.time()
            self.close_controller_session(stop_reason="dashboard_reset")
            self.repo.clear_monitoring_tables()
            self._delete_monitoring_redis_state()
            self._session_id = None
            session_id = self.start_controller_session(version=self._version)

            reseeded_workers = 0
            for worker in worker_snapshots or []:
                worker_id = worker.get("worker_id")
                if not worker_id:
                    continue
                state = self._normalize_reseed_state(worker)
                if not state:
                    continue
                self.set_worker_state(
                    worker_id,
                    state,
                    reason="dashboard_reset_reseed",
                    task_key=worker.get("current_task"),
                    event_time=reset_time,
                )
                current_task = worker.get("current_task")
                if current_task:
                    task_metadata = self._resolve_task_metadata(current_task)
                    retry_count = 0
                    status_json = self.redis.hget("task:status", current_task)
                    if status_json:
                        status_payload = _safe_json_loads(status_json)
                        retry_count = int(status_payload.get("retry_count") or 0)
                    execution_id = f"{current_task}:{int(reset_time * 1000)}"
                    self.repo.upsert_task_execution(
                        execution_id=execution_id,
                        task_key=current_task,
                        worker_id=worker_id,
                        session_id=self.current_session_id,
                        stat_date=_date_str_from_ts(reset_time),
                        task_started_at=reset_time,
                        app_name=task_metadata.get("app_name"),
                        package_name=task_metadata.get("package_name"),
                        last_updated=task_metadata.get("last_updated"),
                        collection_task_type=task_metadata.get("collection_task_type"),
                        retry_count=retry_count,
                    )
                    self.redis.set(self._execution_key(current_task), execution_id)
                reseeded_workers += 1

            return {
                "ok": True,
                "reset_at": datetime.fromtimestamp(reset_time, SHANGHAI_TZ).strftime("%Y-%m-%d %H:%M:%S"),
                "session_id": session_id,
                "reseeded_workers": reseeded_workers,
            }

    def _update_worker_info_state(
        self,
        worker_id: str,
        state: str,
        task_key: Optional[str],
        reason: Optional[str],
        event_time: float,
    ):
        info_json = self.redis.hget("workers:info", worker_id)
        if not info_json:
            return
        info = _safe_json_loads(info_json)
        info["state"] = state
        info["state_since"] = event_time
        info["pause_reason"] = reason
        if task_key is not None:
            info["current_task"] = task_key
        self.redis.hset("workers:info", worker_id, _safe_json_dumps(info))

    def set_worker_state(
        self,
        worker_id: str,
        state: str,
        reason: Optional[str] = None,
        task_key: Optional[str] = None,
        event_time: Optional[float] = None,
    ):
        with self._lock:
            event_time = event_time or time.time()
            current = self._get_current_state(worker_id)
            if (
                current.get("state") == state
                and current.get("reason") == reason
                and current.get("task_key") == task_key
            ):
                return
            payload = {"state": state, "reason": reason, "task_key": task_key, "since": event_time}
            self._set_current_state(worker_id, payload)
            self._update_worker_info_state(worker_id, state, task_key, reason, event_time)
            self.repo.insert_worker_state_event(
                worker_id=worker_id,
                session_id=self.current_session_id,
                event_time=event_time,
                state=state,
                reason=reason,
                task_key=task_key,
            )

    def register_worker(self, worker_id: str, task: Optional[Dict[str, Any]] = None, event_time: Optional[float] = None):
        with self._lock:
            event_time = event_time or time.time()
            if task and task.get("task_key"):
                self._update_worker_info_state(worker_id, "idle_waiting_task", task.get("task_key"), None, event_time)
            else:
                self.set_worker_state(worker_id, "idle_waiting_task", reason="waiting_task", event_time=event_time)

    def handle_task_assigned(self, worker_id: str, task_key: str, event_time: Optional[float] = None):
        with self._lock:
            event_time = event_time or time.time()
            info_json = self.redis.hget("workers:info", worker_id)
            if not info_json:
                return
            info = _safe_json_loads(info_json)
            info["current_task"] = task_key
            info["last_update"] = event_time
            self.redis.hset("workers:info", worker_id, _safe_json_dumps(info))
            # 任务分发后将 worker state 标记为 running（等 stage_started 事件进一步细化为 download/collect）
            self.set_worker_state(worker_id, "running_download", task_key=task_key, event_time=event_time)
            if not self.redis.get(self._execution_key(task_key)):
                status_payload = _safe_json_loads(self.redis.hget("task:status", task_key))
                retry_count = int(status_payload.get("retry_count") or 0)
                task_metadata = self._resolve_task_metadata(task_key)
                execution_id = f"{task_key}:{int(event_time * 1000)}"
                self.repo.upsert_task_execution(
                    execution_id=execution_id,
                    task_key=task_key,
                    worker_id=worker_id,
                    session_id=self.current_session_id,
                    stat_date=_date_str_from_ts(event_time),
                    task_started_at=event_time,
                    app_name=task_metadata.get("app_name"),
                    package_name=task_metadata.get("package_name"),
                    last_updated=task_metadata.get("last_updated"),
                    collection_task_type=task_metadata.get("collection_task_type"),
                    retry_count=retry_count,
                )
                self.redis.set(self._execution_key(task_key), execution_id)

    def _resolve_task_details(self, task_key: str) -> Tuple[Optional[str], Optional[str]]:
        details = self._resolve_task_metadata(task_key)
        return details.get("app_name"), details.get("package_name")

    def _resolve_task_metadata(self, task_key: str) -> Dict[str, Any]:
        raw = self.redis.hget("task:details", task_key)
        if not raw:
            return {}
        details = _safe_json_loads(raw)
        return details if isinstance(details, dict) else {}

    def _ensure_execution_id(
        self,
        worker_id: str,
        task_key: str,
        event_time: float,
        retry_count: int = 0,
    ) -> str:
        with self._lock:
            execution_id = self.redis.get(self._execution_key(task_key))
            if execution_id:
                return execution_id
            execution_id = f"{task_key}:{int(event_time * 1000)}"
            task_metadata = self._resolve_task_metadata(task_key)
            self.repo.upsert_task_execution(
                execution_id=execution_id,
                task_key=task_key,
                worker_id=worker_id,
                session_id=self.current_session_id,
                stat_date=_date_str_from_ts(event_time),
                task_started_at=event_time,
                app_name=task_metadata.get("app_name"),
                package_name=task_metadata.get("package_name"),
                last_updated=task_metadata.get("last_updated"),
                collection_task_type=task_metadata.get("collection_task_type"),
                retry_count=retry_count,
            )
            self.redis.set(self._execution_key(task_key), execution_id)
            return execution_id

    def _resolve_report_execution_id(
        self,
        worker_id: str,
        task_key: str,
        event_time: float,
        retry_count: int = 0,
    ) -> str:
        execution_id = self.redis.get(self._execution_key(task_key))
        if execution_id:
            return execution_id
        execution_id = self.repo.get_latest_execution_id(task_key)
        if execution_id:
            return execution_id
        return self._ensure_execution_id(worker_id, task_key, event_time, retry_count=retry_count)

    def _resolve_failure_fields(
        self,
        error_type: Optional[str],
        error_message: Optional[str],
        download_errors: Any = None,
        trace_payload: Optional[Dict[str, Any]] = None,
    ) -> Tuple[Optional[str], Optional[str], Optional[Dict[str, Any]]]:
        trace_payload = dict(trace_payload) if isinstance(trace_payload, dict) else {}
        normalized_download_errors = _normalize_download_errors(download_errors)
        if normalized_download_errors:
            trace_payload["download_errors"] = normalized_download_errors
        failure_domain = trace_payload.get("failure_domain")
        failure_subtype = trace_payload.get("failure_subtype")
        has_failure = bool(error_type or error_message or normalized_download_errors)
        if has_failure and (not failure_domain or not failure_subtype):
            failure_domain, failure_subtype = _classify_failure(
                error_type,
                error_message,
                download_errors=normalized_download_errors,
            )
        elif not has_failure:
            failure_domain = None
            failure_subtype = None
        if not trace_payload:
            trace_payload = None
        elif has_failure and not trace_payload.get("failure_domain"):
            trace_payload["failure_domain"] = failure_domain
            trace_payload["failure_subtype"] = failure_subtype
        return failure_domain, failure_subtype, trace_payload

    def handle_worker_event(self, payload: Dict[str, Any]):
        with self._lock:
            worker_id = normalize_worker_id(payload.get("worker_id"))
            task_key = payload.get("task_key")
            event_type = payload.get("event_type")
            if not worker_id or not event_type:
                return
            event_time = float(payload.get("event_time") or time.time())
            stage = payload.get("stage")
            state = payload.get("state")
            status = payload.get("status")
            reason = payload.get("reason")
            metrics = payload.get("metrics") or {}

            if event_type == "worker_state_changed" and state:
                self.set_worker_state(worker_id, state, reason=reason, task_key=task_key, event_time=event_time)
                return

            if not task_key:
                return

            retry_count = int(metrics.get("retry_count") or 0)
            execution_id = self._ensure_execution_id(worker_id, task_key, event_time, retry_count=retry_count)
            self.repo.insert_task_stage_event(
                execution_id=execution_id,
                task_key=task_key,
                worker_id=worker_id,
                session_id=self.current_session_id,
                event_time=event_time,
                event_type=event_type,
                stage=stage,
                status=status,
                reason=reason,
                metrics=metrics,
            )

            if event_type == "task_started":
                task_metadata = self._resolve_task_metadata(task_key)
                self.repo.upsert_task_execution(
                    execution_id=execution_id,
                    task_key=task_key,
                    worker_id=worker_id,
                    session_id=self.current_session_id,
                    stat_date=_date_str_from_ts(event_time),
                    task_started_at=event_time,
                    app_name=task_metadata.get("app_name"),
                    package_name=task_metadata.get("package_name"),
                    last_updated=task_metadata.get("last_updated"),
                    collection_task_type=task_metadata.get("collection_task_type"),
                    retry_count=retry_count,
                )
                self.handle_task_assigned(worker_id, task_key, event_time)
                return

            if event_type == "stage_started":
                if stage == "download":
                    self.set_worker_state(worker_id, "running_download", task_key=task_key, event_time=event_time)
                elif stage == "collect":
                    self.set_worker_state(worker_id, "running_collect", task_key=task_key, event_time=event_time)
                return

            if event_type == "stage_finished":
                self.repo.update_task_execution_stage(
                    execution_id=execution_id,
                    stage=stage or "",
                    duration_seconds=metrics.get("duration_seconds"),
                    download_source=metrics.get("download_source"),
                )
                return

            if event_type == "task_finished":
                error_type = metrics.get("error_type")
                error_message = metrics.get("error_message") or reason
                total_duration = metrics.get("total_duration_seconds")
                failed_total_duration = metrics.get("failed_total_duration_seconds")
                failed_stage = metrics.get("failed_stage")
                failure_domain, failure_subtype, trace_payload = self._resolve_failure_fields(
                    error_type,
                    error_message,
                    download_errors=metrics.get("download_errors"),
                    trace_payload=metrics.get("trace"),
                )
                self.repo.finish_task_execution(
                    execution_id=execution_id,
                    task_ended_at=event_time,
                    status=status or "unknown",
                    failed_stage=failed_stage,
                    failure_bucket="failed" if (status or "unknown") == "failed" else status,
                    total_duration_seconds=total_duration,
                    failed_total_duration_seconds=failed_total_duration,
                    error_type=error_type,
                    error_message=error_message,
                    result_detail=error_message,
                    droidbot_steps=None,
                    guiagent_steps=None,
                    total_steps=None,
                    num_nodes=None,
                    num_reached_activities=None,
                    app_num_total_activities=None,
                    failure_domain=failure_domain,
                    failure_subtype=failure_subtype,
                    trace_payload=trace_payload,
                    retry_count=retry_count,
                )
                self.redis.delete(self._execution_key(task_key))
                self.set_worker_state(worker_id, "idle_waiting_task", reason="waiting_task", event_time=event_time)

    def handle_worker_report(
        self,
        worker_id: str,
        previous_task_key: str,
        report_data: Dict[str, Any],
        current_time: Optional[float] = None,
        retry_count: int = 0,
        resolved_message: Optional[str] = None,
    ):
        worker_id = normalize_worker_id(worker_id)
        with self._lock:
            if not previous_task_key:
                return
            current_time = current_time or time.time()
            execution_id = self._resolve_report_execution_id(
                worker_id,
                previous_task_key,
                current_time,
                retry_count=retry_count,
            )
            statistics = report_data.get("statistics") or {}
            error_data = report_data.get("error") or {}
            trace_payload = report_data.get("trace")
            status = report_data.get("status", "unknown")
            current_state = self._get_current_state(worker_id).get("state")
            failed_stage = "download" if not statistics.get("download_source") else "collect"
            if status == "success":
                failed_stage = None
            elif current_state == "running_download":
                failed_stage = "download"
            elif current_state == "running_collect":
                failed_stage = "collect"
            error_type = (
                f"{error_data.get('category')}/{error_data.get('code')}"
                if error_data.get("category") is not None and error_data.get("code") is not None
                else None
            )
            error_message = resolved_message or error_data.get("reason") or statistics.get("error_reason")
            failure_domain, failure_subtype, trace_payload = self._resolve_failure_fields(
                error_type,
                error_message,
                download_errors=report_data.get("download_errors"),
                trace_payload=trace_payload,
            )
            self.repo.finish_task_execution(
                execution_id=execution_id,
                task_ended_at=current_time,
                status=status,
                failed_stage=failed_stage,
                failure_bucket=status if status != "failed" else "failed",
                total_duration_seconds=statistics.get("duration_seconds"),
                failed_total_duration_seconds=statistics.get("duration_seconds") if status != "success" else None,
                error_type=error_type,
                error_message=error_message,
                result_detail=resolved_message or error_message,
                droidbot_steps=statistics.get("droidbot_steps"),
                guiagent_steps=statistics.get("guiagent_steps"),
                total_steps=statistics.get("total_steps"),
                num_nodes=statistics.get("num_nodes"),
                num_reached_activities=statistics.get("num_reached_activities"),
                app_num_total_activities=statistics.get("app_num_total_activities"),
                failure_domain=failure_domain,
                failure_subtype=failure_subtype,
                trace_payload=trace_payload,
                retry_count=retry_count,
            )
            self.redis.delete(self._execution_key(previous_task_key))

    def mark_worker_disabled(self, worker_id: str, reason: str):
        self.set_worker_state(worker_id, "disabled", reason=reason, event_time=time.time())

    def mark_worker_offline(self, worker_id: str, reason: str):
        self.set_worker_state(worker_id, "offline", reason=reason, event_time=time.time())

    def set_failure_bucket(self, task_key: str, failure_bucket: str):
        with self._lock:
            execution_id = self.redis.get(self._execution_key(task_key))
            if execution_id:
                self.repo.update_failure_bucket(execution_id, failure_bucket)
                return
            self.repo.update_failure_bucket_by_task_key(task_key, failure_bucket)

    def _controller_up_seconds(self, date_str: str, now_ts: Optional[float] = None) -> float:
        day_start, day_end, _ = _day_bounds(date_str, now_ts)
        total = 0.0
        for row in self.repo.list_sessions_overlapping(day_start, day_end):
            start = max(float(row["started_at"]), day_start)
            end = min(float(row["ended_at"] or row["last_heartbeat_at"] or day_end), day_end)
            total += max(0.0, end - start)
        return round(total, 2)

    def _resolve_worker_ids(
        self,
        date_str: str,
        dispatcher=None,
        worker_id: Optional[str] = None,
    ) -> List[str]:
        if worker_id:
            return [worker_id]
        _, day_end, is_today = _day_bounds(date_str)
        # 以 managed_worker_ids（inventory.json）为准
        worker_ids = set(self.managed_worker_ids)
        if not is_today:
            # 仅在查看历史日期时合并 SQLite 中的 worker（兼容查看旧数据）
            worker_ids.update(self.repo.list_worker_ids_up_to(day_end))
        if dispatcher is not None and is_today:
            for worker in dispatcher.get_dashboard_workers():
                worker_ids.add(worker.get("worker_id"))
        worker_ids.discard(None)
        worker_ids.discard("")
        return sorted(worker_ids)

    def _build_worker_options(self, date_str: str, dispatcher=None) -> List[Dict[str, Any]]:
        worker_ids = self._resolve_worker_ids(date_str, dispatcher=dispatcher)
        current_workers: Dict[str, Dict[str, Any]] = {}
        if dispatcher is not None:
            current_workers = {worker.get("worker_id"): worker for worker in dispatcher.get_dashboard_workers()}

        options: List[Dict[str, Any]] = []
        for worker_id in worker_ids:
            current = current_workers.get(worker_id, {})
            ip_address = current.get("ip_address") or worker_id.split("_", 1)[0]
            hostname = current.get("hostname") or ""
            label = f"{ip_address} · {hostname}" if hostname else ip_address
            options.append(
                {
                    "worker_id": worker_id,
                    "ip_address": ip_address,
                    "hostname": hostname,
                    "label": label,
                }
            )
        return options

    def _build_worker_intervals(
        self,
        date_str: str,
        dispatcher=None,
        worker_id: Optional[str] = None,
    ) -> Dict[str, List[Tuple[float, float, str]]]:
        day_start, day_end, _ = _day_bounds(date_str)
        worker_ids = self._resolve_worker_ids(date_str, dispatcher=dispatcher, worker_id=worker_id)
        rows = self.repo.list_state_events_before(day_end, worker_ids=worker_ids)
        grouped: Dict[str, List[sqlite3.Row]] = defaultdict(list)
        for row in rows:
            grouped[row["worker_id"]].append(row)

        intervals: Dict[str, List[Tuple[float, float, str]]] = {}
        for worker_id in worker_ids:
            events = grouped.get(worker_id, [])
            prev_state = "offline"
            prev_time = day_start
            worker_intervals: List[Tuple[float, float, str]] = []
            for row in events:
                event_time = float(row["event_time"])
                if event_time < day_start:
                    prev_state = row["state"]
                    continue
                if event_time > day_end:
                    break
                if event_time > prev_time:
                    worker_intervals.append((prev_time, event_time, prev_state))
                prev_state = row["state"]
                prev_time = event_time
            if day_end > prev_time:
                worker_intervals.append((prev_time, day_end, prev_state))
            intervals[worker_id] = worker_intervals
        return intervals

    def _summarize_worker_seconds(
        self,
        date_str: str,
        dispatcher=None,
        worker_id: Optional[str] = None,
    ) -> Dict[str, Counter]:
        intervals = self._build_worker_intervals(date_str, dispatcher=dispatcher, worker_id=worker_id)
        summary: Dict[str, Counter] = {}
        for worker_id, worker_intervals in intervals.items():
            counts = Counter()
            for start, end, state in worker_intervals:
                counts[state] += max(0.0, end - start)
            summary[worker_id] = counts
        return summary

    def _runtime_state_breakdown(
        self,
        date_str: str,
        dispatcher=None,
        worker_id: Optional[str] = None,
    ) -> List[Dict[str, Any]]:
        totals = Counter()
        for counts in self._summarize_worker_seconds(date_str, dispatcher=dispatcher, worker_id=worker_id).values():
            for state in RUNTIME_STATE_ORDER:
                totals[state] += float(counts.get(state, 0.0))
        total_seconds = sum(totals.values())
        return [
            {
                "state": state,
                "label": RUNTIME_STATE_LABELS[state],
                "duration_seconds": round(float(totals[state]), 2),
                "share_percent": round((float(totals[state]) / total_seconds) * 100, 2) if total_seconds > 0 else 0.0,
            }
            for state in RUNTIME_STATE_ORDER
        ]

    def get_overview(
        self,
        date_str: Optional[str] = None,
        dispatcher=None,
        worker_id: Optional[str] = None,
    ) -> Dict[str, Any]:
        now_ts = time.time()
        date_str = date_str or _date_str_from_ts(now_ts)
        day_start, day_end, is_today = _day_bounds(date_str, now_ts=now_ts)
        up_seconds = self._controller_up_seconds(date_str, now_ts=now_ts)
        down_seconds = round(max(0.0, day_end - day_start - up_seconds), 2)
        metrics = self.repo.get_task_overview_metrics(date_str, worker_id=worker_id)
        overall_totals = self.repo.get_task_execution_totals(worker_id=worker_id)
        overall_success_total = overall_totals["task_success"]
        # The global dashboard should reflect the dispatcher's in-memory success set,
        # including tasks preloaded from the current active catalog queue.
        if dispatcher is not None and not worker_id:
            overall_success_total = int(dispatcher.get_statistics().get("completed") or 0)

        state_counts = {
            "online_workers": 0,
            "running_workers": 0,
            "running_download_workers": 0,
            "running_collect_workers": 0,
            "idle_workers": 0,
            "offline_workers": 0,
            "disabled_workers": 0,
            "pending_tasks": 0,
        }
        if is_today and dispatcher is not None:
            dashboard_workers = dispatcher.get_dashboard_workers()
            if worker_id:
                dashboard_workers = [worker for worker in dashboard_workers if worker.get("worker_id") == worker_id]
            state_counts["pending_tasks"] = 0 if worker_id else dispatcher.get_statistics().get("pending", 0)
            for worker in dashboard_workers:
                current_worker_id = worker.get("worker_id", "")
                state = worker.get("state") or self._get_current_state(current_worker_id).get("state") or "offline"
                if not worker.get("online"):
                    state = "offline"
                if state in ONLINE_STATES:
                    state_counts["online_workers"] += 1
                if state in RUNNING_STATES:
                    state_counts["running_workers"] += 1
                if state == "running_download":
                    state_counts["running_download_workers"] += 1
                elif state == "running_collect":
                    state_counts["running_collect_workers"] += 1
                elif state == "idle_waiting_task":
                    state_counts["idle_workers"] += 1
                elif state == "disabled":
                    state_counts["disabled_workers"] += 1
                else:
                    state_counts["offline_workers"] += 1
        else:
            timeline = self.get_timeline(date_str=date_str, dispatcher=dispatcher, worker_id=worker_id)
            last = timeline["points"][-1] if timeline["points"] else {}
            state_counts = {
                "online_workers": last.get("online_workers", 0),
                "running_workers": last.get("running_workers", 0),
                "running_download_workers": last.get("running_download_workers", 0),
                "running_collect_workers": last.get("running_collect_workers", 0),
                "idle_workers": last.get("idle_workers", 0),
                "offline_workers": last.get("offline_workers", 0),
                "disabled_workers": last.get("disabled_workers", 0),
            }

        return {
            "date": date_str,
            "is_today": is_today,
            "controller_status": "running" if (is_today and self.current_session_id) else "stopped",
            "controller_up_seconds": up_seconds,
            "controller_down_seconds": down_seconds,
            "task_total": metrics["task_total"],
            "task_success": metrics["task_success"],
            "task_non_success": metrics["task_non_success"],
            "task_failed": metrics["task_failed"],
            "task_total_all": overall_totals["task_total"],
            "task_success_all": overall_success_total,
            "task_non_success_all": overall_totals["task_non_success"],
            "task_failed_all": overall_totals["task_failed"],
            "avg_download_duration_seconds": metrics["avg_download_duration_seconds"],
            "avg_collect_duration_seconds": metrics["avg_collect_duration_seconds"],
            "p50_download_duration_seconds": metrics["p50_download_duration_seconds"],
            "p90_download_duration_seconds": metrics["p90_download_duration_seconds"],
            "p50_collect_duration_seconds": metrics["p50_collect_duration_seconds"],
            "p90_collect_duration_seconds": metrics["p90_collect_duration_seconds"],
            **state_counts,
            "selected_worker_id": worker_id or "",
            "worker_options": self._build_worker_options(date_str, dispatcher=dispatcher),
            "generated_at": datetime.fromtimestamp(now_ts, SHANGHAI_TZ).strftime("%Y-%m-%d %H:%M:%S"),
        }

    def get_distributions(
        self,
        date_str: Optional[str] = None,
        worker_id: Optional[str] = None,
        dispatcher=None,
    ) -> Dict[str, Any]:
        date_str = date_str or _date_str_from_ts(time.time())
        task_rows = self.repo.list_task_executions(date_str, worker_id=worker_id, limit=100000)
        _, severe_items = self._split_rows_by_quality(task_rows)
        download_values = [float(row["download_duration_seconds"]) for row in task_rows if row["download_duration_seconds"] is not None]
        collect_values = [float(row["collect_duration_seconds"]) for row in task_rows if row["collect_duration_seconds"] is not None]
        total_values = [float(row["total_duration_seconds"]) for row in task_rows if row["total_duration_seconds"] is not None]
        severe_values = [
            float(row["failed_total_duration_seconds"] or row["total_duration_seconds"] or 0.0)
            for row in severe_items
            if (row["failed_total_duration_seconds"] is not None or row["total_duration_seconds"] is not None)
        ]
        bucket_edges = [60, 180, 300, 600, 1200, 1800, 3600]
        return {
            "date": date_str,
            "download_duration": _build_histogram(download_values, bucket_edges),
            "collect_duration": _build_histogram(collect_values, bucket_edges),
            "total_duration": _build_histogram(total_values, bucket_edges),
            "non_success_total_duration": _build_histogram(severe_values, bucket_edges),
            **self.get_stage_analysis(date_str, worker_id=worker_id, dispatcher=dispatcher, rows=task_rows),
        }

    def get_stage_analysis(
        self,
        date_str: Optional[str] = None,
        worker_id: Optional[str] = None,
        dispatcher=None,
        rows: Optional[List[sqlite3.Row]] = None,
    ) -> Dict[str, Any]:
        date_str = date_str or _date_str_from_ts(time.time())
        rows = rows if rows is not None else self.repo.list_task_executions(date_str, worker_id=worker_id, limit=100000)
        worker_options = self._build_worker_options(date_str, dispatcher=dispatcher)
        worker_label_map = {
            item["worker_id"]: item.get("label") or item["worker_id"]
            for item in worker_options
        }
        qualified_items, severe_items, collection_quality = self._classify_rows_by_collection_quality(rows)

        def row_total(row: sqlite3.Row) -> float:
            return _task_row_total_duration(row)

        def summarize(items: List[sqlite3.Row]) -> List[Dict[str, Any]]:
            if not items:
                return []
            total_sum = sum(max(row_total(item), 0.0) for item in items)
            result = []
            for stage in STAGE_ORDER:
                key = f"{stage}_duration_seconds"
                stage_sum = sum(float(item[key] or 0.0) for item in items)
                result.append(
                    {
                        "stage": stage,
                        "label": STAGE_LABELS.get(stage, stage),
                        "avg_duration_seconds": round(stage_sum / len(items), 2),
                        "share_percent": round((stage_sum / total_sum) * 100, 2) if total_sum > 0 else 0.0,
                        "total_duration_seconds": round(stage_sum, 2),
                    }
                )
            return result

        qualified_total = round(sum(row_total(item) for item in qualified_items), 2)
        severe_total = round(sum(row_total(item) for item in severe_items), 2)
        total_runtime = qualified_total + severe_total
        time_composition = [
            {
                "bucket": "qualified",
                "label": "合格",
                "duration_seconds": qualified_total,
                "share_percent": round((qualified_total / total_runtime) * 100, 2) if total_runtime > 0 else 0.0,
            },
            {
                "bucket": "severe_restricted",
                "label": "严重受限",
                "duration_seconds": severe_total,
                "share_percent": round((severe_total / total_runtime) * 100, 2) if total_runtime > 0 else 0.0,
            },
        ]

        domain_buckets: Dict[str, Dict[str, Any]] = {}
        for row in severe_items:
            failure_domain = row["failure_domain"]
            failure_subtype = row["failure_subtype"]
            if not failure_domain or not failure_subtype:
                failure_domain, failure_subtype = _classify_failure(row["error_type"], row["error_message"])
            duration_seconds = row_total(row)
            reason_label = _humanize_failure_reason(row["error_type"], row["error_message"])
            message_label = row["error_message"] or reason_label
            sample = _build_failure_task_sample(row)

            domain_entry = domain_buckets.setdefault(
                failure_domain,
                {
                    "domain": failure_domain,
                    "label": _humanize_failure_domain(failure_domain),
                    "duration_seconds": 0.0,
                    "task_count": 0,
                    "_subtypes": {},
                    "_reasons": {},
                    "_messages": {},
                    "_tasks": [],
                },
            )
            domain_entry["duration_seconds"] += duration_seconds
            domain_entry["task_count"] += 1
            domain_entry["_tasks"].append(sample)

            domain_reason = domain_entry["_reasons"].setdefault(reason_label, {"label": reason_label, "duration_seconds": 0.0, "task_count": 0})
            domain_reason["duration_seconds"] += duration_seconds
            domain_reason["task_count"] += 1

            domain_message = domain_entry["_messages"].setdefault(message_label, {"label": message_label, "duration_seconds": 0.0, "task_count": 0})
            domain_message["duration_seconds"] += duration_seconds
            domain_message["task_count"] += 1

            subtype_entry = domain_entry["_subtypes"].setdefault(
                failure_subtype,
                {
                    "subtype": failure_subtype,
                    "label": _humanize_failure_subtype(failure_domain, failure_subtype, row["error_type"], row["error_message"]),
                    "duration_seconds": 0.0,
                    "task_count": 0,
                    "_reasons": {},
                    "_messages": {},
                    "_tasks": [],
                },
            )
            subtype_entry["duration_seconds"] += duration_seconds
            subtype_entry["task_count"] += 1
            subtype_entry["_tasks"].append(sample)

            subtype_reason = subtype_entry["_reasons"].setdefault(reason_label, {"label": reason_label, "duration_seconds": 0.0, "task_count": 0})
            subtype_reason["duration_seconds"] += duration_seconds
            subtype_reason["task_count"] += 1

            subtype_message = subtype_entry["_messages"].setdefault(message_label, {"label": message_label, "duration_seconds": 0.0, "task_count": 0})
            subtype_message["duration_seconds"] += duration_seconds
            subtype_message["task_count"] += 1

        failure_domain_breakdown = []
        failure_subtype_breakdown = []
        for domain_entry in sorted(domain_buckets.values(), key=lambda item: item["duration_seconds"], reverse=True):
            domain_duration = float(domain_entry["duration_seconds"])
            subtypes = []
            for subtype_entry in sorted(domain_entry["_subtypes"].values(), key=lambda item: item["duration_seconds"], reverse=True):
                subtype_duration = float(subtype_entry["duration_seconds"])
                subtype_reasons = sorted(subtype_entry["_reasons"].values(), key=lambda item: item["duration_seconds"], reverse=True)
                subtype_messages = sorted(subtype_entry["_messages"].values(), key=lambda item: item["duration_seconds"], reverse=True)
                subtype_tasks = sorted(
                    subtype_entry["_tasks"],
                    key=lambda item: (
                        float(item.get("failed_total_duration_seconds") or item.get("total_duration_seconds") or 0.0),
                        item.get("task_ended_at", ""),
                    ),
                    reverse=True,
                )
                reasons = [
                    {
                        "label": item["label"],
                        "duration_seconds": round(float(item["duration_seconds"]), 2),
                        "task_count": int(item["task_count"]),
                        "overall_share_percent": round((float(item["duration_seconds"]) / total_runtime) * 100, 2) if total_runtime > 0 else 0.0,
                        "error_share_percent": round((float(item["duration_seconds"]) / severe_total) * 100, 2) if severe_total > 0 else 0.0,
                        "domain_share_percent": round((float(item["duration_seconds"]) / domain_duration) * 100, 2) if domain_duration > 0 else 0.0,
                        "subtype_share_percent": round((float(item["duration_seconds"]) / subtype_duration) * 100, 2) if subtype_duration > 0 else 0.0,
                    }
                    for item in subtype_reasons
                ]
                messages = [
                    {
                        "label": item["label"],
                        "message": item["label"],
                        "duration_seconds": round(float(item["duration_seconds"]), 2),
                        "task_count": int(item["task_count"]),
                        "overall_share_percent": round((float(item["duration_seconds"]) / total_runtime) * 100, 2) if total_runtime > 0 else 0.0,
                        "error_share_percent": round((float(item["duration_seconds"]) / severe_total) * 100, 2) if severe_total > 0 else 0.0,
                        "domain_share_percent": round((float(item["duration_seconds"]) / domain_duration) * 100, 2) if domain_duration > 0 else 0.0,
                        "subtype_share_percent": round((float(item["duration_seconds"]) / subtype_duration) * 100, 2) if subtype_duration > 0 else 0.0,
                    }
                    for item in subtype_messages
                ]
                subtype_item = {
                    "domain": domain_entry["domain"],
                    "subtype": subtype_entry["subtype"],
                    "label": subtype_entry["label"],
                    "duration_seconds": round(subtype_duration, 2),
                    "task_count": int(subtype_entry["task_count"]),
                    "avg_duration_seconds": round(subtype_duration / subtype_entry["task_count"], 2) if subtype_entry["task_count"] else 0.0,
                    "overall_share_percent": round((subtype_duration / total_runtime) * 100, 2) if total_runtime > 0 else 0.0,
                    "error_share_percent": round((subtype_duration / severe_total) * 100, 2) if severe_total > 0 else 0.0,
                    "domain_share_percent": round((subtype_duration / domain_duration) * 100, 2) if domain_duration > 0 else 0.0,
                    "reasons": reasons,
                    "messages": messages,
                    "tasks": subtype_tasks,
                }
                subtypes.append(subtype_item)
                failure_subtype_breakdown.append(subtype_item)

            domain_reasons = sorted(domain_entry["_reasons"].values(), key=lambda item: item["duration_seconds"], reverse=True)
            domain_messages = sorted(domain_entry["_messages"].values(), key=lambda item: item["duration_seconds"], reverse=True)
            domain_tasks = sorted(
                domain_entry["_tasks"],
                key=lambda item: (
                    float(item.get("failed_total_duration_seconds") or item.get("total_duration_seconds") or 0.0),
                    item.get("task_ended_at", ""),
                ),
                reverse=True,
            )
            failure_domain_breakdown.append(
                {
                    "domain": domain_entry["domain"],
                    "label": domain_entry["label"],
                    "duration_seconds": round(domain_duration, 2),
                    "task_count": int(domain_entry["task_count"]),
                    "overall_share_percent": round((domain_duration / total_runtime) * 100, 2) if total_runtime > 0 else 0.0,
                    "error_share_percent": round((domain_duration / severe_total) * 100, 2) if severe_total > 0 else 0.0,
                    "share_percent": round((domain_duration / severe_total) * 100, 2) if severe_total > 0 else 0.0,
                    "subtypes": subtypes,
                    "reasons": [
                        {
                            "label": item["label"],
                            "duration_seconds": round(float(item["duration_seconds"]), 2),
                            "task_count": int(item["task_count"]),
                            "overall_share_percent": round((float(item["duration_seconds"]) / total_runtime) * 100, 2) if total_runtime > 0 else 0.0,
                            "error_share_percent": round((float(item["duration_seconds"]) / severe_total) * 100, 2) if severe_total > 0 else 0.0,
                            "domain_share_percent": round((float(item["duration_seconds"]) / domain_duration) * 100, 2) if domain_duration > 0 else 0.0,
                        }
                        for item in domain_reasons
                    ],
                    "messages": [
                        {
                            "label": item["label"],
                            "message": item["label"],
                            "duration_seconds": round(float(item["duration_seconds"]), 2),
                            "task_count": int(item["task_count"]),
                            "overall_share_percent": round((float(item["duration_seconds"]) / total_runtime) * 100, 2) if total_runtime > 0 else 0.0,
                            "error_share_percent": round((float(item["duration_seconds"]) / severe_total) * 100, 2) if severe_total > 0 else 0.0,
                            "domain_share_percent": round((float(item["duration_seconds"]) / domain_duration) * 100, 2) if domain_duration > 0 else 0.0,
                        }
                        for item in domain_messages
                    ],
                    "tasks": domain_tasks,
                }
            )

        failure_reasons = Counter()
        for row in severe_items:
            label = _humanize_failure_reason(row["error_type"], row["error_message"])
            failure_reasons[label] += 1

        if len(time_composition) > 1:
            time_composition[1]["details"] = [
                {
                    "domain": item["domain"],
                    "label": item["label"],
                    "duration_seconds": item["duration_seconds"],
                    "overall_share_percent": item["overall_share_percent"],
                    "error_share_percent": item["error_share_percent"],
                    "task_count": item["task_count"],
                }
                for item in failure_domain_breakdown
            ]

        worker_stage_buckets: Dict[str, Dict[str, Any]] = {}
        total_collect_seconds = 0.0
        for row in rows:
            current_worker_id = row["worker_id"]
            if not current_worker_id:
                continue
            download_seconds = float(row["download_duration_seconds"] or 0.0)
            collect_seconds = float(row["collect_duration_seconds"] or 0.0)
            if download_seconds <= 0 and collect_seconds <= 0:
                continue
            stage_total_seconds = download_seconds + collect_seconds
            bucket = worker_stage_buckets.setdefault(
                current_worker_id,
                {
                    "worker_id": current_worker_id,
                    "label": worker_label_map.get(current_worker_id, current_worker_id),
                    "download_duration_seconds": 0.0,
                    "collect_duration_seconds": 0.0,
                    "stage_total_seconds": 0.0,
                    "task_count": 0,
                    "success_count": 0,
                    "non_success_count": 0,
                },
            )
            bucket["download_duration_seconds"] += download_seconds
            bucket["collect_duration_seconds"] += collect_seconds
            bucket["stage_total_seconds"] += stage_total_seconds
            bucket["task_count"] += 1
            if row["status"] == "success":
                bucket["success_count"] += 1
            else:
                bucket["non_success_count"] += 1
            total_collect_seconds += collect_seconds

        worker_collect_share_breakdown = [
            {
                "worker_id": item["worker_id"],
                "label": item["label"],
                "download_duration_seconds": round(float(item["download_duration_seconds"]), 2),
                "collect_duration_seconds": round(float(item["collect_duration_seconds"]), 2),
                "stage_total_seconds": round(float(item["stage_total_seconds"]), 2),
                "collect_share_percent": round((float(item["collect_duration_seconds"]) / float(item["stage_total_seconds"])) * 100, 2)
                if item["stage_total_seconds"] > 0
                else 0.0,
                "overall_collect_share_percent": round((float(item["collect_duration_seconds"]) / total_collect_seconds) * 100, 2)
                if total_collect_seconds > 0
                else 0.0,
                "task_count": int(item["task_count"]),
                "success_count": int(item["success_count"]),
                "non_success_count": int(item["non_success_count"]),
            }
            for item in sorted(
                worker_stage_buckets.values(),
                key=lambda current: (
                    round((float(current["collect_duration_seconds"]) / float(current["stage_total_seconds"])) * 100, 6)
                    if current["stage_total_seconds"] > 0
                    else 0.0,
                    float(current["collect_duration_seconds"]),
                ),
                reverse=True,
            )
        ]

        return {
            "success_stage_breakdown": summarize(qualified_items),
            "non_success_stage_breakdown": summarize(severe_items),
            "time_composition": time_composition,
            "daily_collection_overview": {
                "qualified_task_count": int(collection_quality["qualified_task_count"]),
                "qualified_success_task_count": int(collection_quality["qualified_success_task_count"]),
                "qualified_light_task_count": int(collection_quality["qualified_light_task_count"]),
                "non_success_task_count": int(collection_quality["non_success_task_count"]),
                "non_retryable_task_count": int(collection_quality["non_retryable_task_count"]),
            },
            "runtime_state_breakdown": self._runtime_state_breakdown(
                date_str,
                dispatcher=dispatcher,
                worker_id=worker_id,
            ),
            "failure_domain_breakdown": failure_domain_breakdown,
            "failure_subtype_breakdown": failure_subtype_breakdown,
            "failure_reasons": [
                {"label": label, "count": count}
                for label, count in failure_reasons.most_common()
            ],
            "success_task_count": len(qualified_items),
            "non_success_task_count": len(severe_items),
            "worker_collect_share_breakdown": worker_collect_share_breakdown,
            "total_collect_duration_seconds": round(total_collect_seconds, 2),
            "selected_worker_id": worker_id or "",
            "worker_options": worker_options,
        }

    def get_timeline(
        self,
        date_str: Optional[str] = None,
        dispatcher=None,
        bucket_minutes: int = MONITORING_TIMELINE_BUCKET_MINUTES,
        worker_id: Optional[str] = None,
    ) -> Dict[str, Any]:
        date_str = date_str or _date_str_from_ts(time.time())
        day_start, day_end, _ = _day_bounds(date_str)
        intervals = self._build_worker_intervals(date_str, dispatcher=dispatcher, worker_id=worker_id)
        bucket_seconds = max(1, bucket_minutes) * 60
        buckets: List[Dict[str, Any]] = []
        bucket_start = day_start
        while bucket_start < day_end:
            buckets.append(
                {
                    "timestamp": bucket_start,
                    "label": datetime.fromtimestamp(bucket_start, SHANGHAI_TZ).strftime("%H:%M"),
                    "online_workers": 0,
                    "running_workers": 0,
                    "running_download_workers": 0,
                    "running_collect_workers": 0,
                    "idle_workers": 0,
                    "offline_workers": 0,
                    "disabled_workers": 0,
                }
            )
            bucket_start += bucket_seconds

        for worker_intervals in intervals.values():
            for interval_start, interval_end, state in worker_intervals:
                for bucket in buckets:
                    bucket_start = bucket["timestamp"]
                    bucket_end = bucket_start + bucket_seconds
                    overlap_start = max(interval_start, bucket_start)
                    overlap_end = min(interval_end, bucket_end)
                    if overlap_start >= overlap_end:
                        continue
                    weight = (overlap_end - overlap_start) / bucket_seconds
                    if state in ONLINE_STATES:
                        bucket["online_workers"] += weight
                    if state in RUNNING_STATES:
                        bucket["running_workers"] += weight
                    if state == "running_download":
                        bucket["running_download_workers"] += weight
                    elif state == "running_collect":
                        bucket["running_collect_workers"] += weight
                    elif state == "idle_waiting_task":
                        bucket["idle_workers"] += weight
                    elif state == "disabled":
                        bucket["disabled_workers"] += weight
                    else:
                        bucket["offline_workers"] += weight
        for bucket in buckets:
            for key in (
                "online_workers",
                "running_workers",
                "running_download_workers",
                "running_collect_workers",
                "idle_workers",
                "offline_workers",
                "disabled_workers",
            ):
                bucket[key] = round(float(bucket[key]), 2)
        return {"date": date_str, "bucket_minutes": bucket_minutes, "points": buckets}

    def get_worker_stats(
        self,
        date_str: Optional[str] = None,
        dispatcher=None,
    ) -> List[Dict[str, Any]]:
        now_ts = time.time()
        date_str = date_str or _date_str_from_ts(now_ts)
        is_today = _date_str_from_ts(now_ts) == date_str
        seconds_by_worker = self._summarize_worker_seconds(date_str, dispatcher=dispatcher)
        task_rows = self.repo.list_task_executions(date_str, limit=100000)
        task_stats: Dict[str, Dict[str, Any]] = defaultdict(
            lambda: {
                "task_count": 0,
                "success_count": 0,
                "failed_count": 0,
                "download_values": [],
                "collect_values": [],
                "last_task_at": None,
            }
        )
        for row in task_rows:
            entry = task_stats[row["worker_id"]]
            entry["task_count"] += 1
            if row["status"] == "success":
                entry["success_count"] += 1
            elif row["status"] == "failed":
                entry["failed_count"] += 1
            if row["download_duration_seconds"] is not None:
                entry["download_values"].append(float(row["download_duration_seconds"]))
            if row["collect_duration_seconds"] is not None:
                entry["collect_values"].append(float(row["collect_duration_seconds"]))
            last_task_at = row["task_ended_at"] or row["task_started_at"]
            if last_task_at is not None:
                if entry["last_task_at"] is None or float(last_task_at) > entry["last_task_at"]:
                    entry["last_task_at"] = float(last_task_at)

        current_workers = {}
        if dispatcher is not None and is_today:
            current_workers = {worker.get("worker_id"): worker for worker in dispatcher.get_dashboard_workers()}

        results: List[Dict[str, Any]] = []
        for worker_id in sorted(set(self._resolve_worker_ids(date_str, dispatcher=dispatcher))):
            seconds = seconds_by_worker.get(worker_id, Counter())
            metrics = task_stats.get(worker_id, {})
            current_worker = current_workers.get(worker_id, {})
            current_state = (
                current_worker.get("state")
                or self._get_current_state(worker_id).get("state")
                or ("offline" if not current_worker.get("online") else "idle_waiting_task")
            )
            if is_today and current_worker and not current_worker.get("online"):
                current_state = "offline"
            last_task_at = metrics.get("last_task_at")
            results.append(
                {
                    "worker_id": worker_id,
                    "ip_address": current_worker.get("ip_address") or worker_id.split("_", 1)[0],
                    "status": DISPLAY_STATE_LABELS.get(current_state, current_state),
                    "running_seconds": round(seconds.get("running_download", 0.0) + seconds.get("running_collect", 0.0), 2),
                    "idle_waiting_seconds": round(seconds.get("idle_waiting_task", 0.0), 2),
                    "offline_seconds": round(seconds.get("offline", 0.0), 2),
                    "disabled_seconds": round(seconds.get("disabled", 0.0), 2),
                    "task_count": metrics.get("task_count", 0),
                    "success_count": metrics.get("success_count", 0),
                    "failed_count": metrics.get("failed_count", 0),
                    "avg_download_duration_seconds": round(sum(metrics.get("download_values", [])) / len(metrics.get("download_values", [])), 2)
                    if metrics.get("download_values")
                    else 0.0,
                    "avg_collect_duration_seconds": round(sum(metrics.get("collect_values", [])) / len(metrics.get("collect_values", [])), 2)
                    if metrics.get("collect_values")
                    else 0.0,
                    "last_task_at": datetime.fromtimestamp(last_task_at, SHANGHAI_TZ).strftime("%Y-%m-%d %H:%M:%S")
                    if last_task_at
                    else "--",
                }
            )
        return results

    def get_task_rows(
        self,
        date_str: Optional[str] = None,
        worker_id: Optional[str] = None,
        limit: int = 200,
    ) -> List[Dict[str, Any]]:
        date_str = date_str or _date_str_from_ts(time.time())
        rows = self.repo.list_task_executions(date_str, worker_id=worker_id, limit=limit)
        return [_build_failure_task_sample(row) for row in rows]