diff --git a/burr/core/persistence.py b/burr/core/persistence.py
index c32bf8e96..0423d0d4f 100644
--- a/burr/core/persistence.py
+++ b/burr/core/persistence.py
@@ -491,11 +491,11 @@ def load(
         cursor = self.connection.cursor()
         try:
             if app_id is None:
-                # get latest for all app_ids
+                # CURRENT_TIMESTAMP has second precision; break ties by insertion order.
                 cursor.execute(
                     f"SELECT position, state, sequence_id, app_id, created_at, status FROM {self.table_name} "
                     f"WHERE partition_key = ? "
-                    f"ORDER BY CREATED_AT DESC LIMIT 1",
+                    f"ORDER BY created_at DESC, rowid DESC LIMIT 1",
                     (partition_key,),
                 )
             elif sequence_id is None:
diff --git a/burr/integrations/persisters/b_aiosqlite.py b/burr/integrations/persisters/b_aiosqlite.py
index 8439b965b..3508c442c 100644
--- a/burr/integrations/persisters/b_aiosqlite.py
+++ b/burr/integrations/persisters/b_aiosqlite.py
@@ -229,11 +229,11 @@ async def load(
         logger.debug("Loading %s, %s, %s", partition_key, app_id, sequence_id)
         cursor = await self.connection.cursor()
         if app_id is None:
-            # get latest for all app_ids
+            # CURRENT_TIMESTAMP has second precision; break ties by insertion order.
             await cursor.execute(
                 f"SELECT position, state, sequence_id, app_id, created_at, status FROM {self.table_name} "
                 f"WHERE partition_key = ? "
-                f"ORDER BY CREATED_AT DESC LIMIT 1",
+                f"ORDER BY created_at DESC, rowid DESC LIMIT 1",
                 (partition_key,),
             )
         elif sequence_id is None:
diff --git a/tests/core/test_persistence.py b/tests/core/test_persistence.py
index 18c65a0d1..88dbc78bc 100644
--- a/tests/core/test_persistence.py
+++ b/tests/core/test_persistence.py
@@ -131,6 +131,46 @@ def test_sqlite_persister_list_app_ids_without_initialize_raises_runtime_error()
         persister.cleanup()
 
 
+@pytest.mark.parametrize("partition_key", [None, "partition"])
+@pytest.mark.parametrize("latest_app,latest_sequence", [("a-app", 12), ("z-app", 0)])
+def test_sqlite_load_latest_breaks_timestamp_ties(
+    tmp_path, partition_key, latest_app, latest_sequence
+):
+    db_path = str(tmp_path / "checkpoints.db")
+    with SQLLitePersister(db_path=db_path, table_name="test_table") as persister:
+        persister.initialize()
+        persister.save(partition_key, "a-app", 11, "first", State({"step": 1}), "completed")
+        persister.save(
+            partition_key, latest_app, latest_sequence, "second", State({"step": 2}), "completed"
+        )
+        persister.save("other-partition", "other-app", 999, "other", State({}), "completed")
+        # Model saves within the same second without depending on wall-clock timing.
+        persister.connection.execute("UPDATE test_table SET created_at = '2026-01-01 00:00:00'")
+        persister.connection.commit()
+
+    with SQLLitePersister(db_path=db_path, table_name="test_table") as persister:
+        loaded = persister.load(partition_key, None)
+        assert loaded["app_id"] == latest_app
+        assert loaded["sequence_id"] == latest_sequence
+        assert loaded["position"] == "second"
+        assert loaded["state"] == State({"step": 2})
+        assert loaded["status"] == "completed"
+        assert persister.load(partition_key, "a-app", 11)["state"] == State({"step": 1})
+
+
+def test_sqlite_load_latest_prioritizes_timestamp(initializing_persistence):
+    persister = initializing_persistence
+    persister.initialize()
+    persister.save("partition", "a-app", 1, "first", State({"step": 1}), "completed")
+    persister.save("partition", "z-app", 0, "second", State({"step": 2}), "completed")
+    persister.connection.execute(
+        "UPDATE test_table SET created_at = CASE app_id "
+        "WHEN 'a-app' THEN '2026-01-02 00:00:00' ELSE '2026-01-01 00:00:00' END"
+    )
+    persister.connection.commit()
+    assert persister.load("partition", None)["state"] == State({"step": 1})
+
+
 @pytest.mark.parametrize(
     "method_name,kwargs",
     [
diff --git a/tests/integrations/persisters/test_b_aiosqlite.py b/tests/integrations/persisters/test_b_aiosqlite.py
index adb97532c..f1945a472 100644
--- a/tests/integrations/persisters/test_b_aiosqlite.py
+++ b/tests/integrations/persisters/test_b_aiosqlite.py
@@ -73,6 +73,52 @@ async def test_async_persistence_lists_app_ids(async_persistence):
     assert set(app_ids) == set(["app_id1", "app_id2"])
 
 
+@pytest.mark.parametrize("partition_key", [None, "partition"])
+@pytest.mark.parametrize("latest_app,latest_sequence", [("a-app", 12), ("z-app", 0)])
+async def test_async_sqlite_load_latest_breaks_timestamp_ties(
+    tmp_path, partition_key, latest_app, latest_sequence
+):
+    db_path = str(tmp_path / "checkpoints.db")
+    async with AsyncSQLitePersister.from_values(
+        db_path=db_path, table_name="test_table"
+    ) as persister:
+        await persister.initialize()
+        await persister.save(partition_key, "a-app", 11, "first", State({"step": 1}), "completed")
+        await persister.save(
+            partition_key, latest_app, latest_sequence, "second", State({"step": 2}), "completed"
+        )
+        await persister.save("other-partition", "other-app", 999, "other", State({}), "completed")
+        # Model saves within the same second without depending on wall-clock timing.
+        await persister.connection.execute(
+            "UPDATE test_table SET created_at = '2026-01-01 00:00:00'"
+        )
+        await persister.connection.commit()
+
+    async with AsyncSQLitePersister.from_values(
+        db_path=db_path, table_name="test_table"
+    ) as persister:
+        loaded = await persister.load(partition_key, None)
+        assert loaded["app_id"] == latest_app
+        assert loaded["sequence_id"] == latest_sequence
+        assert loaded["position"] == "second"
+        assert loaded["state"] == State({"step": 2})
+        assert loaded["status"] == "completed"
+        assert (await persister.load(partition_key, "a-app", 11))["state"] == State({"step": 1})
+
+
+async def test_async_sqlite_load_latest_prioritizes_timestamp(async_persistence):
+    persister = async_persistence
+    await persister.initialize()
+    await persister.save("partition", "a-app", 1, "first", State({"step": 1}), "completed")
+    await persister.save("partition", "z-app", 0, "second", State({"step": 2}), "completed")
+    await persister.connection.execute(
+        "UPDATE test_table SET created_at = CASE app_id "
+        "WHEN 'a-app' THEN '2026-01-02 00:00:00' ELSE '2026-01-01 00:00:00' END"
+    )
+    await persister.connection.commit()
+    assert (await persister.load("partition", None))["state"] == State({"step": 1})
+
+
 @pytest.mark.parametrize(
     "method_name,kwargs",
     [