fix(migration): address Open SWE review findings on migrate_tables.py

- Validate the truncate backup on dry-run as well as --execute so a
  missing/stale/wrong-incarnation --backup-arn surfaces on the rehearsal
  run (finding f_24a48b8900).
- Assert configured keys match the live key schema of both tables before
  any key projection, turning config/schema drift into a descriptive
  abort instead of a mid-backfill KeyError (finding f_cb6b5a6c59).
- Clarify why key-existence spot-checks are sound for WorkOrderComments:
  the copy Puts source items verbatim and the sample uses the same
  cutoff filter, so per-account comment_id divergence never enters the
  check (finding f_390b7d6c3b is a false positive; comment hardened).

scripts/migrate_tables.py

@@ -33,7 +33,9 @@ Per-table modes (encode the plan's dual-delivery-safe semantics):
                                 Take an on-demand backup of the dest table and
                                 wait for it to be AVAILABLE before running
                                 (the script refuses without a fresh --backup-arn
-                                of THIS table).
+                                of THIS table, on DRY-RUN as well as --execute:
+                                the rehearsal validates the same preconditions
+                                the real run will).
 
 Usage:
     # copy (dry-run first, then --execute) -- run tables INDIVIDUALLY, in order
@@ -94,6 +96,21 @@ def session_table(profile, expected_account, table_name):
     return session.resource("dynamodb").Table(table_name)
 
 
+def check_key_schema(table, name, keys):
+    """Abort with a clear message if the configured keys drift from the live
+    table's key schema. Every DynamoDB item necessarily carries its table's
+    key attributes, so once this passes the `item[k]` key projections below
+    cannot KeyError; a mismatch here is the only way they could. Set (not
+    ordered) comparison is deliberate: keys are only ever used to build dict
+    Key= projections, which are order-insensitive."""
+    live = {k["AttributeName"] for k in table.key_schema}
+    if live != set(keys):
+        sys.exit(
+            f"ERROR: {name} live key schema {sorted(live)} != configured "
+            f"{sorted(keys)}; refusing to build keys from a stale TABLES entry."
+        )
+
+
 def scan_items(table, filter_kwargs=None):
     kwargs = dict(filter_kwargs or {})
     while True:
@@ -197,8 +214,9 @@ def _validate_truncate_backup(name, dst, args):
 
 
 def _truncate_destination(name, dst, args, keys):
-    if args.execute:
-        _validate_truncate_backup(name, dst, args)
+    # Validate in dry-run too: a stale/missing --backup-arn should surface on
+    # the rehearsal run, not first appear when the operator adds --execute.
+    _validate_truncate_backup(name, dst, args)
     dest_keys = [{k: item[k] for k in keys} for item in scan_items(dst)]
     print(f"[{name}] truncate: {len(dest_keys)} destination rows to delete")
     if args.execute:
@@ -213,6 +231,8 @@ def _truncate_destination(name, dst, args, keys):
 def copy_table(name, cfg, args):
     src = session_table(args.source_profile, SOURCE_ACCOUNT, name)
     dst = session_table(args.dest_profile, DEST_ACCOUNT, name)
+    check_key_schema(src, name, cfg["keys"])
+    check_key_schema(dst, name, cfg["keys"])
 
     filter_kwargs = None
     if cfg["mode"] == "cutoff":
@@ -259,6 +279,8 @@ def copy_table(name, cfg, args):
 def verify_table(name, cfg, args):
     src = session_table(args.source_profile, SOURCE_ACCOUNT, name)
     dst = session_table(args.dest_profile, DEST_ACCOUNT, name)
+    check_key_schema(src, name, cfg["keys"])
+    check_key_schema(dst, name, cfg["keys"])
 
     src_filter = None
     if cfg["mode"] == "cutoff":
@@ -279,9 +301,12 @@ def verify_table(name, cfg, args):
     print(f"[{name}] source={src_count} dest={dst_count} -> {status}")
 
     ok = relation
-    # Spot-check every mode. For cutoff tables the source scan is cutoff-filtered
-    # (so no dual-window rows are sampled); for truncate_load keys are copied
-    # verbatim, so key-existence is a sound check there too.
+    # Spot-check every mode. Key-existence lookups are sound for ALL modes
+    # because the copy batch-Puts source items VERBATIM (source comment_id
+    # included) -- per-account comment_id divergence only affects rows each
+    # pipeline ingests itself, and those dual-window rows are excluded here by
+    # the same cutoff filter the copy used. For truncate_load keys are likewise
+    # copied verbatim.
     if args.sample:
         sampled = mismatched = 0
         for item in scan_items(src, src_filter):
@@ -351,7 +376,8 @@ def main():
             p.add_argument("--execute", action="store_true")
             p.add_argument(
                 "--backup-arn",
-                help="required with --execute for truncate-and-load tables",
+                help="required for truncate-and-load tables (validated on "
+                "dry-run as well as --execute)",
             )
         else:
             p.add_argument(