Unify compare search via live source lookup; stratify success samples.

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

netx_api/biz_state/compare_engine.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+from collections import defaultdict, deque
 from typing import Any, Mapping, Sequence
 
 from .compare_rules import field_rule_map, values_equal, explain_diff
@@ -11,6 +12,53 @@ from .iface_normalize import (
     resolve_mapped_iface,
 )
 
+# Prefer these fields when stratifying success samples (BGP multipath / multi-cmd).
+_STRATUM_FIELDS = ("neighbor", "direction", "afi", "vrf")
+
+
+def stratum_key(row: Mapping[str, Any] | None) -> str:
+    """Bucket key for stratified unchanged sampling."""
+    if not isinstance(row, Mapping):
+        return "_"
+    parts: list[str] = []
+    for f in _STRATUM_FIELDS:
+        v = str(row.get(f) or "").strip()
+        if v:
+            parts.append(f"{f}={v}")
+    return "|".join(parts) if parts else "_"
+
+
+def stratify_take(items: list[Any], limit: int, *, key_fn) -> list[Any]:
+    """Round-robin across strata so one neighbor/direction cannot consume the whole sample."""
+    lim = max(0, int(limit))
+    if lim <= 0 or not items:
+        return []
+    if len(items) <= lim:
+        return list(items)
+    buckets: dict[str, deque[Any]] = defaultdict(deque)
+    order: list[str] = []
+    for it in items:
+        sk = str(key_fn(it) or "_")
+        if sk not in buckets:
+            order.append(sk)
+        buckets[sk].append(it)
+    out: list[Any] = []
+    while len(out) < lim and buckets:
+        drained: list[str] = []
+        for sk in order:
+            q = buckets.get(sk)
+            if not q:
+                drained.append(sk)
+                continue
+            out.append(q.popleft())
+            if len(out) >= lim:
+                break
+        for sk in drained:
+            buckets.pop(sk, None)
+            if sk in order:
+                order = [x for x in order if x != sk]
+    return out
+
 
 def apply_port_map(
     row: dict[str, Any],
@@ -111,9 +159,14 @@ def compare_rows(
 ) -> dict[str, Any]:
     """Return summary + diffs list.
 
-    Diff kinds: added | removed | changed | unchanged | duplicate
+    Diff kinds: added | removed | changed | unchanged
 
-    Pipeline: iface normalize (both sides) → port map (before) → match.
+    Pipeline: iface normalize (both sides) → port map (before) → ordered
+    same-key pairing (load / list order within each match key).
+
+    Same match key with N before and M after rows: zip by index
+    ``0..min(N,M)-1`` for field compare; extras become ``removed`` (before)
+    or ``added`` (after). Example: 5 vs 2 → 2 compared + 3 removed.
 
     ``include_unchanged``: when False, matching rows still increment
     ``summary.unchanged`` but are omitted from ``diffs``.
@@ -129,8 +182,8 @@ def compare_rows(
       - ``True``: force drop iface from match key when a non-empty candidate exists.
       - ``False``: never drop iface from match key.
 
-    Duplicate match keys are not silently discarded: extras become ``duplicate``
-    diffs and ``summary.duplicate_key_list`` lists the colliding keys.
+    ``summary.duplicate`` is always 0. ``duplicate_keys_*`` count match keys
+    that appear more than once on a side (diagnostic only).
 
     ``field_rules`` drives normalize / numeric tolerance / per-field compare mode
     (template-driven; no metric-specific branches here).
@@ -177,26 +230,22 @@ def compare_rows(
         apply_port_map(r, iface_fields=iface_list, port_map=pmap) for r in before_norm
     ]
 
-    after_index: dict[tuple[str, ...], dict[str, Any]] = {}
-    after_dup = 0
-    after_dup_keys: list[tuple[str, ...]] = []
-    after_dup_rows: list[tuple[tuple[str, ...], dict[str, Any]]] = []
+    before_groups: dict[tuple[str, ...], list[tuple[dict[str, Any], dict[str, Any]]]] = (
+        defaultdict(list)
+    )
+    after_groups: dict[tuple[str, ...], list[dict[str, Any]]] = defaultdict(list)
+    for orig, mapped in zip(before_rows, before_mapped):
+        before_groups[row_key(mapped, match_keys)].append((orig, mapped))
     for r in after_norm:
-        k = row_key(r, match_keys)
-        if k in after_index:
-            after_dup += 1
-            after_dup_keys.append(k)
-            after_dup_rows.append((k, r))
-            continue  # first wins — do not overwrite
-        after_index[k] = r
+        after_groups[row_key(r, match_keys)].append(r)
 
-    before_keys: set[tuple[str, ...]] = set()
-    before_dup = 0
-    before_dup_keys: list[tuple[str, ...]] = []
     diffs: list[dict[str, Any]] = []
-    added = removed = changed = unchanged = duplicate = 0
+    added = removed = changed = unchanged = 0
     unchanged_listed = 0
     limit_n = None if unchanged_limit is None else max(0, int(unchanged_limit))
+    multi_before_keys: list[tuple[str, ...]] = []
+    multi_after_keys: list[tuple[str, ...]] = []
+    unchanged_candidates: list[tuple[dict[str, Any], dict[str, Any], dict[str, Any]]] = []
 
     def _key_obj(row: dict[str, Any]) -> dict[str, Any]:
         return {f: row.get(f, "") for f in key_fields}
@@ -209,12 +258,10 @@ def compare_rows(
             return str(netx.get("row_id") or "")
         return ""
 
-    def _emit_unchanged(orig: dict[str, Any], mapped: dict[str, Any], after_row: dict[str, Any]) -> None:
+    def _append_unchanged_diff(
+        orig: dict[str, Any], mapped: dict[str, Any], after_row: dict[str, Any]
+    ) -> None:
         nonlocal unchanged_listed
-        if not include_unchanged:
-            return
-        if limit_n is not None and unchanged_listed >= limit_n:
-            return
         unchanged_listed += 1
         before_rid = _row_id(orig)
         after_rid = _row_id(after_row)
@@ -246,96 +293,103 @@ def compare_rows(
                 }
             )
 
+    # Stable key order: before encounter order, then after-only keys
+    seen_keys: set[tuple[str, ...]] = set()
+    ordered_keys: list[tuple[str, ...]] = []
     for orig, mapped in zip(before_rows, before_mapped):
         k = row_key(mapped, match_keys)
-        if k in before_keys:
-            before_dup += 1
-            before_dup_keys.append(k)
-            duplicate += 1
-            diffs.append(
-                {
-                    "kind": "duplicate",
-                    "side": "before",
-                    "key": _key_obj(mapped),
-                    "before": orig,
-                    "after": after_index.get(k),
-                    "mapped_before": mapped,
-                    "changes": {},
-                }
-            )
-            continue  # only first before row participates in match
-        before_keys.add(k)
-        after = after_index.get(k)
-        if after is None:
-            removed += 1
-            diffs.append(
-                {
-                    "kind": "removed",
-                    "key": _key_obj(mapped),
-                    "before": orig,
-                    "after": None,
-                    "mapped_before": mapped,
-                    "changes": {},
-                }
-            )
-            continue
-        # Empty compare_fields = presence-only: keyed rows that exist on both
-        # sides are unchanged (no value checks).
-        field_changes: dict[str, dict[str, Any]] = {}
-        for f in compare_fields:
-            bv = mapped.get(f, "")
-            av = after.get(f, "")
-            rule = rules.get(f)
-            if not values_equal(bv, av, rule=rule):
-                entry: dict[str, Any] = {"before": bv, "after": av}
-                reason = explain_diff(bv, av, rule=rule)
-                if reason:
-                    entry["reason"] = reason
-                field_changes[f] = entry
-        if field_changes:
-            changed += 1
-            diffs.append(
-                {
-                    "kind": "changed",
-                    "key": _key_obj(mapped),
-                    "before": orig,
-                    "after": after,
-                    "mapped_before": mapped,
-                    "changes": field_changes,
-                }
-            )
+        if k not in seen_keys:
+            seen_keys.add(k)
+            ordered_keys.append(k)
+    for r in after_norm:
+        k = row_key(r, match_keys)
+        if k not in seen_keys:
+            seen_keys.add(k)
+            ordered_keys.append(k)
+    for k in ordered_keys:
+        b_list = before_groups.get(k) or []
+        a_list = after_groups.get(k) or []
+        if len(b_list) > 1:
+            multi_before_keys.append(k)
+        if len(a_list) > 1:
+            multi_after_keys.append(k)
+        n = max(len(b_list), len(a_list))
+        for i in range(n):
+            if i >= len(b_list):
+                after = a_list[i]
+                added += 1
+                diffs.append(
+                    {
+                        "kind": "added",
+                        "key": _key_obj(after),
+                        "before": None,
+                        "after": after,
+                        "mapped_before": None,
+                        "changes": {},
+                        "before_row_id": "",
+                        "after_row_id": _row_id(after),
+                    }
+                )
+                continue
+            if i >= len(a_list):
+                orig, mapped = b_list[i]
+                removed += 1
+                diffs.append(
+                    {
+                        "kind": "removed",
+                        "key": _key_obj(mapped),
+                        "before": orig,
+                        "after": None,
+                        "mapped_before": mapped,
+                        "changes": {},
+                        "before_row_id": _row_id(orig),
+                        "after_row_id": "",
+                    }
+                )
+                continue
+            orig, mapped = b_list[i]
+            after = a_list[i]
+            field_changes: dict[str, dict[str, Any]] = {}
+            for f in compare_fields:
+                bv = mapped.get(f, "")
+                av = after.get(f, "")
+                rule = rules.get(f)
+                if not values_equal(bv, av, rule=rule):
+                    entry: dict[str, Any] = {"before": bv, "after": av}
+                    reason = explain_diff(bv, av, rule=rule)
+                    if reason:
+                        entry["reason"] = reason
+                    field_changes[f] = entry
+            if field_changes:
+                changed += 1
+                diffs.append(
+                    {
+                        "kind": "changed",
+                        "key": _key_obj(mapped),
+                        "before": orig,
+                        "after": after,
+                        "mapped_before": mapped,
+                        "changes": field_changes,
+                        "before_row_id": _row_id(orig),
+                        "after_row_id": _row_id(after),
+                    }
+                )
+            else:
+                unchanged += 1
+                if include_unchanged:
+                    unchanged_candidates.append((orig, mapped, after))
+
+    if include_unchanged and unchanged_candidates:
+        if limit_n is None:
+            picked = unchanged_candidates
         else:
-            unchanged += 1
-            _emit_unchanged(orig, mapped, after)
-
-    for k, after in after_index.items():
-        if k in before_keys:
-            continue
-        added += 1
-        diffs.append(
-            {
-                "kind": "added",
-                "key": _key_obj(after),
-                "before": None,
-                "after": after,
-                "mapped_before": None,
-                "changes": {},
-            }
-        )
-
-    for k, after in after_dup_rows:
-        duplicate += 1
-        diffs.append(
-            {
-                "kind": "duplicate",
-                "side": "after",
-                "key": _key_obj(after),
-                "before": None,
-                "after": after,
-                "mapped_before": None,
-                "changes": {},
-            }
-        )
+            picked = stratify_take(
+                unchanged_candidates,
+                limit_n,
+                key_fn=lambda t: stratum_key(t[1]),
+            )
+        for orig, mapped, after_row in picked:
+            _append_unchanged_diff(orig, mapped, after_row)
 
     def _fmt_keys(keys: list[tuple[str, ...]]) -> list[str]:
         seen: set[str] = set()
@@ -345,7 +399,7 @@ def compare_rows(
             if s not in seen:
                 seen.add(s)
                 out.append(s)
-        return out
+        return out[:64]
 
     stats = mapping_stats(
         before_rows=before_norm,
@@ -363,16 +417,21 @@ def compare_rows(
             "removed": removed,
             "changed": changed,
             "unchanged": unchanged,
-            "duplicate": duplicate,
+            "duplicate": 0,
             "match_key_fields": match_keys,
-            "duplicate_keys_before": before_dup,
-            "duplicate_keys_after": after_dup,
-            "duplicate_key_list": _fmt_keys(before_dup_keys + after_dup_keys),
+            "duplicate_keys_before": len(multi_before_keys),
+            "duplicate_keys_after": len(multi_after_keys),
+            "duplicate_key_list": _fmt_keys(multi_before_keys + multi_after_keys),
             "unchanged_listed": unchanged_listed,
             "unchanged_truncated": bool(
                 include_unchanged and limit_n is not None and unchanged > unchanged_listed
             ),
             "unchanged_compact": bool(compact_unchanged and unchanged_listed > 0),
+            "unchanged_sample_mode": (
+                "stratified"
+                if include_unchanged and limit_n is not None and unchanged > unchanged_listed
+                else ("full" if include_unchanged else "none")
+            ),
         },
         "diffs": diffs,
         "mapping_stats": stats,

netx_api/biz_state/compare_service.py

@@ -186,6 +186,13 @@ def _compare_side(
 _DIFF_CHUNK = 2000
 _LOAD_YIELD_PER = 5000
 _SEARCH_TEXT_MAX = 4000
+# Heartbeat while bulk-inserting large fail/ok diff sets (vpnv4-scale).
+_PERSIST_PROGRESS_EVERY = 10_000
+# Above this, store fail diffs as key + row_id + changes (hydrate sides on read).
+_FAIL_COMPACT_MIN = 50_000
+# Live search (kw): load at most this many matching rows per side, return ≤ this many pairs.
+_LIVE_SEARCH_LOAD_CAP = 2_000
+_LIVE_SEARCH_RESULT_CAP = 200
 # Success-row persist policy (see resolve_unchanged_policy)
 _STORE_UNCHANGED_MODES = frozenset({"auto", "always", "never", "sample", "keys"})
 _UNCHANGED_FULL_MAX = 20_000
@@ -273,15 +280,41 @@ def _persist_sheet_diffs(
     metric_id: str,
     diffs: list[dict[str, Any]],
     seq_start: int = 0,
+    on_progress: Callable[[int, int], None] | None = None,
 ) -> int:
     """Bulk-insert diff rows already selected by the engine policy.
 
     Compact success rows carry key + before/after_row_id; JSON sides stay empty
     and are hydrated from metric tables on read.
+
+    ``on_progress(written, total)`` fires periodically so UI elapsed time moves
+    during multi-minute inserts (e.g. large vpnv4 fail sets).
     """
     buf: list[dict[str, Any]] = []
     seq = int(seq_start or 0)
     written = 0
+    total = len(diffs)
+    last_prog = 0
+    last_prog_t = time.monotonic()
+
+    def _maybe_prog(force: bool = False) -> None:
+        nonlocal last_prog, last_prog_t
+        if not on_progress:
+            return
+        now = time.monotonic()
+        if (
+            not force
+            and written - last_prog < _PERSIST_PROGRESS_EVERY
+            and now - last_prog_t < 2.0
+        ):
+            return
+        last_prog = written
+        last_prog_t = now
+        try:
+            on_progress(written, total)
+        except Exception:
+            _log.exception("persist progress callback failed run=%s metric=%s", run_id, metric_id)
+
     for d in diffs:
         kind = str(d.get("kind") or "")
         before = _strip_netx(d.get("before"))
@@ -295,10 +328,14 @@ def _persist_sheet_diffs(
             "mapped_before": mapped,
             "changes": dict(d.get("changes") or {}),
         }
-        # Compact success: search_text = kind + key only (no fat sides)
+        # Compact rows: search_text = kind + key (+ changes) only — no fat sides
         search_src = (
-            {"kind": kind, "key": payload["key"]}
-            if kind == "unchanged" and bool(d.get("compact"))
+            {
+                "kind": kind,
+                "key": payload["key"],
+                **({"changes": payload["changes"]} if payload["changes"] else {}),
+            }
+            if bool(d.get("compact"))
             else payload
         )
         buf.append(
@@ -323,8 +360,14 @@ def _persist_sheet_diffs(
         if len(buf) >= _DIFF_CHUNK:
             db.bulk_insert_mappings(BizCompareDiff, buf)
             buf.clear()
+            # Commit chunks so progress/UI can see mid-write fail rows and
+            # elapsed_ms advances (otherwise persisting_* looks frozen).
+            db.commit()
+            _maybe_prog()
     if buf:
         db.bulk_insert_mappings(BizCompareDiff, buf)
+        db.commit()
+    _maybe_prog(force=True)
     return written
 
 
@@ -514,6 +557,7 @@ def _pending_sheet_meta(sheet: dict[str, Any]) -> dict[str, Any]:
         "compare_fields": compare_fields,
         "display_fields": list(sheet.get("display_fields") or []),
         "field_rules": list(sheet.get("field_rules") or []),
+        "row_filters": list(sheet.get("row_filters") or []),
         "ignore_port_changes": sheet.get("ignore_port_changes"),
         "mode": "presence" if not compare_fields else "fields",
         "status": "pending",
@@ -2423,6 +2467,39 @@ def _execute_compare_into_run(db: Session, run_id: str) -> dict[str, Any]:
             fail_diffs = [d for d in diffs if str(d.get("kind") or "") != "unchanged"]
             ok_diffs = [d for d in diffs if str(d.get("kind") or "") == "unchanged"]
             mid = sheet_key(one)
+            # Million-row vpnv4 with many diffs: keep key/row_id/changes only.
+            if len(fail_diffs) >= _FAIL_COMPACT_MIN:
+                for d in fail_diffs:
+                    d["before"] = {}
+                    d["after"] = {}
+                    d["mapped_before"] = {}
+                    d["compact"] = True
+
+            def _on_persist(
+                written: int,
+                total_n: int,
+                *,
+                phase: str,
+                kind_key: str,
+            ) -> None:
+                _set_run_progress(
+                    db,
+                    run,
+                    phase=phase,
+                    sheet_index=idx,
+                    sheet_total=total,
+                    sheet=sheet,
+                    started_mono=started_mono,
+                    extra={
+                        kind_key: total_n,
+                        "persisted": written,
+                        "persist_total": total_n,
+                    },
+                    # Separate session so chunk commits in persist do not race
+                    # with progress JSON writes on the worker session.
+                    detach=True,
+                )
+
             _set_run_progress(
                 db,
                 run,
@@ -2431,12 +2508,22 @@ def _execute_compare_into_run(db: Session, run_id: str) -> dict[str, Any]:
                 sheet_total=total,
                 sheet=sheet,
                 started_mono=started_mono,
-                extra={"fail_rows": len(fail_diffs)},
+                extra={
+                    "fail_rows": len(fail_diffs),
+                    "persisted": 0,
+                    "persist_total": len(fail_diffs),
+                },
             )
             n_fail = _persist_sheet_diffs(
-                db, run_id=run.id, metric_id=mid, diffs=fail_diffs, seq_start=0
+                db,
+                run_id=run.id,
+                metric_id=mid,
+                diffs=fail_diffs,
+                seq_start=0,
+                on_progress=lambda w, n: _on_persist(
+                    w, n, phase="persisting_fail", kind_key="fail_rows"
+                ),
             )
-            db.commit()
             if ok_diffs:
                 _set_run_progress(
                     db,
@@ -2446,7 +2533,11 @@ def _execute_compare_into_run(db: Session, run_id: str) -> dict[str, Any]:
                     sheet_total=total,
                     sheet=sheet,
                     started_mono=started_mono,
-                    extra={"ok_rows": len(ok_diffs)},
+                    extra={
+                        "ok_rows": len(ok_diffs),
+                        "persisted": 0,
+                        "persist_total": len(ok_diffs),
+                    },
                 )
                 _persist_sheet_diffs(
                     db,
@@ -2454,6 +2545,9 @@ def _execute_compare_into_run(db: Session, run_id: str) -> dict[str, Any]:
                     metric_id=mid,
                     diffs=ok_diffs,
                     seq_start=n_fail,
+                    on_progress=lambda w, n: _on_persist(
+                        w, n, phase="persisting_ok", kind_key="ok_rows"
+                    ),
                 )
             for k in agg:
                 agg[k] += int(s.get(k) or 0)
@@ -2467,6 +2561,7 @@ def _execute_compare_into_run(db: Session, run_id: str) -> dict[str, Any]:
                 "compare_fields": one["compare_fields"],
                 "display_fields": one.get("display_fields") or [],
                 "field_rules": one.get("field_rules") or [],
+                "row_filters": one.get("row_filters") or [],
                 "ignore_port_changes": one.get("ignore_port_changes"),
                 "mode": one["mode"],
                 "status": "done",
@@ -2948,6 +3043,270 @@ def _lookup_sheet(sheets: list[dict[str, Any]], key: str) -> dict[str, Any] | No
     return None
 
 
+def _kind_allows(kind_n: str, diff_kind: str) -> bool:
+    dk = str(diff_kind or "")
+    if kind_n == "all":
+        return True
+    if kind_n == "diff":
+        return dk in ("removed", "changed")
+    return dk == kind_n
+
+
+def _kw_match_sql(key_fields: list[str], *, param: str = "kw") -> str:
+    """OR of ILIKE on key fields (and data_json::text fallback)."""
+    from .compare_sql import _FIELD_RE, _safe_field
+
+    parts: list[str] = []
+    for f in key_fields:
+        name = str(f or "").strip()
+        if not name or not _FIELD_RE.match(name):
+            continue
+        sf = _safe_field(name)
+        parts.append(f"lower(trim(both from coalesce(data_json->>'{sf}', ''))) LIKE :{param}")
+    # Broad fallback so free-text still hits non-key columns (path, next_hop, …)
+    parts.append(f"lower(data_json::text) LIKE :{param}")
+    return "(" + " OR ".join(parts) + ")" if parts else f"(lower(data_json::text) LIKE :{param})"
+
+
+def _load_metric_rows_for_search(
+    db: Session,
+    *,
+    batch_id: str,
+    metric_id: str,
+    row_filters: list[dict[str, Any]] | None,
+    key_fields: list[str],
+    kw: str,
+    cap: int = _LIVE_SEARCH_LOAD_CAP,
+) -> tuple[list[dict[str, Any]], bool]:
+    """Load rows matching sheet filters + kw. Returns (rows, truncated)."""
+    from .compare_sql import (
+        _dialect_is_postgres,
+        _filters_sql_compatible,
+        compile_row_filters_sql,
+    )
+    from ..models import BizStateMetricRow
+    from sqlalchemy import text as sql_text
+
+    bid = str(batch_id or "").strip()
+    mid = str(metric_id or "").strip()
+    needle = str(kw or "").strip()
+    if not bid or not mid or not needle:
+        return [], False
+    lim = max(1, min(int(cap), _LIVE_SEARCH_LOAD_CAP))
+    filters = [f for f in (row_filters or []) if isinstance(f, dict)]
+    like = f"%{needle.lower()}%"
+
+    if _dialect_is_postgres(db):
+        filter_sql, filter_params = ("TRUE", {})
+        if filters and _filters_sql_compatible(filters):
+            filter_sql, filter_params = compile_row_filters_sql(filters)
+        kw_sql = _kw_match_sql(key_fields)
+        # Fetch lim+1 to detect truncation
+        params = {
+            "bid": bid,
+            "mid": mid,
+            "kw": like,
+            "lim": lim + 1,
+            **filter_params,
+        }
+        rows = db.execute(
+            sql_text(
+                f"""
+                SELECT id, batch_command_id, task_id, ne_id, seq, data_json, collected_at
+                FROM biz_state_metric_row
+                WHERE batch_id = :bid
+                  AND metric_id = :mid
+                  AND ({filter_sql})
+                  AND ({kw_sql})
+                ORDER BY seq ASC, id ASC
+                LIMIT :lim
+                """
+            ),
+            params,
+        ).mappings().all()
+        truncated = len(rows) > lim
+        rows = rows[:lim]
+        out: list[dict[str, Any]] = []
+        for r in rows:
+            data = dict(r["data_json"] or {})
+            collected = r["collected_at"]
+            out.append(
+                {
+                    **data,
+                    "_netx": {
+                        "batch_id": bid,
+                        "batch_command_id": str(r["batch_command_id"] or ""),
+                        "task_id": str(r["task_id"] or ""),
+                        "ne_id": str(r["ne_id"] or ""),
+                        "collected_at": collected.isoformat() + "Z"
+                        if collected is not None
+                        else None,
+                        "row_id": str(r["id"]),
+                    },
+                }
+            )
+        return out, truncated
+
+    # Non-PG / fallback: scan with early stop (OK for tests / small sheets)
+    q = (
+        db.query(BizStateMetricRow)
+        .filter(
+            BizStateMetricRow.batch_id == bid,
+            BizStateMetricRow.metric_id == mid,
+        )
+        .order_by(BizStateMetricRow.seq.asc(), BizStateMetricRow.id.asc())
+    )
+    out = []
+    truncated = False
+    needle_l = needle.lower()
+    key_set = [str(k).strip() for k in key_fields if str(k).strip()]
+    for r in q.yield_per(500):
+        data = dict(r.data_json or {})
+        row = {
+            **data,
+            "_netx": {
+                "batch_id": bid,
+                "batch_command_id": r.batch_command_id or "",
+                "task_id": r.task_id or "",
+                "ne_id": r.ne_id or "",
+                "collected_at": r.collected_at.isoformat() + "Z"
+                if r.collected_at
+                else None,
+                "row_id": r.id,
+            },
+        }
+        if filters and not all(row_matches_filter(row, f) for f in filters):
+            db.expunge(r)
+            continue
+        hit = False
+        for kf in key_set:
+            if needle_l in str(row.get(kf) or "").lower():
+                hit = True
+                break
+        if not hit:
+            blob = json.dumps(data, ensure_ascii=False, default=str).lower()
+            hit = needle_l in blob
+        if not hit:
+            db.expunge(r)
+            continue
+        out.append(row)
+        db.expunge(r)
+        if len(out) >= lim:
+            # peek one more?
+            truncated = True
+            break
+    return out, truncated
+
+
+def _live_search_sheet_diffs(
+    db: Session,
+    run: BizCompareRun,
+    sheet: dict[str, Any],
+    *,
+    kind: str,
+    kw: str,
+    page: int,
+    page_size: int,
+) -> dict[str, Any]:
+    """Search before/after metric tables, zip-compare, filter by kind tab."""
+    mid_src = str(sheet.get("metric_id") or "").strip()
+    sid = sheet_key(sheet)
+    key_fields = list(sheet.get("key_fields") or [])
+    iface_fields = list(sheet.get("iface_fields") or [])
+    field_rules = list(sheet.get("field_rules") or [])
+    compare_fields = effective_compare_fields(
+        list(sheet.get("compare_fields") or []),
+        field_rules,
+    )
+    row_filters = list(sheet.get("row_filters") or [])
+    # Older runs may lack row_filters on sheet meta — fall back to template
+    if not row_filters and run.template_id:
+        tpl = db.get(BizCompareTemplate, run.template_id)
+        if tpl:
+            for s in template_metrics(tpl):
+                if sheet_key(s) == sid:
+                    row_filters = list(s.get("row_filters") or [])
+                    if not key_fields:
+                        key_fields = list(s.get("key_fields") or [])
+                    if not field_rules:
+                        field_rules = list(s.get("field_rules") or [])
+                        compare_fields = effective_compare_fields(
+                            list(s.get("compare_fields") or compare_fields),
+                            field_rules,
+                        )
+                    break
+
+    if not key_fields or not mid_src:
+        return {
+            "total": 0,
+            "page": page,
+            "page_size": page_size,
+            "metric_id": sid,
+            "items": [],
+            "source": "live",
+            "truncated": False,
+        }
+
+    ignore_ports = sheet.get("ignore_port_changes")
+    if ignore_ports is not None:
+        ignore_ports = bool(ignore_ports)
+    pmap = _port_map_dict(db, str(run.mapping_id or ""))
+    tpl = db.get(BizCompareTemplate, run.template_id) if run.template_id else None
+    norm_rules = template_iface_normalize(tpl)
+
+    before_rows, trunc_b = _load_metric_rows_for_search(
+        db,
+        batch_id=str(run.before_batch_id or ""),
+        metric_id=mid_src,
+        row_filters=row_filters,
+        key_fields=key_fields,
+        kw=kw,
+    )
+    after_rows, trunc_a = _load_metric_rows_for_search(
+        db,
+        batch_id=str(run.after_batch_id or ""),
+        metric_id=mid_src,
+        row_filters=row_filters,
+        key_fields=key_fields,
+        kw=kw,
+    )
+    result = compare_rows(
+        before_rows=before_rows,
+        after_rows=after_rows,
+        key_fields=key_fields,
+        iface_fields=iface_fields,
+        compare_fields=compare_fields,
+        port_map=pmap,
+        field_rules=field_rules,
+        iface_normalize_rules=norm_rules,
+        ignore_port_changes=ignore_ports,
+        include_unchanged=True,
+        unchanged_limit=None,
+        compact_unchanged=False,
+    )
+    kind_n = (kind or "diff").strip().lower()
+    filtered = [
+        d for d in list(result.get("diffs") or []) if _kind_allows(kind_n, str(d.get("kind") or ""))
+    ]
+    # Cap pairs returned to keep UI snappy
+    truncated = bool(trunc_b or trunc_a or len(filtered) > _LIVE_SEARCH_RESULT_CAP)
+    filtered = filtered[:_LIVE_SEARCH_RESULT_CAP]
+    total = len(filtered)
+    start = (page - 1) * page_size
+    page_items = filtered[start : start + page_size]
+    return {
+        "total": total,
+        "page": page,
+        "page_size": page_size,
+        "metric_id": sid,
+        "items": page_items,
+        "source": "live",
+        "truncated": truncated,
+        "live_before_matched": len(before_rows),
+        "live_after_matched": len(after_rows),
+    }
+
+
 def list_run_diffs(
     db: Session,
     run_id: str,
@@ -2973,6 +3332,18 @@ def list_run_diffs(
     sheet = _lookup_sheet(sheets, asked) if asked else (sheets[0] if sheets else None)
     mid = sheet_key(sheet) if sheet else (asked or str(r.metric_id or ""))
 
+    # Unified search: any kw → live source lookup; kind tab only filters.
+    if kw_n and sheet:
+        return _live_search_sheet_diffs(
+            db,
+            r,
+            sheet,
+            kind=kind_n,
+            kw=kw_n,
+            page=page_n,
+            page_size=size_n,
+        )
+
     if _run_has_diff_rows(db, run_id):
         q = db.query(BizCompareDiff).filter(
             BizCompareDiff.run_id == run_id,
@@ -2980,10 +3351,12 @@ def list_run_diffs(
         )
         if kind_n == "diff":
             q = q.filter(BizCompareDiff.kind.in_(("removed", "changed")))
+        elif kind_n == "removed":
+            q = q.filter(BizCompareDiff.kind == "removed")
+        elif kind_n == "changed":
+            q = q.filter(BizCompareDiff.kind == "changed")
         elif kind_n != "all":
             q = q.filter(BizCompareDiff.kind == kind_n)
-        if kw_n:
-            q = q.filter(BizCompareDiff.search_text.ilike(f"%{kw_n}%"))
         total = q.count()
         rows = (
             q.order_by(BizCompareDiff.seq.asc(), BizCompareDiff.id.asc())
@@ -2998,6 +3371,8 @@ def list_run_diffs(
             "page_size": size_n,
             "metric_id": mid,
             "items": items,
+            "source": "stored",
+            "truncated": False,
         }
 
     # Legacy: diffs embedded in summary_json / diffs_json
@@ -3007,7 +3382,7 @@ def list_run_diffs(
     inline = list((sheet or {}).get("diffs") or [])
     if not inline and mid == r.metric_id:
         inline = list(r.diffs_json or [])
-    filtered = _filter_inline_diffs(inline, kind=kind_n, kw=kw_n)
+    filtered = _filter_inline_diffs(inline, kind=kind_n, kw="")
     total = len(filtered)
     start = (page_n - 1) * size_n
     page_items = filtered[start : start + size_n]
@@ -3017,6 +3392,8 @@ def list_run_diffs(
         "page_size": size_n,
         "metric_id": mid,
         "items": page_items,
+        "source": "stored",
+        "truncated": False,
     }
 
 

netx_api/biz_state/compare_sql.py

@@ -406,6 +406,9 @@ def run_sql_sheet_compare(
 ) -> dict[str, Any]:
     """Compare one sheet via PostgreSQL TEMP tables + FULL OUTER JOIN.
 
+    Same-key multipath: rows get ``dup_rn`` by ``seq,id`` and join on
+    ``(rk, dup_rn)`` (ordered zip). Extras are added/removed, not duplicate.
+
     Returns the same ``{summary, diffs, mapping_stats}`` shape as ``compare_rows``.
     ``on_progress(side, n, *, engine=\"sql\", note=..., phase=...)``.
     """
@@ -503,9 +506,7 @@ def run_sql_sheet_compare(
             ),
             params,
         )
-        db.execute(
-            text(f"CREATE INDEX IF NOT EXISTS {tname}_rk ON {tname} (rk) WHERE dup_rn = 1")
-        )
+        db.execute(text(f"CREATE INDEX IF NOT EXISTS {tname}_rk ON {tname} (rk, dup_rn)"))
 
     before_n = int(
         db.execute(text(f"SELECT count(*) FROM {tb}")).scalar() or 0
@@ -525,14 +526,14 @@ def run_sql_sheet_compare(
       END
     """
 
-    # Aggregate primary match kinds (first-wins keys only)
+    # Ordered same-key pairing: join on (rk, dup_rn) so 5 vs 2 → 2 compared + 3 removed
     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
+            FROM {tb} b
+            FULL OUTER JOIN {ta} a
+              ON b.rk = a.rk AND b.dup_rn = a.dup_rn
             GROUP BY 1
             """
         )
@@ -543,13 +544,19 @@ def run_sql_sheet_compare(
     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
+    # Diagnostic: count of match keys with >1 row (not fail rows)
+    multi_b = int(
+        db.execute(
+            text(f"SELECT count(*) FROM (SELECT rk FROM {tb} GROUP BY rk HAVING count(*) > 1) t")
+        ).scalar()
+        or 0
     )
-    dup_a = int(
-        db.execute(text(f"SELECT count(*) FROM {ta} WHERE dup_rn > 1")).scalar() or 0
+    multi_a = int(
+        db.execute(
+            text(f"SELECT count(*) FROM (SELECT rk FROM {ta} GROUP BY rk HAVING count(*) > 1) t")
+        ).scalar()
+        or 0
     )
-    duplicate = dup_b + dup_a
 
     # Lazy import — avoid circular import with compare_service
     from .compare_service import resolve_unchanged_policy
@@ -559,7 +566,7 @@ def run_sql_sheet_compare(
     )
     diffs: list[dict[str, Any]] = []
 
-    # Fail + duplicate rows (stream into Python — should be << million)
+    # Fail rows (stream into Python — should be << million)
     fail_sql = text(
         f"""
         SELECT
@@ -569,12 +576,14 @@ def run_sql_sheet_compare(
           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
+        FROM {tb} b
+        FULL OUTER JOIN {ta} a
+          ON b.rk = a.rk AND b.dup_rn = a.dup_rn
         WHERE {kind_expr} IN ('added', 'removed', 'changed')
         """
     )
+    fail_n = 0
+    _prog("after", after_n, phase="sql_fail_fetch", note="fetch_fails")
     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
@@ -593,54 +602,77 @@ def run_sql_sheet_compare(
         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 "",
-                }
-            )
+        fail_n += 1
+        if fail_n == 1 or fail_n % 25_000 == 0:
+            _prog("fail", fail_n, phase="sql_fail_fetch", note="fetch_fails")
+    if fail_n:
+        _prog("fail", fail_n, phase="sql_fail_fetch", note="fetch_fails")
 
     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] = {}
+        # Stratified sample: round-robin by neighbor|direction|afi|vrf so one
+        # BGP command cannot consume the whole success sample.
+        stratum_expr = """
+          concat_ws('|',
+            coalesce(nullif(trim(both from coalesce(b.data->>'neighbor','')), ''), '_'),
+            coalesce(nullif(trim(both from coalesce(b.data->>'direction','')), ''), '_'),
+            coalesce(nullif(trim(both from coalesce(b.data->>'afi','')), ''), '_'),
+            coalesce(nullif(trim(both from coalesce(b.data->>'vrf','')), ''), '_')
+          )
+        """
         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}
-            """
-        )
+            u_sql = text(
+                f"""
+                WITH u AS (
+                  SELECT
+                    b.id AS before_row_id,
+                    a.id AS after_row_id,
+                    b.data AS before_data,
+                    a.data AS after_data,
+                    {stratum_expr} AS stratum
+                  FROM {tb} b
+                  INNER JOIN {ta} a
+                    ON b.rk = a.rk AND b.dup_rn = a.dup_rn
+                  WHERE NOT ({changed_pred})
+                ),
+                ranked AS (
+                  SELECT *,
+                    row_number() OVER (
+                      PARTITION BY stratum ORDER BY before_row_id ASC
+                    ) AS rn_in
+                  FROM u
+                ),
+                picked AS (
+                  SELECT *,
+                    row_number() OVER (
+                      ORDER BY rn_in ASC, stratum ASC, before_row_id ASC
+                    ) AS pick_ord
+                  FROM ranked
+                )
+                SELECT before_row_id, after_row_id, before_data, after_data
+                FROM picked
+                WHERE pick_ord <= :lim
+                """
+            )
+        else:
+            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 {tb} b
+                INNER JOIN {ta} a
+                  ON b.rk = a.rk AND b.dup_rn = a.dup_rn
+                WHERE NOT ({changed_pred})
+                """
+            )
         for row in db.execute(u_sql, params_u).mappings():
             before = dict(row["before_data"] or {})
             after = dict(row["after_data"] or {})
@@ -677,7 +709,6 @@ def run_sql_sheet_compare(
     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,
@@ -685,16 +716,21 @@ def run_sql_sheet_compare(
         "removed": removed,
         "changed": changed,
         "unchanged": unchanged,
-        "duplicate": duplicate,
+        "duplicate": 0,
         "match_key_fields": list(key_fields),
-        "duplicate_keys_before": dup_b,
-        "duplicate_keys_after": dup_a,
-        "duplicate_key_list": dup_key_list,
+        "duplicate_keys_before": multi_b,
+        "duplicate_keys_after": multi_a,
+        "duplicate_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),
+        "unchanged_sample_mode": (
+            "stratified"
+            if include_u and limit_n is not None and unchanged > unchanged_listed
+            else ("full" if include_u else "none")
+        ),
         "engine": "sql",
         "before_raw_count": raw_b,
         "after_raw_count": raw_a,

tests/test_biz_state_compare.py

@@ -5,7 +5,8 @@ from __future__ import annotations
 import copy
 import unittest
 
-from netx_api.biz_state.compare_engine import compare_rows, mapping_stats
+from netx_api.biz_state.compare_engine import compare_rows, mapping_stats, stratify_take
+from netx_api.biz_state.compare_service import _kind_allows
 from netx_api.biz_state.compare_rules import (
     apply_row_filters,
     arp_dynamic_row_filters,
@@ -198,12 +199,70 @@ class CompareEngineTests(unittest.TestCase):
         self.assertEqual(out["summary"]["unchanged_listed"], 2)
         self.assertTrue(out["summary"]["unchanged_truncated"])
         self.assertTrue(out["summary"]["unchanged_compact"])
+        self.assertEqual(out["summary"]["unchanged_sample_mode"], "stratified")
         self.assertEqual(len(out["diffs"]), 2)
         self.assertEqual(out["diffs"][0]["before"], {})
         self.assertEqual(out["diffs"][0]["after"], {})
         self.assertEqual(out["diffs"][0]["before_row_id"], "b0")
         self.assertEqual(out["diffs"][0]["after_row_id"], "a0")
 
+    def test_stratify_take_round_robin(self) -> None:
+        items = [("a", i) for i in range(5)] + [("b", i) for i in range(5)]
+        got = stratify_take(items, 4, key_fn=lambda x: x[0])
+        self.assertEqual([x[0] for x in got], ["a", "b", "a", "b"])
+
+    def test_kind_allows_tabs(self) -> None:
+        self.assertTrue(_kind_allows("all", "unchanged"))
+        self.assertTrue(_kind_allows("diff", "removed"))
+        self.assertTrue(_kind_allows("diff", "changed"))
+        self.assertFalse(_kind_allows("diff", "added"))
+        self.assertFalse(_kind_allows("diff", "unchanged"))
+        self.assertTrue(_kind_allows("unchanged", "unchanged"))
+        self.assertFalse(_kind_allows("unchanged", "removed"))
+        self.assertTrue(_kind_allows("removed", "removed"))
+        self.assertTrue(_kind_allows("changed", "changed"))
+
+    def test_unchanged_sample_stratified_across_neighbors(self) -> None:
+        """Sample must cover multiple neighbors, not only the first command's rows."""
+        before = []
+        after = []
+        for n_i, neigh in enumerate(("1.1.1.1", "2.2.2.2", "3.3.3.3")):
+            for j in range(10):
+                net = f"10.{n_i}.{j}.0/24"
+                before.append(
+                    {
+                        "neighbor": neigh,
+                        "direction": "in",
+                        "network": net,
+                        "v": "1",
+                        "_netx": {"row_id": f"b-{neigh}-{j}"},
+                    }
+                )
+                after.append(
+                    {
+                        "neighbor": neigh,
+                        "direction": "in",
+                        "network": net,
+                        "v": "1",
+                        "_netx": {"row_id": f"a-{neigh}-{j}"},
+                    }
+                )
+        out = compare_rows(
+            before_rows=before,
+            after_rows=after,
+            key_fields=["neighbor", "direction", "network"],
+            iface_fields=[],
+            compare_fields=["v"],
+            port_map={},
+            include_unchanged=True,
+            unchanged_limit=6,
+            compact_unchanged=True,
+        )
+        self.assertEqual(out["summary"]["unchanged"], 30)
+        self.assertEqual(out["summary"]["unchanged_listed"], 6)
+        keys = [d["key"]["neighbor"] for d in out["diffs"]]
+        self.assertEqual(set(keys), {"1.1.1.1", "2.2.2.2", "3.3.3.3"})
+
     def test_mac_normalize_via_field_rules(self) -> None:
         before = [{"ip": "1.1.1.1", "mac": "00:11:22:33:44:55", "iface": "gei-0/1"}]
         after = [{"ip": "1.1.1.1", "mac": "0011.2233.4455", "iface": "gei-0/1"}]
@@ -475,6 +534,13 @@ class CompareSheetDefaultsTests(unittest.TestCase):
                 for r in (route4.get("field_rules") or [])
             )
         )
+        route_vpn = next(s for s in sheets if sheet_key(s) == "bgp_route.vpnv4")
+        self.assertTrue(
+            any(
+                f.get("field") == "afi" and f.get("value") == "vpnv4"
+                for f in (route_vpn.get("row_filters") or [])
+            )
+        )
         vpnv4 = next(s for s in sheets if sheet_key(s) == "bgp_peer.vpnv4")
         self.assertEqual(
             vpnv4["row_filters"],
@@ -483,7 +549,8 @@ class CompareSheetDefaultsTests(unittest.TestCase):
         isis4 = next(s for s in sheets if sheet_key(s) == "isis_adjacency.ipv4")
         self.assertEqual(isis4["row_filters"][0]["op"], "contains")
 
-    def test_duplicate_match_keys_are_reported(self) -> None:
+    def test_ordered_same_key_pairing_2v2(self) -> None:
+        """Same key, two rows each: zip by order (not first-wins + duplicate)."""
         before = [
             {"local_if": "a", "remote_sys": "X", "remote_if": "1", "remote_ip": "1"},
             {"local_if": "a", "remote_sys": "X", "remote_if": "1", "remote_ip": "9"},
@@ -500,15 +567,47 @@ class CompareSheetDefaultsTests(unittest.TestCase):
             compare_fields=["remote_ip"],
             port_map={"a": "a"},
         )
+        self.assertEqual(out["summary"]["before_count"], 2)
+        self.assertEqual(out["summary"]["after_count"], 2)
+        self.assertEqual(out["summary"]["duplicate"], 0)
         self.assertEqual(out["summary"]["duplicate_keys_before"], 1)
         self.assertEqual(out["summary"]["duplicate_keys_after"], 1)
-        self.assertEqual(out["summary"]["duplicate"], 2)
         self.assertIn("a|X|1", out["summary"]["duplicate_key_list"])
         kinds = [d["kind"] for d in out["diffs"]]
-        self.assertEqual(kinds.count("duplicate"), 2)
-        # First before wins → matches first after → unchanged (same remote_ip)
+        self.assertNotIn("duplicate", kinds)
+        # Pair 0: 1==1 unchanged; pair 1: 9!=2 changed
         self.assertEqual(out["summary"]["unchanged"], 1)
+        self.assertEqual(out["summary"]["changed"], 1)
+        self.assertEqual(out["summary"]["added"], 0)
+        self.assertEqual(out["summary"]["removed"], 0)
+
+    def test_ordered_multipath_5_vs_2(self) -> None:
+        """5 before / 2 after same key → 2 compared + 3 removed failures."""
+        key = {"local_if": "a", "remote_sys": "X", "remote_if": "1"}
+        before = [{**key, "remote_ip": str(i)} for i in range(5)]
+        after = [{**key, "remote_ip": "0"}, {**key, "remote_ip": "1"}]
+        out = compare_rows(
+            before_rows=before,
+            after_rows=after,
+            key_fields=["local_if", "remote_sys", "remote_if"],
+            iface_fields=["local_if"],
+            compare_fields=["remote_ip"],
+            port_map={"a": "a"},
+        )
+        self.assertEqual(out["summary"]["before_count"], 5)
+        self.assertEqual(out["summary"]["after_count"], 2)
+        self.assertEqual(out["summary"]["removed"], 3)
+        self.assertEqual(out["summary"]["added"], 0)
+        self.assertEqual(
+            out["summary"]["unchanged"] + out["summary"]["changed"],
+            2,
+        )
+        self.assertEqual(out["summary"]["unchanged"], 2)
         self.assertEqual(out["summary"]["changed"], 0)
+        self.assertEqual(out["summary"]["duplicate"], 0)
+        kinds = [d["kind"] for d in out["diffs"]]
+        self.assertEqual(kinds.count("removed"), 3)
+        self.assertNotIn("duplicate", kinds)
 
     def test_ignore_port_changes_false_keeps_iface(self) -> None:
         before = [

web/src/i18n/en.ts

@@ -502,20 +502,23 @@ const en = {
     runProgress: "{{phase}} · sheet {{i}}/{{n}} · {{sheet}} · {{s}}s elapsed",
     runRowsLoaded: "Loaded {{side}} {{n}} rows",
     runRowsSqlCount: "DB count {{side}} {{n}} rows",
+    runPersisting: "Wrote {{done}}/{{total}} 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.",
-    unchangedSampleHint: "{{total}} success rows total; browsing a sample of {{listed}} (hydrated from source tables for cutover spot-check). Pass rate uses all {{total}}.",
+    unchangedSampleHint: "{{total}} success rows total; browsing a stratified sample of {{listed}}. Search any route for live source lookup. Pass rate uses all {{total}}.",
+    liveSearchHint: "Search: live verdict from source batches (tabs only filter kind). Works for fail, success, and added.",
+    liveSearchTruncatedHint: "Search results truncated — narrow the query (full prefix / neighbor).",
     storeUnchanged: "Store success rows",
     storeUnchangedAuto: "Auto (full if small / sample 5k + hydrate if large)",
     storeUnchangedSample: "Sample only (up to 5k keys + hydrate)",
     storeUnchangedKeys: "All success keys (full browse; large write still slow)",
     storeUnchangedAlways: "Always full JSON (slow on huge sheets)",
     storeUnchangedNever: "Never (counts only)",
-    storeUnchangedHint: "Cutover focuses on fails and pass rate; success defaults to a sample. Pass rate always uses the full success count.",
+    storeUnchangedHint: "Cutover focuses on fails and pass rate; success defaults to a stratified sample. Type a filter to look up any route from source tables. Pass rate uses the full success count.",
     saveJob: "Save config",
     jobSaved: "Job config saved",
     jobDeleted: "Compare job deleted",
@@ -542,7 +545,7 @@ const en = {
     pickRun: "Select run…",
     runCount: "{{n}} runs",
     resultEmpty: "No matching diff rows",
-    resultFilterPh: "Filter key / values…",
+    resultFilterPh: "Search prefix / neighbor / RD… (live source lookup)",
     kindAll: "All",
     kindDiff: "Fail",
     kindAdded: "Added",
@@ -574,7 +577,7 @@ const en = {
     filterFailField: "Failed compare field",
     filterFailFieldAll: "Any failed field",
     resultEmpty: "No matching rows",
-    resultFilterPh: "Filter identity / values…",
+    resultFilterPh: "Search prefix / neighbor / RD… (live source lookup)",
     diffCount: "{{n}} failed",
     passRate: "Pass rate",
     passOk: "All passed",

web/src/i18n/zh.ts

@@ -501,20 +501,23 @@ const zh = {
     runProgress: "{{phase}} · 表 {{i}}/{{n}} · {{sheet}} · 已用 {{s}}s",
     runRowsLoaded: "已加载 {{side}} {{n}} 行",
     runRowsSqlCount: "库内统计 {{side}} {{n}} 行",
+    runPersisting: "已写入 {{done}}/{{total}} 行",
     runEngineSql: "引擎 SQL",
     runEnginePython: "引擎 Python",
     runEngineNote: "原因 {{note}}",
     sheetPending: "等待中",
     ranWithDuration: "比对完成（耗时 {{s}} 秒）",
     unchangedNotStored: "成功行仅统计数量未落库。可在任务配置将「成功行保存」改为抽样后重新比对（抽查用）。",
-    unchangedSampleHint: "成功共 {{total}} 条，明细抽样 {{listed}} 条（从原表补全显示，割接抽查用）。通过率按全部 {{total}} 计。",
+    unchangedSampleHint: "成功共 {{total}} 条，明细抽样 {{listed}} 条（分层抽查）。要查任意路由请输入筛选条件——将按原表即时判定。通过率按全部 {{total}} 计。",
+    liveSearchHint: "搜索：原表即时判定（页签只过滤种类）。失败/成功/新增均可查到。",
+    liveSearchTruncatedHint: "搜索结果已截断，请收窄条件（如完整前缀 / neighbor）。",
     storeUnchanged: "成功行保存",
     storeUnchangedAuto: "自动（小表全量 / 大表抽样 5000+原表补全）",
     storeUnchangedSample: "仅抽样（最多 5000，身份键+原表补全）",
     storeUnchangedKeys: "全量身份键（可翻完全部成功，大表写入仍较久）",
     storeUnchangedAlways: "始终全量 JSON（大表很慢，慎用）",
     storeUnchangedNever: "不保存（只计数量）",
-    storeUnchangedHint: "割接优先看失败与通过率；成功默认抽样抽查。通过率始终按全部成功计数。",
+    storeUnchangedHint: "割接优先看失败与通过率；成功默认分层抽样。输入筛选可回查原表（不依赖抽样）。通过率按全部成功计数。",
     saveJob: "保存配置",
     jobSaved: "任务配置已保存",
     jobDeleted: "比对任务已删除",
@@ -541,7 +544,7 @@ const zh = {
     pickRun: "选择比对记录…",
     runCount: "{{n}} 次",
     resultEmpty: "无匹配失败/结果行",
-    resultFilterPh: "筛选身份 / 字段值…",
+    resultFilterPh: "搜索前缀 / neighbor / RD…（回查原表）",
     kindAll: "全部",
     kindDiff: "失败",
     kindAdded: "新增",

web/src/pages/network/BizComparePage.tsx

@@ -38,6 +38,7 @@ import { jobChipColor, NmStatusChip } from "./nmChips";
 type PageTab = "templates" | "jobs";
 type JobDetailTab = "config" | "runs" | "result";
 type KindFilter = "diff" | "all" | "added" | "removed" | "changed" | "unchanged";
+type DiffListSource = "stored" | "live" | "";
 type CreateJobStep = 0 | 1 | 2 | 3;
 const CREATE_JOB_STEPS = 4;
 
@@ -847,6 +848,8 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage
   const [resultTotal, setResultTotal] = useState(0);
   const [pagedDiffs, setPagedDiffs] = useState<DiffRow[]>([]);
   const [diffsLoading, setDiffsLoading] = useState(false);
+  const [diffsSource, setDiffsSource] = useState<DiffListSource>("");
+  const [diffsTruncated, setDiffsTruncated] = useState(false);
   const boardRef = useRef<HTMLDivElement | null>(null);
   const tableScrollRef = useRef<HTMLDivElement | null>(null);
   const tableScrollPosRef = useRef({ top: 0, left: 0 });
@@ -1069,6 +1072,8 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage
     if (!runId || !mid || jobDetailTab !== "result" || sheetStillRunning) {
       setPagedDiffs([]);
       setResultTotal(0);
+      setDiffsSource("");
+      setDiffsTruncated(false);
       return;
     }
     let cancelled = false;
@@ -1086,6 +1091,10 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage
         if (cancelled) return;
         setPagedDiffs((res.items || []) as DiffRow[]);
         setResultTotal(Number(res.total || 0));
+        setDiffsSource(
+          String((res as any).source || "").toLowerCase() === "live" ? "live" : "stored",
+        );
+        setDiffsTruncated(Boolean((res as any).truncated));
         const pages = Math.max(
           1,
           Math.ceil(Number(res.total || 0) / Number(res.page_size || resultPageSize)),
@@ -1257,8 +1266,15 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage
     String(activeSheetCard?.status || ""),
   );
   const activeAdded = Number(activeSheetCard?.added || 0);
+  const activeRemoved = Number(activeSheetCard?.removed || 0);
+  const activeChanged = Number(activeSheetCard?.changed || 0);
   const showFailCol =
-    kindFilter === "diff" || kindFilter === "all" || kindFilter === "added";
+    kindFilter === "diff" ||
+    kindFilter === "all" ||
+    kindFilter === "added" ||
+    kindFilter === "removed" ||
+    kindFilter === "changed";
+  const isLiveSearch = diffsSource === "live" && Boolean(debouncedResultKw.trim());
   const resultEmptyColSpan =
     1 +
     (showFailCol ? 1 : 0) +
@@ -1895,6 +1911,10 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage
     rows_loaded?: number;
     engine?: string;
     engine_note?: string;
+    persisted?: number;
+    persist_total?: number;
+    fail_rows?: number;
+    ok_rows?: number;
   };
   const runEngine = String(runProgress.engine || "").toLowerCase();
 
@@ -3292,6 +3312,15 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage
                           })}
                     </span>
                   ) : null}
+                  {String(runProgress.phase || "").startsWith("persisting") &&
+                  Number(runProgress.persist_total || 0) > 0 ? (
+                    <span className="muted">
+                      {t("bizCompare.runPersisting", {
+                        done: String(runProgress.persisted || 0),
+                        total: String(runProgress.persist_total || 0),
+                      })}
+                    </span>
+                  ) : null}
                   {runDetail.message ? (
                     <span className="muted bs-cmp-progress__msg">{String(runDetail.message)}</span>
                   ) : null}
@@ -3491,6 +3520,8 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage
                         {(
                           [
                             ["diff", activeFail, "diff"],
+                            ["removed", activeRemoved, "removed"],
+                            ["changed", activeChanged, "changed"],
                             ["added", activeAdded, "added"],
                             ["unchanged", activeSuccess, "unchanged"],
                             ["all", null, "all"],
@@ -3510,7 +3541,11 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage
                                 ? t("bizCompare.kindAll")
                                 : id === "added"
                                   ? t("bizCompare.kindAddedShort")
-                                  : t("bizCompare.kindSuccess")}
+                                  : id === "removed"
+                                    ? t("bizCompare.kindRemovedShort")
+                                    : id === "changed"
+                                      ? t("bizCompare.kindChangedShort")
+                                      : t("bizCompare.kindSuccess")}
                             {n !== null ? (
                               <>
                                 {" "}
@@ -3530,11 +3565,31 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage
                           {diffsLoading ? "…" : `${pagedDiffs.length}/${resultTotal}`}
                         </span>
                       </div>
-                      {kindFilter === "unchanged" &&
+                      {isLiveSearch ? (
+                        <p className="muted bm-hint" style={{ margin: "4px 0 0" }}>
+                          {diffsTruncated
+                            ? t("bizCompare.liveSearchTruncatedHint")
+                            : t("bizCompare.liveSearchHint")}
+                        </p>
+                      ) : null}
+                      {!isLiveSearch &&
+                      kindFilter === "unchanged" &&
                       (runDetail?.summary?.unchanged_truncated ||
-                        (Number(runDetail?.summary?.unchanged || 0) >
-                          Number(runDetail?.summary?.unchanged_listed || 0) &&
-                          Number(runDetail?.summary?.unchanged_listed || 0) > 0)) ? (
+                        (Number(
+                          (activeRunSheet?.summary as any)?.unchanged ||
+                            runDetail?.summary?.unchanged ||
+                            0,
+                        ) >
+                          Number(
+                            (activeRunSheet?.summary as any)?.unchanged_listed ??
+                              runDetail?.summary?.unchanged_listed ??
+                              0,
+                          ) &&
+                          Number(
+                            (activeRunSheet?.summary as any)?.unchanged_listed ??
+                              runDetail?.summary?.unchanged_listed ??
+                              0,
+                          ) > 0)) ? (
                         <p className="muted bm-hint" style={{ margin: "4px 0 0" }}>
                           {t("bizCompare.unchangedSampleHint", {
                             listed: String(

web/src/services/api.ts

@@ -2185,6 +2185,10 @@ export const bizCompareListRunDiffs = (params: {
     page_size: number;
     metric_id: string;
     items: Record<string, unknown>[];
+    source?: string;
+    truncated?: boolean;
+    live_before_matched?: number;
+    live_after_matched?: number;
   }>(`/v1/biz-state/compare/runs/${encodeURIComponent(params.runId)}/diffs?${p.toString()}`);
 };