From 737c2b5f3968df4d1a29149cdc80c10c3423982a Mon Sep 17 00:00:00 2001
From: bradjin8 <headit74@hotmail.com>
Date: Thu, 30 Jul 2026 16:26:02 -0400
Subject: [PATCH 1/2] initial implementation of #100

---
 CHANGELOG.md              |  1 +
 cli/localci/core/queue.py |  3 +-
 cli/tests/test_queue.py   | 71 +++++++++++++++++++++++++++++++++++++++
 3 files changed, 74 insertions(+), 1 deletion(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index bf1941c..0dfb4c1 100644
--- a/CHANGELOG.md
+++ b/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
 
diff --git a/cli/localci/core/queue.py b/cli/localci/core/queue.py
index 22441fd..5905b93 100644
--- a/cli/localci/core/queue.py
+++ b/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:
diff --git a/cli/tests/test_queue.py b/cli/tests/test_queue.py
index fcea09d..4c0bde8 100644
--- a/cli/tests/test_queue.py
+++ b/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,65 @@ 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 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):

From bc5707062ebe55f67e67a71c1558def1fb51bd8a Mon Sep 17 00:00:00 2001
From: bradjin8 <headit74@hotmail.com>
Date: Thu, 30 Jul 2026 17:23:57 -0400
Subject: [PATCH 2/2] fix: TOCTOU race

---
 cli/tests/test_queue.py | 3 +++
 1 file changed, 3 insertions(+)

diff --git a/cli/tests/test_queue.py b/cli/tests/test_queue.py
index 4c0bde8..f03711e 100644
--- a/cli/tests/test_queue.py
+++ b/cli/tests/test_queue.py
@@ -368,6 +368,9 @@ 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):