from __future__ import annotations

import json
import os
import threading
import time
from datetime import datetime, timedelta
from pathlib import Path

from app.core.config import settings
from app.core.db import get_db
from app.core.files import read_json
from app.core.redis_client import get_redis
from app.services.build_info_service import get_runtime_build_info
from app.services.cluster_runtime_service import get_cluster_snapshot
from app.services.detect_service import get_detect_status
from app.services.detect_job_service import (
    _load_latest_runtime_active_job_snapshot,
    get_detect_capacity_plan,
    get_detect_queue_health,
    get_active_detect_job_summary,
)
from app.services.sync_record_service import (
    _pick_latest_projection_row,
    append_runtime_projection_if_changed,
    get_sync_summary,
)
from app.services.runtime_settings_service import get_runtime_settings
from app.services.worker_control_service import detect_sync_agent_runtime, detect_worker_runtime

_DOMAIN_INVENTORY_CACHE_LOCK = threading.Lock()
_DOMAIN_INVENTORY_CACHE_TTL_SECONDS = 30.0
_DOMAIN_INVENTORY_CACHE_VALUE: dict | None = None
_DOMAIN_INVENTORY_CACHE_EXPIRES_AT = 0.0
_RUNTIME_STATUS_CACHE_LOCK = threading.Lock()
_RUNTIME_STATUS_CACHE_TTL_SECONDS = 3.0
_RUNTIME_STATUS_CACHE_VALUE: dict | None = None
_RUNTIME_STATUS_CACHE_EXPIRES_AT = 0.0
_RUNTIME_STATUS_CACHE_SIGNATURE: tuple[object, ...] = ()


def _clone_runtime_status_payload(value: dict | None) -> dict:
    try:
        return json.loads(json.dumps(dict(value or {}), ensure_ascii=False))
    except Exception:
        return dict(value or {})


def _runtime_status_cache_signature() -> tuple[object, ...]:
    return (
        str(settings.node_code or ""),
        str(settings.node_region or ""),
        str(settings.node_role or ""),
        int(settings.api_port or 0),
        id(_build_runtime_detect_context),
        id(detect_sync_agent_runtime),
        id(append_runtime_projection_if_changed),
        id(get_sync_summary),
        id(_build_multi_region_readiness),
        id(get_runtime_build_info),
    )


def _align_queue_health_with_backlog(queue_health: dict | None, backlog_snapshot: dict | None) -> dict:
    normalized = dict(queue_health or {})
    queue = dict(normalized.get("queue") or {})
    backlog = dict(backlog_snapshot or {})

    pending_total = max(int(queue.get("pending", 0) or 0), int(backlog.get("pending_total", 0) or 0))
    claimed_total = max(int(queue.get("claimed", 0) or 0), int(backlog.get("claimed_total", 0) or 0))
    running_total = max(int(queue.get("running", 0) or 0), int(backlog.get("running_total", 0) or 0))
    completed_total = max(int(queue.get("completed", 0) or 0), int(backlog.get("completed_total", 0) or 0))
    blacklisted_total = max(int(queue.get("blacklisted", 0) or 0), int(backlog.get("blacklisted_total", 0) or 0))
    failed_total = max(int(queue.get("failed", 0) or 0), int(backlog.get("failed_total", 0) or 0))
    terminal_total = max(
        int(queue.get("terminal", 0) or 0),
        completed_total + blacklisted_total + failed_total,
    )

    normalized["has_active_job"] = bool(
        normalized.get("has_active_job")
        or pending_total > 0
        or claimed_total > 0
        or running_total > 0
        or terminal_total > 0
    )
    normalized["queue"] = {
        **queue,
        "items_total": pending_total + claimed_total + running_total + terminal_total,
        "pending": pending_total,
        "claimed": claimed_total,
        "running": running_total,
        "completed": completed_total,
        "blacklisted": blacklisted_total,
        "failed": failed_total,
        "terminal": terminal_total,
    }
    return normalized


def _decode_projection_payload(value: object) -> dict:
    if isinstance(value, dict):
        return dict(value)
    if value in (None, ""):
        return {}
    try:
        import json

        return dict(json.loads(value))
    except Exception:
        return {}


def _build_backlog_snapshot_from_active_job(active_job: dict | None) -> dict:
    normalized_job = dict(active_job or {})
    if not normalized_job:
        return {}

    pending_total = max(0, int(normalized_job.get("items_pending", 0) or 0))
    claimed_total = max(
        max(
            int(normalized_job.get("items_claimed", 0) or 0),
            int(normalized_job.get("display_items_claimed", 0) or 0),
        ),
        0,
    )
    running_total = max(
        max(
            int(normalized_job.get("items_running", 0) or 0),
            int(normalized_job.get("display_items_running", 0) or 0),
        ),
        int(normalized_job.get("display_active_threads", 0) or 0),
        0,
    )
    completed_total = max(0, int(normalized_job.get("items_completed", 0) or 0))
    blacklisted_total = max(0, int(normalized_job.get("items_blacklisted", 0) or 0))
    failed_total = max(0, int(normalized_job.get("items_failed", 0) or 0))

    register_pending = 0
    downstream_pending = 0
    for raw_step in list(normalized_job.get("step_stats") or normalized_job.get("raw_step_stats") or []):
        if not isinstance(raw_step, dict):
            continue
        step_code = str(raw_step.get("step_code") or raw_step.get("code") or "").strip()
        step_pending = max(
            int(raw_step.get("items_pending", raw_step.get("pending", 0)) or 0),
            0,
        )
        if step_pending <= 0:
            continue
        if step_code == "detect_register":
            register_pending += step_pending
        else:
            downstream_pending += step_pending

    if register_pending <= 0 and downstream_pending <= 0 and pending_total > 0:
        downstream_pending = pending_total

    if (
        pending_total <= 0
        and claimed_total <= 0
        and running_total <= 0
        and completed_total <= 0
        and blacklisted_total <= 0
        and failed_total <= 0
    ):
        return {}

    return {
        "pending_total": pending_total,
        "claimed_total": claimed_total,
        "running_total": running_total,
        "completed_total": completed_total,
        "blacklisted_total": blacklisted_total,
        "failed_total": failed_total,
        "register_pending": register_pending,
        "downstream_pending": downstream_pending,
    }


def _load_detect_backlog_snapshot() -> dict:
    active_job_snapshot = _build_backlog_snapshot_from_active_job(
        get_active_detect_job_summary(event_limit=1)
    )
    if active_job_snapshot:
        return active_job_snapshot

    with get_db() as conn:
        with conn.cursor() as cur:
            cur.execute(
                """
                SELECT id, status, COALESCE(started_at, created_at) AS activity_at
                FROM detect_jobs
                WHERE status IN ('pending', 'running')
                ORDER BY
                    CASE WHEN status = 'running' THEN 0 ELSE 1 END,
                    COALESCE(started_at, created_at) DESC,
                    id DESC
                LIMIT %s
                """,
                (32,),
            )
            job_rows = list(cur.fetchall() or [])
            selected_job_ids: list[int] = []
            fallback_job_id = 0
            for raw_job_id, raw_status, raw_activity_at in job_rows:
                try:
                    job_id = int(raw_job_id or 0)
                except (TypeError, ValueError):
                    continue
                if job_id <= 0:
                    continue
                if fallback_job_id <= 0:
                    fallback_job_id = job_id
                if job_id in selected_job_ids:
                    continue
                status = str(raw_status or "").strip().lower()
                keep = status == "running"
                if not keep and raw_activity_at is not None:
                    now = datetime.now(raw_activity_at.tzinfo) if getattr(raw_activity_at, "tzinfo", None) else datetime.now()
                    keep = now - raw_activity_at <= timedelta(hours=6)
                if not keep:
                    continue
                selected_job_ids.append(job_id)
                if len(selected_job_ids) >= 4:
                    break
            if not selected_job_ids and fallback_job_id > 0:
                selected_job_ids.append(fallback_job_id)
            if not selected_job_ids:
                return {
                    "pending_total": 0,
                    "claimed_total": 0,
                    "running_total": 0,
                    "completed_total": 0,
                    "blacklisted_total": 0,
                    "failed_total": 0,
                    "register_pending": 0,
                    "downstream_pending": 0,
                }
            cur.execute(
                """
                SELECT
                    COUNT(*) FILTER (WHERE item.status = 'pending') AS pending_total,
                    COUNT(*) FILTER (WHERE item.status = 'claimed') AS claimed_total,
                    COUNT(*) FILTER (WHERE item.status = 'running') AS running_total,
                    COUNT(*) FILTER (WHERE item.status = 'completed') AS completed_total,
                    COUNT(*) FILTER (WHERE item.status = 'blacklisted') AS blacklisted_total,
                    COUNT(*) FILTER (WHERE item.status = 'failed') AS failed_total,
                    COUNT(*) FILTER (WHERE item.status = 'pending' AND item.step_code = 'detect_register') AS register_pending,
                    COUNT(*) FILTER (WHERE item.status = 'pending' AND item.step_code <> 'detect_register') AS downstream_pending
                FROM detect_job_items item
                WHERE item.job_id = ANY(%s)
                """,
                (selected_job_ids,),
            )
            row = cur.fetchone() or (0, 0, 0, 0, 0, 0, 0, 0)
    return {
        "pending_total": int(row[0] or 0),
        "claimed_total": int(row[1] or 0),
        "running_total": int(row[2] or 0),
        "completed_total": int(row[3] or 0),
        "blacklisted_total": int(row[4] or 0),
        "failed_total": int(row[5] or 0),
        "register_pending": int(row[6] or 0),
        "downstream_pending": int(row[7] or 0),
    }


def _load_latest_remote_runtime_projection_backlog() -> dict:
    if not (settings.node_region == "overseas" and settings.node_role == "control"):
        return {}
    with get_db() as conn:
        with conn.cursor() as cur:
            cur.execute(
                """
                SELECT payload_json, COALESCE(updated_at, created_at)
                FROM detect_sync_records
                WHERE sync_type IN ('runtime_ingest', 'runtime_projection')
                  AND source_region = 'mainland'
                  AND target_region = 'overseas'
                  AND status IN ('received', 'projected', 'pushing', 'synced')
                ORDER BY
                    CASE WHEN sync_type = 'runtime_ingest' THEN 0 ELSE 1 END ASC,
                    COALESCE(updated_at, created_at) DESC,
                    id DESC
                LIMIT 50
                """
            )
            row = _pick_latest_projection_row(list(cur.fetchall() or []), created_at_index=1)
    if not row:
        return {}
    payload = _decode_projection_payload(row[0])
    projection = payload.get("projection") if isinstance(payload, dict) else {}
    backlog = projection.get("backlog") if isinstance(projection, dict) else {}
    if not isinstance(backlog, dict):
        return {}
    return {
        "pending_total": int(backlog.get("pending_total", 0) or 0),
        "claimed_total": int(backlog.get("claimed_total", 0) or 0),
        "running_total": int(backlog.get("running_total", 0) or 0),
        "completed_total": int(backlog.get("completed_total", 0) or 0),
        "blacklisted_total": int(backlog.get("blacklisted_total", 0) or 0),
        "failed_total": int(backlog.get("failed_total", 0) or 0),
        "register_pending": int(backlog.get("register_pending", 0) or 0),
        "downstream_pending": int(backlog.get("downstream_pending", 0) or 0),
    }


def _merge_backlog_snapshots(primary: dict, secondary: dict) -> dict:
    merged = dict(primary or {})
    for key in (
        "pending_total",
        "claimed_total",
        "running_total",
        "completed_total",
        "blacklisted_total",
        "failed_total",
        "register_pending",
        "downstream_pending",
    ):
        merged[key] = max(int(merged.get(key, 0) or 0), int((secondary or {}).get(key, 0) or 0))
    return merged


def _safe_int_value(value: object, default: int = 0) -> int:
    try:
        return int(value or 0)
    except Exception:
        return int(default)


def _safe_float_value(value: object, default: float = 0.0) -> float:
    try:
        return float(value or 0.0)
    except Exception:
        return float(default)


def _safe_text_value(value: object) -> str:
    return str(value or "").strip()


def _load_domain_inventory_summary() -> dict:
    global _DOMAIN_INVENTORY_CACHE_VALUE, _DOMAIN_INVENTORY_CACHE_EXPIRES_AT

    now_ts = time.monotonic()
    with _DOMAIN_INVENTORY_CACHE_LOCK:
        if _DOMAIN_INVENTORY_CACHE_VALUE is not None and now_ts < _DOMAIN_INVENTORY_CACHE_EXPIRES_AT:
            return dict(_DOMAIN_INVENTORY_CACHE_VALUE)

    authoritative = bool(settings.node_region == "overseas" and settings.node_role == "control")
    summary = {
        "scope_label": "海外主库总盘子" if authoritative else "当前节点本地库存",
        "authoritative": authoritative,
        "domains_total": 0,
        "pending_total": 0,
        "completed_total": 0,
        "running_total": 0,
        "blacklist_total": 0,
        "failed_total": 0,
        "processed_total": 0,
        "remaining_total": 0,
    }
    try:
        with get_db() as conn:
            with conn.cursor() as cur:
                cur.execute(
                    """
                    SELECT
                        COUNT(*) AS domains_total,
                        COUNT(*) FILTER (WHERE detect_status = 0) AS pending_total,
                        COUNT(*) FILTER (WHERE detect_status = 1) AS completed_total,
                        COUNT(*) FILTER (WHERE detect_status = 2) AS running_total,
                        COUNT(*) FILTER (WHERE detect_status = 3) AS blacklist_total,
                        COUNT(*) FILTER (WHERE detect_status = 4) AS failed_total
                    FROM domains
                    """
                )
                row = cur.fetchone() or (0, 0, 0, 0, 0, 0)
        summary.update(
            {
                "domains_total": int(row[0] or 0),
                "pending_total": int(row[1] or 0),
                "completed_total": int(row[2] or 0),
                "running_total": int(row[3] or 0),
                "blacklist_total": int(row[4] or 0),
                "failed_total": int(row[5] or 0),
            }
        )
        summary["processed_total"] = (
            int(summary["completed_total"])
            + int(summary["blacklist_total"])
            + int(summary["failed_total"])
        )
        summary["remaining_total"] = int(summary["pending_total"]) + int(summary["running_total"])
    except Exception as exc:
        summary["error"] = str(exc)

    with _DOMAIN_INVENTORY_CACHE_LOCK:
        _DOMAIN_INVENTORY_CACHE_VALUE = dict(summary)
        _DOMAIN_INVENTORY_CACHE_EXPIRES_AT = time.monotonic() + _DOMAIN_INVENTORY_CACHE_TTL_SECONDS
    return dict(summary)


def _build_detect_observation_top_nodes(participating_nodes: list[dict] | None) -> list[dict]:
    rows: list[dict] = []
    for raw_item in list(participating_nodes or []):
        item = dict(raw_item or {})
        node_code = _safe_text_value(item.get("node_code"))
        if not node_code:
            continue
        current_load = max(
            _safe_int_value(item.get("current_load")),
            _safe_int_value(item.get("active_threads")),
            _safe_int_value(item.get("items_running")),
            _safe_int_value(item.get("items_claimed")),
        )
        processed_recent = _safe_int_value(item.get("processed_recent"))
        if current_load <= 0 and processed_recent <= 0 and not bool(item.get("is_current_participant", False)):
            continue
        rows.append(
            {
                "node_code": node_code,
                "participation_label": _safe_text_value(item.get("participation_label")) or "参与中",
                "items_claimed": _safe_int_value(item.get("items_claimed")),
                "items_running": _safe_int_value(item.get("items_running")),
                "active_threads": _safe_int_value(item.get("active_threads")),
                "max_threads": _safe_int_value(item.get("max_threads")),
                "processed_recent": processed_recent,
                "processed_per_minute": round(_safe_float_value(item.get("processed_per_minute")), 2),
                "current_load": current_load,
            }
        )
    rows.sort(
        key=lambda item: (
            -int(item.get("current_load", 0) or 0),
            -int(item.get("processed_recent", 0) or 0),
            -int(item.get("items_claimed", 0) or 0),
            str(item.get("node_code") or ""),
        )
    )
    return rows[:6]


def _build_detect_observation_top_steps(detect_payload: dict) -> list[dict]:
    queue_steps = list(((detect_payload.get("queue_health") or {}).get("steps") or []))
    if not queue_steps:
        queue_steps = list(((detect_payload.get("active_job") or {}).get("step_stats") or []))
    rows: list[dict] = []
    for raw_item in queue_steps:
        item = dict(raw_item or {})
        step_code = _safe_text_value(item.get("step_code") or item.get("code"))
        if not step_code:
            continue
        rows.append(
            {
                "step_code": step_code,
                "step_name": _safe_text_value(item.get("step_name")) or step_code,
                "pending": max(
                    _safe_int_value(item.get("items_pending")),
                    _safe_int_value(item.get("pending")),
                ),
                "running": max(
                    _safe_int_value(item.get("items_running")),
                    _safe_int_value(item.get("running")),
                ),
                "processed_recent": _safe_int_value(item.get("processed_recent")),
                "failed_recent": _safe_int_value(item.get("failed_recent")),
                "blacklisted_recent": _safe_int_value(item.get("blacklisted_recent")),
            }
        )
    rows.sort(
        key=lambda item: (
            -int(item.get("pending", 0) or 0),
            -int(item.get("running", 0) or 0),
            -int(item.get("processed_recent", 0) or 0),
            str(item.get("step_name") or ""),
        )
    )
    return [item for item in rows if item["pending"] > 0 or item["running"] > 0 or item["processed_recent"] > 0][:6]


def _build_detect_observation_summary(
    *,
    detect_payload: dict,
    cluster_snapshot: dict,
    inventory_summary: dict | None = None,
) -> dict:
    normalized_detect = dict(detect_payload or {})
    active_job = dict(normalized_detect.get("active_job") or {})
    queue_health = dict(normalized_detect.get("queue_health") or {})
    queue = dict(queue_health.get("queue") or {})
    throughput = dict(queue_health.get("throughput") or {})
    participation_summary = dict(normalized_detect.get("participation_summary") or {})
    source_inventory = dict(inventory_summary or {})
    participating_nodes = [
        dict(item)
        for item in list(normalized_detect.get("participating_nodes") or [])
        if isinstance(item, dict)
    ]
    top_active_nodes = _build_detect_observation_top_nodes(participating_nodes)
    top_steps = _build_detect_observation_top_steps(normalized_detect)

    job_id = _safe_int_value(active_job.get("job_id"))
    job_code = _safe_text_value(active_job.get("job_code"))
    job_status = _safe_text_value(active_job.get("status")) or "-"
    progress_percent = round(
        _safe_float_value(active_job.get("progress_percent"), _safe_float_value(normalized_detect.get("progress_percent"))),
        2,
    )

    pending = max(
        _safe_int_value(active_job.get("items_pending")),
        _safe_int_value(queue.get("pending")),
        _safe_int_value((normalized_detect.get("backlog") or {}).get("pending_total")),
    )
    claimed = max(
        _safe_int_value(active_job.get("display_items_claimed")),
        _safe_int_value(active_job.get("items_claimed")),
        _safe_int_value(queue.get("display_claimed")),
        _safe_int_value(queue.get("claimed")),
        _safe_int_value((normalized_detect.get("backlog") or {}).get("claimed_total")),
    )
    running = max(
        _safe_int_value(active_job.get("display_items_running")),
        _safe_int_value(active_job.get("display_active_threads")),
        _safe_int_value(active_job.get("items_running")),
        _safe_int_value(queue.get("display_running")),
        _safe_int_value(queue.get("running")),
        _safe_int_value((normalized_detect.get("backlog") or {}).get("running_total")),
    )
    completed = max(
        _safe_int_value(active_job.get("items_completed")),
        _safe_int_value(queue.get("completed")),
        _safe_int_value((normalized_detect.get("backlog") or {}).get("completed_total")),
    )
    failed = max(
        _safe_int_value(active_job.get("items_failed")),
        _safe_int_value(queue.get("failed")),
        _safe_int_value((normalized_detect.get("backlog") or {}).get("failed_total")),
    )
    blacklisted = max(
        _safe_int_value(active_job.get("items_blacklisted")),
        _safe_int_value(queue.get("blacklisted")),
        _safe_int_value((normalized_detect.get("backlog") or {}).get("blacklisted_total")),
    )
    effective_items_total = max(
        _safe_int_value(active_job.get("items_total")),
        _safe_int_value(queue.get("items_total")),
        pending + claimed + running + completed + failed + blacklisted,
    )
    raw_items_total = max(
        _safe_int_value(active_job.get("raw_items_total")),
        effective_items_total,
    )
    raw_pending = max(_safe_int_value(active_job.get("raw_items_pending")), pending)
    raw_claimed = max(_safe_int_value(active_job.get("raw_items_claimed")), claimed)
    raw_running = max(_safe_int_value(active_job.get("raw_items_running")), running)
    raw_completed = max(_safe_int_value(active_job.get("raw_items_completed")), completed)
    raw_failed = max(_safe_int_value(active_job.get("raw_items_failed")), failed)
    raw_blacklisted = max(_safe_int_value(active_job.get("raw_items_blacklisted")), blacklisted)

    derived_active_processes = sum(
        1
        for item in participating_nodes
        if bool(item.get("is_dispatch_active", False))
        or _safe_int_value(item.get("items_running")) > 0
        or _safe_int_value(item.get("items_claimed")) > 0
        or _safe_int_value(item.get("active_threads")) > 0
    )
    if derived_active_processes <= 0:
        derived_active_processes = max(
            _safe_int_value(participation_summary.get("dispatch_active_nodes")),
            _safe_int_value(normalized_detect.get("aggregate_process_count")),
        )
    active_threads = max(
        _safe_int_value(normalized_detect.get("active_thread_count")),
        _safe_int_value(active_job.get("display_active_threads")),
        _safe_int_value(active_job.get("display_items_running")),
        _safe_int_value(queue.get("display_running")),
        _safe_int_value(queue.get("running")),
    )
    active_max_threads = max(
        _safe_int_value(normalized_detect.get("max_thread_count")),
        _safe_int_value(active_job.get("display_max_threads")),
        sum(_safe_int_value(item.get("max_threads")) for item in top_active_nodes if _safe_int_value(item.get("max_threads")) > 0),
    )
    thread_utilization_percent = round((active_threads / active_max_threads) * 100, 2) if active_max_threads > 0 else 0.0

    processed_recent = max(
        _safe_int_value(active_job.get("processed_recent")),
        _safe_int_value(throughput.get("processed_recent")),
    )
    processed_per_minute = max(
        round(_safe_float_value(active_job.get("processed_per_minute")), 2),
        round(_safe_float_value(throughput.get("processed_per_minute")), 2),
    )
    completed_recent = max(
        _safe_int_value(active_job.get("completed_recent")),
        _safe_int_value(throughput.get("completed_recent")),
    )
    failed_recent = max(
        _safe_int_value(active_job.get("failed_recent")),
        _safe_int_value(throughput.get("failed_recent")),
    )
    blacklisted_recent = max(
        _safe_int_value(active_job.get("blacklisted_recent")),
        _safe_int_value(throughput.get("blacklisted_recent")),
    )

    online_worker_nodes = _safe_int_value(((cluster_snapshot.get("summary") or {}).get("online_worker_nodes")))
    participating_node_count = max(
        _safe_int_value(participation_summary.get("participating_nodes")),
        len(participating_nodes),
    )
    dispatch_active_nodes = max(
        _safe_int_value(participation_summary.get("dispatch_active_nodes")),
        derived_active_processes,
    )
    source_domains_total = _safe_int_value(source_inventory.get("domains_total"))
    source_pending_total = _safe_int_value(source_inventory.get("pending_total"))
    source_completed_total = _safe_int_value(source_inventory.get("completed_total"))
    source_running_total = _safe_int_value(source_inventory.get("running_total"))
    source_blacklist_total = _safe_int_value(source_inventory.get("blacklist_total"))
    source_failed_total = _safe_int_value(source_inventory.get("failed_total"))
    source_remaining_total = _safe_int_value(source_inventory.get("remaining_total"))

    if job_id <= 0 and not job_code and pending <= 0 and claimed <= 0 and running <= 0 and active_threads <= 0:
        state = "idle"
        state_label = "当前空闲"
        state_reason = "当前没有活跃检测任务，执行面处于待命或暂时没有可观察样本。"
    elif derived_active_processes > 0 and active_threads > 0 and processed_recent > 0:
        state = "running"
        state_label = "真跑中"
        state_reason = "已经看到真实执行进程、活跃线程和近窗吞吐，不是只剩日志残影。"
    elif derived_active_processes > 0 and active_threads > 0 and max(running, claimed) > 0:
        state = "slow"
        state_label = "在跑但偏慢"
        state_reason = "执行面已经起来了，但最近吞吐还没完全拉起来，瓶颈更像外部步骤或领取效率。"
    elif pending > 0 and derived_active_processes <= 0 and active_threads <= 0 and processed_recent <= 0:
        state = "not_running"
        state_label = "没跑起来"
        state_reason = "队列里还有积压，但当前没看到真实执行进程和线程在持续消化。"
    elif pending > 0 and processed_recent <= 0:
        state = "stale"
        state_label = "疑似残影"
        state_reason = "还能看到积压或运行中计数，但最近没有吞吐增量，需要继续查执行链或状态回传。"
    else:
        state = "watching"
        state_label = "正在观察"
        state_reason = "当前已有部分运行信号，但还需要继续观察吞吐和结果产出是否稳定。"

    if source_domains_total > 0 and raw_items_total > 0:
        scope_hint = (
            f"{_safe_text_value(source_inventory.get('scope_label')) or '当前总盘子'} {source_domains_total}；"
            f"当前活跃批次原始 {raw_items_total} 项，展示口径 {effective_items_total} 项，不等于全盘累计。"
        )
    elif source_domains_total > 0:
        scope_hint = (
            f"{_safe_text_value(source_inventory.get('scope_label')) or '当前总盘子'} {source_domains_total}；"
            "当前还没有可对齐的活跃批次。"
        )
    else:
        scope_hint = "当前总盘子摘要暂时不可用。"

    if failed_recent > max(completed_recent, blacklisted_recent) and failed_recent > 0:
        focus_hint = (
            f"最近失败 {failed_recent} 明显高于完成 {completed_recent}，更像外部步骤超时、代理/RDAP 或链路异常。"
        )
    elif blacklisted_recent > 0 and blacklisted_recent >= failed_recent:
        focus_hint = f"最近黑名单命中 {blacklisted_recent}，说明当前推进里有一部分是被规则直接拦截。"
    elif pending > 0 and active_threads <= 0:
        focus_hint = "队列还有积压，但当前看不到真实执行线程，先别盯日志，先盯任务领取和 worker 存活。"
    elif active_max_threads > 0 and active_threads > 0 and thread_utilization_percent < 10:
        focus_hint = f"线程利用率只有 {thread_utilization_percent}% ，执行面没有吃满，瓶颈更像外部链路或任务领取。"
    elif processed_recent > 0:
        focus_hint = f"近 15 分钟已处理 {processed_recent} 项，约 {processed_per_minute} 项/分钟。"
    else:
        focus_hint = "当前还没有明显的近窗吞吐样本，继续盯任务是否持续出结果。"

    summary_lines = [
        (
            f"{_safe_text_value(source_inventory.get('scope_label')) or '总盘子'} {source_domains_total}，"
            f"库存待检测 {source_pending_total} / 已通过 {source_completed_total} / 运行中 {source_running_total} / "
            f"失败 {source_failed_total} / 黑名单 {source_blacklist_total}。"
        ),
        (
            f"当前任务 {job_code or '-'}，展示 {effective_items_total} 项 / 原始 {raw_items_total} 项；"
            f"待 {pending} / 领 {claimed} / 跑 {running} / 完 {completed} / 失败 {failed} / 黑名单 {blacklisted}。"
        ),
        (
            f"真实执行面 {derived_active_processes} 个进程，活跃线程 {active_threads} / {active_max_threads}，"
            f"线程利用率 {thread_utilization_percent}%。"
        ),
        (
            f"近 15 分钟处理 {processed_recent} 项，约 {processed_per_minute} 项/分钟，"
            f"其中完成 {completed_recent}、失败 {failed_recent}、黑名单 {blacklisted_recent}。"
        ),
    ]

    return {
        "state": state,
        "state_label": state_label,
        "state_reason": state_reason,
        "scope_hint": scope_hint,
        "focus_hint": focus_hint,
        "source_inventory": {
            "scope_label": _safe_text_value(source_inventory.get("scope_label")) or "当前总盘子",
            "authoritative": bool(source_inventory.get("authoritative", False)),
            "domains_total": source_domains_total,
            "pending_total": source_pending_total,
            "completed_total": source_completed_total,
            "running_total": source_running_total,
            "blacklist_total": source_blacklist_total,
            "failed_total": source_failed_total,
            "processed_total": _safe_int_value(source_inventory.get("processed_total")),
            "remaining_total": source_remaining_total,
            "error": _safe_text_value(source_inventory.get("error")),
        },
        "job": {
            "job_id": job_id,
            "job_code": job_code,
            "status": job_status,
            "progress_percent": progress_percent,
        },
        "active_batch": {
            "job_id": job_id,
            "job_code": job_code,
            "status": job_status,
            "started_at": _safe_text_value(active_job.get("started_at")),
            "progress_percent": progress_percent,
            "effective_items_total": effective_items_total,
            "raw_items_total": raw_items_total,
            "pending": pending,
            "claimed": claimed,
            "running": running,
            "completed": completed,
            "failed": failed,
            "blacklisted": blacklisted,
            "raw_pending": raw_pending,
            "raw_claimed": raw_claimed,
            "raw_running": raw_running,
            "raw_completed": raw_completed,
            "raw_failed": raw_failed,
            "raw_blacklisted": raw_blacklisted,
        },
        "backlog": {
            "pending": pending,
            "claimed": claimed,
            "running": running,
            "completed": completed,
            "failed": failed,
            "blacklisted": blacklisted,
        },
        "execution": {
            "active_processes": derived_active_processes,
            "dispatch_active_nodes": dispatch_active_nodes,
            "participating_nodes": participating_node_count,
            "online_worker_nodes": online_worker_nodes,
            "active_threads": active_threads,
            "max_threads": active_max_threads,
            "thread_utilization_percent": thread_utilization_percent,
        },
        "throughput": {
            "processed_recent": processed_recent,
            "processed_per_minute": processed_per_minute,
            "completed_recent": completed_recent,
            "failed_recent": failed_recent,
            "blacklisted_recent": blacklisted_recent,
        },
        "top_active_nodes": top_active_nodes,
        "top_steps": top_steps,
        "summary_lines": summary_lines,
    }


def _runtime_log_path(filename: str) -> str:
    path = Path(__file__).resolve().parents[2] / "runtime" / "logs" / filename
    return str(path)


def _domain_cookie_status(filename: str) -> tuple[bool, str]:
    path = Path(settings.domain_root) / filename
    return path.exists(), str(path)


def _bloom_filter_status() -> tuple[bool, str]:
    try:
        redis_client = get_redis()
        modules = redis_client.execute_command("MODULE", "LIST")
        for module in modules or []:
            module_parts = module[::2]
            module_values = module[1::2]
            module_info = dict(zip(module_parts, module_values))
            module_name = str(module_info.get("name", "")).lower()
            if module_name in {"bf", "redisbloom"}:
                return True, "RedisBloom 已安装"
        return False, "Redis 未安装 RedisBloom，当前将使用普通缓存"
    except Exception as exc:
        return False, f"RedisBloom 检查失败: {exc}"


def _build_multi_region_readiness(
    *,
    cluster_snapshot: dict,
    sync_summary: dict,
    worker_runtime: dict,
    sync_agent_runtime: dict,
) -> dict:
    nodes = list(cluster_snapshot.get("nodes") or [])
    summary = cluster_snapshot.get("summary") or {}
    batch_summary = (sync_summary.get("detect_result_batches") or {})
    batch_states = batch_summary.get("state_counts") or {}
    batch_applicable = bool(batch_summary.get("applicable", True))

    online_control_nodes = int(summary.get("online_control_nodes", 0) or 0)
    online_worker_nodes = int(summary.get("online_worker_nodes", 0) or 0)
    mainland_control_nodes = [
        node for node in nodes
        if str(node.get("region") or "") == "mainland"
        and str(node.get("role") or "") == "control"
        and str(node.get("status") or "") in {"online", "busy"}
    ]
    mainland_worker_nodes = [
        node for node in nodes
        if str(node.get("region") or "") == "mainland"
        and bool(node.get("is_effective_worker", False))
        and str(node.get("status") or "") in {"online", "busy"}
    ]

    blocking_issues: list[str] = []
    warning_issues: list[str] = []
    info_items: list[str] = []

    online_pairs = {
        (
            str(node.get("region") or "").strip(),
            str(node.get("role") or "").strip(),
        )
        for node in nodes
        if str(node.get("status") or "") in {"online", "busy"}
    }
    critical_stale_nodes: list[str] = []
    redundant_stale_nodes: list[str] = []
    critical_offline_nodes: list[str] = []
    redundant_offline_nodes: list[str] = []

    if online_control_nodes <= 0:
        blocking_issues.append("当前没有在线控制面节点，无法视为正式可用集群。")

    if settings.node_region == "mainland" and settings.node_role == "control":
        if not sync_agent_runtime.get("running", False):
            blocking_issues.append("当前节点应承载 sync-agent，但服务未运行。")
        if str(settings.sync_target_api_base_url or "").strip() == "":
            blocking_issues.append("当前节点未配置 SYNC_TARGET_API_BASE_URL，无法向海外控制面推送。")
        if not bool(settings.sync_push_enabled):
            blocking_issues.append("当前节点未启用 SYNC_PUSH_ENABLED，结果同步不会自动执行。")

    if settings.node_region == "overseas" and online_control_nodes > 0 and not mainland_control_nodes:
        warning_issues.append("当前尚未观察到在线的大陆 controller 节点，后续自动结果同步仍未进入正式双地域态。")

    if online_worker_nodes <= 0:
        warning_issues.append("当前没有在线 Worker 节点，检测任务无法在多机状态下继续推进。")

    for node in nodes:
        node_status = str(node.get("status") or "").strip()
        if node_status not in {"stale", "offline"}:
            continue
        node_code = str(node.get("node_code") or "").strip()
        node_region = str(node.get("region") or "").strip()
        node_role = str(node.get("role") or "").strip()
        metadata = node.get("metadata") or {}
        replaced_by_online_peer = (node_region, node_role) in online_pairs
        redundant = replaced_by_online_peer and (
            node_role == "control"
            or not bool(node.get("is_effective_worker", False))
            or str(metadata.get("service") or "") == "runtime-ingest"
        )
        if node_status == "stale":
            (redundant_stale_nodes if redundant else critical_stale_nodes).append(node_code)
        else:
            (redundant_offline_nodes if redundant else critical_offline_nodes).append(node_code)

    if critical_stale_nodes:
        warning_issues.append(f"存在失活节点: {'、'.join(critical_stale_nodes)}")
    if critical_offline_nodes:
        warning_issues.append(f"存在离线节点: {'、'.join(critical_offline_nodes)}")
    if redundant_stale_nodes:
        info_items.append(f"存在历史失活节点（已被在线同类节点覆盖）: {'、'.join(redundant_stale_nodes)}")
    if redundant_offline_nodes:
        info_items.append(f"存在历史离线节点（已被在线同类节点覆盖）: {'、'.join(redundant_offline_nodes)}")

    failed_batches = int(batch_states.get("failed", 0) or 0)
    projected_batches = int(batch_states.get("projected", 0) or 0)
    pushing_batches = int(batch_states.get("pushing", 0) or 0)
    synced_batches = int(batch_states.get("synced", 0) or 0)
    if failed_batches > 0 and batch_applicable:
        warning_issues.append(f"存在 {failed_batches} 个结果批次同步失败，需要检查 sync-agent 或目标接收面。")
    if projected_batches > 0 and batch_applicable:
        warning_issues.append(f"存在 {projected_batches} 个结果批次仍待推送。")
    if pushing_batches > 0 and batch_applicable:
        info_items.append(f"当前有 {pushing_batches} 个结果批次正在推送。")
    if synced_batches > 0 and batch_applicable:
        info_items.append(f"最近已接收 {synced_batches} 个结果批次。")
    if not batch_applicable:
        info_items.append(str(batch_summary.get("reason") or "当前节点不承载本地结果批次推送。"))

    if worker_runtime.get("running", False):
        info_items.append("当前节点本机 Worker 进程在线。")
    if settings.node_region == "mainland" and settings.node_role == "control" and sync_agent_runtime.get("running", False):
        info_items.append("当前节点本机 sync-agent 在线。")
    if mainland_worker_nodes:
        info_items.append(f"在线大陆 Worker {len(mainland_worker_nodes)} 台。")
    if mainland_control_nodes:
        info_items.append(f"在线大陆 controller {len(mainland_control_nodes)} 台。")

    if blocking_issues:
        status = "blocking"
        summary_text = blocking_issues[0]
    elif warning_issues:
        status = "attention"
        summary_text = warning_issues[0]
    else:
        status = "ready"
        summary_text = "当前多机与跨地域骨架已进入可联调、可持续观察状态。"

    return {
        "status": status,
        "ready": status == "ready",
        "summary": summary_text,
        "blocking_issues": blocking_issues,
        "warnings": warning_issues,
        "info": info_items,
        "cluster": {
            "online_control_nodes": online_control_nodes,
            "online_worker_nodes": online_worker_nodes,
            "mainland_control_nodes": len(mainland_control_nodes),
            "mainland_worker_nodes": len(mainland_worker_nodes),
        },
        "sync": {
            "enabled": bool(sync_summary.get("enabled", False)),
            "push_expected_on_this_node": bool(sync_summary.get("push_expected_on_this_node", False)),
            "source_region": sync_summary.get("source_region", ""),
            "target_region": sync_summary.get("target_region", ""),
            "applicable": batch_applicable,
            "projected_batches": projected_batches,
            "pushing_batches": pushing_batches,
            "failed_batches": failed_batches,
            "synced_batches": synced_batches,
        },
    }


def _detect_participation_snapshot(*, row: dict) -> dict:
    items_running = int(row.get("items_running", 0) or 0)
    items_claimed = int(row.get("items_claimed", 0) or 0)
    active_threads = int(row.get("active_threads", 0) or 0)
    max_threads = int(row.get("max_threads", 0) or 0)
    processed_recent = int(row.get("processed_recent", 0) or 0)
    current_load = int(row.get("current_load", 0) or 0)
    status = str(row.get("status") or "").strip().lower()

    if items_running > 0:
        return {
            "participation_state": "running",
            "participation_label": "执行中",
            "participation_reason": f"当前正在执行 {items_running} 项检测任务。",
            "is_current_participant": True,
            "is_dispatch_active": True,
        }
    if items_claimed > 0:
        return {
            "participation_state": "claimed",
            "participation_label": "已领待跑",
            "participation_reason": f"已领取 {items_claimed} 项任务，等待线程继续执行。",
            "is_current_participant": True,
            "is_dispatch_active": True,
        }
    if active_threads > 0:
        detail = f"当前活跃线程 {active_threads}"
        if max_threads > 0:
            detail = f"{detail}/{max_threads}"
        return {
            "participation_state": "runtime_active",
            "participation_label": "执行中",
            "participation_reason": detail,
            "is_current_participant": True,
            "is_dispatch_active": True,
        }
    if processed_recent > 0:
        return {
            "participation_state": "recent_throughput",
            "participation_label": "近窗有吞吐",
            "participation_reason": f"近 15 分钟内已处理 {processed_recent} 项任务。",
            "is_current_participant": True,
            "is_dispatch_active": False,
        }
    if current_load > 0 or status == "busy":
        load_value = max(current_load, 1)
        return {
            "participation_state": "load_syncing",
            "participation_label": "负载待确认",
            "participation_reason": f"节点当前负载为 {load_value}，但还未观察到已领、执行中或近窗吞吐数据，先归入在线未参与观察。",
            "is_current_participant": False,
            "is_dispatch_active": False,
        }
    return {
        "participation_state": "standby",
        "participation_label": "在线待命",
        "participation_reason": "当前未领任务、未执行任务，也没有近窗吞吐。",
        "is_current_participant": False,
        "is_dispatch_active": False,
    }


def _build_detect_node_row(*, node_code: str, cluster_node: dict, job_node: dict, queue_node: dict) -> dict:
    metadata = cluster_node.get("metadata") or {}
    role = str(cluster_node.get("role") or job_node.get("role") or "worker")
    region = str(cluster_node.get("region") or settings.node_region)
    is_effective_worker = bool(cluster_node.get("is_effective_worker", False) or role == "worker")
    items_total = max(
        int(job_node.get("items_total", 0) or 0),
        int(queue_node.get("items_total", 0) or 0),
        int(metadata.get("job_items_total", 0) or 0),
    )
    items_claimed = max(
        int(job_node.get("items_claimed", 0) or 0),
        int(queue_node.get("items_claimed", 0) or 0),
        int(metadata.get("job_items_claimed", 0) or 0),
    )
    items_running = max(
        int(job_node.get("items_running", 0) or 0),
        int(queue_node.get("items_running", 0) or 0),
        int(metadata.get("job_items_running", 0) or 0),
    )
    items_completed = max(
        int(job_node.get("items_completed", 0) or 0),
        int(queue_node.get("items_completed", 0) or 0),
        int(metadata.get("job_items_completed", 0) or 0),
    )
    items_failed = max(
        int(job_node.get("items_failed", 0) or 0),
        int(queue_node.get("items_failed", 0) or 0),
        int(metadata.get("job_items_failed", 0) or 0),
    )
    items_blacklisted = max(
        int(job_node.get("items_blacklisted", 0) or 0),
        int(queue_node.get("items_blacklisted", 0) or 0),
    )
    active_threads = max(
        int((cluster_node.get("metadata") or {}).get("active_threads", 0) or 0),
        int(job_node.get("active_threads", 0) or 0),
        int(queue_node.get("active_threads", 0) or 0),
    )
    max_threads = max(
        int((cluster_node.get("metadata") or {}).get("max_threads", 0) or 0),
        int(job_node.get("max_threads", 0) or 0),
        int(queue_node.get("max_threads", 0) or 0),
    )
    current_load = max(items_running, active_threads, 0)
    derived_pending = max(items_total - items_claimed - items_running - items_completed - items_failed - items_blacklisted, 0)
    items_pending = max(
        int(job_node.get("items_pending", 0) or 0),
        int(queue_node.get("items_pending", 0) or 0),
        derived_pending,
    )
    row = {
        "node_code": node_code,
        "role": role,
        "region": region,
        "status": str(cluster_node.get("status") or "unknown"),
        "is_effective_worker": is_effective_worker,
        "detect_participating": False,
        "current_load": current_load,
        "items_total": items_total,
        "items_pending": items_pending,
        "items_claimed": items_claimed,
        "items_running": items_running,
        "items_completed": items_completed,
        "items_blacklisted": items_blacklisted,
        "items_failed": items_failed,
        "active_threads": active_threads,
        "max_threads": max_threads,
        "processed_recent": int(queue_node.get("processed_recent", 0) or 0),
        "processed_per_minute": float(queue_node.get("processed_per_minute", 0) or 0),
        "last_heartbeat_at": str(cluster_node.get("last_heartbeat_at") or ""),
        "metrics_source": "active_job" if bool(job_node) else ("cluster_metadata" if metadata else "derived"),
    }
    row.update(_detect_participation_snapshot(row=row))
    row["detect_participating"] = bool(row.get("is_current_participant", False))
    return row


def _build_participating_detect_nodes(*, cluster_snapshot: dict, detect_snapshot: dict, worker_runtime: dict) -> list[dict]:
    cluster_nodes = list(cluster_snapshot.get("nodes") or [])
    active_job = detect_snapshot.get("active_job") or {}
    queue_nodes = list((detect_snapshot.get("queue_health") or {}).get("nodes") or [])

    cluster_map = {
        str(item.get("node_code") or "").strip(): item
        for item in cluster_nodes
        if str(item.get("node_code") or "").strip()
    }
    queue_map = {
        str(item.get("node_code") or "").strip(): item
        for item in queue_nodes
        if str(item.get("node_code") or "").strip()
    }

    merged: list[dict] = []
    job_map = {
        str(item.get("node_code") or "").strip(): item
        for item in list(active_job.get("node_stats") or [])
        if str(item.get("node_code") or "").strip() and str(item.get("node_code") or "").strip() != "unassigned"
    }
    candidate_codes = sorted(set(job_map) | set(queue_map) | set(cluster_map))
    for node_code in candidate_codes:
        cluster_node = cluster_map.get(node_code, {})
        if cluster_node and not bool(cluster_node.get("is_effective_worker", False)):
            continue
        row = _build_detect_node_row(
            node_code=node_code,
            cluster_node=cluster_node,
            job_node=job_map.get(node_code, {}),
            queue_node=queue_map.get(node_code, {}),
        )
        if not bool(row.get("is_current_participant", False)):
            continue
        merged.append(row)

    return sorted(
        merged,
        key=lambda item: (
            -int(item.get("items_running", 0) or 0),
            -int(item.get("items_claimed", 0) or 0),
            -int(item.get("processed_recent", 0) or 0),
            -int(item.get("current_load", 0) or 0),
            str(item.get("node_code") or ""),
        ),
    )


def _build_standby_detect_nodes(*, cluster_snapshot: dict, participating_nodes: list[dict]) -> list[dict]:
    cluster_nodes = list(cluster_snapshot.get("nodes") or [])
    participating_codes = {
        str(item.get("node_code") or "").strip()
        for item in list(participating_nodes or [])
        if str(item.get("node_code") or "").strip()
    }
    standby_rows: list[dict] = []
    for node in cluster_nodes:
        node_code = str(node.get("node_code") or "").strip()
        node_status = str(node.get("status") or "").strip()
        node_current_load = int(node.get("current_load", 0) or 0)
        metadata = node.get("metadata") or {}
        if not node_code:
            continue
        if not bool(node.get("is_effective_worker", False)):
            continue
        if node_status not in {"online", "busy"}:
            continue
        if node_code in participating_codes:
            continue
        standby_state = "load_syncing" if node_current_load > 0 or node_status == "busy" else "standby"
        standby_label = "负载待确认" if standby_state == "load_syncing" else "在线待命"
        standby_reason = (
            str(metadata.get("detail") or "").strip()
            or str(metadata.get("phase_detail") or "").strip()
            or (
                f"当前阶段：{str(metadata.get('phase') or metadata.get('phase_label') or '').strip()}"
                if str(metadata.get("phase") or metadata.get("phase_label") or "").strip()
                else ""
            )
            or (
                f"节点当前负载为 {node_current_load}，但还未观察到已领、执行中或近窗吞吐数据。"
                if standby_state == "load_syncing"
                else "节点在线，当前未领任务、未执行任务，也没有近窗吞吐。"
            )
        )
        standby_rows.append(
            {
                "node_code": node_code,
                "role": str(node.get("role") or "worker"),
                "region": str(node.get("region") or settings.node_region),
                "status": node_status,
                "is_effective_worker": True,
                "detect_participating": False,
                "participation_state": standby_state,
                "participation_label": standby_label,
                "participation_reason": standby_reason,
                "current_load": node_current_load,
                "last_heartbeat_at": str(node.get("last_heartbeat_at") or ""),
                "phase": str(metadata.get("phase") or metadata.get("phase_label") or "").strip(),
                "detail": str(metadata.get("detail") or metadata.get("phase_detail") or "").strip(),
                "worker_online": bool(metadata.get("worker_online", False)),
                "standby_reason": standby_reason,
            }
        )
    return sorted(
        standby_rows,
        key=lambda item: (
            str(item.get("status") or ""),
            str(item.get("role") or ""),
            str(item.get("node_code") or ""),
        ),
    )


def _build_detect_participation_summary(
    *,
    participating_nodes: list[dict],
    standby_nodes: list[dict],
    cluster_snapshot: dict,
) -> dict:
    cluster_nodes = list(cluster_snapshot.get("nodes") or [])
    effective_online_nodes = [
        node
        for node in cluster_nodes
        if bool(node.get("is_effective_worker", False)) and str(node.get("status") or "").strip() in {"online", "busy"}
    ]
    dispatch_active_nodes = [
        row for row in participating_nodes
        if bool(row.get("is_dispatch_active", False))
    ]
    recent_only_nodes = [
        row for row in participating_nodes
        if str(row.get("participation_state") or "").strip() == "recent_throughput"
    ]
    load_syncing_nodes = [
        row for row in standby_nodes
        if str(row.get("participation_state") or "").strip() == "load_syncing"
    ]
    pure_standby_nodes = [
        row for row in standby_nodes
        if str(row.get("participation_state") or "").strip() == "standby"
    ]
    dedicated_worker_nodes = [
        node for node in effective_online_nodes
        if str(node.get("role") or "").strip() == "worker"
    ]
    controller_worker_nodes = [
        node for node in effective_online_nodes
        if str(node.get("role") or "").strip() == "control"
    ]

    summary_parts = [
        f"有效执行节点 {len(effective_online_nodes)} 台",
        f"正在执行/领任务 {len(dispatch_active_nodes)} 台",
        f"近窗刚有吞吐 {len(recent_only_nodes)} 台",
        f"在线但未参与 {len(standby_nodes)} 台",
    ]
    if load_syncing_nodes:
        summary_parts.append(f"其中负载待确认 {len(load_syncing_nodes)} 台")

    return {
        "effective_online_nodes": len(effective_online_nodes),
        "participating_nodes": len(participating_nodes),
        "dispatch_active_nodes": len(dispatch_active_nodes),
        "recent_only_nodes": len(recent_only_nodes),
        "non_participating_nodes": len(standby_nodes),
        "standby_nodes": len(pure_standby_nodes),
        "load_syncing_nodes": len(load_syncing_nodes),
        "dedicated_worker_nodes": len(dedicated_worker_nodes),
        "controller_worker_nodes": len(controller_worker_nodes),
        "dispatch_active_node_codes": [str(item.get("node_code") or "") for item in dispatch_active_nodes],
        "recent_only_node_codes": [str(item.get("node_code") or "") for item in recent_only_nodes],
        "non_participating_node_codes": [str(item.get("node_code") or "") for item in standby_nodes],
        "standby_node_codes": [str(item.get("node_code") or "") for item in pure_standby_nodes],
        "load_syncing_node_codes": [str(item.get("node_code") or "") for item in load_syncing_nodes],
        "summary": "；".join(summary_parts),
    }


def _merge_detect_payload_with_queue_health(detect_payload: dict, queue_health: dict) -> dict:
    normalized_detect = dict(detect_payload or {})
    normalized_queue_health = dict(queue_health or {})
    if not normalized_queue_health.get("has_active_job"):
        return normalized_detect

    queue_payload = dict(normalized_queue_health.get("queue") or {})
    queue_job = dict(normalized_queue_health.get("job") or {})
    queue_nodes = [dict(item) for item in list(normalized_queue_health.get("nodes") or []) if isinstance(item, dict)]
    active_nodes = [
        item
        for item in queue_nodes
        if int(item.get("display_running", item.get("items_running", 0)) or 0) > 0
        or int(item.get("items_claimed", 0) or 0) > 0
        or int(item.get("active_threads", 0) or 0) > 0
    ]
    display_max_threads = sum(int(item.get("max_threads", 0) or 0) for item in active_nodes or queue_nodes)
    normalized_detect["active_job"] = {
        **dict(normalized_detect.get("active_job") or {}),
        "job_id": queue_job.get("job_id"),
        "job_code": queue_job.get("job_code", ""),
        "status": queue_job.get("status", ""),
        "progress_percent": float(queue_job.get("progress_percent", 0) or 0),
        "items_total": int(queue_payload.get("items_total", 0) or 0),
        "items_pending": int(queue_payload.get("pending", 0) or 0),
        "items_claimed": int(queue_payload.get("claimed", 0) or 0),
        "items_running": int(queue_payload.get("running", 0) or 0),
        "items_completed": int(queue_payload.get("completed", 0) or 0),
        "items_blacklisted": int(queue_payload.get("blacklisted", 0) or 0),
        "items_failed": int(queue_payload.get("failed", 0) or 0),
        "display_items_running": int(queue_payload.get("display_running", queue_payload.get("running", 0)) or 0),
        "display_active_threads": int(queue_payload.get("display_running", queue_payload.get("running", 0)) or 0),
        "display_max_threads": int(display_max_threads or 0),
        "node_stats": list(queue_nodes),
        "distributed_node_stats": list(queue_nodes),
    }
    normalized_detect["active_thread_count"] = int(queue_payload.get("display_running", queue_payload.get("running", 0)) or 0)
    normalized_detect["max_thread_count"] = int(display_max_threads or normalized_detect.get("max_thread_count", 0) or 0)
    normalized_detect["aggregate_participating_node_count"] = len(active_nodes)
    normalized_detect["aggregate_participating_node_codes"] = [str(item.get("node_code") or "") for item in active_nodes]
    normalized_detect["aggregate_process_count"] = len(active_nodes)
    normalized_detect["queue_health"] = normalized_queue_health
    return normalized_detect


def _build_runtime_detect_context(*, window_minutes: int = 15) -> dict:
    runtime_settings = get_runtime_settings()
    worker_runtime = detect_worker_runtime()
    worker_expected_on_this_node = not (settings.node_region == "overseas" and settings.node_role == "control")
    inventory_summary = _load_domain_inventory_summary()
    detect_snapshot = get_detect_status()
    latest_run = (detect_snapshot.get("runs") or [None])[0] or {}

    cluster_snapshot = get_cluster_snapshot()
    queue_health = get_detect_queue_health(window_minutes=window_minutes)
    effective_online_worker_nodes = int((cluster_snapshot.get("summary") or {}).get("online_worker_nodes", 0) or 0)
    if effective_online_worker_nodes <= 0 and worker_runtime.get("running", False):
        effective_online_worker_nodes = max(1, worker_runtime.get("process_count", 1) or 1)
    backlog_snapshot = _load_detect_backlog_snapshot()
    remote_backlog_snapshot = _load_latest_remote_runtime_projection_backlog()
    runtime_snapshot_backlog = dict(_load_latest_runtime_active_job_snapshot(window_minutes).get("backlog") or {})
    backlog_snapshot = _merge_backlog_snapshots(backlog_snapshot, remote_backlog_snapshot)
    backlog_snapshot = _merge_backlog_snapshots(backlog_snapshot, runtime_snapshot_backlog)
    capacity_plan = get_detect_capacity_plan(
        queue_health=_align_queue_health_with_backlog(queue_health, backlog_snapshot),
        online_worker_nodes=effective_online_worker_nodes,
        target_finish_hours=6,
    )
    detect_payload = {
        "phase_label": latest_run.get("phase_label", ""),
        "phase_detail": latest_run.get("phase_detail", ""),
        "recent_event": detect_snapshot.get("recent_event", ""),
        "recent_warning": detect_snapshot.get("recent_warning", ""),
        "progress_percent": detect_snapshot.get("progress_percent", 0),
        "progress": detect_snapshot.get("progress", {}),
        "active_thread_count": detect_snapshot.get("active_thread_count", 0),
        "max_thread_count": detect_snapshot.get("max_thread_count", 0),
        "aggregate_process_count": detect_snapshot.get("aggregate_process_count", 0),
        "aggregate_participating_node_count": detect_snapshot.get("aggregate_participating_node_count", 0),
        "aggregate_participating_node_codes": detect_snapshot.get("aggregate_participating_node_codes", []),
        "aggregate_max_thread_count": detect_snapshot.get("aggregate_max_thread_count", 0),
        "aggregate_thread_count_per_process": detect_snapshot.get("aggregate_thread_count_per_process", 0),
        "available_proxy_count": detect_snapshot.get("available_proxy_count", 0),
        "proxy_pool_count": detect_snapshot.get("proxy_pool_count", 0),
        "proxy_runtime_label": detect_snapshot.get("proxy_runtime_label", ""),
        "proxy_runtime_detail": detect_snapshot.get("proxy_runtime_detail", ""),
        "proxy_runtime_reason": detect_snapshot.get("proxy_runtime_reason", ""),
        "proxy_supplier_empty": detect_snapshot.get("proxy_supplier_empty", False),
        "proxy_last_refresh_status": detect_snapshot.get("proxy_last_refresh_status", ""),
        "proxy_last_refresh_time": detect_snapshot.get("proxy_last_refresh_time", ""),
        "proxy_last_refresh_source_count": detect_snapshot.get("proxy_last_refresh_source_count", 0),
        "proxy_last_refresh_total_items": detect_snapshot.get("proxy_last_refresh_total_items", 0),
        "proxy_last_validated_count": detect_snapshot.get("proxy_last_validated_count", 0),
        "proxy_last_available_count": detect_snapshot.get("proxy_last_available_count", 0),
        "proxy_source_stats": detect_snapshot.get("proxy_source_stats", []),
        "dependency_alerts": detect_snapshot.get("dependency_alerts", []),
        "active_job": detect_snapshot.get("active_job"),
        "runs_count": len(detect_snapshot.get("runs") or []),
        "worker_online": worker_runtime.get("running", False),
        "worker_mode": worker_runtime.get("mode", runtime_settings.get("worker_mode", "windows-local")),
        "aggregate_detect_view": bool(detect_snapshot.get("aggregate_detect_view", False)),
        "queue_health": queue_health,
        "backlog": backlog_snapshot,
        "capacity_plan": capacity_plan,
        "log_sync": {
            "enabled": bool(runtime_settings.get("worker_log_sync_enabled", False)),
            "mode": str(runtime_settings.get("worker_log_sync_mode", "key") or "key"),
            "line_count": int(detect_snapshot.get("remote_log_line_count", 0) or 0),
            "source_node_count": int(detect_snapshot.get("remote_log_node_count", 0) or 0),
            "source_nodes": list(detect_snapshot.get("remote_log_nodes") or []),
            "source_node_summaries": list(detect_snapshot.get("remote_log_node_summaries") or []),
            "last_at": str(detect_snapshot.get("remote_log_last_at") or ""),
            "last_line": str(detect_snapshot.get("remote_log_last_line") or ""),
            "preview_lines": list(detect_snapshot.get("remote_log_lines") or [])[-20:],
        },
        "_detect_snapshot": detect_snapshot,
    }
    if settings.node_region == "overseas" and settings.node_role == "control":
        detect_payload = _merge_detect_payload_with_queue_health(detect_payload, queue_health)
    aggregate_remote_detect = bool(detect_payload.get("aggregate_detect_view") and detect_payload.get("active_job"))
    if not worker_expected_on_this_node:
        detect_payload.update(
            {
                "phase_label": (
                    detect_snapshot.get("phase_label")
                    or ("集群执行中" if aggregate_remote_detect else "当前节点不承载")
                ),
                "phase_detail": (
                    "当前节点为海外控制面，仅承载 API 控制与同步接收，不执行本机检测任务。"
                    + (
                        f" 当前汇总执行由 {int(detect_payload.get('aggregate_participating_node_count', 0) or 0)} 台参与节点、"
                        f"{int(detect_payload.get('aggregate_process_count', 0) or 0)} 个检测进程承担。"
                        if aggregate_remote_detect
                        else ""
                    )
                ),
                "recent_warning": "",
                "available_proxy_count": 0,
                "proxy_pool_count": 0,
                "proxy_runtime_label": "不适用",
                "proxy_runtime_detail": "当前节点不承载本机 Worker，代理、线程与检测阶段信息在此节点上不适用。",
                "proxy_runtime_reason": "not_applicable",
                "proxy_supplier_empty": False,
                "proxy_last_refresh_status": "",
                "proxy_last_refresh_time": "",
                "proxy_last_refresh_source_count": 0,
                "proxy_last_refresh_total_items": 0,
                "proxy_last_validated_count": 0,
                "proxy_last_available_count": 0,
                "proxy_source_stats": [],
                "dependency_alerts": [],
                "worker_online": False,
            }
        )
        if not aggregate_remote_detect:
            detect_payload.update(
                {
                    "recent_event": "",
                    "progress_percent": 0,
                    "active_thread_count": 0,
                    "max_thread_count": 0,
                    "active_job": None,
                    "runs_count": 0,
                }
            )
    detect_payload["participating_nodes"] = _build_participating_detect_nodes(
        cluster_snapshot=cluster_snapshot,
        detect_snapshot=detect_payload,
        worker_runtime=worker_runtime,
    )
    detect_payload["standby_nodes"] = _build_standby_detect_nodes(
        cluster_snapshot=cluster_snapshot,
        participating_nodes=detect_payload["participating_nodes"],
    )
    detect_payload["non_participating_nodes"] = list(detect_payload["standby_nodes"] or [])
    detect_payload["participation_summary"] = _build_detect_participation_summary(
        participating_nodes=detect_payload["participating_nodes"],
        standby_nodes=detect_payload["non_participating_nodes"],
        cluster_snapshot=cluster_snapshot,
    )
    detect_payload["observation_summary"] = _build_detect_observation_summary(
        detect_payload=detect_payload,
        cluster_snapshot=cluster_snapshot,
        inventory_summary=inventory_summary,
    )
    return {
        "runtime_settings": runtime_settings,
        "worker_runtime": worker_runtime,
        "worker_expected_on_this_node": worker_expected_on_this_node,
        "detect_payload": detect_payload,
        "cluster_snapshot": cluster_snapshot,
    }


def _build_lightweight_runtime_projection_context() -> dict:
    runtime_settings = get_runtime_settings()
    worker_runtime = detect_worker_runtime()
    active_job = dict(get_active_detect_job_summary(event_limit=10) or {})
    cluster_snapshot = get_cluster_snapshot()

    backlog_snapshot = _load_detect_backlog_snapshot()
    active_job_backlog = _build_backlog_snapshot_from_active_job(active_job)
    if active_job_backlog:
        backlog_snapshot = _merge_backlog_snapshots(backlog_snapshot, active_job_backlog)

    active_job_payload: dict | None = None
    progress_payload = {
        "pending": int(backlog_snapshot.get("pending_total", 0) or 0),
        "running": int(backlog_snapshot.get("running_total", 0) or 0),
        "completed": int(backlog_snapshot.get("completed_total", 0) or 0),
        "blacklisted": int(backlog_snapshot.get("blacklisted_total", 0) or 0),
        "failed": int(backlog_snapshot.get("failed_total", 0) or 0),
    }
    queue_payload = {
        "items_total": sum(progress_payload.values()),
        "pending": progress_payload["pending"],
        "claimed": int(backlog_snapshot.get("claimed_total", 0) or 0),
        "running": progress_payload["running"],
        "display_claimed": int(backlog_snapshot.get("claimed_total", 0) or 0),
        "display_running": progress_payload["running"],
        "completed": progress_payload["completed"],
        "blacklisted": progress_payload["blacklisted"],
        "failed": progress_payload["failed"],
        "terminal": progress_payload["completed"] + progress_payload["blacklisted"] + progress_payload["failed"],
    }

    display_running = progress_payload["running"]
    display_claimed = queue_payload["claimed"]
    display_max_threads = int(worker_runtime.get("thread_count", 0) or runtime_settings.get("thread_count", 0) or 0)
    if active_job:
        display_running = max(
            int(active_job.get("display_items_running", active_job.get("items_running", 0)) or 0),
            int(active_job.get("display_active_threads", 0) or 0),
            progress_payload["running"],
        )
        display_claimed = max(
            int(active_job.get("display_items_claimed", active_job.get("items_claimed", 0)) or 0),
            queue_payload["claimed"],
        )
        display_max_threads = max(
            int(active_job.get("display_max_threads", 0) or 0),
            display_max_threads,
        )
        active_job_payload = {
            "job_id": active_job.get("job_id"),
            "job_code": active_job.get("job_code", ""),
            "status": active_job.get("status", ""),
            "progress_percent": active_job.get("progress_percent", 0),
            "items_total": int(active_job.get("items_total", queue_payload["items_total"]) or queue_payload["items_total"]),
            "items_terminal": int(active_job.get("items_terminal", queue_payload["terminal"]) or queue_payload["terminal"]),
            "items_pending": int(active_job.get("items_pending", queue_payload["pending"]) or queue_payload["pending"]),
            "items_claimed": int(active_job.get("items_claimed", queue_payload["claimed"]) or queue_payload["claimed"]),
            "items_running": int(active_job.get("items_running", queue_payload["running"]) or queue_payload["running"]),
            "items_completed": int(active_job.get("items_completed", queue_payload["completed"]) or queue_payload["completed"]),
            "items_blacklisted": int(active_job.get("items_blacklisted", queue_payload["blacklisted"]) or queue_payload["blacklisted"]),
            "items_failed": int(active_job.get("items_failed", queue_payload["failed"]) or queue_payload["failed"]),
            "display_items_claimed": display_claimed,
            "display_items_running": display_running,
            "display_active_threads": display_running,
            "display_max_threads": display_max_threads,
            "node_stats": list(active_job.get("node_stats") or []),
            "distributed_node_stats": list(active_job.get("distributed_node_stats") or active_job.get("node_stats") or []),
            "step_stats": list(active_job.get("step_stats") or []),
            "raw_step_stats": list(active_job.get("raw_step_stats") or []),
        }

    phase_label = str((active_job_payload or {}).get("status") or "").strip() or (
        "running" if worker_runtime.get("running", False) else "idle"
    )
    phase_detail = ""
    if active_job_payload:
        phase_detail = f"active job {active_job_payload.get('job_code', '')}"
    elif worker_runtime.get("running", False):
        phase_detail = str(worker_runtime.get("message") or "").strip() or "worker running"

    detect_payload = {
        "phase_label": phase_label,
        "phase_detail": phase_detail,
        "recent_event": "",
        "recent_warning": "",
        "progress_percent": float((active_job_payload or {}).get("progress_percent", 0) or 0),
        "progress": progress_payload,
        "active_thread_count": display_running,
        "max_thread_count": display_max_threads,
        "aggregate_process_count": int(worker_runtime.get("process_count", 0) or 0),
        "aggregate_participating_node_count": int((cluster_snapshot.get("summary") or {}).get("online_worker_nodes", 0) or 0),
        "aggregate_participating_node_codes": [],
        "aggregate_max_thread_count": int(worker_runtime.get("max_threads", 0) or display_max_threads or 0),
        "aggregate_thread_count_per_process": int(runtime_settings.get("thread_count", 0) or 0),
        "available_proxy_count": 0,
        "proxy_pool_count": 0,
        "proxy_runtime_label": "",
        "proxy_runtime_detail": "",
        "proxy_runtime_reason": "",
        "proxy_supplier_empty": False,
        "proxy_last_refresh_status": "",
        "proxy_last_refresh_time": "",
        "proxy_last_refresh_source_count": 0,
        "proxy_last_refresh_total_items": 0,
        "proxy_last_validated_count": 0,
        "proxy_last_available_count": 0,
        "proxy_source_stats": [],
        "dependency_alerts": [],
        "active_job": active_job_payload,
        "runs_count": 1 if active_job_payload else 0,
        "worker_online": bool(worker_runtime.get("running", False)),
        "worker_mode": worker_runtime.get("mode", runtime_settings.get("worker_mode", "linux-systemd")),
        "aggregate_detect_view": False,
        "queue_health": {"queue": queue_payload},
        "backlog": backlog_snapshot,
        "capacity_plan": {},
        "log_sync": {"enabled": False, "mode": "", "line_count": 0, "source_node_count": 0, "source_nodes": [], "source_node_summaries": [], "last_at": "", "last_line": "", "preview_lines": []},
    }
    return {
        "detect_payload": detect_payload,
        "cluster_snapshot": cluster_snapshot,
    }


def refresh_runtime_projection_snapshot(*, window_minutes: int = 15) -> dict:
    context = _build_lightweight_runtime_projection_context()
    record_id = append_runtime_projection_if_changed(
        detect=context["detect_payload"],
        cluster=context["cluster_snapshot"],
    )
    return {
        "record_id": record_id,
        "active_thread_count": int((context["detect_payload"] or {}).get("active_thread_count", 0) or 0),
        "max_thread_count": int((context["detect_payload"] or {}).get("max_thread_count", 0) or 0),
        "queue_display_running": int(
            ((((context["detect_payload"] or {}).get("queue_health") or {}).get("queue") or {}).get("display_running", 0) or 0)
        ),
    }


def get_runtime_status() -> dict:
    global _RUNTIME_STATUS_CACHE_EXPIRES_AT, _RUNTIME_STATUS_CACHE_SIGNATURE, _RUNTIME_STATUS_CACHE_VALUE

    now_ts = time.monotonic()
    cache_signature = _runtime_status_cache_signature()
    with _RUNTIME_STATUS_CACHE_LOCK:
        if (
            _RUNTIME_STATUS_CACHE_VALUE is not None
            and now_ts < _RUNTIME_STATUS_CACHE_EXPIRES_AT
            and _RUNTIME_STATUS_CACHE_SIGNATURE == cache_signature
        ):
            return _clone_runtime_status_payload(_RUNTIME_STATUS_CACHE_VALUE)

    context = _build_runtime_detect_context(window_minutes=15)
    runtime_settings = context["runtime_settings"]
    worker_runtime = context["worker_runtime"]
    worker_expected_on_this_node = bool(context["worker_expected_on_this_node"])
    detect_payload = dict(context["detect_payload"] or {})
    cluster_snapshot = dict(context["cluster_snapshot"] or {})
    sync_agent_runtime = detect_sync_agent_runtime()
    api_pid = os.getpid()
    detect_snapshot = dict(detect_payload.pop("_detect_snapshot", {}) or {})
    append_runtime_projection_if_changed(detect=detect_payload, cluster=cluster_snapshot)
    sync_summary = get_sync_summary(record_limit=5)
    readiness = _build_multi_region_readiness(
        cluster_snapshot=cluster_snapshot,
        sync_summary=sync_summary,
        worker_runtime=worker_runtime,
        sync_agent_runtime=sync_agent_runtime,
    )
    build_info = get_runtime_build_info()

    api_payload = {
            "service": "domain-api",
            "version": "0.1.0",
            "api_prefix": settings.api_prefix,
            "pid": api_pid,
            "host": settings.api_host,
            "port": settings.api_port,
            "mode": runtime_settings.get("worker_mode", "windows-local"),
            "service_name": runtime_settings.get("api_service_name", settings.api_service_name),
            "health_url": f"http://127.0.0.1:{settings.api_port}/health",
            "stdout_log": _runtime_log_path("domain-api.stdout.log"),
            "stderr_log": _runtime_log_path("domain-api.stderr.log"),
            "build": build_info,
        }
    node_payload = {
            "code": settings.node_code,
            "region": settings.node_region,
            "role": settings.node_role,
        }
    worker_payload = {
            "mode": worker_runtime.get("mode", runtime_settings.get("worker_mode", "windows-local")),
            "service_name": runtime_settings.get("worker_service_name", settings.worker_service_name),
            "running": worker_runtime.get("running", False),
            "expected_on_this_node": worker_expected_on_this_node,
            "process_count": worker_runtime.get("process_count", 0),
            "latest_start_time": worker_runtime.get("latest_start_time", ""),
            "message": worker_runtime.get("message", ""),
            "log_path": str(Path(settings.domain_root) / "detect_worker.log"),
        }
    sync_agent_payload = {
            "mode": sync_agent_runtime.get("mode", runtime_settings.get("worker_mode", "windows-local")),
            "service_name": runtime_settings.get("sync_agent_service_name", settings.sync_agent_service_name),
            "running": sync_agent_runtime.get("running", False),
            "process_count": sync_agent_runtime.get("process_count", 0),
            "latest_start_time": sync_agent_runtime.get("latest_start_time", ""),
            "message": sync_agent_runtime.get("message", ""),
            "expected_on_this_node": settings.node_region == "mainland" and settings.node_role == "control",
    }
    compatibility_payload = {
        # Backward-compatible flat fields for older pages / stale built assets.
        "api_online": bool(api_payload.get("pid")),
        "api_service_name": api_payload.get("service_name", ""),
        "worker_online": bool(worker_payload.get("running", False)),
        "worker_mode": worker_payload.get("mode", ""),
        "worker_service_name": worker_payload.get("service_name", ""),
        "worker_process_count": worker_payload.get("process_count", 0),
        "worker_latest_start_time": worker_payload.get("latest_start_time", ""),
        "worker_runtime_message": worker_payload.get("message", ""),
        "thread_count": detect_snapshot.get("thread_count", 0),
        "thread_count_default": detect_snapshot.get("thread_count_default", 0),
        "thread_count_source": detect_snapshot.get("thread_count_source", ""),
        "thread_count_override": detect_snapshot.get("thread_count_override"),
        "active_thread_count": detect_payload.get("active_thread_count", 0),
        "max_thread_count": detect_payload.get("max_thread_count", 0),
        "progress": detect_payload.get("progress", {}),
        "backlog": detect_payload.get("backlog", {}),
        "progress_percent": detect_payload.get("progress_percent", 0),
        "available_proxy_count": detect_payload.get("available_proxy_count", 0),
        "proxy_pool_count": detect_payload.get("proxy_pool_count", 0),
        "proxy_runtime_label": detect_payload.get("proxy_runtime_label", ""),
        "proxy_runtime_detail": detect_payload.get("proxy_runtime_detail", ""),
        "proxy_runtime_reason": detect_payload.get("proxy_runtime_reason", ""),
        "proxy_last_refresh_time": detect_payload.get("proxy_last_refresh_time", ""),
        "recent_event": detect_payload.get("recent_event", ""),
        "recent_warning": detect_payload.get("recent_warning", ""),
        "runtime_state": worker_runtime.get("runtime_state") or {},
        "cluster_summary": cluster_snapshot.get("summary") or {},
    }

    result = {
        **compatibility_payload,
        "api": api_payload,
        "node": node_payload,
        "worker": worker_payload,
        "sync_agent": sync_agent_payload,
        "detect": detect_payload,
        "cluster": cluster_snapshot,
        "sync": sync_summary,
        "readiness": readiness,
    }
    with _RUNTIME_STATUS_CACHE_LOCK:
        _RUNTIME_STATUS_CACHE_SIGNATURE = cache_signature
        _RUNTIME_STATUS_CACHE_VALUE = _clone_runtime_status_payload(result)
        _RUNTIME_STATUS_CACHE_EXPIRES_AT = time.monotonic() + _RUNTIME_STATUS_CACHE_TTL_SECONDS
    return result


def get_runtime_preflight() -> dict:
    runtime_settings = get_runtime_settings()
    checks: list[dict[str, object]] = []
    detect_options = read_json("detect_options.json", default={})

    domain_root = Path(settings.domain_root)
    checks.append(
        {
            "key": "domain_root",
            "label": "domainCheck 目录",
            "ok": domain_root.exists(),
            "message": str(domain_root),
        }
    )

    try:
        with get_db() as conn:
            with conn.cursor() as cur:
                cur.execute("select 1")
                cur.fetchone()
        checks.append({"key": "database", "label": "PostgreSQL", "ok": True, "message": f"{settings.db_host}:{settings.db_port}/{settings.db_database}"})
    except Exception as exc:
        checks.append({"key": "database", "label": "PostgreSQL", "ok": False, "message": str(exc)})

    try:
        redis_client = get_redis()
        redis_client.ping()
        checks.append({"key": "redis", "label": "Redis", "ok": True, "message": f"{settings.redis_host}:{settings.redis_port}/{settings.redis_db}"})
    except Exception as exc:
        checks.append({"key": "redis", "label": "Redis", "ok": False, "message": str(exc)})
    else:
        bloom_ok, bloom_message = _bloom_filter_status()
        checks.append(
            {
                "key": "redis_bloom",
                "label": "RedisBloom",
                "ok": True,
                "message": bloom_message,
                "level": "info" if bloom_ok else "warn",
                "degraded": not bloom_ok,
            }
        )

    worker_mode = runtime_settings.get("worker_mode", "windows-local")
    checks.append({"key": "worker_mode", "label": "运行模式", "ok": True, "message": worker_mode})

    jucha_enabled = bool(detect_options.get("detect_jucha"))
    jucha_cookie_ok, jucha_cookie_path = _domain_cookie_status("jucha_cookies.pkl")
    checks.append(
        {
            "key": "detect_jucha",
            "label": "聚查检测",
            "ok": True if not jucha_enabled else jucha_cookie_ok,
            "message": "已启用" if jucha_enabled else "未启用",
        }
    )
    checks.append(
        {
            "key": "jucha_cookie",
            "label": "聚查 Cookie",
            "ok": True if not jucha_enabled else jucha_cookie_ok,
            "message": jucha_cookie_path if jucha_enabled else "未启用聚查检测，无需本地 Cookie",
            "level": "info" if (not jucha_enabled or jucha_cookie_ok) else "warn",
        }
    )

    juziseo_enabled = bool(detect_options.get("detect_juziseo"))
    juziseo_cookie_ok, juziseo_cookie_path = _domain_cookie_status("juziseo_cookies.pkl")
    checks.append(
        {
            "key": "detect_juziseo",
            "label": "桔子SEO检测",
            "ok": True if not juziseo_enabled else juziseo_cookie_ok,
            "message": "已启用" if juziseo_enabled else "未启用",
        }
    )
    checks.append(
        {
            "key": "juziseo_cookie",
            "label": "桔子SEO Cookie",
            "ok": True if not juziseo_enabled else juziseo_cookie_ok,
            "message": juziseo_cookie_path if juziseo_enabled else "未启用桔子SEO检测，无需本地 Cookie",
            "level": "info" if (not juziseo_enabled or juziseo_cookie_ok) else "warn",
        }
    )

    if worker_mode == "linux-systemd":
        checks.append(
            {
                "key": "worker_service_name",
                "label": "Worker service 名",
                "ok": bool(runtime_settings.get("worker_service_name")),
                "message": runtime_settings.get("worker_service_name", ""),
            }
        )
        checks.append(
            {
                "key": "api_service_name",
                "label": "API service 名",
                "ok": bool(runtime_settings.get("api_service_name")),
                "message": runtime_settings.get("api_service_name", ""),
            }
        )
        checks.append(
            {
                "key": "sync_agent_service_name",
                "label": "Sync agent service 名",
                "ok": bool(runtime_settings.get("sync_agent_service_name")),
                "message": runtime_settings.get("sync_agent_service_name", ""),
            }
        )
    else:
        checks.append(
            {
                "key": "windows_scripts",
                "label": "Windows 启停脚本",
                "ok": (Path(settings.domain_root).parent / "start_domain_api.ps1").exists() and (Path(settings.domain_root).parent / "stop_domain_api.ps1").exists(),
                "message": "start_domain_api.ps1 / stop_domain_api.ps1",
            }
        )

    overall_ok = all(bool(item["ok"]) for item in checks)
    return {
        "ok": overall_ok,
        "checks": checks,
    }