fix: take lock in PriorityJobQueue.is_done

CHANGELOG.md

@@ -25,6 +25,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Fixed
 
 - **`derive_image_tag()` prefix** — honours `project.native_image_prefix` instead of hardcoding `capy-`, completing the generic-profile default behaviour documented for Week 30.
+- **`PriorityJobQueue.is_done`** — acquires `self._lock`, fixing a torn-read race when the dispatch loop polls completion during concurrent enqueue/complete.
 
 ## [0.1.0] - TBD
 

cli/localci/core/queue.py

@@ -351,7 +351,8 @@ def is_empty(self) -> bool:
 
     @property
     def is_done(self) -> bool:
-        return len(self._jobs) > 0 and len(self._completed_keys) >= len(self._jobs)
+        with self._lock:
+            return len(self._jobs) > 0 and len(self._completed_keys) >= len(self._jobs)
 
     @property
     def total_jobs(self) -> int:

cli/tests/test_queue.py

@@ -10,6 +10,7 @@
 from localci.core.models import (
     JobEventType,
     QueuedJob,
+    QueuedJobStatus,
 )
 from localci.core.queue import (
     CyclicDependencyError,
@@ -322,6 +323,17 @@ def test_from_config_reads_priorities(self):
 # Thread safety
 # ---------------------------------------------------------------------------
 
+_TERMINAL_STATUSES = frozenset(
+    {
+        QueuedJobStatus.PASSED,
+        QueuedJobStatus.FAILED,
+        QueuedJobStatus.CANCELLED,
+        QueuedJobStatus.SKIPPED,
+        QueuedJobStatus.ERROR,
+        QueuedJobStatus.TIMEOUT,
+    }
+)
+
 
 class TestThreadSafety:
     def test_concurrent_enqueue(self):
@@ -341,6 +353,68 @@ def enqueue_batch(start: int, count: int) -> None:
             t.join()
         assert queue.total_jobs == 100
 
+    def test_concurrent_is_done_invariant(self):
+        queue = PriorityJobQueue()
+        stop = threading.Event()
+        producers_done = threading.Event()
+        violations: list[str] = []
+        violations_lock = threading.Lock()
+
+        def record(msg: str) -> None:
+            with violations_lock:
+                violations.append(msg)
+
+        def monitor() -> None:
+            while not stop.is_set():
+                if queue.is_done:
+                    jobs = queue.get_all_jobs()
+                    if not queue.is_done:
+                        time.sleep(0.001)
+                        continue
+                    if not jobs:
+                        record("is_done True with empty queue")
+                    elif not all(j.status in _TERMINAL_STATUSES for j in jobs):
+                        record("is_done True with non-terminal jobs")
+                time.sleep(0.001)
+
+        def enqueue_batch(start: int, count: int) -> None:
+            for i in range(start, start + count):
+                queue.enqueue(make_job(f"Job {i}", priority=1, index=i))
+                time.sleep(0.001)
+
+        def consume() -> None:
+            while not (producers_done.is_set() and queue.is_done):
+                job = queue.next_ready()
+                if job is None:
+                    time.sleep(0.001)
+                    continue
+                queue.mark_running(job)
+                queue.mark_completed(job, success=True)
+
+        producers = [
+            threading.Thread(target=enqueue_batch, args=(0, 50)),
+            threading.Thread(target=enqueue_batch, args=(50, 50)),
+        ]
+        consumers = [threading.Thread(target=consume) for _ in range(4)]
+        monitor_thread = threading.Thread(target=monitor)
+
+        monitor_thread.start()
+        for t in producers:
+            t.start()
+        for t in consumers:
+            t.start()
+
+        for t in producers:
+            t.join()
+        producers_done.set()
+        for t in consumers:
+            t.join()
+        stop.set()
+        monitor_thread.join()
+
+        assert not violations, violations
+        assert queue.is_done is True
+
     def test_concurrent_dequeue(self):
         queue = PriorityJobQueue()
         for i in range(20):