V2 merge: purpose_agent/unified.py

purpose_agent/unified.py

@@ -73,6 +73,11 @@ END = "__END__"
 START = "__START__"
 
 
+import threading
+
+# Global lock for shared replay/optimizer in parallel execution
+_parallel_lock = threading.Lock()
+
 # ═══════════════════════════════════════════════════════════════════════════
 # 1. PLUG-AND-PLAY — Agent() one-liner factory (OpenAI Agents SDK simplicity)
 # ═══════════════════════════════════════════════════════════════════════════
@@ -447,7 +452,22 @@ def parallel(
             agent_list = agents
 
     states = initial_states or [None] * len(normalized)
-    workers = max_workers or min(len(normalized), 8)
+
+    # Thread safety: detect backend type for concurrency limit
+    # Local backends (Ollama, llama-cpp) share one GPU/CPU — serialize
+    # Cloud/API backends can parallelize
+    if max_workers is None:
+        sample_agent = agent_list[0] if agent_list else None
+        if sample_agent and hasattr(sample_agent, 'llm'):
+            backend_type = type(sample_agent.llm).__name__
+            if backend_type in ("OllamaBackend", "LlamaCppBackend", "MockLLMBackend"):
+                workers = 1  # Local model — serialize to avoid contention
+            else:
+                workers = min(len(normalized), 8)
+        else:
+            workers = min(len(normalized), 8)
+    else:
+        workers = max_workers
 
     logger.info(f"Parallel: Running {len(normalized)} tasks with {workers} workers")
 
@@ -455,7 +475,9 @@ def parallel(
         task = normalized[idx]
         agent = agent_list[idx]
         state = states[idx]
-        return agent.run(task["purpose"], state=state)
+        # Lock around shared replay/optimizer writes
+        with _parallel_lock:
+            return agent.run(task["purpose"], state=state)
 
     results: list[TaskResult | None] = [None] * len(normalized)