fix(companion): protect direct offline messages

repeater/companion/frame_server.py

@@ -76,13 +76,19 @@ class CompanionFrameServer(_BaseFrameServer):
         if retention == 0:
             self.bridge.message_queue.pop_last()
             return
-        await asyncio.to_thread(
+        persisted = await asyncio.to_thread(
             self.sqlite_handler.companion_push_message,
             self.companion_hash,
             msg_dict,
             retention,
         )
-        self.bridge.message_queue.pop_last()
+        if persisted:
+            self.bridge.message_queue.pop_last()
+        else:
+            logger.debug(
+                "Companion %s: retaining message in memory after SQLite queue rejection",
+                self.companion_hash,
+            )
 
     def _sync_next_from_persistence(self) -> Optional[QueuedMessage]:
         """Retrieve next message from SQLite when bridge queue is empty."""

repeater/data_acquisition/sqlite_handler.py

@@ -3358,12 +3358,17 @@ class SQLiteHandler:
         previous SELECT + INSERT round-trip (two statements, two SD-card reads)
         with a single atomic statement.
 
-        When ``max_messages`` is set, the oldest rows beyond that retention limit
-        are trimmed after a successful insert (power-user ``offline_queue_size``).
+        When ``max_messages`` is set, capacity follows MeshCore's offline queue
+        policy: evict the oldest channel message first and never displace a
+        retained direct message. The insert and any eviction share one
+        transaction.
 
-        Returns True if inserted, False if the message was a duplicate (skipped).
+        Returns True if the message is retained, False if it is a duplicate or
+        the protected queue cannot make room for it.
         """
         try:
+            if max_messages is not None and max_messages <= 0:
+                return False
             packet_hash = msg.get("packet_hash") or None
             if isinstance(packet_hash, bytes):
                 packet_hash = packet_hash.decode("utf-8", errors="replace") if packet_hash else None
@@ -3372,6 +3377,7 @@ class SQLiteHandler:
             if not isinstance(sender_prefix, str):
                 sender_prefix = bytes(sender_prefix or b"").hex()
             with self._connect() as conn:
+                conn.execute("SAVEPOINT companion_message_push")
                 cursor = conn.execute(
                     """
                     INSERT OR IGNORE INTO companion_messages
@@ -3394,22 +3400,49 @@ class SQLiteHandler:
                     ),
                 )
                 inserted = cursor.rowcount > 0
-                if inserted and max_messages is not None:
-                    # Keep the newest `max_messages` rows; drop older overflow.
-                    conn.execute(
-                        """
-                        DELETE FROM companion_messages
-                        WHERE companion_hash = ? AND id NOT IN (
+                if not inserted:
+                    conn.execute("RELEASE SAVEPOINT companion_message_push")
+                    conn.commit()
+                    return False
+                if max_messages is not None:
+                    count = conn.execute(
+                        "SELECT COUNT(*) FROM companion_messages WHERE companion_hash = ?",
+                        (companion_hash,),
+                    ).fetchone()[0]
+                    while count > max_messages:
+                        oldest_channel = conn.execute(
+                            """
                             SELECT id FROM companion_messages
-                            WHERE companion_hash = ?
-                            ORDER BY created_at DESC, id DESC
-                            LIMIT ?
+                            WHERE companion_hash = ? AND is_channel = 1
+                            ORDER BY created_at ASC, id ASC LIMIT 1
+                            """,
+                            (companion_hash,),
+                        ).fetchone()
+                        if oldest_channel is None:
+                            # The just-inserted row is not retainable without
+                            # sacrificing a direct message. Keep every prior
+                            # row intact, including any channel rows already
+                            # considered while satisfying a lowered limit.
+                            conn.execute("ROLLBACK TO SAVEPOINT companion_message_push")
+                            conn.execute("RELEASE SAVEPOINT companion_message_push")
+                            conn.commit()
+                            return False
+                        if oldest_channel[0] == cursor.lastrowid:
+                            # A new channel message is itself the only
+                            # evictable row; retain the protected directs and
+                            # report that the incoming message was rejected.
+                            conn.execute("ROLLBACK TO SAVEPOINT companion_message_push")
+                            conn.execute("RELEASE SAVEPOINT companion_message_push")
+                            conn.commit()
+                            return False
+                        conn.execute(
+                            "DELETE FROM companion_messages WHERE id = ?",
+                            (oldest_channel[0],),
                         )
-                        """,
-                        (companion_hash, companion_hash, max_messages),
-                    )
+                        count -= 1
+                conn.execute("RELEASE SAVEPOINT companion_message_push")
                 conn.commit()
-                return inserted
+                return True
         except Exception as e:
             logger.error(f"Failed to push companion message: {e}")
             return False

tests/test_companion_bridge_frame_utils.py

@@ -5,7 +5,7 @@ from typing import Any, cast
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
-from openhop_core.companion.constants import RESP_CODE_NO_MORE_MESSAGES
+from openhop_core.companion.constants import PUSH_CODE_MSG_WAITING, RESP_CODE_NO_MORE_MESSAGES
 
 from repeater.companion.bridge import RepeaterCompanionBridge, _to_json_safe
 from repeater.companion.frame_server import CompanionFrameServer
@@ -201,6 +201,24 @@ async def test_frame_server_no_more_messages_response_when_empty():
         assert srv._write_frame.call_args[0][0] == bytes([RESP_CODE_NO_MORE_MESSAGES])
 
 
+@pytest.mark.asyncio
+async def test_rejected_queue_callback_skips_sqlite_persistence_but_notifies_client():
+    server = object.__new__(CompanionFrameServer)
+    server._persist_companion_message = AsyncMock()
+    server._enqueue_frame = MagicMock()
+
+    await server._on_message_received(
+        b"\x01" * 32,
+        "rejected",
+        1,
+        0,
+        queued=False,
+    )
+
+    server._persist_companion_message.assert_not_awaited()
+    server._enqueue_frame.assert_called_once_with(bytes([PUSH_CODE_MSG_WAITING]))
+
+
 def test_companion_utils_validation_and_normalization():
     assert normalize_companion_identity_key(" 0xAABB ") == "AABB"
     assert validate_companion_node_name("  node-1  ") == "node-1"

tests/test_companion_settings.py

@@ -176,9 +176,54 @@ class TestSqliteRetentionTrim:
 
     def test_trims_to_max_messages(self, tmp_path):
         h = self._handler(tmp_path)
-        for i in range(5):
-            self._push(h, "0x01", i, max_messages=3)
-        assert len(h.companion_load_messages("0x01")) == 3
+        results = [self._push(h, "0x01", i, max_messages=3) for i in range(5)]
+        assert results == [True, True, True, False, False]
+        assert [m["text"] for m in h.companion_load_messages("0x01")] == ["m0", "m1", "m2"]
+
+    def test_evicts_oldest_channel_message_before_direct_message(self, tmp_path):
+        h = self._handler(tmp_path)
+        direct_one = {"text": "direct one", "packet_hash": "d1", "is_channel": False}
+        channel_one = {"text": "channel one", "packet_hash": "c1", "is_channel": True}
+        direct_two = {"text": "direct two", "packet_hash": "d2", "is_channel": False}
+
+        assert h.companion_push_message("0x01", direct_one, max_messages=2)
+        assert h.companion_push_message("0x01", channel_one, max_messages=2)
+        assert h.companion_push_message("0x01", direct_two, max_messages=2)
+
+        messages = h.companion_load_messages("0x01")
+        assert [m["text"] for m in messages] == ["direct one", "direct two"]
+        assert [m["is_channel"] for m in messages] == [0, 0]
+
+    def test_rejects_channel_when_queue_contains_only_direct_messages(self, tmp_path):
+        h = self._handler(tmp_path)
+        for packet_hash in ("d1", "d2"):
+            assert h.companion_push_message(
+                "0x01", {"text": packet_hash, "packet_hash": packet_hash}, max_messages=2
+            )
+
+        assert not h.companion_push_message(
+            "0x01", {"text": "channel", "packet_hash": "c1", "is_channel": True}, max_messages=2
+        )
+        assert [m["text"] for m in h.companion_load_messages("0x01")] == ["d1", "d2"]
+
+    def test_rejected_insert_keeps_existing_channels_when_limit_is_lowered(self, tmp_path):
+        h = self._handler(tmp_path)
+        existing = [
+            {"text": "direct one", "packet_hash": "d1", "is_channel": False},
+            {"text": "channel one", "packet_hash": "c1", "is_channel": True},
+            {"text": "direct two", "packet_hash": "d2", "is_channel": False},
+        ]
+        for message in existing:
+            assert h.companion_push_message("0x01", message)
+
+        assert not h.companion_push_message(
+            "0x01", {"text": "incoming", "packet_hash": "d3"}, max_messages=2
+        )
+        assert [m["text"] for m in h.companion_load_messages("0x01")] == [
+            "direct one",
+            "channel one",
+            "direct two",
+        ]
 
     def test_none_keeps_all(self, tmp_path):
         h = self._handler(tmp_path)
@@ -381,6 +426,14 @@ class TestPersistSkipWhenOff:
         asyncio.run(fs._persist_companion_message({"text": "x"}))
         fs.sqlite_handler.companion_push_message.assert_called_once_with("0x01", {"text": "x"}, 7)
 
+    def test_keeps_memory_message_when_sqlite_rejects_it(self):
+        import asyncio
+
+        fs = self._frame_server(7)
+        fs.sqlite_handler.companion_push_message.return_value = False
+        asyncio.run(fs._persist_companion_message({"text": "x"}))
+        fs.bridge.message_queue.pop_last.assert_not_called()
+
 
 class TestImportRepeaterContactsCap:
     """The import endpoint must never leave persisted contacts above max_contacts.