diff --git a/keepercommander/commands/pam/vault_target.py b/keepercommander/commands/pam/vault_target.py
index dae5e96cb..821a87291 100644
--- a/keepercommander/commands/pam/vault_target.py
+++ b/keepercommander/commands/pam/vault_target.py
@@ -581,6 +581,9 @@ def update_pam_record(params, record, command='pam', force_nsf=False):
         sync_down_preserving_nsf_keys(params)
     else:
         record_management.update_record(params, record)
+        # Defer vault refresh so a second classic edit in the same session does
+        # not send a stale record_cache revision (no immediate sync_down here).
+        params.sync_data = True
 
 
 def execute_record_add_in_folder(params, args, folder_uid, command='pam'):
diff --git a/keepercommander/commands/tunnel_and_connections.py b/keepercommander/commands/tunnel_and_connections.py
index 32cb55b9a..f3dab0117 100644
--- a/keepercommander/commands/tunnel_and_connections.py
+++ b/keepercommander/commands/tunnel_and_connections.py
@@ -45,7 +45,7 @@
     stop_tunnel_process,
     unregister_tunnel,
 )
-from .. import api, vault, record_management
+from .. import api, vault
 from ..display import bcolors
 from ..error import CommandError
 from ..params import LAST_RECORD_UID
@@ -2983,8 +2983,7 @@ def _get_effective_protocol():
                         logging.debug(f'security is already {target_sec} on record={record_uid}')
 
             if dirty:
-                record_management.update_record(params, record)
-                api.sync_down(params)
+                update_pam_record(params, record, command='pam connection edit')
 
                 traffic_encryption_key = record.get_typed_field('trafficEncryptionSeed')
                 if not traffic_encryption_key:
@@ -4055,8 +4054,7 @@ def update_connection_choice(field_name, value):
             update_connection_choice('sessionPersistence', session_persistence)
 
         if dirty:
-            record_management.update_record(params, record)
-            api.sync_down(params)
+            update_pam_record(params, record, command='pam rbi edit')
 
             traffic_encryption_key = record.get_typed_field('trafficEncryptionSeed')
             if not traffic_encryption_key:
@@ -4235,7 +4233,7 @@ def execute(self, params, **kwargs):
                     pam_settings = vault.TypedField.new_field('pamSettings', "", "")
                     record.fields.append(pam_settings)
 
-                record_management.update_record(params, record)
+                update_pam_record(params, record, command='pam-split')
                 params.sync_data = True
 
                 print(f"{bcolors.WARNING}Record {record_uid} has no data to split and "
@@ -4291,7 +4289,7 @@ def execute(self, params, **kwargs):
             pam_settings = vault.TypedField.new_field('pamSettings', "", "")
             record.fields.append(pam_settings)
 
-        record_management.update_record(params, record)
+        update_pam_record(params, record, command='pam-split')
         params.sync_data = True
 
         if pam_config_uid:
diff --git a/keepercommander/nested_share_folder/__init__.py b/keepercommander/nested_share_folder/__init__.py
index 14c005919..3894b474b 100644
--- a/keepercommander/nested_share_folder/__init__.py
+++ b/keepercommander/nested_share_folder/__init__.py
@@ -19,7 +19,8 @@
     ],
     'common': [
         'get_folder_key', 'get_record_key', 'get_user_public_key',
-        'get_record_from_cache', 'parse_sharing_status', 'get_record_key_type',
+        'get_record_from_cache', 'get_record_revision', 'patch_record_revision',
+        'parse_sharing_status', 'get_record_key_type',
         'encrypt_record_key_for_folder', 'encrypt_for_recipient',
         'handle_share_invite', 'resolve_user_uid_bytes',
         'load_user_public_key', 'parse_folder_access_result',
diff --git a/keepercommander/nested_share_folder/common.py b/keepercommander/nested_share_folder/common.py
index 261f47651..790a08d36 100644
--- a/keepercommander/nested_share_folder/common.py
+++ b/keepercommander/nested_share_folder/common.py
@@ -56,6 +56,32 @@ def get_record_from_cache(params, record_uid: str) -> Optional[dict]:
     return None
 
 
+def get_record_revision(params, record_uid: str, default: int = 0) -> int:
+    """Return the highest known revision across NSF and classic caches.
+
+    NSF metadata and classic vault sync can diverge; always prefer the
+    freshest revision when sending optimistic-concurrency updates.
+    """
+    revisions = []
+    for attr in ('nested_share_records', 'record_cache'):
+        cache = getattr(params, attr, None) or {}
+        if record_uid in cache:
+            rev = cache[record_uid].get('revision')
+            if rev is not None:
+                revisions.append(rev)
+    return max(revisions) if revisions else default
+
+
+def patch_record_revision(params, record_uid: str, revision: int) -> None:
+    """Write *revision* into both NSF and classic caches when the UID is present."""
+    if not revision:
+        return
+    for attr in ('nested_share_records', 'record_cache'):
+        cache = getattr(params, attr, None)
+        if cache and record_uid in cache:
+            cache[record_uid]['revision'] = revision
+
+
 def get_record_key_type(params, record_uid: str) -> Optional[int]:
     """Return the record key type if available (legacy AES-CBC vs AES-GCM)."""
     meta = getattr(params, 'meta_data_cache', {}).get(record_uid)
diff --git a/keepercommander/nested_share_folder/record_api.py b/keepercommander/nested_share_folder/record_api.py
index 1ac17c2fc..d5239e1ed 100644
--- a/keepercommander/nested_share_folder/record_api.py
+++ b/keepercommander/nested_share_folder/record_api.py
@@ -15,6 +15,7 @@
 
 from .common import (
     get_folder_key, get_record_key, get_record_from_cache,
+    get_record_revision, patch_record_revision,
     get_user_public_key, encrypt_for_recipient, handle_share_invite,
     parse_sharing_status,
 )
@@ -173,62 +174,116 @@ def _load_existing_record_data(params, record_uid, rec=None):
     return None
 
 
+def _apply_record_update_overrides(data, title=None, record_type=None, fields=None, notes=None):
+    """Apply optional title/type/fields/notes overrides onto a record data dict."""
+    if title is not None:
+        data['title'] = title
+    if record_type is not None:
+        data['type'] = record_type
+    if fields is not None:
+        by_type = {}
+        for ef in data.get('fields', []):
+            by_type.setdefault(ef.get('type'), []).append(ef)
+        for ft, fv in fields.items():
+            fv = fv if isinstance(fv, list) else [fv]
+            if ft in by_type and by_type[ft]:
+                by_type[ft][0]['value'] = fv
+            else:
+                data.setdefault('fields', []).append({'type': ft, 'value': fv})
+    if notes is not None:
+        data['notes'] = notes
+    return data
+
+
+def _sync_down_for_nsf_update(params):
+    """Sync while preserving NSF folder keys when possible."""
+    try:
+        from ..commands.pam_import.nsf_helpers import sync_down_preserving_nsf_keys
+        sync_down_preserving_nsf_keys(params)
+    except Exception:
+        from .. import sync_down as sync_down_mod
+        sync_down_mod.sync_down(params)
+
+
 def update_record_v3(params, record_uid, data=None, title=None,
                      record_type=None, fields=None, notes=None,
                      non_shared_data=None, revision=None):
+    """Update an NSF/classic-cached record via vault/records/v3/update.
+
+    On ``RS_OUT_OF_SYNC`` with no explicit *revision*, syncs once and retries.
+    When *data* was omitted, the retry rebuilds payload from refreshed caches
+    so a concurrent content edit is not silently overwritten. When the caller
+    supplied *data* (e.g. ``update_pam_record``), the retry re-sends that
+    payload at the freshest revision (force / last-write-wins for that call).
+    """
     rec = get_record_from_cache(params, record_uid)
     if not rec:
-        from .. import sync_down
-        sync_down.sync_down(params)
+        _sync_down_for_nsf_update(params)
         rec = get_record_from_cache(params, record_uid)
     if not rec:
         raise ValueError(f"Record {record_uid} not found")
 
     rk = rec.get('record_key_unencrypted') or get_record_key(params, record_uid)
+    caller_supplied_data = data is not None
 
     if data is None:
         existing = _load_existing_record_data(params, record_uid, rec)
         data = existing.copy() if existing else {'fields': []}
-        if title is not None:
-            data['title'] = title
-        if record_type is not None:
-            data['type'] = record_type
-        if fields is not None:
-            by_type = {}
-            for ef in data.get('fields', []):
-                by_type.setdefault(ef.get('type'), []).append(ef)
-            for ft, fv in fields.items():
-                fv = fv if isinstance(fv, list) else [fv]
-                if ft in by_type and by_type[ft]:
-                    by_type[ft][0]['value'] = fv
-                else:
-                    data.setdefault('fields', []).append({'type': ft, 'value': fv})
-        if notes is not None:
-            data['notes'] = notes
-
-    ru = record_pb2.RecordUpdate()
-    ru.record_uid = utils.base64_url_decode(record_uid)
-    ru.client_modified_time = utils.current_milli_time()
-    ru.revision = revision if revision is not None else rec.get('revision', 0)
-
-    dj = pad_aes_gcm(json.dumps(data))
-    db = dj.encode() if isinstance(dj, str) else dj
-    ru.data = crypto.encrypt_aes_v2(db, rk)
-
-    if non_shared_data:
-        nsj = pad_aes_gcm(json.dumps(non_shared_data))
-        nsb = nsj.encode() if isinstance(nsj, str) else nsj
-        ru.non_shared_data = crypto.encrypt_aes_v2(nsb, rk)
-
+        _apply_record_update_overrides(data, title, record_type, fields, notes)
+
+    def _build_update(payload, rev):
+        ru = record_pb2.RecordUpdate()
+        ru.record_uid = utils.base64_url_decode(record_uid)
+        ru.client_modified_time = utils.current_milli_time()
+        # Prefer the highest known revision: NSF metadata can lag classic sync_down.
+        ru.revision = (rev if rev is not None
+                       else get_record_revision(params, record_uid, rec.get('revision', 0)))
+        dj = pad_aes_gcm(json.dumps(payload))
+        db = dj.encode() if isinstance(dj, str) else dj
+        ru.data = crypto.encrypt_aes_v2(db, rk)
+        if non_shared_data:
+            nsj = pad_aes_gcm(json.dumps(non_shared_data))
+            nsb = nsj.encode() if isinstance(nsj, str) else nsj
+            ru.non_shared_data = crypto.encrypt_aes_v2(nsb, rk)
+        return ru
+
+    ru = _build_update(data, revision)
     response = record_update_v3(params, [ru])
     if response.records:
         r = response.records[0]
+        # After PAM/classic edits, sync can leave NSF metadata lagging. Refresh
+        # once and retry with the freshest known revision.
+        if r.status == record_pb2.RS_OUT_OF_SYNC and revision is None:
+            _sync_down_for_nsf_update(params)
+            rec = get_record_from_cache(params, record_uid) or rec
+            rk = rec.get('record_key_unencrypted') or rk
+            if caller_supplied_data:
+                logging.warning(
+                    'NSF record update retry for %s after RS_OUT_OF_SYNC may overwrite '
+                    'a concurrent content edit (caller-supplied data).',
+                    record_uid,
+                )
+                retry_data = data
+            else:
+                existing = _load_existing_record_data(params, record_uid, rec)
+                retry_data = existing.copy() if existing else {'fields': []}
+                _apply_record_update_overrides(
+                    retry_data, title, record_type, fields, notes)
+            ru = _build_update(retry_data, None)
+            response = record_update_v3(params, [ru])
+            if not response.records:
+                raise KeeperApiError('no_results', 'No results from record update')
+            r = response.records[0]
+        success = r.status == record_pb2.RS_SUCCESS
+        new_revision = getattr(response, 'revision', 0)
+        if success:
+            patch_record_revision(params, record_uid, new_revision)
         return {
             'record_uid': record_uid,
             'status': record_pb2.RecordModifyResult.Name(r.status),
             'message': r.message,
-            'success': r.status == record_pb2.RS_SUCCESS,
-            'revision': getattr(response, 'revision', 0),
+            'success': success,
+            'revision': new_revision,
         }
     raise KeeperApiError('no_results', 'No results from record update')
 
diff --git a/keepercommander/nested_share_folder/sync.py b/keepercommander/nested_share_folder/sync.py
index 69c884579..64922815b 100644
--- a/keepercommander/nested_share_folder/sync.py
+++ b/keepercommander/nested_share_folder/sync.py
@@ -386,9 +386,14 @@ def _process_records(params, records):
     """Store DriveRecord metadata (no encrypted content)."""
     for record in records:
         record_uid = utils.base64_url_encode(record.recordUid)
+        existing = params.nested_share_records.get(record_uid) or {}
+        # Classic vault updates can bump revision before NSF drive metadata
+        # catches up. Never allow a lagging DriveRecord to downgrade the cache.
+        incoming_rev = record.revision or 0
+        existing_rev = existing.get('revision', 0) or 0
         record_obj = {
             'record_uid': record_uid,
-            'revision': record.revision,
+            'revision': max(incoming_rev, existing_rev),
             'version': record.version,
             'shared': record.shared if record.shared else False,
             'client_modified_time': record.clientModifiedTime if record.clientModifiedTime else 0,
@@ -397,6 +402,9 @@ def _process_records(params, records):
             record_obj['file_size'] = record.fileSize
         if record.thumbnailSize:
             record_obj['thumbnail_size'] = record.thumbnailSize
+        # Preserve decrypted key material across metadata refreshes.
+        if 'record_key_unencrypted' in existing:
+            record_obj['record_key_unencrypted'] = existing['record_key_unencrypted']
         params.nested_share_records[record_uid] = record_obj
 
 
@@ -1136,9 +1144,32 @@ def _reconstruct_nested_share_folder_entities(params):
         if 'data_json' not in rd_obj:
             continue
 
+        classic = params.record_cache.get(record_uid) or {}
+        classic_rev = classic.get('revision', 0) or 0
+        nsf_rev = record_obj.get('revision', 0) or 0
+        # Prefer the freshest revision across classic sync and NSF metadata.
+        revision = max(classic_rev, nsf_rev)
+        if revision != nsf_rev:
+            record_obj['revision'] = revision
+
+        # Classic response.records payloads use encrypted 'data' and are not tagged
+        # source=nested_share_folder. Prefer that copy when it is at least as fresh
+        # so lagging keeperDriveData cannot roll content/revision backwards after PAM.
+        classic_from_vault = (
+            'data' in classic and classic.get('source') != 'nested_share_folder'
+        )
+        if classic_from_vault and classic_rev >= nsf_rev:
+            if 'record_key_unencrypted' not in classic:
+                classic['record_key_unencrypted'] = record_obj['record_key_unencrypted']
+            # Still backfill meta/owner caches — NSF is often the only source on
+            # fresh login, and this branch is taken whenever classic sync returns
+            # the record (e.g. after PAM edits).
+            _backfill_nsf_record_access_caches(params, record_uid, record_obj, rd_obj)
+            continue
+
         record_entry = {
             'record_uid': record_uid,
-            'revision': record_obj.get('revision', 0),
+            'revision': revision,
             'version': record_obj.get('version', 0),
             'shared': record_obj.get('shared', False),
             'record_key_unencrypted': record_obj['record_key_unencrypted'],
@@ -1149,24 +1180,28 @@ def _reconstruct_nested_share_folder_entities(params):
         }
 
         params.record_cache[record_uid] = record_entry
+        _backfill_nsf_record_access_caches(params, record_uid, record_obj, rd_obj)
 
-        if record_uid not in params.meta_data_cache:
-            meta_data = {
-                'record_uid': record_uid,
-                'record_key_unencrypted': record_obj['record_key_unencrypted'],
-                'can_share': True,
-                'can_edit': True,
-            }
-            if 'user_account_uid' in rd_obj:
-                meta_data['owner_account_uid'] = rd_obj['user_account_uid']
-                if rd_obj['user_account_uid'] in params.user_cache:
-                    meta_data['owner_username'] = params.user_cache[rd_obj['user_account_uid']]
-            params.meta_data_cache[record_uid] = meta_data
-
-        if record_uid not in params.record_owner_cache:
-            if 'user_account_uid' in rd_obj:
-                is_owner = (rd_obj['user_account_uid'] == utils.base64_url_encode(params.account_uid_bytes))
-                params.record_owner_cache[record_uid] = RecordOwner(
-                    is_owner,
-                    rd_obj['user_account_uid']
-                )
\ No newline at end of file
+
+def _backfill_nsf_record_access_caches(params, record_uid, record_obj, rd_obj):
+    """Populate meta_data_cache / record_owner_cache from NSF data when missing."""
+    if record_uid not in params.meta_data_cache:
+        meta_data = {
+            'record_uid': record_uid,
+            'record_key_unencrypted': record_obj['record_key_unencrypted'],
+            'can_share': True,
+            'can_edit': True,
+        }
+        if 'user_account_uid' in rd_obj:
+            meta_data['owner_account_uid'] = rd_obj['user_account_uid']
+            if rd_obj['user_account_uid'] in params.user_cache:
+                meta_data['owner_username'] = params.user_cache[rd_obj['user_account_uid']]
+        params.meta_data_cache[record_uid] = meta_data
+
+    if record_uid not in params.record_owner_cache:
+        if 'user_account_uid' in rd_obj:
+            is_owner = (rd_obj['user_account_uid'] == utils.base64_url_encode(params.account_uid_bytes))
+            params.record_owner_cache[record_uid] = RecordOwner(
+                is_owner,
+                rd_obj['user_account_uid']
+            )
\ No newline at end of file
diff --git a/keepercommander/sync_down.py b/keepercommander/sync_down.py
index 45fd3a510..fd1396475 100644
--- a/keepercommander/sync_down.py
+++ b/keepercommander/sync_down.py
@@ -287,7 +287,24 @@ def convert_record(r):
 
             for r in response.records:
                 record = convert_record(r)
-                params.record_cache[record['record_uid']] = record
+                record_uid = record['record_uid']
+                params.record_cache[record_uid] = record
+                # Classic vault updates bump revision without NSF keeperDriveData.
+                # Keep nested_share_records.revision aligned so NSF updates do not
+                # send a stale revision (RS_OUT_OF_SYNC / "This object no longer exists").
+                nsf_records = getattr(params, 'nested_share_records', None)
+                if nsf_records and record_uid in nsf_records:
+                    # Direct assignment is intentional: classic response.records is
+                    # vault-authoritative for this UID after a classic update. NSF
+                    # _process_records uses max() because keeperDriveData can lag;
+                    # here the classic stream is the fresher source of truth.
+                    nsf_rec = nsf_records[record_uid]
+                    nsf_rec['revision'] = record['revision']
+                    nsf_rec['version'] = record['version']
+                    # shared/client_modified_time follow the classic payload for the
+                    # same reason; NSF-only sharing state is refreshed from drive data.
+                    nsf_rec['shared'] = record['shared']
+                    nsf_rec['client_modified_time'] = record['client_modified_time']
 
         if len(response.nonSharedData) > 0:
             for nsd in response.nonSharedData:
diff --git a/unit-tests/pam/test_pam_connection_edit_scrollback.py b/unit-tests/pam/test_pam_connection_edit_scrollback.py
index 2ebd61573..090fe1ad5 100644
--- a/unit-tests/pam/test_pam_connection_edit_scrollback.py
+++ b/unit-tests/pam/test_pam_connection_edit_scrollback.py
@@ -325,7 +325,7 @@ def _mock_record(self, record_type='pamMachine', protocol='ssh'):
         return rec
 
     @mock.patch('keepercommander.commands.tunnel_and_connections.RecordMixin.resolve_single_record')
-    @mock.patch('keepercommander.commands.tunnel_and_connections.record_management.update_record')
+    @mock.patch('keepercommander.commands.tunnel_and_connections.update_pam_record')
     @mock.patch('keepercommander.commands.tunnel_and_connections.api.sync_down')
     @mock.patch('keepercommander.commands.tunnel_and_connections.get_keeper_tokens',
                 return_value=(b'st', b'tk', b'tr'))
@@ -344,7 +344,7 @@ def test_scrollback_alone_skips_dag(self, mock_tdag, mock_get_config_uid,
         mock_update.assert_called_once()
 
     @mock.patch('keepercommander.commands.tunnel_and_connections.RecordMixin.resolve_single_record')
-    @mock.patch('keepercommander.commands.tunnel_and_connections.record_management.update_record')
+    @mock.patch('keepercommander.commands.tunnel_and_connections.update_pam_record')
     @mock.patch('keepercommander.commands.tunnel_and_connections.api.sync_down')
     @mock.patch('keepercommander.commands.tunnel_and_connections.get_keeper_tokens',
                 return_value=(b'st', b'tk', b'tr'))
@@ -361,7 +361,7 @@ def test_key_events_alone_skips_dag(self, mock_tdag, mock_get_config_uid,
         mock_tdag.assert_not_called()
 
     @mock.patch('keepercommander.commands.tunnel_and_connections.RecordMixin.resolve_single_record')
-    @mock.patch('keepercommander.commands.tunnel_and_connections.record_management.update_record')
+    @mock.patch('keepercommander.commands.tunnel_and_connections.update_pam_record')
     @mock.patch('keepercommander.commands.tunnel_and_connections.api.sync_down')
     @mock.patch('keepercommander.commands.tunnel_and_connections.get_keeper_tokens',
                 return_value=(b'st', b'tk', b'tr'))
diff --git a/unit-tests/pam/test_pam_connection_edit_security.py b/unit-tests/pam/test_pam_connection_edit_security.py
index 50735c68e..d4dd332d7 100644
--- a/unit-tests/pam/test_pam_connection_edit_security.py
+++ b/unit-tests/pam/test_pam_connection_edit_security.py
@@ -220,7 +220,7 @@ def _mock_record(self, record_type='pamMachine', protocol='rdp', existing_connec
         return rec, ps_field
 
     @mock.patch('keepercommander.commands.tunnel_and_connections.RecordMixin.resolve_single_record')
-    @mock.patch('keepercommander.commands.tunnel_and_connections.record_management.update_record')
+    @mock.patch('keepercommander.commands.tunnel_and_connections.update_pam_record')
     @mock.patch('keepercommander.commands.tunnel_and_connections.api.sync_down')
     @mock.patch('keepercommander.commands.tunnel_and_connections.get_keeper_tokens',
                 return_value=(b'st', b'tk', b'tr'))
@@ -236,7 +236,7 @@ def test_ignore_server_cert_on_writes_true(self, mock_tdag, mock_get_config_uid,
         mock_update.assert_called_once()
 
     @mock.patch('keepercommander.commands.tunnel_and_connections.RecordMixin.resolve_single_record')
-    @mock.patch('keepercommander.commands.tunnel_and_connections.record_management.update_record')
+    @mock.patch('keepercommander.commands.tunnel_and_connections.update_pam_record')
     @mock.patch('keepercommander.commands.tunnel_and_connections.api.sync_down')
     @mock.patch('keepercommander.commands.tunnel_and_connections.get_keeper_tokens',
                 return_value=(b'st', b'tk', b'tr'))
@@ -251,7 +251,7 @@ def test_ignore_server_cert_off_writes_false(self, mock_tdag, mock_get_config_ui
         self.assertEqual(ps_field.value[0]['connection'].get('ignoreCert'), False)
 
     @mock.patch('keepercommander.commands.tunnel_and_connections.RecordMixin.resolve_single_record')
-    @mock.patch('keepercommander.commands.tunnel_and_connections.record_management.update_record')
+    @mock.patch('keepercommander.commands.tunnel_and_connections.update_pam_record')
     @mock.patch('keepercommander.commands.tunnel_and_connections.api.sync_down')
     @mock.patch('keepercommander.commands.tunnel_and_connections.get_keeper_tokens',
                 return_value=(b'st', b'tk', b'tr'))
@@ -266,7 +266,7 @@ def test_ignore_server_cert_default_removes_key(self, mock_tdag, mock_get_config
         self.assertNotIn('ignoreCert', ps_field.value[0]['connection'])
 
     @mock.patch('keepercommander.commands.tunnel_and_connections.RecordMixin.resolve_single_record')
-    @mock.patch('keepercommander.commands.tunnel_and_connections.record_management.update_record')
+    @mock.patch('keepercommander.commands.tunnel_and_connections.update_pam_record')
     @mock.patch('keepercommander.commands.tunnel_and_connections.api.sync_down')
     @mock.patch('keepercommander.commands.tunnel_and_connections.get_keeper_tokens',
                 return_value=(b'st', b'tk', b'tr'))
@@ -281,7 +281,7 @@ def test_security_mode_writes_lowercase_value(self, mock_tdag, mock_get_config_u
         self.assertEqual(ps_field.value[0]['connection'].get('security'), 'nla')
 
     @mock.patch('keepercommander.commands.tunnel_and_connections.RecordMixin.resolve_single_record')
-    @mock.patch('keepercommander.commands.tunnel_and_connections.record_management.update_record')
+    @mock.patch('keepercommander.commands.tunnel_and_connections.update_pam_record')
     @mock.patch('keepercommander.commands.tunnel_and_connections.api.sync_down')
     @mock.patch('keepercommander.commands.tunnel_and_connections.get_keeper_tokens',
                 return_value=(b'st', b'tk', b'tr'))
@@ -296,7 +296,7 @@ def test_security_mode_default_removes_key(self, mock_tdag, mock_get_config_uid,
         self.assertNotIn('security', ps_field.value[0]['connection'])
 
     @mock.patch('keepercommander.commands.tunnel_and_connections.RecordMixin.resolve_single_record')
-    @mock.patch('keepercommander.commands.tunnel_and_connections.record_management.update_record')
+    @mock.patch('keepercommander.commands.tunnel_and_connections.update_pam_record')
     @mock.patch('keepercommander.commands.tunnel_and_connections.api.sync_down')
     @mock.patch('keepercommander.commands.tunnel_and_connections.get_keeper_tokens',
                 return_value=(b'st', b'tk', b'tr'))
@@ -339,7 +339,7 @@ def _mock_record(self, record_type='pamMachine', protocol='rdp'):
         return rec
 
     @mock.patch('keepercommander.commands.tunnel_and_connections.RecordMixin.resolve_single_record')
-    @mock.patch('keepercommander.commands.tunnel_and_connections.record_management.update_record')
+    @mock.patch('keepercommander.commands.tunnel_and_connections.update_pam_record')
     @mock.patch('keepercommander.commands.tunnel_and_connections.api.sync_down')
     @mock.patch('keepercommander.commands.tunnel_and_connections.get_keeper_tokens',
                 return_value=(b'st', b'tk', b'tr'))
@@ -356,7 +356,7 @@ def test_ignore_server_cert_alone_skips_dag(self, mock_tdag, mock_get_config_uid
         mock_update.assert_called_once()
 
     @mock.patch('keepercommander.commands.tunnel_and_connections.RecordMixin.resolve_single_record')
-    @mock.patch('keepercommander.commands.tunnel_and_connections.record_management.update_record')
+    @mock.patch('keepercommander.commands.tunnel_and_connections.update_pam_record')
     @mock.patch('keepercommander.commands.tunnel_and_connections.api.sync_down')
     @mock.patch('keepercommander.commands.tunnel_and_connections.get_keeper_tokens',
                 return_value=(b'st', b'tk', b'tr'))
diff --git a/unit-tests/pam/test_pam_nsf_config.py b/unit-tests/pam/test_pam_nsf_config.py
index aa75e930c..09e80f50c 100644
--- a/unit-tests/pam/test_pam_nsf_config.py
+++ b/unit-tests/pam/test_pam_nsf_config.py
@@ -113,6 +113,20 @@ def test_create_record_in_folder_uses_v3_add_for_nsf_folder(self, _extract, mock
         self.assertEqual(record.record_uid, 'nsf_record_uid')
         _sync.assert_called_once_with(params)
 
+    @mock.patch('keepercommander.commands.pam.vault_target.record_management.update_record')
+    def test_update_pam_record_classic_sets_sync_data(self, mock_update):
+        from keepercommander.commands.pam.vault_target import update_pam_record
+
+        params = _make_params()
+        params.sync_data = False
+        record = vault.TypedRecord()
+        record.record_uid = 'classic_uid'
+
+        update_pam_record(params, record, command='pam connection edit')
+
+        mock_update.assert_called_once_with(params, record)
+        self.assertTrue(params.sync_data)
+
     def test_resolve_pam_folder_uid_finds_root_nsf_folder_by_name(self):
         params = _make_params()
         folder = NestedShareFolderNode()
diff --git a/unit-tests/pam/test_pam_rbi_edit.py b/unit-tests/pam/test_pam_rbi_edit.py
index d3a451b20..bb877bb9a 100644
--- a/unit-tests/pam/test_pam_rbi_edit.py
+++ b/unit-tests/pam/test_pam_rbi_edit.py
@@ -250,7 +250,7 @@ def test_no_param_raises_error_with_new_settings_check(self):
         self.assertIn('At least one parameter is required', str(context.exception))
 
     @mock.patch('keepercommander.commands.tunnel_and_connections.RecordMixin.resolve_single_record')
-    @mock.patch('keepercommander.commands.tunnel_and_connections.record_management.update_record')
+    @mock.patch('keepercommander.commands.tunnel_and_connections.update_pam_record')
     @mock.patch('keepercommander.commands.tunnel_and_connections.api.sync_down')
     def test_allow_url_navigation_on_sets_true(self, mock_sync, mock_update, mock_resolve):
         mock_resolve.return_value = self.mock_record
@@ -258,7 +258,7 @@ def test_allow_url_navigation_on_sets_true(self, mock_sync, mock_update, mock_re
         self.assertEqual(self.pam_settings['connection'].get('allowUrlManipulation'), True)
 
     @mock.patch('keepercommander.commands.tunnel_and_connections.RecordMixin.resolve_single_record')
-    @mock.patch('keepercommander.commands.tunnel_and_connections.record_management.update_record')
+    @mock.patch('keepercommander.commands.tunnel_and_connections.update_pam_record')
     @mock.patch('keepercommander.commands.tunnel_and_connections.api.sync_down')
     def test_allow_url_navigation_off_sets_false(self, mock_sync, mock_update, mock_resolve):
         mock_resolve.return_value = self.mock_record
@@ -266,7 +266,7 @@ def test_allow_url_navigation_off_sets_false(self, mock_sync, mock_update, mock_
         self.assertEqual(self.pam_settings['connection'].get('allowUrlManipulation'), False)
 
     @mock.patch('keepercommander.commands.tunnel_and_connections.RecordMixin.resolve_single_record')
-    @mock.patch('keepercommander.commands.tunnel_and_connections.record_management.update_record')
+    @mock.patch('keepercommander.commands.tunnel_and_connections.update_pam_record')
     @mock.patch('keepercommander.commands.tunnel_and_connections.api.sync_down')
     def test_allow_url_navigation_default_removes_field(self, mock_sync, mock_update, mock_resolve):
         mock_resolve.return_value = self.mock_record
@@ -275,7 +275,7 @@ def test_allow_url_navigation_default_removes_field(self, mock_sync, mock_update
         self.assertNotIn('allowUrlManipulation', self.pam_settings['connection'])
 
     @mock.patch('keepercommander.commands.tunnel_and_connections.RecordMixin.resolve_single_record')
-    @mock.patch('keepercommander.commands.tunnel_and_connections.record_management.update_record')
+    @mock.patch('keepercommander.commands.tunnel_and_connections.update_pam_record')
     @mock.patch('keepercommander.commands.tunnel_and_connections.api.sync_down')
     def test_ignore_server_cert_on_sets_true(self, mock_sync, mock_update, mock_resolve):
         mock_resolve.return_value = self.mock_record
@@ -283,7 +283,7 @@ def test_ignore_server_cert_on_sets_true(self, mock_sync, mock_update, mock_reso
         self.assertEqual(self.pam_settings['connection'].get('ignoreInitialSslCert'), True)
 
     @mock.patch('keepercommander.commands.tunnel_and_connections.RecordMixin.resolve_single_record')
-    @mock.patch('keepercommander.commands.tunnel_and_connections.record_management.update_record')
+    @mock.patch('keepercommander.commands.tunnel_and_connections.update_pam_record')
     @mock.patch('keepercommander.commands.tunnel_and_connections.api.sync_down')
     def test_allow_file_uploads_on_sets_true(self, mock_sync, mock_update, mock_resolve):
         mock_resolve.return_value = self.mock_record
@@ -291,7 +291,7 @@ def test_allow_file_uploads_on_sets_true(self, mock_sync, mock_update, mock_reso
         self.assertEqual(self.pam_settings['connection'].get('allowFileUploads'), True)
 
     @mock.patch('keepercommander.commands.tunnel_and_connections.RecordMixin.resolve_single_record')
-    @mock.patch('keepercommander.commands.tunnel_and_connections.record_management.update_record')
+    @mock.patch('keepercommander.commands.tunnel_and_connections.update_pam_record')
     @mock.patch('keepercommander.commands.tunnel_and_connections.api.sync_down')
     def test_allow_file_uploads_off_sets_false(self, mock_sync, mock_update, mock_resolve):
         mock_resolve.return_value = self.mock_record
@@ -299,7 +299,7 @@ def test_allow_file_uploads_off_sets_false(self, mock_sync, mock_update, mock_re
         self.assertEqual(self.pam_settings['connection'].get('allowFileUploads'), False)
 
     @mock.patch('keepercommander.commands.tunnel_and_connections.RecordMixin.resolve_single_record')
-    @mock.patch('keepercommander.commands.tunnel_and_connections.record_management.update_record')
+    @mock.patch('keepercommander.commands.tunnel_and_connections.update_pam_record')
     @mock.patch('keepercommander.commands.tunnel_and_connections.api.sync_down')
     def test_allow_file_uploads_default_removes_field(self, mock_sync, mock_update, mock_resolve):
         mock_resolve.return_value = self.mock_record
@@ -308,7 +308,7 @@ def test_allow_file_uploads_default_removes_field(self, mock_sync, mock_update,
         self.assertNotIn('allowFileUploads', self.pam_settings['connection'])
 
     @mock.patch('keepercommander.commands.tunnel_and_connections.RecordMixin.resolve_single_record')
-    @mock.patch('keepercommander.commands.tunnel_and_connections.record_management.update_record')
+    @mock.patch('keepercommander.commands.tunnel_and_connections.update_pam_record')
     @mock.patch('keepercommander.commands.tunnel_and_connections.api.sync_down')
     def test_allow_file_downloads_on_sets_true(self, mock_sync, mock_update, mock_resolve):
         mock_resolve.return_value = self.mock_record
@@ -316,7 +316,7 @@ def test_allow_file_downloads_on_sets_true(self, mock_sync, mock_update, mock_re
         self.assertEqual(self.pam_settings['connection'].get('allowFileDownloads'), True)
 
     @mock.patch('keepercommander.commands.tunnel_and_connections.RecordMixin.resolve_single_record')
-    @mock.patch('keepercommander.commands.tunnel_and_connections.record_management.update_record')
+    @mock.patch('keepercommander.commands.tunnel_and_connections.update_pam_record')
     @mock.patch('keepercommander.commands.tunnel_and_connections.api.sync_down')
     def test_allow_file_downloads_off_sets_false(self, mock_sync, mock_update, mock_resolve):
         mock_resolve.return_value = self.mock_record
@@ -324,7 +324,7 @@ def test_allow_file_downloads_off_sets_false(self, mock_sync, mock_update, mock_
         self.assertEqual(self.pam_settings['connection'].get('allowFileDownloads'), False)
 
     @mock.patch('keepercommander.commands.tunnel_and_connections.RecordMixin.resolve_single_record')
-    @mock.patch('keepercommander.commands.tunnel_and_connections.record_management.update_record')
+    @mock.patch('keepercommander.commands.tunnel_and_connections.update_pam_record')
     @mock.patch('keepercommander.commands.tunnel_and_connections.api.sync_down')
     def test_allow_file_downloads_default_removes_field(self, mock_sync, mock_update, mock_resolve):
         mock_resolve.return_value = self.mock_record
@@ -333,7 +333,7 @@ def test_allow_file_downloads_default_removes_field(self, mock_sync, mock_update
         self.assertNotIn('allowFileDownloads', self.pam_settings['connection'])
 
     @mock.patch('keepercommander.commands.tunnel_and_connections.RecordMixin.resolve_single_record')
-    @mock.patch('keepercommander.commands.tunnel_and_connections.record_management.update_record')
+    @mock.patch('keepercommander.commands.tunnel_and_connections.update_pam_record')
     @mock.patch('keepercommander.commands.tunnel_and_connections.api.sync_down')
     def test_allowed_urls_joins_with_newlines(self, mock_sync, mock_update, mock_resolve):
         mock_resolve.return_value = self.mock_record
@@ -341,7 +341,7 @@ def test_allowed_urls_joins_with_newlines(self, mock_sync, mock_update, mock_res
         self.assertEqual(self.pam_settings['connection'].get('allowedUrlPatterns'), '*.example.com\n*.test.com')
 
     @mock.patch('keepercommander.commands.tunnel_and_connections.RecordMixin.resolve_single_record')
-    @mock.patch('keepercommander.commands.tunnel_and_connections.record_management.update_record')
+    @mock.patch('keepercommander.commands.tunnel_and_connections.update_pam_record')
     @mock.patch('keepercommander.commands.tunnel_and_connections.api.sync_down')
     def test_allowed_resource_urls_joins_with_newlines(self, mock_sync, mock_update, mock_resolve):
         mock_resolve.return_value = self.mock_record
@@ -349,7 +349,7 @@ def test_allowed_resource_urls_joins_with_newlines(self, mock_sync, mock_update,
         self.assertEqual(self.pam_settings['connection'].get('allowedResourceUrlPatterns'), '*.cdn.example.com')
 
     @mock.patch('keepercommander.commands.tunnel_and_connections.RecordMixin.resolve_single_record')
-    @mock.patch('keepercommander.commands.tunnel_and_connections.record_management.update_record')
+    @mock.patch('keepercommander.commands.tunnel_and_connections.update_pam_record')
     @mock.patch('keepercommander.commands.tunnel_and_connections.api.sync_down')
     def test_autofill_targets_joins_with_newlines(self, mock_sync, mock_update, mock_resolve):
         mock_resolve.return_value = self.mock_record
@@ -357,7 +357,7 @@ def test_autofill_targets_joins_with_newlines(self, mock_sync, mock_update, mock
         self.assertEqual(self.pam_settings['connection'].get('autofillConfiguration'), '#username\n#password')
 
     @mock.patch('keepercommander.commands.tunnel_and_connections.RecordMixin.resolve_single_record')
-    @mock.patch('keepercommander.commands.tunnel_and_connections.record_management.update_record')
+    @mock.patch('keepercommander.commands.tunnel_and_connections.update_pam_record')
     @mock.patch('keepercommander.commands.tunnel_and_connections.api.sync_down')
     def test_session_persistence_sets_value(self, mock_sync, mock_update, mock_resolve):
         mock_resolve.return_value = self.mock_record
@@ -365,7 +365,7 @@ def test_session_persistence_sets_value(self, mock_sync, mock_update, mock_resol
         self.assertEqual(self.pam_settings['connection'].get('sessionPersistence'), 'user')
 
     @mock.patch('keepercommander.commands.tunnel_and_connections.RecordMixin.resolve_single_record')
-    @mock.patch('keepercommander.commands.tunnel_and_connections.record_management.update_record')
+    @mock.patch('keepercommander.commands.tunnel_and_connections.update_pam_record')
     @mock.patch('keepercommander.commands.tunnel_and_connections.api.sync_down')
     def test_session_persistence_none_sets_literal(self, mock_sync, mock_update, mock_resolve):
         # 'none' is a real enum value (no persistence), not a removal sentinel
@@ -374,7 +374,7 @@ def test_session_persistence_none_sets_literal(self, mock_sync, mock_update, moc
         self.assertEqual(self.pam_settings['connection'].get('sessionPersistence'), 'none')
 
     @mock.patch('keepercommander.commands.tunnel_and_connections.RecordMixin.resolve_single_record')
-    @mock.patch('keepercommander.commands.tunnel_and_connections.record_management.update_record')
+    @mock.patch('keepercommander.commands.tunnel_and_connections.update_pam_record')
     @mock.patch('keepercommander.commands.tunnel_and_connections.api.sync_down')
     def test_session_persistence_default_removes_field(self, mock_sync, mock_update, mock_resolve):
         mock_resolve.return_value = self.mock_record
@@ -383,7 +383,7 @@ def test_session_persistence_default_removes_field(self, mock_sync, mock_update,
         self.assertNotIn('sessionPersistence', self.pam_settings['connection'])
 
     @mock.patch('keepercommander.commands.tunnel_and_connections.RecordMixin.resolve_single_record')
-    @mock.patch('keepercommander.commands.tunnel_and_connections.record_management.update_record')
+    @mock.patch('keepercommander.commands.tunnel_and_connections.update_pam_record')
     @mock.patch('keepercommander.commands.tunnel_and_connections.api.sync_down')
     def test_session_persistence_default_removes_present_but_null(self, mock_sync, mock_update, mock_resolve):
         # A present-but-null value must still be removed (membership check, not None check)
@@ -409,7 +409,7 @@ def setUp(self):
         self.mock_params = mock.MagicMock()
 
     @mock.patch('keepercommander.commands.tunnel_and_connections.RecordMixin.resolve_single_record')
-    @mock.patch('keepercommander.commands.tunnel_and_connections.record_management.update_record')
+    @mock.patch('keepercommander.commands.tunnel_and_connections.update_pam_record')
     @mock.patch('keepercommander.commands.tunnel_and_connections.api.sync_down')
     def test_allow_copy_on_sets_disable_copy_false(self, mock_sync, mock_update, mock_resolve):
         mock_resolve.return_value = self.mock_record
@@ -417,7 +417,7 @@ def test_allow_copy_on_sets_disable_copy_false(self, mock_sync, mock_update, moc
         self.assertEqual(self.pam_settings['connection'].get('disableCopy'), False)
 
     @mock.patch('keepercommander.commands.tunnel_and_connections.RecordMixin.resolve_single_record')
-    @mock.patch('keepercommander.commands.tunnel_and_connections.record_management.update_record')
+    @mock.patch('keepercommander.commands.tunnel_and_connections.update_pam_record')
     @mock.patch('keepercommander.commands.tunnel_and_connections.api.sync_down')
     def test_allow_copy_off_sets_disable_copy_true(self, mock_sync, mock_update, mock_resolve):
         mock_resolve.return_value = self.mock_record
@@ -425,7 +425,7 @@ def test_allow_copy_off_sets_disable_copy_true(self, mock_sync, mock_update, moc
         self.assertEqual(self.pam_settings['connection'].get('disableCopy'), True)
 
     @mock.patch('keepercommander.commands.tunnel_and_connections.RecordMixin.resolve_single_record')
-    @mock.patch('keepercommander.commands.tunnel_and_connections.record_management.update_record')
+    @mock.patch('keepercommander.commands.tunnel_and_connections.update_pam_record')
     @mock.patch('keepercommander.commands.tunnel_and_connections.api.sync_down')
     def test_allow_copy_default_removes_field(self, mock_sync, mock_update, mock_resolve):
         mock_resolve.return_value = self.mock_record
@@ -434,7 +434,7 @@ def test_allow_copy_default_removes_field(self, mock_sync, mock_update, mock_res
         self.assertNotIn('disableCopy', self.pam_settings['connection'])
 
     @mock.patch('keepercommander.commands.tunnel_and_connections.RecordMixin.resolve_single_record')
-    @mock.patch('keepercommander.commands.tunnel_and_connections.record_management.update_record')
+    @mock.patch('keepercommander.commands.tunnel_and_connections.update_pam_record')
     @mock.patch('keepercommander.commands.tunnel_and_connections.api.sync_down')
     def test_allow_paste_on_sets_disable_paste_false(self, mock_sync, mock_update, mock_resolve):
         mock_resolve.return_value = self.mock_record
@@ -442,7 +442,7 @@ def test_allow_paste_on_sets_disable_paste_false(self, mock_sync, mock_update, m
         self.assertEqual(self.pam_settings['connection'].get('disablePaste'), False)
 
     @mock.patch('keepercommander.commands.tunnel_and_connections.RecordMixin.resolve_single_record')
-    @mock.patch('keepercommander.commands.tunnel_and_connections.record_management.update_record')
+    @mock.patch('keepercommander.commands.tunnel_and_connections.update_pam_record')
     @mock.patch('keepercommander.commands.tunnel_and_connections.api.sync_down')
     def test_allow_paste_off_sets_disable_paste_true(self, mock_sync, mock_update, mock_resolve):
         mock_resolve.return_value = self.mock_record
@@ -450,7 +450,7 @@ def test_allow_paste_off_sets_disable_paste_true(self, mock_sync, mock_update, m
         self.assertEqual(self.pam_settings['connection'].get('disablePaste'), True)
 
     @mock.patch('keepercommander.commands.tunnel_and_connections.RecordMixin.resolve_single_record')
-    @mock.patch('keepercommander.commands.tunnel_and_connections.record_management.update_record')
+    @mock.patch('keepercommander.commands.tunnel_and_connections.update_pam_record')
     @mock.patch('keepercommander.commands.tunnel_and_connections.api.sync_down')
     def test_clipboard_both_on(self, mock_sync, mock_update, mock_resolve):
         mock_resolve.return_value = self.mock_record
@@ -459,7 +459,7 @@ def test_clipboard_both_on(self, mock_sync, mock_update, mock_resolve):
         self.assertEqual(self.pam_settings['connection'].get('disablePaste'), False)
 
     @mock.patch('keepercommander.commands.tunnel_and_connections.RecordMixin.resolve_single_record')
-    @mock.patch('keepercommander.commands.tunnel_and_connections.record_management.update_record')
+    @mock.patch('keepercommander.commands.tunnel_and_connections.update_pam_record')
     @mock.patch('keepercommander.commands.tunnel_and_connections.api.sync_down')
     def test_clipboard_both_off(self, mock_sync, mock_update, mock_resolve):
         mock_resolve.return_value = self.mock_record
@@ -484,16 +484,17 @@ def setUp(self):
         self.mock_params = mock.MagicMock()
 
     @mock.patch('keepercommander.commands.tunnel_and_connections.RecordMixin.resolve_single_record')
-    @mock.patch('keepercommander.commands.tunnel_and_connections.record_management.update_record')
+    @mock.patch('keepercommander.commands.tunnel_and_connections.update_pam_record')
     @mock.patch('keepercommander.commands.tunnel_and_connections.api.sync_down')
     def test_record_update_called_when_field_changes(self, mock_sync, mock_update, mock_resolve):
         mock_resolve.return_value = self.mock_record
         self.command.execute(self.mock_params, record='test-record', allow_url_navigation='on')
         mock_update.assert_called_once()
-        mock_sync.assert_called_once()
+        # Sync (and deferred classic sync_data) is handled inside update_pam_record.
+        mock_sync.assert_not_called()
 
     @mock.patch('keepercommander.commands.tunnel_and_connections.RecordMixin.resolve_single_record')
-    @mock.patch('keepercommander.commands.tunnel_and_connections.record_management.update_record')
+    @mock.patch('keepercommander.commands.tunnel_and_connections.update_pam_record')
     @mock.patch('keepercommander.commands.tunnel_and_connections.api.sync_down')
     def test_multiple_fields_single_update(self, mock_sync, mock_update, mock_resolve):
         mock_resolve.return_value = self.mock_record
@@ -616,7 +617,7 @@ def test_disable_audio_default(self):
         self.assertEqual(args.disable_audio, 'default')
 
     @mock.patch('keepercommander.commands.tunnel_and_connections.RecordMixin.resolve_single_record')
-    @mock.patch('keepercommander.commands.tunnel_and_connections.record_management.update_record')
+    @mock.patch('keepercommander.commands.tunnel_and_connections.update_pam_record')
     @mock.patch('keepercommander.commands.tunnel_and_connections.api.sync_down')
     def test_disable_audio_on_sets_true(self, mock_sync, mock_update, mock_resolve):
         mock_resolve.return_value = self.mock_record
@@ -624,7 +625,7 @@ def test_disable_audio_on_sets_true(self, mock_sync, mock_update, mock_resolve):
         self.assertEqual(self.pam_settings['connection'].get('disableAudio'), True)
 
     @mock.patch('keepercommander.commands.tunnel_and_connections.RecordMixin.resolve_single_record')
-    @mock.patch('keepercommander.commands.tunnel_and_connections.record_management.update_record')
+    @mock.patch('keepercommander.commands.tunnel_and_connections.update_pam_record')
     @mock.patch('keepercommander.commands.tunnel_and_connections.api.sync_down')
     def test_disable_audio_off_sets_false(self, mock_sync, mock_update, mock_resolve):
         mock_resolve.return_value = self.mock_record
@@ -632,7 +633,7 @@ def test_disable_audio_off_sets_false(self, mock_sync, mock_update, mock_resolve
         self.assertEqual(self.pam_settings['connection'].get('disableAudio'), False)
 
     @mock.patch('keepercommander.commands.tunnel_and_connections.RecordMixin.resolve_single_record')
-    @mock.patch('keepercommander.commands.tunnel_and_connections.record_management.update_record')
+    @mock.patch('keepercommander.commands.tunnel_and_connections.update_pam_record')
     @mock.patch('keepercommander.commands.tunnel_and_connections.api.sync_down')
     def test_disable_audio_default_removes_field(self, mock_sync, mock_update, mock_resolve):
         mock_resolve.return_value = self.mock_record
@@ -649,7 +650,7 @@ def test_audio_channels_mono(self):
         self.assertEqual(args.audio_channels, 1)
 
     @mock.patch('keepercommander.commands.tunnel_and_connections.RecordMixin.resolve_single_record')
-    @mock.patch('keepercommander.commands.tunnel_and_connections.record_management.update_record')
+    @mock.patch('keepercommander.commands.tunnel_and_connections.update_pam_record')
     @mock.patch('keepercommander.commands.tunnel_and_connections.api.sync_down')
     def test_audio_channels_sets_field(self, mock_sync, mock_update, mock_resolve):
         mock_resolve.return_value = self.mock_record
@@ -669,7 +670,7 @@ def test_audio_bit_depth_invalid(self):
             self.parser.parse_args(['--record', 'test-record', '--audio-bit-depth', '24'])
 
     @mock.patch('keepercommander.commands.tunnel_and_connections.RecordMixin.resolve_single_record')
-    @mock.patch('keepercommander.commands.tunnel_and_connections.record_management.update_record')
+    @mock.patch('keepercommander.commands.tunnel_and_connections.update_pam_record')
     @mock.patch('keepercommander.commands.tunnel_and_connections.api.sync_down')
     def test_audio_bit_depth_sets_field(self, mock_sync, mock_update, mock_resolve):
         mock_resolve.return_value = self.mock_record
@@ -685,7 +686,7 @@ def test_audio_sample_rate_48000(self):
         self.assertEqual(args.audio_sample_rate, 48000)
 
     @mock.patch('keepercommander.commands.tunnel_and_connections.RecordMixin.resolve_single_record')
-    @mock.patch('keepercommander.commands.tunnel_and_connections.record_management.update_record')
+    @mock.patch('keepercommander.commands.tunnel_and_connections.update_pam_record')
     @mock.patch('keepercommander.commands.tunnel_and_connections.api.sync_down')
     def test_audio_sample_rate_sets_field(self, mock_sync, mock_update, mock_resolve):
         mock_resolve.return_value = self.mock_record
@@ -693,7 +694,7 @@ def test_audio_sample_rate_sets_field(self, mock_sync, mock_update, mock_resolve
         self.assertEqual(self.pam_settings['connection'].get('audioSampleRate'), 48000)
 
     @mock.patch('keepercommander.commands.tunnel_and_connections.RecordMixin.resolve_single_record')
-    @mock.patch('keepercommander.commands.tunnel_and_connections.record_management.update_record')
+    @mock.patch('keepercommander.commands.tunnel_and_connections.update_pam_record')
     @mock.patch('keepercommander.commands.tunnel_and_connections.api.sync_down')
     def test_all_audio_settings_combined(self, mock_sync, mock_update, mock_resolve):
         mock_resolve.return_value = self.mock_record
diff --git a/unit-tests/pam/test_pam_split_nsf.py b/unit-tests/pam/test_pam_split_nsf.py
index 7ba568c90..a37b74501 100644
--- a/unit-tests/pam/test_pam_split_nsf.py
+++ b/unit-tests/pam/test_pam_split_nsf.py
@@ -48,7 +48,7 @@ class TestPamSplitNsfPlacement(unittest.TestCase):
     @mock.patch('keepercommander.commands.tunnel_and_connections.get_keeper_tokens',
                 return_value=(b'encrypted_session', b'encrypted_key', b'transmission_key'))
     @mock.patch('keepercommander.commands.tunnel_and_connections.api.sync_down')
-    @mock.patch('keepercommander.commands.tunnel_and_connections.record_management.update_record')
+    @mock.patch('keepercommander.commands.tunnel_and_connections.update_pam_record')
     @mock.patch('keepercommander.commands.pam.vault_target.create_record_in_folder')
     @mock.patch('keepercommander.commands.pam.vault_target.resolve_pam_folder_uid',
                 return_value='nsf_folder')
@@ -83,7 +83,7 @@ def test_split_resolves_explicit_nsf_folder_name(
     @mock.patch('keepercommander.commands.tunnel_and_connections.get_keeper_tokens',
                 return_value=(b'encrypted_session', b'encrypted_key', b'transmission_key'))
     @mock.patch('keepercommander.commands.tunnel_and_connections.api.sync_down')
-    @mock.patch('keepercommander.commands.tunnel_and_connections.record_management.update_record')
+    @mock.patch('keepercommander.commands.tunnel_and_connections.update_pam_record')
     @mock.patch('keepercommander.commands.pam.vault_target.create_record_in_folder')
     @mock.patch('keepercommander.commands.tunnel_and_connections.vault.KeeperRecord.create')
     @mock.patch('keepercommander.commands.tunnel_and_connections.vault.KeeperRecord.load')
diff --git a/unit-tests/test_nested_share_folder.py b/unit-tests/test_nested_share_folder.py
index d48edcbf0..9bae3a371 100644
--- a/unit-tests/test_nested_share_folder.py
+++ b/unit-tests/test_nested_share_folder.py
@@ -1472,6 +1472,239 @@ def test_update_record_v3_preserves_login_fields_from_nsf_record_data(self, mock
         self.assertEqual(by_type['login'], ['bob2'])
         self.assertEqual(by_type['password'], ['KeepMe'])
 
+    @patch('keepercommander.nested_share_folder.record_api.record_update_v3')
+    def test_update_record_v3_patches_revision_in_both_caches(self, mock_update):
+        """Successful NSF update must refresh nested_share_records and record_cache revision."""
+        from keepercommander.nested_share_folder.record_api import update_record_v3
+        from keepercommander.proto import record_pb2
+
+        ruid, robj = _make_record()
+        robj['revision'] = 3
+        params = _make_params(
+            nested_share_records={ruid: robj},
+            record_cache={ruid: {
+                'revision': 3,
+                'record_key_unencrypted': robj['record_key_unencrypted'],
+                'data_unencrypted': json.dumps({
+                    'type': 'login', 'title': 'T', 'fields': [],
+                }).encode('utf-8'),
+            }},
+        )
+        mock_rs = Mock()
+        mock_rec = Mock()
+        mock_rec.status = record_pb2.RS_SUCCESS
+        mock_rec.message = ''
+        mock_rs.records = [mock_rec]
+        mock_rs.revision = 7
+        mock_update.return_value = mock_rs
+
+        result = update_record_v3(params, ruid, title='Updated')
+        self.assertTrue(result['success'])
+        self.assertEqual(result['revision'], 7)
+        self.assertEqual(params.nested_share_records[ruid]['revision'], 7)
+        self.assertEqual(params.record_cache[ruid]['revision'], 7)
+
+    @patch('keepercommander.nested_share_folder.record_api.record_update_v3')
+    def test_update_record_v3_uses_max_revision_across_caches(self, mock_update):
+        """Stale NSF metadata must not win over a newer classic record_cache revision."""
+        from keepercommander.nested_share_folder.record_api import update_record_v3
+        from keepercommander.proto import record_pb2
+
+        ruid, robj = _make_record()
+        robj['revision'] = 4  # stale NSF cache (preferred by get_record_from_cache)
+        params = _make_params(
+            nested_share_records={ruid: robj},
+            record_cache={ruid: {
+                'revision': 9,  # fresher after classic PAM edit / sync_down
+                'record_key_unencrypted': robj['record_key_unencrypted'],
+                'data_unencrypted': json.dumps({
+                    'type': 'login', 'title': 'T', 'fields': [],
+                }).encode('utf-8'),
+            }},
+        )
+        mock_rs = Mock()
+        mock_rec = Mock()
+        mock_rec.status = record_pb2.RS_SUCCESS
+        mock_rec.message = ''
+        mock_rs.records = [mock_rec]
+        mock_rs.revision = 10
+        mock_update.return_value = mock_rs
+
+        result = update_record_v3(params, ruid, title='Updated')
+        self.assertTrue(result['success'])
+
+        ru = mock_update.call_args[0][1][0]
+        self.assertEqual(ru.revision, 9)
+
+    def test_get_record_revision_prefers_max(self):
+        from keepercommander.nested_share_folder.common import get_record_revision
+
+        ruid = utils.generate_uid()
+        params = _make_params(
+            nested_share_records={ruid: {'revision': 2}},
+            record_cache={ruid: {'revision': 5}},
+        )
+        self.assertEqual(get_record_revision(params, ruid), 5)
+        self.assertEqual(get_record_revision(params, 'missing', default=1), 1)
+
+    @patch('keepercommander.nested_share_folder.record_api.record_update_v3')
+    @patch('keepercommander.nested_share_folder.record_api._sync_down_for_nsf_update')
+    def test_update_record_v3_retries_on_out_of_sync(self, mock_sync, mock_update):
+        """RS_OUT_OF_SYNC should sync once and retry with refreshed revision."""
+        from keepercommander.nested_share_folder.record_api import update_record_v3
+        from keepercommander.proto import record_pb2
+
+        ruid, robj = _make_record()
+        robj['revision'] = 4
+        params = _make_params(
+            nested_share_records={ruid: robj},
+            record_cache={ruid: {
+                'revision': 4,
+                'record_key_unencrypted': robj['record_key_unencrypted'],
+                'data_unencrypted': json.dumps({
+                    'type': 'login', 'title': 'T', 'fields': [],
+                }).encode('utf-8'),
+            }},
+        )
+
+        def _sync_side_effect(_params):
+            _params.nested_share_records[ruid]['revision'] = 9
+            _params.record_cache[ruid]['revision'] = 9
+            _params.record_cache[ruid]['data_unencrypted'] = json.dumps({
+                'type': 'login', 'title': 'FromSync', 'fields': [],
+            }).encode('utf-8')
+
+        mock_sync.side_effect = _sync_side_effect
+
+        stale = Mock()
+        stale.status = record_pb2.RS_OUT_OF_SYNC
+        stale.message = 'This object no longer exists.'
+        stale_rs = Mock()
+        stale_rs.records = [stale]
+        stale_rs.revision = 0
+
+        ok = Mock()
+        ok.status = record_pb2.RS_SUCCESS
+        ok.message = ''
+        ok_rs = Mock()
+        ok_rs.records = [ok]
+        ok_rs.revision = 10
+        sent = []
+
+        def _update_side_effect(params_arg, records):
+            ru = records[0]
+            decrypted = json.loads(
+                crypto.decrypt_aes_v2(ru.data, robj['record_key_unencrypted'])
+                .decode('utf-8').rstrip('\x00')
+            )
+            sent.append({'revision': ru.revision, 'title': decrypted.get('title')})
+            if len(sent) == 1:
+                return stale_rs
+            return ok_rs
+
+        mock_update.side_effect = _update_side_effect
+
+        result = update_record_v3(params, ruid, title='Updated')
+        self.assertTrue(result['success'])
+        self.assertEqual(sent[0]['revision'], 4)
+        self.assertEqual(sent[0]['title'], 'Updated')
+        # Retry rebuilds from refreshed cache then re-applies title override.
+        self.assertEqual(sent[1]['revision'], 9)
+        self.assertEqual(sent[1]['title'], 'Updated')
+        mock_sync.assert_called_once()
+        self.assertEqual(params.nested_share_records[ruid]['revision'], 10)
+
+    @patch('keepercommander.nested_share_folder.record_api.record_update_v3')
+    @patch('keepercommander.nested_share_folder.record_api._sync_down_for_nsf_update')
+    def test_update_record_v3_no_retry_when_revision_explicit(self, mock_sync, mock_update):
+        """Explicit revision means caller owns concurrency — do not auto-retry."""
+        from keepercommander.nested_share_folder.record_api import update_record_v3
+        from keepercommander.proto import record_pb2
+
+        ruid, robj = _make_record()
+        params = _make_params(
+            nested_share_records={ruid: robj},
+            record_cache={ruid: {
+                'revision': 4,
+                'record_key_unencrypted': robj['record_key_unencrypted'],
+                'data_unencrypted': json.dumps({
+                    'type': 'login', 'title': 'T', 'fields': [],
+                }).encode('utf-8'),
+            }},
+        )
+        stale = Mock()
+        stale.status = record_pb2.RS_OUT_OF_SYNC
+        stale.message = 'This object no longer exists.'
+        stale_rs = Mock()
+        stale_rs.records = [stale]
+        stale_rs.revision = 0
+        mock_update.return_value = stale_rs
+
+        result = update_record_v3(params, ruid, title='Updated', revision=4)
+        self.assertFalse(result['success'])
+        mock_sync.assert_not_called()
+        self.assertEqual(mock_update.call_count, 1)
+
+    def test_process_records_does_not_downgrade_revision(self):
+        from keepercommander.nested_share_folder.sync import _process_records
+        from types import SimpleNamespace
+
+        ruid, robj = _make_record()
+        robj['revision'] = 9
+        params = _make_params(nested_share_records={ruid: robj})
+        drive_rec = SimpleNamespace(
+            recordUid=utils.base64_url_decode(ruid),
+            revision=4,
+            version=3,
+            shared=False,
+            clientModifiedTime=0,
+            fileSize=0,
+            thumbnailSize=0,
+        )
+        _process_records(params, [drive_rec])
+        self.assertEqual(params.nested_share_records[ruid]['revision'], 9)
+        self.assertIn('record_key_unencrypted', params.nested_share_records[ruid])
+
+    def test_reconstruct_classic_preferred_still_backfills_meta_owner(self):
+        """Classic-preferred continue must still populate meta/owner caches."""
+        from keepercommander.nested_share_folder.sync import (
+            _reconstruct_nested_share_folder_entities,
+        )
+        from keepercommander.params import RecordOwner
+
+        ruid, robj = _make_record()
+        owner_uid = _ACCOUNT_UID
+        robj['revision'] = 5
+        params = _make_params(
+            nested_share_records={ruid: robj},
+            nested_share_record_data={ruid: {
+                'data_json': {'type': 'login', 'title': 'NSF', 'fields': []},
+                'user_account_uid': owner_uid,
+            }},
+            record_cache={ruid: {
+                'record_uid': ruid,
+                'revision': 9,
+                'data': 'encrypted-classic-payload',
+                'version': 3,
+            }},
+            meta_data_cache={},
+            record_owner_cache={},
+            user_cache={owner_uid: 'owner@example.com'},
+        )
+        params.account_uid_bytes = utils.base64_url_decode(_ACCOUNT_UID)
+
+        _reconstruct_nested_share_folder_entities(params)
+
+        # Keep classic vault payload; do not overwrite with NSF data_json.
+        self.assertEqual(params.record_cache[ruid].get('data'), 'encrypted-classic-payload')
+        self.assertNotIn('source', params.record_cache[ruid])
+        self.assertIn(ruid, params.meta_data_cache)
+        self.assertEqual(params.meta_data_cache[ruid]['can_edit'], True)
+        self.assertEqual(params.meta_data_cache[ruid]['owner_account_uid'], owner_uid)
+        self.assertIn(ruid, params.record_owner_cache)
+        self.assertIsInstance(params.record_owner_cache[ruid], RecordOwner)
+        self.assertEqual(params.nested_share_records[ruid]['revision'], 9)
+
     @patch('keepercommander.nested_share_folder.record_api.api.communicate_rest')
     @patch('keepercommander.nested_share_folder.record_api.encrypt_for_recipient')
     @patch('keepercommander.nested_share_folder.record_api.get_user_public_key')
diff --git a/unit-tests/test_sync_down.py b/unit-tests/test_sync_down.py
index 71f52adff..1d9f754fc 100644
--- a/unit-tests/test_sync_down.py
+++ b/unit-tests/test_sync_down.py
@@ -95,6 +95,39 @@ def test_sync_remove_team_shared_folder(self):
         self.assertEqual(len(params.team_cache), 0)
         self.assert_key_unencrypted(params)
 
+    def test_classic_records_update_nested_share_revision(self):
+        """Classic response.records must refresh nested_share_records.revision when present."""
+        params = get_synced_params()
+        record_uid = next(iter(params.record_cache))
+        params.nested_share_records = {
+            record_uid: {
+                'record_uid': record_uid,
+                'revision': 1,
+                'version': 3,
+                'shared': False,
+                'client_modified_time': 0,
+            }
+        }
+        new_revision = 42
+
+        with mock.patch('keepercommander.api.communicate_rest') as mock_comm:
+            rs = SyncDown_pb2.SyncDownResponse()
+            rs.continuationToken = crypto.get_random_bytes(64)
+            rec = rs.records.add()
+            rec.recordUid = utils.base64_url_decode(record_uid)
+            rec.revision = new_revision
+            rec.version = 3
+            rec.shared = True
+            rec.clientModifiedTime = 123456
+            rec.data = b'\x00' * 16
+            mock_comm.return_value = rs
+            sync_down(params)
+
+        self.assertEqual(params.record_cache[record_uid]['revision'], new_revision)
+        self.assertEqual(params.nested_share_records[record_uid]['revision'], new_revision)
+        self.assertEqual(params.nested_share_records[record_uid]['shared'], True)
+        self.assertEqual(params.nested_share_records[record_uid]['client_modified_time'], 123456)
+
     def assert_key_unencrypted(self, params):
         for r in params.record_cache.values():
             self.assertTrue('record_key_unencrypted' in r)