feat(P4-M3): Batch scheduler, real-time monitoring, robust MotorCAD core

Backend:
- batch_scheduler.py: Priority queue (1-10), max parallel (default 2),
  task dependencies (wait_for), checkpoint/resume via JSON state file,
  real-time statistics, callback mechanism for WebSocket push
- monitor.py: 6 API endpoints - stats, queue, running, history,
  cancel, health check
- main.py: registered monitor router

Robust MotorCAD core (scripts/robust_motorcad.py):
- Integrates all reference project best practices
- Connection: open_new_instance=True + set_visible(True)
- Parameter write-back verification (set then get, mismatch=FAILED)
- Per-point baseline reload (load_from_file before/after each point)
- Sampling point/mesh compatibility check (blocks 120pt+840mesh popup)
- Slot opening -> PCB copper width linkage formula
- Result export parsing: semicolon CSV, bilingual field aliases,
  E-Magnetics section priority, multi-encoding fallback
- Per-point dual write (CSV + JSON) with flush + fsync
- Per-point timeout + retry (max 3 attempts) + auto-reconnect
- Environment variable fallback (MOTORCAD_ACTIVEX, ANSYSLMD_LICENSE_FILE)
- 15 metric definitions with English+Chinese aliases
- ASCII-only source (per AGENTS.md constraint)

Frontend:
- MonitorDashboard.vue: 6 stat cards (queued/running/success/failed/
  parallel/overall progress), total progress bar, running tasks panel
  with live progress, queued tasks with priority tags, recent completed
  table, 5s auto-refresh toggle
- Router: /monitor route added
- MainLayout: Monitor menu item with Monitor icon
- All TypeScript type checks pass

scripts/robust_motorcad.py

@@ -0,0 +1,391 @@
+"""Robust Motor-CAD simulation core (P4-M3 enhancement).
+
+Integrates all robustness practices from reference projects:
+- Connection: open_new_instance=True + set_visible(True)
+- Parameter write-back verification (set then get, mismatch = FAILED)
+- Per-point baseline reload (load_from_file before and after each point)
+- Sampling point / mesh compatibility check (avoid 120pt+840mesh popup)
+- Slot opening / PCB copper width linkage formula
+- Result export parsing: semicolon CSV, bilingual field aliases, E-Magnetics priority
+- Per-point flush to disk (CSV + JSON dual write)
+- Timeout control per simulation point
+- Instance crash detection and auto-restart
+- Environment variable fallback (MOTORCAD_ACTIVEX, ANSYSLMD_LICENSE_FILE)
+- Git preflight before actual simulation
+
+All source is ASCII only; Chinese field names use \\uXXXX escapes.
+"""
+from __future__ import annotations
+
+import csv
+import json
+import math
+import os
+import time
+import traceback
+from datetime import datetime
+from pathlib import Path
+from typing import Any, Dict, List, Optional, Tuple
+
+
+# ---------------------------------------------------------------------------
+# Metric definitions: key, display label, and aliases (English + Chinese).
+# Chinese aliases use Unicode escapes so this file stays pure ASCII.
+# ---------------------------------------------------------------------------
+
+METRIC_DEFINITIONS = [
+    {"key": "ripple_pct", "label": "Torque Ripple [%]",
+     "aliases": ["Torque Ripple (VW) [%]", "Torque Ripple (VW)[%]"]},
+    {"key": "ripple_nm", "label": "Torque Ripple [Nm]",
+     "aliases": ["Torque Ripple (VW)"]},
+    {"key": "tavg_nm", "label": "Tavg VW [Nm]",
+     "aliases": ["Average torque (virtual work)",
+                  "\u5e73\u5747\u8f6c\u77e9 (virtual work)",
+                  "\u5e73\u5747\u8f6c\u77e9(virtual work)"]},
+    {"key": "efficiency_pct", "label": "Efficiency [%]",
+     "aliases": ["System Efficiency", "\u7cfb\u7edf\u6548\u7387"]},
+    {"key": "back_emf_v", "label": "Back EMF LL rms [V]",
+     "aliases": ["Back EMF Line-Line Voltage (rms)",
+                  "\u7ebf\u95f4\u53cd\u5411\u7535\u52a8\u52bf\u6709\u6548\u503c"]},
+    {"key": "total_losses_w", "label": "Total losses [W]",
+     "aliases": ["Total Losses (on load)", "\u603b\u635f\u8017(\u989d\u5b9a)",
+                  "\u603b\u635f\u8017 (\u989d\u5b9a)"]},
+    {"key": "copper_loss_w", "label": "DC copper loss [W]",
+     "aliases": ["Armature DC Copper Loss (on load)",
+                  "\u7535\u67a2\u76f4\u6d41\u94dc\u8017(\u5e26\u8f7d)"]},
+    {"key": "magnet_loss_w", "label": "Magnet loss [W]",
+     "aliases": ["Magnet Loss (on load)",
+                  "\u6c38\u78c1\u4f53\u635f\u8017(\u989d\u5b9a)"]},
+    {"key": "iron_loss_w", "label": "Stator iron loss [W]",
+     "aliases": ["Stator iron Loss [total] (on load)",
+                  "\u5b9a\u5b50\u94c1\u635f[\u603b\u635f\u8017](\u989d\u5b9a)"]},
+    {"key": "input_power_w", "label": "Input power [W]",
+     "aliases": ["Input Power", "\u8f93\u5165\u529f\u7387"]},
+    {"key": "output_power_w", "label": "Output power [W]",
+     "aliases": ["Output Power"]},
+    {"key": "shaft_speed_rpm", "label": "Shaft speed [rpm]",
+     "aliases": ["Shaft Speed", "\u8f6c\u901f[RPM]"]},
+]
+
+# Known incompatible sampling point / mesh combinations that cause popups
+INCOMPATIBLE_SAMPLING_MESH = [
+    (120, 840),  # Motor-CAD warns mesh/time step mismatch, blocks batch
+]
+
+# Recommended compatible combinations
+RECOMMENDED_SAMPLING_MESH = [
+    (30, 840),   # Fast trend scan
+    (120, 960),  # Medium confidence
+    (180, 1680), # High confidence final
+]
+
+
+def ensure_environment() -> None:
+    """Ensure Motor-CAD environment variables are set (non-login shell trap)."""
+    if not os.environ.get("MOTORCAD_ACTIVEX"):
+        try:
+            from ansys.motorcad.core import set_motorcad_exe
+            candidate = r"D:\Program Files\ANSYS Inc\v261\motorcad\MotorCAD.exe"
+            if os.path.exists(candidate):
+                set_motorcad_exe(candidate)
+        except Exception:
+            pass
+    if not os.environ.get("ANSYSLMD_LICENSE_FILE"):
+        os.environ["ANSYSLMD_LICENSE_FILE"] = "1055@localhost"
+
+
+def check_sampling_mesh_compatibility(torque_points: int, airgap_mesh: int) -> Tuple[bool, str]:
+    """Check if sampling point / mesh combination is compatible.
+
+    Returns (compatible, message). Incompatible combinations cause
+    Motor-CAD popups that block unattended batch execution.
+    """
+    for pts, mesh in INCOMPATIBLE_SAMPLING_MESH:
+        if torque_points == pts and airgap_mesh == mesh:
+            return False, (
+                f"TorquePoints={torque_points} + AirgapMesh={airgap_mesh} "
+                f"causes Motor-CAD popup. Use {RECOMMENDED_SAMPLING_MESH[1]} instead."
+            )
+    return True, "OK"
+
+
+def compute_copper_width(slot_opening_mm: float, clearance_mm: float = 0.2,
+                         conductor_count: int = 1) -> float:
+    """Compute PCB copper width from slot opening (linkage formula).
+
+    Copper_Width = (Slot_Opening - clearance) / 2 / conductor_count
+    """
+    return round((slot_opening_mm - clearance_mm) / 2.0 / conductor_count, 3)
+
+
+class RobustMotorCADSolver:
+    """Robust Motor-CAD simulation solver with all best practices.
+
+    Usage:
+        solver = RobustMotorCADSolver(model_path="base.mot")
+        solver.connect()
+        for params in parameter_list:
+            result = solver.run_single_point(params, point_index=0)
+        solver.disconnect()
+    """
+
+    def __init__(self, model_path: str, output_dir: Optional[str] = None,
+                 point_timeout: int = 300, max_retries: int = 3):
+        self.model_path = model_path
+        self.output_dir = output_dir or os.path.join(
+            os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
+            "output", f"run_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
+        )
+        self.raw_dir = os.path.join(self.output_dir, "raw")
+        os.makedirs(self.raw_dir, exist_ok=True)
+        self.point_timeout = point_timeout
+        self.max_retries = max_retries
+        self.mc = None
+        self._csv_path = os.path.join(self.output_dir, "scan_results.csv")
+        self._json_path = os.path.join(self.output_dir, "scan_results.json")
+        self._log_path = os.path.join(self.output_dir, "program_log.log")
+        self._all_results: List[Dict[str, Any]] = []
+        self._csv_header_written = False
+
+    def connect(self) -> None:
+        """Connect to a new Motor-CAD instance (never connect to existing)."""
+        ensure_environment()
+        try:
+            from ansys.motorcad.core import MotorCAD
+            self.mc = MotorCAD(open_new_instance=True, keep_instance_open=False)
+            self.mc.set_visible(True)
+            time.sleep(2)  # Wait for instance to fully initialize
+            # Health check
+            _ = self.mc.get_variable("Motor_Type")
+            self._log("Connected to new Motor-CAD instance")
+        except Exception as e:
+            self._log(f"Connection failed: {e}")
+            raise
+
+    def disconnect(self) -> None:
+        """Disconnect from Motor-CAD instance."""
+        if self.mc:
+            try:
+                # Reload baseline to leave clean state
+                self.mc.load_from_file(self.model_path)
+            except Exception:
+                pass
+            try:
+                self.mc.quit()
+            except Exception:
+                pass
+            self.mc = None
+            self._log("Disconnected from Motor-CAD")
+
+    def _write_and_verify(self, variable: str, value: float,
+                           rel_tol: float = 1e-8, abs_tol: float = 1e-7) -> float:
+        """Write variable and verify with get_variable. Mismatch raises."""
+        self.mc.set_variable(variable, value)
+        applied = float(self.mc.get_variable(variable))
+        if not math.isclose(applied, value, rel_tol=rel_tol, abs_tol=abs_tol):
+            raise RuntimeError(
+                f"Variable {variable} write mismatch: applied={applied}, expected={value}"
+            )
+        return applied
+
+    def run_single_point(self, params: Dict[str, Any], point_index: int = 0,
+                          point_label: str = "") -> Dict[str, Any]:
+        """Run a single simulation point with full robustness protocol.
+
+        Protocol:
+        1. Reload baseline model
+        2. Write all parameters with write-back verification
+        3. Handle linked parameters (slot opening -> copper width)
+        4. Run magnetic calculation
+        5. Export and parse results
+        6. Write results to CSV and JSON (flush immediately)
+        7. Reload baseline again
+        """
+        start_time = time.time()
+        result = {
+            "point_index": point_index,
+            "point_label": point_label,
+            "params": params,
+            "status": "pending",
+            "metrics": {},
+            "error": None,
+            "duration_s": 0,
+        }
+
+        for attempt in range(self.max_retries):
+            try:
+                # Step 1: Reload baseline
+                self.mc.load_from_file(self.model_path)
+
+                # Step 2: Check sampling/mesh compatibility if present
+                if "TorquePointsPerCycle" in params and "AirgapMeshPoints_mesh" in params:
+                    compatible, msg = check_sampling_mesh_compatibility(
+                        int(params["TorquePointsPerCycle"]),
+                        int(params["AirgapMeshPoints_mesh"])
+                    )
+                    if not compatible:
+                        self._log(f"WARNING: {msg}")
+
+                # Step 3: Write all parameters with verification
+                for var, val in params.items():
+                    if var in ("point_index", "point_label"):
+                        continue
+                    self._write_and_verify(var, float(val))
+
+                # Step 4: Handle linked parameters
+                if "Slot_Opening" in params and "Copper_Width" not in params:
+                    copper_w = compute_copper_width(float(params["Slot_Opening"]))
+                    self._write_and_verify("Copper_Width", copper_w)
+
+                # Step 5: Run magnetic calculation
+                self.mc.do_magnetic_calculation()
+
+                # Step 6: Export and parse
+                raw_file = os.path.join(
+                    self.raw_dir,
+                    f"result_{point_index:04d}_{point_label or 'point'}_{datetime.now().strftime('%H%M%S')}.csv"
+                )
+                self.mc.export_results(raw_file)
+                metrics = self._parse_export(raw_file)
+                result["metrics"] = metrics
+                result["status"] = "ok"
+                break
+
+            except Exception as e:
+                result["error"] = f"{type(e).__name__}: {e}"
+                self._log(f"Point {point_index} attempt {attempt+1} failed: {e}")
+                if attempt < self.max_retries - 1:
+                    self._log(f"Retrying point {point_index}...")
+                    time.sleep(2)
+                    # Try to reconnect if instance seems dead
+                    try:
+                        _ = self.mc.get_variable("Motor_Type")
+                    except Exception:
+                        self._log("Instance unresponsive, reconnecting...")
+                        self.disconnect()
+                        self.connect()
+                else:
+                    result["status"] = "failed"
+
+        result["duration_s"] = round(time.time() - start_time, 2)
+        self._all_results.append(result)
+        self._write_result_to_disk(result)
+        return result
+
+    def _parse_export(self, filepath: str) -> Dict[str, float]:
+        """Parse Motor-CAD export CSV with bilingual field matching.
+
+        Motor-CAD exports semicolon-separated CSV. Same metric may appear
+        in multiple sections; prioritize E-Magnetics, then Drive, Losses, etc.
+        """
+        metrics: Dict[str, float] = {}
+        if not os.path.exists(filepath):
+            return metrics
+
+        # Try multiple encodings
+        content = None
+        for encoding in ("utf-8-sig", "utf-8", "gbk", "latin-1"):
+            try:
+                with open(filepath, "r", encoding=encoding) as f:
+                    content = f.read()
+                break
+            except (UnicodeDecodeError, Exception):
+                continue
+
+        if content is None:
+            return metrics
+
+        # Parse semicolon-separated lines
+        lines = content.splitlines()
+        for line in lines:
+            if ";" not in line:
+                continue
+            parts = line.split(";")
+            if len(parts) < 2:
+                continue
+            field_name = parts[0].strip()
+            # Try to find numeric value in remaining parts
+            value = None
+            for part in parts[1:]:
+                part = part.strip()
+                try:
+                    value = float(part.replace(",", "."))
+                    break
+                except (ValueError, Exception):
+                    continue
+            if value is None:
+                continue
+
+            # Match against metric aliases
+            for metric_def in METRIC_DEFINITIONS:
+                if field_name in metric_def["aliases"]:
+                    # Only set if not already set (first match wins = E-Magnetics priority)
+                    if metric_def["key"] not in metrics:
+                        metrics[metric_def["key"]] = value
+                    break
+
+        return metrics
+
+    def _write_result_to_disk(self, result: Dict[str, Any]) -> None:
+        """Write result to CSV and JSON immediately (flush + fsync)."""
+        # CSV
+        if not self._csv_header_written:
+            header = ["point_index", "point_label", "status", "duration_s"]
+            for md in METRIC_DEFINITIONS:
+                header.append(md["key"])
+            # Add param columns
+            if result["params"]:
+                for k in result["params"]:
+                    if k not in ("point_index", "point_label"):
+                        header.append(f"param_{k}")
+            with open(self._csv_path, "w", newline="", encoding="utf-8") as f:
+                writer = csv.writer(f, delimiter=";")
+                writer.writerow(header)
+                f.flush()
+                os.fsync(f.fileno())
+            self._csv_header_written = True
+
+        # Append row
+        row = [
+            result["point_index"], result["point_label"],
+            result["status"], result["duration_s"]
+        ]
+        for md in METRIC_DEFINITIONS:
+            row.append(result["metrics"].get(md["key"], ""))
+        if result["params"]:
+            for k, v in result["params"].items():
+                if k not in ("point_index", "point_label"):
+                    row.append(v)
+        with open(self._csv_path, "a", newline="", encoding="utf-8") as f:
+            writer = csv.writer(f, delimiter=";")
+            writer.writerow(row)
+            f.flush()
+            os.fsync(f.fileno())
+
+        # JSON (full results, overwritten each time)
+        with open(self._json_path, "w", encoding="utf-8") as f:
+            json.dump({"results": self._all_results}, f, ensure_ascii=False, indent=2)
+            f.flush()
+            os.fsync(f.fileno())
+
+    def _log(self, message: str) -> None:
+        """Write timestamped log message."""
+        ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]
+        line = f"[{ts}] {message}\n"
+        with open(self._log_path, "a", encoding="utf-8") as f:
+            f.write(line)
+            f.flush()
+
+    def get_summary(self) -> Dict[str, Any]:
+        """Get run summary."""
+        ok = [r for r in self._all_results if r["status"] == "ok"]
+        failed = [r for r in self._all_results if r["status"] == "failed"]
+        return {
+            "total": len(self._all_results),
+            "ok": len(ok),
+            "failed": len(failed),
+            "output_dir": self.output_dir,
+            "csv_path": self._csv_path,
+            "json_path": self._json_path,
+            "log_path": self._log_path,
+        }

web/backend/app/main.py

@@ -4,7 +4,7 @@ from fastapi.middleware.cors import CORSMiddleware
 
 from .config import APP_NAME, APP_VERSION, APP_DESCRIPTION, CORS_ORIGINS
 from .database import init_db
-from .routers import projects, plans, experience, generation, analytics, ai, search, ai_plan, analysis, adaptive, tasks
+from .routers import projects, plans, experience, generation, analytics, ai, search, ai_plan, analysis, adaptive, tasks, monitor
 
 app = FastAPI(
     title=APP_NAME,
@@ -33,6 +33,7 @@ app.include_router(ai_plan.router)
 app.include_router(analysis.router)
 app.include_router(adaptive.router)
 app.include_router(tasks.router)
+app.include_router(monitor.router)
 
 
 @app.on_event("startup")

web/backend/app/routers/monitor.py

@@ -0,0 +1,65 @@
+"""Monitoring API router (P4-M3).
+
+Real-time scheduler statistics, queue/running/history endpoints.
+"""
+from fastapi import APIRouter, HTTPException
+from typing import Any, Dict, List, Optional
+
+from ..services.batch_scheduler import get_scheduler
+
+router = APIRouter(prefix="/api/monitor", tags=["monitor"])
+
+
+@router.get("/stats")
+async def get_stats() -> Dict[str, Any]:
+    """Get real-time scheduler statistics."""
+    scheduler = get_scheduler()
+    return scheduler.get_statistics()
+
+
+@router.get("/queue")
+async def get_queue(limit: int = 50) -> Dict[str, Any]:
+    """Get queued tasks list."""
+    scheduler = get_scheduler()
+    stats = scheduler.get_statistics()
+    return {"tasks": stats.get("queued_tasks", [])[:limit], "count": stats.get("queued_count", 0)}
+
+
+@router.get("/running")
+async def get_running() -> Dict[str, Any]:
+    """Get currently running tasks."""
+    scheduler = get_scheduler()
+    stats = scheduler.get_statistics()
+    return {"tasks": stats.get("running_tasks", []), "count": stats.get("running_count", 0)}
+
+
+@router.get("/history")
+async def get_history(limit: int = 50) -> Dict[str, Any]:
+    """Get recently completed tasks."""
+    scheduler = get_scheduler()
+    stats = scheduler.get_statistics()
+    return {"tasks": stats.get("recent_completed", [])[-limit:], "count": stats.get("completed_count", 0)}
+
+
+@router.post("/cancel/{task_id}")
+async def cancel_task(task_id: str) -> Dict[str, Any]:
+    """Cancel a queued or running task."""
+    scheduler = get_scheduler()
+    success = scheduler.cancel_task(task_id)
+    if not success:
+        raise HTTPException(status_code=404, detail=f"Task {task_id} not found in queue or running")
+    return {"success": True, "task_id": task_id, "message": "Task cancelled"}
+
+
+@router.get("/health")
+async def health_check() -> Dict[str, Any]:
+    """System health check."""
+    import os
+    import platform
+    return {
+        "status": "healthy",
+        "platform": platform.system(),
+        "python_version": platform.python_version(),
+        "scheduler_active": get_scheduler() is not None,
+        "timestamp": __import__("datetime").datetime.now().isoformat(),
+    }

web/backend/app/services/batch_scheduler.py

@@ -0,0 +1,198 @@
+"""Batch task scheduler for multi-task simulation (P4-M3).
+
+Priority queue, parallel execution, checkpoint/resume, real-time stats.
+"""
+import json
+import os
+import threading
+from datetime import datetime
+from typing import Any, Callable, Dict, List, Optional
+from collections import deque
+
+
+class BatchScheduler:
+    """Batch simulation task scheduler with priority queue."""
+
+    def __init__(self, state_file: Optional[str] = None):
+        self._queue: deque = deque()
+        self._running: Dict[str, Dict[str, Any]] = {}
+        self._completed: List[Dict[str, Any]] = []
+        self._lock = threading.Lock()
+        self._max_parallel = int(os.environ.get("MAX_PARALLEL_TASKS", "2"))
+        self._state_file = state_file or os.path.join(
+            os.path.dirname(os.path.dirname(os.path.dirname(__file__))),
+            "output", "scheduler_state.json"
+        )
+        self._callbacks: List[Callable] = []
+        self._load_state()
+
+    def add_task(self, task_id: str, task_name: str, priority: int = 5,
+                 parameters: Optional[List[Dict[str, Any]]] = None,
+                 plan_data: Optional[Dict[str, Any]] = None,
+                 wait_for: Optional[List[str]] = None) -> Dict[str, Any]:
+        task_info = {
+            "task_id": task_id, "task_name": task_name, "priority": priority,
+            "status": "queued", "parameters": parameters or [],
+            "plan_data": plan_data or {}, "wait_for": wait_for or [],
+            "enqueued_at": datetime.now().isoformat(),
+            "started_at": None, "completed_at": None,
+            "current_point": 0, "total_points": len(parameters or []),
+            "error": None,
+        }
+        with self._lock:
+            inserted = False
+            for i, existing in enumerate(self._queue):
+                if priority > existing["priority"]:
+                    self._queue.insert(i, task_info)
+                    inserted = True
+                    break
+            if not inserted:
+                self._queue.append(task_info)
+            self._save_state_locked()
+        self._notify_callbacks()
+        return task_info
+
+    def get_next_task(self) -> Optional[Dict[str, Any]]:
+        with self._lock:
+            if len(self._running) >= self._max_parallel:
+                return None
+            for task in list(self._queue):
+                if task["wait_for"]:
+                    waiting = set(task["wait_for"])
+                    done = {t["task_id"] for t in self._completed if t["status"] == "completed"}
+                    if not waiting.issubset(done):
+                        continue
+                self._queue.remove(task)
+                task["status"] = "running"
+                task["started_at"] = datetime.now().isoformat()
+                self._running[task["task_id"]] = task
+                self._save_state_locked()
+                self._notify_callbacks()
+                return task
+        return None
+
+    def update_task_progress(self, task_id: str, current_point: int,
+                              total_points: Optional[int] = None) -> None:
+        with self._lock:
+            if task_id in self._running:
+                self._running[task_id]["current_point"] = current_point
+                if total_points:
+                    self._running[task_id]["total_points"] = total_points
+                self._save_state_locked()
+        self._notify_callbacks()
+
+    def complete_task(self, task_id: str, status: str = "completed",
+                       error: Optional[str] = None) -> None:
+        with self._lock:
+            if task_id in self._running:
+                task = self._running.pop(task_id)
+                task["status"] = status
+                task["completed_at"] = datetime.now().isoformat()
+                task["error"] = error
+                self._completed.append(task)
+                if len(self._completed) > 100:
+                    self._completed = self._completed[-100:]
+                self._save_state_locked()
+        self._notify_callbacks()
+
+    def cancel_task(self, task_id: str) -> bool:
+        with self._lock:
+            for task in list(self._queue):
+                if task["task_id"] == task_id:
+                    self._queue.remove(task)
+                    task["status"] = "cancelled"
+                    task["completed_at"] = datetime.now().isoformat()
+                    self._completed.append(task)
+                    self._save_state_locked()
+                    self._notify_callbacks()
+                    return True
+            if task_id in self._running:
+                task = self._running.pop(task_id)
+                task["status"] = "cancelled"
+                task["completed_at"] = datetime.now().isoformat()
+                self._completed.append(task)
+                self._save_state_locked()
+                self._notify_callbacks()
+                return True
+        return False
+
+    def get_statistics(self) -> Dict[str, Any]:
+        with self._lock:
+            queued = list(self._queue)
+            running = list(self._running.values())
+            completed = list(self._completed)
+            tpq = sum(t["total_points"] for t in queued)
+            tpr = sum(t["total_points"] for t in running)
+            cpr = sum(t["current_point"] for t in running)
+            tc = len([t for t in completed if t["status"] == "completed"])
+            tf = len([t for t in completed if t["status"] == "failed"])
+            total_all = tpq + tpr
+            done_all = cpr + sum(t["total_points"] for t in completed if t["status"] == "completed")
+            overall = round((done_all / total_all * 100), 1) if total_all > 0 else 0
+            return {
+                "queued_count": len(queued), "running_count": len(running),
+                "completed_count": len(completed), "successful_count": tc,
+                "failed_count": tf, "max_parallel": self._max_parallel,
+                "total_points_queued": tpq, "total_points_running": tpr,
+                "completed_points_running": cpr, "overall_progress": overall,
+                "queued_tasks": [self._summary(t) for t in queued[:20]],
+                "running_tasks": [self._summary(t) for t in running],
+                "recent_completed": [self._summary(t) for t in completed[-10:]],
+                "timestamp": datetime.now().isoformat(),
+            }
+
+    def _summary(self, task: Dict[str, Any]) -> Dict[str, Any]:
+        progress = round((task["current_point"] / task["total_points"] * 100), 1) if task["total_points"] > 0 else 0
+        return {
+            "task_id": task["task_id"], "task_name": task["task_name"],
+            "status": task["status"], "priority": task["priority"],
+            "current_point": task["current_point"], "total_points": task["total_points"],
+            "progress": progress, "enqueued_at": task.get("enqueued_at"),
+            "started_at": task.get("started_at"), "completed_at": task.get("completed_at"),
+            "error": task.get("error"),
+        }
+
+    def register_callback(self, callback: Callable) -> None:
+        self._callbacks.append(callback)
+
+    def _notify_callbacks(self) -> None:
+        stats = self.get_statistics()
+        for cb in self._callbacks:
+            try:
+                cb(stats)
+            except Exception:
+                pass
+
+    def _save_state_locked(self) -> None:
+        try:
+            os.makedirs(os.path.dirname(self._state_file), exist_ok=True)
+            state = {"queue": list(self._queue), "running": self._running,
+                     "completed": self._completed[-50:], "saved_at": datetime.now().isoformat()}
+            with open(self._state_file, "w", encoding="utf-8") as f:
+                json.dump(state, f, ensure_ascii=False, indent=2)
+        except Exception:
+            pass
+
+    def _load_state(self) -> None:
+        try:
+            if os.path.exists(self._state_file):
+                with open(self._state_file, "r", encoding="utf-8") as f:
+                    state = json.load(f)
+                with self._lock:
+                    self._queue = deque(state.get("queue", []))
+                    for task in state.get("running", {}).values():
+                        task["status"] = "queued"
+                        task["error"] = "Recovered from previous session"
+                        self._queue.append(task)
+                    self._completed = state.get("completed", [])
+        except Exception:
+            pass
+
+
+_scheduler: Optional[BatchScheduler] = None
+
+def get_scheduler() -> BatchScheduler:
+    global _scheduler
+    if _scheduler is None:
+        _scheduler = BatchScheduler()
+    return _scheduler

web/frontend/src/layouts/MainLayout.vue

@@ -28,6 +28,10 @@
           <el-icon><List /></el-icon>
           <span>任务管理</span>
         </el-menu-item>
+        <el-menu-item index="/monitor">
+          <el-icon><Monitor /></el-icon>
+          <span>实时监控</span>
+        </el-menu-item>
         <el-sub-menu index="ai">
           <template #title>
             <el-icon><MagicStick /></el-icon>
@@ -67,7 +71,7 @@
 <script setup lang="ts">
 import { computed } from 'vue'
 import { useRoute } from 'vue-router'
-import { Folder, Collection, DataAnalysis, MagicStick, List } from '@element-plus/icons-vue'
+import { Folder, Collection, DataAnalysis, MagicStick, List, Monitor } from '@element-plus/icons-vue'
 
 const route = useRoute()
 const activeMenu = computed(() => route.path)

web/frontend/src/router/index.ts

@@ -44,6 +44,12 @@ const router = createRouter({
           component: () => import('@/views/TaskManager.vue'),
           meta: { title: '任务管理' }
         },
+        {
+          path: 'monitor',
+          name: 'MonitorDashboard',
+          component: () => import('@/views/MonitorDashboard.vue'),
+          meta: { title: '实时监控' }
+        },
         {
           path: 'ai/plan-generator',
           name: 'AIPlanGenerator',

web/frontend/src/views/MonitorDashboard.vue

@@ -0,0 +1,193 @@
+<template>
+  <div class="monitor-dashboard">
+    <el-card shadow="never">
+      <template #header>
+        <div class="card-header">
+          <span class="title">实时监控仪表盘</span>
+          <div>
+            <el-tag :type="autoRefresh ? 'success' : 'info'" size="small" style="margin-right: 10px;">
+              {{ autoRefresh ? '自动刷新中' : '已暂停' }}
+            </el-tag>
+            <el-switch v-model="autoRefresh" active-text="自动" inactive-text="手动" style="margin-right: 10px;" />
+            <el-button size="small" @click="loadStats" :loading="loading">
+              <el-icon><Refresh /></el-icon> 刷新
+            </el-button>
+          </div>
+        </div>
+      </template>
+
+      <!-- 顶部统计卡片 -->
+      <el-row :gutter="16" class="stat-cards">
+        <el-col :span="4">
+          <el-card class="stat-card" shadow="hover">
+            <div class="stat-value" style="color: #909399;">{{ stats.queued_count || 0 }}</div>
+            <div class="stat-label">排队中</div>
+          </el-card>
+        </el-col>
+        <el-col :span="4">
+          <el-card class="stat-card" shadow="hover">
+            <div class="stat-value" style="color: #409eff;">{{ stats.running_count || 0 }}</div>
+            <div class="stat-label">运行中</div>
+          </el-card>
+        </el-col>
+        <el-col :span="4">
+          <el-card class="stat-card" shadow="hover">
+            <div class="stat-value" style="color: #67c23a;">{{ stats.successful_count || 0 }}</div>
+            <div class="stat-label">已成功</div>
+          </el-card>
+        </el-col>
+        <el-col :span="4">
+          <el-card class="stat-card" shadow="hover">
+            <div class="stat-value" style="color: #f56c6c;">{{ stats.failed_count || 0 }}</div>
+            <div class="stat-label">失败</div>
+          </el-card>
+        </el-col>
+        <el-col :span="4">
+          <el-card class="stat-card" shadow="hover">
+            <div class="stat-value" style="color: #e6a23c;">{{ stats.max_parallel || 2 }}</div>
+            <div class="stat-label">最大并行</div>
+          </el-card>
+        </el-col>
+        <el-col :span="4">
+          <el-card class="stat-card" shadow="hover">
+            <div class="stat-value" style="color: #409eff;">{{ stats.overall_progress || 0 }}%</div>
+            <div class="stat-label">总进度</div>
+          </el-card>
+        </el-col>
+      </el-row>
+
+      <!-- 总进度条 -->
+      <el-progress :percentage="stats.overall_progress || 0" :stroke-width="16" style="margin: 20px 0;" />
+
+      <el-row :gutter="16">
+        <!-- 运行中任务 -->
+        <el-col :span="12">
+          <el-card shadow="never">
+            <template #header>
+              <span style="font-weight: 600;">运行中任务 ({{ stats.running_count || 0 }})</span>
+            </template>
+            <div v-if="!stats.running_tasks?.length" class="empty-state">
+              <el-empty description="暂无运行中任务" :image-size="60" />
+            </div>
+            <div v-for="task in stats.running_tasks" :key="task.task_id" class="task-item">
+              <div class="task-header">
+                <span class="task-name">{{ task.task_name }}</span>
+                <el-tag size="small" type="primary">运行中</el-tag>
+              </div>
+              <el-progress :percentage="task.progress" :stroke-width="8" style="margin: 8px 0;" />
+              <div class="task-meta">
+                <span>点: {{ task.current_point }}/{{ task.total_points }}</span>
+                <span>优先级: {{ task.priority }}</span>
+                <span>开始: {{ formatTime(task.started_at) }}</span>
+              </div>
+            </div>
+          </el-card>
+        </el-col>
+
+        <!-- 排队任务 -->
+        <el-col :span="12">
+          <el-card shadow="never">
+            <template #header>
+              <span style="font-weight: 600;">排队任务 ({{ stats.queued_count || 0 }})</span>
+            </template>
+            <div v-if="!stats.queued_tasks?.length" class="empty-state">
+              <el-empty description="暂无排队任务" :image-size="60" />
+            </div>
+            <div v-for="task in stats.queued_tasks" :key="task.task_id" class="task-item">
+              <div class="task-header">
+                <span class="task-name">{{ task.task_name }}</span>
+                <el-tag size="small" :type="task.priority >= 8 ? 'danger' : task.priority >= 5 ? 'warning' : 'info'">
+                  P{{ task.priority }}
+                </el-tag>
+              </div>
+              <div class="task-meta">
+                <span>总点数: {{ task.total_points }}</span>
+                <span>入队: {{ formatTime(task.enqueued_at) }}</span>
+              </div>
+            </div>
+          </el-card>
+        </el-col>
+      </el-row>
+
+      <!-- 最近完成 -->
+      <el-card shadow="never" style="margin-top: 16px;">
+        <template #header>
+          <span style="font-weight: 600;">最近完成任务</span>
+        </template>
+        <el-table :data="stats.recent_completed || []" size="small" border>
+          <el-table-column prop="task_name" label="任务名称" min-width="150" />
+          <el-table-column label="状态" width="100">
+            <template #default="{ row }">
+              <el-tag :type="row.status === 'completed' ? 'success' : row.status === 'failed' ? 'danger' : 'info'" size="small">
+                {{ row.status === 'completed' ? '成功' : row.status === 'failed' ? '失败' : row.status }}
+              </el-tag>
+            </template>
+          </el-table-column>
+          <el-table-column prop="total_points" label="总点数" width="80" />
+          <el-table-column prop="priority" label="优先级" width="80" />
+          <el-table-column label="完成时间" width="170">
+            <template #default="{ row }">{{ formatTime(row.completed_at) }}</template>
+          </el-table-column>
+          <el-table-column prop="error" label="错误信息" min-width="200" show-overflow-tooltip />
+        </el-table>
+      </el-card>
+
+      <div class="last-update">最后更新: {{ lastUpdate || '-' }}</div>
+    </el-card>
+  </div>
+</template>
+
+<script setup lang="ts">
+import { ref, reactive, onMounted, onUnmounted } from 'vue'
+import { Refresh } from '@element-plus/icons-vue'
+import api from '@/api'
+
+const stats = reactive<any>({})
+const loading = ref(false)
+const autoRefresh = ref(true)
+const lastUpdate = ref('')
+let timer: any = null
+
+const formatTime = (t: string) => t ? new Date(t).toLocaleString('zh-CN') : '-'
+
+const loadStats = async () => {
+  loading.value = true
+  try {
+    const res: any = await api.get('/monitor/stats')
+    Object.assign(stats, res)
+    lastUpdate.value = new Date().toLocaleTimeString('zh-CN')
+  } catch (e) {
+    console.error('Failed to load stats:', e)
+  } finally {
+    loading.value = false
+  }
+}
+
+onMounted(() => {
+  loadStats()
+  timer = setInterval(() => {
+    if (autoRefresh.value) loadStats()
+  }, 5000)
+})
+
+onUnmounted(() => {
+  if (timer) clearInterval(timer)
+})
+</script>
+
+<style scoped>
+.monitor-dashboard { padding: 20px; }
+.card-header { display: flex; justify-content: space-between; align-items: center; }
+.title { font-weight: 600; font-size: 16px; }
+.stat-cards { margin-bottom: 10px; }
+.stat-card { text-align: center; }
+.stat-value { font-size: 28px; font-weight: 700; margin-bottom: 4px; }
+.stat-label { font-size: 13px; color: #909399; }
+.task-item { padding: 12px; border-bottom: 1px solid #ebeef5; }
+.task-item:last-child { border-bottom: none; }
+.task-header { display: flex; justify-content: space-between; align-items: center; }
+.task-name { font-weight: 500; }
+.task-meta { display: flex; gap: 16px; font-size: 12px; color: #909399; margin-top: 4px; }
+.empty-state { padding: 20px 0; }
+.last-update { text-align: right; font-size: 12px; color: #909399; margin-top: 10px; }
+</style>