From 06c26e0a411c3a5a5ff17973c05df55f48b976f2 Mon Sep 17 00:00:00 2001
From: Vito Sansevero <vito.sansevero@gmail.com>
Date: Mon, 19 Oct 2026 10:23:20 -0700
Subject: [PATCH] test(history): hold the first concurrent writer open past the
 bounded wait (#242)

The concurrent replay tests expected every peer to succeed, but each peer
only waits 250 ms for BEGIN IMMEDIATE by design. On a slow Windows runner
three queued commits exceeded that and a peer raised 'database is
locked'. Holding the first writer's transaction open for 500 ms makes the
same failure deterministic on any host (20/20 locally).
---
 tests/test_execution_context_database.py | 13 +++++++++++-
 tests/test_execution_history_database.py | 25 ++++++++++++++++++++++++
 tests/test_execution_output_database.py  | 12 ++++++++++--
 3 files changed, 47 insertions(+), 3 deletions(-)

diff --git a/tests/test_execution_context_database.py b/tests/test_execution_context_database.py
index 5e9256e..64fcace 100644
--- a/tests/test_execution_context_database.py
+++ b/tests/test_execution_context_database.py
@@ -11,6 +11,7 @@ import textwrap
 from unittest.mock import patch
 import unittest
 
+from database import execution_contexts
 from database.execution_history import HistoryEpochChanged
 from database.execution_context_schema import context_schema_error
 from database.execution_contexts import LIVE_HINT_PROMPT_IDS_SQL
@@ -161,12 +162,22 @@ class ContextDatabaseTests(unittest.TestCase):
         for start in starts:
             db.write_execution_start(start)
 
+        hold = history_tests.FirstWriterHold()
+        write_capture = execution_contexts._write_capture
+
+        def held_open(*args, **kwargs):
+            hold()
+            return write_capture(*args, **kwargs)
+
         def write(start):
             return db.write_execution_contexts(
                 start.run_id, start.storage_epoch, bundle()
             )
 
-        with ThreadPoolExecutor(max_workers=4) as pool:
+        with (
+            patch.object(execution_contexts, "_write_capture", side_effect=held_open),
+            ThreadPoolExecutor(max_workers=4) as pool,
+        ):
             self.assertEqual(list(pool.map(write, starts)), [True] * 8)
         self.assertEqual(self.counts(db), (8, 1, 2, 8))
 
diff --git a/tests/test_execution_history_database.py b/tests/test_execution_history_database.py
index f308701..c43a69b 100644
--- a/tests/test_execution_history_database.py
+++ b/tests/test_execution_history_database.py
@@ -2,6 +2,8 @@
 
 import sqlite3
 import tempfile
+import threading
+import time
 import unittest
 from contextlib import closing
 from dataclasses import replace
@@ -13,6 +15,29 @@ from database.models import PromptModel
 from database.operations import PromptDatabase
 from utils.execution_history_events import make_finish, make_start
 
+# Longer than the 250 ms bounded writer wait, as a slow Windows commit can be.
+WRITER_HOLD_SECONDS = 0.5
+
+
+class FirstWriterHold:
+    """Keep the first write transaction that calls it open for a while.
+
+    Concurrent writer tests use it so the other peers really queue behind an
+    open BEGIN IMMEDIATE, instead of depending on how fast the host commits.
+    """
+
+    def __init__(self, seconds=WRITER_HOLD_SECONDS):
+        self.seconds = seconds
+        self._lock = threading.Lock()
+        self._held = False
+
+    def __call__(self):
+        with self._lock:
+            hold, self._held = not self._held, True
+        if hold:
+            time.sleep(self.seconds)
+
+
 SESSION = "11111111-1111-4111-8111-111111111111"
 JOB = "22222222-2222-4222-8222-222222222222"
 
diff --git a/tests/test_execution_output_database.py b/tests/test_execution_output_database.py
index 080844b..af16bfd 100644
--- a/tests/test_execution_output_database.py
+++ b/tests/test_execution_output_database.py
@@ -144,11 +144,19 @@ class OutputDatabaseTests(unittest.TestCase):
         finish = self.finish(start)
         peers = [self.database() for _ in range(4)]
         barrier = Barrier(len(peers))
+        hold = history_tests.FirstWriterHold()
+
+        @contextmanager
+        def held_open():
+            hold()
+            yield True
 
         def write(index):
-            barrier.wait(timeout=5)
+            barrier.wait(timeout=60)
             return peers[index].write_execution_outputs_for_finish(
-                finish, bundle("first.png" if index % 2 else "second.png")
+                finish,
+                bundle("first.png" if index % 2 else "second.png"),
+                generation_guard=held_open,
             )
 
         with ThreadPoolExecutor(max_workers=4) as executor: