getDomain/domain-api/app/services/detect_job_service.py

from __future__ import annotations

import hashlib
import math
import json
import threading
import time
from datetime import datetime, timedelta
from uuid import uuid4

from psycopg2 import errors

from app.core.config import settings
from app.core.db import db_read_retry, get_db, is_retryable_db_error
from app.services.debug_event_service import push_debug_event
from app.services.settings_service import get_settings_payload


ACTIVE_JOB_STATUSES = ("pending", "running")
_RUNTIME_NODE_STALE_MINUTES = 10
_PREFERRED_RUNTIME_SNAPSHOT_GRACE = timedelta(minutes=10)
_LIVE_RUNTIME_NODE_STALE_AFTER = timedelta(seconds=90)
_DEFAULT_TASK_MODE = "domain_pipeline"
_DETECT_JOBS_LEGACY_SELECT_MODE: bool | None = None
_DETECT_JOB_ITEM_RECYCLE_LOCK = threading.Lock()
_DETECT_JOB_ITEM_RECYCLE_LAST_RUN_TS = 0.0
_DETECT_JOB_ITEM_RECYCLE_INTERVAL_SECONDS = 15.0
_DETECT_JOB_ITEM_RECYCLE_ADVISORY_LOCK_KEY = 0
_DETECT_JOB_ITEM_RECYCLE_BATCH_SIZE = 2000
_DETECT_JOB_ITEM_RECYCLE_MAX_BATCHES = 6
_ACTIVE_JOB_SUMMARY_CACHE_LOCK = threading.Lock()
_ACTIVE_JOB_SUMMARY_CACHE_TTL_SECONDS = 3.0
_ACTIVE_JOB_SUMMARY_CACHE: dict[tuple[int], tuple[float, dict | None]] = {}
_DETECT_QUEUE_HEALTH_CACHE_LOCK = threading.Lock()
_DETECT_QUEUE_HEALTH_CACHE_TTL_SECONDS = 3.0
_DETECT_QUEUE_HEALTH_CACHE: dict[tuple[int], tuple[float, dict]] = {}
_DISABLED_MANAGED_NODE_CACHE: set[str] = set()
_DISABLED_MANAGED_NODE_CACHE_EXPIRES_AT = 0.0
_PIPELINE_STEP_ORDER = [
    "detect_register",
    "detect_baidu_site",
    "detect_360_site",
    "detect_chinaz",
    "detect_aizhan",
    "detect_wayback",
    "detect_jucha",
    "detect_juziseo",
]
_PIPELINE_STEP_FIELDS = {
    "detect_baidu_site": "baidu_site",
    "detect_360_site": "qihu360_site",
    "detect_chinaz": "chinaz_info",
    "detect_aizhan": "aizhan_info",
    "detect_wayback": "wayback_info",
    "detect_jucha": "jucha_info",
    "detect_juziseo": "juziseo_info",
}
_PIPELINE_STEP_LABELS = {
    "detect_register": "注册状态检测",
    "detect_baidu_site": "百度site检测",
    "detect_360_site": "360 site检测",
    "detect_chinaz": "站长之家检测",
    "detect_aizhan": "爱站检测",
    "detect_wayback": "时光机检测",
    "detect_jucha": "聚查检测",
    "detect_juziseo": "桔子检测",
}
_PIPELINE_STEP_RETRY_LIMITS = {
    "default": 3,
    "detect_wayback": 2,
}
_SINGLE_STEP_JOB_SPECS = {
    "detect_baidu_site": {
        "step_code": "detect_baidu_site",
        "task_mode": "single_step",
        "label": "百度site检测",
        "source": "api-step-start",
    },
    "detect_wayback": {
        "step_code": "detect_wayback",
        "task_mode": "single_step",
        "label": "时光机检测",
        "source": "api-step-start",
    },
}


def _build_pg_advisory_lock_key(scope: str) -> int:
    normalized_scope = str(scope or "").strip() or "domaincheck-default"
    digest = hashlib.sha1(normalized_scope.encode("utf-8")).digest()
    raw_value = int.from_bytes(digest[:8], "big", signed=False)
    return raw_value - (1 << 64) if raw_value >= (1 << 63) else raw_value


_DETECT_JOB_ITEM_RECYCLE_ADVISORY_LOCK_KEY = _build_pg_advisory_lock_key("detect-job-items-recycle-expired")


def _clone_cacheable_payload(value):
    try:
        return json.loads(json.dumps(value, ensure_ascii=False))
    except Exception:
        if isinstance(value, dict):
            return dict(value)
        if isinstance(value, list):
            return list(value)
        return value


def _selection_sql() -> str:
    return """
    SELECT id
    FROM domains
    WHERE
        detect_status IN (0, 4)
        OR (use_status = 0 AND detect_status = 1 AND register_status = 3 AND expire_date < CURRENT_DATE)
    ORDER BY id ASC
    LIMIT %s
    """


def _refresh_detect_job_status_with_cursor(cur, job_id: int) -> None:
    cur.execute(
        """
        SELECT
            job.task_mode,
            count(*) FILTER (WHERE item.status = 'pending') AS pending_count,
            count(*) FILTER (WHERE item.status IN ('claimed', 'running')) AS dispatch_active_count,
            count(*) FILTER (WHERE item.status = 'failed') AS failed_count,
            count(*) FILTER (WHERE item.status IN ('completed', 'blacklisted')) AS done_count,
            count(*) FILTER (
                WHERE item.status IN ('completed', 'blacklisted', 'failed')
                  AND COALESCE(item.step_code, '') <> ''
                  AND COALESCE(item.result_payload_json->>'controller_processed', 'false') <> 'true'
            ) AS unprocessed_terminal_count
        FROM detect_job_items AS item
        JOIN detect_jobs AS job ON job.id = item.job_id
        WHERE item.job_id = %s
        GROUP BY job.task_mode
        """,
        (int(job_id),),
    )
    refresh_row = cur.fetchone()
    if not refresh_row:
        return
    task_mode = str((refresh_row or [""])[0] or "").strip()
    pending_count = int((refresh_row or ["", 0])[1] or 0)
    dispatch_active_count = int((refresh_row or ["", 0, 0])[2] or 0)
    failed_count = int((refresh_row or ["", 0, 0, 0])[3] or 0)
    done_count = int((refresh_row or ["", 0, 0, 0, 0])[4] or 0)
    unprocessed_terminal_count = int((refresh_row or ["", 0, 0, 0, 0, 0])[5] or 0)
    if dispatch_active_count > 0 or (task_mode == "domain_pipeline" and unprocessed_terminal_count > 0):
        cur.execute(
            """
            UPDATE detect_jobs
            SET status = 'running',
                started_at = COALESCE(started_at, CURRENT_TIMESTAMP),
                finished_at = NULL
            WHERE id = %s
              AND (
                    status <> 'running'
                 OR started_at IS NULL
                 OR finished_at IS NOT NULL
              )
            """,
            (int(job_id),),
        )
        return
    if pending_count > 0:
        cur.execute(
            """
            UPDATE detect_jobs
            SET status = 'pending',
                finished_at = NULL
            WHERE id = %s
            """,
            (int(job_id),),
        )
        return

    final_status = "completed"
    if failed_count > 0 and done_count > 0:
        final_status = "partial_failed"
    elif failed_count > 0:
        final_status = "failed"
    cur.execute(
        """
        UPDATE detect_jobs
        SET status = %s,
            finished_at = CURRENT_TIMESTAMP,
            started_at = COALESCE(started_at, CURRENT_TIMESTAMP)
        WHERE id = %s
        """,
        (final_status, int(job_id)),
    )


def _recycle_expired_detect_job_items_once() -> int:
    conn = None
    advisory_locked = False
    try:
        with get_db() as conn:
            conn.autocommit = False
            with conn.cursor() as cur:
                cur.execute(
                    "SELECT pg_try_advisory_lock(%s)",
                    (_DETECT_JOB_ITEM_RECYCLE_ADVISORY_LOCK_KEY,),
                )
                lock_row = cur.fetchone()
                advisory_locked = bool((lock_row or [False])[0])
                if not advisory_locked:
                    conn.rollback()
                    return 0
                recycled_count = 0
                touched_job_ids: set[int] = set()
                for _ in range(_DETECT_JOB_ITEM_RECYCLE_MAX_BATCHES):
                    cur.execute(
                        """
                        WITH expired_candidates AS (
                            SELECT id, job_id, status
                            FROM detect_job_items
                            WHERE status IN ('claimed', 'running')
                              AND lease_expires_at IS NOT NULL
                              AND lease_expires_at < CURRENT_TIMESTAMP
                            ORDER BY lease_expires_at ASC, id ASC
                            FOR UPDATE SKIP LOCKED
                            LIMIT %s
                        ),
                        recycled AS (
                            UPDATE detect_job_items AS item
                            SET status = 'pending',
                                claimed_by = '',
                                claim_token = '',
                                lease_expires_at = NULL,
                                updated_at = CURRENT_TIMESTAMP,
                                last_error = CASE
                                    WHEN expired_candidates.status = 'running' THEN 'lease expired while running'
                                    WHEN expired_candidates.status = 'claimed' THEN 'lease expired before running'
                                    ELSE item.last_error
                                END
                            FROM expired_candidates
                            WHERE item.id = expired_candidates.id
                            RETURNING expired_candidates.job_id
                        )
                        SELECT count(*), array_remove(array_agg(DISTINCT job_id), NULL)
                        FROM recycled
                        """,
                        (_DETECT_JOB_ITEM_RECYCLE_BATCH_SIZE,),
                    )
                    row = cur.fetchone()
                    batch_count = int((row or [0])[0] or 0)
                    recycled_count += batch_count
                    touched_job_ids.update(
                        int(item)
                        for item in list((row or [0, []])[1] or [])
                        if item is not None
                    )
                    if batch_count < _DETECT_JOB_ITEM_RECYCLE_BATCH_SIZE:
                        break
                for job_id in sorted(touched_job_ids):
                    _refresh_detect_job_status_with_cursor(cur, job_id)
                conn.commit()

        if recycled_count > 0:
            try:
                push_debug_event(
                    service="detect-job",
                    event_type="expired_item_recycled",
                    level="info",
                    message=f"控制面回收过期检测任务项 {recycled_count} 个",
                    payload={
                        "recycled_count": recycled_count,
                        "job_ids": touched_job_ids,
                    },
                )
            except Exception:
                pass
        return recycled_count
    except Exception as exc:
        try:
            if conn is not None:
                conn.rollback()
        except Exception:
            pass
        try:
            push_debug_event(
                service="detect-job",
                event_type="expired_item_recycle_failed",
                level="warning",
                message="控制面回收过期检测任务项失败",
                payload={"error": str(exc or "")[:500]},
            )
        except Exception:
            pass
        return 0
    finally:
        if conn is not None and advisory_locked:
            try:
                with conn.cursor() as cur:
                    cur.execute(
                        "SELECT pg_advisory_unlock(%s)",
                        (_DETECT_JOB_ITEM_RECYCLE_ADVISORY_LOCK_KEY,),
                    )
                conn.commit()
            except Exception:
                try:
                    conn.rollback()
                except Exception:
                    pass


def _maybe_recycle_expired_detect_job_items() -> int:
    global _DETECT_JOB_ITEM_RECYCLE_LAST_RUN_TS

    now_ts = time.monotonic()
    with _DETECT_JOB_ITEM_RECYCLE_LOCK:
        if now_ts - _DETECT_JOB_ITEM_RECYCLE_LAST_RUN_TS < _DETECT_JOB_ITEM_RECYCLE_INTERVAL_SECONDS:
            return 0
        _DETECT_JOB_ITEM_RECYCLE_LAST_RUN_TS = now_ts
    return _recycle_expired_detect_job_items_once()


def _format_time(value: datetime | None) -> str:
    return value.isoformat(sep=" ", timespec="seconds") if value else ""


def _load_disabled_managed_node_codes(node_codes: list[str] | None = None) -> set[str]:
    global _DISABLED_MANAGED_NODE_CACHE, _DISABLED_MANAGED_NODE_CACHE_EXPIRES_AT
    normalized_codes = [
        str(item or "").strip()
        for item in list(node_codes or [])
        if str(item or "").strip()
    ]
    now_ts = time.time()
    if not normalized_codes and now_ts < _DISABLED_MANAGED_NODE_CACHE_EXPIRES_AT:
        return set(_DISABLED_MANAGED_NODE_CACHE)
    try:
        with get_db() as conn:
            with conn.cursor() as cur:
                if normalized_codes:
                    cur.execute(
                        """
                        SELECT node_code
                        FROM ops_managed_nodes
                        WHERE is_enabled = FALSE
                          AND node_code = ANY(%s)
                        """,
                        (normalized_codes,),
                    )
                else:
                    cur.execute(
                        """
                        SELECT node_code
                        FROM ops_managed_nodes
                        WHERE is_enabled = FALSE
                        """
                    )
                rows = list(cur.fetchall() or [])
    except Exception:
        if normalized_codes:
            return {code for code in normalized_codes if code in _DISABLED_MANAGED_NODE_CACHE}
        return set(_DISABLED_MANAGED_NODE_CACHE)

    disabled_codes = {
        str(row[0] or "").strip()
        for row in rows
        if str(row[0] or "").strip()
    }
    if normalized_codes:
        return disabled_codes
    _DISABLED_MANAGED_NODE_CACHE = disabled_codes
    _DISABLED_MANAGED_NODE_CACHE_EXPIRES_AT = now_ts + 5.0
    return set(disabled_codes)


def _detect_jobs_select_columns(*, alias: str = "", legacy_mode: bool = False) -> str:
    prefix = f"{str(alias).strip()}." if str(alias or "").strip() else ""
    task_mode_expr = f"'{_DEFAULT_TASK_MODE}' AS task_mode" if legacy_mode else f"{prefix}task_mode"
    step_code_expr = "'' AS step_code" if legacy_mode else f"{prefix}step_code"
    return (
        f"{prefix}id, {prefix}job_code, {prefix}source, "
        f"{task_mode_expr}, {step_code_expr}, "
        f"{prefix}status, {prefix}created_by, {prefix}created_at, {prefix}started_at, {prefix}finished_at"
    )


def _execute_detect_jobs_select(
    cur,
    *,
    from_clause: str,
    where_clause: str = "",
    order_clause: str = "",
    limit_clause: str = "",
    params: tuple | list | None = None,
    alias: str = "",
) -> None:
    global _DETECT_JOBS_LEGACY_SELECT_MODE

    safe_params = tuple(params or ())
    candidate_modes = (
        [_DETECT_JOBS_LEGACY_SELECT_MODE]
        if _DETECT_JOBS_LEGACY_SELECT_MODE is not None
        else [False, True]
    )
    last_error: Exception | None = None

    for legacy_mode in candidate_modes:
        parts = [
            f"SELECT {_detect_jobs_select_columns(alias=alias, legacy_mode=bool(legacy_mode))}",
            from_clause,
        ]
        if where_clause:
            parts.append(f"WHERE {where_clause}")
        if order_clause:
            parts.append(f"ORDER BY {order_clause}")
        if limit_clause:
            normalized_limit = str(limit_clause or "").strip()
            if normalized_limit:
                parts.append(
                    normalized_limit
                    if normalized_limit.upper().startswith("LIMIT")
                    else f"LIMIT {normalized_limit}"
                )
        sql = "\n".join(parts)
        try:
            cur.execute(sql, safe_params)
            _DETECT_JOBS_LEGACY_SELECT_MODE = bool(legacy_mode)
            return
        except errors.UndefinedColumn as exc:
            last_error = exc
            _DETECT_JOBS_LEGACY_SELECT_MODE = True
            if bool(legacy_mode):
                raise
    if last_error is not None:
        raise last_error


def _normalize_sync_region(value: str | None, fallback: str) -> str:
    text = str(value or "").strip()
    if not text or text == "unknown":
        return str(fallback or "unknown").strip() or "unknown"
    return text


def normalize_detect_step_code(step_code: object) -> str:
    normalized = str(step_code or "").strip()
    if not normalized:
        return ""
    return normalized if normalized in _SINGLE_STEP_JOB_SPECS else ""


def resolve_detect_job_definition(step_code: object = None) -> dict:
    normalized_step_code = normalize_detect_step_code(step_code)
    if normalized_step_code:
        spec = dict(_SINGLE_STEP_JOB_SPECS[normalized_step_code])
        spec["is_single_step"] = True
        return spec
    return {
        "step_code": "",
        "task_mode": _DEFAULT_TASK_MODE,
        "label": "全流程检测",
        "source": "api-start",
        "is_single_step": False,
    }


def _step_label(step_code: str) -> str:
    return _PIPELINE_STEP_LABELS.get(str(step_code or "").strip(), str(step_code or "").strip() or "unknown")


def _json_state(value: object) -> str:
    if isinstance(value, dict):
        return str(value.get("state") or "").strip().lower()
    return ""


def _pipeline_retry_limit(step_code: str) -> int:
    return int(_PIPELINE_STEP_RETRY_LIMITS.get(str(step_code or "").strip(), _PIPELINE_STEP_RETRY_LIMITS["default"]) or 1)


def _classify_pipeline_item_outcome(
    *,
    item_status: str,
    result_payload: dict | None,
    step_code: str,
    attempt_count: int,
) -> dict:
    normalized_item_status = str(item_status or "").strip().lower()
    payload = dict(result_payload or {})
    result_state = str(payload.get("state") or "").strip().lower()
    retry_limit = _pipeline_retry_limit(step_code)
    retry_recommended = bool(payload.get("retry_recommended"))
    legacy_retry_candidate = normalized_item_status == "failed" and not result_state

    if normalized_item_status == "blacklisted" or result_state == "blacklisted":
        return {
            "action": "black_hit",
            "result_state": result_state or "blacklisted",
            "retry_limit": retry_limit,
            "should_retry": False,
            "reason_code": "blacklisted",
        }

    should_retry = (
        result_state in {"degraded", "error"}
        or retry_recommended
        or legacy_retry_candidate
    )
    if should_retry:
        return {
            "action": "retry" if int(attempt_count or 0) < retry_limit else "reject",
            "result_state": result_state or ("degraded" if retry_recommended else "failed"),
            "retry_limit": retry_limit,
            "should_retry": int(attempt_count or 0) < retry_limit,
            "reason_code": "external_retry" if int(attempt_count or 0) < retry_limit else "retry_exhausted",
        }

    if normalized_item_status == "failed" or result_state in {"failed", "rejected", "reject"}:
        return {
            "action": "reject",
            "result_state": result_state or "failed",
            "retry_limit": retry_limit,
            "should_retry": False,
            "reason_code": "business_reject",
        }

    return {
        "action": "pass",
        "result_state": result_state or "passed",
        "retry_limit": retry_limit,
        "should_retry": False,
        "reason_code": "passed",
    }


def _load_pipeline_order(settings_payload: dict | None = None) -> list[str]:
    payload = settings_payload or get_settings_payload()
    detect_options = dict(payload.get("detect_options") or {})
    configured_order = list(detect_options.get("detect_order") or [])
    normalized_order = [item for item in configured_order if item in _PIPELINE_STEP_ORDER]
    for item in _PIPELINE_STEP_ORDER:
        if item not in normalized_order:
            normalized_order.append(item)
    return [item for item in normalized_order if bool(detect_options.get(item, False))]


def order_step_buckets(
    step_items: list[dict] | None,
    *,
    settings_payload: dict | None = None,
    limit: int | None = None,
) -> list[dict]:
    normalized_items = [dict(item) for item in list(step_items or []) if isinstance(item, dict)]
    if not normalized_items:
        return []

    pipeline_order = _load_pipeline_order(settings_payload=settings_payload)
    order_index = {step_code: index for index, step_code in enumerate(pipeline_order)}

    ordered = sorted(
        normalized_items,
        key=lambda item: (
            int(order_index.get(str(item.get("step_code") or "").strip(), 10_000)),
            str(item.get("step_code") or ""),
        ),
    )
    if limit is not None and int(limit or 0) > 0:
        return ordered[: int(limit)]
    return ordered


def _load_domain_pipeline_snapshot(cur, domain_id: int) -> dict | None:
    cur.execute(
        """
        SELECT
            d.id,
            d.domain,
            d.source_type,
            d.register_status,
            d.detect_status,
            d.use_status,
            d.expire_date,
            dd.baidu_site,
            dd.qihu360_site,
            dd.wayback_info,
            dd.chinaz_info,
            dd.aizhan_info,
            dd.jucha_info,
            dd.juziseo_info
        FROM domains AS d
        LEFT JOIN domain_detections AS dd
          ON dd.domain_id = d.id
        WHERE d.id = %s
        LIMIT 1
        """,
        (int(domain_id),),
    )
    row = cur.fetchone()
    if not row:
        return None
    return {
        "id": int(row[0]),
        "domain": row[1] or "",
        "source_type": int(row[2] or 0),
        "register_status": int(row[3] or 0),
        "detect_status": int(row[4] or 0),
        "use_status": int(row[5] or 0),
        "expire_date": _format_time(row[6]) if row[6] else "",
        "baidu_site": _decode_payload(row[7]),
        "qihu360_site": _decode_payload(row[8]),
        "wayback_info": _decode_payload(row[9]),
        "chinaz_info": _decode_payload(row[10]),
        "aizhan_info": _decode_payload(row[11]),
        "jucha_info": _decode_payload(row[12]),
        "juziseo_info": _decode_payload(row[13]),
    }


def _is_pipeline_step_completed(domain_snapshot: dict, step_code: str) -> bool:
    normalized_step_code = str(step_code or "").strip()
    if normalized_step_code == "detect_register":
        return int(domain_snapshot.get("register_status", 0) or 0) not in {0, 10}
    field_name = _PIPELINE_STEP_FIELDS.get(normalized_step_code)
    if not field_name:
        return False
    state = _json_state(domain_snapshot.get(field_name))
    return state in {"passed", "blacklisted"}


def _should_skip_pipeline_step(domain_snapshot: dict, step_code: str) -> bool:
    normalized_step_code = str(step_code or "").strip()
    if normalized_step_code == "detect_register" and int(domain_snapshot.get("source_type", 0) or 0) == 1:
        return True
    return False


def resolve_domain_pipeline_step(domain_snapshot: dict, *, settings_payload: dict | None = None, after_step_code: str | None = None) -> str:
    pipeline_order = _load_pipeline_order(settings_payload=settings_payload)
    if not pipeline_order:
        return ""
    normalized_after = str(after_step_code or "").strip()
    seen_after = not normalized_after
    for step_code in pipeline_order:
        if not seen_after:
            if step_code == normalized_after:
                seen_after = True
            continue
        if _should_skip_pipeline_step(domain_snapshot, step_code):
            continue
        if _is_pipeline_step_completed(domain_snapshot, step_code):
            continue
        return step_code
    return ""


def resolve_initial_domain_pipeline_item(
    domain_snapshot: dict,
    *,
    settings_payload: dict | None = None,
    retry_count: int = 0,
) -> tuple[str, dict | None]:
    step_code = resolve_domain_pipeline_step(domain_snapshot, settings_payload=settings_payload)
    if not step_code:
        return "", None
    return step_code, _build_step_payload(
        step_code=step_code,
        domain_snapshot=domain_snapshot,
        settings_payload=settings_payload,
        retry_count=retry_count,
    )


def _build_step_payload(*, step_code: str, domain_snapshot: dict, settings_payload: dict | None = None, retry_count: int = 0) -> dict:
    pipeline_order = _load_pipeline_order(settings_payload=settings_payload)
    payload = {
        "step_code": str(step_code or "").strip(),
        "step_name": _step_label(step_code),
        "pipeline_order": pipeline_order,
        "retry_count": int(retry_count or 0),
        "domain": str(domain_snapshot.get("domain") or "").strip(),
        "source_type": int(domain_snapshot.get("source_type", 0) or 0),
    }
    if str(step_code or "").strip() == "detect_wayback":
        payload.update(
            {
                "wayback_strategy": "recent_years",
                "wayback_recent_years": 5,
                "wayback_stop_on_first_hit": True,
            }
        )
    return payload


def _decode_payload(value: object) -> dict:
    if isinstance(value, dict):
        return value
    if value in (None, ""):
        return {}
    try:
        return json.loads(value)
    except Exception:
        return {}


def _upsert_domain_detection_field(cur, domain_id: int, field_name: str, payload: dict) -> bool:
    normalized_field_name = str(field_name or "").strip()
    if not normalized_field_name or normalized_field_name not in _PIPELINE_STEP_FIELDS.values():
        return False
    normalized_payload = dict(payload or {})
    cur.execute("SELECT id FROM domain_detections WHERE domain_id = %s LIMIT 1", (int(domain_id),))
    existing = cur.fetchone()
    payload_json = json.dumps(normalized_payload, ensure_ascii=False)
    if existing:
        cur.execute(
            f"""
            UPDATE domain_detections
            SET {normalized_field_name} = %s::jsonb,
                update_time = NOW()
            WHERE domain_id = %s
            """,
            (payload_json, int(domain_id)),
        )
        return True
    cur.execute(
        f"""
        INSERT INTO domain_detections (domain_id, {normalized_field_name}, create_time, update_time)
        VALUES (%s, %s::jsonb, NOW(), NOW())
        """,
        (int(domain_id), payload_json),
    )
    return True


def _int_value(value: object) -> int:
    try:
        return int(value or 0)
    except Exception:
        return 0


def _max_runtime_metric(*values: object) -> int:
    return max((_int_value(value) for value in values), default=0)


def _effective_runtime_load(*, items_running: object, active_threads: object) -> int:
    return max(_int_value(items_running), _int_value(active_threads), 0)


def _control_node_supports_worker(*, region: object, metadata: dict | None) -> bool:
    normalized_region = str(region or "").strip()
    runtime_metadata = dict(metadata or {})
    active_threads = int(runtime_metadata.get("active_threads", 0) or 0)
    max_threads = int(runtime_metadata.get("max_threads", 0) or 0)
    if normalized_region != "mainland":
        return False
    return bool(
        runtime_metadata.get("worker_online", False)
        or runtime_metadata.get("detect_participating", False)
        or active_threads > 0
        or max_threads > 0
    )


def _build_runtime_display_bucket(row: tuple) -> dict | None:
    node_code = str(row[0] or "").strip()
    if not node_code:
        return None
    region = str(row[1] or "").strip()
    role = str(row[2] or "").strip()
    metadata = _decode_payload(row[5])
    if role == "control" and not _control_node_supports_worker(region=region, metadata=metadata):
        return None
    items_total = _int_value(metadata.get("job_items_total"))
    items_claimed = _int_value(metadata.get("job_items_claimed"))
    items_running = _int_value(metadata.get("job_items_running"))
    items_completed = _int_value(metadata.get("job_items_completed"))
    items_failed = _int_value(metadata.get("job_items_failed"))
    raw_current_load = _int_value(row[4])
    active_threads = _int_value(metadata.get("active_threads"))
    max_threads = _int_value(metadata.get("max_threads"))
    current_load = _effective_runtime_load(items_running=items_running, active_threads=active_threads)
    detect_participating = bool(metadata.get("detect_participating", False))
    if (
        settings.node_region == "overseas"
        and settings.node_role == "control"
        and node_code == str(settings.node_code or "").strip()
        and not detect_participating
    ):
        # Overseas control nodes may report their own control-plane load into
        # detect_worker_nodes; that load should never be counted as worker execution.
        return None
    if (
        items_total <= 0
        and items_claimed <= 0
        and items_running <= 0
        and items_completed <= 0
        and items_failed <= 0
        and current_load <= 0
        and raw_current_load <= 0
        and active_threads <= 0
        and not detect_participating
    ):
        return None
    items_pending = max(items_total - items_claimed - items_running - items_completed - items_failed, 0)
    return {
        "node_code": node_code,
        "items_total": items_total,
        "items_pending": items_pending,
        "items_claimed": items_claimed,
        "items_running": items_running,
        "display_running": current_load,
        "items_completed": items_completed,
        "items_blacklisted": 0,
        "items_failed": items_failed,
        "metrics_source": str(metadata.get("service") or "runtime").strip() or "runtime",
        "region": region,
        "role": role,
        "status": str(row[3] or "").strip(),
        "current_load": current_load,
        "active_threads": active_threads,
        "max_threads": max_threads,
        "last_heartbeat_at": _format_time(row[6]) if row[6] else "",
    }


def _merge_display_node_stats(*, local_node_stats: list[dict], runtime_node_rows: list[tuple]) -> list[dict]:
    merged: dict[str, dict] = {}

    def _ensure_bucket(node_code: str) -> dict:
        return merged.setdefault(
            node_code,
            {
                "node_code": node_code,
                "items_total": 0,
                "items_pending": 0,
                "items_claimed": 0,
                "items_running": 0,
                "display_running": 0,
                "items_completed": 0,
                "items_blacklisted": 0,
                "items_failed": 0,
                "metrics_source": "",
                "region": "",
                "role": "",
                "status": "",
                "current_load": 0,
                "active_threads": 0,
                "max_threads": 0,
                "last_heartbeat_at": "",
            },
        )

    for item in list(local_node_stats or []):
        node_code = str(item.get("node_code") or "").strip()
        if not node_code:
            continue
        bucket = _ensure_bucket(node_code)
        for key in (
            "items_total",
            "items_pending",
            "items_claimed",
            "items_running",
            "display_running",
            "items_completed",
            "items_blacklisted",
            "items_failed",
        ):
            bucket[key] = max(_int_value(bucket.get(key)), _int_value(item.get(key)))
        if node_code == "unassigned":
            bucket["metrics_source"] = "central_queue"

    for row in list(runtime_node_rows or []):
        runtime_bucket = _build_runtime_display_bucket(row)
        if not runtime_bucket:
            continue
        node_code = str(runtime_bucket.get("node_code") or "").strip()
        bucket = _ensure_bucket(node_code)
        for key in (
            "items_total",
            "items_pending",
            "items_claimed",
            "items_running",
            "display_running",
            "items_completed",
            "items_blacklisted",
            "items_failed",
            "current_load",
            "active_threads",
            "max_threads",
        ):
            bucket[key] = max(_int_value(bucket.get(key)), _int_value(runtime_bucket.get(key)))
        for key in ("metrics_source", "region", "role", "status", "last_heartbeat_at"):
            if str(runtime_bucket.get(key) or "").strip():
                bucket[key] = runtime_bucket.get(key)

    return sorted(
        merged.values(),
        key=lambda item: (
            str(item.get("node_code") or "") == "unassigned",
            -_int_value(item.get("active_threads")),
            -_int_value(item.get("items_running")),
            -_int_value(item.get("items_claimed")),
            -_int_value(item.get("items_total")),
            str(item.get("node_code") or ""),
        ),
    )


def _load_runtime_display_rows(cur) -> list[tuple]:
    disabled_node_codes = _load_disabled_managed_node_codes()
    cur.execute(
        f"""
        SELECT node_code, region, role, status, current_load, metadata_json, last_heartbeat_at
        FROM detect_worker_nodes
        WHERE last_heartbeat_at >= CURRENT_TIMESTAMP - interval '{_RUNTIME_NODE_STALE_MINUTES} minutes'
        ORDER BY last_heartbeat_at DESC, node_code ASC
        """
    )
    return [
        row
        for row in list(cur.fetchall() or [])
        if str((row or [""])[0] or "").strip() not in disabled_node_codes
    ]


def _build_display_summary(node_stats: list[dict]) -> dict:
    effective_nodes = [
        item
        for item in list(node_stats or [])
        if str(item.get("node_code") or "").strip()
        and str(item.get("node_code") or "").strip() != "unassigned"
        and not (
            settings.node_region == "overseas"
            and settings.node_role == "control"
            and str(item.get("node_code") or "").strip() == str(settings.node_code or "").strip()
        )
    ]
    return {
        "items_claimed": sum(_int_value(item.get("items_claimed")) for item in effective_nodes),
        "items_running": sum(_int_value(item.get("items_running")) for item in effective_nodes),
        "display_running": sum(
            _effective_runtime_load(
                items_running=item.get("items_running"),
                active_threads=item.get("active_threads"),
            )
            for item in effective_nodes
        ),
        "current_load": sum(
            _effective_runtime_load(
                items_running=item.get("items_running"),
                active_threads=item.get("active_threads"),
            )
            for item in effective_nodes
        ),
        "active_threads": sum(_int_value(item.get("active_threads")) for item in effective_nodes),
        "max_threads": sum(_int_value(item.get("max_threads")) for item in effective_nodes),
        "items_completed": sum(_int_value(item.get("items_completed")) for item in effective_nodes),
        "items_failed": sum(_int_value(item.get("items_failed")) for item in effective_nodes),
        "active_nodes": [
            str(item.get("node_code") or "").strip()
            for item in effective_nodes
            if _effective_runtime_load(
                items_running=item.get("items_running"),
                active_threads=item.get("active_threads"),
            ) > 0
        ],
    }


def _normalize_node_bucket(item: dict) -> dict:
    node_code = str(item.get("node_code") or "").strip()
    items_running = _int_value(item.get("items_running"))
    active_threads = _int_value(item.get("active_threads"))
    max_threads = _int_value(item.get("max_threads"))
    current_load = _effective_runtime_load(items_running=items_running, active_threads=active_threads)
    return {
        "node_code": node_code,
        "items_total": _int_value(item.get("items_total")),
        "items_pending": _int_value(item.get("items_pending")),
        "items_claimed": _int_value(item.get("items_claimed")),
        "items_running": items_running,
        "display_running": current_load,
        "items_completed": _int_value(item.get("items_completed")),
        "items_blacklisted": _int_value(item.get("items_blacklisted")),
        "items_failed": _int_value(item.get("items_failed")),
        "metrics_source": str(item.get("metrics_source") or "").strip(),
        "region": str(item.get("region") or "").strip(),
        "role": str(item.get("role") or "").strip(),
        "status": str(item.get("status") or "").strip(),
        "current_load": current_load,
        "active_threads": active_threads,
        "max_threads": max_threads,
        "last_heartbeat_at": str(item.get("last_heartbeat_at") or "").strip(),
    }


def _build_effective_node_stats(
    *,
    distributed_node_stats: list[dict],
    raw_items_total: int,
) -> list[dict]:
    normalized_total = max(0, _int_value(raw_items_total))
    assigned_buckets: list[dict] = []
    assigned_total = 0
    unassigned_bucket: dict | None = None
    for item in list(distributed_node_stats or []):
        bucket = _normalize_node_bucket(item)
        if not bucket["node_code"]:
            continue
        if bucket["node_code"] == "unassigned":
            unassigned_bucket = bucket
            continue
        if (
            settings.node_region == "overseas"
            and settings.node_role == "control"
            and bucket["node_code"] == str(settings.node_code or "").strip()
        ):
            continue
        assigned_buckets.append(bucket)
        assigned_total += bucket["items_total"]

    if unassigned_bucket and unassigned_bucket["items_total"] > 0:
        assigned_buckets.append(unassigned_bucket)
        assigned_total += unassigned_bucket["items_total"]

    effective_total = max(normalized_total, assigned_total)
    remainder_total = max(effective_total - assigned_total, 0)
    if remainder_total > 0:
        if unassigned_bucket:
            unassigned_bucket["items_total"] += remainder_total
            unassigned_bucket["items_pending"] += remainder_total
            assigned_buckets = [
                unassigned_bucket if str(item.get("node_code") or "").strip() == "unassigned" else item
                for item in assigned_buckets
            ]
        else:
            assigned_buckets.append(
                {
                    "node_code": "unassigned",
                    "items_total": remainder_total,
                    "items_pending": remainder_total,
                    "items_claimed": 0,
                    "items_running": 0,
                    "display_running": 0,
                    "items_completed": 0,
                    "items_blacklisted": 0,
                    "items_failed": 0,
                    "metrics_source": "central_queue",
                    "region": "",
                    "role": "",
                    "status": "",
                    "current_load": 0,
                    "active_threads": 0,
                    "max_threads": 0,
                    "last_heartbeat_at": "",
                }
            )

    return sorted(
        assigned_buckets,
        key=lambda item: (
            str(item.get("node_code") or "") == "unassigned",
            -_int_value(item.get("active_threads")),
            -_int_value(item.get("items_running")),
            -_int_value(item.get("items_claimed")),
            -_int_value(item.get("items_total")),
            str(item.get("node_code") or ""),
        ),
    )


def _build_effective_summary(
    *,
    node_stats: list[dict],
    raw_items_total: int,
    raw_items_pending: int,
    raw_items_claimed: int,
    raw_items_running: int,
    raw_items_completed: int,
    raw_items_blacklisted: int,
    raw_items_failed: int,
) -> dict:
    effective_total = max(0, _int_value(raw_items_total))
    effective_pending = max(0, _int_value(raw_items_pending))
    effective_claimed = max(0, _int_value(raw_items_claimed))
    effective_running = max(0, _int_value(raw_items_running))
    effective_completed = max(0, _int_value(raw_items_completed))
    effective_blacklisted = max(0, _int_value(raw_items_blacklisted))
    effective_failed = max(0, _int_value(raw_items_failed))
    effective_terminal = effective_completed + effective_blacklisted + effective_failed
    return {
        "items_total": effective_total,
        "items_pending": effective_pending,
        "items_claimed": effective_claimed,
        "items_running": effective_running,
        "items_completed": effective_completed,
        "items_blacklisted": effective_blacklisted,
        "items_failed": effective_failed,
        "items_terminal": max(effective_terminal, 0),
    }


def _build_step_bucket(step_code: str) -> dict:
    normalized_step_code = str(step_code or "").strip()
    return {
        "step_code": normalized_step_code,
        "step_name": _step_label(normalized_step_code),
        "items_total": 0,
        "items_pending": 0,
        "items_claimed": 0,
        "items_running": 0,
        "items_completed": 0,
        "items_blacklisted": 0,
        "items_failed": 0,
        "processed_recent": 0,
        "processed_per_minute": 0,
        "started_recent": 0,
        "completed_recent": 0,
        "blacklisted_recent": 0,
        "failed_recent": 0,
    }


def _extract_current_cycle_events(events: list[dict]) -> tuple[str, list[dict]]:
    if not events:
        return "", []

    latest_payload = events[0].get("payload") or {}
    latest_cycle_token = str(latest_payload.get("cycle_token") or "").strip()
    if latest_cycle_token:
        current_cycle = []
        for event in events:
            payload = event.get("payload") or {}
            event_cycle = str(payload.get("cycle_token") or "").strip()
            if event_cycle == latest_cycle_token:
                current_cycle.append(event)
        return latest_cycle_token, current_cycle

    cycle_token = ""
    anchor_index = -1
    for index, event in enumerate(events):
        payload = event.get("payload") or {}
        event_type = str(event.get("event_type") or "")
        candidate = str(payload.get("cycle_token") or "").strip()
        if candidate and event_type in {"job_dispatch_sent", "job_dispatch_requested", "job_dispatch_failed", "job_dispatch_rejected"}:
            cycle_token = candidate
            anchor_index = index
            break
    if not cycle_token:
        return "", events
    current_cycle = []
    for index, event in enumerate(events[: anchor_index + 1]):
        payload = event.get("payload") or {}
        event_cycle = str(payload.get("cycle_token") or "").strip()
        if index == anchor_index or event_cycle == cycle_token:
            current_cycle.append(event)
    return cycle_token, current_cycle


def _classify_runtime_debug_event(*, event_type: str, message: str, payload: dict) -> dict:
    normalized_event_type = str(event_type or "").strip()
    normalized_message = str(message or "").strip()
    normalized_payload = payload if isinstance(payload, dict) else {}
    normalized_job_code = str(normalized_payload.get("job_code") or "").strip()
    normalized_node_code = str(normalized_payload.get("node_code") or "").strip()
    result = {
        "job_code": normalized_job_code,
        "node_code": normalized_node_code,
        "started": False,
        "terminal": False,
        "terminal_status": "",
        "step_code": "",
    }

    if normalized_event_type in {"domain_started", "domain_completed", "domain_failed", "domain_blacklisted"}:
        result["started"] = normalized_event_type == "domain_started"
        result["terminal"] = normalized_event_type in {"domain_completed", "domain_failed", "domain_blacklisted"}
        terminal_map = {
            "domain_completed": "completed",
            "domain_failed": "failed",
            "domain_blacklisted": "blacklisted",
        }
        result["terminal_status"] = terminal_map.get(normalized_event_type, "")
        result["step_code"] = normalize_detect_step_code(normalized_payload.get("detect_key")) or "detect_register"
        return result

    if normalized_event_type != "worker_log":
        return result

    step_markers = (
        ("detect_register", ("| step=注册状态检测 |", "detect_register")),
        ("detect_baidu_site", ("| step=百度site检测 |", "detect_baidu_site", "百度阶段", "detect.baidu")),
        ("detect_360_site", ("| step=360收录检测 |", "detect_360_site", "360阶段", "detect.c360")),
        ("detect_chinaz", ("| step=站长之家检测 |", "detect_chinaz", "站长之家", "detect.chinaz")),
        ("detect_aizhan", ("| step=爱站检测 |", "detect_aizhan", "爱站", "detect.aizhan")),
        ("detect_wayback", ("| step=时光机检测 |", "detect_wayback", "时光机检测")),
        ("detect_jucha", ("| step=聚查检测 |", "detect_jucha", "聚查", "detect.jucha")),
        ("detect_juziseo", ("| step=桔子检测 |", "detect_juziseo", "桔子", "juziseo")),
    )
    for step_code, markers in step_markers:
        if any(marker in normalized_message for marker in markers):
            result["step_code"] = step_code
            break

    if "| stage=started |" in normalized_message or "开始检测域名:" in normalized_message:
        result["started"] = True
        return result

    if "| stage=single_step_finalized |" in normalized_message:
        result["terminal"] = True
        if "final_status=completed" in normalized_message:
            result["terminal_status"] = "completed"
        elif "final_status=failed" in normalized_message:
            result["terminal_status"] = "failed"
        elif "final_status=blacklisted" in normalized_message:
            result["terminal_status"] = "blacklisted"
        return result

    return result


def _load_runtime_activity_snapshot(window_minutes: int) -> dict:
    safe_window_minutes = max(5, min(int(window_minutes or 15), 120))
    snapshot = {
        "processed_recent": 0,
        "completed_recent": 0,
        "failed_recent": 0,
        "blacklisted_recent": 0,
        "focus_job_code": "",
        "job_codes": [],
        "step_code": "detect_register",
        "step_stats": {},
        "nodes": {},
    }
    if not (settings.node_region == "overseas" and settings.node_role == "control"):
        return snapshot

    if settings.node_region == "overseas" and settings.node_role == "control":
        source_region = "mainland"
    else:
        source_region = _normalize_sync_region(settings.sync_source_region, settings.node_region)
    with get_db() as conn:
        with conn.cursor() as cur:
            cur.execute(
                """
                SELECT
                    COALESCE(NULLIF(node_code, ''), 'unknown') AS node_code,
                    event_type,
                    message,
                    payload_json,
                    created_at
                FROM detect_debug_events
                WHERE source_region = %s
                  AND created_at >= CURRENT_TIMESTAMP - (%s || ' minutes')::interval
                  AND event_type IN ('worker_log', 'domain_started', 'domain_completed', 'domain_failed', 'domain_blacklisted')
                ORDER BY created_at DESC, id DESC
                LIMIT 2000
                """,
                (source_region, safe_window_minutes),
            )
            rows = cur.fetchall()

    snapshot_job_codes = _load_recent_runtime_snapshot_job_codes(safe_window_minutes)
    job_activity: dict[str, dict] = {}
    overall_totals = {
        "processed_recent": 0,
        "completed_recent": 0,
        "failed_recent": 0,
        "blacklisted_recent": 0,
    }
    overall_node_stats: dict[str, dict] = {}
    overall_step_activity: dict[str, dict] = {}
    per_job_totals: dict[str, dict] = {}
    per_job_node_stats: dict[str, dict[str, dict]] = {}
    per_job_step_activity: dict[str, dict[str, dict]] = {}
    for row in rows:
        node_code = str(row[0] or "unknown").strip() or "unknown"
        event_type = str(row[1] or "").strip()
        message = str(row[2] or "").strip()
        payload = _decode_payload(row[3])
        created_at = _format_time(row[4])
        classified = _classify_runtime_debug_event(event_type=event_type, message=message, payload=payload)
        job_code = str(classified.get("job_code") or "").strip()
        step_code = str(classified.get("step_code") or "").strip() or "detect_register"

        if job_code:
            bucket = job_activity.setdefault(
                job_code,
                {
                    "job_code": job_code,
                    "activity_count": 0,
                    "started_count": 0,
                    "terminal_count": 0,
                    "latest_at": "",
                },
            )
            bucket["activity_count"] += 1
            if classified.get("started"):
                bucket["started_count"] += 1
            if classified.get("terminal"):
                bucket["terminal_count"] += 1
            if created_at > str(bucket.get("latest_at") or ""):
                bucket["latest_at"] = created_at

        if classified.get("terminal"):
            overall_totals["processed_recent"] += 1
            step_bucket = overall_step_activity.setdefault(
                step_code,
                {
                    "step_code": step_code,
                    "started_recent": 0,
                    "processed_recent": 0,
                    "completed_recent": 0,
                    "failed_recent": 0,
                    "blacklisted_recent": 0,
                },
            )
            step_bucket["processed_recent"] += 1
            node_bucket = overall_node_stats.setdefault(
                node_code,
                {
                    "node_code": node_code,
                    "processed_recent": 0,
                    "completed_recent": 0,
                    "failed_recent": 0,
                    "blacklisted_recent": 0,
                },
            )
            node_bucket["processed_recent"] += 1
            if job_code:
                job_totals = per_job_totals.setdefault(
                    job_code,
                    {
                        "processed_recent": 0,
                        "completed_recent": 0,
                        "failed_recent": 0,
                        "blacklisted_recent": 0,
                    },
                )
                job_totals["processed_recent"] += 1
                job_step_bucket = per_job_step_activity.setdefault(job_code, {}).setdefault(
                    step_code,
                    {
                        "step_code": step_code,
                        "started_recent": 0,
                        "processed_recent": 0,
                        "completed_recent": 0,
                        "failed_recent": 0,
                        "blacklisted_recent": 0,
                    },
                )
                job_step_bucket["processed_recent"] += 1
                job_node_bucket = per_job_node_stats.setdefault(job_code, {}).setdefault(
                    node_code,
                    {
                        "node_code": node_code,
                        "processed_recent": 0,
                        "completed_recent": 0,
                        "failed_recent": 0,
                        "blacklisted_recent": 0,
                    },
                )
                job_node_bucket["processed_recent"] += 1
            terminal_status = str(classified.get("terminal_status") or "").strip()
            if terminal_status == "completed":
                overall_totals["completed_recent"] += 1
                node_bucket["completed_recent"] += 1
                step_bucket["completed_recent"] += 1
                if job_code:
                    job_totals["completed_recent"] += 1
                    job_node_bucket["completed_recent"] += 1
                    job_step_bucket["completed_recent"] += 1
            elif terminal_status == "failed":
                overall_totals["failed_recent"] += 1
                node_bucket["failed_recent"] += 1
                step_bucket["failed_recent"] += 1
                if job_code:
                    job_totals["failed_recent"] += 1
                    job_node_bucket["failed_recent"] += 1
                    job_step_bucket["failed_recent"] += 1
            elif terminal_status == "blacklisted":
                overall_totals["blacklisted_recent"] += 1
                node_bucket["blacklisted_recent"] += 1
                step_bucket["blacklisted_recent"] += 1
                if job_code:
                    job_totals["blacklisted_recent"] += 1
                    job_node_bucket["blacklisted_recent"] += 1
                    job_step_bucket["blacklisted_recent"] += 1

        if classified.get("started"):
            step_bucket = overall_step_activity.setdefault(
                step_code,
                {
                    "step_code": step_code,
                    "started_recent": 0,
                    "processed_recent": 0,
                    "completed_recent": 0,
                    "failed_recent": 0,
                    "blacklisted_recent": 0,
                },
            )
            step_bucket["started_recent"] += 1
            if job_code:
                job_step_bucket = per_job_step_activity.setdefault(job_code, {}).setdefault(
                    step_code,
                    {
                        "step_code": step_code,
                        "started_recent": 0,
                        "processed_recent": 0,
                        "completed_recent": 0,
                        "failed_recent": 0,
                        "blacklisted_recent": 0,
                    },
                )
                job_step_bucket["started_recent"] += 1

    ordered_jobs = []
    if job_activity:
        ordered_jobs = sorted(
            job_activity.values(),
            key=lambda item: (
                -int(item.get("terminal_count", 0) or 0),
                -int(item.get("activity_count", 0) or 0),
                str(item.get("latest_at") or ""),
                str(item.get("job_code") or ""),
            ),
        )
    ordered_activity_codes = [
        str(item.get("job_code") or "").strip()
        for item in ordered_jobs
        if str(item.get("job_code") or "").strip()
    ]
    merged_job_codes: list[str] = []
    seen_job_codes: set[str] = set()
    for job_code in [*snapshot_job_codes, *ordered_activity_codes]:
        normalized_job_code = str(job_code or "").strip()
        if not normalized_job_code or normalized_job_code in seen_job_codes:
            continue
        merged_job_codes.append(normalized_job_code)
        seen_job_codes.add(normalized_job_code)
    focus_job_code = merged_job_codes[0] if merged_job_codes else ""
    if focus_job_code:
        snapshot["focus_job_code"] = focus_job_code
        snapshot["job_codes"] = merged_job_codes[:5]
    elif ordered_activity_codes:
        snapshot["focus_job_code"] = ordered_activity_codes[0]
        snapshot["job_codes"] = ordered_activity_codes[:5]

    selected_totals = overall_totals
    selected_step_activity = overall_step_activity
    selected_node_stats = overall_node_stats
    if focus_job_code:
        if focus_job_code in per_job_totals:
            selected_totals = dict(per_job_totals.get(focus_job_code) or {})
        if focus_job_code in per_job_step_activity:
            selected_step_activity = dict(per_job_step_activity.get(focus_job_code) or {})
        if focus_job_code in per_job_node_stats:
            selected_node_stats = dict(per_job_node_stats.get(focus_job_code) or {})

    snapshot["processed_recent"] = int(selected_totals.get("processed_recent", 0) or 0)
    snapshot["completed_recent"] = int(selected_totals.get("completed_recent", 0) or 0)
    snapshot["failed_recent"] = int(selected_totals.get("failed_recent", 0) or 0)
    snapshot["blacklisted_recent"] = int(selected_totals.get("blacklisted_recent", 0) or 0)

    if selected_step_activity:
        ordered_steps = sorted(
            selected_step_activity.values(),
            key=lambda item: (
                -int(item.get("processed_recent", 0) or 0),
                -int(item.get("started_recent", 0) or 0),
                str(item.get("step_code") or ""),
            ),
        )
        snapshot["step_code"] = str((ordered_steps[0] or {}).get("step_code") or "detect_register")
        snapshot["step_stats"] = {
            str(item.get("step_code") or ""): {
                "step_code": str(item.get("step_code") or ""),
                "started_recent": int(item.get("started_recent", 0) or 0),
                "processed_recent": int(item.get("processed_recent", 0) or 0),
                "completed_recent": int(item.get("completed_recent", 0) or 0),
                "failed_recent": int(item.get("failed_recent", 0) or 0),
                "blacklisted_recent": int(item.get("blacklisted_recent", 0) or 0),
            }
            for item in ordered_steps
            if str(item.get("step_code") or "").strip()
        }
    snapshot["nodes"] = selected_node_stats
    return snapshot


def _load_recent_runtime_snapshot_job_codes(window_minutes: int, *, limit: int = 5) -> list[str]:
    safe_window_minutes = max(5, min(int(window_minutes or 15), 12 * 60))
    safe_limit = max(1, min(int(limit or 5), 10))
    if not (settings.node_region == "overseas" and settings.node_role == "control"):
        return []

    entries: list[tuple[datetime, str]] = []
    with get_db() as conn:
        with conn.cursor() as cur:
            cur.execute(
                """
                SELECT payload_json, created_at
                FROM detect_debug_events
                WHERE source_region = 'mainland'
                  AND service = 'detect-runtime'
                  AND event_type = 'active_job_snapshot'
                  AND created_at >= CURRENT_TIMESTAMP - (%s || ' minutes')::interval
                ORDER BY created_at DESC, id DESC
                LIMIT 12
                """,
                (safe_window_minutes,),
            )
            for payload_json, created_at in list(cur.fetchall() or []):
                payload = _decode_payload(payload_json)
                job_code = str((payload.get("job") or {}).get("job_code") or "").strip() if isinstance(payload, dict) else ""
                if job_code and created_at:
                    entries.append((created_at, job_code))

            cur.execute(
                """
                SELECT payload_json, COALESCE(updated_at, created_at)
                FROM detect_sync_records
                WHERE sync_type = 'runtime_ingest'
                  AND source_region = 'mainland'
                  AND target_region = 'overseas'
                  AND created_at >= CURRENT_TIMESTAMP - (%s || ' minutes')::interval
                ORDER BY COALESCE(updated_at, created_at) DESC, id DESC
                LIMIT 12
                """,
                (safe_window_minutes,),
            )
            for payload_json, created_at in list(cur.fetchall() or []):
                payload = _decode_payload(payload_json)
                projection = payload.get("projection") if isinstance(payload, dict) else {}
                job_code = str((projection.get("active_job") or {}).get("job_code") or "").strip() if isinstance(projection, dict) else ""
                if job_code and created_at:
                    entries.append((created_at, job_code))

    codes: list[str] = []
    seen_codes: set[str] = set()
    for _created_at, job_code in sorted(entries, key=lambda item: item[0], reverse=True):
        if job_code in seen_codes:
            continue
        codes.append(job_code)
        seen_codes.add(job_code)
        if len(codes) >= safe_limit:
            break
    return codes


def _load_latest_runtime_debug_active_job_snapshot(
    window_minutes: int,
    *,
    preferred_job_codes: list[str] | None = None,
) -> dict:
    safe_window_minutes = max(5, min(int(window_minutes or 15), 120))
    if not (settings.node_region == "overseas" and settings.node_role == "control"):
        return {}

    preferred = {
        str(item or "").strip()
        for item in list(preferred_job_codes or [])
        if str(item or "").strip()
    }
    source_region = "mainland"
    with get_db() as conn:
        with conn.cursor() as cur:
            cur.execute(
                """
                SELECT payload_json, created_at
                FROM detect_debug_events
                WHERE source_region = %s
                  AND service = 'detect-runtime'
                  AND event_type = 'active_job_snapshot'
                  AND created_at >= CURRENT_TIMESTAMP - (%s || ' minutes')::interval
                ORDER BY created_at DESC, id DESC
                LIMIT 12
                """,
                (source_region, safe_window_minutes),
            )
            rows = list(cur.fetchall() or [])

    fallback_snapshot: dict = {}
    for row in rows:
        payload = _decode_payload(row[0])
        if not isinstance(payload, dict):
            continue
        snapshot = dict(payload)
        snapshot["_created_at"] = _format_time(row[1])
        snapshot["_snapshot_source"] = "active_job_snapshot"
        if not fallback_snapshot:
            fallback_snapshot = snapshot
        job_code = str((snapshot.get("job") or {}).get("job_code") or "").strip()
        if preferred and job_code in preferred:
            return snapshot
    return fallback_snapshot


def _normalize_runtime_projection_nodes(
    *,
    active_job: dict,
    progress: dict,
    projection: dict,
) -> list[dict]:
    node_info = dict(projection.get("node") or {})
    fallback_node_code = str(node_info.get("node_code") or "").strip()
    fallback_region = str(node_info.get("region") or "").strip()
    fallback_role = str(node_info.get("role") or "").strip()
    active_thread_count = _int_value(projection.get("active_thread_count"))
    max_thread_count = _int_value(projection.get("max_thread_count"))

    nodes: list[dict] = []
    for raw_item in list(active_job.get("distributed_node_stats") or active_job.get("node_stats") or []):
        if not isinstance(raw_item, dict):
            continue
        node_code = str(raw_item.get("node_code") or "").strip()
        if not node_code:
            continue
        nodes.append(
            _normalize_node_bucket(
                {
                    **raw_item,
                    "node_code": node_code,
                    "metrics_source": str(raw_item.get("metrics_source") or "runtime_ingest").strip() or "runtime_ingest",
                    "region": str(raw_item.get("region") or (fallback_region if node_code == fallback_node_code else "")).strip(),
                    "role": str(raw_item.get("role") or (fallback_role if node_code == fallback_node_code else "")).strip(),
                    "status": str(raw_item.get("status") or "").strip(),
                }
            )
        )

    if fallback_node_code:
        matched = False
        for item in nodes:
            if str(item.get("node_code") or "").strip() != fallback_node_code:
                continue
            matched = True
            item["region"] = str(item.get("region") or fallback_region).strip()
            item["role"] = str(item.get("role") or fallback_role).strip()
            item["active_threads"] = max(_int_value(item.get("active_threads")), active_thread_count)
            item["max_threads"] = max(_int_value(item.get("max_threads")), max_thread_count)
            item["current_load"] = _effective_runtime_load(
                items_running=item.get("items_running"),
                active_threads=item.get("active_threads"),
            )
            item["display_running"] = item["current_load"]
            if not str(item.get("status") or "").strip():
                item["status"] = "busy" if item["current_load"] > 0 else "online"
            break
        has_child_instances = any(
            str(item.get("node_code") or "").strip().startswith(f"{fallback_node_code}-")
            for item in nodes
        )
        if not matched and not has_child_instances and (
            int(active_job.get("items_total", 0) or 0) > 0
            or active_thread_count > 0
            or max_thread_count > 0
        ):
            nodes.append(
                _normalize_node_bucket(
                    {
                        "node_code": fallback_node_code,
                        "items_total": int(active_job.get("items_total", 0) or 0),
                        "items_pending": int(progress.get("pending", 0) or 0),
                        "items_claimed": max(
                            0,
                            int(active_job.get("items_total", 0) or 0)
                            - int(progress.get("pending", 0) or 0)
                            - int(progress.get("running", 0) or 0)
                            - int(active_job.get("items_terminal", 0) or 0),
                        ),
                        "items_running": int(progress.get("running", 0) or 0),
                        "items_completed": int(progress.get("completed", 0) or 0),
                        "items_blacklisted": int(progress.get("blacklisted", 0) or 0),
                        "items_failed": max(
                            int(progress.get("failed", 0) or 0),
                            int(active_job.get("items_failed", 0) or 0),
                        ),
                        "active_threads": active_thread_count,
                        "max_threads": max_thread_count,
                        "metrics_source": "runtime_ingest",
                        "region": fallback_region,
                        "role": fallback_role,
                        "status": "busy" if max(active_thread_count, int(progress.get("running", 0) or 0)) > 0 else "online",
                    }
                )
            )

    items_total = max(0, int(active_job.get("items_total", 0) or 0))
    pending = max(0, int(progress.get("pending", 0) or 0))
    running = max(0, int(progress.get("running", 0) or 0))
    completed = max(0, int(progress.get("completed", 0) or 0))
    blacklisted = max(0, int(progress.get("blacklisted", 0) or 0))
    failed = max(
        max(0, int(progress.get("failed", 0) or 0)),
        max(0, int(active_job.get("items_failed", 0) or 0)),
    )
    terminal = max(
        max(0, int(active_job.get("items_terminal", 0) or 0)),
        completed + blacklisted + failed,
    )
    claimed = max(0, items_total - pending - running - terminal)
    assigned_totals = {
        "items_total": sum(_int_value(item.get("items_total")) for item in nodes),
        "items_pending": sum(_int_value(item.get("items_pending")) for item in nodes),
        "items_claimed": sum(_int_value(item.get("items_claimed")) for item in nodes),
        "items_running": sum(_int_value(item.get("items_running")) for item in nodes),
        "items_completed": sum(_int_value(item.get("items_completed")) for item in nodes),
        "items_blacklisted": sum(_int_value(item.get("items_blacklisted")) for item in nodes),
        "items_failed": sum(_int_value(item.get("items_failed")) for item in nodes),
    }
    unassigned_bucket = {
        "node_code": "unassigned",
        "items_total": max(0, items_total - assigned_totals["items_total"]),
        "items_pending": max(0, pending - assigned_totals["items_pending"]),
        "items_claimed": max(0, claimed - assigned_totals["items_claimed"]),
        "items_running": max(0, running - assigned_totals["items_running"]),
        "items_completed": max(0, completed - assigned_totals["items_completed"]),
        "items_blacklisted": max(0, blacklisted - assigned_totals["items_blacklisted"]),
        "items_failed": max(0, failed - assigned_totals["items_failed"]),
        "metrics_source": "runtime_ingest",
        "status": "online",
    }
    if any(int(unassigned_bucket.get(key, 0) or 0) > 0 for key in (
        "items_total",
        "items_pending",
        "items_claimed",
        "items_running",
        "items_completed",
        "items_blacklisted",
        "items_failed",
    )):
        nodes.append(_normalize_node_bucket(unassigned_bucket))
    return nodes


def _build_runtime_snapshot_from_projection(
    projection: dict,
    *,
    created_at: datetime | None = None,
    window_minutes: int = 15,
) -> dict:
    normalized_projection = dict(projection or {})
    active_job = dict(normalized_projection.get("active_job") or {})
    progress = dict(normalized_projection.get("progress") or {})
    backlog = dict(normalized_projection.get("backlog") or {})
    nodes = _normalize_runtime_projection_nodes(
        active_job=active_job,
        progress=progress,
        projection=normalized_projection,
    )

    items_total = max(0, int(active_job.get("items_total", 0) or 0))
    pending = max(0, int(progress.get("pending", active_job.get("items_pending", 0)) or 0))
    running = max(0, int(progress.get("running", active_job.get("items_running", 0)) or 0))
    completed = max(0, int(progress.get("completed", 0) or 0))
    blacklisted = max(0, int(progress.get("blacklisted", 0) or 0))
    failed = max(
        max(0, int(progress.get("failed", 0) or 0)),
        max(0, int(active_job.get("items_failed", 0) or 0)),
    )
    terminal = max(
        max(0, int(active_job.get("items_terminal", 0) or 0)),
        completed + blacklisted + failed,
    )
    if items_total <= 0:
        items_total = pending + running + terminal
    claimed = max(0, items_total - pending - running - terminal)
    display_running = _max_runtime_metric(
        running,
        normalized_projection.get("active_thread_count"),
        active_job.get("display_active_threads"),
        active_job.get("display_items_running"),
    )
    display_claimed = _max_runtime_metric(claimed, active_job.get("display_items_claimed"))
    progress_percent = float(active_job.get("progress_percent", 0) or 0)
    if progress_percent <= 0 and items_total > 0 and terminal > 0:
        progress_percent = round((terminal / items_total) * 100, 2)

    queue_health = {
        "window_minutes": max(5, min(int(window_minutes or 15), 120)),
        "has_active_job": bool(str(active_job.get("job_code") or "").strip() or items_total > 0),
        "job": {
            "job_id": int(active_job.get("job_id", 0) or 0),
            "job_code": str(active_job.get("job_code") or "").strip(),
            "runtime_job_code": str(active_job.get("job_code") or "").strip(),
            "status": str(active_job.get("status") or "").strip(),
            "progress_percent": progress_percent,
        },
        "queue": {
            "items_total": items_total,
            "pending": pending,
            "claimed": claimed,
            "running": running,
            "display_claimed": display_claimed,
            "display_running": display_running,
            "completed": completed,
            "blacklisted": blacklisted,
            "failed": failed,
            "terminal": terminal,
            "terminal_percent": round((terminal / items_total) * 100, 2) if items_total else 0.0,
            "oldest_pending_at": "",
            "oldest_pending_age_minutes": 0,
            "nearest_lease_expiry_at": "",
            "overdue_leases": 0,
            "expiring_soon_leases": 0,
        },
        "throughput": {
            "processed_recent": 0,
            "processed_per_minute": 0.0,
            "completed_recent": 0,
            "blacklisted_recent": 0,
            "failed_recent": 0,
        },
        "nodes": nodes,
        "steps": [],
        "runtime_activity": {},
    }
    snapshot = {
        "job": {
            "job_id": int(active_job.get("job_id", 0) or 0),
            "job_code": str(active_job.get("job_code") or "").strip(),
            "status": str(active_job.get("status") or "").strip(),
            "items_total": items_total,
            "items_pending": pending,
            "items_claimed": claimed,
            "items_running": running,
            "items_completed": completed,
            "items_blacklisted": blacklisted,
            "items_failed": failed,
            "items_terminal": terminal,
            "progress_percent": progress_percent,
            "node_stats": nodes,
        },
        "queue_health": queue_health,
        "backlog": backlog if isinstance(backlog, dict) else {},
        "recent_events": [],
        "_snapshot_source": "runtime_ingest",
    }
    if created_at:
        snapshot["_created_at"] = _format_time(created_at)
    return snapshot


def _snapshot_created_at(snapshot: dict) -> datetime | None:
    return _parse_runtime_timestamp(snapshot.get("_created_at"))


def _parse_runtime_timestamp(value: object) -> datetime | None:
    raw_value = str(value or "").strip()
    if not raw_value:
        return None
    normalized_value = raw_value.replace("T", " ")
    if normalized_value.endswith("Z"):
        normalized_value = f"{normalized_value[:-1]}+00:00"
    try:
        return datetime.fromisoformat(normalized_value)
    except ValueError:
        for fmt in ("%Y-%m-%d %H:%M:%S.%f", "%Y-%m-%d %H:%M:%S"):
            try:
                return datetime.strptime(normalized_value, fmt)
            except ValueError:
                continue
    return None


def _runtime_timestamp_is_fresh(value: datetime | None, *, max_age: timedelta) -> bool:
    if value is None:
        return False
    now = datetime.now(value.tzinfo) if value.tzinfo else datetime.now()
    return now - value <= max_age


def _runtime_node_display_running(item: dict) -> int:
    return _max_runtime_metric(
        item.get("display_running"),
        item.get("current_load"),
        item.get("active_threads"),
        item.get("items_running"),
    )


def _load_runtime_node_overlay_map(node_codes: list[str]) -> dict[str, dict]:
    normalized_codes = [
        str(item or "").strip()
        for item in list(node_codes or [])
        if str(item or "").strip() and str(item or "").strip() != "unassigned"
    ]
    if not normalized_codes:
        return {}
    if not (settings.node_region == "overseas" and settings.node_role == "control"):
        return {}
    disabled_node_codes = _load_disabled_managed_node_codes(normalized_codes)
    try:
        with get_db() as conn:
            with conn.cursor() as cur:
                cur.execute(
                    """
                    SELECT node_code, region, role, status, current_load, metadata_json, last_heartbeat_at
                    FROM detect_worker_nodes
                    WHERE node_code = ANY(%s)
                    """,
                    (normalized_codes,),
                )
                rows = list(cur.fetchall() or [])
    except Exception:
        return {}

    overlays: dict[str, dict] = {}
    for row in rows:
        node_code = str((row or [""])[0] or "").strip()
        if not node_code or node_code in disabled_node_codes:
            continue
        metadata = _decode_payload((row or ["", "", "", "", 0, {}])[5])
        overlays[node_code] = {
            "region": str((row or ["", "", ""])[1] or "").strip(),
            "role": str((row or ["", "", "", ""])[2] or "").strip(),
            "status": str((row or ["", "", "", "", ""])[3] or "").strip(),
            "current_load": _int_value((row or ["", "", "", "", 0])[4]),
            "active_threads": _int_value(metadata.get("active_threads")),
            "max_threads": _int_value(metadata.get("max_threads")),
            "last_heartbeat_at": _format_time((row or ["", "", "", "", 0, {}, None])[6]) if len(row or []) > 6 and row[6] else "",
            "metrics_source": str(metadata.get("service") or "runtime").strip() or "runtime",
        }
    return overlays


def _merge_runtime_node_overlay(item: dict, overlay: dict | None) -> dict:
    merged = dict(item or {})
    normalized_overlay = dict(overlay or {})
    if not normalized_overlay:
        return merged
    for key in ("region", "role", "status", "last_heartbeat_at", "metrics_source"):
        if not str(merged.get(key) or "").strip() and str(normalized_overlay.get(key) or "").strip():
            merged[key] = normalized_overlay.get(key)
    for key in ("current_load", "active_threads", "max_threads"):
        merged[key] = max(_int_value(merged.get(key)), _int_value(normalized_overlay.get(key)))
    return merged


def _filter_live_runtime_nodes(nodes: list[dict], *, snapshot_created_at: datetime | None) -> list[dict]:
    filtered: list[dict] = []
    for raw_item in list(nodes or []):
        if not isinstance(raw_item, dict):
            continue
        item = dict(raw_item)
        node_code = str(item.get("node_code") or "").strip()
        if not node_code:
            continue
        if node_code == "unassigned":
            filtered.append(item)
            continue
        status = str(item.get("status") or "").strip().lower()
        if status in {"stale", "offline"}:
            continue
        reference_at = _parse_runtime_timestamp(item.get("last_heartbeat_at"))
        if reference_at is None:
            filtered.append(item)
            continue
        if not _runtime_timestamp_is_fresh(reference_at, max_age=_LIVE_RUNTIME_NODE_STALE_AFTER):
            continue
        filtered.append(item)
    return filtered


def _build_live_runtime_snapshot(snapshot: dict | None) -> dict:
    normalized_snapshot = dict(snapshot or {})
    if not normalized_snapshot:
        return {}

    snapshot_created_at = _snapshot_created_at(normalized_snapshot)
    job = dict(normalized_snapshot.get("job") or {})
    queue_health = dict(normalized_snapshot.get("queue_health") or {})
    queue = dict(queue_health.get("queue") or {})
    queue_nodes = [dict(item) for item in list(queue_health.get("nodes") or []) if isinstance(item, dict)]
    job_nodes = [dict(item) for item in list(job.get("node_stats") or []) if isinstance(item, dict)]
    overlay_map = {}
    if str(normalized_snapshot.get("_snapshot_source") or "").strip() == "runtime_ingest":
        overlay_map = _load_runtime_node_overlay_map(
            [
                *(item.get("node_code") for item in queue_nodes),
                *(item.get("node_code") for item in job_nodes),
            ]
        )
    queue_nodes = [
        _merge_runtime_node_overlay(item, overlay_map.get(str(item.get("node_code") or "").strip()))
        for item in queue_nodes
    ]
    job_nodes = [
        _merge_runtime_node_overlay(item, overlay_map.get(str(item.get("node_code") or "").strip()))
        for item in job_nodes
    ]
    live_queue_nodes = _filter_live_runtime_nodes(queue_nodes, snapshot_created_at=snapshot_created_at)
    live_job_nodes = _filter_live_runtime_nodes(job_nodes, snapshot_created_at=snapshot_created_at)
    has_runtime_node_rows = bool(queue_nodes or job_nodes)
    raw_node_count = len([item for item in queue_nodes or job_nodes if str(item.get("node_code") or "").strip() and str(item.get("node_code") or "").strip() != "unassigned"])
    live_nodes_for_display = live_queue_nodes if queue_nodes else live_job_nodes
    live_participant_nodes = [
        item
        for item in live_nodes_for_display
        if str(item.get("node_code") or "").strip()
        and str(item.get("node_code") or "").strip() != "unassigned"
    ]
    dropped_node_count = max(0, raw_node_count - len(live_participant_nodes))

    display_running = sum(_runtime_node_display_running(item) for item in live_participant_nodes)
    display_claimed = sum(
        _max_runtime_metric(item.get("display_claimed"), item.get("items_claimed"))
        for item in live_participant_nodes
    )
    display_max_threads = sum(max(0, _int_value(item.get("max_threads"))) for item in live_participant_nodes)
    allow_raw_display_fallback = dropped_node_count <= 0 and (
        not has_runtime_node_rows or bool(live_participant_nodes)
    )
    if allow_raw_display_fallback:
        display_running = _max_runtime_metric(
            display_running,
            queue.get("display_running"),
            job.get("display_items_running"),
            job.get("display_active_threads"),
            normalized_snapshot.get("active_thread_count"),
            job.get("items_running"),
            queue.get("running"),
        )
        display_claimed = _max_runtime_metric(
            display_claimed,
            queue.get("display_claimed"),
            job.get("display_items_claimed"),
            queue.get("claimed"),
            job.get("items_claimed"),
        )
        display_max_threads = _max_runtime_metric(
            display_max_threads,
            job.get("display_max_threads"),
            normalized_snapshot.get("max_thread_count"),
        )

    if queue_nodes or job_nodes:
        queue_health["nodes"] = list(live_queue_nodes or live_job_nodes)
        queue["display_running"] = display_running
        queue["display_claimed"] = display_claimed
        queue_health["queue"] = queue
        job["node_stats"] = list(live_job_nodes or live_queue_nodes)
        job["display_items_running"] = display_running
        job["display_active_threads"] = display_running
        job["display_max_threads"] = display_max_threads

    normalized_snapshot["job"] = job
    normalized_snapshot["queue_health"] = queue_health
    normalized_snapshot["_raw_runtime_node_count"] = raw_node_count
    normalized_snapshot["_live_runtime_node_count"] = len(live_participant_nodes)
    normalized_snapshot["_dropped_runtime_node_count"] = dropped_node_count
    normalized_snapshot["_has_runtime_node_rows"] = has_runtime_node_rows
    return normalized_snapshot


def _snapshot_display_running(snapshot: dict) -> int:
    queue = dict((snapshot.get("queue_health") or {}).get("queue") or {})
    job = dict(snapshot.get("job") or {})
    return max(
        _int_value(queue.get("display_running")),
        _int_value(job.get("display_items_running")),
        _int_value(job.get("display_active_threads")),
        _int_value(snapshot.get("active_thread_count")),
    )


def _runtime_job_matches_active_job(
    active_job: dict | None,
    *,
    runtime_job_code: object = "",
    runtime_job_id: object = 0,
) -> bool:
    normalized_active_job = dict(active_job or {})
    active_job_code = str(normalized_active_job.get("job_code") or "").strip()
    active_job_id = _int_value(normalized_active_job.get("job_id"))
    normalized_runtime_job_code = str(runtime_job_code or "").strip()
    normalized_runtime_job_id = _int_value(runtime_job_id)
    if active_job_id > 0 and normalized_runtime_job_id > 0 and active_job_id == normalized_runtime_job_id:
        return True
    if active_job_code and normalized_runtime_job_code and active_job_code == normalized_runtime_job_code:
        return True
    return False


def _snapshot_node_count(snapshot: dict) -> int:
    queue_nodes = list((snapshot.get("queue_health") or {}).get("nodes") or [])
    if queue_nodes:
        return len(
            [
                item
                for item in queue_nodes
                if isinstance(item, dict) and str(item.get("node_code") or "").strip() != "unassigned"
            ]
        )
    job_nodes = list((snapshot.get("job") or {}).get("node_stats") or [])
    return len(
        [
            item
            for item in job_nodes
            if isinstance(item, dict) and str(item.get("node_code") or "").strip() != "unassigned"
        ]
    )


def _snapshot_is_stale_for_live_display(snapshot: dict) -> bool:
    normalized_snapshot = dict(snapshot or {})
    if str(normalized_snapshot.get("_snapshot_source") or "").strip() != "runtime_ingest":
        return False
    created_at = _snapshot_created_at(normalized_snapshot)
    if created_at is None:
        return False
    return not _runtime_timestamp_is_fresh(created_at, max_age=_PREFERRED_RUNTIME_SNAPSHOT_GRACE)


def _should_prefer_fallback_runtime_snapshot(*, fallback_snapshot: dict, matched_snapshot: dict) -> bool:
    fallback_nodes = _snapshot_node_count(fallback_snapshot)
    matched_nodes = _snapshot_node_count(matched_snapshot)
    fallback_display_running = _snapshot_display_running(fallback_snapshot)
    matched_display_running = _snapshot_display_running(matched_snapshot)
    if fallback_nodes <= matched_nodes and fallback_display_running <= matched_display_running:
        return False

    fallback_created_at = _snapshot_created_at(fallback_snapshot)
    matched_created_at = _snapshot_created_at(matched_snapshot)
    if fallback_created_at and matched_created_at:
        return fallback_created_at - matched_created_at > _PREFERRED_RUNTIME_SNAPSHOT_GRACE
    return matched_nodes <= 0 and matched_display_running <= 0


def _load_latest_runtime_ingest_active_job_snapshot(
    window_minutes: int,
    *,
    preferred_job_codes: list[str] | None = None,
) -> dict:
    safe_window_minutes = max(5, min(int(window_minutes or 15), 120))
    if not (settings.node_region == "overseas" and settings.node_role == "control"):
        return {}

    preferred = {
        str(item or "").strip()
        for item in list(preferred_job_codes or [])
        if str(item or "").strip()
    }
    query_sql = """
        SELECT payload_json, COALESCE(updated_at, created_at)
        FROM detect_sync_records
        WHERE sync_type = 'runtime_ingest'
          AND source_region = 'mainland'
          AND target_region = 'overseas'
          AND created_at >= CURRENT_TIMESTAMP - (%s || ' minutes')::interval
        ORDER BY COALESCE(updated_at, created_at) DESC, id DESC
        LIMIT 20
    """

    def _select_rows(lookback_minutes: int) -> list[tuple]:
        with get_db() as conn:
            with conn.cursor() as cur:
                cur.execute(query_sql, (int(lookback_minutes),))
                return list(cur.fetchall() or [])

    def _pick_snapshot(rows: list[tuple]) -> tuple[dict, dict]:
        fallback_snapshot: dict = {}
        matched_snapshot: dict = {}
        for row in rows:
            payload = _decode_payload(row[0])
            projection = payload.get("projection") if isinstance(payload, dict) else {}
            if not isinstance(projection, dict):
                continue
            snapshot = _build_runtime_snapshot_from_projection(
                projection,
                created_at=row[1],
                window_minutes=safe_window_minutes,
            )
            if not fallback_snapshot:
                fallback_snapshot = snapshot
            job_code = str((snapshot.get("job") or {}).get("job_code") or "").strip()
            if preferred and job_code in preferred:
                matched_snapshot = snapshot
                break
        return fallback_snapshot, matched_snapshot

    recent_rows = _select_rows(safe_window_minutes)
    fallback_snapshot, matched_snapshot = _pick_snapshot(recent_rows)
    if matched_snapshot:
        if (
            fallback_snapshot
            and fallback_snapshot is not matched_snapshot
            and _should_prefer_fallback_runtime_snapshot(
                fallback_snapshot=fallback_snapshot,
                matched_snapshot=matched_snapshot,
            )
        ):
            return fallback_snapshot
        return matched_snapshot

    if preferred:
        extended_rows = _select_rows(max(safe_window_minutes, 12 * 60))
        extended_fallback, matched_snapshot = _pick_snapshot(extended_rows)
        if matched_snapshot:
            if (
                extended_fallback
                and extended_fallback is not matched_snapshot
                and _should_prefer_fallback_runtime_snapshot(
                    fallback_snapshot=extended_fallback,
                    matched_snapshot=matched_snapshot,
                )
            ):
                return extended_fallback
            return matched_snapshot
        if not fallback_snapshot:
            fallback_snapshot = extended_fallback

    return fallback_snapshot


@db_read_retry()
def _load_latest_runtime_active_job_snapshot(window_minutes: int) -> dict:
    safe_window_minutes = max(5, min(int(window_minutes or 15), 120))
    if not (settings.node_region == "overseas" and settings.node_role == "control"):
        return {}

    runtime_activity = _load_runtime_activity_snapshot(safe_window_minutes)
    focus_job_code = str(runtime_activity.get("focus_job_code") or "").strip()
    preferred_job_codes = [focus_job_code] + [
        str(item or "").strip()
        for item in list(runtime_activity.get("job_codes") or [])
        if str(item or "").strip()
    ]
    debug_snapshot = _load_latest_runtime_debug_active_job_snapshot(
        safe_window_minutes,
        preferred_job_codes=preferred_job_codes,
    )
    ingest_snapshot = _load_latest_runtime_ingest_active_job_snapshot(
        safe_window_minutes,
        preferred_job_codes=preferred_job_codes,
    )
    debug_has_payload = bool(debug_snapshot.get("job")) or bool((debug_snapshot.get("queue_health") or {}).get("queue"))
    ingest_has_payload = bool(ingest_snapshot.get("job")) or bool((ingest_snapshot.get("queue_health") or {}).get("queue"))

    if ingest_has_payload and not debug_has_payload:
        return _build_live_runtime_snapshot(ingest_snapshot)
    if debug_has_payload and not ingest_has_payload:
        return _build_live_runtime_snapshot(debug_snapshot)
    if not debug_has_payload and not ingest_has_payload:
        return {}
    if focus_job_code:
        debug_job_code = str((debug_snapshot.get("job") or {}).get("job_code") or "").strip()
        ingest_job_code = str((ingest_snapshot.get("job") or {}).get("job_code") or "").strip()
        debug_matches_focus = bool(debug_job_code and debug_job_code == focus_job_code)
        ingest_matches_focus = bool(ingest_job_code and ingest_job_code == focus_job_code)
        if ingest_matches_focus and not debug_matches_focus:
            return _build_live_runtime_snapshot(ingest_snapshot)
        if debug_matches_focus and not ingest_matches_focus:
            return _build_live_runtime_snapshot(debug_snapshot)
    debug_queue = dict((debug_snapshot.get("queue_health") or {}).get("queue") or {})
    ingest_queue = dict((ingest_snapshot.get("queue_health") or {}).get("queue") or {})
    debug_nodes = list((debug_snapshot.get("queue_health") or {}).get("nodes") or (debug_snapshot.get("job") or {}).get("node_stats") or [])
    ingest_nodes = list((ingest_snapshot.get("queue_health") or {}).get("nodes") or (ingest_snapshot.get("job") or {}).get("node_stats") or [])
    debug_display_running = max(
        _int_value(debug_queue.get("display_running")),
        _int_value(((debug_snapshot.get("job") or {}).get("display_items_running"))),
        _int_value(((debug_snapshot.get("job") or {}).get("display_active_threads"))),
    )
    ingest_display_running = max(
        _int_value(ingest_queue.get("display_running")),
        _int_value(((ingest_snapshot.get("job") or {}).get("display_items_running"))),
        _int_value(((ingest_snapshot.get("job") or {}).get("display_active_threads"))),
    )
    if len(ingest_nodes) > len(debug_nodes) or ingest_display_running > debug_display_running:
        return _build_live_runtime_snapshot(ingest_snapshot)
    if str(ingest_snapshot.get("_created_at") or "") > str(debug_snapshot.get("_created_at") or ""):
        return _build_live_runtime_snapshot(ingest_snapshot)
    return _build_live_runtime_snapshot(debug_snapshot)


def _select_runtime_snapshot_events(
    runtime_snapshot: dict,
    *,
    job_code: str = "",
    job_id: int = 0,
    event_limit: int = 20,
) -> list[dict]:
    safe_limit = max(1, min(int(event_limit or 20), 100))
    target_job_code = str(job_code or "").strip()
    target_job_id = int(job_id or 0)
    selected: list[dict] = []
    for raw_event in list(runtime_snapshot.get("recent_events") or []):
        if not isinstance(raw_event, dict):
            continue
        payload = raw_event.get("payload")
        normalized_payload = payload if isinstance(payload, dict) else _decode_payload(payload)
        event_job_code = str(normalized_payload.get("job_code") or "").strip()
        event_job_id = _int_value(raw_event.get("job_id"))
        if target_job_code:
            if event_job_code == target_job_code or (target_job_id > 0 and event_job_id == target_job_id):
                selected.append(
                    {
                        "node_code": str(raw_event.get("node_code") or "").strip(),
                        "event_type": str(raw_event.get("event_type") or "").strip(),
                        "level": str(raw_event.get("level") or "info").strip() or "info",
                        "message": str(raw_event.get("message") or "").strip(),
                        "payload": normalized_payload,
                        "created_at": str(raw_event.get("created_at") or "").strip(),
                    }
                )
        else:
            selected.append(
                {
                    "node_code": str(raw_event.get("node_code") or "").strip(),
                    "event_type": str(raw_event.get("event_type") or "").strip(),
                    "level": str(raw_event.get("level") or "info").strip() or "info",
                    "message": str(raw_event.get("message") or "").strip(),
                    "payload": normalized_payload,
                    "created_at": str(raw_event.get("created_at") or "").strip(),
                }
            )
        if len(selected) >= safe_limit:
            break
    return selected


def _filter_events_for_job(events: list[dict], *, job_code: str = "", job_id: int = 0, limit: int = 20) -> list[dict]:
    target_job_code = str(job_code or "").strip()
    target_job_id = int(job_id or 0)
    safe_limit = max(1, min(int(limit or 20), 100))
    filtered: list[dict] = []
    for raw_event in list(events or []):
        if not isinstance(raw_event, dict):
            continue
        payload = raw_event.get("payload")
        normalized_payload = payload if isinstance(payload, dict) else _decode_payload(payload)
        event_job_code = str(normalized_payload.get("job_code") or "").strip()
        event_job_id = _int_value(raw_event.get("job_id"))
        if target_job_code:
            if event_job_code != target_job_code and (target_job_id <= 0 or event_job_id != target_job_id):
                continue
        filtered.append(
            {
                "node_code": str(raw_event.get("node_code") or "").strip(),
                "event_type": str(raw_event.get("event_type") or "").strip(),
                "level": str(raw_event.get("level") or "info").strip() or "info",
                "message": str(raw_event.get("message") or "").strip(),
                "payload": normalized_payload,
                "created_at": str(raw_event.get("created_at") or "").strip(),
            }
        )
        if len(filtered) >= safe_limit:
            break
    return filtered


def _build_active_job_summary_from_runtime_snapshot(runtime_snapshot: dict, *, event_limit: int = 20) -> dict | None:
    normalized_snapshot = _build_live_runtime_snapshot(runtime_snapshot)
    runtime_job = dict(normalized_snapshot.get("job") or {})
    runtime_queue = dict((normalized_snapshot.get("queue_health") or {}).get("queue") or {})
    if not runtime_job and not runtime_queue:
        return None

    runtime_nodes = list((normalized_snapshot.get("queue_health") or {}).get("nodes") or runtime_job.get("node_stats") or [])
    runtime_steps = list((normalized_snapshot.get("queue_health") or {}).get("steps") or [])
    runtime_throughput = dict((normalized_snapshot.get("queue_health") or {}).get("throughput") or {})
    normalized_nodes = [
        _normalize_node_bucket(item)
        for item in runtime_nodes
        if isinstance(item, dict) and str(item.get("node_code") or "").strip()
    ]
    display_summary = _build_display_summary(normalized_nodes)
    runtime_allow_raw_display_fallback = (
        int(normalized_snapshot.get("_dropped_runtime_node_count", 0) or 0) <= 0
        and (
            not bool(normalized_snapshot.get("_has_runtime_node_rows"))
            or bool(display_summary.get("active_nodes") or [])
            or any(str(item.get("node_code") or "").strip() != "unassigned" for item in normalized_nodes)
        )
    )
    display_running_candidates = [
        runtime_queue.get("display_running"),
        display_summary.get("display_running"),
        runtime_job.get("display_items_running"),
        runtime_job.get("display_active_threads"),
    ]
    if runtime_allow_raw_display_fallback:
        display_running_candidates.extend((runtime_job.get("items_running"), runtime_queue.get("running")))
    display_items_running = _max_runtime_metric(
        *display_running_candidates,
    )
    display_active_thread_candidates = [
        display_summary.get("active_threads"),
        runtime_job.get("display_active_threads"),
        runtime_job.get("display_items_running"),
        runtime_queue.get("display_running"),
    ]
    if runtime_allow_raw_display_fallback:
        display_active_thread_candidates.extend((runtime_job.get("items_running"), runtime_queue.get("running")))
    display_active_threads = _max_runtime_metric(
        *display_active_thread_candidates,
    )
    display_max_threads = _max_runtime_metric(
        display_summary.get("max_threads"),
        runtime_job.get("display_max_threads"),
    )

    items_total = _int_value(runtime_queue.get("items_total", runtime_job.get("items_total", 0)))
    items_pending = _int_value(runtime_queue.get("pending", runtime_job.get("items_pending", 0)))
    items_claimed = _int_value(runtime_queue.get("claimed", runtime_job.get("items_claimed", 0)))
    items_running = _int_value(runtime_queue.get("running", runtime_job.get("items_running", 0)))
    items_completed = _int_value(runtime_queue.get("completed", runtime_job.get("items_completed", 0)))
    items_blacklisted = _int_value(runtime_queue.get("blacklisted", runtime_job.get("items_blacklisted", 0)))
    items_failed = _int_value(runtime_queue.get("failed", runtime_job.get("items_failed", 0)))
    items_terminal = _int_value(
        runtime_queue.get(
            "terminal",
            runtime_job.get("items_terminal", items_completed + items_blacklisted + items_failed),
        )
    )
    display_current_load = _max_runtime_metric(
        display_summary.get("current_load"),
        display_items_running,
        display_active_threads,
        items_running,
    )
    recent_events = _select_runtime_snapshot_events(
        normalized_snapshot,
        job_code=str(runtime_job.get("job_code") or "").strip(),
        job_id=int(runtime_job.get("job_id", 0) or 0),
        event_limit=event_limit,
    )
    cycle_token, current_cycle_events = _extract_current_cycle_events(recent_events)
    return {
        "job_id": int(runtime_job.get("job_id", 0) or 0),
        "job_code": str(runtime_job.get("job_code") or "").strip(),
        "source": str(normalized_snapshot.get("_snapshot_source") or "runtime").strip() or "runtime",
        "task_mode": _DEFAULT_TASK_MODE,
        "step_code": "",
        "status": str(runtime_job.get("status") or "").strip() or "running",
        "created_by": str(normalized_snapshot.get("_snapshot_source") or "runtime").strip() or "runtime",
        "created_at": str(normalized_snapshot.get("_created_at") or "").strip(),
        "started_at": "",
        "finished_at": "",
        "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,
        "items_terminal": items_terminal,
        "progress_percent": float(runtime_job.get("progress_percent", 0) or 0),
        "raw_items_total": items_total,
        "raw_items_pending": items_pending,
        "raw_items_claimed": items_claimed,
        "raw_items_running": items_running,
        "raw_items_completed": items_completed,
        "raw_items_blacklisted": items_blacklisted,
        "raw_items_failed": items_failed,
        "raw_items_terminal": items_terminal,
        "raw_node_stats": list(normalized_nodes),
        "raw_step_stats": list(runtime_steps),
        "node_stats": list(normalized_nodes),
        "distributed_node_stats": list(normalized_nodes),
        "step_stats": list(runtime_steps),
        "display_items_claimed": _max_runtime_metric(
            runtime_queue.get("display_claimed"),
            display_summary.get("items_claimed"),
            items_claimed,
        ),
        "display_items_running": display_items_running,
        "display_current_load": display_current_load,
        "display_active_threads": display_active_threads,
        "display_max_threads": display_max_threads,
        "display_items_completed": int(display_summary.get("items_completed", items_completed) or 0),
        "display_items_failed": int(display_summary.get("items_failed", items_failed) or 0),
        "display_active_node_codes": list(display_summary.get("active_nodes") or []),
        "processed_recent": int(runtime_throughput.get("processed_recent", 0) or 0),
        "processed_per_minute": float(runtime_throughput.get("processed_per_minute", 0) or 0),
        "completed_recent": int(runtime_throughput.get("completed_recent", 0) or 0),
        "blacklisted_recent": int(runtime_throughput.get("blacklisted_recent", 0) or 0),
        "failed_recent": int(runtime_throughput.get("failed_recent", 0) or 0),
        "recent_events": recent_events,
        "latest_event": recent_events[0] if recent_events else None,
        "current_cycle_token": cycle_token,
        "current_cycle_events": current_cycle_events,
        "latest_cycle_event": current_cycle_events[0] if current_cycle_events else (recent_events[0] if recent_events else None),
        "recent_domain_events": [],
    }


def _fetch_job_summary(cur, job_row, event_limit: int = 20) -> dict:
    job_id = job_row[0]
    cur.execute(
        """
        SELECT status, count(*)
        FROM detect_job_items
        WHERE job_id = %s
        GROUP BY status
        """,
        (job_id,),
    )
    counts = {status: int(count) for status, count in cur.fetchall()}
    cur.execute(
        """
        SELECT COALESCE(NULLIF(claimed_by, ''), 'unassigned') AS node_code,
               status,
               count(*)
        FROM detect_job_items
        WHERE job_id = %s
        GROUP BY COALESCE(NULLIF(claimed_by, ''), 'unassigned'), status
        ORDER BY node_code ASC, status ASC
        """,
        (job_id,),
    )
    node_buckets: dict[str, dict] = {}
    for node_code, status, count in cur.fetchall():
        bucket = node_buckets.setdefault(
            node_code,
            {
                "node_code": node_code,
                "items_total": 0,
                "items_pending": 0,
                "items_claimed": 0,
                "items_running": 0,
                "items_completed": 0,
                "items_blacklisted": 0,
                "items_failed": 0,
            },
        )
        field_name = f"items_{status}"
        if field_name in bucket:
            bucket[field_name] += int(count)
        bucket["items_total"] += int(count)

    cur.execute(
        """
        SELECT COALESCE(NULLIF(step_code, ''), 'domain_pipeline') AS step_code, status, count(*)
        FROM detect_job_items
        WHERE job_id = %s
        GROUP BY COALESCE(NULLIF(step_code, ''), 'domain_pipeline'), status
        ORDER BY step_code ASC, status ASC
        """,
        (job_id,),
    )
    step_buckets: dict[str, dict] = {}
    for step_code, status, count in cur.fetchall():
        bucket = step_buckets.setdefault(str(step_code or ""), _build_step_bucket(str(step_code or "")))
        field_name = f"items_{status}"
        if field_name in bucket:
            bucket[field_name] += int(count)
        bucket["items_total"] += int(count)

    cur.execute(
        """
        SELECT node_code, event_type, level, message, payload_json, created_at
        FROM detect_run_events
        WHERE job_id = %s
        ORDER BY created_at DESC, id DESC
        LIMIT %s
        """,
        (job_id, max(1, int(event_limit or 20))),
    )
    events = [
        {
            "node_code": item[0],
            "event_type": item[1],
            "level": item[2],
            "message": item[3],
            "payload": _decode_payload(item[4]),
            "created_at": _format_time(item[5]),
        }
        for item in cur.fetchall()
    ]
    cycle_token, current_cycle_events = _extract_current_cycle_events(events)
    distributed_node_stats = _merge_display_node_stats(
        local_node_stats=list(node_buckets.values()),
        runtime_node_rows=_load_runtime_display_rows(cur),
    )
    raw_total = sum(counts.values())
    raw_pending = int(counts.get("pending", 0))
    raw_claimed = int(counts.get("claimed", 0))
    raw_running = int(counts.get("running", 0))
    raw_completed = int(counts.get("completed", 0))
    raw_blacklisted = int(counts.get("blacklisted", 0))
    raw_failed = int(counts.get("failed", 0))
    raw_terminal = raw_completed + raw_blacklisted + raw_failed
    effective_node_stats = _build_effective_node_stats(
        distributed_node_stats=distributed_node_stats,
        raw_items_total=raw_total,
    )
    effective_summary = _build_effective_summary(
        node_stats=effective_node_stats,
        raw_items_total=raw_total,
        raw_items_pending=raw_pending,
        raw_items_claimed=raw_claimed,
        raw_items_running=raw_running,
        raw_items_completed=raw_completed,
        raw_items_blacklisted=raw_blacklisted,
        raw_items_failed=raw_failed,
    )
    display_summary = _build_display_summary(effective_node_stats)
    return {
        "job_id": job_id,
        "job_code": job_row[1],
        "source": job_row[2],
        "task_mode": job_row[3] or _DEFAULT_TASK_MODE,
        "step_code": job_row[4] or "",
        "status": job_row[5],
        "created_by": job_row[6],
        "created_at": _format_time(job_row[7]),
        "started_at": _format_time(job_row[8]),
        "finished_at": _format_time(job_row[9]),
        "items_total": int(effective_summary.get("items_total", 0) or 0),
        "items_pending": int(effective_summary.get("items_pending", 0) or 0),
        "items_claimed": int(effective_summary.get("items_claimed", 0) or 0),
        "items_running": int(effective_summary.get("items_running", 0) or 0),
        "items_completed": int(effective_summary.get("items_completed", 0) or 0),
        "items_blacklisted": int(effective_summary.get("items_blacklisted", 0) or 0),
        "items_failed": int(effective_summary.get("items_failed", 0) or 0),
        "items_terminal": int(effective_summary.get("items_terminal", 0) or 0),
        "progress_percent": round(
            (int(effective_summary.get("items_terminal", 0) or 0) / int(effective_summary.get("items_total", 0) or 0)) * 100,
            2,
        ) if int(effective_summary.get("items_total", 0) or 0) else 0,
        "raw_items_total": raw_total,
        "raw_items_pending": raw_pending,
        "raw_items_claimed": raw_claimed,
        "raw_items_running": raw_running,
        "raw_items_completed": raw_completed,
        "raw_items_blacklisted": raw_blacklisted,
        "raw_items_failed": raw_failed,
        "raw_items_terminal": raw_terminal,
        "raw_node_stats": list(node_buckets.values()),
        "raw_step_stats": list(step_buckets.values()),
        "node_stats": effective_node_stats,
        "distributed_node_stats": effective_node_stats,
        "step_stats": sorted(
            step_buckets.values(),
            key=lambda item: (
                -_int_value(item.get("items_running")),
                -_int_value(item.get("items_pending")),
                str(item.get("step_code") or ""),
            ),
        ),
        "display_items_claimed": int(display_summary.get("items_claimed", 0) or 0),
        "display_items_running": int(display_summary.get("display_running", display_summary.get("items_running", 0)) or 0),
        "display_current_load": int(display_summary.get("current_load", 0) or 0),
        "display_active_threads": int(display_summary.get("active_threads", 0) or 0),
        "display_max_threads": int(display_summary.get("max_threads", 0) or 0),
        "display_items_completed": int(display_summary.get("items_completed", 0) or 0),
        "display_items_failed": int(display_summary.get("items_failed", 0) or 0),
        "display_active_node_codes": list(display_summary.get("active_nodes") or []),
        "recent_events": events,
        "latest_event": events[0] if events else None,
        "current_cycle_token": cycle_token,
        "current_cycle_events": current_cycle_events,
        "latest_cycle_event": current_cycle_events[0] if current_cycle_events else None,
    }


def _enrich_active_job_summary_with_runtime(
    summary: dict | None,
    *,
    event_limit: int = 20,
    window_minutes: int = 15,
    runtime_activity: dict | None = None,
    runtime_snapshot: dict | None = None,
    recent_domain_events: list[dict] | None = None,
) -> dict | None:
    if not summary:
        return summary
    enriched = dict(summary)
    runtime_activity = dict(runtime_activity or _load_runtime_activity_snapshot(window_minutes))
    runtime_snapshot = _build_live_runtime_snapshot(runtime_snapshot or _load_latest_runtime_active_job_snapshot(window_minutes))
    runtime_job_code = str(runtime_activity.get("focus_job_code") or "").strip()
    runtime_job_codes = [
        str(item or "").strip()
        for item in list(runtime_activity.get("job_codes") or [])
        if str(item or "").strip()
    ]
    if runtime_job_code:
        enriched["runtime_job_code"] = runtime_job_code
    if runtime_job_codes:
        enriched["runtime_job_codes"] = runtime_job_codes

    enriched["processed_recent"] = max(
        int(enriched.get("processed_recent", 0) or 0),
        int(runtime_activity.get("processed_recent", 0) or 0),
    )
    enriched["processed_per_minute"] = max(
        float(enriched.get("processed_per_minute", 0) or 0),
        round(int(runtime_activity.get("processed_recent", 0) or 0) / max(1, int(window_minutes or 15)), 2),
    )
    enriched["completed_recent"] = max(
        int(enriched.get("completed_recent", 0) or 0),
        int(runtime_activity.get("completed_recent", 0) or 0),
    )
    enriched["failed_recent"] = max(
        int(enriched.get("failed_recent", 0) or 0),
        int(runtime_activity.get("failed_recent", 0) or 0),
    )
    enriched["blacklisted_recent"] = max(
        int(enriched.get("blacklisted_recent", 0) or 0),
        int(runtime_activity.get("blacklisted_recent", 0) or 0),
    )
    runtime_allow_raw_display_fallback = True
    runtime_missing_participant_attribution = False

    if recent_domain_events is not None:
        enriched["recent_domain_events"] = list(recent_domain_events)
    elif not enriched.get("recent_domain_events"):
        try:
            from app.services.sync_record_service import _collect_recent_domain_events

            enriched["recent_domain_events"] = _collect_recent_domain_events(
                enriched,
                limit=max(10, min(int(event_limit or 20), 60)),
            )
        except Exception:
            enriched["recent_domain_events"] = []
    if runtime_snapshot:
        runtime_job = dict(runtime_snapshot.get("job") or {})
        runtime_queue = dict((runtime_snapshot.get("queue_health") or {}).get("queue") or {})
        runtime_nodes = list((runtime_snapshot.get("queue_health") or {}).get("nodes") or [])
        runtime_steps = list((runtime_snapshot.get("queue_health") or {}).get("steps") or [])
        runtime_job_nodes = list(runtime_job.get("node_stats") or [])
        runtime_snapshot_source = str(runtime_snapshot.get("_snapshot_source") or "").strip()
        runtime_snapshot_stale_for_live_display = _snapshot_is_stale_for_live_display(runtime_snapshot)
        if runtime_snapshot_stale_for_live_display and str(enriched.get("created_by") or "").strip() == runtime_snapshot_source:
            runtime_snapshot_stale_for_live_display = False
        has_runtime_participant_nodes = any(
            str(item.get("node_code") or "").strip() and str(item.get("node_code") or "").strip() != "unassigned"
            for item in [*runtime_nodes, *runtime_job_nodes]
            if isinstance(item, dict)
        )
        runtime_missing_participant_attribution = bool(runtime_snapshot.get("_has_runtime_node_rows")) and not has_runtime_participant_nodes
        runtime_allow_raw_display_fallback = (
            int(runtime_snapshot.get("_dropped_runtime_node_count", 0) or 0) <= 0
            and (
                not bool(runtime_snapshot.get("_has_runtime_node_rows"))
                or has_runtime_participant_nodes
            )
        )
        if runtime_snapshot_stale_for_live_display:
            runtime_missing_participant_attribution = True
            runtime_allow_raw_display_fallback = False
        if runtime_job and not runtime_snapshot_stale_for_live_display:
            runtime_snapshot_job_id = int(runtime_job.get("job_id", 0) or 0)
            enriched["runtime_snapshot_job_id"] = runtime_snapshot_job_id
            runtime_snapshot_job_code = str(runtime_job.get("job_code") or "").strip()
            enriched["runtime_snapshot_job_code"] = runtime_snapshot_job_code
            if runtime_snapshot_job_code:
                enriched["job_code"] = runtime_snapshot_job_code
                enriched["runtime_job_code"] = runtime_snapshot_job_code
                merged_runtime_codes = [runtime_snapshot_job_code] + [
                    item for item in list(enriched.get("runtime_job_codes") or []) if str(item or "").strip() != runtime_snapshot_job_code
                ]
                enriched["runtime_job_codes"] = merged_runtime_codes[:5]
            if runtime_snapshot_job_id > 0:
                enriched["job_id"] = runtime_snapshot_job_id
            runtime_status = str(runtime_job.get("status") or "").strip()
            if runtime_status:
                enriched["status"] = runtime_status
            if float(runtime_job.get("progress_percent", 0) or 0) > float(enriched.get("progress_percent", 0) or 0):
                enriched["progress_percent"] = float(runtime_job.get("progress_percent", 0) or 0)
        if runtime_queue and not runtime_snapshot_stale_for_live_display:
            enriched["runtime_snapshot_queue"] = runtime_queue
            runtime_items_total = int(runtime_queue.get("items_total", 0) or 0)
            if runtime_items_total > 0:
                enriched["items_total"] = runtime_items_total
                enriched["items_pending"] = int(runtime_queue.get("pending", 0) or 0)
                enriched["items_claimed"] = int(runtime_queue.get("claimed", 0) or 0)
                enriched["items_running"] = int(runtime_queue.get("running", 0) or 0)
                enriched["items_completed"] = int(runtime_queue.get("completed", 0) or 0)
                enriched["items_blacklisted"] = int(runtime_queue.get("blacklisted", 0) or 0)
                enriched["items_failed"] = int(runtime_queue.get("failed", 0) or 0)
                enriched["items_terminal"] = int(runtime_queue.get("terminal", 0) or 0)
                enriched["display_items_claimed"] = _max_runtime_metric(
                    enriched.get("display_items_claimed"),
                    runtime_queue.get("display_claimed"),
                    enriched.get("items_claimed"),
                )
                display_running_candidates = [
                    enriched.get("display_items_running"),
                    runtime_queue.get("display_running"),
                    enriched.get("display_active_threads"),
                ]
                if runtime_allow_raw_display_fallback:
                    display_running_candidates.append(enriched.get("items_running"))
                enriched["display_items_running"] = _max_runtime_metric(*display_running_candidates)
        if runtime_job_nodes and not runtime_snapshot_stale_for_live_display:
            enriched["node_stats"] = list(runtime_job_nodes)
            enriched["distributed_node_stats"] = list(runtime_job_nodes)
            runtime_display_summary = _build_display_summary(list(runtime_job_nodes))
            enriched["display_items_claimed"] = _max_runtime_metric(
                enriched.get("display_items_claimed"),
                runtime_display_summary.get("items_claimed"),
                enriched.get("items_claimed"),
            )
            enriched["display_items_running"] = _max_runtime_metric(
                enriched.get("display_items_running"),
                runtime_display_summary.get("display_running"),
                enriched.get("display_active_threads"),
                *([] if not runtime_allow_raw_display_fallback else [enriched.get("items_running")]),
            )
            enriched["display_current_load"] = _max_runtime_metric(
                enriched.get("display_current_load"),
                runtime_display_summary.get("current_load"),
                enriched.get("display_items_running"),
            )
            enriched["display_active_threads"] = _max_runtime_metric(
                enriched.get("display_active_threads"),
                runtime_display_summary.get("active_threads"),
                enriched.get("display_items_running"),
                *([] if not runtime_allow_raw_display_fallback else [enriched.get("items_running")]),
            )
            enriched["display_max_threads"] = _max_runtime_metric(
                enriched.get("display_max_threads"),
                runtime_display_summary.get("max_threads"),
            )
            enriched["display_items_completed"] = int(runtime_display_summary.get("items_completed", enriched.get("display_items_completed", 0)) or 0)
            enriched["display_items_failed"] = int(runtime_display_summary.get("items_failed", enriched.get("display_items_failed", 0)) or 0)
            enriched["display_active_node_codes"] = list(runtime_display_summary.get("active_nodes") or [])
        elif runtime_nodes and not runtime_snapshot_stale_for_live_display:
            enriched["node_stats"] = list(runtime_nodes)
            enriched["distributed_node_stats"] = list(runtime_nodes)
        if runtime_steps and not runtime_snapshot_stale_for_live_display:
            enriched["step_stats"] = list(runtime_steps)
        if not runtime_snapshot_stale_for_live_display:
            runtime_recent_events = _select_runtime_snapshot_events(
                runtime_snapshot,
                job_code=str(runtime_job.get("job_code") or "").strip(),
                job_id=int(runtime_job.get("job_id", 0) or 0),
                event_limit=event_limit,
            )
            if runtime_recent_events:
                enriched["recent_events"] = runtime_recent_events
                current_cycle_token, current_cycle_events = _extract_current_cycle_events(runtime_recent_events)
                enriched["current_cycle_token"] = current_cycle_token
                enriched["current_cycle_events"] = current_cycle_events
                enriched["latest_event"] = runtime_recent_events[0]
                enriched["latest_cycle_event"] = current_cycle_events[0] if current_cycle_events else runtime_recent_events[0]
    aligned_job_code = str(enriched.get("job_code") or enriched.get("runtime_job_code") or "").strip()
    aligned_job_id = int(enriched.get("job_id", 0) or 0)
    aligned_recent_events = _filter_events_for_job(
        list(enriched.get("recent_events") or []),
        job_code=aligned_job_code,
        job_id=aligned_job_id,
        limit=event_limit,
    )
    if list(enriched.get("recent_events") or []) and not aligned_recent_events:
        enriched["recent_events"] = []
        enriched["current_cycle_token"] = ""
        enriched["current_cycle_events"] = []
        enriched["latest_event"] = None
        enriched["latest_cycle_event"] = None
    elif aligned_recent_events:
        enriched["recent_events"] = aligned_recent_events
        current_cycle_token, current_cycle_events = _extract_current_cycle_events(aligned_recent_events)
        enriched["current_cycle_token"] = current_cycle_token
        enriched["current_cycle_events"] = current_cycle_events
        enriched["latest_event"] = aligned_recent_events[0]
        enriched["latest_cycle_event"] = current_cycle_events[0] if current_cycle_events else aligned_recent_events[0]
    enriched["display_items_running"] = _max_runtime_metric(
        enriched.get("display_items_running"),
        enriched.get("display_active_threads"),
        *([] if not runtime_allow_raw_display_fallback else [enriched.get("items_running")]),
    )
    enriched["display_current_load"] = _max_runtime_metric(
        enriched.get("display_current_load"),
        enriched.get("display_items_running"),
        enriched.get("display_active_threads"),
        *([] if not runtime_allow_raw_display_fallback else [enriched.get("items_running")]),
    )
    enriched["display_active_threads"] = _max_runtime_metric(
        enriched.get("display_active_threads"),
        enriched.get("display_items_running"),
        *([] if not runtime_allow_raw_display_fallback else [enriched.get("items_running")]),
    )
    if runtime_missing_participant_attribution:
        enriched["display_items_running"] = 0
        enriched["display_current_load"] = 0
        enriched["display_active_threads"] = 0
        enriched["display_active_node_codes"] = []
        enriched["display_max_threads"] = 0
    enriched["display_max_threads"] = _max_runtime_metric(
        enriched.get("display_max_threads"),
        sum(
            _int_value(item.get("max_threads"))
            for item in list(enriched.get("distributed_node_stats") or [])
            if isinstance(item, dict)
        ),
    )
    return enriched


@db_read_retry()
def get_detect_job_summary(job_id: int, event_limit: int = 20) -> dict | None:
    with get_db() as conn:
        with conn.cursor() as cur:
            _execute_detect_jobs_select(
                cur,
                from_clause="FROM detect_jobs",
                where_clause="id = %s",
                limit_clause="1",
                params=(int(job_id),),
            )
            row = cur.fetchone()
            if not row:
                return None
            return _enrich_active_job_summary_with_runtime(_fetch_job_summary(cur, row, event_limit=event_limit), event_limit=event_limit)


def _select_runtime_aligned_active_job_row(
    job_rows: list[tuple],
    runtime_activity: dict | None = None,
    runtime_snapshot: dict | None = None,
):
    rows = list(job_rows or [])
    if not rows:
        return None
    activity = dict(runtime_activity or {})
    snapshot = dict(runtime_snapshot or {})
    snapshot_job = dict(snapshot.get("job") or {})
    preferred_job_codes = [
        str(snapshot_job.get("job_code") or "").strip(),
        str(activity.get("focus_job_code") or "").strip(),
        *[
            str(item or "").strip()
            for item in list(activity.get("job_codes") or [])
            if str(item or "").strip()
        ],
    ]
    normalized_codes: list[str] = []
    seen_codes: set[str] = set()
    for item in preferred_job_codes:
        if not item or item in seen_codes:
            continue
        normalized_codes.append(item)
        seen_codes.add(item)
    if not normalized_codes:
        return rows[0]
    for preferred_code in normalized_codes:
        for row in rows:
            if str(row[1] or "").strip() == preferred_code:
                return row
    return rows[0]


@db_read_retry()
def get_active_detect_job_summary(event_limit: int = 20) -> dict | None:
    safe_event_limit = max(1, min(int(event_limit or 20), 100))
    cache_key = (safe_event_limit,)
    now_ts = time.monotonic()
    with _ACTIVE_JOB_SUMMARY_CACHE_LOCK:
        cached = _ACTIVE_JOB_SUMMARY_CACHE.get(cache_key)
        if cached and now_ts < float(cached[0] or 0.0):
            return _clone_cacheable_payload(cached[1])

    _maybe_recycle_expired_detect_job_items()
    result: dict | None = None
    with get_db() as conn:
        with conn.cursor() as cur:
            runtime_snapshot = _load_latest_runtime_active_job_snapshot(15)
            runtime_activity = _load_runtime_activity_snapshot(15)
            _execute_detect_jobs_select(
                cur,
                from_clause="FROM detect_jobs",
                where_clause="status IN ('pending', 'running')",
                order_clause="""
                    CASE WHEN status = 'running' THEN 0 ELSE 1 END ASC,
                    COALESCE(started_at, created_at) DESC,
                    id DESC
                """,
                limit_clause="20",
            )
            rows = list(cur.fetchall() or [])
            if not rows:
                runtime_summary = _build_active_job_summary_from_runtime_snapshot(
                    runtime_snapshot,
                    event_limit=safe_event_limit,
                )
                if not runtime_summary:
                    result = None
                else:
                    result = _enrich_active_job_summary_with_runtime(
                        runtime_summary,
                        event_limit=safe_event_limit,
                        runtime_activity=runtime_activity,
                        runtime_snapshot=runtime_snapshot,
                    )
            else:
                row = _select_runtime_aligned_active_job_row(
                    rows,
                    runtime_activity=runtime_activity,
                    runtime_snapshot=runtime_snapshot,
                )
                result = _enrich_active_job_summary_with_runtime(
                    _fetch_job_summary(cur, row, event_limit=safe_event_limit),
                    event_limit=safe_event_limit,
                    runtime_activity=runtime_activity,
                    runtime_snapshot=runtime_snapshot,
                )
    with _ACTIVE_JOB_SUMMARY_CACHE_LOCK:
        _ACTIVE_JOB_SUMMARY_CACHE[cache_key] = (
            time.monotonic() + _ACTIVE_JOB_SUMMARY_CACHE_TTL_SECONDS,
            _clone_cacheable_payload(result),
        )
    return _clone_cacheable_payload(result)


@db_read_retry()
def get_latest_detect_job_summary(
    *,
    event_limit: int = 20,
    statuses: tuple[str, ...] | list[str] | None = None,
    recent_minutes: int | None = None,
) -> dict | None:
    normalized_statuses = [
        str(item or "").strip()
        for item in list(statuses or [])
        if str(item or "").strip()
    ]
    if not normalized_statuses:
        normalized_statuses = ["pending", "running", "completed", "partial_failed", "failed"]

    where_clauses = ["status = ANY(%s)"]
    params: list[object] = [normalized_statuses]
    if recent_minutes is not None:
        safe_recent_minutes = max(1, min(int(recent_minutes or 1), 24 * 60))
        where_clauses.append(
            """
            COALESCE(finished_at, started_at, created_at) >=
            CURRENT_TIMESTAMP - (%s || ' minutes')::interval
            """
        )
        params.append(safe_recent_minutes)

    with get_db() as conn:
        with conn.cursor() as cur:
            _execute_detect_jobs_select(
                cur,
                from_clause="FROM detect_jobs",
                where_clause=" AND ".join(where_clauses),
                order_clause="""
                    CASE
                        WHEN status = 'running' THEN 0
                        WHEN status = 'pending' THEN 1
                        WHEN status = 'completed' THEN 2
                        WHEN status = 'partial_failed' THEN 3
                        WHEN status = 'failed' THEN 4
                        ELSE 5
                    END ASC,
                    COALESCE(finished_at, started_at, created_at) DESC,
                    id DESC
                """,
                limit_clause="1",
                params=tuple(params),
            )
            row = cur.fetchone()
            if not row:
                return None
            return _enrich_active_job_summary_with_runtime(_fetch_job_summary(cur, row, event_limit=event_limit), event_limit=event_limit)


@db_read_retry()
def get_latest_unprojected_detect_job_summary(
    *,
    event_limit: int = 20,
    statuses: tuple[str, ...] | list[str] | None = None,
    recent_minutes: int | None = None,
    source_region: str | None = None,
    target_region: str | None = None,
) -> dict | None:
    normalized_statuses = [
        str(item or "").strip()
        for item in list(statuses or [])
        if str(item or "").strip()
    ]
    if not normalized_statuses:
        normalized_statuses = ["completed", "partial_failed", "failed"]

    normalized_source_region = _normalize_sync_region(
        source_region,
        _normalize_sync_region(settings.sync_source_region, settings.node_region),
    )
    normalized_target_region = _normalize_sync_region(
        target_region,
        _normalize_sync_region(settings.sync_target_region, "overseas"),
    )

    where_clauses = [
        "job.status = ANY(%s)",
        """
        NOT EXISTS (
            SELECT 1
            FROM detect_sync_records AS sync_record
            WHERE sync_record.sync_type = 'detect_result_projection'
              AND sync_record.source_region = %s
              AND sync_record.target_region = %s
              AND (sync_record.payload_json->'projection'->'job'->>'job_id') = job.id::text
        )
        """,
    ]
    params: list[object] = [normalized_statuses, normalized_source_region, normalized_target_region]
    if recent_minutes is not None:
        safe_recent_minutes = max(1, min(int(recent_minutes or 1), 24 * 60))
        where_clauses.append(
            """
            COALESCE(job.finished_at, job.started_at, job.created_at) >=
            CURRENT_TIMESTAMP - (%s || ' minutes')::interval
            """
        )
        params.append(safe_recent_minutes)

    with get_db() as conn:
        with conn.cursor() as cur:
            _execute_detect_jobs_select(
                cur,
                from_clause="FROM detect_jobs AS job",
                where_clause=" AND ".join(where_clauses),
                order_clause="""
                    COALESCE(job.finished_at, job.started_at, job.created_at) DESC,
                    job.id DESC
                """,
                limit_clause="1",
                params=tuple(params),
                alias="job",
            )
            row = cur.fetchone()
            if not row:
                return None
            return _enrich_active_job_summary_with_runtime(_fetch_job_summary(cur, row, event_limit=event_limit), event_limit=event_limit)


def list_detect_jobs(limit: int = 20) -> list[dict]:
    with get_db() as conn:
        with conn.cursor() as cur:
            _execute_detect_jobs_select(
                cur,
                from_clause="FROM detect_jobs",
                order_clause="created_at DESC, id DESC",
                limit_clause="%s",
                params=(max(1, min(int(limit or 20), 100)),),
            )
            rows = cur.fetchall()
            return [_fetch_job_summary(cur, row, event_limit=10) for row in rows]


def list_recent_detect_run_events(limit: int = 20) -> list[dict]:
    safe_limit = max(1, min(int(limit or 20), 200))
    with get_db() as conn:
        with conn.cursor() as cur:
            cur.execute(
                """
                SELECT job_id, node_code, event_type, level, message, payload_json, created_at
                FROM detect_run_events
                ORDER BY created_at DESC, id DESC
                LIMIT %s
                """,
                (safe_limit,),
            )
            rows = cur.fetchall()
    return [
        {
            "job_id": int(row[0]) if row[0] is not None else None,
            "node_code": row[1] or "",
            "event_type": row[2] or "",
            "level": row[3] or "info",
            "message": row[4] or "",
            "payload": _decode_payload(row[5]),
            "created_at": _format_time(row[6]),
        }
        for row in rows
    ]


@db_read_retry()
def get_detect_queue_health(window_minutes: int = 15) -> dict:
    window_minutes = max(5, min(int(window_minutes or 15), 120))
    cache_key = (window_minutes,)
    now_ts = time.monotonic()
    with _DETECT_QUEUE_HEALTH_CACHE_LOCK:
        cached = _DETECT_QUEUE_HEALTH_CACHE.get(cache_key)
        if cached and now_ts < float(cached[0] or 0.0):
            return _clone_cacheable_payload(cached[1])

    active_job = get_active_detect_job_summary(event_limit=10)
    runtime_activity = _load_runtime_activity_snapshot(window_minutes)
    runtime_snapshot = _load_latest_runtime_active_job_snapshot(window_minutes)
    active_job_runtime_code = str(
        (active_job or {}).get("runtime_snapshot_job_code")
        or (active_job or {}).get("runtime_job_code")
        or (active_job or {}).get("job_code")
        or ""
    ).strip()
    if active_job_runtime_code:
        merged_runtime_codes = [active_job_runtime_code] + [
            str(item or "").strip()
            for item in list(runtime_activity.get("job_codes") or [])
            if str(item or "").strip() and str(item or "").strip() != active_job_runtime_code
        ]
        runtime_activity = {
            **dict(runtime_activity or {}),
            "focus_job_code": active_job_runtime_code,
            "job_codes": merged_runtime_codes[:5],
        }

    runtime_snapshot_job = dict((runtime_snapshot or {}).get("job") or {})
    runtime_snapshot_matches_active_job = _runtime_job_matches_active_job(
        active_job,
        runtime_job_code=runtime_snapshot_job.get("job_code"),
        runtime_job_id=runtime_snapshot_job.get("job_id"),
    )
    runtime_activity_matches_active_job = _runtime_job_matches_active_job(
        active_job,
        runtime_job_code=(runtime_activity or {}).get("focus_job_code"),
    )
    if runtime_snapshot and not runtime_snapshot_matches_active_job:
        runtime_snapshot = {}
    if runtime_activity and not runtime_activity_matches_active_job and not runtime_snapshot_matches_active_job:
        runtime_activity = {
            **dict(runtime_activity or {}),
            "focus_job_code": "",
            "job_codes": [],
            "processed_recent": 0,
            "completed_recent": 0,
            "failed_recent": 0,
            "blacklisted_recent": 0,
            "step_code": "",
            "step_stats": {},
            "nodes": {},
        }

    if not active_job:
        result = {
            "window_minutes": window_minutes,
            "has_active_job": False,
            "job": None,
            "queue": {
                "items_total": 0,
                "pending": 0,
                "claimed": 0,
                "running": 0,
                "completed": 0,
                "blacklisted": 0,
                "failed": 0,
                "terminal": 0,
                "terminal_percent": 0,
                "oldest_pending_at": "",
                "oldest_pending_age_minutes": 0,
                "nearest_lease_expiry_at": "",
                "overdue_leases": 0,
                "expiring_soon_leases": 0,
            },
            "throughput": {
                "processed_recent": int(runtime_activity.get("processed_recent", 0) or 0),
                "processed_per_minute": round(int(runtime_activity.get("processed_recent", 0) or 0) / window_minutes, 2),
                "completed_recent": int(runtime_activity.get("completed_recent", 0) or 0),
                "blacklisted_recent": int(runtime_activity.get("blacklisted_recent", 0) or 0),
                "failed_recent": int(runtime_activity.get("failed_recent", 0) or 0),
            },
            "nodes": [],
            "steps": [],
            "runtime_activity": runtime_activity,
        }
        with _DETECT_QUEUE_HEALTH_CACHE_LOCK:
            _DETECT_QUEUE_HEALTH_CACHE[cache_key] = (
                time.monotonic() + _DETECT_QUEUE_HEALTH_CACHE_TTL_SECONDS,
                _clone_cacheable_payload(result),
            )
        return _clone_cacheable_payload(result)

    job_id = int(active_job["job_id"])
    with get_db() as conn:
        with conn.cursor() as cur:
            cur.execute(
                """
                SELECT
                    MIN(create_time) FILTER (WHERE status = 'pending') AS oldest_pending_at,
                    MIN(lease_expires_at) FILTER (WHERE status IN ('claimed', 'running') AND lease_expires_at IS NOT NULL) AS nearest_lease_expiry_at,
                    COUNT(*) FILTER (WHERE status IN ('claimed', 'running') AND lease_expires_at IS NOT NULL AND lease_expires_at < CURRENT_TIMESTAMP) AS overdue_leases,
                    COUNT(*) FILTER (
                        WHERE status IN ('claimed', 'running')
                          AND lease_expires_at IS NOT NULL
                          AND lease_expires_at >= CURRENT_TIMESTAMP
                          AND lease_expires_at < CURRENT_TIMESTAMP + interval '5 minutes'
                    ) AS expiring_soon_leases
                FROM detect_job_items
                WHERE job_id = %s
                """,
                (job_id,),
            )
            lease_row = cur.fetchone()

            cur.execute(
                """
                SELECT
                    COALESCE(NULLIF(node_code, ''), 'unknown') AS node_code,
                    COUNT(*) AS processed_recent,
                    COUNT(*) FILTER (WHERE event_type = 'domain_completed') AS completed_recent,
                    COUNT(*) FILTER (WHERE event_type = 'domain_blacklisted') AS blacklisted_recent,
                    COUNT(*) FILTER (WHERE event_type = 'domain_failed') AS failed_recent
                FROM detect_run_events
                WHERE job_id = %s
                  AND event_type IN ('domain_completed', 'domain_blacklisted', 'domain_failed')
                  AND created_at >= CURRENT_TIMESTAMP - (%s || ' minutes')::interval
                GROUP BY COALESCE(NULLIF(node_code, ''), 'unknown')
                ORDER BY processed_recent DESC, node_code ASC
                """,
                (job_id, window_minutes),
            )
            throughput_rows = cur.fetchall()

            cur.execute(
                """
                SELECT
                    COALESCE(NULLIF(step_code, ''), 'domain_pipeline') AS step_code,
                    COUNT(*) FILTER (
                        WHERE status IN ('completed', 'blacklisted', 'failed')
                          AND finished_at >= CURRENT_TIMESTAMP - (%s || ' minutes')::interval
                    ) AS processed_recent,
                    COUNT(*) FILTER (
                        WHERE status = 'completed'
                          AND finished_at >= CURRENT_TIMESTAMP - (%s || ' minutes')::interval
                    ) AS completed_recent,
                    COUNT(*) FILTER (
                        WHERE status = 'blacklisted'
                          AND finished_at >= CURRENT_TIMESTAMP - (%s || ' minutes')::interval
                    ) AS blacklisted_recent,
                    COUNT(*) FILTER (
                        WHERE status = 'failed'
                          AND finished_at >= CURRENT_TIMESTAMP - (%s || ' minutes')::interval
                    ) AS failed_recent
                FROM detect_job_items
                WHERE job_id = %s
                GROUP BY COALESCE(NULLIF(step_code, ''), 'domain_pipeline')
                ORDER BY step_code ASC
                """,
                (window_minutes, window_minutes, window_minutes, window_minutes, job_id),
            )
            step_throughput_rows = cur.fetchall()
            runtime_display_rows = _load_runtime_display_rows(cur)

    oldest_pending_at = _format_time(lease_row[0]) if lease_row and lease_row[0] else ""
    nearest_lease_expiry_at = _format_time(lease_row[1]) if lease_row and lease_row[1] else ""
    oldest_pending_age_minutes = 0
    if lease_row and lease_row[0]:
        oldest_pending_age_minutes = max(0, int((datetime.now() - lease_row[0]).total_seconds() // 60))

    node_map = {
        str(item.get("node_code") or "unknown"): {
            "node_code": str(item.get("node_code") or "unknown"),
            "items_total": int(item.get("items_total", 0) or 0),
            "items_pending": int(item.get("items_pending", 0) or 0),
            "items_claimed": int(item.get("items_claimed", 0) or 0),
            "items_running": int(item.get("items_running", 0) or 0),
            "items_completed": int(item.get("items_completed", 0) or 0),
            "items_blacklisted": int(item.get("items_blacklisted", 0) or 0),
            "items_failed": int(item.get("items_failed", 0) or 0),
            "processed_recent": 0,
            "processed_per_minute": 0,
            "completed_recent": 0,
            "blacklisted_recent": 0,
            "failed_recent": 0,
            "display_running": _int_value(item.get("display_running")),
            "current_load": _int_value(item.get("current_load")),
            "active_threads": _int_value(item.get("active_threads")),
            "max_threads": _int_value(item.get("max_threads")),
            "region": str(item.get("region") or "").strip(),
            "role": str(item.get("role") or "").strip(),
            "status": str(item.get("status") or "").strip(),
            "last_heartbeat_at": str(item.get("last_heartbeat_at") or "").strip(),
        }
        for item in active_job.get("node_stats") or []
    }
    distributed_nodes = list(active_job.get("distributed_node_stats") or [])
    if distributed_nodes:
        node_map = {
            str(item.get("node_code") or "unknown"): {
                "node_code": str(item.get("node_code") or "unknown"),
                "items_total": _int_value(item.get("items_total")),
                "items_pending": _int_value(item.get("items_pending")),
                "items_claimed": _int_value(item.get("items_claimed")),
                "items_running": _int_value(item.get("items_running")),
                "items_completed": _int_value(item.get("items_completed")),
                "items_blacklisted": _int_value(item.get("items_blacklisted")),
                "items_failed": _int_value(item.get("items_failed")),
                "processed_recent": 0,
                "processed_per_minute": 0,
                "completed_recent": 0,
                "blacklisted_recent": 0,
                "failed_recent": 0,
                "metrics_source": str(item.get("metrics_source") or "").strip(),
                "display_running": _int_value(item.get("display_running")),
                "current_load": _int_value(item.get("current_load")),
                "active_threads": _int_value(item.get("active_threads")),
                "max_threads": _int_value(item.get("max_threads")),
                "region": str(item.get("region") or "").strip(),
                "role": str(item.get("role") or "").strip(),
                "status": str(item.get("status") or "").strip(),
                "last_heartbeat_at": str(item.get("last_heartbeat_at") or "").strip(),
            }
            for item in distributed_nodes
        }
    total_processed_recent = 0
    total_completed_recent = 0
    total_blacklisted_recent = 0
    total_failed_recent = 0
    step_map = {
        str(item.get("step_code") or "domain_pipeline"): {
            **_build_step_bucket(str(item.get("step_code") or "domain_pipeline")),
            "items_total": _int_value(item.get("items_total")),
            "items_pending": _int_value(item.get("items_pending")),
            "items_claimed": _int_value(item.get("items_claimed")),
            "items_running": _int_value(item.get("items_running")),
            "items_completed": _int_value(item.get("items_completed")),
            "items_blacklisted": _int_value(item.get("items_blacklisted")),
            "items_failed": _int_value(item.get("items_failed")),
        }
        for item in active_job.get("step_stats") or []
    }
    for row in throughput_rows:
        node_code = str(row[0] or "unknown")
        bucket = node_map.setdefault(
            node_code,
            {
                "node_code": node_code,
                "items_total": 0,
                "items_pending": 0,
                "items_claimed": 0,
                "items_running": 0,
                "items_completed": 0,
                "items_blacklisted": 0,
                "items_failed": 0,
                "processed_recent": 0,
                "processed_per_minute": 0,
                "completed_recent": 0,
                "blacklisted_recent": 0,
                "failed_recent": 0,
            },
        )
        processed_recent = int(row[1] or 0)
        completed_recent = int(row[2] or 0)
        blacklisted_recent = int(row[3] or 0)
        failed_recent = int(row[4] or 0)
        bucket["processed_recent"] = processed_recent
        bucket["processed_per_minute"] = round(processed_recent / window_minutes, 2)
        bucket["completed_recent"] = completed_recent
        bucket["blacklisted_recent"] = blacklisted_recent
        bucket["failed_recent"] = failed_recent
        total_processed_recent += processed_recent
        total_completed_recent += completed_recent
        total_blacklisted_recent += blacklisted_recent
        total_failed_recent += failed_recent

    for row in list(runtime_display_rows or []):
        runtime_bucket = _build_runtime_display_bucket(row)
        if not runtime_bucket:
            continue
        node_code = str(runtime_bucket.get("node_code") or "unknown")
        bucket = node_map.setdefault(
            node_code,
            {
                "node_code": node_code,
                "items_total": 0,
                "items_pending": 0,
                "items_claimed": 0,
                "items_running": 0,
                "items_completed": 0,
                "items_blacklisted": 0,
                "items_failed": 0,
                "processed_recent": 0,
                "processed_per_minute": 0,
                "completed_recent": 0,
                "blacklisted_recent": 0,
                "failed_recent": 0,
                "metrics_source": "runtime",
                "display_running": 0,
                "current_load": 0,
                "active_threads": 0,
                "max_threads": 0,
                "region": "",
                "role": "",
                "status": "",
                "last_heartbeat_at": "",
            },
        )
        runtime_display_running = _effective_runtime_load(
            items_running=runtime_bucket.get("items_running"),
            active_threads=runtime_bucket.get("active_threads"),
        )
        for key in (
            "items_total",
            "items_pending",
            "items_claimed",
            "items_completed",
            "items_blacklisted",
            "items_failed",
            "current_load",
            "active_threads",
            "max_threads",
        ):
            bucket[key] = max(_int_value(bucket.get(key)), _int_value(runtime_bucket.get(key)))
        bucket["display_running"] = max(
            _int_value(bucket.get("display_running")),
            runtime_display_running,
        )
        bucket["items_running"] = max(
            _int_value(bucket.get("items_running")),
            _int_value(runtime_bucket.get("items_running")),
        )
        for key in ("metrics_source", "region", "role", "status", "last_heartbeat_at"):
            if str(runtime_bucket.get(key) or "").strip():
                bucket[key] = runtime_bucket.get(key)

    for node_code, runtime_bucket in dict(runtime_activity.get("nodes") or {}).items():
        bucket = node_map.setdefault(
            node_code,
            {
                "node_code": node_code,
                "items_total": 0,
                "items_pending": 0,
                "items_claimed": 0,
                "items_running": 0,
                "items_completed": 0,
                "items_blacklisted": 0,
                "items_failed": 0,
                "processed_recent": 0,
                "processed_per_minute": 0,
                "completed_recent": 0,
                "blacklisted_recent": 0,
                "failed_recent": 0,
                "metrics_source": "runtime",
                "display_running": 0,
                "current_load": 0,
                "active_threads": 0,
                "max_threads": 0,
                "region": "",
                "role": "",
                "status": "",
                "last_heartbeat_at": "",
            },
        )
        runtime_processed_recent = int(runtime_bucket.get("processed_recent", 0) or 0)
        if runtime_processed_recent > int(bucket.get("processed_recent", 0) or 0):
            bucket["processed_recent"] = runtime_processed_recent
            bucket["processed_per_minute"] = round(runtime_processed_recent / window_minutes, 2)
            bucket["completed_recent"] = int(runtime_bucket.get("completed_recent", 0) or 0)
            bucket["blacklisted_recent"] = int(runtime_bucket.get("blacklisted_recent", 0) or 0)
            bucket["failed_recent"] = int(runtime_bucket.get("failed_recent", 0) or 0)

    for row in step_throughput_rows:
        step_code = str(row[0] or "domain_pipeline")
        bucket = step_map.setdefault(step_code, _build_step_bucket(step_code))
        processed_recent = int(row[1] or 0)
        completed_recent = int(row[2] or 0)
        blacklisted_recent = int(row[3] or 0)
        failed_recent = int(row[4] or 0)
        bucket["processed_recent"] = processed_recent
        bucket["processed_per_minute"] = round(processed_recent / window_minutes, 2)
        bucket["completed_recent"] = completed_recent
        bucket["blacklisted_recent"] = blacklisted_recent
        bucket["failed_recent"] = failed_recent

    runtime_step_stats = dict(runtime_activity.get("step_stats") or {})
    for runtime_step_code, runtime_step_bucket in runtime_step_stats.items():
        normalized_step_code = str(runtime_step_code or "").strip()
        if not normalized_step_code:
            continue
        bucket = step_map.setdefault(normalized_step_code, _build_step_bucket(normalized_step_code))
        runtime_processed_recent = int(runtime_step_bucket.get("processed_recent", 0) or 0)
        runtime_started_recent = int(runtime_step_bucket.get("started_recent", 0) or 0)
        bucket["started_recent"] = max(int(bucket.get("started_recent", 0) or 0), runtime_started_recent)
        if runtime_processed_recent > int(bucket.get("processed_recent", 0) or 0):
            bucket["processed_recent"] = runtime_processed_recent
            bucket["processed_per_minute"] = round(runtime_processed_recent / window_minutes, 2)
            bucket["completed_recent"] = int(runtime_step_bucket.get("completed_recent", 0) or 0)
            bucket["blacklisted_recent"] = int(runtime_step_bucket.get("blacklisted_recent", 0) or 0)
            bucket["failed_recent"] = int(runtime_step_bucket.get("failed_recent", 0) or 0)
        elif int(bucket.get("processed_recent", 0) or 0) <= 0 and runtime_started_recent > 0:
            bucket["processed_recent"] = runtime_started_recent
            bucket["processed_per_minute"] = round(runtime_started_recent / window_minutes, 2)

    runtime_step_code = str(runtime_activity.get("step_code") or "").strip()
    runtime_processed_recent = int(runtime_activity.get("processed_recent", 0) or 0)
    if runtime_step_code and runtime_processed_recent > 0 and runtime_step_code not in runtime_step_stats:
        runtime_bucket = step_map.setdefault(runtime_step_code, _build_step_bucket(runtime_step_code))
        if runtime_processed_recent > int(runtime_bucket.get("processed_recent", 0) or 0):
            runtime_bucket["processed_recent"] = runtime_processed_recent
            runtime_bucket["processed_per_minute"] = round(runtime_processed_recent / window_minutes, 2)
            runtime_bucket["completed_recent"] = int(runtime_activity.get("completed_recent", 0) or 0)
            runtime_bucket["blacklisted_recent"] = int(runtime_activity.get("blacklisted_recent", 0) or 0)
            runtime_bucket["failed_recent"] = int(runtime_activity.get("failed_recent", 0) or 0)

    total_processed_recent = max(total_processed_recent, int(runtime_activity.get("processed_recent", 0) or 0))
    total_completed_recent = max(total_completed_recent, int(runtime_activity.get("completed_recent", 0) or 0))
    total_blacklisted_recent = max(total_blacklisted_recent, int(runtime_activity.get("blacklisted_recent", 0) or 0))
    total_failed_recent = max(total_failed_recent, int(runtime_activity.get("failed_recent", 0) or 0))

    runtime_snapshot_job = dict(runtime_snapshot.get("job") or {})
    runtime_snapshot_queue = dict((runtime_snapshot.get("queue_health") or {}).get("queue") or {})
    runtime_snapshot_nodes = list((runtime_snapshot.get("queue_health") or {}).get("nodes") or [])
    runtime_snapshot_steps = list((runtime_snapshot.get("queue_health") or {}).get("steps") or [])
    runtime_snapshot_job_code = str(runtime_snapshot_job.get("job_code") or "").strip()

    items_total = int(active_job.get("items_total", 0) or 0)
    pending_items = int(active_job.get("items_pending", 0) or 0)
    claimed_items = int(active_job.get("items_claimed", 0) or 0)
    running_items = int(active_job.get("items_running", 0) or 0)
    completed_items = int(active_job.get("items_completed", 0) or 0)
    blacklisted_items = int(active_job.get("items_blacklisted", 0) or 0)
    failed_items = int(active_job.get("items_failed", 0) or 0)
    display_claimed = int(active_job.get("display_items_claimed", active_job.get("items_claimed", 0)) or 0)
    display_running = int(
        active_job.get("display_active_threads", active_job.get("display_items_running", 0))
        or 0
    )
    progress_percent = float(active_job.get("progress_percent", 0) or 0)

    if runtime_snapshot_queue:
        runtime_items_total = int(runtime_snapshot_queue.get("items_total", 0) or 0)
        if runtime_items_total > 0:
            items_total = runtime_items_total
            pending_items = int(runtime_snapshot_queue.get("pending", 0) or 0)
            claimed_items = int(runtime_snapshot_queue.get("claimed", 0) or 0)
            running_items = int(runtime_snapshot_queue.get("running", 0) or 0)
            completed_items = int(runtime_snapshot_queue.get("completed", 0) or 0)
            blacklisted_items = int(runtime_snapshot_queue.get("blacklisted", 0) or 0)
            failed_items = int(runtime_snapshot_queue.get("failed", 0) or 0)
            display_claimed = int(runtime_snapshot_queue.get("display_claimed", claimed_items) or 0)
            display_running = int(runtime_snapshot_queue.get("display_running", display_running) or 0)
            progress_percent = float(runtime_snapshot_job.get("progress_percent", progress_percent) or 0)

            if runtime_snapshot_nodes:
                node_map = {
                    str(item.get("node_code") or "unknown"): {
                        "node_code": str(item.get("node_code") or "unknown"),
                        "items_total": _int_value(item.get("items_total")),
                        "items_pending": _int_value(item.get("items_pending")),
                        "items_claimed": _int_value(item.get("items_claimed")),
                        "items_running": _int_value(item.get("items_running")),
                        "items_completed": _int_value(item.get("items_completed")),
                        "items_blacklisted": _int_value(item.get("items_blacklisted")),
                        "items_failed": _int_value(item.get("items_failed")),
                        "processed_recent": _int_value(item.get("processed_recent")),
                        "processed_per_minute": float(item.get("processed_per_minute", 0) or 0),
                        "completed_recent": _int_value(item.get("completed_recent")),
                        "blacklisted_recent": _int_value(item.get("blacklisted_recent")),
                        "failed_recent": _int_value(item.get("failed_recent")),
                        "metrics_source": str(item.get("metrics_source") or "").strip(),
                        "display_running": _int_value(item.get("display_running")),
                        "current_load": _int_value(item.get("current_load")),
                        "active_threads": _int_value(item.get("active_threads")),
                        "max_threads": _int_value(item.get("max_threads")),
                        "region": str(item.get("region") or "").strip(),
                        "role": str(item.get("role") or "").strip(),
                        "status": str(item.get("status") or "").strip(),
                        "last_heartbeat_at": str(item.get("last_heartbeat_at") or "").strip(),
                    }
                    for item in runtime_snapshot_nodes
                }
            if runtime_snapshot_steps:
                step_map = {
                    str(item.get("step_code") or "domain_pipeline"): {
                        **_build_step_bucket(str(item.get("step_code") or "domain_pipeline")),
                        "items_total": _int_value(item.get("items_total")),
                        "items_pending": _int_value(item.get("items_pending")),
                        "items_claimed": _int_value(item.get("items_claimed")),
                        "items_running": _int_value(item.get("items_running")),
                        "items_completed": _int_value(item.get("items_completed")),
                        "items_blacklisted": _int_value(item.get("items_blacklisted")),
                        "items_failed": _int_value(item.get("items_failed")),
                        "processed_recent": _int_value(item.get("processed_recent")),
                        "processed_per_minute": float(item.get("processed_per_minute", 0) or 0),
                        "completed_recent": _int_value(item.get("completed_recent")),
                        "blacklisted_recent": _int_value(item.get("blacklisted_recent")),
                        "failed_recent": _int_value(item.get("failed_recent")),
                    }
                    for item in runtime_snapshot_steps
                }

    for row in list(runtime_display_rows or []):
        runtime_bucket = _build_runtime_display_bucket(row)
        if not runtime_bucket:
            continue
        node_code = str(runtime_bucket.get("node_code") or "unknown")
        bucket = node_map.setdefault(
            node_code,
            {
                "node_code": node_code,
                "items_total": 0,
                "items_pending": 0,
                "items_claimed": 0,
                "items_running": 0,
                "items_completed": 0,
                "items_blacklisted": 0,
                "items_failed": 0,
                "processed_recent": 0,
                "processed_per_minute": 0,
                "completed_recent": 0,
                "blacklisted_recent": 0,
                "failed_recent": 0,
                "metrics_source": "runtime",
                "display_running": 0,
                "current_load": 0,
                "active_threads": 0,
                "max_threads": 0,
                "region": "",
                "role": "",
                "status": "",
                "last_heartbeat_at": "",
            },
        )
        runtime_display_running = _effective_runtime_load(
            items_running=runtime_bucket.get("items_running"),
            active_threads=runtime_bucket.get("active_threads"),
        )
        for key in (
            "items_total",
            "items_pending",
            "items_claimed",
            "items_completed",
            "items_blacklisted",
            "items_failed",
            "current_load",
            "active_threads",
            "max_threads",
        ):
            bucket[key] = max(_int_value(bucket.get(key)), _int_value(runtime_bucket.get(key)))
        bucket["display_running"] = max(
            _int_value(bucket.get("display_running")),
            runtime_display_running,
        )
        bucket["items_running"] = max(
            _int_value(bucket.get("items_running")),
            _int_value(runtime_bucket.get("items_running")),
        )
        for key in ("metrics_source", "region", "role", "status", "last_heartbeat_at"):
            if str(runtime_bucket.get(key) or "").strip():
                bucket[key] = runtime_bucket.get(key)

    for node_code, runtime_bucket in dict(runtime_activity.get("nodes") or {}).items():
        bucket = node_map.setdefault(
            node_code,
            {
                "node_code": node_code,
                "items_total": 0,
                "items_pending": 0,
                "items_claimed": 0,
                "items_running": 0,
                "items_completed": 0,
                "items_blacklisted": 0,
                "items_failed": 0,
                "processed_recent": 0,
                "processed_per_minute": 0,
                "completed_recent": 0,
                "blacklisted_recent": 0,
                "failed_recent": 0,
                "metrics_source": "runtime",
            },
        )
        runtime_processed_recent = int(runtime_bucket.get("processed_recent", 0) or 0)
        if runtime_processed_recent > int(bucket.get("processed_recent", 0) or 0):
            bucket["processed_recent"] = runtime_processed_recent
            bucket["processed_per_minute"] = round(runtime_processed_recent / window_minutes, 2)
            bucket["completed_recent"] = int(runtime_bucket.get("completed_recent", 0) or 0)
            bucket["blacklisted_recent"] = int(runtime_bucket.get("blacklisted_recent", 0) or 0)
            bucket["failed_recent"] = int(runtime_bucket.get("failed_recent", 0) or 0)
        if not str(bucket.get("metrics_source") or "").strip():
            bucket["metrics_source"] = "runtime"

    for bucket in node_map.values():
        active_threads = int(bucket.get("active_threads", 0) or 0)
        current_load = int(bucket.get("current_load", 0) or 0)
        display_running_value = max(
            int(bucket.get("display_running", 0) or 0),
            active_threads,
            current_load,
        )
        bucket["display_running"] = display_running_value
        if active_threads > 0:
            bucket["items_running"] = min(
                max(int(bucket.get("items_running", 0) or 0), active_threads),
                active_threads,
            )
        bucket["current_load"] = max(current_load, active_threads)

    for bucket in step_map.values():
        runtime_bucket = dict(runtime_activity.get("step_stats") or {}).get(str(bucket.get("step_code") or "").strip()) or {}
        if runtime_bucket:
            bucket["started_recent"] = max(
                int(bucket.get("started_recent", 0) or 0),
                int(runtime_bucket.get("started_recent", 0) or 0),
            )
            runtime_processed_recent = int(runtime_bucket.get("processed_recent", 0) or 0)
            if runtime_processed_recent > int(bucket.get("processed_recent", 0) or 0):
                bucket["processed_recent"] = runtime_processed_recent
                bucket["processed_per_minute"] = round(runtime_processed_recent / window_minutes, 2)
                bucket["completed_recent"] = int(runtime_bucket.get("completed_recent", 0) or 0)
                bucket["blacklisted_recent"] = int(runtime_bucket.get("blacklisted_recent", 0) or 0)
                bucket["failed_recent"] = int(runtime_bucket.get("failed_recent", 0) or 0)

    snapshot_created_at = _snapshot_created_at(runtime_snapshot)
    raw_nodes = list(node_map.values())
    live_nodes = _filter_live_runtime_nodes(raw_nodes, snapshot_created_at=snapshot_created_at)
    raw_participant_count = sum(
        1
        for item in raw_nodes
        if str(item.get("node_code") or "").strip()
        and str(item.get("node_code") or "").strip() != "unassigned"
    )
    live_participant_count = sum(
        1
        for item in live_nodes
        if str(item.get("node_code") or "").strip()
        and str(item.get("node_code") or "").strip() != "unassigned"
    )
    dropped_runtime_node_count = max(0, raw_participant_count - live_participant_count)
    if raw_nodes:
        node_map = {
            str(item.get("node_code") or "unknown"): dict(item)
            for item in live_nodes
        }

    normalized_running_items = sum(
        int(item.get("items_running", 0) or 0)
        for item in node_map.values()
        if str(item.get("node_code") or "").strip() and str(item.get("node_code") or "").strip() != "unassigned"
    )
    normalized_display_running = sum(
        int(item.get("display_running", 0) or 0)
        for item in node_map.values()
        if str(item.get("node_code") or "").strip() and str(item.get("node_code") or "").strip() != "unassigned"
    )
    # Prefer normalized per-node sums once runtime heartbeats are present.
    # Stale snapshot queue rows can temporarily over-report thread load from an
    # older cycle, which is exactly what misleads the ops dashboard.
    if normalized_running_items > 0 or dropped_runtime_node_count > 0:
        running_items = normalized_running_items
    if normalized_display_running > 0 or dropped_runtime_node_count > 0:
        display_running = normalized_display_running

    nodes = sorted(
        node_map.values(),
        key=lambda item: (
            -int(item.get("processed_recent", 0) or 0),
            -int(item.get("display_running", item.get("items_running", 0)) or 0),
            -int(item.get("items_running", 0) or 0),
            str(item.get("node_code") or ""),
        ),
    )
    terminal = completed_items + blacklisted_items + failed_items
    steps = order_step_buckets(list(step_map.values()))
    result = {
        "window_minutes": window_minutes,
        "has_active_job": True,
        "job": {
            "job_id": job_id,
            "job_code": active_job.get("job_code", ""),
            "runtime_job_code": runtime_snapshot_job_code or str(runtime_activity.get("focus_job_code") or "").strip(),
            "runtime_job_codes": list(runtime_activity.get("job_codes") or []),
            "status": active_job.get("status", ""),
            "progress_percent": progress_percent,
        },
        "queue": {
            "items_total": items_total,
            "pending": pending_items,
            "claimed": claimed_items,
            "running": running_items,
            "display_claimed": display_claimed,
            "display_running": display_running,
            "completed": completed_items,
            "blacklisted": blacklisted_items,
            "failed": failed_items,
            "terminal": terminal,
            "terminal_percent": round((terminal / items_total) * 100, 2) if items_total else 0,
            "oldest_pending_at": oldest_pending_at,
            "oldest_pending_age_minutes": oldest_pending_age_minutes,
            "nearest_lease_expiry_at": nearest_lease_expiry_at,
            "overdue_leases": int(lease_row[2] or 0) if lease_row else 0,
            "expiring_soon_leases": int(lease_row[3] or 0) if lease_row else 0,
        },
        "throughput": {
            "processed_recent": total_processed_recent,
            "processed_per_minute": round(total_processed_recent / window_minutes, 2),
            "completed_recent": total_completed_recent,
            "blacklisted_recent": total_blacklisted_recent,
            "failed_recent": total_failed_recent,
        },
        "nodes": nodes,
        "steps": steps,
        "runtime_activity": runtime_activity,
        "runtime_snapshot_backlog": dict(runtime_snapshot.get("backlog") or {}),
    }
    with _DETECT_QUEUE_HEALTH_CACHE_LOCK:
        _DETECT_QUEUE_HEALTH_CACHE[cache_key] = (
            time.monotonic() + _DETECT_QUEUE_HEALTH_CACHE_TTL_SECONDS,
            _clone_cacheable_payload(result),
        )
    return _clone_cacheable_payload(result)


def get_detect_capacity_plan(*, queue_health: dict | None = None, online_worker_nodes: int = 0, target_finish_hours: int = 6) -> dict:
    queue_health = queue_health or get_detect_queue_health(window_minutes=15)
    target_finish_hours = max(1, min(int(target_finish_hours or 6), 72))
    online_worker_nodes = max(0, int(online_worker_nodes or 0))

    if not queue_health.get("has_active_job"):
        return {
            "has_active_job": False,
            "online_worker_nodes": online_worker_nodes,
            "target_finish_hours": target_finish_hours,
            "estimated_hours_remaining": 0,
            "recommended_total_workers": max(1, online_worker_nodes),
            "recommended_additional_workers": 0,
            "current_processed_per_hour": 0,
            "pending_items": 0,
            "terminal_items": 0,
            "summary": "当前没有活跃任务，无需扩容建议。",
        }

    queue = queue_health.get("queue") or {}
    throughput = queue_health.get("throughput") or {}
    pending_items = int(queue.get("pending", 0) or 0)
    claimed_items = int(queue.get("claimed", 0) or 0)
    running_items = int(queue.get("running", 0) or 0)
    remaining_items = pending_items + claimed_items + running_items
    current_processed_per_hour = round(float(throughput.get("processed_per_minute", 0) or 0) * 60, 2)
    estimated_hours_remaining = round((remaining_items / current_processed_per_hour), 2) if current_processed_per_hour > 0 else 0

    recommended_total_workers = max(1, online_worker_nodes or 1)
    recommended_additional_workers = 0
    if remaining_items > 0 and target_finish_hours > 0:
        required_per_hour = remaining_items / target_finish_hours
        if current_processed_per_hour > 0 and max(1, online_worker_nodes) > 0:
            per_worker_per_hour = current_processed_per_hour / max(1, online_worker_nodes)
            recommended_total_workers = max(1, int(math.ceil(required_per_hour / per_worker_per_hour)))
            recommended_additional_workers = max(0, recommended_total_workers - online_worker_nodes)
        elif remaining_items > 0:
            recommended_total_workers = max(1, online_worker_nodes or 1)
            recommended_additional_workers = 0

    summary = (
        f"当前在线 Worker {online_worker_nodes} 台，近窗吞吐约 {current_processed_per_hour} 项/小时，"
        f"剩余待处理约 {remaining_items} 项，预计还需 {estimated_hours_remaining} 小时。"
    )
    if recommended_additional_workers > 0:
        summary = (
            f"{summary} 若希望在 {target_finish_hours} 小时内收敛，建议总 Worker 数达到 "
            f"{recommended_total_workers} 台，至少再加 {recommended_additional_workers} 台。"
        )
    else:
        summary = f"{summary} 按当前目标 {target_finish_hours} 小时看，现有 Worker 数量暂时够用。"

    return {
        "has_active_job": True,
        "online_worker_nodes": online_worker_nodes,
        "target_finish_hours": target_finish_hours,
        "estimated_hours_remaining": estimated_hours_remaining,
        "recommended_total_workers": recommended_total_workers,
        "recommended_additional_workers": recommended_additional_workers,
        "current_processed_per_hour": current_processed_per_hour,
        "pending_items": pending_items,
        "remaining_items": remaining_items,
        "terminal_items": int(queue.get("terminal", 0) or 0),
        "summary": summary,
    }


def process_detect_pipeline(limit: int = 5000, job_id: int | None = None) -> dict:
    safe_limit = max(1, min(int(limit or 5000), 5000))
    normalized_job_id = int(job_id or 0) if job_id not in (None, "", 0, "0") else 0
    summary = {
        "processed_items": 0,
        "advanced_items": 0,
        "retried_items": 0,
        "terminal_items": 0,
        "skipped_items": 0,
        "created_next_items": 0,
        "retry_reset_items": 0,
        "legacy_items_upgraded": 0,
        "legacy_items_completed": 0,
        "job_ids": [],
    }

    with get_db() as conn:
        conn.autocommit = False
        with conn.cursor() as cur:
            settings_payload = get_settings_payload()
            touched_job_ids: set[int] = set()
            cur.execute(
                """
                SELECT item.id, item.job_id, item.domain_id
                FROM detect_job_items AS item
                JOIN detect_jobs AS job
                  ON job.id = item.job_id
                WHERE job.task_mode = 'domain_pipeline'
                  AND (%s = 0 OR item.job_id = %s)
                  AND job.status IN ('pending', 'running', 'completed', 'failed', 'partial_failed')
                  AND COALESCE(item.step_code, '') = ''
                  AND item.status = 'pending'
                ORDER BY item.id ASC
                LIMIT %s
                FOR UPDATE SKIP LOCKED
                """,
                (normalized_job_id, normalized_job_id, safe_limit),
            )
            legacy_rows = cur.fetchall()
            for row in legacy_rows:
                item_id = int(row[0])
                legacy_job_id = int(row[1])
                domain_id = int(row[2])
                domain_snapshot = _load_domain_pipeline_snapshot(cur, domain_id)
                if not domain_snapshot:
                    continue
                step_code, step_payload = resolve_initial_domain_pipeline_item(
                    domain_snapshot,
                    settings_payload=settings_payload,
                )
                if step_code and step_payload:
                    cur.execute(
                        """
                        UPDATE detect_job_items
                        SET step_code = %s,
                            step_payload_json = %s::jsonb,
                            updated_at = CURRENT_TIMESTAMP
                        WHERE id = %s
                        """,
                        (step_code, json.dumps(step_payload, ensure_ascii=False), item_id),
                    )
                    touched_job_ids.add(legacy_job_id)
                    summary["processed_items"] += 1
                    summary["legacy_items_upgraded"] += 1
                    continue

                processed_payload = {
                    "controller_processed": True,
                    "controller_action": "pipeline_already_complete",
                    "controller_processed_at": datetime.now().isoformat(timespec="seconds"),
                }
                cur.execute(
                    """
                    UPDATE detect_job_items
                    SET status = 'completed',
                        finished_at = COALESCE(finished_at, CURRENT_TIMESTAMP),
                        result_payload_json = %s::jsonb,
                        updated_at = CURRENT_TIMESTAMP
                    WHERE id = %s
                    """,
                    (json.dumps(processed_payload, ensure_ascii=False), item_id),
                )
                touched_job_ids.add(legacy_job_id)
                summary["processed_items"] += 1
                summary["legacy_items_completed"] += 1

            cur.execute(
                """
                SELECT item.id, item.job_id, item.domain_id, item.step_code, item.status, item.attempt_count, item.result_payload_json, job.task_mode
                FROM detect_job_items AS item
                JOIN detect_jobs AS job
                  ON job.id = item.job_id
                WHERE job.task_mode IN ('domain_pipeline', 'single_step')
                  AND (%s = 0 OR item.job_id = %s)
                  AND job.status IN ('pending', 'running', 'completed', 'failed', 'partial_failed')
                  AND COALESCE(item.step_code, '') <> ''
                  AND item.status IN ('completed', 'blacklisted', 'failed')
                  AND COALESCE(item.result_payload_json->>'controller_processed', 'false') <> 'true'
                ORDER BY COALESCE(item.finished_at, item.updated_at) ASC, item.id ASC
                LIMIT %s
                FOR UPDATE SKIP LOCKED
                """,
                (normalized_job_id, normalized_job_id, safe_limit),
            )
            rows = cur.fetchall()

            for row in rows:
                item_id = int(row[0])
                job_id = int(row[1])
                domain_id = int(row[2])
                current_step_code = str(row[3] or "").strip()
                item_status = str(row[4] or "").strip()
                attempt_count = int(row[5] or 0)
                result_payload = _decode_payload(row[6])
                task_mode = str(row[7] or "").strip()
                domain_snapshot = _load_domain_pipeline_snapshot(cur, domain_id)
                if not domain_snapshot:
                    cur.execute(
                        """
                        UPDATE detect_job_items
                        SET result_payload_json = jsonb_set(COALESCE(result_payload_json, '{}'::jsonb), '{controller_processed}', 'true'::jsonb, true),
                            updated_at = CURRENT_TIMESTAMP
                        WHERE id = %s
                        """,
                        (item_id,),
                    )
                    summary["processed_items"] += 1
                    summary["skipped_items"] += 1
                    continue

                if task_mode == "single_step":
                    result_field_name = str(
                        result_payload.get("field_name") or _PIPELINE_STEP_FIELDS.get(current_step_code) or ""
                    ).strip()
                    persisted_payload = {
                        key: value
                        for key, value in dict(result_payload or {}).items()
                        if key not in {"controller_processed", "controller_action", "controller_processed_at"}
                    }
                    action = "single_step_ignored"
                    if result_field_name:
                        persisted_payload.setdefault("step_code", current_step_code)
                        persisted_payload.setdefault("step_name", _step_label(current_step_code))
                        persisted_payload.setdefault("field_name", result_field_name)
                        persisted_payload.setdefault("task_mode", "single_step")
                        persisted_payload.setdefault("domain_id", domain_id)
                        _upsert_domain_detection_field(cur, domain_id, result_field_name, persisted_payload)
                        action = "single_step_persisted"
                    processed_payload = {
                        **dict(result_payload or {}),
                        "controller_processed": True,
                        "controller_action": action,
                        "controller_processed_at": datetime.now().isoformat(timespec="seconds"),
                    }
                    cur.execute(
                        """
                        UPDATE detect_job_items
                        SET result_payload_json = %s::jsonb,
                            updated_at = CURRENT_TIMESTAMP
                        WHERE id = %s
                        """,
                        (json.dumps(processed_payload, ensure_ascii=False), item_id),
                    )
                    touched_job_ids.add(job_id)
                    summary["processed_items"] += 1
                    if action == "single_step_persisted":
                        summary["advanced_items"] += 1
                    else:
                        summary["skipped_items"] += 1
                    continue

                result_message = str(result_payload.get("message") or "").strip()
                outcome = _classify_pipeline_item_outcome(
                    item_status=item_status,
                    result_payload=result_payload,
                    step_code=current_step_code,
                    attempt_count=attempt_count,
                )
                action = str(outcome.get("action") or "pass").strip()
                result_state = str(outcome.get("result_state") or "").strip().lower()
                retry_limit = int(outcome.get("retry_limit", 0) or 0)

                if action == "retry":
                    retry_payload = {
                        **result_payload,
                        "controller_action": "retry",
                        "controller_processed": False,
                        "controller_retry_count": int(attempt_count),
                        "controller_reason_code": str(outcome.get("reason_code") or "external_retry"),
                    }
                    cur.execute(
                        """
                        UPDATE detect_job_items
                        SET status = 'pending',
                            claimed_by = '',
                            claim_token = '',
                            lease_expires_at = NULL,
                            finished_at = NULL,
                            last_error = %s,
                            result_payload_json = %s::jsonb,
                            updated_at = CURRENT_TIMESTAMP
                        WHERE id = %s
                        """,
                        (
                            result_message or f"{_step_label(current_step_code)} 执行失败，等待重试",
                            json.dumps(retry_payload, ensure_ascii=False),
                            item_id,
                        ),
                    )
                    _append_detect_job_event_with_cursor(
                        cur,
                        job_id,
                        event_type="step_retry_scheduled",
                        message=f"{domain_snapshot['domain']} 的 {_step_label(current_step_code)} 已重投",
                        payload={
                            "domain_id": domain_id,
                            "domain": domain_snapshot["domain"],
                            "step_code": current_step_code,
                            "attempt_count": attempt_count,
                            "retry_limit": retry_limit,
                            "result_state": result_state,
                        },
                    )
                    touched_job_ids.add(job_id)
                    summary["processed_items"] += 1
                    summary["retried_items"] += 1
                    summary["retry_reset_items"] += 1
                    continue

                if action == "pass":
                    next_step_code = resolve_domain_pipeline_step(
                        domain_snapshot,
                        settings_payload=settings_payload,
                        after_step_code=current_step_code,
                    )
                    if next_step_code:
                        next_payload = _build_step_payload(
                            step_code=next_step_code,
                            domain_snapshot=domain_snapshot,
                            settings_payload=settings_payload,
                        )
                        cur.execute(
                            """
                            INSERT INTO detect_job_items (job_id, domain_id, step_code, status, step_payload_json)
                            VALUES (%s, %s, %s, 'pending', %s::jsonb)
                            ON CONFLICT (job_id, domain_id, step_code) DO NOTHING
                            RETURNING id
                            """,
                            (
                                job_id,
                                domain_id,
                                next_step_code,
                                json.dumps(next_payload, ensure_ascii=False),
                            ),
                        )
                        created_next = bool(cur.fetchone())
                        action = "pass"
                        summary["advanced_items"] += 1
                        if created_next:
                            summary["created_next_items"] += 1
                        _append_detect_job_event_with_cursor(
                            cur,
                            job_id,
                            event_type="pipeline_step_advanced",
                            message=f"{domain_snapshot['domain']} 进入下一步骤: {_step_label(next_step_code)}",
                            payload={
                                "domain_id": domain_id,
                                "domain": domain_snapshot["domain"],
                                "current_step_code": current_step_code,
                                "next_step_code": next_step_code,
                            },
                        )
                    else:
                        action = "pipeline_completed"
                        _append_detect_job_event_with_cursor(
                            cur,
                            job_id,
                            event_type="pipeline_completed",
                            message=f"{domain_snapshot['domain']} 本次 pipeline 已完成",
                            payload={
                                "domain_id": domain_id,
                                "domain": domain_snapshot["domain"],
                                "current_step_code": current_step_code,
                            },
                        )

                elif action == "black_hit":
                    _append_detect_job_event_with_cursor(
                        cur,
                        job_id,
                        event_type="pipeline_black_hit",
                        message=f"{domain_snapshot['domain']} 在 {_step_label(current_step_code)} 命中黑名单并终止后续步骤",
                        level="warning",
                        payload={
                            "domain_id": domain_id,
                            "domain": domain_snapshot["domain"],
                            "current_step_code": current_step_code,
                            "result_state": result_state or "blacklisted",
                        },
                    )
                elif action == "reject":
                    _append_detect_job_event_with_cursor(
                        cur,
                        job_id,
                        event_type="pipeline_rejected",
                        message=f"{domain_snapshot['domain']} 在 {_step_label(current_step_code)} 被判定终止",
                        level="warning",
                        payload={
                            "domain_id": domain_id,
                            "domain": domain_snapshot["domain"],
                            "current_step_code": current_step_code,
                            "result_state": result_state or "failed",
                            "attempt_count": attempt_count,
                            "retry_limit": retry_limit,
                            "reason_code": str(outcome.get("reason_code") or ""),
                        },
                    )

                processed_payload = {
                    **result_payload,
                    "controller_processed": True,
                    "controller_action": action,
                    "controller_reason_code": str(outcome.get("reason_code") or ""),
                    "controller_processed_at": datetime.now().isoformat(timespec="seconds"),
                }
                if action == "pass":
                    processed_payload["controller_result_state"] = result_state or "passed"
                cur.execute(
                    """
                    UPDATE detect_job_items
                    SET result_payload_json = %s::jsonb,
                        updated_at = CURRENT_TIMESTAMP
                    WHERE id = %s
                    """,
                    (json.dumps(processed_payload, ensure_ascii=False), item_id),
                )
                touched_job_ids.add(job_id)
                summary["processed_items"] += 1
                if action in {"black_hit", "reject", "pipeline_completed"}:
                    summary["terminal_items"] += 1

            for job_id in touched_job_ids:
                _refresh_detect_job_status_with_cursor(cur, job_id)
                summary["job_ids"].append(job_id)

        conn.commit()

    summary["job_ids"] = sorted(set(int(item) for item in summary["job_ids"]))
    return summary


def process_detect_pipeline_now(limit: int = 2000, job_id: int | None = None) -> tuple[bool, str, dict]:
    attempts = 3
    delay_seconds = 0.05
    for attempt in range(1, attempts + 1):
        try:
            data = process_detect_pipeline(limit=limit, job_id=job_id)
            processed = int(data.get("processed_items", 0) or 0)
            advanced = int(data.get("advanced_items", 0) or 0)
            retried = int(data.get("retried_items", 0) or 0)
            message = f"pipeline 处理完成：processed={processed}, advanced={advanced}, retried={retried}"
            if attempt > 1:
                message = f"{message}（deadlock 自动重试 {attempt - 1} 次后成功）"
            return True, message, {"action": "process_pipeline", "retry_attempts": attempt - 1, **data}
        except Exception as exc:
            if not is_retryable_db_error(exc) or attempt >= attempts:
                raise
            time.sleep(delay_seconds)
            delay_seconds *= 2

    data = process_detect_pipeline(limit=limit, job_id=job_id)
    processed = int(data.get("processed_items", 0) or 0)
    advanced = int(data.get("advanced_items", 0) or 0)
    retried = int(data.get("retried_items", 0) or 0)
    message = f"pipeline 处理完成：processed={processed}, advanced={advanced}, retried={retried}"
    return True, message, {"action": "process_pipeline", **data}


def _append_detect_job_event_with_cursor(
    cur,
    job_id: int,
    *,
    event_type: str,
    message: str,
    level: str = "info",
    payload: dict | None = None,
    node_code: str | None = None,
) -> bool:
    cur.execute(
        "SELECT 1 FROM detect_jobs WHERE id = %s LIMIT 1",
        (int(job_id),),
    )
    if cur.fetchone() is None:
        return False
    cur.execute(
        """
        INSERT INTO detect_run_events (job_id, node_code, event_type, level, message, payload_json)
        VALUES (%s, %s, %s, %s, %s, %s::jsonb)
        """,
        (
            int(job_id),
            node_code or settings.node_code,
            str(event_type or "").strip() or "info",
            str(level or "info").strip() or "info",
            str(message or "").strip(),
            json.dumps(payload or {}, ensure_ascii=False),
        ),
    )
    return True


def append_detect_job_event(
    job_id: int,
    *,
    event_type: str,
    message: str,
    level: str = "info",
    payload: dict | None = None,
    node_code: str | None = None,
) -> None:
    inserted = False
    with get_db() as conn:
        with conn.cursor() as cur:
            inserted = _append_detect_job_event_with_cursor(
                cur,
                job_id,
                event_type=event_type,
                message=message,
                level=level,
                payload=payload,
                node_code=node_code,
            )
        if inserted:
            conn.commit()
    try:
        push_debug_event(
            service="detect-job",
            event_type=str(event_type or "").strip() or "info",
            level=str(level or "info").strip() or "info",
            message=str(message or "").strip(),
            payload={
                "job_id": int(job_id),
                "node_code": node_code or settings.node_code,
                **(payload or {}),
            },
        )
    except Exception:
        pass


def create_detect_job_if_needed(limit: int = 1000, created_by: str = "system", step_code: str | None = None) -> dict | None:
    job_definition = resolve_detect_job_definition(step_code)
    existing = get_active_detect_job_summary()
    if existing:
        existing_task_mode = str(existing.get("task_mode") or "").strip()
        existing_step_code = normalize_detect_step_code(existing.get("step_code"))
        if (
            existing_task_mode == str(job_definition["task_mode"]).strip()
            and existing_step_code == normalize_detect_step_code(job_definition["step_code"])
        ):
            return existing
    settings_payload = get_settings_payload()

    with get_db() as conn:
        conn.autocommit = False
        with conn.cursor() as cur:
            cur.execute(_selection_sql(), (max(1, int(limit or 1000)),))
            domain_ids = [row[0] for row in cur.fetchall()]
            if not domain_ids:
                conn.rollback()

    if not domain_ids:
        try:
            from app.services.sync_push_service import pull_detect_task_batch_now

            pull_detect_task_batch_now(limit=max(1, int(limit or 1000)))
        except Exception:
            pass

        with get_db() as conn:
            conn.autocommit = False
            with conn.cursor() as cur:
                cur.execute(_selection_sql(), (max(1, int(limit or 1000)),))
                domain_ids = [row[0] for row in cur.fetchall()]
                if not domain_ids:
                    conn.rollback()
                    return None

    with get_db() as conn:
        conn.autocommit = False
        with conn.cursor() as cur:
            domain_snapshots: list[dict] = []
            for domain_id in domain_ids:
                snapshot = _load_domain_pipeline_snapshot(cur, int(domain_id))
                if snapshot:
                    domain_snapshots.append(snapshot)
            if not domain_snapshots:
                conn.rollback()
                return None

            job_prefix = "step" if job_definition["is_single_step"] else "detect"
            job_code = f"{job_prefix}-{datetime.now().strftime('%Y%m%d%H%M%S')}-{uuid4().hex[:6]}"
            plan_hash = uuid4().hex
            cur.execute(
                """
                INSERT INTO detect_jobs (job_code, source, plan_hash, task_mode, step_code, status, remark, created_by)
                VALUES (%s, %s, %s, %s, %s, 'pending', %s, %s)
                RETURNING id
                """,
                (
                    job_code,
                    job_definition["source"],
                    plan_hash,
                    job_definition["task_mode"],
                    job_definition["step_code"],
                    f"API 创建{job_definition['label']}任务，待检测域名 {len(domain_ids)} 个",
                    created_by,
                ),
            )
            job_id = cur.fetchone()[0]
            queued_count = 0
            skipped_count = 0
            for snapshot in domain_snapshots:
                item_step_code = (
                    job_definition["step_code"]
                    if job_definition["is_single_step"]
                    else resolve_initial_domain_pipeline_item(snapshot, settings_payload=settings_payload)[0]
                )
                if not item_step_code:
                    skipped_count += 1
                    continue
                _, step_payload = resolve_initial_domain_pipeline_item(snapshot, settings_payload=settings_payload)
                cur.execute(
                    """
                    INSERT INTO detect_job_items (job_id, domain_id, step_code, status, step_payload_json)
                    VALUES (%s, %s, %s, 'pending', %s::jsonb)
                    ON CONFLICT (job_id, domain_id, step_code) DO NOTHING
                    RETURNING id
                    """,
                    (
                        job_id,
                        int(snapshot["id"]),
                        item_step_code,
                        json.dumps(step_payload, ensure_ascii=False),
                    ),
                )
                if cur.fetchone():
                    queued_count += 1
                else:
                    skipped_count += 1

            if queued_count <= 0:
                conn.rollback()
                return None
            cur.execute(
                """
                INSERT INTO detect_run_events (job_id, node_code, event_type, level, message, payload_json)
                VALUES (%s, %s, %s, %s, %s, %s::jsonb)
                """,
                (
                    job_id,
                    settings.node_code,
                    "job_created",
                    "info",
                    f"创建检测任务 {job_code}，已入队 {queued_count} 个任务项",
                    json.dumps(
                        {
                            "count": len(domain_ids),
                            "queued_count": queued_count,
                            "skipped_count": skipped_count,
                            "task_mode": job_definition["task_mode"],
                            "step_code": job_definition["step_code"],
                        },
                        ensure_ascii=False,
                    ),
                ),
            )
        conn.commit()
    return get_detect_job_summary(job_id)