fix: race condition bug for threading lock example

Author: celesteanglm

plugins/acp/examples/reactive/buyer.py

@@ -102,12 +102,25 @@ def job_worker():
         while True:
             job_event.wait()  # Wait for job
 
-            job = safe_pop_job()
-            while job:
-                process_job(job)
+            # Process all available jobs
+            while True:
                 job = safe_pop_job()
-
-            job_event.clear()  # Go back to wait
+                if not job:
+                    break
+                try:
+                    process_job(job)
+                except Exception as e:
+                    print(f"❌ Error processing job: {e}")
+                    # Continue processing other jobs even if one fails
+
+            # Clear event only after ensuring no jobs remain
+            if use_thread_lock:
+                with job_queue_lock:
+                    if not job_queue:
+                        job_event.clear()
+            else:
+                if not job_queue:
+                    job_event.clear()
 
     # Event-triggered job task receiver
     def on_new_task(job: ACPJob):
@@ -200,7 +213,7 @@ def post_tweet(content: str, reasoning: str) -> Tuple[FunctionResultStatus, str,
         agent_goal="Finding the best meme to do tweet posting",
         agent_description=f"""
         Agent that gain market traction by posting meme. Your interest are in cats and AI. 
-        You can head to acp to look for agents to help you generating meme.
+        You can head to acp to look for devrel_seller to help you generating meme.
         Do not look for a relevant validator to validate the deliverable.
 
         {acp_plugin.agent_description}

plugins/acp/examples/reactive/seller.py

@@ -8,6 +8,7 @@
 from virtuals_acp import ACPJob, ACPJobPhase
 from game_sdk.game.custom_types import Argument, Function, FunctionResultStatus
 from game_sdk.game.agent import Agent
+from collections import deque
 from rich import print, box
 from rich.panel import Panel
 from dotenv import load_dotenv
@@ -55,7 +56,7 @@ def seller(use_thread_lock: bool = True):
         return
 
     # Thread-safe job queue setup
-    job_queue = []
+    job_queue = deque()
     job_queue_lock = threading.Lock()
     job_event = threading.Event()
 
@@ -78,14 +79,14 @@ def safe_pop_job():
             with job_queue_lock:
                 print("[pop] Lock acquired.")
                 if job_queue:
-                    job = job_queue.pop(0)
+                    job = job_queue.popleft()
                     print(f"[pop] Job popped: {job.id}")
                     return job
                 else:
                     print("[pop] Queue is empty.")
         else:
             if job_queue:
-                job = job_queue.pop(0)
+                job = job_queue.popleft()
                 print(f"[pop] Job popped (no lock): {job.id}")
                 return job
             else:
@@ -94,19 +95,28 @@ def safe_pop_job():
 
     # Background thread worker: process jobs one by one
     def job_worker():
-        print("[worker] Job worker started, waiting for jobs.")
         while True:
             job_event.wait()
-            print("[worker] job_event triggered.")
 
-            job = safe_pop_job()
-            while job:
-                print(f"[worker] Processing job {job.id}")
-                process_job(job)
+            # Process all available jobs
+            while True:
                 job = safe_pop_job()
-
-            job_event.clear()
-            print("[worker] All jobs processed. Waiting again.")
+                if not job:
+                    break
+                try:
+                    process_job(job)
+                except Exception as e:
+                    print(f"❌ Error processing job: {e}")
+                    # Continue processing other jobs even if one fails
+
+            # Clear event only after ensuring no jobs remain
+            if use_thread_lock:
+                with job_queue_lock:
+                    if not job_queue:
+                        job_event.clear()
+            else:
+                if not job_queue:
+                    job_event.clear()
 
     # Event-triggered job task receiver
     def on_new_task(job: ACPJob):