Show compare engine in progress and SQL skip reasons.

Co-authored-by: Cursor <cursoragent@cursor.com>

netx_api/app_startup.py

@@ -159,6 +159,17 @@ def run_api_startup() -> None:
                 )
         except Exception:
             _log.exception("startup: biz_state collect recovery failed")
+        try:
+            from .biz_state.compare_service import recover_interrupted_compares_on_startup
+
+            cmp_rec = recover_interrupted_compares_on_startup(db)
+            if cmp_rec.get("runs"):
+                _log.info(
+                    "startup: cancelled %s interrupted biz compare run(s)",
+                    cmp_rec.get("runs"),
+                )
+        except Exception:
+            _log.exception("startup: biz compare recovery failed")
         try:
             from .port_traffic_migrate import backfill_port_traffic_series
 

netx_api/biz_state/compare_service.py

@@ -453,6 +453,53 @@ def _sheet_meta_from_summary(summary: dict[str, Any], run: BizCompareRun, tpl: A
     ]
 
 
+def _metric_row_estimate(
+    db: Session, batch_ids: list[str], metric_id: str
+) -> int:
+    """Cheap size hint from batch command row_count (max across sides)."""
+    mid = str(metric_id or "").strip()
+    if not mid:
+        return 0
+    best = 0
+    for bid in batch_ids:
+        bid = str(bid or "").strip()
+        if not bid:
+            continue
+        rows = (
+            db.query(BizStateBatchCommand)
+            .filter(
+                BizStateBatchCommand.batch_id == bid,
+                BizStateBatchCommand.metric_id == mid,
+            )
+            .all()
+        )
+        if not rows:
+            continue
+        n = sum(int(c.row_count or 0) for c in rows)
+        if n > best:
+            best = n
+    return best
+
+
+def _order_sheets_small_first(
+    db: Session,
+    sheets: list[dict[str, Any]],
+    *,
+    before_batch_id: str,
+    after_batch_id: str,
+) -> list[dict[str, Any]]:
+    """Run smaller metrics first so field engineers can review early results."""
+    if len(sheets) <= 1:
+        return list(sheets)
+    batches = [before_batch_id, after_batch_id]
+    scored: list[tuple[int, int, dict[str, Any]]] = []
+    for i, sheet in enumerate(sheets):
+        n = _metric_row_estimate(db, batches, str(sheet.get("metric_id") or ""))
+        scored.append((n, i, sheet))
+    scored.sort(key=lambda x: (x[0], x[1]))
+    return [s for _, _, s in scored]
+
+
 def _pending_sheet_meta(sheet: dict[str, Any]) -> dict[str, Any]:
     """Placeholder meta so the UI lists all check items while a run is in flight."""
     key_fields = list(sheet.get("key_fields") or [])
@@ -1672,6 +1719,40 @@ def _resolve_after_batch(db: Session, job: BizCompareJob) -> str:
     return str(latest.id) if latest else ""
 
 
+def _sheet_result_envelope(
+    sheet: dict[str, Any],
+    *,
+    key_fields: list[str],
+    iface_fields: list[str],
+    compare_fields: list[str],
+    display_fields: list[str],
+    row_filters: list[Any],
+    field_rules: list[Any],
+    ignore_ports: bool | None,
+    mode: str,
+    summary: dict[str, Any],
+    diffs: list[Any],
+    mapping_stats: dict[str, Any],
+) -> dict[str, Any]:
+    return {
+        "sheet_id": sheet_key(sheet),
+        "title": sheet_title(sheet),
+        "metric_id": sheet["metric_id"],
+        "key_fields": key_fields,
+        "iface_fields": iface_fields,
+        "compare_fields": compare_fields,
+        "display_fields": display_fields,
+        "row_filters": row_filters,
+        "field_rules": field_rules,
+        "ignore_port_changes": ignore_ports,
+        "mode": mode,
+        "status": "done",
+        "summary": summary,
+        "diffs": diffs,
+        "mapping_stats": mapping_stats,
+    }
+
+
 def _run_sheet(
     db: Session,
     *,
@@ -1681,7 +1762,7 @@ def _run_sheet(
     port_map: dict[str, str],
     iface_normalize_rules: list[dict[str, str]] | None = None,
     store_unchanged: str = "auto",
-    on_load_progress: Callable[[str, int], None] | None = None,
+    on_load_progress: Callable[..., None] | None = None,
 ) -> dict[str, Any]:
     key_fields = list(sheet.get("key_fields") or [])
     iface_fields = list(sheet.get("iface_fields") or [])
@@ -1704,13 +1785,77 @@ def _run_sheet(
         ignore_ports = bool(ignore_ports)
     mid = sheet["metric_id"]
 
+    def _emit_load(side: str, n: int, **meta: Any) -> None:
+        if not on_load_progress:
+            return
+        try:
+            on_load_progress(side, n, **meta)
+        except TypeError:
+            on_load_progress(side, n)
+
+    # PostgreSQL path: pushdown-safe sheets join in-DB (BGP-scale).
+    from .compare_sql import run_sql_sheet_compare, sql_compare_skip_reason
+
+    skip_reason = sql_compare_skip_reason(
+        db,
+        sheet,
+        port_map=port_map,
+        iface_normalize_rules=iface_normalize_rules,
+    )
+    if not skip_reason:
+        try:
+            result = run_sql_sheet_compare(
+                db,
+                sheet=sheet,
+                before_batch_id=before_batch_id,
+                after_batch_id=after_batch_id,
+                store_unchanged=store_unchanged,
+                on_progress=_emit_load,
+            )
+            summary = dict(result["summary"])
+            # Engine already sets raw counts / policy; keep keys stable
+            if "unchanged_policy" not in summary:
+                summary["unchanged_policy"] = resolve_unchanged_policy(
+                    store_unchanged,
+                    before_n=int(summary.get("before_count") or 0),
+                    after_n=int(summary.get("after_count") or 0),
+                )
+            return _sheet_result_envelope(
+                sheet,
+                key_fields=key_fields,
+                iface_fields=iface_fields,
+                compare_fields=compare_fields,
+                display_fields=display_fields,
+                row_filters=row_filters,
+                field_rules=field_rules,
+                ignore_ports=ignore_ports,
+                mode=mode,
+                summary=summary,
+                diffs=list(result.get("diffs") or []),
+                mapping_stats=dict(result.get("mapping_stats") or {}),
+            )
+        except Exception:
+            _log.exception(
+                "sql compare fallback sheet=%s metric=%s — using Python engine",
+                sheet_key(sheet),
+                mid,
+            )
+            skip_reason = "sql_error_fallback"
+    else:
+        _log.info(
+            "python compare sheet=%s metric=%s skip_sql=%s",
+            sheet_key(sheet),
+            mid,
+            skip_reason,
+        )
+
+    _emit_load("before", 0, engine="python", note=skip_reason or "python", phase="loading")
+
     def _before_chunk(n: int) -> None:
-        if on_load_progress:
-            on_load_progress("before", n)
+        _emit_load("before", n, engine="python", note=skip_reason or "python", phase="loading")
 
     def _after_chunk(n: int) -> None:
-        if on_load_progress:
-            on_load_progress("after", n)
+        _emit_load("after", n, engine="python", note=skip_reason or "python", phase="loading")
 
     before_raw = _load_metric_rows(
         db, batch_id=before_batch_id, metric_id=mid, on_chunk=_before_chunk
@@ -1742,23 +1887,21 @@ def _run_sheet(
     summary["after_raw_count"] = len(after_raw)
     summary["row_filters"] = len(row_filters)
     summary["unchanged_policy"] = policy
-    return {
-        "sheet_id": sheet_key(sheet),
-        "title": sheet_title(sheet),
-        "metric_id": sheet["metric_id"],
-        "key_fields": key_fields,
-        "iface_fields": iface_fields,
-        "compare_fields": compare_fields,
-        "display_fields": display_fields,
-        "row_filters": row_filters,
-        "field_rules": field_rules,
-        "ignore_port_changes": ignore_ports,
-        "mode": mode,
-        "status": "done",
-        "summary": summary,
-        "diffs": result["diffs"],
-        "mapping_stats": result["mapping_stats"],
-    }
+    summary.setdefault("engine", "python")
+    return _sheet_result_envelope(
+        sheet,
+        key_fields=key_fields,
+        iface_fields=iface_fields,
+        compare_fields=compare_fields,
+        display_fields=display_fields,
+        row_filters=row_filters,
+        field_rules=field_rules,
+        ignore_ports=ignore_ports,
+        mode=mode,
+        summary=summary,
+        diffs=list(result.get("diffs") or []),
+        mapping_stats=dict(result.get("mapping_stats") or {}),
+    )
 
 
 def _validate_compare_job(
@@ -1784,6 +1927,12 @@ def _validate_compare_job(
     sheets_cfg = _filter_enabled_sheets(sheets_cfg, getattr(j, "enabled_sheet_ids", None))
     if not sheets_cfg:
         raise HTTPException(status_code=400, detail="no_enabled_sheets")
+    sheets_cfg = _order_sheets_small_first(
+        db,
+        sheets_cfg,
+        before_batch_id=before_batch_id,
+        after_batch_id=after_batch_id,
+    )
     return {
         "job": j,
         "template": tpl,
@@ -1868,13 +2017,17 @@ def _set_run_progress(
         progress.update(extra)
     prev["progress"] = progress
     run.summary_json = prev
-    run.status = "running"
-    title = progress["sheet_title"] or progress["sheet_id"] or ""
-    run.message = (
-        f"{phase} {sheet_index}/{sheet_total}"
-        + (f" · {title}" if title else "")
-        + f" · {elapsed_ms // 1000}s"
-    )[:1024]
+    # Re-read status from DB — cancel may have been committed by another session
+    # (UI cancel / startup recovery) while this worker still holds a stale "running".
+    db.expire(run, ["status", "message"])
+    if str(run.status or "") != "cancelled":
+        run.status = "running"
+        title = progress["sheet_title"] or progress["sheet_id"] or ""
+        run.message = (
+            f"{phase} {sheet_index}/{sheet_total}"
+            + (f" · {title}" if title else "")
+            + f" · {elapsed_ms // 1000}s"
+        )[:1024]
     db.commit()
 
 
@@ -1906,6 +2059,12 @@ def _execute_compare_into_run(db: Session, run_id: str) -> dict[str, Any]:
         run.message = "no_enabled_sheets"
         db.commit()
         raise HTTPException(status_code=400, detail="no_enabled_sheets")
+    sheets_cfg = _order_sheets_small_first(
+        db,
+        sheets_cfg,
+        before_batch_id=before_batch_id,
+        after_batch_id=after_batch_id,
+    )
 
     started_mono = time.monotonic()
     store_mode = normalize_store_unchanged(getattr(j, "store_unchanged", None))
@@ -1927,20 +2086,30 @@ def _execute_compare_into_run(db: Session, run_id: str) -> dict[str, Any]:
     field_counts: dict[str, int] = {}
     total = len(sheets_cfg)
 
-    # Prefer seeded pending sheets from create; rebuild if missing
-    sheet_metas: list[dict[str, Any]] = list((run.summary_json or {}).get("sheets") or [])
-    if len(sheet_metas) != total:
-        sheet_metas = [_pending_sheet_meta(s) for s in sheets_cfg]
+    # Prefer seeded pending sheets from create; realign to small-first order
+    prev_metas = list((run.summary_json or {}).get("sheets") or [])
+    by_key = {sheet_key(m): m for m in prev_metas}
+    sheet_metas = [
+        by_key.get(sheet_key(s)) or _pending_sheet_meta(s) for s in sheets_cfg
+    ]
 
     def _publish_sheets() -> None:
         prev = dict(run.summary_json or {})
         prev["sheets"] = list(sheet_metas)
         prev.update({k: agg[k] for k in agg})
         run.summary_json = prev
+        db.expire(run, ["status"])
+        # Never resurrect cancelled while publishing incremental sheet metas
+        if str(run.status or "") == "cancelled":
+            db.commit()
+            return
         db.commit()
 
     try:
         for idx, sheet in enumerate(sheets_cfg, start=1):
+            if _run_is_cancelled(db, run_id):
+                run = db.get(BizCompareRun, run_id) or run
+                return get_run(db, run.id)
             sid = sheet_key(sheet)
             # Mark current sheet running in the sidebar list
             for meta in sheet_metas:
@@ -1949,27 +2118,43 @@ def _execute_compare_into_run(db: Session, run_id: str) -> dict[str, Any]:
                     break
             _publish_sheets()
 
-            _load_pub = {"t": 0.0, "n": -1}
+            _load_pub = {"t": 0.0, "n": -1, "engine": ""}
 
-            def _on_load(side: str, n: int, _idx: int = idx, _sheet: dict = sheet) -> None:
+            def _on_load(
+                side: str,
+                n: int,
+                *,
+                engine: str = "python",
+                note: str = "",
+                phase: str | None = None,
+                _idx: int = idx,
+                _sheet: dict = sheet,
+            ) -> None:
                 now = time.monotonic()
-                # Avoid committing every 5k on million-row BGP loads
-                if n - _load_pub["n"] < 25_000 and now - _load_pub["t"] < 2.0:
-                    return
+                eng = str(engine or "python")
+                # SQL emits sparse updates; Python still throttle chunk spam
+                if eng != "sql":
+                    if n - _load_pub["n"] < 25_000 and now - _load_pub["t"] < 2.0:
+                        return
                 _load_pub["t"] = now
                 _load_pub["n"] = n
+                _load_pub["engine"] = eng
+                extra: dict[str, Any] = {
+                    "load_side": side,
+                    "rows_loaded": n,
+                    "engine": eng,
+                }
+                if note:
+                    extra["engine_note"] = str(note)[:128]
                 _set_run_progress(
                     db,
                     run,
-                    phase="loading",
+                    phase=str(phase or ("loading" if eng != "sql" else "sql_count")),
                     sheet_index=_idx,
                     sheet_total=total,
                     sheet=_sheet,
                     started_mono=started_mono,
-                    extra={
-                        "load_side": side,
-                        "rows_loaded": n,
-                    },
+                    extra=extra,
                 )
 
             _set_run_progress(
@@ -1980,6 +2165,7 @@ def _execute_compare_into_run(db: Session, run_id: str) -> dict[str, Any]:
                 sheet_total=total,
                 sheet=sheet,
                 started_mono=started_mono,
+                extra={"engine": "", "engine_note": ""},
             )
             one = _run_sheet(
                 db,
@@ -2091,6 +2277,13 @@ def _execute_compare_into_run(db: Session, run_id: str) -> dict[str, Any]:
             fail_diffs.clear()
             ok_diffs.clear()
             db.commit()
+            if _run_is_cancelled(db, run_id):
+                run = db.get(BizCompareRun, run_id) or run
+                return get_run(db, run.id)
+
+        if _run_is_cancelled(db, run_id):
+            run = db.get(BizCompareRun, run_id) or run
+            return get_run(db, run.id)
 
         duration_ms = int((time.monotonic() - started_mono) * 1000)
         top_fields = sorted(
@@ -2117,6 +2310,9 @@ def _execute_compare_into_run(db: Session, run_id: str) -> dict[str, Any]:
             },
             "sheets": sheet_metas,
         }
+        run = db.get(BizCompareRun, run_id) or run
+        if str(run.status or "") == "cancelled":
+            return get_run(db, run.id)
         run.status = "success"
         run.summary_json = summary_payload
         run.mapping_stats_json = mapping_by_metric
@@ -2129,34 +2325,123 @@ def _execute_compare_into_run(db: Session, run_id: str) -> dict[str, Any]:
         db.commit()
         return get_run(db, run.id)
     except HTTPException as exc:
-        run.status = "failed"
-        run.message = str(getattr(exc, "detail", "") or exc)[:1024]
-        prev = dict(run.summary_json or {})
-        prog = dict(prev.get("progress") or {})
-        prog["phase"] = "failed"
-        prog["elapsed_ms"] = int((time.monotonic() - started_mono) * 1000)
-        prev["progress"] = prog
-        prev["duration_ms"] = prog["elapsed_ms"]
-        run.summary_json = prev
-        db.commit()
+        run = db.get(BizCompareRun, run_id) or run
+        if str(run.status or "") != "cancelled":
+            run.status = "failed"
+            run.message = str(getattr(exc, "detail", "") or exc)[:1024]
+            prev = dict(run.summary_json or {})
+            prog = dict(prev.get("progress") or {})
+            prog["phase"] = "failed"
+            prog["elapsed_ms"] = int((time.monotonic() - started_mono) * 1000)
+            prev["progress"] = prog
+            prev["duration_ms"] = prog["elapsed_ms"]
+            run.summary_json = prev
+            db.commit()
         raise
     except Exception as exc:
         _log.exception("compare run failed run=%s job=%s", run_id, j.id)
-        run.status = "failed"
-        run.message = str(exc)[:1024]
-        prev = dict(run.summary_json or {})
-        prog = dict(prev.get("progress") or {})
-        prog["phase"] = "failed"
-        prog["elapsed_ms"] = int((time.monotonic() - started_mono) * 1000)
-        prev["progress"] = prog
-        prev["duration_ms"] = prog["elapsed_ms"]
-        run.summary_json = prev
-        db.commit()
+        run = db.get(BizCompareRun, run_id) or run
+        if str(run.status or "") != "cancelled":
+            run.status = "failed"
+            run.message = str(exc)[:1024]
+            prev = dict(run.summary_json or {})
+            prog = dict(prev.get("progress") or {})
+            prog["phase"] = "failed"
+            prog["elapsed_ms"] = int((time.monotonic() - started_mono) * 1000)
+            prev["progress"] = prog
+            prev["duration_ms"] = prog["elapsed_ms"]
+            run.summary_json = prev
+            db.commit()
         raise
 
 
+_INTERRUPT_MARK = "interrupted_by_restart"
+
+
+def _job_has_active_run(db: Session, job_id: str) -> BizCompareRun | None:
+    return (
+        db.query(BizCompareRun)
+        .filter(
+            BizCompareRun.job_id == str(job_id or ""),
+            BizCompareRun.status.in_(("running", "queued")),
+        )
+        .order_by(BizCompareRun.created_at.desc())
+        .first()
+    )
+
+
+def _run_is_cancelled(db: Session, run_id: str) -> bool:
+    """Re-read status so user/startup cancel is visible to the worker thread."""
+    db.expire_all()
+    r = db.get(BizCompareRun, run_id)
+    return bool(r and str(r.status or "") == "cancelled")
+
+
+def recover_interrupted_compares_on_startup(db: Session) -> dict[str, Any]:
+    """Mark orphaned running/queued compare runs as cancelled after process restart.
+
+    In-memory job locks die with the process; without this, the UI stays on
+    「比对中」and blocks a new run.
+    """
+    now = _utcnow()
+    rows = (
+        db.query(BizCompareRun)
+        .filter(BizCompareRun.status.in_(("running", "queued")))
+        .all()
+    )
+    n = 0
+    for r in rows:
+        r.status = "cancelled"
+        msg = str(r.message or "").strip()
+        if _INTERRUPT_MARK not in msg:
+            r.message = f"{msg} | {_INTERRUPT_MARK}".strip(" |")[:1024]
+        prev = dict(r.summary_json or {})
+        prog = dict(prev.get("progress") or {})
+        prog["phase"] = "cancelled"
+        prog["elapsed_ms"] = int(prog.get("elapsed_ms") or 0)
+        prev["progress"] = prog
+        # Mark in-flight sheet placeholders so UI does not show fake pass
+        sheets = list(prev.get("sheets") or [])
+        for sh in sheets:
+            st = str(sh.get("status") or "")
+            if st in ("pending", "running", "queued"):
+                sh["status"] = "cancelled"
+        prev["sheets"] = sheets
+        r.summary_json = prev
+        n += 1
+    if n:
+        db.commit()
+        _log.info("startup: cancelled %s interrupted compare run(s)", n)
+    return {"runs": n, "at": now.isoformat() + "Z"}
+
+
+def cancel_compare_run(db: Session, run_id: str) -> dict[str, Any]:
+    """Cancel a running/queued compare so a new run can start."""
+    r = db.get(BizCompareRun, run_id)
+    if not r:
+        raise HTTPException(status_code=404, detail="run_not_found")
+    st = str(r.status or "")
+    if st not in ("running", "queued"):
+        return get_run(db, run_id)
+    r.status = "cancelled"
+    msg = str(r.message or "").strip()
+    r.message = f"{msg} | cancelled_by_user".strip(" |")[:1024]
+    prev = dict(r.summary_json or {})
+    prog = dict(prev.get("progress") or {})
+    prog["phase"] = "cancelled"
+    prev["progress"] = prog
+    for sh in list(prev.get("sheets") or []):
+        if str(sh.get("status") or "") in ("pending", "running", "queued"):
+            sh["status"] = "cancelled"
+    r.summary_json = prev
+    db.commit()
+    return get_run(db, run_id)
+
+
 def run_compare(db: Session, job_id: str, *, force_after_batch_id: str = "") -> dict[str, Any]:
     """Synchronous compare (auto-compare / tests). Blocks the caller until done."""
+    if _job_has_active_run(db, job_id):
+        raise HTTPException(status_code=409, detail="compare_already_running")
     lock = _job_compare_lock(job_id)
     if not lock.acquire(blocking=False):
         raise HTTPException(status_code=409, detail="compare_already_running")
@@ -2183,6 +2468,8 @@ def enqueue_compare(
     Returns immediately so the HTTP worker / UI stay responsive. Poll
     ``GET /compare/runs/{id}`` for progress (``summary.progress``).
     """
+    if _job_has_active_run(db, job_id):
+        raise HTTPException(status_code=409, detail="compare_already_running")
     lock = _job_compare_lock(job_id)
     if not lock.acquire(blocking=False):
         raise HTTPException(status_code=409, detail="compare_already_running")

netx_api/biz_state/compare_sql.py

@@ -0,0 +1,646 @@
+"""PostgreSQL-backed sheet compare (FULL OUTER JOIN) for pushdown-safe sheets.
+
+Falls back to the Python engine when port-map / complex rules cannot be expressed
+in SQL. See ``can_sql_compare``.
+"""
+
+from __future__ import annotations
+
+import logging
+import re
+from typing import Any, Callable, Mapping, Sequence
+from uuid import uuid4
+
+from sqlalchemy import text
+from sqlalchemy.orm import Session
+
+from .compare_rules import effective_compare_fields, field_rule_map
+
+_log = logging.getLogger("netx.biz_state.compare_sql")
+
+_FIELD_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
+_SQL_FILTER_OPS = frozenset(
+    {"eq", "==", "ne", "!=", "in", "not_in", "nin", "contains", "empty", "not_empty", "nonempty", "ci_eq"}
+)
+_SQL_NORMALIZE = frozenset({"", "none", "strip", "lower", "upper", "empty_as_blank"})
+_SQL_COMPARE_MODES = frozenset({"", "eq", "ignore", "skip", "off"})
+_EMPTY_AS_BLANK = ("n/a", "na", "-", "--", "none", "null")
+
+
+def _dialect_is_postgres(db: Session) -> bool:
+    bind = db.get_bind()
+    if bind is None:
+        return False
+    return str(getattr(bind.dialect, "name", "") or "").lower() in ("postgresql", "postgres")
+
+
+def _safe_field(name: str) -> str:
+    f = str(name or "").strip()
+    if not f or not _FIELD_RE.match(f):
+        raise ValueError(f"unsafe_json_field:{name!r}")
+    return f
+
+
+def _json_text_expr(alias: str, field: str) -> str:
+    """SQL expression: trimmed text from JSONB column ``alias.data`` / ``data_json``."""
+    f = _safe_field(field)
+    # alias is caller-controlled (_b / data_json) — not user input
+    return f"trim(both from coalesce({alias}->>'{f}', ''))"
+
+
+def _norm_expr(alias: str, field: str, normalize: str) -> str:
+    base = _json_text_expr(alias, field)
+    mode = str(normalize or "strip").strip().lower() or "strip"
+    if mode in ("", "none", "strip"):
+        return base
+    if mode == "lower":
+        return f"lower({base})"
+    if mode == "upper":
+        return f"upper({base})"
+    if mode == "empty_as_blank":
+        opts = ", ".join(f"'{x}'" for x in _EMPTY_AS_BLANK)
+        return (
+            f"CASE WHEN lower({base}) IN ({opts}) THEN '' ELSE {base} END"
+        )
+    raise ValueError(f"unsupported_normalize:{mode}")
+
+
+def _filters_sql_compatible(filters: Sequence[Mapping[str, Any]] | None) -> bool:
+    fl = [f for f in (filters or []) if isinstance(f, dict)]
+    return all(_filter_node_compatible(f) for f in fl)
+
+
+def _filter_node_compatible(filt: Mapping[str, Any]) -> bool:
+    if "any" in filt:
+        kids = filt.get("any") or []
+        return isinstance(kids, list) and all(
+            isinstance(k, dict) and _filter_node_compatible(k) for k in kids
+        )
+    if "all" in filt:
+        kids = filt.get("all") or []
+        return isinstance(kids, list) and all(
+            isinstance(k, dict) and _filter_node_compatible(k) for k in kids
+        )
+    field = str(filt.get("field") or "").strip()
+    op = str(filt.get("op") or "eq").strip().lower()
+    if op not in _SQL_FILTER_OPS:
+        return False
+    if op in ("empty", "not_empty", "nonempty"):
+        return bool(field) and bool(_FIELD_RE.match(field))
+    if not field or not _FIELD_RE.match(field):
+        return False
+    if op in ("in", "not_in", "nin"):
+        vals = filt.get("value")
+        if vals is None:
+            return True
+        if not isinstance(vals, (list, tuple)):
+            vals = [vals]
+        return all(isinstance(x, (str, int, float, bool)) or x is None for x in vals)
+    return True
+
+
+def _field_rules_sql_compatible(rules: Sequence[Mapping[str, Any]] | None) -> bool:
+    for raw in rules or []:
+        if not isinstance(raw, dict):
+            return False
+        name = str(raw.get("field") or "").strip()
+        if name and not _FIELD_RE.match(name):
+            return False
+        mode = str(raw.get("compare") or "eq").strip().lower() or "eq"
+        if mode not in _SQL_COMPARE_MODES:
+            return False
+        if raw.get("ignore") is True:
+            continue
+        if mode in ("ignore", "skip", "off"):
+            continue
+        norm = str(raw.get("normalize") or "strip").strip().lower() or "strip"
+        if norm not in _SQL_NORMALIZE:
+            return False
+    return True
+
+
+def sql_compare_skip_reason(
+    db: Session,
+    sheet: Mapping[str, Any],
+    *,
+    port_map: Mapping[str, str] | None = None,
+    iface_normalize_rules: Sequence[Mapping[str, str]] | None = None,
+) -> str:
+    """Empty string when SQL is allowed; otherwise a short reason for progress/logs.
+
+    Simple ``row_filters`` (eq/in/contains/…) used for BGP sheet splits are fine —
+    they do **not** force Python by themselves.
+    """
+    if not _dialect_is_postgres(db):
+        return "not_postgres"
+    if port_map:
+        return "port_map"
+    key_fields = [str(k).strip() for k in (sheet.get("key_fields") or []) if str(k).strip()]
+    if not key_fields:
+        return "no_key_fields"
+    if any(not _FIELD_RE.match(k) for k in key_fields):
+        return "unsafe_key_field"
+    iface_fields = [str(f).strip() for f in (sheet.get("iface_fields") or []) if str(f).strip()]
+    iface_set = set(iface_fields)
+    ignore_ports = sheet.get("ignore_port_changes")
+    if ignore_ports is True:
+        return "ignore_port_changes"
+    if ignore_ports is None and iface_set and any(k in iface_set for k in key_fields):
+        return "auto_ignore_ports"
+    field_rules = list(sheet.get("field_rules") or [])
+    if not _field_rules_sql_compatible(field_rules):
+        return "field_rules"
+    compare_fields = effective_compare_fields(
+        list(sheet.get("compare_fields") or []),
+        field_rules,
+    )
+    if any(not _FIELD_RE.match(str(f).strip()) for f in compare_fields if str(f).strip()):
+        return "unsafe_compare_field"
+    used = set(key_fields) | {str(f).strip() for f in compare_fields if str(f).strip()}
+    if iface_normalize_rules and (used & iface_set):
+        return "iface_normalize"
+    if not _filters_sql_compatible(list(sheet.get("row_filters") or [])):
+        return "row_filters"
+    if not str(sheet.get("metric_id") or "").strip():
+        return "no_metric_id"
+    return ""
+
+
+def can_sql_compare(
+    db: Session,
+    sheet: Mapping[str, Any],
+    *,
+    port_map: Mapping[str, str] | None = None,
+    iface_normalize_rules: Sequence[Mapping[str, str]] | None = None,
+) -> bool:
+    """True when this sheet can run entirely as a PostgreSQL JOIN."""
+    return not bool(
+        sql_compare_skip_reason(
+            db,
+            sheet,
+            port_map=port_map,
+            iface_normalize_rules=iface_normalize_rules,
+        )
+    )
+
+
+def compile_row_filters_sql(
+    filters: Sequence[Mapping[str, Any]] | None,
+    *,
+    json_col: str = "data_json",
+    param_prefix: str = "f",
+) -> tuple[str, dict[str, Any]]:
+    """Compile template row_filters to SQL AND-clause + bind params.
+
+    Returns ``(sql_fragment, params)``. Empty filters → ``(\"TRUE\", {})``.
+    ``json_col`` is the JSONB column/expression name (trusted).
+    """
+    fl = [f for f in (filters or []) if isinstance(f, dict)]
+    if not fl:
+        return "TRUE", {}
+    params: dict[str, Any] = {}
+    counter = {"n": 0}
+
+    def _next(name: str) -> str:
+        counter["n"] += 1
+        return f"{param_prefix}_{counter['n']}_{name}"
+
+    def _node(filt: Mapping[str, Any]) -> str:
+        if "any" in filt:
+            kids = [k for k in (filt.get("any") or []) if isinstance(k, dict)]
+            if not kids:
+                return "TRUE"
+            return "(" + " OR ".join(_node(k) for k in kids) + ")"
+        if "all" in filt:
+            kids = [k for k in (filt.get("all") or []) if isinstance(k, dict)]
+            if not kids:
+                return "TRUE"
+            return "(" + " AND ".join(_node(k) for k in kids) + ")"
+        field = _safe_field(str(filt.get("field") or ""))
+        op = str(filt.get("op") or "eq").strip().lower()
+        expr = _json_text_expr(json_col, field)
+        # _json_text_expr uses alias->> ; for bare column use data_json directly
+        if json_col == "data_json":
+            expr = f"trim(both from coalesce(data_json->>'{field}', ''))"
+        if op in ("empty",):
+            return f"({expr} = '')"
+        if op in ("not_empty", "nonempty"):
+            return f"({expr} <> '')"
+        expect = filt.get("value")
+        if op in ("eq", "==", "ci_eq"):
+            key = _next("v")
+            params[key] = str(expect or "").strip()
+            if op == "ci_eq" or op in ("eq", "=="):
+                # Python eq is case-insensitive via .lower()
+                params[key] = str(expect or "").strip().lower()
+                return f"(lower({expr}) = :{key})"
+        if op in ("ne", "!="):
+            key = _next("v")
+            params[key] = str(expect or "").strip().lower()
+            return f"(lower({expr}) <> :{key})"
+        if op == "contains":
+            key = _next("v")
+            needle = str(expect or "").strip().lower()
+            params[key] = f"%{needle}%"
+            return f"(lower({expr}) LIKE :{key})"
+        if op in ("in", "not_in", "nin"):
+            vals = expect
+            if vals is None:
+                vals = []
+            if not isinstance(vals, (list, tuple)):
+                vals = [vals]
+            clean = [str(x).strip().lower() for x in vals if str(x).strip()]
+            if not clean:
+                return "FALSE" if op == "in" else "TRUE"
+            keys = []
+            for i, v in enumerate(clean):
+                k = _next(f"in{i}")
+                params[k] = v
+                keys.append(f":{k}")
+            inside = f"lower({expr}) IN ({', '.join(keys)})"
+            return f"({inside})" if op == "in" else f"(NOT {inside})"
+        raise ValueError(f"unsupported_filter_op:{op}")
+
+    parts = [_node(f) for f in fl]
+    return "(" + " AND ".join(parts) + ")", params
+
+
+def _rk_sql(key_fields: list[str], *, json_col: str = "data_json") -> str:
+    parts = []
+    for k in key_fields:
+        f = _safe_field(k)
+        if json_col == "data_json":
+            parts.append(f"trim(both from coalesce(data_json->>'{f}', ''))")
+        else:
+            parts.append(f"trim(both from coalesce({json_col}->>'{f}', ''))")
+    if len(parts) == 1:
+        return parts[0]
+    return "concat_ws('|', " + ", ".join(parts) + ")"
+
+
+def _changed_predicate(
+    compare_fields: list[str],
+    rules: dict[str, dict[str, Any]],
+    *,
+    before_alias: str = "b",
+    after_alias: str = "a",
+) -> str:
+    """SQL boolean: True when any compare field differs (IS DISTINCT FROM)."""
+    if not compare_fields:
+        return "FALSE"
+    clauses: list[str] = []
+    for f in compare_fields:
+        name = str(f).strip()
+        if not name:
+            continue
+        rule = rules.get(name) or {}
+        mode = str(rule.get("compare") or "eq").strip().lower() or "eq"
+        if mode in ("ignore", "skip", "off") or rule.get("ignore") is True:
+            continue
+        norm = str(rule.get("normalize") or "strip").strip().lower() or "strip"
+        bv = _norm_expr(f"{before_alias}.data", name, norm)
+        av = _norm_expr(f"{after_alias}.data", name, norm)
+        clauses.append(f"({bv} IS DISTINCT FROM {av})")
+    if not clauses:
+        return "FALSE"
+    return "(" + " OR ".join(clauses) + ")"
+
+
+def _key_obj_from_data(data: dict[str, Any] | None, key_fields: list[str]) -> dict[str, Any]:
+    row = data if isinstance(data, dict) else {}
+    return {f: row.get(f, "") for f in key_fields}
+
+
+def _changes_from_rows(
+    before: dict[str, Any] | None,
+    after: dict[str, Any] | None,
+    compare_fields: list[str],
+    rules: dict[str, dict[str, Any]],
+) -> dict[str, dict[str, Any]]:
+    from .compare_rules import explain_diff, values_equal
+
+    out: dict[str, dict[str, Any]] = {}
+    b = before or {}
+    a = after or {}
+    for f in compare_fields:
+        rule = rules.get(f)
+        bv = b.get(f, "")
+        av = a.get(f, "")
+        if values_equal(bv, av, rule=rule):
+            continue
+        entry: dict[str, Any] = {"before": bv, "after": av}
+        reason = explain_diff(bv, av, rule=rule)
+        if reason:
+            entry["reason"] = reason
+        out[f] = entry
+    return out
+
+
+def run_sql_sheet_compare(
+    db: Session,
+    *,
+    sheet: Mapping[str, Any],
+    before_batch_id: str,
+    after_batch_id: str,
+    store_unchanged: str = "auto",
+    on_progress: Callable[..., None] | None = None,
+) -> dict[str, Any]:
+    """Compare one sheet via PostgreSQL TEMP tables + FULL OUTER JOIN.
+
+    Returns the same ``{summary, diffs, mapping_stats}`` shape as ``compare_rows``.
+    ``on_progress(side, n, *, engine=\"sql\", note=..., phase=...)``.
+    """
+    if not _dialect_is_postgres(db):
+        raise RuntimeError("sql_compare_requires_postgres")
+
+    def _prog(side: str, n: int, *, phase: str = "sql_count", note: str = "") -> None:
+        if not on_progress:
+            return
+        try:
+            on_progress(side, n, engine="sql", note=note, phase=phase)
+        except TypeError:
+            on_progress(side, n)
+
+    key_fields = [str(k).strip() for k in (sheet.get("key_fields") or []) if str(k).strip()]
+    field_rules = list(sheet.get("field_rules") or [])
+    rules = field_rule_map(field_rules)
+    compare_fields = effective_compare_fields(
+        list(sheet.get("compare_fields") or []),
+        field_rules,
+    )
+    row_filters = list(sheet.get("row_filters") or [])
+    mid = str(sheet.get("metric_id") or "").strip()
+    bid_b = str(before_batch_id or "").strip()
+    bid_a = str(after_batch_id or "").strip()
+    if not mid or not bid_b or not bid_a or not key_fields:
+        raise ValueError("sql_compare_missing_args")
+
+    filter_sql, filter_params = compile_row_filters_sql(row_filters)
+    rk = _rk_sql(key_fields)
+    tag = uuid4().hex[:8]
+    tb = f"_netx_cmp_b_{tag}"
+    ta = f"_netx_cmp_a_{tag}"
+
+    _prog("before", 0, phase="sql_count", note="count")
+
+    # Raw counts (no row_filters)
+    raw_b = int(
+        db.execute(
+            text(
+                "SELECT count(*) FROM biz_state_metric_row "
+                "WHERE batch_id = :bid AND metric_id = :mid"
+            ),
+            {"bid": bid_b, "mid": mid},
+        ).scalar()
+        or 0
+    )
+    _prog("before", raw_b, phase="sql_count")
+    raw_a = int(
+        db.execute(
+            text(
+                "SELECT count(*) FROM biz_state_metric_row "
+                "WHERE batch_id = :bid AND metric_id = :mid"
+            ),
+            {"bid": bid_a, "mid": mid},
+        ).scalar()
+        or 0
+    )
+    _prog("after", raw_a, phase="sql_count")
+
+    base_params = {"bid": bid_b, "mid": mid, **filter_params}
+    # Build TEMP sides
+    for tname, batch_id, side in ((tb, bid_b, "before"), (ta, bid_a, "after")):
+        db.execute(text(f"DROP TABLE IF EXISTS {tname}"))
+        params = {**base_params, "bid": batch_id}
+        _prog(side, raw_b if side == "before" else raw_a, phase="sql_project", note="temp")
+        # PRESERVE ROWS: compare progress commits must not drop temps mid-run
+        db.execute(
+            text(
+                f"""
+                CREATE TEMP TABLE {tname} ON COMMIT PRESERVE ROWS AS
+                SELECT
+                  id,
+                  ({rk}) AS rk,
+                  data_json AS data,
+                  row_number() OVER (
+                    PARTITION BY ({rk})
+                    ORDER BY seq ASC, id ASC
+                  ) AS dup_rn
+                FROM biz_state_metric_row
+                WHERE batch_id = :bid
+                  AND metric_id = :mid
+                  AND ({filter_sql})
+                """
+            ),
+            params,
+        )
+        db.execute(text(f"CREATE INDEX ON {tname} (rk) WHERE dup_rn = 1"))
+
+    before_n = int(
+        db.execute(text(f"SELECT count(*) FROM {tb}")).scalar() or 0
+    )
+    after_n = int(
+        db.execute(text(f"SELECT count(*) FROM {ta}")).scalar() or 0
+    )
+    _prog("after", after_n, phase="sql_join", note="join")
+
+    changed_pred = _changed_predicate(compare_fields, rules)
+    kind_expr = f"""
+      CASE
+        WHEN b.id IS NULL THEN 'added'
+        WHEN a.id IS NULL THEN 'removed'
+        WHEN {changed_pred} THEN 'changed'
+        ELSE 'unchanged'
+      END
+    """
+
+    # Aggregate primary match kinds (first-wins keys only)
+    agg_rows = db.execute(
+        text(
+            f"""
+            SELECT {kind_expr} AS kind, count(*)::bigint AS n
+            FROM (SELECT * FROM {tb} WHERE dup_rn = 1) b
+            FULL OUTER JOIN (SELECT * FROM {ta} WHERE dup_rn = 1) a
+              ON b.rk = a.rk
+            GROUP BY 1
+            """
+        )
+    ).mappings().all()
+    counts = {str(r["kind"]): int(r["n"] or 0) for r in agg_rows}
+    added = counts.get("added", 0)
+    removed = counts.get("removed", 0)
+    changed = counts.get("changed", 0)
+    unchanged = counts.get("unchanged", 0)
+
+    dup_b = int(
+        db.execute(text(f"SELECT count(*) FROM {tb} WHERE dup_rn > 1")).scalar() or 0
+    )
+    dup_a = int(
+        db.execute(text(f"SELECT count(*) FROM {ta} WHERE dup_rn > 1")).scalar() or 0
+    )
+    duplicate = dup_b + dup_a
+
+    # Lazy import — avoid circular import with compare_service
+    from .compare_service import resolve_unchanged_policy
+
+    policy = resolve_unchanged_policy(
+        store_unchanged, before_n=before_n, after_n=after_n
+    )
+    diffs: list[dict[str, Any]] = []
+
+    # Fail + duplicate rows (stream into Python — should be << million)
+    fail_sql = text(
+        f"""
+        SELECT
+          {kind_expr} AS kind,
+          b.id AS before_row_id,
+          a.id AS after_row_id,
+          b.data AS before_data,
+          a.data AS after_data,
+          COALESCE(b.rk, a.rk) AS rk
+        FROM (SELECT * FROM {tb} WHERE dup_rn = 1) b
+        FULL OUTER JOIN (SELECT * FROM {ta} WHERE dup_rn = 1) a
+          ON b.rk = a.rk
+        WHERE {kind_expr} IN ('added', 'removed', 'changed')
+        """
+    )
+    for row in db.execute(fail_sql).mappings():
+        kind = str(row["kind"] or "")
+        before = dict(row["before_data"] or {}) if row["before_data"] is not None else None
+        after = dict(row["after_data"] or {}) if row["after_data"] is not None else None
+        key_src = after if kind == "added" else (before or after or {})
+        item: dict[str, Any] = {
+            "kind": kind,
+            "key": _key_obj_from_data(key_src, key_fields),
+            "before": before,
+            "after": after,
+            "mapped_before": before,
+            "changes": {},
+            "before_row_id": str(row["before_row_id"] or ""),
+            "after_row_id": str(row["after_row_id"] or ""),
+        }
+        if kind == "changed":
+            item["changes"] = _changes_from_rows(before, after, compare_fields, rules)
+        diffs.append(item)
+
+    # Duplicate extras
+    for side, tname in (("before", tb), ("after", ta)):
+        q = text(
+            f"""
+            SELECT id, data, rk FROM {tname} WHERE dup_rn > 1
+            """
+        )
+        for row in db.execute(q).mappings():
+            data = dict(row["data"] or {})
+            diffs.append(
+                {
+                    "kind": "duplicate",
+                    "side": side,
+                    "key": _key_obj_from_data(data, key_fields),
+                    "before": data if side == "before" else None,
+                    "after": data if side == "after" else None,
+                    "mapped_before": data if side == "before" else None,
+                    "changes": {},
+                    "before_row_id": str(row["id"] or "") if side == "before" else "",
+                    "after_row_id": str(row["id"] or "") if side == "after" else "",
+                }
+            )
+
+    unchanged_listed = 0
+    include_u = bool(policy.get("include"))
+    compact = bool(policy.get("compact"))
+    limit_n = policy.get("limit")
+    if include_u and unchanged > 0:
+        lim_sql = ""
+        params_u: dict[str, Any] = {}
+        if limit_n is not None:
+            lim_sql = " LIMIT :lim"
+            params_u["lim"] = max(0, int(limit_n))
+        u_sql = text(
+            f"""
+            SELECT
+              b.id AS before_row_id,
+              a.id AS after_row_id,
+              b.data AS before_data,
+              a.data AS after_data
+            FROM (SELECT * FROM {tb} WHERE dup_rn = 1) b
+            INNER JOIN (SELECT * FROM {ta} WHERE dup_rn = 1) a
+              ON b.rk = a.rk
+            WHERE NOT ({changed_pred})
+            {lim_sql}
+            """
+        )
+        for row in db.execute(u_sql, params_u).mappings():
+            before = dict(row["before_data"] or {})
+            after = dict(row["after_data"] or {})
+            unchanged_listed += 1
+            if compact:
+                diffs.append(
+                    {
+                        "kind": "unchanged",
+                        "key": _key_obj_from_data(before, key_fields),
+                        "before": {},
+                        "after": {},
+                        "mapped_before": {},
+                        "changes": {},
+                        "compact": True,
+                        "before_row_id": str(row["before_row_id"] or ""),
+                        "after_row_id": str(row["after_row_id"] or ""),
+                    }
+                )
+            else:
+                diffs.append(
+                    {
+                        "kind": "unchanged",
+                        "key": _key_obj_from_data(before, key_fields),
+                        "before": before,
+                        "after": after,
+                        "mapped_before": before,
+                        "changes": {},
+                        "before_row_id": str(row["before_row_id"] or ""),
+                        "after_row_id": str(row["after_row_id"] or ""),
+                    }
+                )
+
+    # Cleanup (also ON COMMIT DROP)
+    db.execute(text(f"DROP TABLE IF EXISTS {tb}"))
+    db.execute(text(f"DROP TABLE IF EXISTS {ta}"))
+
+    dup_key_list: list[str] = []
+    summary = {
+        "before_count": before_n,
+        "after_count": after_n,
+        "added": added,
+        "removed": removed,
+        "changed": changed,
+        "unchanged": unchanged,
+        "duplicate": duplicate,
+        "match_key_fields": list(key_fields),
+        "duplicate_keys_before": dup_b,
+        "duplicate_keys_after": dup_a,
+        "duplicate_key_list": dup_key_list,
+        "unchanged_listed": unchanged_listed,
+        "unchanged_truncated": bool(
+            include_u and limit_n is not None and unchanged > unchanged_listed
+        ),
+        "unchanged_compact": bool(compact and unchanged_listed > 0),
+        "engine": "sql",
+        "before_raw_count": raw_b,
+        "after_raw_count": raw_a,
+        "row_filters": len(row_filters),
+        "unchanged_policy": policy,
+    }
+    mapping_stats = {
+        "before_iface_count": 0,
+        "after_iface_count": 0,
+        "map_pairs": 0,
+        "hit_before": [],
+        "miss_before": [],
+        "hit_after": [],
+        "miss_after": [],
+        "unused_before_keys": [],
+        "ok": True,
+        "ignore_port_changes": False,
+        "engine": "sql",
+    }
+    return {"summary": summary, "diffs": diffs, "mapping_stats": mapping_stats}

netx_api/biz_state_router.py

@@ -746,6 +746,12 @@ def api_get_run(run_id: str, db: Session = Depends(get_db)) -> dict[str, Any]:
     return cmp_svc.get_run(db, run_id)
 
 
+@router.post("/compare/runs/{run_id}/cancel")
+def api_cancel_run(run_id: str, db: Session = Depends(get_db)) -> dict[str, Any]:
+    """Cancel a stuck/running compare so a new run can start."""
+    return cmp_svc.cancel_compare_run(db, run_id)
+
+
 @router.delete("/compare/runs/{run_id}")
 def api_delete_run(run_id: str, db: Session = Depends(get_db)) -> dict[str, Any]:
     return cmp_svc.delete_run(db, run_id)

tests/test_biz_state_compare_recovery.py

@@ -0,0 +1,98 @@
+"""Startup recovery for interrupted biz-state compare runs."""
+
+from __future__ import annotations
+
+import unittest
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+from netx_api.biz_state.compare_service import (
+    cancel_compare_run,
+    recover_interrupted_compares_on_startup,
+)
+from netx_api.db import Base
+from netx_api.models import BizCompareJob, BizCompareRun
+
+
+class BizStateCompareRecoveryTests(unittest.TestCase):
+    def setUp(self) -> None:
+        engine = create_engine("sqlite+pysqlite:///:memory:", future=True)
+        TestingSession = sessionmaker(
+            bind=engine, autoflush=False, autocommit=False, expire_on_commit=False
+        )
+        Base.metadata.create_all(bind=engine)
+        self.db = TestingSession()
+        self.job = BizCompareJob(
+            id="j1",
+            name="cutover",
+            template_id="tpl1",
+            status="active",
+        )
+        self.db.add(self.job)
+        self.running = BizCompareRun(
+            id="r_run",
+            job_id="j1",
+            status="running",
+            message="loading 1/2 · BGP",
+            summary_json={
+                "progress": {"phase": "loading", "sheet_index": 1, "sheet_total": 2},
+                "sheets": [
+                    {"sheet_id": "bgp", "status": "running"},
+                    {"sheet_id": "isis", "status": "pending"},
+                ],
+            },
+        )
+        self.queued = BizCompareRun(
+            id="r_q",
+            job_id="j1",
+            status="queued",
+            message="queued",
+            summary_json={"sheets": [{"sheet_id": "bgp", "status": "pending"}]},
+        )
+        self.ok = BizCompareRun(
+            id="r_ok",
+            job_id="j1",
+            status="success",
+            message="done",
+            summary_json={"added": 0, "removed": 0, "changed": 0},
+        )
+        self.db.add_all([self.running, self.queued, self.ok])
+        self.db.commit()
+
+    def tearDown(self) -> None:
+        self.db.close()
+
+    def test_startup_cancels_running_and_queued(self) -> None:
+        out = recover_interrupted_compares_on_startup(self.db)
+        self.assertEqual(out["runs"], 2)
+
+        self.db.refresh(self.running)
+        self.db.refresh(self.queued)
+        self.db.refresh(self.ok)
+        self.assertEqual(self.running.status, "cancelled")
+        self.assertIn("interrupted_by_restart", self.running.message or "")
+        self.assertEqual(
+            (self.running.summary_json or {}).get("progress", {}).get("phase"),
+            "cancelled",
+        )
+        sheets = list((self.running.summary_json or {}).get("sheets") or [])
+        self.assertEqual(sheets[0].get("status"), "cancelled")
+        self.assertEqual(sheets[1].get("status"), "cancelled")
+        self.assertEqual(self.queued.status, "cancelled")
+        self.assertEqual(self.ok.status, "success")
+
+    def test_cancel_compare_run_user(self) -> None:
+        out = cancel_compare_run(self.db, "r_run")
+        self.assertEqual(out["status"], "cancelled")
+        self.db.refresh(self.running)
+        self.assertEqual(self.running.status, "cancelled")
+        self.assertIn("cancelled_by_user", self.running.message or "")
+
+    def test_cancel_idempotent_on_success(self) -> None:
+        out = cancel_compare_run(self.db, "r_ok")
+        self.assertEqual(out["status"], "success")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_biz_state_compare_sql.py

@@ -0,0 +1,218 @@
+"""Unit tests for SQL compare eligibility and filter compilation."""
+
+from __future__ import annotations
+
+import unittest
+from unittest.mock import MagicMock
+
+from netx_api.biz_state.compare_sql import (
+    can_sql_compare,
+    compile_row_filters_sql,
+    sql_compare_skip_reason,
+    _field_rules_sql_compatible,
+    _filters_sql_compatible,
+    _safe_field,
+)
+
+
+class _FakeDialect:
+    def __init__(self, name: str) -> None:
+        self.name = name
+
+
+class _FakeBind:
+    def __init__(self, name: str) -> None:
+        self.dialect = _FakeDialect(name)
+
+
+def _db(dialect: str = "postgresql") -> MagicMock:
+    db = MagicMock()
+    db.get_bind.return_value = _FakeBind(dialect)
+    return db
+
+
+class CompareSqlGateTests(unittest.TestCase):
+    def test_safe_field_rejects_injection(self) -> None:
+        with self.assertRaises(ValueError):
+            _safe_field("a'; drop table x;--")
+        with self.assertRaises(ValueError):
+            _safe_field("x.y")
+        self.assertEqual(_safe_field("network"), "network")
+
+    def test_requires_postgres(self) -> None:
+        sheet = {
+            "metric_id": "bgp_route",
+            "key_fields": ["network", "next_hop"],
+            "compare_fields": ["path"],
+            "iface_fields": [],
+            "field_rules": [],
+            "row_filters": [],
+        }
+        self.assertFalse(can_sql_compare(_db("sqlite"), sheet, port_map={}))
+        self.assertTrue(can_sql_compare(_db("postgresql"), sheet, port_map={}))
+
+    def test_rejects_port_map(self) -> None:
+        sheet = {
+            "metric_id": "lldp_neighbor",
+            "key_fields": ["local_if", "remote_sys"],
+            "compare_fields": [],
+            "iface_fields": ["local_if"],
+            "ignore_port_changes": False,
+            "field_rules": [],
+            "row_filters": [],
+        }
+        self.assertFalse(
+            can_sql_compare(_db(), sheet, port_map={"gei-0/1": "gei-0/2"})
+        )
+
+    def test_rejects_auto_ignore_ports_with_iface_key(self) -> None:
+        sheet = {
+            "metric_id": "lldp_neighbor",
+            "key_fields": ["local_if", "remote_sys"],
+            "compare_fields": [],
+            "iface_fields": ["local_if"],
+            "ignore_port_changes": None,
+            "field_rules": [],
+            "row_filters": [],
+        }
+        self.assertFalse(can_sql_compare(_db(), sheet, port_map={}))
+
+    def test_rejects_mac_normalize(self) -> None:
+        sheet = {
+            "metric_id": "arp",
+            "key_fields": ["ip", "vrf"],
+            "compare_fields": ["mac"],
+            "iface_fields": [],
+            "field_rules": [{"field": "mac", "normalize": "mac"}],
+            "row_filters": [],
+        }
+        self.assertFalse(can_sql_compare(_db(), sheet, port_map={}))
+
+    def test_rejects_age_timer_filter(self) -> None:
+        sheet = {
+            "metric_id": "arp",
+            "key_fields": ["ip"],
+            "compare_fields": [],
+            "iface_fields": [],
+            "field_rules": [],
+            "row_filters": [{"field": "age", "op": "age_timer"}],
+        }
+        self.assertFalse(can_sql_compare(_db(), sheet, port_map={}))
+
+    def test_accepts_bgp_route_style(self) -> None:
+        sheet = {
+            "metric_id": "bgp_route",
+            "key_fields": [
+                "local_as",
+                "afi",
+                "vrf",
+                "neighbor",
+                "direction",
+                "rd",
+                "network",
+                "next_hop",
+            ],
+            "compare_fields": ["path", "as_num"],
+            "iface_fields": [],
+            "field_rules": [],
+            "row_filters": [{"field": "afi", "op": "eq", "value": "ipv4"}],
+        }
+        self.assertTrue(can_sql_compare(_db(), sheet, port_map={}))
+        # BGP afi/vrf sheet splits via row_filters must not force Python
+        self.assertEqual(sql_compare_skip_reason(_db(), sheet, port_map={}), "")
+
+    def test_rejects_iface_normalize_when_key_uses_iface(self) -> None:
+        sheet = {
+            "metric_id": "interface_brief",
+            "key_fields": ["interface"],
+            "compare_fields": ["admin"],
+            "iface_fields": ["interface"],
+            "ignore_port_changes": False,
+            "field_rules": [],
+            "row_filters": [],
+        }
+        self.assertFalse(
+            can_sql_compare(
+                _db(),
+                sheet,
+                port_map={},
+                iface_normalize_rules=[{"from": "GE", "to": "gei"}],
+            )
+        )
+
+    def test_accepts_ignore_port_changes_false_without_normalize(self) -> None:
+        sheet = {
+            "metric_id": "interface_brief",
+            "key_fields": ["interface"],
+            "compare_fields": ["admin"],
+            "iface_fields": ["interface"],
+            "ignore_port_changes": False,
+            "field_rules": [],
+            "row_filters": [],
+        }
+        self.assertTrue(can_sql_compare(_db(), sheet, port_map={}, iface_normalize_rules=[]))
+
+
+class CompareSqlFilterCompileTests(unittest.TestCase):
+    def test_empty_filters(self) -> None:
+        sql, params = compile_row_filters_sql([])
+        self.assertEqual(sql, "TRUE")
+        self.assertEqual(params, {})
+
+    def test_eq_case_insensitive(self) -> None:
+        sql, params = compile_row_filters_sql(
+            [{"field": "afi", "op": "eq", "value": "IPv4"}]
+        )
+        self.assertIn("lower(", sql)
+        self.assertIn("afi", sql)
+        self.assertEqual(list(params.values()), ["ipv4"])
+
+    def test_contains(self) -> None:
+        sql, params = compile_row_filters_sql(
+            [{"field": "af", "op": "contains", "value": "IPv4"}]
+        )
+        self.assertIn("LIKE", sql)
+        self.assertEqual(list(params.values()), ["%ipv4%"])
+
+    def test_any_all_nesting(self) -> None:
+        sql, params = compile_row_filters_sql(
+            [
+                {
+                    "any": [
+                        {"field": "entry_type", "op": "eq", "value": "dynamic"},
+                        {
+                            "all": [
+                                {"field": "entry_type", "op": "empty"},
+                                {"field": "ip", "op": "not_empty"},
+                            ]
+                        },
+                    ]
+                }
+            ]
+        )
+        self.assertIn(" OR ", sql)
+        self.assertIn(" AND ", sql)
+        self.assertTrue(_filters_sql_compatible([{"any": [{"field": "a", "op": "eq", "value": "1"}]}]))
+
+    def test_in_list(self) -> None:
+        sql, params = compile_row_filters_sql(
+            [{"field": "afi", "op": "in", "value": ["ipv4", "ipv6"]}]
+        )
+        self.assertIn(" IN (", sql)
+        self.assertEqual(sorted(params.values()), ["ipv4", "ipv6"])
+
+    def test_field_rules_matrix(self) -> None:
+        self.assertTrue(_field_rules_sql_compatible([{"field": "mac", "normalize": "lower"}]))
+        self.assertFalse(_field_rules_sql_compatible([{"field": "mac", "normalize": "mac"}]))
+        self.assertFalse(
+            _field_rules_sql_compatible(
+                [{"field": "rx", "compare": "numeric", "tolerance": 1}]
+            )
+        )
+        self.assertTrue(
+            _field_rules_sql_compatible([{"field": "x", "compare": "ignore"}])
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

web/src/i18n/en.ts

@@ -501,6 +501,10 @@ const en = {
     runStatusDoneSec: "Done {{s}}s",
     runProgress: "{{phase}} · sheet {{i}}/{{n}} · {{sheet}} · {{s}}s elapsed",
     runRowsLoaded: "Loaded {{side}} {{n}} rows",
+    runRowsSqlCount: "DB count {{side}} {{n}} rows",
+    runEngineSql: "engine SQL",
+    runEnginePython: "engine Python",
+    runEngineNote: "reason {{note}}",
     sheetPending: "Pending",
     ranWithDuration: "Compare finished ({{s}}s)",
     unchangedNotStored: "Success rows were counted but not stored. Set “Store success rows” to sample and re-run for spot-check.",
@@ -520,10 +524,20 @@ const en = {
     edit: "Edit",
     delete: "Delete",
     tabConfig: "Job config",
+    tabRuns: "Compare batches",
     tabResult: "Result",
     runs: "Run history",
+    runsHint: "Each “Run now” creates a batch. Stuck “Running” after restart is auto-cancelled; you can also cancel manually and re-run.",
     noRuns: "No runs yet — run a compare first",
     result: "Result",
+    viewResult: "View result",
+    cancelRun: "Cancel",
+    confirmCancelRun: "Cancel this compare batch? You can start a new run afterward.",
+    runCancelled: "Compare cancelled",
+    runStatusCancelled: "Cancelled",
+    compareAlreadyRunning: "A compare is already running — cancel it under Compare batches or wait",
+    runBatchSummary: "Fail {{fail}} · Pass {{ok}} · Added {{added}}",
+    colTime: "Time",
     pickBatchRun: "Select compare run",
     pickRun: "Select run…",
     runCount: "{{n}} runs",

web/src/i18n/zh.ts

@@ -500,6 +500,10 @@ const zh = {
     runStatusDoneSec: "完成 {{s}}s",
     runProgress: "{{phase}} · 表 {{i}}/{{n}} · {{sheet}} · 已用 {{s}}s",
     runRowsLoaded: "已加载 {{side}} {{n}} 行",
+    runRowsSqlCount: "库内统计 {{side}} {{n}} 行",
+    runEngineSql: "引擎 SQL",
+    runEnginePython: "引擎 Python",
+    runEngineNote: "原因 {{note}}",
     sheetPending: "等待中",
     ranWithDuration: "比对完成（耗时 {{s}} 秒）",
     unchangedNotStored: "成功行仅统计数量未落库。可在任务配置将「成功行保存」改为抽样后重新比对（抽查用）。",
@@ -519,10 +523,20 @@ const zh = {
     edit: "编辑",
     delete: "删除",
     tabConfig: "任务配置",
+    tabRuns: "对比批次",
     tabResult: "比对结果",
     runs: "历史比对",
+    runsHint: "每次「立即比对」生成一条批次。重启后卡住的「比对中」会自动标为已取消；也可手动终止后重跑。",
     noRuns: "尚无比对记录，请先执行比对",
     result: "比对结果",
+    viewResult: "查看结果",
+    cancelRun: "终止",
+    confirmCancelRun: "确定终止该比对批次？终止后可重新发起比对。",
+    runCancelled: "比对已取消",
+    runStatusCancelled: "已取消",
+    compareAlreadyRunning: "已有比对在进行中，请先在「对比批次」中终止或等待完成",
+    runBatchSummary: "失败 {{fail}} · 成功 {{ok}} · 新增 {{added}}",
+    colTime: "时间",
     pickBatchRun: "选择比对记录",
     pickRun: "选择比对记录…",
     runCount: "{{n}} 次",

web/src/pages/network/BizComparePage.tsx

@@ -7,6 +7,7 @@ import { useDebouncedValue } from "../../hooks/useDebouncedValue";
 import { useToast } from "../../hooks/useToast";
 import { useI18n } from "../../i18n";
 import {
+  bizCompareCancelRun,
   bizCompareCreateJob,
   bizCompareCreateMapping,
   bizCompareCreateTemplate,
@@ -35,7 +36,7 @@ import { cutoverCachedGet, cutoverCachedGetSWR, invalidateCutoverCache } from ".
 import { jobChipColor, NmStatusChip } from "./nmChips";
 
 type PageTab = "templates" | "jobs";
-type JobDetailTab = "config" | "result";
+type JobDetailTab = "config" | "runs" | "result";
 type KindFilter = "diff" | "all" | "added" | "removed" | "changed" | "unchanged";
 type CreateJobStep = 0 | 1 | 2 | 3;
 const CREATE_JOB_STEPS = 4;
@@ -137,6 +138,8 @@ type RunSheet = {
   display_fields?: string[];
   field_rules?: FieldRule[];
   mode?: string;
+  /** pending|running|queued|done|cancelled — set while overall run is active */
+  status?: string;
   summary?: Record<string, number>;
   diffs?: DiffRow[];
 };
@@ -1060,14 +1063,10 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage
   useEffect(() => {
     const runId = String(runDetail?.id || "");
     const mid = resultSheetId || sheetIdentity(activeRunSheet) || "";
-    const st = String(runDetail?.status || "");
-    if (
-      !runId ||
-      !mid ||
-      jobDetailTab !== "result" ||
-      st === "running" ||
-      st === "queued"
-    ) {
+    const sheetSt = String(activeRunSheet?.status || "");
+    // Block only while *this* sheet is still in flight — done sheets are readable mid-run
+    const sheetStillRunning = ["pending", "running", "queued"].includes(sheetSt);
+    if (!runId || !mid || jobDetailTab !== "result" || sheetStillRunning) {
       setPagedDiffs([]);
       setResultTotal(0);
       return;
@@ -1109,6 +1108,7 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage
     runDetail?.status,
     resultSheetId,
     activeRunSheet?.metric_id,
+    activeRunSheet?.status,
     kindFilter,
     debouncedResultKw,
     resultPage,
@@ -1802,12 +1802,14 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage
     const job = jobs.find((x) => x.id === id);
     if (job) resetJobForm(job);
     try {
-      const r = await bizCompareListRuns(id);
-      setRuns(r.items || []);
-      if ((r.items || []).length) {
-        const latest = await bizCompareGetRun(String((r.items as any[])[0].id));
+      const r = await bizCompareListRuns(id, 50);
+      const items = r.items || [];
+      setRuns(items);
+      if (items.length) {
+        const latest = await bizCompareGetRun(String((items as any[])[0].id));
         setRunDetail(latest);
-        setJobDetailTab("result");
+        // Land on batch list so stuck/cancelled runs are visible and actionable
+        setJobDetailTab("runs");
       }
     } catch (e) {
       showError(formatErr(e));
@@ -1843,32 +1845,42 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage
     }
   };
 
+  const runStatus = String(runDetail?.status || "");
+  const runIsActive = runStatus === "running" || runStatus === "queued";
+  const jobHasActiveRun = runs.some((r) => {
+    const st = String((r as any).status || "");
+    return st === "running" || st === "queued";
+  });
+
   const runNow = async () => {
     if (!jobId) return;
     if (!enabledJobSheetCount) {
       showError(t("bizCompare.needSheets"));
       return;
     }
+    if (jobHasActiveRun) {
+      showError(t("bizCompare.compareAlreadyRunning"));
+      setJobDetailTab("runs");
+      return;
+    }
     setBusy(true);
     try {
       await bizCompareUpdateJob(jobId, jobConfigBody());
       // Async enqueue — returns immediately with status=running; poll below.
       const run = await bizCompareRunJob(jobId);
       setRunDetail(run);
-      setJobDetailTab("result");
+      setJobDetailTab("runs");
       showOk(t("bizCompare.runStarted"));
-      const r = await bizCompareListRuns(jobId);
+      const r = await bizCompareListRuns(jobId, 50);
       setRuns(r.items || []);
       await refresh({ force: true });
     } catch (e) {
       showError(formatErr(e));
+      setJobDetailTab("runs");
     } finally {
       setBusy(false);
     }
   };
-
-  const runStatus = String(runDetail?.status || "");
-  const runIsActive = runStatus === "running" || runStatus === "queued";
   const runProgress = (runDetail?.summary?.progress || {}) as {
     phase?: string;
     sheet_index?: number;
@@ -1881,7 +1893,10 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage
     diff_rows?: number;
     load_side?: string;
     rows_loaded?: number;
+    engine?: string;
+    engine_note?: string;
   };
+  const runEngine = String(runProgress.engine || "").toLowerCase();
 
   // Poll active compare runs so the modal can be closed and reopened safely.
   useEffect(() => {
@@ -1906,6 +1921,9 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage
         } else if (!notified && st === "failed") {
           notified = true;
           showError(String(d.message || t("bizCompare.runFailed")));
+        } else if (!notified && st === "cancelled") {
+          notified = true;
+          showError(String(d.message || t("bizCompare.runCancelled")));
         }
       } catch (e) {
         if (!cancelled) showError(formatErr(e));
@@ -1919,6 +1937,31 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage
     };
   }, [runDetail?.id, runIsActive, jobId, showOk, showError, t]);
 
+  // Keep the runs list fresh while any batch on this job is active
+  useEffect(() => {
+    if (!jobId || !jobHasActiveRun || jobDetailTab !== "runs") return;
+    let cancelled = false;
+    const tick = async () => {
+      try {
+        const r = await bizCompareListRuns(jobId, 50);
+        if (!cancelled) setRuns(r.items || []);
+        const curId = String(runDetail?.id || "");
+        if (curId) {
+          const d = await bizCompareGetRun(curId);
+          if (!cancelled) setRunDetail(d);
+        }
+      } catch {
+        /* ignore list poll errors */
+      }
+    };
+    void tick();
+    const id = window.setInterval(() => void tick(), 2500);
+    return () => {
+      cancelled = true;
+      window.clearInterval(id);
+    };
+  }, [jobId, jobHasActiveRun, jobDetailTab, runDetail?.id]);
+
   const loadRun = async (runId: string) => {
     try {
       const d = await bizCompareGetRun(runId);
@@ -1931,6 +1974,25 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage
     }
   };
 
+  const cancelRun = async (runId: string) => {
+    if (!runId) return;
+    if (!window.confirm(t("bizCompare.confirmCancelRun"))) return;
+    setBusy(true);
+    try {
+      const d = await bizCompareCancelRun(runId);
+      if (String(runDetail?.id || "") === runId) setRunDetail(d);
+      if (jobId) {
+        const r = await bizCompareListRuns(jobId, 50);
+        setRuns(r.items || []);
+      }
+      showOk(t("bizCompare.runCancelled"));
+    } catch (e) {
+      showError(formatErr(e));
+    } finally {
+      setBusy(false);
+    }
+  };
+
   const removeRun = async (runId: string) => {
     if (!runId) return;
     if (!window.confirm(t("bizCompare.confirmDeleteRun"))) return;
@@ -1940,19 +2002,16 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage
       await bizCompareDeleteRun(runId);
       let nextRuns: typeof runs = [];
       if (jobId) {
-        const r = await bizCompareListRuns(jobId);
+        const r = await bizCompareListRuns(jobId, 50);
         nextRuns = r.items || [];
       } else {
         nextRuns = (runs || []).filter((r) => String(r.id) !== runId);
       }
       setRuns(nextRuns);
       if (wasCurrent) {
-        if (nextRuns.length) {
-          await loadRun(String(nextRuns[0].id));
-        } else {
-          setRunDetail(null);
-          setResultSheetId("");
-        }
+        setRunDetail(null);
+        setResultSheetId("");
+        setJobDetailTab("runs");
       }
       showOk(t("bizCompare.runDeleted"));
     } catch (e) {
@@ -2864,6 +2923,18 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage
             >
               {t("bizCompare.tabConfig")}
             </Button>
+            <Button
+              size="sm"
+              variant={jobDetailTab === "runs" ? "primary" : "secondary"}
+              className={jobDetailTab === "runs" ? "is-active" : undefined}
+              onPress={() => {
+                setJobDetailTab("runs");
+                if (jobId) void bizCompareListRuns(jobId, 50).then((r) => setRuns(r.items || []));
+              }}
+            >
+              {t("bizCompare.tabRuns")}
+              {runs.length ? ` (${runs.length})` : ""}
+            </Button>
             <Button
               size="sm"
               variant={jobDetailTab === "result" ? "primary" : "secondary"}
@@ -2876,6 +2947,169 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage
 
           {jobDetailTab === "config" ? (
             renderJobForm()
+          ) : jobDetailTab === "runs" ? (
+            <div className="bs-cmp-runs">
+              <div className="btn-row" style={{ marginBottom: 8, justifyContent: "space-between" }}>
+                <p className="muted bm-hint" style={{ margin: 0 }}>
+                  {t("bizCompare.runsHint")}
+                </p>
+                <Button
+                  size="sm"
+                  variant="primary"
+                  isDisabled={busy || jobHasActiveRun}
+                  onPress={() => void runNow()}
+                >
+                  {jobHasActiveRun ? t("bizCompare.runStatusRunning") : t("bizCompare.runNow")}
+                </Button>
+              </div>
+              {!runs.length ? (
+                <div className="pt-list-empty">{t("bizCompare.noRuns")}</div>
+              ) : (
+                <div className="table-wrap">
+                  <table className="data-table">
+                    <thead>
+                      <tr>
+                        <th>{t("bizCompare.colStatus")}</th>
+                        <th>{t("bizCompare.colTime")}</th>
+                        <th>{t("bizCompare.sidesTitle")}</th>
+                        <th>{t("bizCompare.colResult")}</th>
+                        <th>{t("bizCompare.colActions")}</th>
+                      </tr>
+                    </thead>
+                    <tbody>
+                      {runs.map((r) => {
+                        const st = String((r as any).status || "");
+                        const active = st === "running" || st === "queued";
+                        const selected = String(runDetail?.id || "") === String(r.id);
+                        const before = enrichSide(
+                          (r as any).before,
+                          beforeTaskId,
+                          tasks,
+                          beforeBatches,
+                        );
+                        const after = enrichSide(
+                          (r as any).after,
+                          afterTaskId || beforeTaskId,
+                          tasks,
+                          afterBatches.length ? afterBatches : beforeBatches,
+                        );
+                        const when = formatSystemTime((r as any).created_at) || "—";
+                        const sum = ((r as any).summary || {}) as Record<string, number>;
+                        const fail =
+                          Number(sum.removed || 0) + Number(sum.changed || 0);
+                        const ok = Number(sum.unchanged || 0);
+                        const durMs = Number(sum.duration_ms || 0);
+                        const prog = ((r as any).progress || {}) as {
+                          phase?: string;
+                          sheet_index?: number;
+                          sheet_total?: number;
+                          sheet_title?: string;
+                          rows_loaded?: number;
+                          load_side?: string;
+                          engine?: string;
+                          engine_note?: string;
+                        };
+                        const eng = String(prog.engine || "").toLowerCase();
+                        const stLabel =
+                          st === "running" || st === "queued"
+                            ? t("bizCompare.runStatusRunning")
+                            : st === "failed"
+                              ? t("bizCompare.runStatusFailed")
+                              : st === "cancelled"
+                                ? t("bizCompare.runStatusCancelled")
+                                : t("bizCompare.runStatusDoneSec", {
+                                    s: String(durMs > 0 ? Math.round(durMs / 1000) : 0),
+                                  });
+                        return (
+                          <tr key={String(r.id)} className={selected ? "is-selected" : undefined}>
+                            <td>
+                              <NmStatusChip color={jobChipColor(st)}>{stLabel}</NmStatusChip>
+                              {active && (prog.phase || eng) ? (
+                                <div className="muted" style={{ fontSize: "0.75rem", marginTop: 4 }}>
+                                  {eng === "sql"
+                                    ? t("bizCompare.runEngineSql")
+                                    : eng === "python"
+                                      ? t("bizCompare.runEnginePython")
+                                      : ""}
+                                  {eng && prog.engine_note
+                                    ? ` · ${prog.engine_note}`
+                                    : ""}
+                                  {prog.phase ? `${eng ? " · " : ""}${prog.phase}` : ""}
+                                  {prog.sheet_total
+                                    ? ` · ${prog.sheet_index || 0}/${prog.sheet_total}`
+                                    : ""}
+                                  {prog.sheet_title ? ` · ${prog.sheet_title}` : ""}
+                                  {Number(prog.rows_loaded || 0) > 0
+                                    ? eng === "sql"
+                                      ? ` · ${prog.load_side || ""} ${prog.rows_loaded}`
+                                      : ` · ${prog.load_side || ""} ${prog.rows_loaded}`
+                                    : ""}
+                                </div>
+                              ) : null}
+                              {(r as any).message ? (
+                                <div
+                                  className="muted"
+                                  style={{ fontSize: "0.75rem", marginTop: 2 }}
+                                  title={String((r as any).message)}
+                                >
+                                  {String((r as any).message).slice(0, 80)}
+                                </div>
+                              ) : null}
+                            </td>
+                            <td>{when}</td>
+                            <td>
+                              {sideDeviceName(before)} {sideCollectTime(before)}
+                              {" → "}
+                              {sideDeviceName(after)} {sideCollectTime(after)}
+                            </td>
+                            <td>
+                              {active
+                                ? "…"
+                                : t("bizCompare.runBatchSummary", {
+                                    fail: String(fail),
+                                    ok: String(ok),
+                                    added: String(Number(sum.added || 0)),
+                                  })}
+                            </td>
+                            <td>
+                              <div className="btn-row">
+                                <Button
+                                  size="sm"
+                                  variant="secondary"
+                                  isDisabled={busy || active}
+                                  onPress={() => void loadRun(String(r.id))}
+                                >
+                                  {t("bizCompare.viewResult")}
+                                </Button>
+                                {active ? (
+                                  <Button
+                                    size="sm"
+                                    variant="danger"
+                                    isDisabled={busy}
+                                    onPress={() => void cancelRun(String(r.id))}
+                                  >
+                                    {t("bizCompare.cancelRun")}
+                                  </Button>
+                                ) : (
+                                  <Button
+                                    size="sm"
+                                    variant="danger"
+                                    isDisabled={busy}
+                                    onPress={() => void removeRun(String(r.id))}
+                                  >
+                                    {t("bizCompare.deleteRun")}
+                                  </Button>
+                                )}
+                              </div>
+                            </td>
+                          </tr>
+                        );
+                      })}
+                    </tbody>
+                  </table>
+                </div>
+              )}
+            </div>
           ) : (
             <div
               ref={boardRef}
@@ -3022,6 +3256,18 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage
               {runDetail && runIsActive ? (
                 <div className="bs-cmp-progress" role="status" aria-live="polite">
                   <strong>{t("bizCompare.runStatusRunning")}</strong>
+                  {runEngine === "sql" || runEngine === "python" ? (
+                    <span className="muted">
+                      {runEngine === "sql"
+                        ? t("bizCompare.runEngineSql")
+                        : t("bizCompare.runEnginePython")}
+                      {runProgress.engine_note
+                        ? ` · ${t("bizCompare.runEngineNote", {
+                            note: String(runProgress.engine_note),
+                          })}`
+                        : ""}
+                    </span>
+                  ) : null}
                   <span className="muted">
                     {t("bizCompare.runProgress", {
                       phase: String(runProgress.phase || "…"),
@@ -3035,10 +3281,15 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage
                   </span>
                   {Number(runProgress.rows_loaded || 0) > 0 ? (
                     <span className="muted">
-                      {t("bizCompare.runRowsLoaded", {
-                        side: String(runProgress.load_side || "—"),
-                        n: String(runProgress.rows_loaded || 0),
-                      })}
+                      {runEngine === "sql"
+                        ? t("bizCompare.runRowsSqlCount", {
+                            side: String(runProgress.load_side || "—"),
+                            n: String(runProgress.rows_loaded || 0),
+                          })
+                        : t("bizCompare.runRowsLoaded", {
+                            side: String(runProgress.load_side || "—"),
+                            n: String(runProgress.rows_loaded || 0),
+                          })}
                     </span>
                   ) : null}
                   {runDetail.message ? (
@@ -3495,7 +3746,7 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage
                                       !Number(runDetail?.summary?.unchanged_listed || 0)
                                       ? t("bizCompare.unchangedNotStored")
                                       : t("bizCompare.resultEmpty")
-                                    : runIsActive
+                                    : activeSheetPending
                                       ? t("bizCompare.runStatusRunning")
                                       : t("bizCompare.resultEmpty")}
                                 </div>
@@ -3536,10 +3787,10 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage
               <Button
                 size="sm"
                 variant="primary"
-                isDisabled={busy || runIsActive}
+                isDisabled={busy || jobHasActiveRun}
                 onPress={() => void runNow()}
               >
-                {runIsActive ? t("bizCompare.runStatusRunning") : t("bizCompare.runNow")}
+                {jobHasActiveRun ? t("bizCompare.runStatusRunning") : t("bizCompare.runNow")}
               </Button>
               {jobId ? (
                 <Button size="sm" variant="danger" isDisabled={busy} onPress={() => void removeJob(jobId)}>
@@ -3547,12 +3798,21 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage
                 </Button>
               ) : null}
             </>
+          ) : jobDetailTab === "runs" ? (
+            <Button
+              size="sm"
+              variant="primary"
+              isDisabled={busy || jobHasActiveRun}
+              onPress={() => void runNow()}
+            >
+              {jobHasActiveRun ? t("bizCompare.runStatusRunning") : t("bizCompare.runNow")}
+            </Button>
           ) : (
             <>
               <Button
                 size="sm"
                 variant="secondary"
-                isDisabled={busy || !runDetail?.id}
+                isDisabled={busy || runIsActive || !runDetail?.id}
                 onPress={() => void downloadRunTables()}
               >
                 {t("bizCompare.exportTables")}
@@ -3568,10 +3828,10 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage
               <Button
                 size="sm"
                 variant="primary"
-                isDisabled={busy || runIsActive}
+                isDisabled={busy || jobHasActiveRun}
                 onPress={() => void runNow()}
               >
-                {runIsActive ? t("bizCompare.runStatusRunning") : t("bizCompare.runNow")}
+                {jobHasActiveRun ? t("bizCompare.runStatusRunning") : t("bizCompare.runNow")}
               </Button>
             </>
           )}

web/src/services/api.ts

@@ -2154,6 +2154,12 @@ export const bizCompareListRuns = (jobId: string, limit = 20) =>
 export const bizCompareGetRun = (runId: string) =>
   apiGet<Record<string, unknown>>(`/v1/biz-state/compare/runs/${encodeURIComponent(runId)}`);
 
+export const bizCompareCancelRun = (runId: string) =>
+  apiPost<Record<string, unknown>>(
+    `/v1/biz-state/compare/runs/${encodeURIComponent(runId)}/cancel`,
+    {},
+  );
+
 export const bizCompareDeleteRun = (runId: string) =>
   apiDelete<{ ok: boolean; job_id?: string; run_id?: string }>(
     `/v1/biz-state/compare/runs/${encodeURIComponent(runId)}`,