test(history): bound batch drains by progress and total time

tests/history_waits.py

@@ -0,0 +1,36 @@
+"""Finite test waits for real SQLite batches without a throughput requirement."""
+
+import threading
+import time
+from typing import Callable
+
+from utils.execution_history_writer import HistoryWriter
+
+
+def drain_writer(
+    writer: HistoryWriter, *, clock: Callable[[], float] = time.monotonic
+) -> None:
+    initial = current = writer.status()
+    started = last_progress = clock()
+    lowest_pending = initial.pending
+    while current.pending:
+        now = clock()
+        if current.pending < lowest_pending:
+            lowest_pending = current.pending
+            last_progress = now
+        reason = None
+        if not current.alive:
+            reason = "worker stopped"
+        elif now - started >= 30:
+            reason = "batch deadline"
+        elif now - last_progress >= 3:
+            reason = "worker stalled"
+        if reason is not None:
+            raise AssertionError(
+                f"history drain {reason}: elapsed={now - started:.3f}s, "
+                f"stalled={now - last_progress:.3f}s, "
+                f"initial={initial}, current={current}, "
+                f"charged_bytes={writer.context_budget.charged_bytes}"
+            )
+        threading.Event().wait(0.005)
+        current = writer.status()

tests/test_execution_history_hooks.py

@@ -16,6 +16,7 @@ from types import SimpleNamespace
 import unittest
 from unittest.mock import patch
 
+from tests.history_waits import drain_writer
 from database import PromptDatabase
 from utils import usage_tracking
 from utils.execution_history_writer import HistoryWriter
@@ -117,7 +118,7 @@ class HistoryHooksTests(unittest.TestCase):
         ]
 
     def drain(self, registration):
-        self.wait_for(lambda: registration.writer.status().pending == 0)
+        drain_writer(registration.writer, clock=time.monotonic)
 
     def test_package_history_initializes_when_monitor_and_usage_setup_fail(self):
         try:

tests/test_execution_history_writer.py

@@ -13,6 +13,7 @@ import unittest
 from unittest.mock import patch
 from uuid import uuid4
 
+from tests.history_waits import drain_writer
 from database import PromptDatabase
 from utils.execution_history_events import (
     LifecycleEvent,
@@ -696,7 +697,7 @@ class HistoryWriterTests(unittest.TestCase):
                 "prompt_manager.execution_history_writer", "WARNING"
             ) as logs:
                 release.set()
-                self.wait_for(lambda: writer.status().pending == 0)
+                drain_writer(writer)
                 self.assertTrue(writer.close(2))
             self.assertEqual(len(logs.records), 1)
 

tests/test_history_waits.py

@@ -0,0 +1,90 @@
+"""Batch drain detects stalls without imposing a SQLite throughput target."""
+
+from dataclasses import replace
+from types import SimpleNamespace
+import unittest
+from unittest.mock import patch
+
+from tests import test_execution_history_hooks as hooks
+from utils.execution_history_writer import HistoryWriterStatus
+
+
+class StepClock:
+    def __init__(self, step: float):
+        self.value = 0.0
+        self.step = step
+
+    def __call__(self) -> float:
+        self.value += self.step
+        return self.value
+
+
+class ObservedWriter:
+    def __init__(self, pending: int, *, progress: bool = False, alive: bool = True):
+        self.pending = pending
+        self.progress = progress
+        self.alive = alive
+        self.context_budget = SimpleNamespace(charged_bytes=281)
+
+    def status(self) -> HistoryWriterStatus:
+        status = HistoryWriterStatus("epoch", True, self.pending, 0, self.alive, 0)
+        if self.progress and self.pending:
+            self.pending -= 1
+        return status
+
+
+class HistoryDrainTests(unittest.TestCase):
+    wait_for = hooks.HistoryHooksTests.wait_for
+
+    def drain(self, writer: ObservedWriter, clock: StepClock) -> None:
+        with patch.object(hooks, "time", SimpleNamespace(monotonic=clock)):
+            hooks.HistoryHooksTests.drain(self, SimpleNamespace(writer=writer))
+
+    def test_batch_with_progress_can_exceed_short_stall_deadline(self):
+        writer = ObservedWriter(128, progress=True)
+        clock = StepClock(0.04)
+        self.drain(writer, clock)
+        self.assertEqual(writer.pending, 0)
+        self.assertGreater(clock.value, 3)
+        self.assertLess(clock.value, 30)
+
+    def test_no_progress_fails_with_worker_and_lease_diagnostics(self):
+        writer = ObservedWriter(52)
+        clock = StepClock(0.5)
+        with self.assertRaisesRegex(
+            AssertionError, r"worker stalled.*pending=52.*charged_bytes=281"
+        ):
+            self.drain(writer, clock)
+        self.assertLessEqual(clock.value, 4)
+
+    def test_progress_cannot_extend_batch_past_hard_ceiling(self):
+        writer = ObservedWriter(1000, progress=True)
+        clock = StepClock(0.5)
+        with self.assertRaisesRegex(AssertionError, "batch deadline"):
+            self.drain(writer, clock)
+        self.assertGreater(writer.pending, 0)
+        self.assertGreaterEqual(clock.value, 30)
+        self.assertLessEqual(clock.value, 31)
+
+    def test_dead_worker_with_pending_work_fails_immediately(self):
+        writer = ObservedWriter(1, alive=False)
+        clock = StepClock(0.5)
+        with self.assertRaisesRegex(AssertionError, r"worker stopped.*alive=False"):
+            self.drain(writer, clock)
+        self.assertLess(clock.value, 3)
+
+    def test_oscillating_pending_is_not_new_progress(self):
+        writer = ObservedWriter(1)
+        original = writer.status
+        calls = 0
+
+        def oscillating() -> HistoryWriterStatus:
+            nonlocal calls
+            calls += 1
+            return replace(original(), pending=1 + calls % 2)
+
+        writer.status = oscillating
+        clock = StepClock(0.5)
+        with self.assertRaisesRegex(AssertionError, "worker stalled"):
+            self.drain(writer, clock)
+        self.assertLessEqual(clock.value, 5)