Merge pull request #180 from game-by-virtuals/refactor/example-threading-lock

Refactor/example threading lock

Author: celesteanglm

plugins/acp/examples/reactive/buyer.py

@@ -1,3 +1,5 @@
+import threading
+
 from typing import Tuple
 from game_sdk.game.agent import Agent, WorkerConfig
 from game_sdk.game.custom_types import Argument, Function, FunctionResultStatus
@@ -53,32 +55,81 @@ def on_evaluate(job: ACPJob):
 #     },
 # }
 
-def buyer():
-    # upon phase change, the buyer agent will respond to the transaction
+def buyer(use_thread_lock: bool = True):
+    if env.WHITELISTED_WALLET_PRIVATE_KEY is None:
+        return
+    
+    if env.BUYER_ENTITY_ID is None:
+        return
+    
+    # Thread-safe job queue setup
+    job_queue = []
+    job_queue_lock = threading.Lock()
+    job_event = threading.Event()
+
+    # Thread-safe append with optional lock
+    def safe_append_job(job):
+        if use_thread_lock:
+            print(f"[safe_append_job] Acquiring lock to append job {job.id}")
+            with job_queue_lock:
+                print(f"[safe_append_job] Lock acquired, appending job {job.id} to queue")
+                job_queue.append(job)
+        else:
+            job_queue.append(job)
+
+    # Thread-safe pop with optional lock
+    def safe_pop_job():
+        if use_thread_lock:
+            print(f"[safe_pop_job] Acquiring lock to pop job")
+            with job_queue_lock:
+                if job_queue:
+                    job = job_queue.pop(0)
+                    print(f"[safe_pop_job] Lock acquired, popped job {job.id}")
+                    return job
+                else:
+                    print("[safe_pop_job] Queue is empty after acquiring lock")
+        else:
+            if job_queue:
+                job = job_queue.pop(0)
+                print(f"[safe_pop_job] Popped job {job.id} without lock")
+                return job
+            else:
+                print("[safe_pop_job] Queue is empty (no lock)")
+        return None
+
+    # Background thread worker: process jobs one by one
+    def job_worker():
+        while True:
+            job_event.wait()  # Wait for job
+
+            job = safe_pop_job()
+            while job:
+                process_job(job)
+                job = safe_pop_job()
+
+            job_event.clear()  # Go back to wait
+
+    # Event-triggered job task receiver
     def on_new_task(job: ACPJob):
+        print(f"[on_new_task] Received job {job.id} (phase: {job.phase})")
+        safe_append_job(job)
+        job_event.set()
+
+    def process_job(job: ACPJob):
         out = ""
         print(job.phase, "job.phase")
         if job.phase == ACPJobPhase.NEGOTIATION:
             for memo in job.memos:
                 print(memo.next_phase, "memo.next_phase")
                 if memo.next_phase == ACPJobPhase.TRANSACTION:
                     out += f"Buyer agent is reacting to job:\n{job}\n\n"
-        
                     buyer_agent.get_worker("acp_worker").run(
                         f"Respond to the following transaction: {job}",
                     )
-        
                     out += "Buyer agent has responded to the job\n"
+
         print(Panel(out, title="🔁 Reaction", box=box.ROUNDED, title_align="left", border_style="red"))
 
-    
-    if env.WHITELISTED_WALLET_PRIVATE_KEY is None:
-        return
-    
-    if env.BUYER_ENTITY_ID is None:
-        return
-    
-    
     acp_plugin = AcpPlugin(
         options=AcpPluginOptions(
             api_key=env.GAME_API_KEY,
@@ -185,15 +236,24 @@ def post_tweet(content: str, reasoning: str) -> Tuple[FunctionResultStatus, str,
     buyer_agent.compile()
     agent.compile()
 
+    # Start background job thread
+    threading.Thread(target=job_worker, daemon=True).start()
+
     while True:
         print("🟢"*40)
         init_state = AcpState.model_validate(agent.agent_state)
         print(Panel(f"{init_state}", title="Agent State", box=box.ROUNDED, title_align="left"))
-        agent.step()
+        
+        print("[agent.step] Attempting to acquire lock for agent.step()")
+        with job_queue_lock:
+            print("[agent.step] Lock acquired, executing agent.step()")
+            agent.step()
+        print("[agent.step] Released lock after agent.step()")
+
         end_state = AcpState.model_validate(agent.agent_state)
         print(Panel(f"{end_state}", title="End Agent State", box=box.ROUNDED, title_align="left"))
         print("🔴"*40)
         input("\nPress any key to continue...\n")
 
 if __name__ == "__main__":
-    buyer()
+    buyer(use_thread_lock=True)

plugins/acp/examples/reactive/seller.py

@@ -19,6 +19,7 @@
 # from twitter_plugin_gamesdk.twitter_plugin import TwitterPlugin
 
 load_dotenv(override=True)
+
 env = PluginEnvSettings()
 
 
@@ -46,11 +47,79 @@
 #     },
 # }
 
-def seller():
+def seller(use_thread_lock: bool = True):
+    if env.WHITELISTED_WALLET_PRIVATE_KEY is None:
+        return
+    
+    if env.SELLER_ENTITY_ID is None:
+        return
+
+    # Thread-safe job queue setup
+    job_queue = []
+    job_queue_lock = threading.Lock()
+    job_event = threading.Event()
+
+    # Thread-safe append wrapper
+    def safe_append_job(job):
+        if use_thread_lock:
+            print("[append] Attempting to acquire job_queue_lock")
+            with job_queue_lock:
+                print("[append] Lock acquired. Appending job to queue:", job.id)
+                job_queue.append(job)
+                print(f"[append] Queue size is now {len(job_queue)}")
+        else:
+            job_queue.append(job)
+            print(f"[append] Appended job (no lock). Queue size is now {len(job_queue)}")
+
+    # Thread-safe pop wrapper
+    def safe_pop_job():
+        if use_thread_lock:
+            print("[pop] Attempting to acquire job_queue_lock")
+            with job_queue_lock:
+                print("[pop] Lock acquired.")
+                if job_queue:
+                    job = job_queue.pop(0)
+                    print(f"[pop] Job popped: {job.id}")
+                    return job
+                else:
+                    print("[pop] Queue is empty.")
+        else:
+            if job_queue:
+                job = job_queue.pop(0)
+                print(f"[pop] Job popped (no lock): {job.id}")
+                return job
+            else:
+                print("[pop] Queue is empty (no lock).")
+        return None
+
+    # 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)
+                job = safe_pop_job()
+
+            job_event.clear()
+            print("[worker] All jobs processed. Waiting again.")
+
+    # Event-triggered job task receiver
     def on_new_task(job: ACPJob):
+        print(f"[on_new_task] New job received: {job.id}")
+        safe_append_job(job)
+        job_event.set()
+        print("[on_new_task] job_event set.")
+
+    def process_job(job: ACPJob):
         out = ""
         out += f"Reacting to job:\n{job}\n\n"
         prompt = ""
+
         if job.phase == ACPJobPhase.REQUEST:
             for memo in job.memos:
                 if memo.next_phase == ACPJobPhase.NEGOTIATION:
@@ -81,12 +150,6 @@ def on_new_task(job: ACPJob):
             out += "✅ Seller has responded to job.\n"
 
         print(Panel(out, title="🔁 Reaction", box=box.ROUNDED, title_align="left", border_style="red"))
-
-    if env.WHITELISTED_WALLET_PRIVATE_KEY is None:
-        return
-    
-    if env.SELLER_ENTITY_ID is None:
-        return
 
     acp_plugin = AcpPlugin(
         options=AcpPluginOptions(
@@ -187,9 +250,12 @@ def generate_meme(description: str, job_id: int, reasoning: str) -> Tuple[Functi
     init_state = AcpState.model_validate(agent.agent_state)
     print(Panel(f"{init_state}", title="Agent State", box=box.ROUNDED, title_align="left"))
     print("🔴"*40)
-    print("\nListening\n")
+
+     # Start background thread
+    threading.Thread(target=job_worker, daemon=True).start()
+    print("\nListening...\n")
     threading.Event().wait()
 
 
 if __name__ == "__main__":
-    seller()
+    seller(use_thread_lock=True)

plugins/acp/pyproject.toml

@@ -12,7 +12,7 @@ game-sdk = ">=0.1.5"
 python-dotenv = "^1.1.0"
 dacite = "^1.9.2"
 rich = ">=13.9.4,<15.0.0"
-virtuals-acp = "^0.1.16"
+virtuals-acp = "^0.1.17"
 
 [build-system]
 requires = ["poetry-core"]