debug

domain-api/app/api/routes/detect.py

@@ -19,6 +19,39 @@ from app.services.worker_control_service import send_worker_command, start_worke
 router = APIRouter(tags=["detect"])
 
 
+def _build_detect_action_result(
+    *,
+    action: str,
+    ok: bool,
+    message: str,
+    poll_after_seconds: int = 2,
+    refresh_status: bool = True,
+    data: dict | None = None,
+) -> dict:
+    result = {
+        "action": action,
+        "poll_after_seconds": poll_after_seconds,
+        "refresh_status": refresh_status,
+        **(data or {}),
+    }
+    lowered = str(message or "").strip().lower()
+    if "ui_level" not in result:
+        if not ok:
+            result["ui_level"] = "warning" if any(keyword in lowered for keyword in ("当前没有", "无需", "未运行", "未启动")) else "error"
+        elif any(keyword in str(message or "") for keyword in ("命令已发送", "已发送检测启动请求", "控制指令")):
+            result["ui_level"] = "warning"
+        else:
+            result["ui_level"] = "success"
+    if "poll_schedule_seconds" not in result:
+        if ok:
+            result["poll_schedule_seconds"] = [1, max(2, poll_after_seconds)]
+        elif poll_after_seconds > 0:
+            result["poll_schedule_seconds"] = [poll_after_seconds]
+        else:
+            result["poll_schedule_seconds"] = []
+    return result
+
+
 def _build_settings_summary(settings_payload: dict) -> dict:
     thread_count_resolution = resolve_thread_count(settings_payload=settings_payload)
     return {
@@ -65,15 +98,16 @@ def detect_job_detail(job_id: int) -> ApiResponse:
 def start_detect() -> ApiResponse:
     job_summary = create_detect_job_if_needed(limit=1000, created_by="api")
     if not job_summary:
+        result = _build_detect_action_result(
+            action="start",
+            ok=False,
+            message="当前没有可创建的检测任务",
+            data={"job": None},
+        )
         return ApiResponse(
             code=0,
             message="当前没有可创建的检测任务",
-            data={
-                "action": "start",
-                "job": None,
-                "poll_after_seconds": 2,
-                "refresh_status": True,
-            },
+            data=result,
         )
     cycle_token = uuid4().hex[:10]
     append_detect_job_event(
@@ -91,6 +125,12 @@ def start_detect() -> ApiResponse:
 
     ok, message = start_worker()
     if not ok:
+        result = _build_detect_action_result(
+            action="start",
+            ok=False,
+            message=message,
+            data={"job": job_summary},
+        )
         append_detect_job_event(
             job_summary["job_id"],
             event_type="job_dispatch_failed",
@@ -101,12 +141,7 @@ def start_detect() -> ApiResponse:
         return ApiResponse(
             code=1,
             message=message,
-            data={
-                "action": "start",
-                "job": job_summary,
-                "poll_after_seconds": 2,
-                "refresh_status": True,
-            },
+            data=result,
         )
 
     command_ok, command_message = send_worker_command(
@@ -140,16 +175,14 @@ def start_detect() -> ApiResponse:
             progress=snapshot.get("progress", {}),
             settings_summary=settings_summary,
         )
-    return ApiResponse(
-        code=0 if command_ok else 1,
-        message=f"{message}；{command_message}" if command_ok else command_message,
-        data={
-            "action": "start",
-            "job": job_summary,
-            "poll_after_seconds": 2,
-            "refresh_status": True,
-        },
+    response_message = f"{message}；{command_message}" if command_ok else command_message
+    result = _build_detect_action_result(
+        action="start",
+        ok=command_ok,
+        message=response_message,
+        data={"job": job_summary},
     )
+    return ApiResponse(code=0 if command_ok else 1, message=response_message, data=result)
 
 
 @router.post("/detect/stop", response_model=ApiResponse)
@@ -193,12 +226,5 @@ def stop_detect() -> ApiResponse:
             settings_summary=settings_summary,
             active_job=active_job,
         )
-    return ApiResponse(
-        code=0 if ok else 1,
-        message=message,
-        data={
-            "action": "stop",
-            "poll_after_seconds": 2,
-            "refresh_status": True,
-        },
-    )
+    result = _build_detect_action_result(action="stop", ok=ok, message=message)
+    return ApiResponse(code=0 if ok else 1, message=message, data=result)

domain-api/app/core/files.py

@@ -11,6 +11,31 @@ def domain_root() -> Path:
     return Path(settings.domain_root)
 
 
+def resolve_domain_path(*relative_paths: str) -> Path | None:
+    root = domain_root()
+    candidates: list[Path] = []
+    seen: set[str] = set()
+    for relative_path in relative_paths:
+        text = str(relative_path or "").strip()
+        if not text:
+            continue
+        candidate = root / text
+        candidate_key = str(candidate)
+        if candidate_key not in seen:
+            seen.add(candidate_key)
+            candidates.append(candidate)
+        if "/" not in text and "\\" not in text:
+            nested_candidate = root / "logs" / text
+            nested_key = str(nested_candidate)
+            if nested_key not in seen:
+                seen.add(nested_key)
+                candidates.append(nested_candidate)
+    for candidate in candidates:
+        if candidate.exists():
+            return candidate
+    return candidates[0] if candidates else None
+
+
 def read_json(relative_path: str, default: dict | list | None = None):
     path = domain_root() / relative_path
     if not path.exists():
@@ -30,8 +55,8 @@ def write_json(relative_path: str, payload) -> None:
 
 
 def tail_lines(relative_path: str, max_lines: int = 120) -> list[str]:
-    path = domain_root() / relative_path
-    if not path.exists():
+    path = resolve_domain_path(relative_path)
+    if path is None or not path.exists():
         return []
     with path.open("r", encoding="utf-8", errors="replace") as handle:
         return handle.read().splitlines()[-max_lines:]

domain-api/app/services/cluster_runtime_service.py

@@ -86,6 +86,11 @@ CREATE TABLE IF NOT EXISTS detect_sync_records (
 );
 """
 
+_STALE_AFTER_SECONDS = 90
+_OFFLINE_AFTER_MINUTES = 5
+_PRUNE_IMPORTED_AFTER_MINUTES = 30
+_PRUNE_GENERAL_AFTER_HOURS = 6
+
 
 def _resolve_local_ip() -> str:
     try:
@@ -190,6 +195,28 @@ def cleanup_imported_runtime_nodes(*, region: str, role: str, keep_node_code: st
         conn.commit()
 
 
+def prune_expired_runtime_nodes() -> None:
+    imported_cutoff = datetime.now() - timedelta(minutes=_PRUNE_IMPORTED_AFTER_MINUTES)
+    general_cutoff = datetime.now() - timedelta(hours=_PRUNE_GENERAL_AFTER_HOURS)
+    with get_db() as conn:
+        with conn.cursor() as cur:
+            cur.execute(
+                """
+                DELETE FROM detect_worker_nodes
+                WHERE (
+                    (metadata_json->>'service') = 'runtime-ingest'
+                    AND last_heartbeat_at < %s
+                )
+                OR (
+                    COALESCE(metadata_json->>'service', '') <> 'runtime-ingest'
+                    AND last_heartbeat_at < %s
+                )
+                """,
+                (imported_cutoff, general_cutoff),
+            )
+        conn.commit()
+
+
 def register_local_control_heartbeat() -> None:
     from app.services.detect_job_service import get_active_detect_job_summary
     from app.services.worker_control_service import detect_worker_runtime
@@ -240,14 +267,15 @@ def _normalize_node_status(raw_status: str, last_heartbeat_at: datetime | None)
         return status
     now = datetime.now(last_heartbeat_at.tzinfo) if last_heartbeat_at.tzinfo else datetime.now()
     age = now - last_heartbeat_at
-    if age > timedelta(minutes=5):
+    if age > timedelta(minutes=_OFFLINE_AFTER_MINUTES):
         return "offline"
-    if age > timedelta(seconds=90):
+    if age > timedelta(seconds=_STALE_AFTER_SECONDS):
         return "stale"
     return status
 
 
 def get_cluster_snapshot() -> dict:
+    prune_expired_runtime_nodes()
     register_local_control_heartbeat()
     with get_db() as conn:
         with conn.cursor() as cur:

domain-api/app/services/dashboard.py

@@ -18,10 +18,21 @@ def fetch_overview() -> dict:
     with get_db() as conn:
         with conn.cursor() as cur:
             for key, query in queries.items():
-                cur.execute(query)
-                result[key] = cur.fetchone()[0]
+                try:
+                    cur.execute(query)
+                    result[key] = cur.fetchone()[0]
+                except Exception:
+                    result[key] = 0
     runtime = get_runtime_status()
+    cluster_summary = ((runtime.get("cluster") or {}).get("summary") or {})
+    online_worker_nodes = int(cluster_summary.get("online_worker_nodes", 0) or 0)
+    dedicated_online_worker_nodes = int(cluster_summary.get("dedicated_online_worker_nodes", 0) or 0)
     result["worker_status"] = "online" if runtime["worker"]["running"] else "offline"
+    result["local_worker_status"] = result["worker_status"]
+    result["cluster_worker_status"] = "online" if online_worker_nodes > 0 else "offline"
+    result["cluster_online_worker_nodes"] = online_worker_nodes
+    result["cluster_dedicated_online_worker_nodes"] = dedicated_online_worker_nodes
+    result["cluster_online_control_nodes"] = int(cluster_summary.get("online_control_nodes", 0) or 0)
     result["api_status"] = "online"
     result["worker_mode"] = runtime["worker"]["mode"]
     return result

domain-api/app/services/detect_service.py

@@ -3,10 +3,8 @@ from __future__ import annotations
 import json
 import re
 from datetime import datetime, timezone
-from pathlib import Path
-
 from app.core.db import get_db
-from app.core.files import tail_lines
+from app.core.files import resolve_domain_path, tail_lines
 from app.core.redis_client import get_redis
 from app.services.runtime_settings_service import get_runtime_settings
 from app.services.detect_run_service import sync_detect_runs
@@ -20,6 +18,7 @@ _PROXY_COUNT_RE = re.compile(r"当前可用代理数[:：]\s*(\d+)")
 _THREAD_COUNT_RE = re.compile(r"当前实际线程数量[:：]\s*(\d+)\s*/\s*(\d+)")
 _RUNTIME_STATE_KEY = "domain_tool:detect_runtime_state"
 _TIMESTAMP_FORMATS = ("%Y-%m-%d %H:%M:%S.%f", "%Y-%m-%d %H:%M:%S")
+_REMOTE_LOG_MAX_CHARS = 500
 
 
 def _extract_dependency_alerts(lines: list[str]) -> list[dict]:
@@ -141,6 +140,61 @@ def _recent_event(lines: list[str]) -> str:
     return ""
 
 
+def _build_remote_log_lines(
+    active_job: dict | None,
+    *,
+    enabled: bool,
+    mode: str,
+    limit: int = 240,
+) -> list[str]:
+    if not enabled:
+        return []
+    if not active_job:
+        return []
+    events = list(active_job.get("current_cycle_events") or active_job.get("recent_events") or [])
+    if not events:
+        return []
+
+    lines: list[str] = []
+    normalized_mode = str(mode or "key").strip().lower()
+    if normalized_mode not in {"key", "full"}:
+        normalized_mode = "key"
+    current_cycle_token = str(active_job.get("current_cycle_token") or "").strip()
+    for event in reversed(events):
+        event_type = str(event.get("event_type") or "").strip()
+        if event_type != "worker_log":
+            continue
+        created_at = str(event.get("created_at") or "").strip()
+        node_code = str(event.get("node_code") or "").strip() or "unknown"
+        message = str(event.get("message") or "").strip()
+        if not message:
+            continue
+        payload = event.get("payload") if isinstance(event.get("payload"), dict) else {}
+        event_cycle_token = str(payload.get("cycle_token") or "").strip()
+        if current_cycle_token and event_cycle_token and event_cycle_token != current_cycle_token:
+            continue
+        event_mode = str(payload.get("log_mode") or "key").strip().lower()
+        if event_mode not in {"key", "full"}:
+            event_mode = "key"
+        if normalized_mode != "full" and event_mode == "full":
+            continue
+        if len(message) > _REMOTE_LOG_MAX_CHARS:
+            message = f"{message[:_REMOTE_LOG_MAX_CHARS]}..."
+        lines.append(f"[{created_at}] [{node_code}] {message}")
+    return lines[-max(1, int(limit or 240)) :]
+
+
+def _resolve_remote_log_lines(
+    active_job: dict | None,
+    runs: list[dict],
+    *,
+    enabled: bool,
+    mode: str,
+    limit: int = 240,
+) -> list[str]:
+    return _build_remote_log_lines(active_job, enabled=enabled, mode=mode, limit=limit)
+
+
 def _load_runtime_state() -> dict:
     try:
         redis_client = get_redis()
@@ -282,17 +336,18 @@ def get_detect_status() -> dict:
     with get_db() as conn:
         with conn.cursor() as cur:
             for key, query in queries.items():
-                cur.execute(query)
-                progress[key] = cur.fetchone()[0]
+                try:
+                    cur.execute(query)
+                    progress[key] = cur.fetchone()[0]
+                except Exception:
+                    progress[key] = 0
 
     settings_payload = get_settings_payload()
-    worker_log = Path(__file__).resolve().parents[3] / "domainCheck" / "detect_worker.log"
-    if not worker_log.exists():
-        worker_log = Path(__file__).resolve().parents[3] / "domainCheck" / "logs" / "detect_worker.log"
+    worker_log = resolve_domain_path("detect_worker.log", "logs/detect_worker.log")
 
     worker_online = False
     last_log_time = None
-    if worker_log.exists():
+    if worker_log and worker_log.exists():
         modified = datetime.fromtimestamp(worker_log.stat().st_mtime, tz=timezone.utc)
         last_log_time = modified.isoformat()
         worker_online = (datetime.now(timezone.utc) - modified).total_seconds() < 180
@@ -335,7 +390,7 @@ def get_detect_status() -> dict:
         "allow_direct": settings_payload["proxy_config"].get("allow_direct", False),
         "proxy_pool_count": len(settings_payload["proxy_config"].get("proxy_urls", [])),
     }
-    active_job = get_active_detect_job_summary()
+    active_job = get_active_detect_job_summary(event_limit=240)
     runtime_snapshot = {
         **runtime,
         "detecting": runtime_state.get("detecting", False),
@@ -347,6 +402,15 @@ def get_detect_status() -> dict:
         "proxy_supplier_empty": proxy_runtime["supplier_empty"],
     }
     runs = sync_detect_runs(runtime_snapshot, progress, settings_summary, active_job=active_job)
+    worker_log_sync_enabled = bool(runtime_settings.get("worker_log_sync_enabled", False))
+    worker_log_sync_mode = str(runtime_settings.get("worker_log_sync_mode", "key") or "key")
+    remote_log_lines = _resolve_remote_log_lines(
+        active_job,
+        runs,
+        enabled=worker_log_sync_enabled,
+        mode=worker_log_sync_mode,
+        limit=240,
+    )
     dependency_alerts = _extract_dependency_alerts(recent_lines)
     append_detect_result_projection_if_changed(
         detect={
@@ -400,6 +464,9 @@ def get_detect_status() -> dict:
         "recent_event": runtime_state.get("detail") or _recent_event(recent_lines),
         "recent_warning": recent_proxy_warning,
         "log_lines": recent_lines,
+        "remote_log_lines": remote_log_lines,
         "runs": runs,
         "active_job": active_job,
+        "worker_log_sync_enabled": worker_log_sync_enabled,
+        "worker_log_sync_mode": worker_log_sync_mode,
     }

domain-api/app/services/runtime_control_service.py

@@ -8,7 +8,7 @@ from app.core.config import settings
 from app.services.debug_event_service import push_debug_event
 from app.services.sync_push_service import pull_detect_task_batch_now, push_runtime_projection_now
 from app.services.runtime_settings_service import get_runtime_settings
-from app.services.worker_control_service import _run_systemctl, start_worker, stop_worker
+from app.services.worker_control_service import _run_systemctl, normalize_systemctl_error, start_worker, stop_worker
 
 
 def _workspace_root() -> Path:
@@ -29,7 +29,7 @@ def restart_api() -> tuple[bool, str]:
         # blocking the HTTP request until uvicorn is torn down.
         result = _run_systemctl(["restart", api_service_name], timeout=5, no_block=True)
         if result.returncode != 0:
-            return False, (result.stderr or result.stdout or "重启 Linux API 失败").strip()
+            return False, normalize_systemctl_error(result.stderr or result.stdout or "重启 Linux API 失败", service_name=api_service_name)
         return True, f"Linux API 重启命令已发送: {api_service_name}"
 
     if os.name != "nt":
@@ -59,7 +59,7 @@ def _run_systemd_action(service_name: str, action: str, *, no_block: bool = Fals
     systemctl_command.extend([action, service_name])
     result = _run_systemctl(systemctl_command, timeout=10 if no_block else 30)
     if result.returncode != 0:
-        return False, (result.stderr or result.stdout or f"{action} {service_name} 失败").strip()
+        return False, normalize_systemctl_error(result.stderr or result.stdout or f"{action} {service_name} 失败", service_name=service_name)
     return True, f"{service_name} {action} 命令已发送"
 
 
@@ -80,6 +80,42 @@ def _emit_runtime_action_event(action: str, *, stage: str, ok: bool, message: st
         pass
 
 
+def _build_runtime_action_result(
+    *,
+    action: str,
+    poll_after_seconds: int,
+    refresh_runtime: bool,
+    ok: bool,
+    message: str,
+    data: dict | None = None,
+) -> dict:
+    result = {
+        "action": action,
+        "poll_after_seconds": poll_after_seconds,
+        "refresh_runtime": refresh_runtime,
+        **(data or {}),
+    }
+    lowered = str(message or "").strip().lower()
+    if "ui_level" not in result:
+        if not ok:
+            result["ui_level"] = "warning" if any(keyword in lowered for keyword in ("当前没有", "无需", "未启用", "未配置")) else "error"
+        elif any(keyword in str(message or "") for keyword in ("命令已发送", "确认失败")):
+            result["ui_level"] = "warning"
+        else:
+            result["ui_level"] = "success"
+
+    if "poll_schedule_seconds" not in result:
+        if action in {"start_worker", "stop_worker", "restart_api", "start_sync_agent", "stop_sync_agent"}:
+            result["poll_schedule_seconds"] = [1, max(2, poll_after_seconds)]
+        elif action in {"push_sync", "pull_tasks"}:
+            result["poll_schedule_seconds"] = [1, max(2, poll_after_seconds)]
+        elif poll_after_seconds > 0:
+            result["poll_schedule_seconds"] = [poll_after_seconds]
+        else:
+            result["poll_schedule_seconds"] = []
+    return result
+
+
 def start_sync_agent() -> tuple[bool, str]:
     runtime = get_runtime_settings()
     worker_mode = runtime.get("worker_mode", settings.worker_mode)
@@ -109,47 +145,37 @@ def runtime_action(action: str) -> tuple[bool, str, dict]:
         )
     if normalized_action == "start_worker":
         ok, message = start_worker()
-        result = {"action": normalized_action, "poll_after_seconds": 2, "refresh_runtime": True}
+        result = _build_runtime_action_result(action=normalized_action, poll_after_seconds=2, refresh_runtime=True, ok=ok, message=message)
         _emit_runtime_action_event(normalized_action, stage="finished", ok=ok, message=message, data=result)
         return ok, message, result
     if normalized_action == "stop_worker":
         ok, message = stop_worker()
-        result = {"action": normalized_action, "poll_after_seconds": 2, "refresh_runtime": True}
+        result = _build_runtime_action_result(action=normalized_action, poll_after_seconds=2, refresh_runtime=True, ok=ok, message=message)
         _emit_runtime_action_event(normalized_action, stage="finished", ok=ok, message=message, data=result)
         return ok, message, result
     if normalized_action == "restart_api":
         ok, message = restart_api()
-        result = {"action": normalized_action, "poll_after_seconds": 4, "refresh_runtime": True}
+        result = _build_runtime_action_result(action=normalized_action, poll_after_seconds=4, refresh_runtime=True, ok=ok, message=message)
         _emit_runtime_action_event(normalized_action, stage="finished", ok=ok, message=message, data=result)
         return ok, message, result
     if normalized_action == "start_sync_agent":
         ok, message = start_sync_agent()
-        result = {"action": normalized_action, "poll_after_seconds": 2, "refresh_runtime": True}
+        result = _build_runtime_action_result(action=normalized_action, poll_after_seconds=2, refresh_runtime=True, ok=ok, message=message)
         _emit_runtime_action_event(normalized_action, stage="finished", ok=ok, message=message, data=result)
         return ok, message, result
     if normalized_action == "stop_sync_agent":
         ok, message = stop_sync_agent()
-        result = {"action": normalized_action, "poll_after_seconds": 2, "refresh_runtime": True}
+        result = _build_runtime_action_result(action=normalized_action, poll_after_seconds=2, refresh_runtime=True, ok=ok, message=message)
         _emit_runtime_action_event(normalized_action, stage="finished", ok=ok, message=message, data=result)
         return ok, message, result
     if normalized_action == "push_sync":
         ok, message, data = push_runtime_projection_now()
-        result = {
-            "action": normalized_action,
-            "poll_after_seconds": 2,
-            "refresh_runtime": True,
-            **(data or {}),
-        }
+        result = _build_runtime_action_result(action=normalized_action, poll_after_seconds=2, refresh_runtime=True, ok=ok, message=message, data=data)
         _emit_runtime_action_event(normalized_action, stage="finished", ok=ok, message=message, data=result)
         return ok, message, result
     if normalized_action == "pull_tasks":
         ok, message, data = pull_detect_task_batch_now()
-        result = {
-            "action": normalized_action,
-            "poll_after_seconds": 2,
-            "refresh_runtime": True,
-            **(data or {}),
-        }
+        result = _build_runtime_action_result(action=normalized_action, poll_after_seconds=2, refresh_runtime=True, ok=ok, message=message, data=data)
         _emit_runtime_action_event(normalized_action, stage="finished", ok=ok, message=message, data=result)
         return ok, message, result
     if normalized_action == "push_debug_probe":
@@ -160,18 +186,9 @@ def runtime_action(action: str) -> tuple[bool, str, dict]:
             message="manual debug probe",
             payload={"node_code": settings.node_code, "node_region": settings.node_region},
         )
-        result = {
-            "action": normalized_action,
-            "poll_after_seconds": 1,
-            "refresh_runtime": False,
-            **(data or {}),
-        }
+        result = _build_runtime_action_result(action=normalized_action, poll_after_seconds=1, refresh_runtime=False, ok=ok, message=message, data=data)
         _emit_runtime_action_event(normalized_action, stage="finished", ok=ok, message=message, data=result)
         return ok, message, result
-    result = {
-        "action": normalized_action,
-        "poll_after_seconds": 0,
-        "refresh_runtime": False,
-    }
+    result = _build_runtime_action_result(action=normalized_action, poll_after_seconds=0, refresh_runtime=False, ok=False, message=f"不支持的运行时动作: {action}")
     _emit_runtime_action_event(normalized_action, stage="finished", ok=False, message=f"不支持的运行时动作: {action}", data=result)
     return False, f"不支持的运行时动作: {action}", result

domain-api/app/services/runtime_settings_service.py

@@ -9,21 +9,41 @@ DEFAULT_RUNTIME_SETTINGS = {
     "worker_service_name": settings.worker_service_name,
     "api_service_name": settings.api_service_name,
     "sync_agent_service_name": settings.sync_agent_service_name,
+    "worker_log_sync_enabled": False,
+    "worker_log_sync_mode": "key",
 }
 
 
+def _normalize_worker_log_sync_mode(value: object) -> str:
+    return "full" if str(value or "").strip().lower() == "full" else "key"
+
+
+def normalize_runtime_settings(payload: dict | None) -> dict:
+    merged = dict(DEFAULT_RUNTIME_SETTINGS)
+    if isinstance(payload, dict):
+        for key in DEFAULT_RUNTIME_SETTINGS:
+            if key in payload and payload[key] is not None:
+                merged[key] = payload[key]
+
+    worker_mode = str(merged.get("worker_mode") or "").strip()
+    if worker_mode not in {"windows-local", "linux-systemd"}:
+        merged["worker_mode"] = DEFAULT_RUNTIME_SETTINGS["worker_mode"]
+
+    for key in ("worker_service_name", "api_service_name", "sync_agent_service_name"):
+        value = str(merged.get(key) or "").strip()
+        merged[key] = value or DEFAULT_RUNTIME_SETTINGS[key]
+
+    merged["worker_log_sync_enabled"] = bool(merged.get("worker_log_sync_enabled", False))
+    merged["worker_log_sync_mode"] = _normalize_worker_log_sync_mode(merged.get("worker_log_sync_mode"))
+    return merged
+
+
 def get_runtime_settings() -> dict:
     stored = read_runtime_json("runtime_settings.json", default={})
-    result = dict(DEFAULT_RUNTIME_SETTINGS)
-    result.update(stored or {})
-    return result
+    return normalize_runtime_settings(stored)
 
 
 def update_runtime_settings(payload: dict) -> dict:
-    current = get_runtime_settings()
-    merged = dict(current)
-    for key in DEFAULT_RUNTIME_SETTINGS:
-        if key in payload and payload[key] is not None:
-            merged[key] = payload[key]
+    merged = normalize_runtime_settings({**get_runtime_settings(), **(payload or {})})
     write_runtime_json("runtime_settings.json", merged)
     return merged

domain-api/app/services/runtime_status_service.py

@@ -72,6 +72,19 @@ def _build_multi_region_readiness(
     warning_issues: list[str] = []
     info_items: list[str] = []
 
+    online_pairs = {
+        (
+            str(node.get("region") or "").strip(),
+            str(node.get("role") or "").strip(),
+        )
+        for node in nodes
+        if str(node.get("status") or "") in {"online", "busy"}
+    }
+    critical_stale_nodes: list[str] = []
+    redundant_stale_nodes: list[str] = []
+    critical_offline_nodes: list[str] = []
+    redundant_offline_nodes: list[str] = []
+
     if online_control_nodes <= 0:
         blocking_issues.append("当前没有在线控制面节点，无法视为正式可用集群。")
 
@@ -89,12 +102,33 @@ def _build_multi_region_readiness(
     if online_worker_nodes <= 0:
         warning_issues.append("当前没有在线 Worker 节点，检测任务无法在多机状态下继续推进。")
 
-    offline_nodes = list(summary.get("offline_nodes") or [])
-    stale_nodes = list(summary.get("stale_nodes") or [])
-    if stale_nodes:
-        warning_issues.append(f"存在失活节点: {'、'.join(stale_nodes)}")
-    if offline_nodes:
-        warning_issues.append(f"存在离线节点: {'、'.join(offline_nodes)}")
+    for node in nodes:
+        node_status = str(node.get("status") or "").strip()
+        if node_status not in {"stale", "offline"}:
+            continue
+        node_code = str(node.get("node_code") or "").strip()
+        node_region = str(node.get("region") or "").strip()
+        node_role = str(node.get("role") or "").strip()
+        metadata = node.get("metadata") or {}
+        replaced_by_online_peer = (node_region, node_role) in online_pairs
+        redundant = replaced_by_online_peer and (
+            node_role == "control"
+            or not bool(node.get("is_effective_worker", False))
+            or str(metadata.get("service") or "") == "runtime-ingest"
+        )
+        if node_status == "stale":
+            (redundant_stale_nodes if redundant else critical_stale_nodes).append(node_code)
+        else:
+            (redundant_offline_nodes if redundant else critical_offline_nodes).append(node_code)
+
+    if critical_stale_nodes:
+        warning_issues.append(f"存在失活节点: {'、'.join(critical_stale_nodes)}")
+    if critical_offline_nodes:
+        warning_issues.append(f"存在离线节点: {'、'.join(critical_offline_nodes)}")
+    if redundant_stale_nodes:
+        info_items.append(f"存在历史失活节点（已被在线同类节点覆盖）: {'、'.join(redundant_stale_nodes)}")
+    if redundant_offline_nodes:
+        info_items.append(f"存在历史离线节点（已被在线同类节点覆盖）: {'、'.join(redundant_offline_nodes)}")
 
     failed_batches = int(batch_states.get("failed", 0) or 0)
     projected_batches = int(batch_states.get("projected", 0) or 0)

domain-api/app/services/settings_service.py

@@ -15,6 +15,7 @@ REDIS_KEYS = {
     "thread_count": "domain_tool:thread_count",
     "node_thread_counts": "domain_tool:node_thread_counts",
     "credentials": "domain_tool:credentials",
+    "runtime_settings": "domain_tool:runtime_settings",
 }
 
 DETECT_OPTION_KEYS = {
@@ -198,7 +199,6 @@ def update_settings_payload(payload: dict) -> dict:
     write_json("proxy_config.json", proxy_config)
     write_json("thread_count.json", {"thread_count": str(thread_count)})
     write_json("node_thread_counts.json", node_thread_counts)
-
     redis_client = get_redis()
     try:
         redis_client.set(REDIS_KEYS["detect_options"], json.dumps(detect_options, ensure_ascii=False))
@@ -209,6 +209,12 @@ def update_settings_payload(payload: dict) -> dict:
         redis_client.publish("domain_tool:thread_count:update", str(thread_count))
         redis_client.set(REDIS_KEYS["node_thread_counts"], json.dumps(node_thread_counts, ensure_ascii=False))
         redis_client.publish("domain_tool:node_thread_counts:update", json.dumps(node_thread_counts, ensure_ascii=False))
+        redis_client.set(REDIS_KEYS["runtime_settings"], json.dumps(runtime_settings, ensure_ascii=False))
+        redis_client.publish("domain_tool:config_update", "runtime_settings")
+        redis_client.publish("domain_tool:config_update", "node_thread_counts")
+        redis_client.publish("domain_tool:config_update", "detect_options")
+        redis_client.publish("domain_tool:config_update", "proxy_config")
+        redis_client.publish("domain_tool:config_update", "thread_count")
     except Exception:
         pass
 
@@ -278,6 +284,12 @@ def validate_settings_payload(payload: dict) -> None:
         for key in ("worker_service_name", "api_service_name", "sync_agent_service_name"):
             if key in runtime_settings and runtime_settings[key] is not None and not str(runtime_settings[key]).strip():
                 raise ValueError(f"{key} must not be empty")
+        if "worker_log_sync_enabled" in runtime_settings and not isinstance(runtime_settings["worker_log_sync_enabled"], bool):
+            raise ValueError("worker_log_sync_enabled must be a boolean")
+        if "worker_log_sync_mode" in runtime_settings:
+            worker_log_sync_mode = str(runtime_settings.get("worker_log_sync_mode") or "").strip().lower()
+            if worker_log_sync_mode not in {"key", "full"}:
+                raise ValueError("worker_log_sync_mode must be key or full")
 
 
 def backup_current_settings(reason: str = "manual") -> dict:

domain-api/app/services/sync_push_service.py

@@ -908,11 +908,11 @@ def _push_projection_record(source_record: dict, sync_type: str, ingest_url: str
 
 def push_runtime_projection_now() -> tuple[bool, str, dict]:
     if not settings.sync_push_enabled:
-        return False, "未启用同步推送", {"action": "push_sync"}
+        return False, "未启用同步推送", {"action": "push_sync", "sync_state": "disabled", "ui_level": "warning", "poll_schedule_seconds": []}
 
     ingest_url = _ingest_url(settings.sync_target_api_base_url)
     if not ingest_url:
-        return False, "未配置同步目标地址", {"action": "push_sync"}
+        return False, "未配置同步目标地址", {"action": "push_sync", "sync_state": "misconfigured", "ui_level": "warning", "poll_schedule_seconds": []}
 
     results = []
     ok, message, data = _push_projection_now("runtime_projection", ingest_url)
@@ -922,19 +922,52 @@ def push_runtime_projection_now() -> tuple[bool, str, dict]:
     results.append({"sync_type": "detect_result_projection", "ok": ok, "message": message, "data": data})
 
     success_count = sum(1 for item in results if item["ok"])
+    warning_count = sum(
+        1
+        for item in results
+        if not item["ok"] and any(keyword in str(item.get("message") or "") for keyword in ("当前没有可推送", "已全部同步完成", "无需重复发送", "进行中", "等待下个重试窗口"))
+    )
     if success_count == 0:
-        return False, "当前没有成功推送的同步投影", {"action": "push_sync", "results": results}
-    return True, f"同步推送完成，成功 {success_count}/{len(results)}", {"action": "push_sync", "results": results}
+        if warning_count == len(results):
+            return False, "当前没有需要立即推送的同步投影", {
+                "action": "push_sync",
+                "sync_state": "idle",
+                "ui_level": "warning",
+                "poll_schedule_seconds": [],
+                "results": results,
+            }
+        return False, "同步推送未成功，请检查明细结果", {
+            "action": "push_sync",
+            "sync_state": "failed",
+            "ui_level": "error",
+            "poll_schedule_seconds": [2],
+            "results": results,
+        }
+    if success_count < len(results):
+        return True, f"同步推送部分完成，成功 {success_count}/{len(results)}", {
+            "action": "push_sync",
+            "sync_state": "partial_success",
+            "ui_level": "warning",
+            "poll_schedule_seconds": [1, 3],
+            "results": results,
+        }
+    return True, f"同步推送完成，成功 {success_count}/{len(results)}", {
+        "action": "push_sync",
+        "sync_state": "success",
+        "ui_level": "success",
+        "poll_schedule_seconds": [1, 3],
+        "results": results,
+    }
 
 
 def pull_detect_task_batch_now(limit: int | None = None) -> tuple[bool, str, dict]:
     if settings.node_region != "mainland" or settings.node_role != "control":
-        return False, "当前节点无需拉取待检测任务批次", {"action": "pull_tasks"}
+        return False, "当前节点无需拉取待检测任务批次", {"action": "pull_tasks", "pull_state": "not_applicable", "ui_level": "warning", "poll_schedule_seconds": []}
 
     export_url = _task_export_url(settings.sync_target_api_base_url)
     ack_url = _task_ack_url(settings.sync_target_api_base_url)
     if not export_url or not ack_url:
-        return False, "未配置任务拉取目标地址", {"action": "pull_tasks"}
+        return False, "未配置任务拉取目标地址", {"action": "pull_tasks", "pull_state": "misconfigured", "ui_level": "warning", "poll_schedule_seconds": []}
 
     safe_limit = max(1, min(int(limit or settings.sync_batch_size or 200), max(1, int(settings.sync_batch_size or 200))))
     request_url = f"{export_url}?limit={safe_limit}"
@@ -948,27 +981,30 @@ def pull_detect_task_batch_now(limit: int | None = None) -> tuple[bool, str, dic
             raw = response.read().decode("utf-8")
             data = json.loads(raw) if raw else {}
     except json.JSONDecodeError as exc:
-        return False, f"拉取待检测任务失败: 远端响应不是合法 JSON ({exc})", {"action": "pull_tasks"}
+        return False, f"拉取待检测任务失败: 远端响应不是合法 JSON ({exc})", {"action": "pull_tasks", "pull_state": "failed", "ui_level": "error", "poll_schedule_seconds": [2]}
     except urllib.error.HTTPError as exc:
         error_body = exc.read().decode("utf-8", errors="ignore") if hasattr(exc, "read") else ""
         return False, f"拉取待检测任务失败: HTTP {getattr(exc, 'code', 500)}", {
             "action": "pull_tasks",
+            "pull_state": "failed",
+            "ui_level": "error",
+            "poll_schedule_seconds": [2],
             "http_status": getattr(exc, "code", 500),
             "response_text": error_body[:500],
         }
     except Exception as exc:
-        return False, f"拉取待检测任务失败: {exc}", {"action": "pull_tasks"}
+        return False, f"拉取待检测任务失败: {exc}", {"action": "pull_tasks", "pull_state": "failed", "ui_level": "error", "poll_schedule_seconds": [2]}
 
     payload = data.get("data") or {}
     response_code = data.get("code", 0)
     if response_code not in (0, "0", None) and not payload:
-        return False, str(data.get("message") or "拉取待检测任务失败"), {"action": "pull_tasks", "response": data}
+        return False, str(data.get("message") or "拉取待检测任务失败"), {"action": "pull_tasks", "pull_state": "failed", "ui_level": "error", "poll_schedule_seconds": [2], "response": data}
 
     source_record_id = int(payload.get("source_record_id") or 0)
     projection_hash = str(payload.get("projection_hash") or "").strip()
     projection = payload.get("projection") or {}
     if source_record_id <= 0 or not projection_hash or not projection:
-        return False, "远端当前没有可拉取的待检测任务批次", {"action": "pull_tasks", "batch_size": 0}
+        return False, "远端当前没有可拉取的待检测任务批次", {"action": "pull_tasks", "pull_state": "idle", "ui_level": "warning", "poll_schedule_seconds": [], "batch_size": 0}
 
     ingest_ok, ingest_message, ingest_data = ingest_detect_task_projection(
         {
@@ -981,7 +1017,7 @@ def pull_detect_task_batch_now(limit: int | None = None) -> tuple[bool, str, dic
         shared_token=settings.sync_shared_token,
     )
     if not ingest_ok:
-        return False, ingest_message, {"action": "pull_tasks", **(ingest_data or {})}
+        return False, ingest_message, {"action": "pull_tasks", "pull_state": "failed", "ui_level": "error", "poll_schedule_seconds": [2], **(ingest_data or {})}
 
     ack_request = urllib.request.Request(
         ack_url,
@@ -1008,6 +1044,9 @@ def pull_detect_task_batch_now(limit: int | None = None) -> tuple[bool, str, dic
             if ack_code not in (0, "0", None, ""):
                 return True, f"{ingest_message}；但远端确认失败: {str(ack_response.get('message') or 'ack business error')}", {
                     "action": "pull_tasks",
+                    "pull_state": "ack_warning",
+                    "ui_level": "warning",
+                    "poll_schedule_seconds": [1, 3],
                     "source_record_id": source_record_id,
                     "projection_hash": projection_hash,
                     **(ingest_data or {}),
@@ -1017,18 +1056,27 @@ def pull_detect_task_batch_now(limit: int | None = None) -> tuple[bool, str, dic
     except json.JSONDecodeError as exc:
         return True, f"{ingest_message}；但远端确认失败: ack 响应不是合法 JSON ({exc})", {
             "action": "pull_tasks",
+            "pull_state": "ack_warning",
+            "ui_level": "warning",
+            "poll_schedule_seconds": [1, 3],
             "source_record_id": source_record_id,
             **(ingest_data or {}),
         }
     except Exception as exc:
         return True, f"{ingest_message}；但远端确认失败: {exc}", {
             "action": "pull_tasks",
+            "pull_state": "ack_warning",
+            "ui_level": "warning",
+            "poll_schedule_seconds": [1, 3],
             "source_record_id": source_record_id,
             **(ingest_data or {}),
         }
 
     return True, "待检测任务批次拉取并入库成功", {
         "action": "pull_tasks",
+        "pull_state": "success",
+        "ui_level": "success",
+        "poll_schedule_seconds": [1, 3],
         "source_record_id": source_record_id,
         "projection_hash": projection_hash,
         "batch_code": str(projection.get("batch_code") or "").strip(),

domain-api/app/services/sync_record_service.py

@@ -99,7 +99,7 @@ def _should_append_runtime_projection(previous_payload: dict, current_projection
 
     previous_cluster = previous_projection.get("cluster_summary") or {}
     current_cluster = current_projection.get("cluster_summary") or {}
-    for key in ("online_worker_nodes", "online_control_nodes", "busy_nodes", "stale_nodes", "offline_nodes"):
+    for key in ("online_worker_nodes", "dedicated_online_worker_nodes", "online_control_nodes", "busy_nodes", "stale_nodes", "offline_nodes"):
         if previous_cluster.get(key) != current_cluster.get(key):
             return True
 
@@ -480,6 +480,7 @@ def append_runtime_projection_if_changed(
         "cluster_summary": {
             "nodes_total": int(cluster.get("nodes_total", 0) or 0),
             "online_worker_nodes": int((cluster.get("summary") or {}).get("online_worker_nodes", 0) or 0),
+            "dedicated_online_worker_nodes": int((cluster.get("summary") or {}).get("dedicated_online_worker_nodes", 0) or 0),
             "online_control_nodes": int((cluster.get("summary") or {}).get("online_control_nodes", 0) or 0),
             "busy_nodes": list((cluster.get("summary") or {}).get("busy_nodes") or []),
             "stale_nodes": list((cluster.get("summary") or {}).get("stale_nodes") or []),

domain-api/app/services/worker_control_service.py

@@ -51,6 +51,28 @@ def _run_systemctl(
     return _run_shell(["sudo", "-n", *systemctl_command], timeout=timeout)
 
 
+def normalize_systemctl_error(raw_message: str, *, service_name: str = "") -> str:
+    message = str(raw_message or "").strip()
+    normalized_service_name = str(service_name or "").strip()
+    if not message:
+        target = normalized_service_name or "目标服务"
+        return f"{target} 控制失败，未返回可用错误信息"
+
+    lowered = message.lower()
+    if "sudo: a password is required" in lowered or "authentication is required" in lowered:
+        target = normalized_service_name or "systemd 服务"
+        return f"{target} 控制失败：当前运行用户没有免密 systemctl 权限，请为 API 进程授予对应 sudo/systemd 权限"
+    if "unit " in lowered and " could not be found" in lowered:
+        target = normalized_service_name or "目标服务"
+        return f"{target} 控制失败：systemd 中未找到该服务，请检查服务名配置是否正确"
+    if "access denied" in lowered or "permission denied" in lowered:
+        target = normalized_service_name or "systemd 服务"
+        return f"{target} 控制失败：权限不足，请检查当前 API 进程用户是否具备 systemctl 控制权限"
+    if "host is down" in lowered or "failed to connect to bus" in lowered:
+        return "systemctl 调用失败：当前环境未接入可用的 systemd/dbus，请确认部署方式与 worker_mode 是否匹配"
+    return message
+
+
 def _parse_systemd_timestamp(raw_timestamp: str) -> str:
     raw_timestamp = (raw_timestamp or "").strip()
     if not raw_timestamp:
@@ -206,7 +228,7 @@ def start_worker() -> tuple[bool, str]:
     if worker_mode == "linux-systemd":
         result = _run_systemctl(["start", service_name], timeout=30)
         if result.returncode != 0:
-            return False, (result.stderr or result.stdout or "启动 Linux Worker 失败").strip()
+            return False, normalize_systemctl_error(result.stderr or result.stdout or "启动 Linux Worker 失败", service_name=service_name)
         return True, f"Linux Worker 启动命令已发送: {service_name}"
 
     if os.name != "nt":
@@ -234,7 +256,7 @@ def stop_worker() -> tuple[bool, str]:
     if worker_mode == "linux-systemd":
         result = _run_systemctl(["stop", service_name], timeout=30)
         if result.returncode != 0:
-            return False, (result.stderr or result.stdout or "停止 Linux Worker 失败").strip()
+            return False, normalize_systemctl_error(result.stderr or result.stdout or "停止 Linux Worker 失败", service_name=service_name)
         return True, f"Linux Worker 停止命令已发送: {service_name}"
 
     if os.name != "nt":