feat(migration): refuse a full re-import that would delete native rows

A full run_all.py pass truncates and rebuilds every table it owns from the
Access extract. That was harmless while the platform was a read-only mirror --
every row came from the extract, so wiping and rebuilding lost nothing. It
stopped being harmless once the platform started minting rows Access has never
heard of: allocated portal NUMids, customers created in the staff UI,
OCR-captured policies, app-booked ledger rows, uploaded documents.

REIMPORT is a button in /operaciones, so that was one click away.

native_guard.py counts what only exists here and exits 3; run_all.py runs it
before the first truncate and stops. Detecting an allocated NUMid needs the
staged Parquet -- the customer holds an ordinary-looking (utilities, DATGRAL,
'1172') ref, so "customer has no refs" cannot see it and only comparing against
the extract can. Missing staging is therefore treated as blocking rather than
as "nothing to protect".

The guard does not teach full mode to preserve anything: --sync already upserts
legacy rows against the existing refs and leaves the rest alone, and rebuilding
that inside full mode would re-implement it. --force-full (checkbox in the
REIMPORT confirm, recorded in the audit log) deletes them deliberately.

Verified against dev: clean before, exit 3 listing utilities/1172 with a
synthetic ref present, clean again after removing it.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

apps/api/src/customers/numid.service.ts

@@ -33,7 +33,9 @@ export const UTILITIES_TABLE = "DATGRAL";
  * insurance/DATGRAL is a SEPARATE id space that reuses the same sourceTable name
  * and runs past 4,000. It must never be read as a NUMid, and never allocated
  * from: the portal cannot resolve those ids. Every query here filters on BOTH
- * columns for that reason, never on sourceTable alone.
+ * columns for that reason, never on sourceTable alone. A customer can also hold
+ * more than one insurance ref — 16 of them do, where several insurance rows
+ * folded into one customer — so those are tested with EXISTS rather than joined.
  */
 const POOL = {
   sourceSystem: UTILITIES_SYSTEM,

apps/api/src/ops/ops.controller.ts

@@ -117,11 +117,17 @@ export class OpsController {
   @Post("jobs")
   async startJob(@Body() dto: StartJobDto, @Req() req: Request) {
     const userId = this.actingId(req);
-    const job = await this.ops.startJob(dto.kind, { file: dto.file }, userId);
+    const job = await this.ops.startJob(
+      dto.kind,
+      { file: dto.file, forceFull: dto.forceFull },
+      userId,
+    );
     void this.audit.log(userId, "ops.job.start", {
       jobId: job.id,
       kind: dto.kind,
       file: dto.file,
+      // Recorded because this is the flag that authorised deleting native rows.
+      forceFull: dto.forceFull,
     });
     return job;
   }

apps/api/src/ops/ops.service.ts

@@ -415,12 +415,19 @@ export class OpsService implements OnModuleInit {
       const out = shq(path.join(this.backupDir, file));
       const py = await this.pythonBin();
       const runAll = shq(path.join(this.migrationDir, "run_all.py"));
+      // run_all.py runs native_guard.py before it truncates anything and exits
+      // without touching the database when the target holds rows that only
+      // exist here — allocated portal NUMids, app-created customers, OCR
+      // captures. --force-full is what the operator ticks to delete them
+      // anyway; without it the job fails with the list.
+      const force = params.forceFull === true;
       const cmd =
         `${PIPEFAIL}echo '== Respaldo de seguridad previo ==' && ` +
         `${this.dumpCommand(flags, db, out)} && ` +
         `echo '== Reimportación desde carpeta de ingesta ==' && ` +
-        `${shq(py)} ${runAll} --env ${shq(this.migrationEnv)} --stage`;
-      return { cmd, resolvedParams: { safetyBackup: file } };
+        `${shq(py)} ${runAll} --env ${shq(this.migrationEnv)} --stage` +
+        (force ? " --force-full" : "");
+      return { cmd, resolvedParams: { safetyBackup: file, forceFull: force } };
     }
 
     throw new BadRequestException(`Operación no soportada: ${kind}`);

apps/api/src/ops/start-job.dto.ts

@@ -1,4 +1,4 @@
-import { IsEnum, IsOptional, IsString } from "class-validator";
+import { IsBoolean, IsEnum, IsOptional, IsString } from "class-validator";
 import { OpsJobKind } from "@jorgecuadros/database";
 
 export class StartJobDto {
@@ -9,4 +9,13 @@ export class StartJobDto {
   @IsOptional()
   @IsString()
   file?: string;
+
+  /**
+   * REIMPORT only: proceed even though the rebuild deletes rows that exist only
+   * in the platform. Off by default, so the guard in run_all.py stops the job
+   * and lists what would be lost rather than the operator finding out after.
+   */
+  @IsOptional()
+  @IsBoolean()
+  forceFull?: boolean;
 }

apps/web/src/app/operaciones/page.tsx

@@ -59,6 +59,7 @@ function Operaciones() {
   const [notice, setNotice] = useState<string | null>(null);
   const [confirm, setConfirm] = useState<ConfirmState>(null);
   const [confirmText, setConfirmText] = useState("");
+  const [forceFull, setForceFull] = useState(false);
   const [uploading, setUploading] = useState<string | null>(null);
   const [progress, setProgress] = useState<UploadProgress | null>(null);
   const [starting, setStarting] = useState(false);
@@ -159,12 +160,12 @@ function Operaciones() {
     }
   }
 
-  async function start(kind: OpsJobKind, file?: string) {
+  async function start(kind: OpsJobKind, file?: string, force?: boolean) {
     setError(null);
     setNotice(null);
     setStarting(true);
     try {
-      const job = await startOpsJob(kind, file);
+      const job = await startOpsJob(kind, file, force);
       setActiveJob(job);
       setJobs((prev) => (prev ? [job, ...prev] : [job]));
     } catch (e) {
@@ -177,6 +178,9 @@ function Operaciones() {
   function askConfirm(state: ConfirmState) {
     setConfirm(state);
     setConfirmText("");
+    // Always re-armed: ticking "delete native rows" once must not carry into
+    // the next reimport.
+    setForceFull(false);
     setError(null);
     setNotice(null);
   }
@@ -185,7 +189,7 @@ function Operaciones() {
     if (!confirm) return;
     const c = confirm;
     setConfirm(null);
-    if (c.kind === "REIMPORT") await start("REIMPORT");
+    if (c.kind === "REIMPORT") await start("REIMPORT", undefined, forceFull);
     else if (c.kind === "SYNC") await start("SYNC");
     else await start("RESTORE", c.file);
   }
@@ -478,11 +482,24 @@ function Operaciones() {
             </h2>
             <p className="inline-form-note">
               {confirm.kind === "REIMPORT"
-                ? "Esto BORRA todos los datos actuales (incluidos los capturados a mano) y reconstruye desde los archivos de ingesta. Se creará un respaldo previo automático."
+                ? "Esto BORRA todos los datos actuales y reconstruye desde los archivos de ingesta. Se creará un respaldo previo automático. Si la base contiene registros que sólo existen en la plataforma (clientes creados aquí, números de portal asignados, pólizas capturadas por OCR, movimientos capturados), la operación se detiene y los enumera sin tocar nada."
                 : confirm.kind === "SYNC"
                   ? "Se creará un respaldo previo automático. Luego se importarán al sistema los registros nuevos del legado y se eliminarán los del legado que ya no aparezcan en los archivos de ingesta. Los datos capturados a mano NO se borran."
                   : `Esto sobreescribe la base de datos completa con “${confirm.file}”. Se recomienda crear un respaldo antes.`}
             </p>
+            {confirm.kind === "REIMPORT" && (
+              <label className="inline-form-note" style={{ display: "block" }}>
+                <input
+                  type="checkbox"
+                  checked={forceFull}
+                  onChange={(e) => setForceFull(e.target.checked)}
+                  style={{ marginRight: 8 }}
+                />
+                Borrar también los registros que sólo existen en la plataforma
+                (ignorar la verificación). Sólo marque esto si de verdad quiere
+                perderlos.
+              </label>
+            )}
             <label className="field">
               <span className="field-label">Escriba CONFIRMAR para continuar</span>
               <input

apps/web/src/lib/api.ts

@@ -973,10 +973,16 @@ export function getOpsJob(id: string): Promise<OpsJob> {
 }
 
 /** Start a mutating op. `file` is required for RESTORE. 409 if one is running. */
-export function startOpsJob(kind: OpsJobKind, file?: string): Promise<OpsJob> {
+/** `forceFull` applies to REIMPORT only: proceed even though the rebuild
+ *  deletes rows that exist only in the platform. */
+export function startOpsJob(
+  kind: OpsJobKind,
+  file?: string,
+  forceFull?: boolean,
+): Promise<OpsJob> {
   return apiFetch<OpsJob>("/ops/jobs", {
     method: "POST",
-    body: JSON.stringify({ kind, file }),
+    body: JSON.stringify({ kind, file, forceFull }),
   });
 }
 

docs/BACKLOG.md

@@ -195,13 +195,14 @@ insurance-only customer has no reason to hold a utilities id.
   (`transform_customers.py:327`), so an id recycled today is handed back to its
   Access owner on the next sync and the customer given it loses portal access.
   **Flip it on after utilities cuts over**, or for ids deleted at the source.
-- **A full re-import destroys every natively allocated id.**
-  `transform_customers.py:246` truncates `customers` and `customer_legacy_refs`,
-  then rebuilds the pool from Access alone. Until that is fixed — either by
-  mandating `--sync` for all future utilities loads, or by teaching the transform
-  to preserve refs with no Access counterpart — a NUMid issued here survives only
-  until the next full load. This is a prerequisite for the cutover, not a
-  nice-to-have.
+- **A full re-import would destroy every natively allocated id — now guarded.**
+  `transform_customers.py:246` truncates `customers` and `customer_legacy_refs`
+  (and the other transforms truncate everything they own), then rebuild from
+  Access alone. `migration/native_guard.py` runs before any of it and refuses
+  when the target holds rows Access has never seen; `run_all.py --force-full`,
+  or the checkbox in the REIMPORT confirm, overrides and deletes them. **`--sync`
+  remains the correct path for any database with native rows** — the guard stops
+  the loss, it does not make full mode preserve anything.
 - **The empty-id rule exists twice**: enforced in `numid.service.ts`
   (`EMPTY_NUMID_SQL`) and reported by `scripts/numid-audit.sql`. They agree today
   (both return 1089, 1094, 1134, 1143 on dev); they are not mechanically kept in

migration/native_guard.py

@@ -0,0 +1,210 @@
+"""
+Refuse a full re-import that would delete platform-native data.
+
+A full `run_all.py` pass truncates and rebuilds every table it owns from the
+Access extract:
+
+    transform_customers.py    customers, customer_legacy_refs
+    transform_properties.py   properties, property_services, service_documents,
+                              trust_accounts
+    transform_policies.py     policies + installments, vehicles, drivers,
+                              beneficiaries, claims, adjusters, policy_types,
+                              insurance_providers
+    transform_transactions.py transactions, type_transactions, exchange_rates
+    transform_bank.py         bank tables
+    blob_extract.py           service_documents, policy_documents
+
+That was harmless while the platform was a read-only mirror of Access: every
+row came from the extract, so wiping and rebuilding lost nothing. It stopped
+being harmless when the platform started minting rows Access has never heard
+of — portal NUMids from the allocator (apps/api/src/customers/numid.service.ts),
+customers created in the staff UI, OCR-captured policies, app-booked ledger
+rows, uploaded documents. None of those come back.
+
+`--sync` already avoids all of it: it upserts legacy rows against the existing
+refs and leaves everything else alone. So this guard does not try to teach the
+full path to preserve anything — it stops the full path when there is something
+to preserve, and points at the additive one.
+
+    python native_guard.py --env prod          # report only, exit 3 if blocking
+    python run_all.py --env prod --stage       # runs this first, refuses on 3
+    python run_all.py --env prod --force-full  # ignore the guard (deletes them)
+"""
+
+from __future__ import annotations
+
+import argparse
+import re
+from pathlib import Path
+
+import pandas as pd
+
+from dbenv import connect
+
+STG = Path(__file__).parent / "output"
+
+# Exit code the orchestrator looks for. Distinct from 1 so a connection failure
+# or a bad query is not silently read as "native rows found".
+BLOCKED = 3
+
+# transform_customers.py mints these when a legacy row carries no id of its own.
+# They are regenerated by every full pass, so they are legacy-owned, not native.
+SYNTHETIC_REF_PREFIXES = ("rownum_", "insrow_")
+
+# App-uploaded documents are stored as `<prefix>/<parent>/<uuid>.<ext>`, while
+# blob_extract writes `<prefix>/<parent>/<stagedtable>_<row>_<col>.<ext>`.
+# service_documents carries no provenance column, so the key shape is the only
+# signal available — approximate, and reported as such. Matched in MySQL rather
+# than in Python so the whole scan stays one round trip per table.
+UUID_KEY_SQL = "/[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}\\."
+
+
+def staged_legacy_ids() -> dict[str, set[str]] | None:
+    """The ids Access will re-create, per source system.
+
+    None when the staged Parquet is absent — which is not the same as "no legacy
+    ids". Returning an empty set there would mark all 1,171 refs as native and
+    block every run; returning None lets the caller say "cannot verify" instead.
+    """
+    sources = {"utilities": "stg_utilities", "insurance": "stg_seguros"}
+    out: dict[str, set[str]] = {}
+    for system, folder in sources.items():
+        path = STG / folder / "datgral.parquet"
+        if not path.exists():
+            return None
+        df = pd.read_parquet(path, columns=["num_id"])
+        ids = set()
+        for v in df["num_id"].astype("string"):
+            if v is None or pd.isna(v):
+                continue
+            v = str(v).strip()
+            if v.endswith(".0"):  # some numeric ids serialize as "521.0"
+                v = v[:-2]
+            if v and v != "0":
+                ids.add(v)
+        out[system] = ids
+    return out
+
+
+def native_refs(cur, staged: dict[str, set[str]] | None) -> tuple[int, list[str]]:
+    """Legacy refs with no counterpart in the Access extract.
+
+    This is the check that catches an allocated portal NUMid: the customer holds
+    a perfectly ordinary-looking (utilities, DATGRAL, '1172') ref, so "customer
+    has no refs" does not see it. Only comparing against staging does.
+    """
+    cur.execute(
+        "SELECT sourceSystem, legacyId FROM customer_legacy_refs ORDER BY sourceSystem, legacyId"
+    )
+    rows = cur.fetchall()
+    if staged is None:
+        return 0, []
+
+    found = []
+    for system, legacy_id in rows:
+        if legacy_id.startswith(SYNTHETIC_REF_PREFIXES):
+            continue
+        known = staged.get(system)
+        # An unknown source system has no extract to compare against, so it
+        # cannot be re-created either — treat it as native rather than ignoring.
+        if known is None or legacy_id not in known:
+            found.append(f"{system}/{legacy_id}")
+    return len(found), found
+
+
+def scan(conn) -> tuple[list[tuple[str, int, str]], bool]:
+    """(label, count, detail) per source of native rows, plus whether staging
+    was available to verify the refs."""
+    cur = conn.cursor()
+    staged = staged_legacy_ids()
+    findings: list[tuple[str, int, str]] = []
+
+    ref_count, ref_examples = native_refs(cur, staged)
+    if ref_count:
+        shown = ", ".join(ref_examples[:8])
+        more = f" (+{ref_count - 8} more)" if ref_count > 8 else ""
+        findings.append(("customer_legacy_refs", ref_count, f"{shown}{more}"))
+
+    cur.execute(
+        "SELECT COUNT(*) FROM customers c"
+        " WHERE NOT EXISTS (SELECT 1 FROM customer_legacy_refs r WHERE r.customerId = c.id)"
+    )
+    n = cur.fetchone()[0]
+    if n:
+        findings.append(("customers", n, "created in the staff UI, no legacy ref"))
+
+    # Every transform writes legacyId on what it loads, so a NULL is the app's.
+    for table, detail in (
+        ("transactions", "booked in the app (captura, OCR, manual)"),
+        ("policies", "created in the app or captured by policy OCR"),
+        ("properties", "created in the app"),
+        ("vehicles", "created in the app"),
+        ("bank_transactions", "booked in the chequera"),
+    ):
+        cur.execute(f"SELECT COUNT(*) FROM {table} WHERE legacyId IS NULL")
+        n = cur.fetchone()[0]
+        if n:
+            findings.append((table, n, detail))
+
+    # blob_extract always writes originalColumn; the app never does.
+    cur.execute("SELECT COUNT(*) FROM policy_documents WHERE originalColumn IS NULL")
+    n = cur.fetchone()[0]
+    if n:
+        findings.append(("policy_documents", n, "uploaded in the app"))
+
+    cur.execute("SELECT COUNT(*) FROM service_documents WHERE storageKey REGEXP %s",
+                (UUID_KEY_SQL,))
+    n = cur.fetchone()[0]
+    if n:
+        findings.append(("service_documents", n, "uploaded in the app (key shape, approximate)"))
+
+    return findings, staged is not None
+
+
+def report(findings, staged_ok: bool, env: str) -> int:
+    print(f"=== Verificación de datos nativos (env={env}) ===", flush=True)
+
+    if not staged_ok:
+        print(
+            "  ! No hay Parquet en migration/output, así que no se pueden verificar\n"
+            "    los refs contra el extracto de Access. Ejecute con --stage.",
+            flush=True,
+        )
+
+    if not findings:
+        print("  Sin filas nativas. Una reimportación completa no destruye nada.", flush=True)
+        return 0 if staged_ok else BLOCKED
+
+    total = sum(n for _, n, _ in findings)
+    print(f"  {total} filas existen SÓLO en la plataforma y se perderían:", flush=True)
+    for label, n, detail in findings:
+        print(f"    {n:>7}  {label:<22} {detail}", flush=True)
+    print(
+        "\n  Una reimportación completa vacía estas tablas y las reconstruye desde\n"
+        "  Access, que no conoce ninguna de estas filas.\n"
+        "\n  Use la sincronización aditiva (run_all.py --sync), que respeta lo\n"
+        "  capturado en la plataforma. Para reimportar de todos modos y BORRARLAS,\n"
+        "  ejecute run_all.py --force-full.",
+        flush=True,
+    )
+    return BLOCKED
+
+
+def main() -> None:
+    ap = argparse.ArgumentParser(
+        description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
+    )
+    ap.add_argument("--env", default="dev")
+    args = ap.parse_args()
+
+    conn = connect(args.env)
+    try:
+        findings, staged_ok = scan(conn)
+    finally:
+        conn.close()
+
+    raise SystemExit(report(findings, staged_ok, args.env))
+
+
+if __name__ == "__main__":
+    main()

migration/run_all.py

@@ -15,6 +15,12 @@ Then:
   ./.venv/bin/python run_all.py --env dev            # data only (staging already present)
   ./.venv/bin/python run_all.py --env prod --stage   # re-extract from Access first, then load
 
+A full pass truncates and rebuilds every table it owns from the Access extract,
+so anything the platform minted itself — allocated portal NUMids, customers
+created in the staff UI, OCR-captured policies, app-booked ledger rows, uploaded
+documents — is destroyed. native_guard.py runs first and refuses when the target
+database holds any of it; --force-full overrides and deletes them.
+
 --sync swaps the truncate+rebuild steps for the additive upsert ones. It reads
 the same staged Parquet, so it needs --stage too unless a previous run left
 migration/output populated on this machine — which is never true in a
@@ -83,6 +89,33 @@ SYNC_STEPS = [
 ]
 
 
+# native_guard.py exits with this when the target database holds rows that only
+# exist in the platform. Kept in step with the constant there.
+GUARD_BLOCKED = 3
+
+
+def guard(env: str, force: bool) -> None:
+    """Stop a full pass that would delete platform-native rows.
+
+    Only the full path needs this: --sync upserts legacy rows against the
+    existing refs and leaves everything else alone, so it cannot lose anything.
+    Run after staging, because the guard verifies legacy refs against the staged
+    Parquet and cannot tell an allocated NUMid from an Access one without it.
+    """
+    cmd = [PY, str(HERE / "native_guard.py"), "--env", env]
+    print("+ " + " ".join(cmd), flush=True)
+    r = subprocess.run(cmd)
+    if r.returncode == GUARD_BLOCKED and not force:
+        sys.exit(r.returncode)
+    if r.returncode == GUARD_BLOCKED and force:
+        print(
+            "\n! --force-full: continuando y BORRANDO las filas nativas listadas.\n",
+            flush=True,
+        )
+    elif r.returncode:
+        sys.exit(r.returncode)
+
+
 def run(cmd: list[str], step: int | None = None, total: int | None = None) -> None:
     # The "[paso i/N] name" marker is a contract with the Operaciones screen,
     # which parses the last one to show progress. Emitting it here rather than
@@ -103,6 +136,8 @@ def main() -> None:
                     help="re-run the raw staging load first (needs the Access files + mdbtools)")
     ap.add_argument("--sync", action="store_true",
                     help="upsert legacy rows and archive removed legacy rows; preserve manual rows")
+    ap.add_argument("--force-full", action="store_true",
+                    help="run the full truncate+rebuild even when it deletes platform-native rows")
     args = ap.parse_args()
 
     steps = SYNC_STEPS if args.sync else STEPS
@@ -116,6 +151,11 @@ def main() -> None:
         run([PY, str(HERE / "load_staging.py"), "--output-dir", str(HERE / "output")],
             step=1, total=total)
 
+    # Deliberately not counted as a step: it is a precondition, it takes a
+    # second, and the Operaciones progress bar parses those numbers.
+    if not args.sync:
+        guard(args.env, args.force_full)
+
     for i, step in enumerate(steps, start=1 + offset):
         cmd = [PY, str(HERE / step), "--env", args.env]
         if args.sync:

scripts/numid-audit.sql

@@ -56,8 +56,9 @@ pool AS (
 )
 SELECT
   p.numid, p.cid AS customerUuid, p.name, p.archived, p.hasEmail,
-  -- EXISTS, not a join: one customer can hold several insurance refs (DATGRAL and
-  -- COBRO3 both), and joining them fans this result out past one row per NUMid.
+  -- EXISTS, not a join: 16 customers hold more than one insurance ref (several
+  -- insurance rows folded into one customer), and joining them fans this result
+  -- out past one row per NUMid — 1,188 rows for a 1,171-id pool.
   EXISTS(SELECT 1 FROM customer_legacy_refs i
          WHERE i.customerId=p.cid AND i.sourceSystem='insurance') AS insRef,