fix: enforce business continuity and live cutover monitoring

netx_api/biz_migration/auto_monitor.py

@@ -0,0 +1,104 @@
+"""Coalesced collect-completion events for live cutover evaluation.
+
+One worker per process bounds concurrency. Project row locks in run_evaluate
+serialize persistence across processes. No device collection runs here.
+"""
+from __future__ import annotations
+
+import hashlib
+import json
+import logging
+import threading
+
+from sqlalchemy.orm import Session
+
+from ..models import BizMigrationBatch, BizMigrationProject, BizMigrationRun, BizMonitorTemplate, BizStateBatch
+from . import service
+
+_log = logging.getLogger(__name__)
+_lock = threading.Lock()
+_pending: set[str] = set()
+_worker_running = False
+
+
+def try_auto_monitor_for_task(db: Session, task_id: str) -> int:
+    count = 0
+    # Bindings are JSON for multiple sampling intervals; match in Python after
+    # limiting to projects with an active operation batch.
+    projects = db.query(BizMigrationProject).filter(
+        BizMigrationProject.status != "done",
+        BizMigrationProject.id.in_(db.query(BizMigrationBatch.project_id).filter(
+            BizMigrationBatch.status == "active")),
+    ).all()
+    for candidate in projects:
+        tids = service._all_hf_task_ids(candidate, "old") + service._all_hf_task_ids(candidate, "new")
+        if task_id not in tids or not candidate.old_baseline_batch_id or not candidate.new_baseline_batch_id:
+            continue
+        # Lock before deduplication so two worker processes cannot save the same
+        # snapshot. run_evaluate holds the same lock through the transaction.
+        proj = db.query(BizMigrationProject).filter(
+            BizMigrationProject.id == candidate.id).with_for_update().populate_existing().first()
+        batch = db.query(BizMigrationBatch).filter(
+            BizMigrationBatch.project_id == proj.id, BizMigrationBatch.status == "active"
+        ).order_by(BizMigrationBatch.started_at.desc()).first()
+        if not batch or proj.status == "done":
+            db.rollback()
+            continue
+        mt = db.get(BizMonitorTemplate, proj.monitor_template_id) if proj.monitor_template_id else None
+        if not mt:
+            db.rollback()
+            _log.error("cutover monitor template missing project=%s", proj.id)
+            continue
+        sheets, overrides, defaults, norms = service.resolve_evaluate_sheets(db, mt)
+        samples = []
+        for tid in tids:
+            sample = db.query(BizStateBatch).filter(
+                BizStateBatch.task_id == tid, BizStateBatch.status.notin_(("running", "queued"))
+            ).order_by(BizStateBatch.started_at.desc(), BizStateBatch.id.desc()).first()
+            samples.append([tid, sample.id if sample else "", sample.status if sample else ""])
+        payload = [samples, proj.old_baseline_batch_id, proj.new_baseline_batch_id,
+                   batch.expect_set_json, service._port_map_dict(db, proj.mapping_id),
+                   sheets, overrides, defaults, norms, service.resolve_collect_metric_ids(db, proj)]
+        fingerprint = hashlib.sha256(json.dumps(payload, sort_keys=True, ensure_ascii=False).encode()).hexdigest()
+        last = db.query(BizMigrationRun).filter(BizMigrationRun.batch_id == batch.id).order_by(
+            BizMigrationRun.created_at.desc()).first()
+        if last and (last.summary_json or {}).get("auto_fingerprint") == fingerprint:
+            db.rollback()
+            continue
+        try:
+            result = service.run_evaluate(db, batch_id=batch.id, purpose="auto", _commit=False)
+            run = db.get(BizMigrationRun, result["id"])
+            run.summary_json = {**run.summary_json, "auto_fingerprint": fingerprint}
+            db.commit()
+            count += 1
+        except Exception:
+            db.rollback()
+            _log.exception("cutover monitor evaluate failed project=%s", proj.id)
+    return count
+
+
+def schedule_auto_monitor_for_task(task_id: str) -> None:
+    global _worker_running
+    with _lock:
+        _pending.add(str(task_id))
+        if _worker_running:
+            return
+        _worker_running = True
+
+    def work() -> None:
+        global _worker_running
+        from ..db import SessionLocal
+
+        while True:
+            with _lock:
+                if not _pending:
+                    _worker_running = False
+                    return
+                tid = _pending.pop()
+            try:
+                with SessionLocal() as db:
+                    try_auto_monitor_for_task(db, tid)
+            except Exception:
+                _log.exception("cutover monitor worker failed task=%s", tid)
+
+    threading.Thread(target=work, name="biz-cutover-monitor", daemon=True).start()

netx_api/biz_migration/evaluate.py

@@ -5,6 +5,9 @@ from __future__ import annotations
 from typing import Any
 
 from ..biz_state.compare_engine import apply_port_map, compare_rows
+from ..biz_state.compare_rules import (
+    effective_compare_fields, field_rule_map, scalar_text, values_equal,
+)
 
 PORT_METRIC_ID = "interface_brief"
 PORT_STATUS_FIELDS = ("admin", "phy", "prot")
@@ -22,7 +25,7 @@ def normalize_expect_key(raw: str) -> str:
 
     Accepts legacy ``a|b`` multi-field keys and already-normalized ``\\x1f`` keys.
     """
-    s = str(raw or "").strip()
+    s = scalar_text(raw).strip()
     if not s:
         return ""
     if KEY_SEP in s:
@@ -65,13 +68,13 @@ def classify_status(row: dict[str, Any] | None, sheet_override: dict[str, Any] |
         return "none"
     down_vals = {str(x).lower() for x in (ov.get("down_values") or ["down"])}
     up_vals = {str(x).lower() for x in (ov.get("up_values") or ["up"])}
-    vals = [str(row.get(f) or "").strip().lower() for f in fields]
+    vals = [scalar_text(row.get(f)).strip().lower() for f in fields]
     vals = [v for v in vals if v]
     if not vals:
         return "none"
     if any(v in down_vals for v in vals):
         return "down"
-    if vals and all(v in up_vals for v in vals):
+    if len(vals) == len(fields) and all(v in up_vals for v in vals):
         return "up"
     return "other"
 
@@ -219,7 +222,7 @@ def _field_tokens_from_row(row: dict[str, Any] | None, fields: set[str]) -> set[
         return set()
     out: set[str] = set()
     for f in fields:
-        v = str(row.get(f) or "").strip()
+        v = scalar_text(row.get(f)).strip()
         if v:
             out.add(field_token(f, v))
     return out
@@ -245,8 +248,8 @@ def _side_tokens(
 
 def _norm_list_values(raw: Any) -> list[str]:
     if isinstance(raw, list):
-        return [str(x).strip().lower() for x in raw if str(x).strip()]
-    s = str(raw or "").strip()
+        return [scalar_text(x).strip().lower() for x in raw if scalar_text(x).strip()]
+    s = scalar_text(raw).strip()
     if not s:
         return []
     if "," in s:
@@ -266,18 +269,18 @@ def _eval_field_op(
     f = str(field or "").strip()
     if not f:
         return False
-    raw = "" if not row else str(row.get(f) or "").strip()
+    raw = "" if not row else scalar_text(row.get(f)).strip()
     val = raw.lower()
     o = str(op or "eq").strip().lower()
     if o in ("changed", "diff"):
-        base = "" if not base_row else str(base_row.get(f) or "").strip().lower()
+        base = "" if not base_row else scalar_text(base_row.get(f)).strip().lower()
         if not row:
             return False
         return base != val
     if o in ("unchanged", "same"):
         if not row:
             return False
-        base = "" if not base_row else str(base_row.get(f) or "").strip().lower()
+        base = "" if not base_row else scalar_text(base_row.get(f)).strip().lower()
         return base == val
     if o == "empty":
         return not val
@@ -568,7 +571,7 @@ def _match_field_token_rules(
         if not field or not allowed:
             continue
         row = old_row if side == "old" else new_row if side == "new" else None
-        val = str((row or {}).get(field) or "").strip().lower()
+        val = scalar_text((row or {}).get(field)).strip().lower()
         if val not in allowed:
             return False
     return True
@@ -858,6 +861,14 @@ def _port_identity(key_fields: list[str], iface_fields: list[str]) -> bool:
     return all(str(f) in iface_set for f in key_fields)
 
 
+def _iface_selected(value: str, ports: set[str]) -> bool:
+    """Set lookup plus parent interface prefixes; independent of scope size."""
+    if value in ports:
+        return True
+    parts = value.split(".")
+    return any(".".join(parts[:i]) in ports for i in range(len(parts) - 1, 0, -1))
+
+
 def _unmapped_same_iface_anomaly(
     *,
     old_kind: str,
@@ -892,6 +903,8 @@ def _remap_key_str(
         return key_str
     if not port_map:
         return key_str
+    if key_fields and not set(key_fields).intersection(iface_fields):
+        return key_str
     if (not key_fields or len(key_fields) == 1) and (
         key_str in port_map
         or any(
@@ -928,6 +941,8 @@ def _reverse_remap_key_str(
         return key_str
     if not rev_map:
         return key_str
+    if key_fields and not set(key_fields).intersection(iface_fields):
+        return key_str
     if (not key_fields or len(key_fields) == 1) and (
         key_str in rev_map
         or any(
@@ -1009,7 +1024,7 @@ def row_match_key(
     data = strip_netx(row)
     if iface_fields and rules:
         data = apply_iface_normalize(dict(data), iface_fields=iface_fields, rules=rules)
-    return KEY_SEP.join(str(data.get(k) or "").strip() for k in key_fields)
+    return KEY_SEP.join(scalar_text(data.get(k)).strip() for k in key_fields)
 
 
 def index_raw_by_match_key(
@@ -1033,7 +1048,7 @@ def display_key_from_raw(raw: dict[str, Any] | None, key_fields: list[str]) -> s
     if not raw or not key_fields:
         return ""
     data = strip_netx(raw)
-    return "|".join(str(data.get(k) or "").strip() for k in key_fields)
+    return "|".join(scalar_text(data.get(k)).strip() for k in key_fields)
 
 
 def iface_lineage(
@@ -1053,7 +1068,7 @@ def iface_lineage(
     pmap = port_map or {}
     out: list[dict[str, Any]] = []
     for f in iface_fields:
-        raw_v = str(data.get(f) or "").strip()
+        raw_v = scalar_text(data.get(f)).strip()
         norm_v = normalize_iface_name(raw_v, rules) if rules else raw_v
         mapped = False
         map_to = ""
@@ -1092,6 +1107,8 @@ def evaluate_metric_dual(
     out_of_expect: str = "strict",
     sheet_id: str = "",
     iface_normalize_rules: list[dict[str, str]] | None = None,
+    previous_expect: dict[str, set[str]] | None = None,
+    allow_mapping_scope: bool = True,
 ) -> dict[str, Any]:
     """Run old vs old-baseline, new vs new-baseline (or mapped old baseline), dual merge.
 
@@ -1108,6 +1125,9 @@ def evaluate_metric_dual(
     expect_keys = expect_keys_for_metric(
         expect, metric_id=metric_id, iface_fields=iface_fields, sheet_id=sid
     )
+    previous_keys = expect_keys_for_metric(
+        previous_expect or {}, metric_id=metric_id, iface_fields=iface_fields, sheet_id=sid
+    )
     ov = sheet_override or {}
     if ov.get("skip_dual"):
         return {
@@ -1178,6 +1198,7 @@ def evaluate_metric_dual(
         port_map=None,
         field_rules=field_rules,
         iface_normalize_rules=None,  # already applied
+        include_unchanged=True,
     )
     old_idx = build_diff_index_from_compare(old_cmp)
 
@@ -1186,18 +1207,8 @@ def evaluate_metric_dual(
     if new_baseline_rows is not None:
         new_before = list(new_baseline_rows)
         new_baseline_mode = "provided"
-        # Explicit empty batch is as unusable as "no baseline" for presence semantics.
-        if not new_before and (old_baseline_rows or new_current_rows):
-            if port_map and iface_fields:
-                new_before = [
-                    apply_port_map(r, iface_fields=iface_fields, port_map=port_map)
-                    for r in old_baseline_rows
-                ]
-                new_baseline_mode = "port_mapped"
-                new_baseline_missing = False
-            else:
-                new_baseline_missing = True
-                new_baseline_mode = "empty"
+        # A successfully collected empty baseline is valid on a replacement NE.
+        # Collection completeness is checked by the service, not inferred from rows.
     elif port_map and iface_fields:
         new_before = [
             apply_port_map(r, iface_fields=iface_fields, port_map=port_map)
@@ -1218,12 +1229,22 @@ def evaluate_metric_dual(
         compare_fields=compare_fields,
         port_map=None,
         field_rules=field_rules,
+        include_unchanged=True,
     )
     new_idx = build_diff_index_from_compare(new_cmp)
 
     rev_map = {v: k for k, v in port_map.items()}
     map_scoped = _map_defines_iface_expect(metric_id, iface_fields, port_map)
-    if map_scoped:
+    # Mapping describes identity; an explicit batch selection defines scope.
+    # Port selection also covers business rows attached to those interfaces.
+    for scope, keys in ((expect, expect_keys), (previous_expect or {}, previous_keys)):
+        ports = scope.get("_ports") or set()
+        if ports and not keys and iface_fields:
+            for ks, diff in old_idx.items():
+                row = diff.get("before") or diff.get("after") or {}
+                if any(_iface_selected(scalar_text(row.get(f)), ports) for f in iface_fields):
+                    keys.add(ks)
+    if allow_mapping_scope and map_scoped and not any(expect.values()):
         scoped: set[str] = set()
         for ks in old_idx:
             if _key_in_port_map(
@@ -1253,6 +1274,8 @@ def evaluate_metric_dual(
 
     for ok in sorted(expect_keys):
         _add(ok)
+    for ok in sorted(previous_keys):
+        _add(ok)
     for ks in sorted(old_idx):
         _add(ks)
     for ks in sorted(new_idx):
@@ -1270,6 +1293,10 @@ def evaluate_metric_dual(
     progress_total = 0
     anomaly = 0
     anomaly_in_expect = 0
+    rules_by_field = field_rule_map(field_rules)
+    integrity_fields = effective_compare_fields(compare_fields, field_rules)
+    # Status transitions are interpreted by the monitor's correction rules.
+    integrity_fields = [f for f in integrity_fields if f not in (ov.get("status_fields") or [])]
 
     for old_ks in canon_keys:
         new_ks = _remap_key_str(
@@ -1279,11 +1306,12 @@ def evaluate_metric_dual(
             port_map=port_map,
         )
         in_exp = old_ks in expect_keys
+        previous = old_ks in previous_keys and not in_exp
         if in_exp:
             progress_total += 1
 
         od = old_idx.get(old_ks)
-        nd = new_idx.get(new_ks) or new_idx.get(old_ks)
+        nd = new_idx.get(new_ks)
 
         old_kind = str((od or {}).get("kind") or "")
         new_kind = str((nd or {}).get("kind") or "")
@@ -1301,11 +1329,7 @@ def evaluate_metric_dual(
         old_st = classify_status(old_cur, ov)
         new_st = classify_status(new_cur, ov)
 
-        # Mapped iface rows are the expect set. Unmapped rows stay out of the
-        # migration compare, except same-name ports where old went down and the
-        # new device's same port is up — that is anomaly, not a successful cutover.
-        if map_scoped and not in_exp:
-            if not (
+        if map_scoped and not (in_exp or previous) and (
                 _port_identity(key_fields, iface_fields)
                 and _unmapped_same_iface_anomaly(
                     old_kind=old_kind,
@@ -1314,26 +1338,62 @@ def evaluate_metric_dual(
                     new_status=new_st,
                 )
             ):
-                continue
             verdict, color, rule_hit = "anomaly", "red", "unmapped_same_iface"
         else:
             verdict, color, rule_hit = dual_verdict_ex(
                 old_kind=old_kind or "",
                 new_kind=new_kind or "",
-                in_expect=in_exp,
+                in_expect=in_exp or previous,
                 window_active=window_active,
                 acceptance=acceptance,
                 old_status=old_st,
                 new_status=new_st,
                 success_patterns=success_patterns or None,
                 anomaly_patterns=anomaly_patterns or None,
-                out_of_expect=out_of_expect,
+                # Business drift outside this batch must remain visible.
+                out_of_expect="strict",
                 old_row=old_cur or None,
                 new_row=new_cur or None,
                 old_base=old_base or None,
                 new_base=new_base or None,
                 sheet_override=ov,
             )
+        integrity_changes: list[dict[str, Any]] = []
+        if in_exp or previous:
+            old_departed = old_kind == "removed" or old_st == "down"
+            new_healthy = bool(new_cur) and (
+                new_st == "up" if ov.get("status_fields") else True
+            )
+            if old_departed and not new_healthy:
+                verdict, color, rule_hit = "lost", "red", "service_not_received"
+            elif verdict == "migrated":
+                if not new_healthy:
+                    verdict, color, rule_hit = "anomaly", "red", "target_unhealthy"
+                elif not old_base:
+                    verdict, color, rule_hit = "anomaly", "red", "old_service_not_in_baseline"
+                else:
+                    reference = apply_port_map(old_base, iface_fields=iface_fields, port_map=port_map)
+                    integrity_changes = [
+                        {"field": f, "before": reference.get(f), "after": new_cur.get(f)}
+                        for f in integrity_fields
+                        if not values_equal(reference.get(f), new_cur.get(f), rule=rules_by_field.get(f))
+                    ]
+                    if integrity_changes:
+                        verdict, color, rule_hit = "anomaly", "red", "service_content_changed"
+                    elif not old_departed and not success_patterns:
+                        verdict, color, rule_hit = (
+                            ("unfinished", "red", "old_service_still_present") if acceptance
+                            else ("migrating", "yellow", "old_service_still_present")
+                        )
+        if previous:
+            if verdict == "migrated":
+                verdict, rule_hit = "migrated_previous", "previous_batch_received"
+            elif color != "red":
+                verdict, color, rule_hit = "unfinished_previous", "red", "previous_batch_not_received"
+        # Retain all drift outside scope, but avoid filling the board with stable
+        # unmapped inventory when a port map defines the legacy migration scope.
+        if map_scoped and not in_exp and verdict == "not_involved":
+            continue
         if in_exp and verdict == "migrated" and color == "green":
             progress_ok += 1
         if color == "red":
@@ -1411,7 +1471,9 @@ def evaluate_metric_dual(
                 "verdict": verdict,
                 "color": color,
                 "rule_hit": rule_hit,
+                "integrity_changes": integrity_changes,
                 "in_expect": in_exp,
+                "previous_batch": previous,
                 "old_kind": old_kind,
                 "new_kind": new_kind,
                 "old": old_disp,

netx_api/biz_migration/monitor_templates.py

@@ -8,7 +8,8 @@ from uuid import uuid4
 from fastapi import HTTPException
 from sqlalchemy.orm import Session
 
-from ..models import BizCompareTemplate, BizMonitorTemplate
+from ..models import BizCompareTemplate, BizMigrationProject, BizMonitorTemplate
+from .validation import validate_monitor_rules
 from ..timeutil import utcnow_naive
 from ..biz_state import compare_service as cmp_svc
 from ..biz_state.iface_normalize import default_zte_iface_normalize_rules
@@ -456,6 +457,7 @@ def create_monitor_template(db: Session, body: dict[str, Any]) -> dict[str, Any]
     overrides = body.get("sheet_overrides")
     if not isinstance(overrides, list):
         overrides = []
+    validate_monitor_rules(defaults, overrides)
     row = BizMonitorTemplate(
         id=uuid4().hex,
         name=name[:256],
@@ -479,6 +481,10 @@ def update_monitor_template(db: Session, template_id: str, body: dict[str, Any])
     row = db.get(BizMonitorTemplate, template_id)
     if not row:
         raise HTTPException(status_code=404, detail="monitor_template_not_found")
+    validate_monitor_rules(body.get("defaults", row.defaults_json or {}),
+                           body.get("sheet_overrides", row.sheet_overrides_json or []))
+    if body.get("compare_template_id") and not db.get(BizCompareTemplate, body["compare_template_id"]):
+        raise HTTPException(status_code=404, detail="compare_template_not_found")
     if "name" in body and body["name"] is not None:
         name = str(body["name"] or "").strip()
         if not name:
@@ -515,5 +521,7 @@ def delete_monitor_template(db: Session, template_id: str) -> None:
     row = db.get(BizMonitorTemplate, template_id)
     if not row:
         raise HTTPException(status_code=404, detail="monitor_template_not_found")
+    if db.query(BizMigrationProject.id).filter(BizMigrationProject.monitor_template_id == template_id).first():
+        raise HTTPException(status_code=409, detail="monitor_template_in_use")
     db.delete(row)
     db.commit()

netx_api/biz_migration/service.py

@@ -13,7 +13,6 @@ from ..biz_state.compare_rules import apply_row_filters
 from ..biz_state.compare_service import (
     _load_metric_rows,
     _port_map_dict,
-    batch_metric_collect_ok,
     sheet_key,
     sheet_title,
     template_metrics,
@@ -28,6 +27,7 @@ from ..models import (
     BizMonitorTemplate,
     BizPortMapping,
     BizStateBatch,
+    BizStateBatchCommand,
     BizStateTask,
 )
 from ..timeutil import utcnow_naive
@@ -149,7 +149,7 @@ def hf_task_id(proj: BizMigrationProject, side: str) -> str:
 
 def current_task_id(proj: BizMigrationProject, side: str) -> str:
     """Evaluate current batches: prefer HF, fall back to portrait."""
-    return hf_task_id(proj, side) or portrait_task_id(proj, side)
+    return hf_task_id(proj, side)
 
 
 def _metric_interval_map(proj: BizMigrationProject) -> dict[str, int]:
@@ -268,15 +268,54 @@ def _current_batch_for_metric(
         # Only honor pin when it belongs to this metric's HF task (multi-interval safe)
         if pinned and (not tid or str(pinned.task_id or "") == tid):
             # Pinned partial / failed metric → treat as missing (avoid false red)
-            if not batch_metric_collect_ok(db, pinned.id, metric_id):
+            if not _metric_sample_ok(db, pinned, metric_id):
                 return None
             return pinned
-    batch = _latest_success_batch(db, tid) if tid else None
-    if batch and not batch_metric_collect_ok(db, batch.id, metric_id):
+    batch = (
+        db.query(BizStateBatch)
+        .filter(BizStateBatch.task_id == tid, BizStateBatch.status.notin_(("running", "queued")))
+        .order_by(BizStateBatch.started_at.desc(), BizStateBatch.id.desc()).first()
+    ) if tid else None
+    if batch and not _metric_sample_ok(db, batch, metric_id):
         return None
     return batch
 
 
+def _metric_sample_ok(db: Session, batch: BizStateBatch, metric_id: str) -> bool:
+    """An empty result requires a successful command receipt, just like a full one.
+
+    Batch success alone cannot prove this metric was collected. Unmatched and
+    skipped parsers are uncertainty, never evidence that a business disappeared.
+    """
+    if batch.status not in ("success", "partial"):
+        return False
+    statuses = [str(s or "").lower() for (s,) in db.query(BizStateBatchCommand.parse_status)
+                .filter(BizStateBatchCommand.batch_id == batch.id,
+                        BizStateBatchCommand.metric_id == metric_id).all()]
+    return bool(statuses) and all(s in ("ok", "ok_aux") for s in statuses)
+
+
+def _sample_timing(proj: BizMigrationProject, mid: str, old: BizStateBatch,
+                   new: BizStateBatch, *, check_age: bool) -> dict[str, Any]:
+    interval = max(60, _metric_interval_map(proj).get(mid) or proj.hf_interval_sec or 60)
+    old_at, new_at = old.ended_at or old.started_at, new.ended_at or new.started_at
+    age_limit, skew_limit = interval * 3, interval * 2
+    reasons: list[str] = []
+    if not old_at or not new_at:
+        reasons.append("sample_time_missing")
+        skew = None
+    else:
+        skew = abs((old_at - new_at).total_seconds())
+        if skew > skew_limit:
+            reasons.append("sample_skew")
+        if check_age and max((utcnow_naive() - old_at).total_seconds(),
+                             (utcnow_naive() - new_at).total_seconds()) > age_limit:
+            reasons.append("sample_stale")
+    return {"old_collected_at": _dt_iso(old_at), "new_collected_at": _dt_iso(new_at),
+            "age_limit_sec": age_limit, "skew_limit_sec": skew_limit,
+            "skew_sec": skew, "sample_issues": reasons}
+
+
 def find_portrait_task_for_ne(db: Session, *, source: str, ne_id: str) -> BizStateTask | None:
     """Best portrait (non-HF) biz_state task for an NE — for baseline binding."""
     src = str(source or "managed").strip().lower() or "managed"
@@ -655,6 +694,14 @@ def patch_project(db: Session, project_id: str, body: dict[str, Any]) -> dict[st
     p = db.get(BizMigrationProject, project_id)
     if not p:
         raise HTTPException(status_code=404, detail="project_not_found")
+    # Validate incoming baseline ownership before mutating any project field.
+    for side in ("old", "new"):
+        bid = str(body.get(f"{side}_baseline_batch_id", getattr(p, f"{side}_baseline_batch_id")) or "").strip()
+        tid = str(body.get(f"{side}_task_id", getattr(p, f"{side}_task_id")) or "").strip()
+        if bid:
+            sample = db.get(BizStateBatch, bid)
+            if not sample or sample.task_id != tid or sample.status not in ("success", "partial"):
+                raise HTTPException(status_code=400, detail=f"{side}_baseline_invalid")
     if "name" in body and body["name"] is not None:
         p.name = str(body["name"]).strip() or p.name
     if "note" in body and body["note"] is not None:
@@ -768,11 +815,13 @@ def create_batch(db: Session, project_id: str, body: dict[str, Any]) -> dict[str
         raise HTTPException(status_code=404, detail="project_not_found")
     label = str(body.get("batch_label") or "").strip() or "batch"
     expect = body.get("expect_set") if isinstance(body.get("expect_set"), dict) else {}
+    status = str(body.get("status") or "pending").strip()
+    _validate_batch_status(db, project_id, "", status)
     b = BizMigrationBatch(
         id=uuid4().hex,
         project_id=project_id,
         batch_label=label,
-        status=str(body.get("status") or "pending").strip() or "pending",
+        status=status,
         expect_set_json=dict(expect or {}),
         note=str(body.get("note") or "")[:500],
     )
@@ -819,17 +868,35 @@ def run_evaluate(
     new_batch_id: str = "",
     acceptance: bool = False,
     purpose: str = "manual",
+    _commit: bool = True,
 ) -> dict[str, Any]:
     mb = db.get(BizMigrationBatch, batch_id)
     if not mb:
         raise HTTPException(status_code=404, detail="batch_not_found")
-    proj = db.get(BizMigrationProject, mb.project_id)
+    # PostgreSQL serializes evaluate/finish for this project across API workers.
+    proj = db.query(BizMigrationProject).filter(BizMigrationProject.id == mb.project_id).with_for_update().populate_existing().first()
     if not proj:
         raise HTTPException(status_code=404, detail="project_not_found")
+    db.refresh(mb)
+    if acceptance and mb.status not in ("active", "review"):
+        raise HTTPException(status_code=409, detail="monitor_window_closed")
+    if purpose == "auto" and (mb.status != "active" or proj.status == "done"):
+        raise HTTPException(status_code=409, detail="monitor_window_closed")
     if not proj.old_baseline_batch_id:
         raise HTTPException(status_code=400, detail="old_baseline_required")
     if not proj.new_baseline_batch_id:
         raise HTTPException(status_code=400, detail="new_baseline_required")
+    for side in ("old", "new"):
+        baseline = db.get(BizStateBatch, getattr(proj, f"{side}_baseline_batch_id"))
+        if not baseline or baseline.task_id != portrait_task_id(proj, side):
+            raise HTTPException(status_code=400, detail=f"{side}_baseline_task_mismatch")
+        if baseline.status not in ("success", "partial"):
+            raise HTTPException(status_code=400, detail=f"{side}_baseline_not_complete")
+        pin = (old_batch_id if side == "old" else new_batch_id).strip()
+        if pin:
+            pinned = db.get(BizStateBatch, pin)
+            if not pinned or pinned.task_id not in _all_hf_task_ids(proj, side):
+                raise HTTPException(status_code=400, detail=f"{side}_current_batch_task_mismatch")
 
     old_batch = db.get(BizStateBatch, old_batch_id.strip()) if old_batch_id.strip() else None
     new_batch = db.get(BizStateBatch, new_batch_id.strip()) if new_batch_id.strip() else None
@@ -854,11 +921,41 @@ def run_evaluate(
     pin_new = new_batch_id.strip()
 
     port_map = _port_map_dict(db, proj.mapping_id)
+    if proj.mapping_id and not db.get(BizPortMapping, proj.mapping_id):
+        raise HTTPException(status_code=400, detail="mapping_missing")
+    if len(set(port_map.values())) != len(port_map):
+        raise HTTPException(status_code=400, detail="mapping_target_ambiguous")
     expect = parse_expect_set(mb.expect_set_json if isinstance(mb.expect_set_json, dict) else {})
+    previous_expect: dict[str, set[str]] = {}
+    for prior in db.query(BizMigrationBatch).filter(
+        BizMigrationBatch.project_id == proj.id, BizMigrationBatch.id != mb.id,
+        BizMigrationBatch.status.in_(("review", "done")),
+        BizMigrationBatch.created_at <= mb.created_at,
+    ).all():
+        for mid, keys in parse_expect_set(prior.expect_set_json).items():
+            previous_expect.setdefault(mid, set()).update(keys)
     # Final acceptance: window closed → unfinished expect = red
     window_active = (mb.status == "active") and (not acceptance)
-    mt = resolve_project_monitor_template(db, proj)
+    # Evaluations never seed or commit default configuration inside the lock.
+    mt = db.get(BizMonitorTemplate, proj.monitor_template_id) if proj.monitor_template_id else None
+    if not mt:
+        raise HTTPException(status_code=400, detail="monitor_template_missing")
+    if mt.compare_template_id and not db.get(BizCompareTemplate, mt.compare_template_id):
+        raise HTTPException(status_code=400, detail="compare_template_missing")
     sheets, sheet_overrides, defaults, iface_norm = resolve_evaluate_sheets(db, mt)
+    from .validation import validate_monitor_rules
+
+    validate_monitor_rules(defaults, sheet_overrides)
+    from ..biz_state.compare_validation import validate_template_body
+
+    validate_template_body({"metrics": sheets, "iface_normalize_rules": iface_norm})
+    known_sheets = {sheet_key(s) for s in sheets}
+    known_metrics = {str(s.get("metric_id") or "") for s in sheets}
+    for override in sheet_overrides:
+        if (override.get("sheet_id") and override["sheet_id"] not in known_sheets) or (
+            not override.get("sheet_id") and override.get("metric_id") not in known_metrics
+        ):
+            raise HTTPException(status_code=400, detail="monitor_override_sheet_missing")
     out_of_expect = str((defaults or {}).get("out_of_expect") or "strict").strip().lower()
     collect_ids = set(resolve_collect_metric_ids(db, proj))
 
@@ -867,6 +964,16 @@ def run_evaluate(
     metric_batches: dict[str, dict[str, str]] = {}
     seq = 0
     verdict_counts: dict[str, int] = {}
+    rows_cache: dict[tuple[str, str], list[dict[str, Any]]] = {}
+    command_cache: dict[str, dict[str, Any]] = {}
+    scope_ids = {str(s.get("metric_id") or "") for s in sheets} | {sheet_key(s) for s in sheets} | {"_ports"}
+    uncovered_scope = sorted(k for k, values in expect.items() if values and k not in scope_ids)
+
+    def metric_rows(bid: str, mid: str) -> list[dict[str, Any]]:
+        key = (bid, mid)
+        if key not in rows_cache:
+            rows_cache[key] = _load_metric_rows(db, batch_id=bid, metric_id=mid)
+        return rows_cache[key]
 
     for sheet in sheets:
         mid = str(sheet.get("metric_id") or "").strip()
@@ -880,6 +987,9 @@ def run_evaluate(
         field_rules = list(sheet.get("field_rules") or [])
         sheet_ov = override_for_sheet(sheet_overrides, sheet_id=sid, metric_id=mid)
         if sheet_ov.get("skip_dual"):
+            sheet_cards.append({"metric_id": mid, "sheet_id": sid, "title": sheet_title(sheet),
+                                "progress_ok": 0, "progress_total": 0, "anomaly": 0,
+                                "collect_skipped": True})
             continue
 
         # Metric not in HF collect set → skip dual eval (avoid false red)
@@ -927,21 +1037,31 @@ def run_evaluate(
             continue
         old_cur_mid = old_cur_batch.id
         new_cur_mid = new_cur_batch.id
+        timing = _sample_timing(proj, mid, old_cur_batch, new_cur_batch,
+                                check_age=acceptance or not (pin_old and pin_new))
+        baseline_missing = [side for side in ("old", "new") if not _metric_sample_ok(
+            db, db.get(BizStateBatch, getattr(proj, f"{side}_baseline_batch_id")), mid)]
+        if timing["sample_issues"] or baseline_missing:
+            sheet_cards.append({"metric_id": mid, "sheet_id": sid, "title": sheet_title(sheet),
+                                "progress_ok": 0, "progress_total": 0, "anomaly": 0,
+                                "current_missing": True, "collect_incomplete": True,
+                                "baseline_missing_sides": baseline_missing, **timing})
+            continue
 
         old_base = apply_row_filters(
-            _load_metric_rows(db, batch_id=proj.old_baseline_batch_id, metric_id=mid),
+            metric_rows(proj.old_baseline_batch_id, mid),
             row_filters,
         )
         old_now = apply_row_filters(
-            _load_metric_rows(db, batch_id=old_cur_mid, metric_id=mid),
+            metric_rows(old_cur_mid, mid),
             row_filters,
         )
         new_base_rows = apply_row_filters(
-            _load_metric_rows(db, batch_id=proj.new_baseline_batch_id, metric_id=mid),
+            metric_rows(proj.new_baseline_batch_id, mid),
             row_filters,
         )
         new_now = apply_row_filters(
-            _load_metric_rows(db, batch_id=new_cur_mid, metric_id=mid),
+            metric_rows(new_cur_mid, mid),
             row_filters,
         )
 
@@ -963,7 +1083,19 @@ def run_evaluate(
             sheet_override=sheet_ov,
             out_of_expect=out_of_expect,
             iface_normalize_rules=iface_norm,
+            previous_expect=previous_expect,
+            allow_mapping_scope=False,
         )
+        # A target snapshot taken before the source's disappearance cannot prove
+        # loss. Wait for the next target receipt; final acceptance stays blocked.
+        awaiting = 0
+        if (new_cur_batch.ended_at or new_cur_batch.started_at) < old_cur_batch.started_at:
+            for row in one["rows"]:
+                if row["in_expect"] and row["rule_hit"] == "service_not_received":
+                    row.update(verdict="awaiting_peer", color="gray", rule_hit="target_sample_predates_source")
+                    one["anomaly"] -= 1
+                    one["anomaly_in_expect"] -= 1
+                    awaiting += 1
         old_tid = _hf_task_id_for_metric(proj, "old", mid) or current_task_id(proj, "old")
         new_tid = _hf_task_id_for_metric(proj, "new", mid) or current_task_id(proj, "new")
         old_task_brief = _task_brief(db, old_tid)
@@ -973,6 +1105,7 @@ def run_evaluate(
             "new_batch_id": new_cur_mid,
             "old_task_id": old_tid,
             "new_task_id": new_tid,
+            **timing,
         }
         sheet_cards.append(
             {
@@ -993,6 +1126,8 @@ def run_evaluate(
                 "duplicate_key_list": list(one.get("duplicate_key_list") or []),
                 "duplicate_keys_before": int(one.get("duplicate_keys_before") or 0),
                 "duplicate_keys_after": int(one.get("duplicate_keys_after") or 0),
+                **timing,
+                "awaiting_peer": awaiting,
             }
         )
         for r in one["rows"]:
@@ -1003,6 +1138,7 @@ def run_evaluate(
                 new_batch_id=new_cur_mid,
                 old_task=old_task_brief,
                 new_task=new_task_brief,
+                command_cache=command_cache,
             )
             r["seq"] = seq
             r["sheet_id"] = sid
@@ -1052,6 +1188,13 @@ def run_evaluate(
                 if c.get("collect_skipped")
             ],
             "verdict_counts": verdict_counts,
+            "anomaly_outside_expect": sum(int(c.get("anomaly") or 0) - int(c.get("anomaly_in_expect") or 0) for c in active_cards),
+            "duplicate_keys": sum(int(c.get("duplicate_keys_before") or 0) + int(c.get("duplicate_keys_after") or 0) for c in active_cards),
+            "uncovered_scope": uncovered_scope,
+            "coverage_complete": bool(sheet_cards) and not uncovered_scope and not any(c.get("collect_skipped") or c.get("current_missing") or c.get("awaiting_peer") or c.get("duplicate_keys_before") or c.get("duplicate_keys_after") for c in sheet_cards),
+            "config_snapshot": {"sheets": sheets, "sheet_overrides": sheet_overrides,
+                                "defaults": defaults, "iface_normalize": iface_norm,
+                                "port_map": port_map, "expect_set": mb.expect_set_json},
             "window_active": window_active,
             "expect_ports": sorted(expect.get("_ports") or ()),
             "old_current_task_id": current_task_id(proj, "old"),
@@ -1098,6 +1241,8 @@ def run_evaluate(
                     "old_status": r.get("old_status"),
                     "new_status": r.get("new_status"),
                     "rule_hit": r.get("rule_hit") or "",
+                    "integrity_changes": r.get("integrity_changes") or [],
+                    "previous_batch": bool(r.get("previous_batch")),
                     "sheet_id": r.get("sheet_id") or "",
                     "evidence": r.get("evidence") or {},
                 },
@@ -1109,8 +1254,10 @@ def run_evaluate(
                 search_text=search[:2000],
             )
         )
-    db.commit()
-    db.refresh(run)
+    db.flush()
+    if _commit:
+        db.commit()
+        db.refresh(run)
     return run_to_dict(db, run)
 
 
@@ -1147,7 +1294,12 @@ def _command_brief(db: Session, command_id: str) -> dict[str, Any]:
     cid = str(command_id or "").strip()
     if not cid:
         return {}
-    c = db.get(BizStateBatchCommand, cid)
+    # Select metadata only: raw CLI output can be several MB per command.
+    c = db.query(BizStateBatchCommand.id, BizStateBatchCommand.raw_command,
+                 BizStateBatchCommand.parse_status, BizStateBatchCommand.row_count,
+                 BizStateBatchCommand.profile_id, BizStateBatchCommand.parser_id,
+                 BizStateBatchCommand.metric_id, BizStateBatchCommand.message).filter(
+                     BizStateBatchCommand.id == cid).first()
     if not c:
         return {"command_id": cid}
     return {
@@ -1170,6 +1322,7 @@ def _enrich_row_evidence(
     new_batch_id: str,
     old_task: dict[str, Any] | None,
     new_task: dict[str, Any] | None,
+    command_cache: dict[str, dict[str, Any]] | None = None,
 ) -> None:
     """Attach device / collect / show-command onto evaluate evidence (in-place)."""
     ev = dict(row.get("evidence") or {})
@@ -1179,7 +1332,11 @@ def _enrich_row_evidence(
     ):
         side_ev = dict(ev.get(side) or {})
         netx = dict(side_ev.get("netx") or {})
-        cmd = _command_brief(db, str(netx.get("batch_command_id") or ""))
+        cid = str(netx.get("batch_command_id") or "")
+        cache = command_cache if command_cache is not None else {}
+        if cid not in cache:
+            cache[cid] = _command_brief(db, cid)
+        cmd = cache[cid]
         side_ev["device"] = {
             "ne_id": str(brief.get("ne_id") or netx.get("ne_id") or ""),
             "ne_name": str(brief.get("ne_name") or ""),
@@ -1221,6 +1378,8 @@ def diff_to_dict(d: BizMigrationDiff) -> dict[str, Any]:
         "old_status": kj.get("old_status") or "",
         "new_status": kj.get("new_status") or "",
         "rule_hit": kj.get("rule_hit") or "",
+        "integrity_changes": kj.get("integrity_changes") or [],
+        "previous_batch": bool(kj.get("previous_batch")),
         "evidence": dict(kj.get("evidence") or {}),
         "old_kind": d.old_kind,
         "new_kind": d.new_kind,
@@ -1922,26 +2081,30 @@ def _persist_red_tickets_from_run(
         )
         .all()
     )
-    by_id: dict[tuple[str, str], BizMigrationRedTicket] = {}
+    by_id: dict[tuple[str, str, str], BizMigrationRedTicket] = {}
     for t in existing:
+        detail = t.detail_json if isinstance(t.detail_json, dict) else {}
         mk = str(getattr(t, "match_key_str", "") or "").strip()
         if not mk:
             detail = t.detail_json if isinstance(t.detail_json, dict) else {}
             mk = str(detail.get("match_old_key") or t.key_str or "").strip()
         if mk:
-            by_id[(str(t.metric_id or ""), mk)] = t
+            by_id[(str(t.metric_id or ""), str(detail.get("sheet_id") or t.metric_id or ""),
+                   str(detail.get("match_old_key") or mk))] = t
 
     created_or_updated: list[BizMigrationRedTicket] = []
-    seen: set[tuple[str, str]] = set()
+    seen: set[tuple[str, str, str]] = set()
     for d in diffs:
         kj = d.key_json if isinstance(d.key_json, dict) else {}
         mid = str(d.metric_id or "")
         match_key = str(
             kj.get("match_old_key") or kj.get("key_str") or kj.get("old_key") or ""
-        ).strip()[:256]
+        ).strip()
         if not match_key:
             match_key = str(kj.get("key_str") or "")[:256]
-        ident = (mid, match_key)
+        ident = (mid, str(kj.get("sheet_id") or mid), match_key)
+        from hashlib import sha256
+        stored_key = match_key if len(match_key) <= 256 else "sha256:" + sha256(match_key.encode()).hexdigest()
         detail = {
             "old_kind": d.old_kind,
             "new_kind": d.new_kind,
@@ -1952,6 +2115,8 @@ def _persist_red_tickets_from_run(
             "new_key": kj.get("new_key") or kj.get("new_key_str") or "",
             "match_old_key": kj.get("match_old_key") or match_key,
             "match_new_key": kj.get("match_new_key") or "",
+            "sheet_id": kj.get("sheet_id") or mid,
+            "integrity_changes": kj.get("integrity_changes") or [],
             "evidence": dict(kj.get("evidence") or {}),
         }
         prev = by_id.get(ident)
@@ -1960,7 +2125,7 @@ def _persist_red_tickets_from_run(
             prev.run_id = run_id
             prev.key_str = str(kj.get("key_str") or "")[:256]
             prev.new_key_str = str(kj.get("new_key_str") or "")[:256]
-            prev.match_key_str = match_key
+            prev.match_key_str = stored_key
             prev.verdict = d.verdict
             prev.color = d.color or "red"
             prev.old_status = str(kj.get("old_status") or "")[:64]
@@ -1980,7 +2145,7 @@ def _persist_red_tickets_from_run(
             metric_id=mid,
             key_str=str(kj.get("key_str") or "")[:256],
             new_key_str=str(kj.get("new_key_str") or "")[:256],
-            match_key_str=match_key,
+            match_key_str=stored_key,
             verdict=d.verdict,
             color=d.color or "red",
             old_status=str(kj.get("old_status") or "")[:64],
@@ -2014,14 +2179,9 @@ def finish_batch(
             detail=f"batch_not_finishable:{mb.status}",
         )
 
-    # Close window first so acceptance uses non-active rules
-    if mb.status == "active":
-        mb.ended_at = utcnow_naive()
-    mb.status = "done" if mark_done else "review"
-    mb.updated_at = utcnow_naive()
-    db.commit()
-
-    run = run_evaluate(db, batch_id=batch_id, acceptance=True, purpose="acceptance")
+    # acceptance=True closes the logical window. Persist batch closure only after
+    # evaluation succeeds, so an API/collection failure keeps the batch operable.
+    run = run_evaluate(db, batch_id=batch_id, acceptance=True, purpose="acceptance", _commit=False)
     summary = dict(run.get("summary") or {})
     progress = dict(summary.get("progress") or {})
     anomaly = int(summary.get("anomaly") or 0)
@@ -2036,6 +2196,10 @@ def finish_batch(
         fail_reasons.append("new_baseline_missing")
     if current_missing:
         fail_reasons.append("current_missing")
+    if not summary.get("coverage_complete", False):
+        fail_reasons.append("coverage_incomplete")
+    if summary.get("duplicate_keys"):
+        fail_reasons.append("duplicate_keys")
     if anomaly:
         fail_reasons.append("anomaly")
     if total > 0 and ok < total:
@@ -2061,6 +2225,9 @@ def finish_batch(
     mb = db.get(BizMigrationBatch, batch_id)
     if not mb:
         raise HTTPException(status_code=404, detail="batch_not_found")
+    if mb.status == "active":
+        mb.ended_at = utcnow_naive()
+    mb.status = "done" if mark_done else "review"
     mb.accept_run_id = str(run.get("id") or "")
     mb.accept_status = "passed" if passed else "failed"
     mb.accept_summary_json = accept_summary
@@ -2086,10 +2253,22 @@ def finish_batch(
     }
 
 
+def _validate_batch_status(db: Session, project_id: str, batch_id: str, status: str) -> None:
+    if status not in ("pending", "active", "review", "done"):
+        raise HTTPException(status_code=400, detail="batch_status_invalid")
+    if status == "active":
+        db.query(BizMigrationProject).filter(BizMigrationProject.id == project_id).with_for_update().first()
+        if db.query(BizMigrationBatch.id).filter(BizMigrationBatch.project_id == project_id,
+            BizMigrationBatch.id != batch_id, BizMigrationBatch.status == "active").first():
+            raise HTTPException(status_code=409, detail="another_batch_active")
+
+
 def patch_batch(db: Session, batch_id: str, body: dict[str, Any]) -> dict[str, Any]:
     b = db.get(BizMigrationBatch, batch_id)
     if not b:
         raise HTTPException(status_code=404, detail="batch_not_found")
+    if body.get("status") is not None:
+        _validate_batch_status(db, b.project_id, b.id, str(body["status"]).strip())
     if "batch_label" in body and body["batch_label"] is not None:
         b.batch_label = str(body["batch_label"]).strip() or b.batch_label
     if "note" in body and body["note"] is not None:
@@ -2135,6 +2314,8 @@ def list_run_diffs(
     run_id: str,
     *,
     metric_id: str = "",
+    sheet_id: str = "",
+    only_expect: bool = False,
     verdict: str = "",
     color: str = "",
     kw: str = "",
@@ -2146,19 +2327,23 @@ def list_run_diffs(
     q = db.query(BizMigrationDiff).filter(BizMigrationDiff.run_id == run_id)
     if metric_id:
         q = q.filter(BizMigrationDiff.metric_id == metric_id)
+    if sheet_id:
+        q = q.filter(BizMigrationDiff.key_json["sheet_id"].as_string() == sheet_id)
+    if only_expect:
+        q = q.filter(BizMigrationDiff.in_expect.is_(True))
     if verdict:
         q = q.filter(BizMigrationDiff.verdict == verdict)
     if color:
         q = q.filter(BizMigrationDiff.color == color)
     if kw.strip():
-        like = f"%{kw.strip()}%"
-        q = q.filter(BizMigrationDiff.search_text.ilike(like))
+        literal = kw.strip().replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
+        q = q.filter(BizMigrationDiff.search_text.ilike(f"%{literal}%", escape="\\"))
     total = q.count()
     rows = q.order_by(BizMigrationDiff.seq.asc()).offset(max(0, offset)).limit(min(500, max(1, limit))).all()
     return {
         "total": total,
-        "offset": offset,
-        "limit": limit,
+        "offset": max(0, offset),
+        "limit": min(500, max(1, limit)),
         "items": [diff_to_dict(d) for d in rows],
     }
 

netx_api/biz_migration/validation.py

@@ -0,0 +1,70 @@
+"""Validate monitor correction rules before they can affect live verdicts."""
+from typing import Any
+
+from fastapi import HTTPException
+
+from .evaluate import _groups_for_side, _normalize_cond_type
+
+
+def validate_monitor_rules(defaults: Any, overrides: Any) -> None:
+    def fail(path: str) -> None:
+        raise HTTPException(status_code=400, detail=f"monitor_rule_invalid:{path}")
+
+    if not isinstance(defaults, dict) or not isinstance(overrides, list):
+        fail("defaults/sheet_overrides")
+    if defaults.get("out_of_expect", "strict") not in ("strict", "warn", "ignore"):
+        fail("defaults.out_of_expect")
+    seen = set()
+    for i, ov in enumerate(overrides):
+        path = f"sheet_overrides[{i}]"
+        if not isinstance(ov, dict) or not (ov.get("sheet_id") or ov.get("metric_id")):
+            fail(path)
+        ident = ov.get("sheet_id") or ov.get("metric_id")
+        if ident in seen:
+            fail(path + ".duplicate")
+        seen.add(ident)
+        for key in ("status_fields", "up_values", "down_values"):
+            if key in ov and (not isinstance(ov[key], list) or any(not str(v).strip() for v in ov[key])):
+                fail(path + "." + key)
+        for key in ("success", "anomaly"):
+            patterns = ov.get(key, [])
+            if not isinstance(patterns, list):
+                fail(path + "." + key)
+            for j, pat in enumerate(patterns):
+                pp = f"{path}.{key}[{j}]"
+                if not isinstance(pat, dict):
+                    fail(pp)
+                for side in ("old", "new"):
+                    for attr in (side, side + "_conds", side + "_groups"):
+                        if attr in pat and not isinstance(pat[attr], list):
+                            fail(pp + "." + attr)
+                    if any(isinstance(g, list) and not g for g in pat.get(side + "_groups", [])):
+                        fail(pp + "." + side + "_groups.empty_group")
+                    groups = _groups_for_side(pat, side)
+                    if key == "success" and not groups:
+                        fail(pp + "." + side + ".required")
+                    for group in groups:
+                        for cond in group:
+                            cp = pp + "." + side
+                            if isinstance(cond, str):
+                                if cond not in ("added", "removed", "changed", "unchanged", "up", "down", "other", "none") and not (cond.startswith("field:") and cond.count(":") >= 2):
+                                    fail(cp + ".token")
+                            elif isinstance(cond, dict):
+                                typ = _normalize_cond_type(cond)
+                                if typ not in ("presence", "status", "value"):
+                                    fail(cp + ".type")
+                                val = cond.get("value", cond.get("kind", cond.get("status")))
+                                if typ == "presence" and val not in ("added", "removed", "changed", "unchanged"):
+                                    fail(cp + ".presence")
+                                if typ == "status" and val not in ("up", "down", "other", "none"):
+                                    fail(cp + ".status")
+                                if typ == "value":
+                                    op = cond.get("op", "eq")
+                                    if op not in ("eq", "ne", "in", "not_in", "empty", "not_empty", "changed", "unchanged", "same", "diff"):
+                                        fail(cp + ".op")
+                                    if op in ("eq", "ne", "in", "not_in") and (val is None or val == "" or val == []):
+                                        fail(cp + ".value")
+                            else:
+                                fail(cp + ".condition")
+                if key == "anomaly" and not any(_groups_for_side(pat, s) for s in ("old", "new")):
+                    fail(pp + ".empty")

netx_api/biz_migration_router.py

@@ -285,6 +285,8 @@ def api_get_run(run_id: str, db: Session = Depends(get_db)):
 def api_list_diffs(
     run_id: str,
     metric_id: str = "",
+    sheet_id: str = "",
+    only_expect: bool = False,
     verdict: str = "",
     color: str = "",
     kw: str = "",
@@ -296,6 +298,8 @@ def api_list_diffs(
         db,
         run_id,
         metric_id=metric_id,
+        sheet_id=sheet_id,
+        only_expect=only_expect,
         verdict=verdict,
         color=color,
         kw=kw,

netx_api/biz_state/collect_runner.py

@@ -1720,6 +1720,14 @@ def _finalize_batch_status(
                 batch.message = ""
         status = str(batch.status or "")
         db.commit()
+        # Failed/partial receipts also update cutover coverage; never keep an old
+        # success looking healthy after the latest collection has failed.
+        try:
+            from ..biz_migration.auto_monitor import schedule_auto_monitor_for_task
+
+            schedule_auto_monitor_for_task(task_id)
+        except Exception:
+            _log.exception("cutover monitor schedule failed task=%s", task_id)
         # Only full success triggers auto compare; never block the collect thread.
         if status == "success" and not stopped:
             try:

tests/test_biz_migration_evaluate.py

@@ -733,7 +733,7 @@ class EvaluateMetricDualTests(unittest.TestCase):
         migrated = [r for r in out["rows"] if r["verdict"] == "migrated"]
         self.assertTrue(migrated, out["rows"])
 
-    def test_out_of_expect_ignore_no_red(self):
+    def test_out_of_expect_ignore_cannot_hide_business_drift(self):
         old_base = [{"interface": "gei-keep", "admin": "up", "phy": "up", "prot": "up"}]
         old_cur: list[dict] = []
         new_cur = [{"interface": "gei-keep", "admin": "up", "phy": "up", "prot": "up"}]
@@ -753,9 +753,9 @@ class EvaluateMetricDualTests(unittest.TestCase):
             out_of_expect="ignore",
         )
         reds = [r for r in out["rows"] if r["color"] == "red"]
-        self.assertFalse(reds, out["rows"])
+        self.assertTrue(reds, out["rows"])
         involved = [r for r in out["rows"] if r["verdict"] != "not_involved"]
-        self.assertFalse(involved, out["rows"])
+        self.assertTrue(involved, out["rows"])
 
     def test_unexpected_loss_is_anomaly(self):
         old_base = [{"interface": "gei-keep", "admin": "up", "phy": "up", "prot": "up"}]
@@ -823,7 +823,7 @@ class EvaluateMetricDualTests(unittest.TestCase):
         self.assertFalse(out["new_baseline_missing"])
         self.assertEqual(out["new_baseline_mode"], "port_mapped")
 
-    def test_empty_provided_new_baseline_falls_back_or_missing(self):
+    def test_empty_collected_new_baseline_is_valid(self):
         old_base = [{"peer": "1.1.1.1", "state": "Established"}]
         new_cur = [{"peer": "1.1.1.1", "state": "Established"}]
         expect = parse_expect_set({"items": [{"metric_id": "bgp_peer", "key": "1.1.1.1"}]})
@@ -841,8 +841,8 @@ class EvaluateMetricDualTests(unittest.TestCase):
             window_active=True,
             sheet_override={},
         )
-        self.assertTrue(out["new_baseline_missing"])
-        self.assertEqual(out["new_baseline_mode"], "empty")
+        self.assertFalse(out["new_baseline_missing"])
+        self.assertEqual(out["new_baseline_mode"], "provided")
         # Still reports presence as added vs empty before — callers should heed the flag.
         kinds = {r["new_kind"] for r in out["rows"] if r.get("in_expect")}
         self.assertIn("added", kinds)
@@ -1153,7 +1153,10 @@ class PortMapScopeTests(unittest.TestCase):
         )
         keys = {r["key_str"] for r in out["rows"]}
         self.assertIn("10.0.0.1|gei-old", keys)
-        self.assertNotIn("10.0.0.2|gei-keep", keys)
+        self.assertIn("10.0.0.2|gei-keep", keys)
+        outside = next(r for r in out["rows"] if r["key_str"] == "10.0.0.2|gei-keep")
+        self.assertEqual(outside["color"], "red")
+        self.assertFalse(outside["in_expect"])
         migrated = [r for r in out["rows"] if r["verdict"] == "migrated"]
         self.assertTrue(migrated, out["rows"])
         self.assertEqual(migrated[0]["new_key_str"], "10.0.0.1|gei-new")

tests/test_biz_migration_evidence.py

@@ -149,6 +149,8 @@ class EvidencePersistTests(unittest.TestCase):
             command_count=1,
         )
         self.db.add(b)
+        self.db.add(BizStateBatchCommand(id=f"receipt-{bid}", batch_id=bid,
+                                         metric_id="interface_brief", parse_status="ok", row_count=0))
         self.db.commit()
         return b
 
@@ -233,7 +235,7 @@ class EvidencePersistTests(unittest.TestCase):
             raw_command="show interface brief",
             interface="GE-old",
         )
-        # New baseline empty → port_mapped mode uses mapped old
+        # Successfully collected empty new baseline is valid
         # Old current: gone (no rows) — migration off old
         # New current: xgei-new up
         self._seed_if_row(

tests/test_biz_migration_hf_decouple.py

@@ -14,6 +14,7 @@ from netx_api.models import (
     BizMigrationProject,
     BizMonitorTemplate,
     BizStateBatch,
+    BizStateBatchCommand,
     BizStateTask,
 )
 from netx_api.biz_migration import service as mig
@@ -103,6 +104,9 @@ class HfDecoupleFlowTests(unittest.TestCase):
             row_count=1,
         )
         self.db.add(b)
+        for mid in ("interface_brief", "arp"):
+            self.db.add(BizStateBatchCommand(id=f"receipt-{bid}-{mid}", batch_id=bid,
+                                             metric_id=mid, parse_status="ok", row_count=0))
         self.db.commit()
         return b
 
@@ -170,7 +174,7 @@ class HfDecoupleFlowTests(unittest.TestCase):
         batch = mig.create_batch(
             self.db, pid, {"batch_label": "n1", "expect_set": {"ports": ["gei-0/1"]}}
         )
-        with mock.patch.object(mig, "resolve_evaluate_sheets", return_value=([], [], {}, [])):
+        with mock.patch.object(mig, "resolve_evaluate_sheets", return_value=([mig.port_sheet_def()], [], {}, [])):
             out = mig.run_evaluate(self.db, batch_id=batch["id"])
         self.assertEqual(out["old_batch_id"], "cur_old")
         self.assertEqual(out["new_batch_id"], "cur_new")

tests/test_biz_migration_integrity.py

@@ -0,0 +1,239 @@
+"""Business continuity, wave scope, sample uncertainty and live-monitor regressions."""
+from datetime import timedelta
+
+import pytest
+from fastapi import HTTPException
+from sqlalchemy import create_engine, event
+from sqlalchemy.orm import sessionmaker
+
+from netx_api.db import Base
+from netx_api.models import (BizCompareTemplate, BizMonitorTemplate, BizMigrationProject,
+    BizMigrationBatch, BizMigrationRun, BizMigrationDiff, BizStateTask, BizStateBatch,
+    BizStateBatchCommand, BizStateMetricRow)
+from netx_api.timeutil import utcnow_naive
+from netx_api.biz_migration import service as svc
+from netx_api.biz_migration.auto_monitor import try_auto_monitor_for_task
+from netx_api.biz_migration.evaluate import evaluate_metric_dual, parse_expect_set, _eval_field_op
+from netx_api.biz_migration.validation import validate_monitor_rules
+
+
+def evaluate(**changes):
+    args = dict(metric_id="arp", sheet_id="arp", key_fields=["ip"], iface_fields=["port"],
+                compare_fields=["mac", "port"], old_baseline_rows=[{"ip": "a", "mac": "good", "port": "old"}],
+                old_current_rows=[], new_baseline_rows=[],
+                new_current_rows=[{"ip": "a", "mac": "good", "port": "new"}],
+                port_map={"old": "new"}, expect=parse_expect_set({"items": [{"metric_id": "arp", "key": "a"}]}),
+                window_active=True, allow_mapping_scope=False)
+    return evaluate_metric_dual(**{**args, **changes})
+
+
+def test_target_content_checked_against_old_business():
+    out = evaluate(new_current_rows=[{"ip": "a", "mac": "wrong", "port": "new"}])
+    assert out["progress_ok"] == 0
+    assert out["rows"][0]["rule_hit"] == "service_content_changed"
+    assert out["rows"][0]["integrity_changes"][0]["field"] == "mac"
+
+
+def test_content_normalization_and_ignore_follow_compare_template():
+    out = evaluate(new_current_rows=[{"ip": "a", "mac": "GOOD", "port": "new"}],
+                   field_rules=[{"field": "mac", "normalize": "upper"}])
+    assert out["progress_ok"] == 1
+
+
+def test_down_target_cannot_pass_permissive_success_rule():
+    out = evaluate(new_current_rows=[{"ip": "a", "state": "down"}],
+                   sheet_override={"status_fields": ["state"], "up_values": ["up"], "down_values": ["down"],
+                                   "success": [{"old": ["removed"], "new": ["added"]}]})
+    assert out["rows"][0]["verdict"] == "lost"
+    assert out["rows"][0]["color"] == "red"
+
+
+def test_mapping_does_not_expand_wave_selection():
+    out = evaluate(old_baseline_rows=[{"ip": "a", "port": "old"}, {"ip": "b", "port": "old2"}],
+                   new_current_rows=[{"ip": "a", "port": "new"}], compare_fields=[],
+                   port_map={"old": "new", "old2": "new2"})
+    assert out["progress_total"] == 1
+    assert next(r for r in out["rows"] if r["old_key"] == "b")["color"] == "red"
+
+
+def test_port_selection_covers_business_with_nonkey_interface():
+    out = evaluate(expect=parse_expect_set({"ports": ["old"]}))
+    assert out["progress_total"] == out["progress_ok"] == 1
+
+
+@pytest.mark.parametrize("healthy", [True, False])
+def test_previous_wave_remains_protected_without_changing_current_progress(healthy):
+    out = evaluate(expect={}, previous_expect=parse_expect_set({"items": [{"metric_id": "arp", "key": "a"}]}),
+                   new_current_rows=[{"ip": "a", "mac": "good", "port": "new"}] if healthy else [])
+    assert out["progress_total"] == 0
+    assert out["rows"][0]["verdict"] == ("migrated_previous" if healthy else "lost")
+    assert out["anomaly"] == (0 if healthy else 1)
+
+
+@pytest.mark.parametrize("value", [0, False])
+def test_zero_false_are_not_blank_in_keys_or_conditions(value):
+    assert _eval_field_op({"id": value}, "id", "eq", value)
+    out = evaluate(old_baseline_rows=[{"ip": value}], new_current_rows=[{"ip": value}],
+                   compare_fields=[], expect=parse_expect_set({"items": [{"metric_id": "arp", "key": [str(value).lower()]}]}))
+    assert out["rows"][0]["old_key"] == str(value).lower()
+
+
+@pytest.fixture
+def db():
+    engine = create_engine("sqlite+pysqlite:///:memory:")
+    Base.metadata.create_all(engine)
+    with sessionmaker(bind=engine, expire_on_commit=False, autoflush=False)() as session:
+        now = utcnow_naive() - timedelta(seconds=1)
+        sheet = {"metric_id": "arp", "sheet_id": "s", "key_fields": ["ip"], "compare_fields": ["mac"]}
+        session.add(BizCompareTemplate(id="ct", name="cmp", metrics_json=[sheet]))
+        session.add(BizMonitorTemplate(id="mt", name="mon", compare_template_id="ct",
+                                      collect_metric_ids_json=["arp"], defaults_json={}, sheet_overrides_json=[]))
+        for tid in ("op", "np", "oh", "nh"):
+            session.add(BizStateTask(id=tid, ne_id=tid, ne_name=tid, purpose="cutover_hf" if tid.endswith("h") else ""))
+        session.add(BizMigrationProject(id="p", name="p", old_task_id="op", new_task_id="np",
+                    old_hf_task_id="oh", new_hf_task_id="nh", old_baseline_batch_id="ob", new_baseline_batch_id="nb",
+                    monitor_template_id="mt", hf_interval_sec=60, status="active"))
+        session.add(BizMigrationBatch(id="wave", project_id="p", batch_label="wave", status="active",
+                    expect_set_json={"items": [{"metric_id": "arp", "sheet_id": "s", "key": "a"}]}))
+        for i, (bid, tid) in enumerate((("ob", "op"), ("nb", "np"), ("oc", "oh"), ("nc", "nh"))):
+            at = now - timedelta(milliseconds=100 * (4 - i))
+            session.add(BizStateBatch(id=bid, task_id=tid, status="success", started_at=at, ended_at=at))
+            session.add(BizStateBatchCommand(id="cmd-" + bid, batch_id=bid, metric_id="arp", parse_status="ok", raw_text="secret CLI payload"))
+        for bid in ("ob", "nc"):
+            session.add(BizStateMetricRow(id="row-" + bid, batch_id=bid, task_id="op" if bid == "ob" else "nh",
+                        batch_command_id="cmd-" + bid, metric_id="arp", seq=0, data_json={"ip": "a", "mac": "good"}))
+        session.commit()
+        yield session
+
+
+def test_collected_empty_baseline_can_pass_acceptance(db):
+    finished = svc.finish_batch(db, "wave")
+    assert finished["accept_summary"]["passed"]
+    assert finished["run"]["summary"]["config_snapshot"]["expect_set"]
+
+
+@pytest.mark.parametrize("problem", ["missing_receipt", "failed_receipt", "stale", "skew", "duplicate", "skipped", "unknown_scope"])
+def test_uncertain_coverage_cannot_pass_acceptance(db, problem):
+    if problem == "missing_receipt":
+        db.delete(db.get(BizStateBatchCommand, "cmd-nb"))
+    elif problem == "failed_receipt":
+        db.get(BizStateBatchCommand, "cmd-nc").parse_status = "failed"
+    elif problem in ("stale", "skew"):
+        batch = db.get(BizStateBatch, "nc")
+        batch.started_at = batch.ended_at = utcnow_naive() - timedelta(minutes=10)
+    elif problem == "duplicate":
+        db.add(BizStateMetricRow(id="dup", batch_id="nc", metric_id="arp", data_json={"ip": "a", "mac": "good"}))
+    elif problem == "skipped":
+        db.get(BizMigrationProject, "p").collect_metric_ids_json = ["other"]
+    else:
+        db.get(BizMigrationBatch, "wave").expect_set_json = {"items": [{"sheet_id": "deleted", "metric_id": "arp", "key": "a"}]}
+    db.commit()
+    assert not svc.finish_batch(db, "wave")["accept_summary"]["passed"]
+
+
+def test_invalid_baseline_and_pinned_batch_do_not_fallback(db):
+    with pytest.raises(HTTPException) as error:
+        svc.run_evaluate(db, batch_id="wave", old_batch_id="nb")
+    assert error.value.detail == "old_current_batch_task_mismatch"
+    db.get(BizMigrationProject, "p").old_baseline_batch_id = "nc"
+    db.commit()
+    with pytest.raises(HTTPException):
+        svc.finish_batch(db, "wave")
+    assert db.get(BizMigrationBatch, "wave").status == "active"
+    assert db.query(BizMigrationRun).count() == 0
+
+
+def test_latest_failed_collection_does_not_reuse_previous_success(db):
+    db.add(BizStateBatch(id="latest-failed", task_id="nh", status="failed", started_at=utcnow_naive(), ended_at=utcnow_naive()))
+    db.commit()
+    run = svc.run_evaluate(db, batch_id="wave")
+    assert run["summary"]["current_missing_metrics"] == ["s"]
+    assert not run["summary"]["coverage_complete"]
+
+
+def test_target_older_than_source_is_uncertain_instead_of_false_loss(db):
+    db.delete(db.get(BizStateMetricRow, "row-nc"))
+    db.get(BizStateBatch, "oc").started_at = utcnow_naive()
+    db.get(BizStateBatch, "oc").ended_at = utcnow_naive()
+    db.commit()
+    run = svc.run_evaluate(db, batch_id="wave")
+    row = svc.list_run_diffs(db, run["id"])["items"][0]
+    assert row["verdict"] == "awaiting_peer"
+    assert run["summary"]["anomaly"] == 0
+    assert not run["summary"]["coverage_complete"]
+
+
+def test_auto_monitor_deduplicates_receipts_and_respects_closed_window(db):
+    assert try_auto_monitor_for_task(db, "oh") == 1
+    assert try_auto_monitor_for_task(db, "nh") == 0
+    assert db.query(BizMigrationRun).one().purpose == "auto"
+    db.get(BizMigrationBatch, "wave").status = "done"
+    db.commit()
+    assert try_auto_monitor_for_task(db, "oh") == 0
+
+
+def test_two_active_waves_are_rejected(db):
+    with pytest.raises(HTTPException) as error:
+        svc.create_batch(db, "p", {"status": "active", "batch_label": "overlap"})
+    assert error.value.detail == "another_batch_active"
+    assert db.query(BizMigrationBatch).count() == 1
+
+
+def test_sheets_with_same_key_keep_separate_red_tickets(db):
+    run = svc.run_evaluate(db, batch_id="wave")
+    for sid in ("one", "two"):
+        db.add(BizMigrationDiff(id=sid, run_id=run["id"], metric_id="arp", color="red",
+                               key_json={"sheet_id": sid, "match_old_key": "a"}, verdict="lost"))
+    db.flush()
+    tickets = svc._persist_red_tickets_from_run(db, project_id="p", batch_id="wave", run_id=run["id"])
+    assert len(tickets) == 2
+    assert {t.detail_json["sheet_id"] for t in tickets} == {"one", "two"}
+
+
+def test_port_mapping_does_not_remap_business_key_with_same_text():
+    out = evaluate(old_baseline_rows=[{"ip": "old", "port": "old"}],
+                   new_current_rows=[{"ip": "old", "port": "new"}], compare_fields=["port"],
+                   expect=parse_expect_set({"items": [{"metric_id": "arp", "key": "old"}]}))
+    assert out["progress_ok"] == 1
+
+
+def test_wave_acceptance_rolls_back_whole_transaction_on_ticket_failure(db):
+    from unittest.mock import patch
+
+    with patch.object(svc, "_persist_red_tickets_from_run", side_effect=RuntimeError("write failed")):
+        with pytest.raises(RuntimeError):
+            svc.finish_batch(db, "wave")
+    db.rollback()
+    assert db.query(BizMigrationRun).count() == 0
+    assert db.get(BizMigrationBatch, "wave").status == "active"
+
+
+def test_command_metadata_is_cached_and_raw_cli_is_not_selected(db):
+    statements = []
+    def capture(_conn, _cursor, statement, *_args):
+        statements.append(statement)
+    event.listen(db.bind, "before_cursor_execute", capture)
+    try:
+        run = svc.run_evaluate(db, batch_id="wave")
+    finally:
+        event.remove(db.bind, "before_cursor_execute", capture)
+    command_selects = [q for q in statements if q.startswith("SELECT") and "biz_state_batch_command" in q]
+    assert all("raw_text" not in q for q in command_selects)
+    assert run["summary"]["coverage_complete"]
+
+
+def test_diff_filters_apply_before_pagination_and_escape_wildcards(db):
+    run = svc.run_evaluate(db, batch_id="wave")
+    db.add(BizMigrationDiff(id="outside", run_id=run["id"], metric_id="arp", seq=999, color="red",
+                           key_json={"sheet_id": "outside"}, in_expect=False, search_text="a_100%"))
+    db.commit()
+    result = svc.list_run_diffs(db, run["id"], sheet_id="outside", color="red", limit=1)
+    assert result["total"] == 1 and result["items"][0]["id"] == "outside"
+    assert svc.list_run_diffs(db, run["id"], only_expect=True)["total"] == 1
+    assert svc.list_run_diffs(db, run["id"], kw="_")["total"] == 1
+
+
+@pytest.mark.parametrize("condition", [{"type": "value", "field": "state", "op": "typo", "value": "up"}, {"type": "presence", "value": "typo"}])
+def test_invalid_correction_rules_fail_before_evaluation(condition):
+    with pytest.raises(HTTPException):
+        validate_monitor_rules({}, [{"metric_id": "arp", "success": [{"old_groups": [[condition]], "new": ["added"]}]}])