diff --git a/datasette/database.py b/datasette/database.py
index dfca179c..589c2fe6 100644
--- a/datasette/database.py
+++ b/datasette/database.py
@@ -148,10 +148,11 @@ class Database:
             if self._write_connection is None:
                 self._write_connection = self.connect(write=True)
                 self.ds._prepare_connection(self._write_connection, self.name)
-            return fn(self._write_connection)
+            result = fn(self._write_connection)
+            return result if block else uuid.uuid4()
 
         # threaded mode
-        task_id = uuid.uuid5(uuid.NAMESPACE_DNS, "datasette.io")
+        task_id = uuid.uuid4()
         if self._write_queue is None:
             self._write_queue = queue.Queue()
         if self._write_thread is None:
diff --git a/docs/changelog.rst b/docs/changelog.rst
index 56d1952b..9cac5297 100644
--- a/docs/changelog.rst
+++ b/docs/changelog.rst
@@ -4,6 +4,11 @@
 Changelog
 =========
 
+0.65.x (unreleased)
+-------------------
+
+* Backported the non-blocking write task ID fixes from :issue:`2861` by `Zain Dana Harper (HarperZ9) <https://github.com/HarperZ9>`__. ``execute_write_fn(block=False)`` now returns a unique UUID for each call, including when ``num_sql_threads=0``.
+
 .. _v0_65_3:
 
 0.65.3 (2026-08-06)
diff --git a/tests/test_internals_database.py b/tests/test_internals_database.py
index ad3f1d7a..4c0873d7 100644
--- a/tests/test_internals_database.py
+++ b/tests/test_internals_database.py
@@ -2,6 +2,7 @@
 Tests for the datasette.database.Database class
 """
 
+from datasette.app import Datasette
 from datasette.database import Database, Results, MultipleValues
 from datasette.utils.sqlite import sqlite3
 from datasette.utils import Column
@@ -458,6 +459,29 @@ async def test_execute_write_fn_block_false(db):
     assert isinstance(task_id, uuid.UUID)
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("num_sql_threads", (0, 3))
+async def test_execute_write_fn_block_false_returns_unique_uuid(num_sql_threads):
+    ds = Datasette([], memory=True, settings={"num_sql_threads": num_sql_threads})
+    db = ds.add_memory_database("test_block_false_{}".format(num_sql_threads))
+    await db.execute_write("create table t (id integer primary key)")
+
+    def write_fn(conn):
+        with conn:
+            conn.execute("insert into t default values")
+
+    first = await db.execute_write_fn(write_fn, block=False)
+    second = await db.execute_write_fn(write_fn, block=False)
+    assert isinstance(first, uuid.UUID)
+    assert isinstance(second, uuid.UUID)
+    assert first != second
+    # A blocking write runs after both queued writes on the same connection.
+    count = await db.execute_write_fn(
+        lambda conn: conn.execute("select count(*) from t").fetchone()[0]
+    )
+    assert count == 2
+
+
 @pytest.mark.asyncio
 async def test_execute_write_fn_block_true(db):
     def write_fn(conn):