diff --git a/config.ini.example b/config.ini.example
index 6f5d57d..9b69836 100644
--- a/config.ini.example
+++ b/config.ini.example
@@ -178,6 +178,15 @@ startup_advert = false
 # false: Manual mode - no automatic actions, use !repeater commands to manage contacts (default)
 auto_manage_contacts = bot
 
+# When auto_manage_contacts = device, the bot schedules one-shot jobs after connect:
+#   - Firmware: manual per-type contact adds + overwrite oldest non-favourite on full + auto-add chat (companions) only (bitmask 0x03)
+#   - Favourite hygiene: Admin_ACL and announcements ACL pubkeys are favourited on the radio; other favourites are cleared so eviction can recycle slots
+# Delays (seconds) before each step so the radio is not flooded at boot (defaults shown)
+#device_mode_firmware_delay_seconds = 30
+#device_mode_favourite_pass1_delay_seconds = 90
+#device_mode_favourite_pass2_delay_seconds = 180
+#contact_flag_update_spacing_ms = 200
+
 # Database path for main bot database
 # Default: meshcore_bot.db
 db_path = meshcore_bot.db
@@ -313,6 +322,7 @@ admin_commands = repeater,webviewer,reload,channelpause
 # Enable companion contact purging
 # true: Purge inactive companions when contact list is full
 # false: Never purge companions (default: false for safety)
+# When [Bot] auto_manage_contacts=device, companion auto-purge from the radio is always skipped (firmware slot policy).
 companion_purge_enabled = false
 
 # Days since last DM to consider companion inactive
diff --git a/docs/command-reference.md b/docs/command-reference.md
index 2dfebba..07b79b8 100644
--- a/docs/command-reference.md
+++ b/docs/command-reference.md
@@ -823,9 +823,9 @@ repeater stats
 - NEW_CONTACT events are automatically monitored
 - Repeaters are automatically cataloged when discovered
 - Contact list capacity is monitored in real-time
-- `auto_manage_contacts = device`: Device handles auto-addition, bot manages capacity
-- `auto_manage_contacts = bot`: Bot automatically adds companion contacts and manages capacity
-- `auto_manage_contacts = false`: Manual mode - use `!repeater` commands to manage contacts
+- `auto_manage_contacts = device`: Firmware auto-adds **chat (companion)** peers only, with **overwrite oldest non-favourite** when the contact table is full; the bot schedules delayed jobs to set that firmware policy and to **favourite** keys in `Admin_ACL` plus the effective announcements ACL (same rules as the announcements command), then clear **favourite** on other contacts. The bot still runs capacity management on NEW_CONTACT (near-limit `manage_contact_list`) and does **not** call `add_contact` for new companions itself. **Contact limit** for logging and capacity is taken from the radio’s `max_contacts` and, if the live table is larger (under-reported max), raised to match the mesh so counts are not shown as over-capacity. **Companion auto-purge** never runs on the radio in this mode. Count-based **repeater** auto-purge only runs if the table grows **strictly above** that synced limit (normally off while the firmware manages slots).
+- `auto_manage_contacts = bot`: Bot adds new companions via `add_contact` (full NEW_CONTACT payload), runs **manage-before-add** when the list is near limit, and **retries once** after `manage_contact_list` if the radio returns `TABLE_FULL`.
+- `auto_manage_contacts = false`: Manual mode - NEW_CONTACT companions are tracked in the database only; use `!repeater` commands to manage the device list.
 
 ---
 
diff --git a/modules/message_handler.py b/modules/message_handler.py
index 59148a0..86c5c72 100644
--- a/modules/message_handler.py
+++ b/modules/message_handler.py
@@ -3273,28 +3273,89 @@ class MessageHandler:
                     self.logger.info(f"✅ Repeater {contact_name} tracked in database - not added to device contacts")
                     return
                 else:
-                    # COMPANION: Track in database AND add to device contact list
-                    self.logger.info(f"👤 New companion discovered: {contact_name} - will be added to device contacts")
+                    # COMPANION: track in DB; device add behaviour depends on auto_manage_contacts
+                    auto_manage_setting = self.bot.config.get('Bot', 'auto_manage_contacts', fallback='false').lower()
+                    self.logger.info(
+                        "👤 New companion discovered: %s — auto_manage_contacts=%s",
+                        contact_name,
+                        auto_manage_setting,
+                    )
 
-                    # Track companion in complete database with signal info
-                    await self.bot.repeater_manager.track_contact_advertisement(contact_data, signal_info, packet_hash=packet_hash)
+                    await self.bot.repeater_manager.track_contact_advertisement(
+                        contact_data, signal_info, packet_hash=packet_hash
+                    )
 
-                    # Add companion to device contact list
-                    try:
-                        self._ensure_contact_meshcore_path_encoding(contact_data)
-                        result = await self.bot.meshcore.commands.add_contact(contact_data)
-                        if hasattr(result, 'type') and result.type.name == 'OK':
-                            self.logger.info(f"✅ Companion {contact_name} added to device contacts")
+                    if auto_manage_setting == 'false':
+                        self.logger.info(
+                            "Manual mode — companion %s tracked in database only (not added to device)",
+                            contact_name,
+                        )
+                    elif auto_manage_setting == 'device':
+                        self.logger.info(
+                            "Device mode — companion %s tracked; firmware handles addition; bot may manage capacity",
+                            contact_name,
+                        )
+                        status = await self.bot.repeater_manager.get_contact_list_status()
+                        if status and status.get('is_near_limit', False):
+                            self.logger.warning(
+                                "Contact list near limit (%.1f%%) — managing capacity",
+                                status['usage_percentage'],
+                            )
+                            await self.bot.repeater_manager.manage_contact_list(auto_cleanup=True)
                         else:
-                            self.logger.warning(f"❌ Failed to add companion {contact_name} to device: {result}")
-                    except Exception as e:
-                        self.logger.error(f"❌ Error adding companion {contact_name} to device: {e}")
+                            self.logger.info(
+                                "New companion %s — contact list has adequate space",
+                                contact_name,
+                            )
+                    elif auto_manage_setting == 'bot':
+                        self.logger.info(
+                            "Bot mode — adding companion %s to device with capacity management",
+                            contact_name,
+                        )
+                        try:
+                            self._ensure_contact_meshcore_path_encoding(contact_data)
+                            ok = await self.bot.repeater_manager.add_companion_from_contact_data(
+                                contact_data, contact_name, public_key
+                            )
+                            if not ok:
+                                self.logger.warning(
+                                    "Failed to add companion contact %s to device after managed add/retry",
+                                    contact_name,
+                                )
+                        except Exception as e:
+                            self.logger.error("Error adding companion %s to device: %s", contact_name, e)
+
+                        status = await self.bot.repeater_manager.get_contact_list_status()
+                        if status and status.get('is_near_limit', False):
+                            self.logger.warning(
+                                "Contact list near limit (%.1f%%) — managing capacity after add",
+                                status['usage_percentage'],
+                            )
+                            await self.bot.repeater_manager.manage_contact_list(auto_cleanup=True)
+                        else:
+                            self.logger.info(
+                                "Companion %s — contact list has adequate space after add attempt",
+                                contact_name,
+                            )
+                    else:
+                        self.logger.warning(
+                            "Unknown auto_manage_contacts value %r — treating as manual for %s",
+                            auto_manage_setting,
+                            contact_name,
+                        )
 
-                    # Check if auto-purge is needed
                     await self.bot.repeater_manager.check_and_auto_purge()
+
+                    self.bot.repeater_manager.db_manager.execute_update(
+                        'INSERT INTO purging_log (action, details) VALUES (?, ?)',
+                        (
+                            'new_contact_discovered',
+                            f'New contact discovered: {contact_name} (key: {public_key[:16]}...)',
+                        ),
+                    )
                     return
 
-            # Fallback: Track in database for unknown contact types
+            # Fallback: Track in database for unknown contact types (no repeater_manager)
             if hasattr(self.bot, 'repeater_manager'):
                 await self.bot.repeater_manager.track_contact_advertisement(contact_data, packet_hash=packet_hash)
                 await self.bot.repeater_manager.check_and_auto_purge()
diff --git a/modules/repeater_manager.py b/modules/repeater_manager.py
index 511f720..2d1410d 100644
--- a/modules/repeater_manager.py
+++ b/modules/repeater_manager.py
@@ -12,10 +12,47 @@ from typing import Any, Optional
 
 from meshcore import EventType
 
-from .security_utils import sanitize_name
+from .security_utils import sanitize_name, validate_pubkey_format
 from .utils import rate_limited_nominatim_reverse_sync
 
 
+def collect_protected_pubkeys_for_device_mode(config: Any, logger: Any) -> set[str]:
+    """Public keys that must stay favourited on the radio (admin + announcements ACL semantics).
+
+    Matches announcements_command._load_announcements_acl: explicit announcements_acl keys plus
+    all Admin_ACL admin_pubkeys (admins always included).
+    """
+    keys: list[str] = []
+    try:
+        if config.has_section('Announcements_Command'):
+            ann = config.get('Announcements_Command', 'announcements_acl', fallback='')
+            if ann and ann.strip():
+                for key in ann.split(','):
+                    key = key.strip()
+                    if not key:
+                        continue
+                    if validate_pubkey_format(key, expected_length=64):
+                        keys.append(key.lower())
+                    else:
+                        logger.warning('Invalid pubkey in announcements_acl: %s...', key[:16])
+        if config.has_section('Admin_ACL'):
+            admin_pubkeys = config.get('Admin_ACL', 'admin_pubkeys', fallback='')
+            if admin_pubkeys and admin_pubkeys.strip():
+                for key in admin_pubkeys.split(','):
+                    key = key.strip()
+                    if not key:
+                        continue
+                    if validate_pubkey_format(key, expected_length=64):
+                        nk = key.lower()
+                        if nk not in keys:
+                            keys.append(nk)
+                    else:
+                        logger.warning('Invalid pubkey in admin_pubkeys: %s...', key[:16])
+    except Exception as e:
+        logger.debug('collect_protected_pubkeys_for_device_mode: %s', e)
+    return set(keys)
+
+
 class RepeaterManager:
     """Manages repeater contacts database and purging operations"""
 
@@ -35,6 +72,7 @@ class RepeaterManager:
         self.auto_purge_threshold = 280  # Start purging when 280+ contacts
         # Respect auto_manage_contacts: manual mode (false) = no auto-purge; device/bot = auto-purge on
         auto_manage = bot.config.get('Bot', 'auto_manage_contacts', fallback='false').lower()
+        self._auto_manage_contacts = auto_manage
         self.auto_purge_enabled = (auto_manage != 'false')
 
         # Initialize companion purge settings
@@ -648,11 +686,26 @@ class RepeaterManager:
                 if not self.auto_purge_enabled:
                     return False
 
+                if not getattr(self.bot, 'meshcore', None) or not hasattr(self.bot.meshcore, 'contacts'):
+                    return False
+
+                await self._update_contact_limit_from_device()
+
                 # Get current contact count
                 current_count = len(self.bot.meshcore.contacts)
 
                 if current_count >= self.auto_purge_threshold:
-                    self.logger.info(f"🔄 Auto-purge triggered: {current_count}/{self.contact_limit} contacts (threshold: {self.auto_purge_threshold})")
+                    if self._auto_manage_contacts == 'device':
+                        self.logger.info(
+                            "Repeater-only auto-purge check: %s/%s contacts (threshold %s)",
+                            current_count,
+                            self.contact_limit,
+                            self.auto_purge_threshold,
+                        )
+                    else:
+                        self.logger.info(
+                            f"🔄 Auto-purge triggered: {current_count}/{self.contact_limit} contacts (threshold: {self.auto_purge_threshold})"
+                        )
 
                     # Calculate how many to purge
                     target_count = self.auto_purge_threshold - 20  # Leave some buffer
@@ -663,19 +716,46 @@ class RepeaterManager:
                         repeater_success = await self._auto_purge_repeaters(purge_count)
                         remaining_count = len(self.bot.meshcore.contacts)
 
-                        # If still above threshold and companion purging is enabled, purge companions
-                        if remaining_count >= self.auto_purge_threshold and self.companion_purge_enabled:
+                        companion_success = False
+                        # If still above threshold and companion purging is enabled, purge companions.
+                        # Device mode: never auto-purge companions from the radio — firmware overwrite + favourite
+                        # hygiene own the contact table; repeater purge above may still free slots.
+                        if (
+                            remaining_count >= self.auto_purge_threshold
+                            and self.companion_purge_enabled
+                            and self._auto_manage_contacts != 'device'
+                        ):
                             remaining_purge_count = remaining_count - target_count
                             self.logger.info(f"Still above threshold after repeater purge, purging {remaining_purge_count} companions...")
                             companion_success = await self._auto_purge_companions(remaining_purge_count)
+                        elif (
+                            remaining_count >= self.auto_purge_threshold
+                            and self.companion_purge_enabled
+                            and self._auto_manage_contacts == 'device'
+                        ):
+                            self.logger.info(
+                                "Still above contact threshold after repeater purge; skipping companion auto-purge "
+                                "(auto_manage_contacts=device — firmware handles slot policy)"
+                            )
 
-                            if repeater_success or companion_success:
-                                final_count = len(self.bot.meshcore.contacts)
-                                self.logger.info(f"✅ Auto-purge completed, now at {final_count}/{self.contact_limit} contacts")
-                                return True
+                        if repeater_success or companion_success:
+                            final_count = len(self.bot.meshcore.contacts)
+                            self.logger.info(f"✅ Auto-purge completed, now at {final_count}/{self.contact_limit} contacts")
+                            return True
                         elif repeater_success:
                             self.logger.info(f"✅ Auto-purged {purge_count} repeaters, now at {remaining_count}/{self.contact_limit} contacts")
                             return True
+                        elif (
+                            self._auto_manage_contacts == 'device'
+                            and remaining_count >= self.auto_purge_threshold
+                        ):
+                            # Not an error: we do not bulk-remove companions on the radio in device mode.
+                            self.logger.info(
+                                "Device mode: %s contacts still at/above threshold; no repeater slots freed. "
+                                "Companion list changes are left to firmware (overwrite non-favourite / favourites).",
+                                remaining_count,
+                            )
+                            return True
                         else:
                             self.logger.warning(f"❌ Auto-purge failed to remove {purge_count} contacts")
                             return False
@@ -695,7 +775,13 @@ class RepeaterManager:
             repeaters_to_purge = await self._get_repeaters_for_purging(count)
 
             if not repeaters_to_purge:
-                self.logger.warning("No repeaters available for auto-purge")
+                if self._auto_manage_contacts == 'device':
+                    self.logger.debug(
+                        "No repeaters on device for repeater-only auto-purge check (%s contacts)",
+                        len(self.bot.meshcore.contacts),
+                    )
+                else:
+                    self.logger.warning("No repeaters available for auto-purge")
                 # Log some debugging info
                 total_contacts = len(self.bot.meshcore.contacts)
                 repeater_count = sum(1 for contact_data in list(self.bot.meshcore.contacts.values()) if self._is_repeater_device(contact_data))
@@ -2680,6 +2766,163 @@ class RepeaterManager:
             self.logger.error(f"Error in aggressive contact cleanup: {e}")
             return 0
 
+    @staticmethod
+    def _is_meshcore_table_full(result: Any) -> bool:
+        if result is None or not hasattr(result, 'type'):
+            return False
+        if result.type != EventType.ERROR:
+            return False
+        payload = getattr(result, 'payload', None) or {}
+        if payload.get('error_code') == 3:
+            return True
+        code_str = str(payload.get('code_string', '') or '')
+        return 'TABLE_FULL' in code_str.upper()
+
+    async def add_companion_from_contact_data(
+        self,
+        contact_data: dict[str, Any],
+        contact_name: str,
+        public_key: str,
+    ) -> bool:
+        """Add companion using full contact_data (paths). Pre-manages when near limit; retries once on TABLE_FULL."""
+        try:
+            if not self.bot.meshcore or not hasattr(self.bot.meshcore, 'commands'):
+                self.logger.error('No meshcore commands — cannot add companion %s', contact_name)
+                return False
+
+            status = await self.get_contact_list_status()
+            if status and status.get('is_near_limit'):
+                self.logger.warning(
+                    'Contact list near limit (%.1f%%) before add — managing capacity for %s',
+                    status.get('usage_percentage', 0.0),
+                    contact_name,
+                )
+                await self.manage_contact_list(auto_cleanup=True)
+
+            result = await self.bot.meshcore.commands.add_contact(contact_data)
+            if hasattr(result, 'type') and result.type == EventType.OK:
+                self.logger.info("Companion %s added to device contacts", contact_name)
+                return True
+
+            if self._is_meshcore_table_full(result):
+                self.logger.warning(
+                    'TABLE_FULL adding companion %s — running manage_contact_list and retrying once',
+                    contact_name,
+                )
+                await self.manage_contact_list(auto_cleanup=True)
+                result = await self.bot.meshcore.commands.add_contact(contact_data)
+                if hasattr(result, 'type') and result.type == EventType.OK:
+                    self.logger.info('Companion %s added after retry', contact_name)
+                    return True
+
+            self.logger.warning('Failed to add companion %s to device: %s', contact_name, result)
+            return False
+        except Exception as e:
+            self.logger.error('Error adding companion %s: %s', contact_name, e)
+            return False
+
+    async def apply_device_mode_firmware_preferences(self) -> bool:
+        """Set companion-radio firmware: manual per-type adds + overwrite oldest non-favourite + chat-only (0x03)."""
+        if not self.bot.meshcore or not hasattr(self.bot.meshcore, 'commands'):
+            return False
+        if self.bot.config.get('Bot', 'auto_manage_contacts', fallback='false').lower() != 'device':
+            self.logger.info('Skipping firmware autoadd setup — auto_manage_contacts is not device')
+            return False
+        try:
+            cmds = self.bot.meshcore.commands
+            r_manual = await cmds.set_manual_add_contacts(True)
+            if hasattr(r_manual, 'type') and r_manual.type != EventType.OK:
+                self.logger.warning('set_manual_add_contacts(True) returned %s', r_manual)
+            r_auto = await cmds.set_autoadd_config(0x03)
+            if hasattr(r_auto, 'type') and r_auto.type != EventType.OK:
+                self.logger.warning('set_autoadd_config(0x03) returned %s', r_auto)
+                return False
+            self.logger.info('Device mode: firmware autoadd_config=0x03 (overwrite oldest + chat only), manual add on')
+            return True
+        except Exception as e:
+            self.logger.error('apply_device_mode_firmware_preferences failed: %s', e)
+            return False
+
+    async def sync_device_mode_favourites_pass1(self) -> None:
+        """Favourite all on-device contacts whose pubkey is in the protected set (admin + announcements ACL)."""
+        if self.bot.config.get('Bot', 'auto_manage_contacts', fallback='false').lower() != 'device':
+            return
+        if not self.bot.meshcore or not hasattr(self.bot.meshcore, 'commands'):
+            return
+        protected = collect_protected_pubkeys_for_device_mode(self.bot.config, self.logger)
+        if not protected:
+            self.logger.debug('sync_device_mode_favourites_pass1: no protected pubkeys configured')
+        spacing_ms = max(0, self.bot.config.getint('Bot', 'contact_flag_update_spacing_ms', fallback=200))
+        spacing = spacing_ms / 1000.0
+        try:
+            await self.bot.meshcore.commands.get_contacts()
+        except Exception as e:
+            self.logger.debug('get_contacts before favourite pass1: %s', e)
+        contacts = getattr(self.bot.meshcore, 'contacts', None) or {}
+        for pub_key, c in list(contacts.items()):
+            if self.bot.config.get('Bot', 'auto_manage_contacts', fallback='false').lower() != 'device':
+                return
+            pk = (pub_key or '').lower()
+            if pk not in protected:
+                continue
+            try:
+                flags = int(c.get('flags', 0))
+            except (TypeError, ValueError):
+                flags = 0
+            if flags & 1:
+                continue
+            contact_copy = dict(c)
+            new_flags = flags | 1
+            try:
+                res = await self.bot.meshcore.commands.change_contact_flags(contact_copy, new_flags)
+                if hasattr(res, 'type') and res.type == EventType.OK:
+                    self.logger.debug('Favourited protected contact %s', pk[:16])
+                else:
+                    self.logger.warning('change_contact_flags (favourite on) failed for %s: %s', pk[:16], res)
+            except Exception as e:
+                self.logger.warning('change_contact_flags error for %s: %s', pk[:16], e)
+            if spacing > 0:
+                await asyncio.sleep(spacing)
+
+    async def sync_device_mode_favourites_pass2(self) -> None:
+        """Clear favourite bit for contacts not in the protected set."""
+        if self.bot.config.get('Bot', 'auto_manage_contacts', fallback='false').lower() != 'device':
+            return
+        if not self.bot.meshcore or not hasattr(self.bot.meshcore, 'commands'):
+            return
+        protected = collect_protected_pubkeys_for_device_mode(self.bot.config, self.logger)
+        spacing_ms = max(0, self.bot.config.getint('Bot', 'contact_flag_update_spacing_ms', fallback=200))
+        spacing = spacing_ms / 1000.0
+        try:
+            await self.bot.meshcore.commands.get_contacts()
+        except Exception as e:
+            self.logger.debug('get_contacts before favourite pass2: %s', e)
+        contacts = getattr(self.bot.meshcore, 'contacts', None) or {}
+        for pub_key, c in list(contacts.items()):
+            if self.bot.config.get('Bot', 'auto_manage_contacts', fallback='false').lower() != 'device':
+                return
+            pk = (pub_key or '').lower()
+            if pk in protected:
+                continue
+            try:
+                flags = int(c.get('flags', 0))
+            except (TypeError, ValueError):
+                flags = 0
+            if not (flags & 1):
+                continue
+            contact_copy = dict(c)
+            new_flags = flags & ~1
+            try:
+                res = await self.bot.meshcore.commands.change_contact_flags(contact_copy, new_flags)
+                if hasattr(res, 'type') and res.type == EventType.OK:
+                    self.logger.debug('Cleared favourite for non-protected contact %s', pk[:16])
+                else:
+                    self.logger.warning('change_contact_flags (favourite off) failed for %s: %s', pk[:16], res)
+            except Exception as e:
+                self.logger.warning('change_contact_flags error for %s: %s', pk[:16], e)
+            if spacing > 0:
+                await asyncio.sleep(spacing)
+
     async def add_discovered_contact(self, contact_name: str, public_key: Optional[str] = None, reason: str = "Manual addition") -> bool:
         """Add a discovered contact to the contact list using multiple methods"""
         try:
@@ -3167,11 +3410,27 @@ class RepeaterManager:
             if not self.auto_purge_enabled:
                 return
 
+            if not getattr(self.bot, 'meshcore', None) or not hasattr(self.bot.meshcore, 'contacts'):
+                return
+
+            await self._update_contact_limit_from_device()
+
             current_count = len(self.bot.meshcore.contacts)
 
             # Log current status
             if current_count >= self.auto_purge_threshold:
-                self.logger.warning(f"⚠️ Contact limit monitoring: {current_count}/{self.contact_limit} contacts (threshold: {self.auto_purge_threshold})")
+                if self._auto_manage_contacts == 'device':
+                    self.logger.info(
+                        "📊 Contact list high (%s/%s, threshold %s); device mode — running repeater-only auto-purge check "
+                        "(companions not bulk-removed by bot)",
+                        current_count,
+                        self.contact_limit,
+                        self.auto_purge_threshold,
+                    )
+                else:
+                    self.logger.warning(
+                        f"⚠️ Contact limit monitoring: {current_count}/{self.contact_limit} contacts (threshold: {self.auto_purge_threshold})"
+                    )
 
                 # Trigger auto-purge
                 await self.check_and_auto_purge()
@@ -3266,31 +3525,76 @@ class RepeaterManager:
             self.logger.debug(f"Background geocoding error: {e}")
 
     async def _update_contact_limit_from_device(self):
-        """Update contact limit from device using proper MeshCore API"""
+        """Update contact limit from device using proper MeshCore API.
+
+        In ``auto_manage_contacts=device`` mode, the effective limit is at least the
+        number of contacts currently on the radio (firmware can report a lower
+        ``max_contacts`` than the live table). Count-based auto-purge uses
+        ``threshold = contact_limit + 1`` so routine companion-heavy meshes do not
+        loop through repeater-only purge when the firmware manages the table.
+        """
+        current_mesh = 0
         try:
+            if self.bot.meshcore and hasattr(self.bot.meshcore, 'contacts'):
+                current_mesh = len(self.bot.meshcore.contacts)
+        except Exception:
+            current_mesh = 0
+
+        try:
+            if not self.bot.meshcore or not hasattr(self.bot.meshcore, 'commands'):
+                if self._auto_manage_contacts == 'device':
+                    self.contact_limit = max(self.contact_limit, current_mesh)
+                    self.auto_purge_threshold = self.contact_limit + 1
+                    self.logger.debug(
+                        "Device mode contact limit from mesh only: %s (threshold %s)",
+                        self.contact_limit,
+                        self.auto_purge_threshold,
+                    )
+                return False
+
             # Use the correct MeshCore API to get device info
             device_info = await self.bot.meshcore.commands.send_device_query()
 
             # Check if the query was successful
-            if hasattr(device_info, 'type') and device_info.type.name == 'DEVICE_INFO':
-                max_contacts = device_info.payload.get("max_contacts")
+            if hasattr(device_info, 'type') and device_info.type == EventType.DEVICE_INFO:
+                raw_max = device_info.payload.get('max_contacts')
+                try:
+                    max_contacts = int(raw_max) if raw_max is not None else 0
+                except (TypeError, ValueError):
+                    max_contacts = 0
 
-                if max_contacts and max_contacts > 100:
+                if max_contacts > 100:
                     self.contact_limit = max_contacts
-                    # Update threshold to be 20 contacts below the limit
-                    self.auto_purge_threshold = max(200, max_contacts - 20)
-                    self.logger.debug(f"Updated contact limit from device query: {self.contact_limit} (threshold: {self.auto_purge_threshold})")
+                    if self._auto_manage_contacts == 'device':
+                        self.contact_limit = max(self.contact_limit, current_mesh)
+                        self.auto_purge_threshold = self.contact_limit + 1
+                    else:
+                        # Update threshold to be 20 contacts below the limit
+                        self.auto_purge_threshold = max(200, self.contact_limit - 20)
+                    self.logger.debug(
+                        "Updated contact limit from device query: %s (threshold: %s)",
+                        self.contact_limit,
+                        self.auto_purge_threshold,
+                    )
                     return True
-                else:
-                    self.logger.debug(f"Device returned invalid max_contacts: {max_contacts}")
+
+                self.logger.debug(f"Device returned invalid max_contacts: {raw_max}")
             else:
                 self.logger.debug(f"Device query failed: {device_info}")
 
         except Exception as e:
             self.logger.debug(f"Could not update contact limit from device: {e}")
 
-        # Keep default values if device query failed
-        self.logger.debug(f"Using default contact limit: {self.contact_limit}")
+        if self._auto_manage_contacts == 'device':
+            self.contact_limit = max(self.contact_limit, current_mesh)
+            self.auto_purge_threshold = self.contact_limit + 1
+            self.logger.debug(
+                "Device mode contact limit after failed query: %s (threshold %s)",
+                self.contact_limit,
+                self.auto_purge_threshold,
+            )
+        else:
+            self.logger.debug(f"Using default contact limit: {self.contact_limit}")
         return False
 
     async def get_auto_purge_status(self) -> dict:
diff --git a/modules/scheduler.py b/modules/scheduler.py
index ad79bad..5248fe5 100644
--- a/modules/scheduler.py
+++ b/modules/scheduler.py
@@ -15,6 +15,7 @@ from typing import Any, Optional
 
 from apscheduler.schedulers.background import BackgroundScheduler
 from apscheduler.triggers.cron import CronTrigger
+from apscheduler.triggers.date import DateTrigger
 
 from .maintenance import MaintenanceRunner
 from .security_utils import validate_external_url
@@ -101,6 +102,8 @@ class MessageScheduler:
         self._apscheduler.start()
         self.logger.info(f"APScheduler started with {len(self.scheduled_messages)} scheduled message(s)")
 
+        self._setup_device_mode_scheduler_jobs()
+
         # Setup interval-based advertising
         self.setup_interval_advertising()
 
@@ -118,6 +121,84 @@ class MessageScheduler:
         except Exception as e:
             self.logger.warning(f"Error setting up interval advertising: {e}")
 
+    def _setup_device_mode_scheduler_jobs(self) -> None:
+        """One-shot jobs for auto_manage_contacts=device: firmware autoadd + favourite hygiene."""
+        if self._apscheduler is None:
+            return
+        if self.bot.config.get('Bot', 'auto_manage_contacts', fallback='false').lower() != 'device':
+            return
+        try:
+            delay_fw = max(0, self.bot.config.getint('Bot', 'device_mode_firmware_delay_seconds', fallback=30))
+            delay_p1 = max(0, self.bot.config.getint('Bot', 'device_mode_favourite_pass1_delay_seconds', fallback=90))
+            delay_p2 = max(0, self.bot.config.getint('Bot', 'device_mode_favourite_pass2_delay_seconds', fallback=180))
+            base = self.get_current_time()
+            self._apscheduler.add_job(
+                self._device_mode_firmware_job_sync,
+                trigger=DateTrigger(run_date=base + datetime.timedelta(seconds=delay_fw)),
+                id='device_mode_firmware_autoadd',
+                replace_existing=True,
+            )
+            self._apscheduler.add_job(
+                self._device_mode_favourite_pass1_job_sync,
+                trigger=DateTrigger(run_date=base + datetime.timedelta(seconds=delay_p1)),
+                id='device_mode_favourite_pass1',
+                replace_existing=True,
+            )
+            self._apscheduler.add_job(
+                self._device_mode_favourite_pass2_job_sync,
+                trigger=DateTrigger(run_date=base + datetime.timedelta(seconds=delay_p2)),
+                id='device_mode_favourite_pass2',
+                replace_existing=True,
+            )
+            self.logger.info(
+                'Scheduled device-mode jobs: firmware +%ss, favourite pass1 +%ss, pass2 +%ss',
+                delay_fw,
+                delay_p1,
+                delay_p2,
+            )
+        except Exception as e:
+            self.logger.warning('Could not schedule device-mode contact jobs: %s', e)
+
+    def _run_async_on_main_loop(self, coro: Any, timeout: float = 300.0) -> None:
+        """Run async coroutine on bot main loop from APScheduler thread (same pattern as send_scheduled_message)."""
+        import asyncio
+
+        if hasattr(self.bot, 'main_event_loop') and self.bot.main_event_loop and self.bot.main_event_loop.is_running():
+            future = asyncio.run_coroutine_threadsafe(coro, self.bot.main_event_loop)
+            try:
+                future.result(timeout=timeout)
+            except RuntimeError as e:
+                self.logger.warning('Event loop gone during device-mode job: %s', e)
+            except Exception as e:
+                self.logger.error('Device-mode scheduled job failed: %s', e)
+        else:
+            self.logger.warning('No running main_event_loop — skipping device-mode scheduled job')
+
+    async def _device_mode_firmware_coro(self) -> None:
+        await self.bot.repeater_manager.apply_device_mode_firmware_preferences()
+
+    async def _device_mode_favourite_pass1_coro(self) -> None:
+        await self.bot.repeater_manager.sync_device_mode_favourites_pass1()
+
+    async def _device_mode_favourite_pass2_coro(self) -> None:
+        await self.bot.repeater_manager.sync_device_mode_favourites_pass2()
+
+    def _device_mode_firmware_job_sync(self) -> None:
+        if self.bot.config.get('Bot', 'auto_manage_contacts', fallback='false').lower() != 'device':
+            self.logger.debug('Skipping device_mode_firmware job — not device mode')
+            return
+        self._run_async_on_main_loop(self._device_mode_firmware_coro(), timeout=120.0)
+
+    def _device_mode_favourite_pass1_job_sync(self) -> None:
+        if self.bot.config.get('Bot', 'auto_manage_contacts', fallback='false').lower() != 'device':
+            return
+        self._run_async_on_main_loop(self._device_mode_favourite_pass1_coro(), timeout=600.0)
+
+    def _device_mode_favourite_pass2_job_sync(self) -> None:
+        if self.bot.config.get('Bot', 'auto_manage_contacts', fallback='false').lower() != 'device':
+            return
+        self._run_async_on_main_loop(self._device_mode_favourite_pass2_coro(), timeout=600.0)
+
     def _is_valid_time_format(self, time_str: str) -> bool:
         """Validate time format (HHMM)"""
         try:
diff --git a/tests/test_message_handler.py b/tests/test_message_handler.py
index 7cb4652..8c666d0 100644
--- a/tests/test_message_handler.py
+++ b/tests/test_message_handler.py
@@ -1817,3 +1817,107 @@ class TestRespondToMentions:
         await mention_handler.process_message(msg)
         assert msg.content == "ping"
         mention_bot.command_manager.execute_commands.assert_called_once()
+
+
+# ---------------------------------------------------------------------------
+# handle_new_contact — auto_manage_contacts (companion path)
+# ---------------------------------------------------------------------------
+
+
+class _NewContactEvent:
+    def __init__(self, payload: dict) -> None:
+        self.payload = payload
+
+
+def _companion_contact_payload() -> dict:
+    pk = "ab" * 32
+    return {
+        "public_key": pk,
+        "adv_name": "Alice",
+        "name": "Alice",
+        "type": 1,
+        "flags": 0,
+        "out_path": "",
+        "out_path_len": 0,
+        "out_path_hash_mode": 0,
+    }
+
+
+@pytest.fixture
+def new_contact_env(mock_logger):
+    """Bot + MessageHandler with mocked repeater_manager and meshcore for NEW_CONTACT tests."""
+    bot = Mock()
+    bot.logger = mock_logger
+    bot.config = configparser.ConfigParser()
+    bot.config.add_section("Bot")
+    bot.config.set("Bot", "enabled", "true")
+    bot.config.set("Bot", "rf_data_timeout", "15.0")
+    bot.config.set("Bot", "message_correlation_timeout", "10.0")
+    bot.config.set("Bot", "enable_enhanced_correlation", "true")
+    bot.config.add_section("Channels")
+    bot.config.set("Channels", "respond_to_dms", "true")
+    bot.connection_time = None
+    bot.prefix_hex_chars = 8
+    bot.command_manager = Mock()
+    bot.command_manager.monitor_channels = ["general"]
+    bot.command_manager.is_user_banned = Mock(return_value=False)
+    bot.command_manager.commands = {}
+
+    handler = MessageHandler(bot)
+    bot.message_handler = handler
+
+    rm = Mock()
+    rm.track_contact_advertisement = AsyncMock()
+    rm.check_and_auto_purge = AsyncMock()
+    rm.get_contact_list_status = AsyncMock(
+        return_value={
+            "is_near_limit": False,
+            "usage_percentage": 10.0,
+            "current_contacts": 5,
+            "estimated_limit": 300,
+        }
+    )
+    rm.manage_contact_list = AsyncMock(return_value={"success": True})
+    rm.add_companion_from_contact_data = AsyncMock(return_value=True)
+    rm.db_manager = Mock()
+    rm.db_manager.execute_update = Mock()
+    rm._is_repeater_device = Mock(return_value=False)
+    bot.repeater_manager = rm
+
+    mesh = Mock()
+    mesh.commands = Mock()
+    mesh.commands.add_contact = AsyncMock()
+    bot.meshcore = mesh
+
+    return bot, handler, rm, mesh
+
+
+@pytest.mark.asyncio
+class TestHandleNewContactAutoManage:
+    async def test_manual_mode_no_device_add(self, new_contact_env):
+        bot, handler, rm, mesh = new_contact_env
+        bot.config.set("Bot", "auto_manage_contacts", "false")
+        ev = _NewContactEvent(_companion_contact_payload())
+        await handler.handle_new_contact(ev, None)
+        rm.track_contact_advertisement.assert_awaited_once()
+        mesh.commands.add_contact.assert_not_called()
+        rm.add_companion_from_contact_data.assert_not_called()
+        rm.db_manager.execute_update.assert_called()
+
+    async def test_device_mode_no_bot_add_contact(self, new_contact_env):
+        bot, handler, rm, mesh = new_contact_env
+        bot.config.set("Bot", "auto_manage_contacts", "device")
+        ev = _NewContactEvent(_companion_contact_payload())
+        await handler.handle_new_contact(ev, None)
+        rm.track_contact_advertisement.assert_awaited_once()
+        mesh.commands.add_contact.assert_not_called()
+        rm.add_companion_from_contact_data.assert_not_called()
+        rm.get_contact_list_status.assert_awaited()
+
+    async def test_bot_mode_uses_add_companion_from_contact_data(self, new_contact_env):
+        bot, handler, rm, mesh = new_contact_env
+        bot.config.set("Bot", "auto_manage_contacts", "bot")
+        ev = _NewContactEvent(_companion_contact_payload())
+        await handler.handle_new_contact(ev, None)
+        rm.add_companion_from_contact_data.assert_awaited_once()
+        mesh.commands.add_contact.assert_not_called()
diff --git a/tests/test_message_handler_contact_path.py b/tests/test_message_handler_contact_path.py
index dfda131..afddcb0 100644
--- a/tests/test_message_handler_contact_path.py
+++ b/tests/test_message_handler_contact_path.py
@@ -3,8 +3,10 @@
 from unittest.mock import AsyncMock, MagicMock
 
 import pytest
+from meshcore import EventType
 
 from modules.message_handler import MessageHandler
+from modules.repeater_manager import RepeaterManager
 
 
 def _make_config_get():
@@ -31,11 +33,20 @@ def message_handler():
 
 @pytest.fixture
 def companion_new_contact_setup():
-    """Bot + MessageHandler wired for companion NEW_CONTACT → add_contact."""
+    """Bot + MessageHandler wired for companion NEW_CONTACT → add_contact (bot auto-manage)."""
     bot = MagicMock()
     bot.logger = MagicMock()
+
+    def _config_get(section, key, **kw):
+        defaults = {
+            ("Bot", "rf_data_timeout"): "15.0",
+            ("Bot", "message_correlation_timeout"): "10.0",
+            ("Bot", "auto_manage_contacts"): "bot",
+        }
+        return defaults.get((section, key), kw.get("fallback", ""))
+
     bot.config = MagicMock()
-    bot.config.get = MagicMock(side_effect=_make_config_get())
+    bot.config.get = MagicMock(side_effect=_config_get)
     bot.config.getboolean = MagicMock(return_value=True)
     bot.prefix_hex_chars = 8
 
@@ -43,14 +54,26 @@ def companion_new_contact_setup():
     bot.message_handler = mh
 
     rm = MagicMock()
+    rm.bot = bot
+    rm.logger = bot.logger
     rm._is_repeater_device = MagicMock(return_value=False)
     rm.track_contact_advertisement = AsyncMock()
     rm.check_and_auto_purge = AsyncMock()
+    rm.get_contact_list_status = AsyncMock(
+        return_value={"is_near_limit": False, "usage_percentage": 0.0}
+    )
+    rm.manage_contact_list = AsyncMock()
+    rm.db_manager = MagicMock()
+    rm.db_manager.execute_update = MagicMock()
+
+    async def _add_companion(contact_data, contact_name, public_key):
+        return await RepeaterManager.add_companion_from_contact_data(rm, contact_data, contact_name, public_key)
+
+    rm.add_companion_from_contact_data = AsyncMock(side_effect=_add_companion)
     bot.repeater_manager = rm
 
     ok = MagicMock()
-    ok.type = MagicMock()
-    ok.type.name = "OK"
+    ok.type = EventType.OK
     bot.meshcore = MagicMock()
     bot.meshcore.commands = MagicMock()
     bot.meshcore.commands.add_contact = AsyncMock(return_value=ok)
diff --git a/tests/test_repeater_manager.py b/tests/test_repeater_manager.py
index 2aa6bce..7f37c34 100644
--- a/tests/test_repeater_manager.py
+++ b/tests/test_repeater_manager.py
@@ -9,7 +9,10 @@ from unittest.mock import AsyncMock, Mock, patch
 import pytest
 from meshcore import EventType
 
-from modules.repeater_manager import RepeaterManager
+from modules.repeater_manager import (
+    RepeaterManager,
+    collect_protected_pubkeys_for_device_mode,
+)
 
 
 @pytest.fixture
@@ -496,6 +499,7 @@ class TestCheckAndAutoPurge:
         rm.auto_purge_enabled = True
         rm.auto_purge_threshold = 10
         rm.companion_purge_enabled = True
+        rm._auto_manage_contacts = 'bot'
         rm.bot.meshcore = Mock()
         rm.bot.meshcore.contacts = {str(i): {} for i in range(15)}
 
@@ -510,6 +514,59 @@ class TestCheckAndAutoPurge:
         mock_comp.assert_called_once()
         assert result is True
 
+    async def test_companion_purge_skipped_when_auto_manage_device(self, bot):
+        """Device mode: count-based purge is suppressed; companions never auto-purged from radio."""
+        bot.config.set("Bot", "auto_manage_contacts", "device")
+        rm = RepeaterManager(bot)
+        rm.companion_purge_enabled = True
+        rm.bot.meshcore = Mock()
+        rm.bot.meshcore.contacts = {str(i): {} for i in range(15)}
+        rm.bot.meshcore.commands = Mock()
+        rm.bot.meshcore.commands.send_device_query = AsyncMock(return_value=Mock(type=Mock()))
+
+        with patch.object(rm, "_auto_purge_repeaters", new_callable=AsyncMock) as mock_rep, \
+             patch.object(rm, "_auto_purge_companions", new_callable=AsyncMock) as mock_comp:
+            result = await rm.check_and_auto_purge()
+
+        mock_rep.assert_not_called()
+        mock_comp.assert_not_called()
+        assert result is False
+        assert rm.contact_limit >= 15
+        assert rm.auto_purge_threshold == rm.contact_limit + 1
+
+    async def test_device_mode_returns_true_when_still_full_no_repeaters(self, bot):
+        """Device mode: no repeater purge path when mesh is within synced limit (no false failure)."""
+        bot.config.set("Bot", "auto_manage_contacts", "device")
+        rm = RepeaterManager(bot)
+        rm.companion_purge_enabled = False
+        rm.bot.meshcore = Mock()
+        rm.bot.meshcore.contacts = {str(i): {} for i in range(15)}
+        rm.bot.meshcore.commands = Mock()
+        rm.bot.meshcore.commands.send_device_query = AsyncMock(return_value=Mock(type=Mock()))
+
+        with patch.object(rm, "_auto_purge_repeaters", new_callable=AsyncMock) as mock_rep, \
+             patch.object(rm, "_auto_purge_companions", new_callable=AsyncMock) as mock_comp:
+            result = await rm.check_and_auto_purge()
+
+        mock_rep.assert_not_called()
+        mock_comp.assert_not_called()
+        assert result is False
+
+    async def test_device_mode_floors_contact_limit_to_mesh_when_firmware_under_reports(self, bot):
+        """If live contact count exceeds DEVICE_INFO max_contacts, raise limit to match the radio."""
+        bot.config.set("Bot", "auto_manage_contacts", "device")
+        rm = RepeaterManager(bot)
+        rm.bot.meshcore = Mock()
+        rm.bot.meshcore.contacts = {str(i): {} for i in range(331)}
+        info = SimpleNamespace(type=EventType.DEVICE_INFO, payload={"max_contacts": 300})
+        rm.bot.meshcore.commands = Mock()
+        rm.bot.meshcore.commands.send_device_query = AsyncMock(return_value=info)
+
+        await rm._update_contact_limit_from_device()
+
+        assert rm.contact_limit == 331
+        assert rm.auto_purge_threshold == 332
+
     async def test_returns_false_when_purge_fails(self, rm):
         """When both purge counts succeed=False, check_and_auto_purge returns False."""
         rm.auto_purge_enabled = True
@@ -1302,3 +1359,240 @@ class TestPurgeDedupConcurrency:
         assert first_result is False
         assert second_result is True
         assert rm.bot.meshcore.commands.remove_contact.await_count == 2
+
+
+class TestCollectProtectedPubkeysForDeviceMode:
+    """collect_protected_pubkeys_for_device_mode matches Admin + announcements ACL union."""
+
+    def test_admin_and_announcements_merged(self, mock_logger):
+        cfg = configparser.ConfigParser()
+        pk_a = "aa" * 32
+        pk_b = "bb" * 32
+        cfg.add_section("Admin_ACL")
+        cfg.set("Admin_ACL", "admin_pubkeys", pk_a)
+        cfg.add_section("Announcements_Command")
+        cfg.set("Announcements_Command", "announcements_acl", pk_b)
+        keys = collect_protected_pubkeys_for_device_mode(cfg, mock_logger)
+        assert keys == {pk_a.lower(), pk_b.lower()}
+
+    def test_admin_only_when_no_announcements_acl(self, mock_logger):
+        cfg = configparser.ConfigParser()
+        pk_a = "cc" * 32
+        cfg.add_section("Admin_ACL")
+        cfg.set("Admin_ACL", "admin_pubkeys", pk_a)
+        cfg.add_section("Announcements_Command")
+        cfg.set("Announcements_Command", "announcements_acl", "")
+        keys = collect_protected_pubkeys_for_device_mode(cfg, mock_logger)
+        assert keys == {pk_a.lower()}
+
+
+class TestAddCompanionFromContactData:
+    """RepeaterManager.add_companion_from_contact_data TABLE_FULL retry."""
+
+    @pytest.mark.asyncio
+    async def test_retries_after_table_full(self, rm, bot):
+        pk = "dd" * 32
+        contact_data = {
+            "public_key": pk,
+            "adv_name": "Bob",
+            "type": 1,
+            "flags": 0,
+            "out_path": "",
+            "out_path_len": 0,
+            "out_path_hash_mode": 0,
+            "last_advert": 0,
+            "adv_lat": 0.0,
+            "adv_lon": 0.0,
+        }
+        bot.meshcore = Mock()
+        bot.meshcore.commands = Mock()
+        err = SimpleNamespace(type=EventType.ERROR, payload={"error_code": 3, "code_string": "ERR_CODE_TABLE_FULL"})
+        ok = SimpleNamespace(type=EventType.OK, payload={})
+        bot.meshcore.commands.add_contact = AsyncMock(side_effect=[err, ok])
+
+        rm.get_contact_list_status = AsyncMock(
+            return_value={
+                "is_near_limit": False,
+                "usage_percentage": 50.0,
+                "current_contacts": 100,
+                "estimated_limit": 300,
+            }
+        )
+        rm.manage_contact_list = AsyncMock(return_value={"success": True})
+
+        result = await rm.add_companion_from_contact_data(contact_data, "Bob", pk)
+        assert result is True
+        assert bot.meshcore.commands.add_contact.await_count == 2
+        rm.manage_contact_list.assert_awaited()
+
+    @pytest.mark.asyncio
+    async def test_success_on_first_ok_without_retry(self, rm, bot):
+        pk = "ee" * 32
+        contact_data = {
+            "public_key": pk,
+            "adv_name": "Ann",
+            "type": 1,
+            "flags": 0,
+            "out_path": "",
+            "out_path_len": 0,
+            "out_path_hash_mode": 0,
+            "last_advert": 0,
+            "adv_lat": 0.0,
+            "adv_lon": 0.0,
+        }
+        bot.meshcore = Mock()
+        bot.meshcore.commands = Mock()
+        ok = SimpleNamespace(type=EventType.OK, payload={})
+        bot.meshcore.commands.add_contact = AsyncMock(return_value=ok)
+        rm.get_contact_list_status = AsyncMock(
+            return_value={"is_near_limit": False, "usage_percentage": 10.0}
+        )
+        rm.manage_contact_list = AsyncMock()
+
+        result = await rm.add_companion_from_contact_data(contact_data, "Ann", pk)
+        assert result is True
+        bot.meshcore.commands.add_contact.assert_awaited_once()
+        rm.manage_contact_list.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_near_limit_triggers_manage_before_add(self, rm, bot):
+        pk = "ff" * 32
+        contact_data = {
+            "public_key": pk,
+            "adv_name": "Near",
+            "type": 1,
+            "flags": 0,
+            "out_path": "",
+            "out_path_len": 0,
+            "out_path_hash_mode": 0,
+            "last_advert": 0,
+            "adv_lat": 0.0,
+            "adv_lon": 0.0,
+        }
+        bot.meshcore = Mock()
+        bot.meshcore.commands = Mock()
+        ok = SimpleNamespace(type=EventType.OK, payload={})
+        bot.meshcore.commands.add_contact = AsyncMock(return_value=ok)
+        rm.get_contact_list_status = AsyncMock(
+            return_value={"is_near_limit": True, "usage_percentage": 85.0}
+        )
+        rm.manage_contact_list = AsyncMock(return_value={"success": True})
+
+        result = await rm.add_companion_from_contact_data(contact_data, "Near", pk)
+        assert result is True
+        rm.manage_contact_list.assert_awaited_once()
+        bot.meshcore.commands.add_contact.assert_awaited_once()
+
+
+class TestApplyDeviceModeFirmwarePreferences:
+    @pytest.mark.asyncio
+    async def test_success_sets_manual_and_autoadd(self, bot):
+        bot.config.set("Bot", "auto_manage_contacts", "device")
+        rm = RepeaterManager(bot)
+        ok = SimpleNamespace(type=EventType.OK)
+        bot.meshcore = Mock()
+        bot.meshcore.commands = Mock()
+        bot.meshcore.commands.set_manual_add_contacts = AsyncMock(return_value=ok)
+        bot.meshcore.commands.set_autoadd_config = AsyncMock(return_value=ok)
+
+        assert await rm.apply_device_mode_firmware_preferences() is True
+        bot.meshcore.commands.set_autoadd_config.assert_awaited_once_with(0x03)
+
+    @pytest.mark.asyncio
+    async def test_returns_false_when_not_device_mode(self, bot):
+        bot.config.set("Bot", "auto_manage_contacts", "bot")
+        rm = RepeaterManager(bot)
+        bot.meshcore = Mock()
+        bot.meshcore.commands = Mock()
+
+        assert await rm.apply_device_mode_firmware_preferences() is False
+
+    @pytest.mark.asyncio
+    async def test_returns_false_when_autoadd_not_ok(self, bot):
+        bot.config.set("Bot", "auto_manage_contacts", "device")
+        rm = RepeaterManager(bot)
+        ok = SimpleNamespace(type=EventType.OK)
+        bad = SimpleNamespace(type=EventType.ERROR, payload={})
+        bot.meshcore = Mock()
+        bot.meshcore.commands = Mock()
+        bot.meshcore.commands.set_manual_add_contacts = AsyncMock(return_value=ok)
+        bot.meshcore.commands.set_autoadd_config = AsyncMock(return_value=bad)
+
+        assert await rm.apply_device_mode_firmware_preferences() is False
+
+
+class TestSyncDeviceModeFavourites:
+    @pytest.mark.asyncio
+    async def test_pass1_no_op_when_not_device(self, bot):
+        bot.config.set("Bot", "auto_manage_contacts", "bot")
+        rm = RepeaterManager(bot)
+        bot.meshcore = Mock()
+        bot.meshcore.commands = Mock()
+        bot.meshcore.commands.get_contacts = AsyncMock()
+        bot.meshcore.commands.change_contact_flags = AsyncMock()
+
+        await rm.sync_device_mode_favourites_pass1()
+
+        bot.meshcore.commands.get_contacts.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_pass1_favourites_protected_contact(self, bot):
+        bot.config.set("Bot", "auto_manage_contacts", "device")
+        bot.config.set("Bot", "contact_flag_update_spacing_ms", "0")
+        pk = "aa" * 32
+        bot.config.add_section("Admin_ACL")
+        bot.config.set("Admin_ACL", "admin_pubkeys", pk)
+        rm = RepeaterManager(bot)
+        ok = SimpleNamespace(type=EventType.OK)
+        bot.meshcore = Mock()
+        bot.meshcore.contacts = {pk: {"public_key": pk, "flags": 0, "adv_name": "A", "type": 1}}
+        bot.meshcore.commands = Mock()
+        bot.meshcore.commands.get_contacts = AsyncMock()
+        bot.meshcore.commands.change_contact_flags = AsyncMock(return_value=ok)
+
+        await rm.sync_device_mode_favourites_pass1()
+
+        bot.meshcore.commands.change_contact_flags.assert_awaited_once()
+        args, _kwargs = bot.meshcore.commands.change_contact_flags.await_args
+        assert args[1] == 1
+
+    @pytest.mark.asyncio
+    async def test_pass2_clears_favourite_for_non_protected(self, bot):
+        bot.config.set("Bot", "auto_manage_contacts", "device")
+        bot.config.set("Bot", "contact_flag_update_spacing_ms", "0")
+        pk_prot = "bb" * 32
+        pk_other = "cc" * 32
+        bot.config.add_section("Admin_ACL")
+        bot.config.set("Admin_ACL", "admin_pubkeys", pk_prot)
+        rm = RepeaterManager(bot)
+        ok = SimpleNamespace(type=EventType.OK)
+        bot.meshcore = Mock()
+        bot.meshcore.contacts = {
+            pk_other: {"public_key": pk_other, "flags": 1, "adv_name": "X", "type": 1},
+        }
+        bot.meshcore.commands = Mock()
+        bot.meshcore.commands.get_contacts = AsyncMock()
+        bot.meshcore.commands.change_contact_flags = AsyncMock(return_value=ok)
+
+        await rm.sync_device_mode_favourites_pass2()
+
+        bot.meshcore.commands.change_contact_flags.assert_awaited_once()
+        args, _kwargs = bot.meshcore.commands.change_contact_flags.await_args
+        assert args[1] == 0
+
+
+class TestUpdateContactLimitFromDevice:
+    @pytest.mark.asyncio
+    async def test_bot_mode_uses_firmware_max_and_threshold(self, bot):
+        bot.config.set("Bot", "auto_manage_contacts", "bot")
+        rm = RepeaterManager(bot)
+        info = SimpleNamespace(type=EventType.DEVICE_INFO, payload={"max_contacts": 300})
+        bot.meshcore = Mock()
+        bot.meshcore.contacts = {str(i): {} for i in range(10)}
+        bot.meshcore.commands = Mock()
+        bot.meshcore.commands.send_device_query = AsyncMock(return_value=info)
+
+        await rm._update_contact_limit_from_device()
+
+        assert rm.contact_limit == 300
+        assert rm.auto_purge_threshold == 280
diff --git a/tests/test_scheduler_logic.py b/tests/test_scheduler_logic.py
index 631a015..a5a7c7b 100644
--- a/tests/test_scheduler_logic.py
+++ b/tests/test_scheduler_logic.py
@@ -1620,3 +1620,72 @@ class TestZombieAlertEmailSsrfGuard:
                 sched.send_zombie_alert_email(fail_count=5, threshold=3, interval=60)
         mock_smtp.assert_not_called()
         mock_smtp_ssl.assert_not_called()
+
+
+class TestDeviceModeSchedulerJobs:
+    """_setup_device_mode_scheduler_jobs registers one-shot jobs when auto_manage_contacts=device."""
+
+    def test_registers_three_jobs_in_device_mode(self, scheduler):
+        scheduler.bot.config.set("Bot", "auto_manage_contacts", "device")
+        mock_ap = MagicMock()
+        scheduler._apscheduler = mock_ap
+        scheduler._setup_device_mode_scheduler_jobs()
+        assert mock_ap.add_job.call_count == 3
+        ids = {call.kwargs["id"] for call in mock_ap.add_job.call_args_list}
+        assert ids == {
+            "device_mode_firmware_autoadd",
+            "device_mode_favourite_pass1",
+            "device_mode_favourite_pass2",
+        }
+        for call in mock_ap.add_job.call_args_list:
+            assert call.kwargs.get("replace_existing") is True
+
+    def test_no_jobs_when_not_device_mode(self, scheduler):
+        scheduler.bot.config.set("Bot", "auto_manage_contacts", "bot")
+        mock_ap = MagicMock()
+        scheduler._apscheduler = mock_ap
+        scheduler._setup_device_mode_scheduler_jobs()
+        mock_ap.add_job.assert_not_called()
+
+    def test_firmware_job_sync_skips_when_not_device(self, scheduler):
+        scheduler.bot.config.set("Bot", "auto_manage_contacts", "bot")
+        with patch.object(scheduler, "_run_async_on_main_loop") as mock_run:
+            scheduler._device_mode_firmware_job_sync()
+        mock_run.assert_not_called()
+
+    def test_run_async_warns_without_running_loop(self, scheduler, mock_logger):
+        scheduler.bot.logger = mock_logger
+        scheduler.bot.main_event_loop = None
+
+        async def trivial():
+            pass
+
+        coro = trivial()
+        try:
+            scheduler._run_async_on_main_loop(coro, timeout=1.0)
+        finally:
+            coro.close()
+        mock_logger.warning.assert_called()
+
+    @pytest.mark.asyncio
+    async def test_device_mode_firmware_coro_calls_repeater_manager(self, scheduler):
+        scheduler.bot.config.set("Bot", "auto_manage_contacts", "device")
+        scheduler.bot.repeater_manager = Mock()
+        scheduler.bot.repeater_manager.apply_device_mode_firmware_preferences = AsyncMock(return_value=True)
+
+        await scheduler._device_mode_firmware_coro()
+
+        scheduler.bot.repeater_manager.apply_device_mode_firmware_preferences.assert_awaited_once()
+
+    @pytest.mark.asyncio
+    async def test_device_mode_favourite_coros_delegate(self, scheduler):
+        scheduler.bot.config.set("Bot", "auto_manage_contacts", "device")
+        scheduler.bot.repeater_manager = Mock()
+        scheduler.bot.repeater_manager.sync_device_mode_favourites_pass1 = AsyncMock()
+        scheduler.bot.repeater_manager.sync_device_mode_favourites_pass2 = AsyncMock()
+
+        await scheduler._device_mode_favourite_pass1_coro()
+        await scheduler._device_mode_favourite_pass2_coro()
+
+        scheduler.bot.repeater_manager.sync_device_mode_favourites_pass1.assert_awaited_once()
+        scheduler.bot.repeater_manager.sync_device_mode_favourites_pass2.assert_awaited_once()