feat(83): Phase 3 — Performance & Scale (#83) (#101)

* feat(83): Phase 3.1 — persistent JVM worker pool in MultiThreadBatchRunner (#83)

* feat(83): Phase 3.2 — async Forge episode generator with producer-consumer pipeline (#83)

* feat(83): Phase 3.3 — deprecate self_play.py synthetic mode for training (#83)

Mark self_play.py synthetic training as deprecated per Phase 3.3:
- Add deprecation notice to module docstring
- Add import warnings and _TRAINING_DEPRECATION_MSG constant
- Add DeprecationWarning in collect_rollouts()
- Synthetic sim retained for local testing/unit tests only
- scope.py already defaults forge_weight=1.0, ppo_weight=0.0

* feat(83): Phase 3.3 — update ml/training/__init__.py exports (#83)

ml/training/__init__.py

@@ -0,0 +1,29 @@
+"""
+Commander AI Lab - Training Subpackage
+======================================
+Provides RL training infrastructure including PPO, rollout buffers,
+Forge episode generation, self-play, and distillation.
+"""
+
+from ml.training.rollout_buffer import RolloutBuffer, ForgeEpisodeQueue
+from ml.training.reward import compute_intermediate_reward, compute_terminal_reward, RewardConfig
+from ml.training.decision_exporter import DecisionExporter
+from ml.training.forge_episode_generator import (
+    ForgeEpisodeGenerator,
+    AsyncForgeEpisodeGenerator,
+)
+
+__all__ = [
+    # Rollout & Buffers
+    "RolloutBuffer",
+    "ForgeEpisodeQueue",
+    # Rewards
+    "compute_intermediate_reward",
+    "compute_terminal_reward",
+    "RewardConfig",
+    # Decision export
+    "DecisionExporter",
+    # Forge episode generation (Phase 3)
+    "ForgeEpisodeGenerator",
+    "AsyncForgeEpisodeGenerator",
+]

ml/training/forge_episode_generator.py

@@ -0,0 +1,352 @@
+"""
+Commander AI Lab — Forge Episode Generator (Phase 3.2)
+
+Async producer-consumer pipeline: runs Forge batch games in parallel
+with PPO training.  Forge produces episodes → shared queue → PPO
+trainer consumes.
+
+Uses the existing watchdog (_global_watchdog_loop in forge_runner.py)
+to monitor Forge worker processes for stalls.
+"""
+
+import asyncio
+import json
+import logging
+import os
+import time
+from collections import deque
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any, Deque, Dict, List, Optional, Tuple
+
+import numpy as np
+
+from ml.config.scope import (
+    ACTION_TO_IDX,
+    GAME_SCOPE,
+    IDX_TO_ACTION,
+    MacroAction,
+    NUM_ACTIONS,
+    STATE_DIMS,
+)
+from ml.training.reward import (
+    RewardConfig,
+    compute_intermediate_reward,
+    compute_terminal_reward,
+)
+
+logger = logging.getLogger("ml.forge_episode_generator")
+
+
+# ══════════════════════════════════════════════════════════
+# Episode / Trajectory data structures
+# ══════════════════════════════════════════════════════════
+
+@dataclass
+class ForgeTransition:
+    """Single (s, a, r, log_prob, value, done) tuple from a Forge game."""
+    state: np.ndarray
+    action: int
+    reward: float
+    log_prob: float
+    value: float
+    done: bool
+
+
+@dataclass
+class ForgeEpisode:
+    """Complete trajectory from a single Forge game."""
+    game_id: str
+    transitions: List[ForgeTransition] = field(default_factory=list)
+    winner: Optional[int] = None
+    total_turns: int = 0
+    deck_names: List[str] = field(default_factory=list)
+
+    @property
+    def length(self) -> int:
+        return len(self.transitions)
+
+    @property
+    def total_reward(self) -> float:
+        return sum(t.reward for t in self.transitions)
+
+
+# ══════════════════════════════════════════════════════════
+# JSONL Parser — reads Forge ML decision snapshots
+# ══════════════════════════════════════════════════════════
+
+def parse_forge_jsonl(path: Path) -> List[Dict[str, Any]]:
+    """Parse a JSONL file of Forge ML decision snapshots.
+
+    Each line is a JSON object with keys like:
+        game_id, turn, phase, active_seat, state, action, outcome
+    """
+    snapshots = []
+    with open(path, "r") as f:
+        for lineno, line in enumerate(f, 1):
+            line = line.strip()
+            if not line:
+                continue
+            try:
+                snapshots.append(json.loads(line))
+            except json.JSONDecodeError as e:
+                logger.warning("Skipping malformed line %d in %s: %s", lineno, path, e)
+    logger.info("Parsed %d snapshots from %s", len(snapshots), path)
+    return snapshots
+
+
+def snapshots_to_episode(
+    snapshots: List[Dict[str, Any]],
+    reward_config: Optional[RewardConfig] = None,
+) -> ForgeEpisode:
+    """Convert a sequence of Forge snapshots into a ForgeEpisode.
+
+    Maps Forge action labels → MacroAction indices and computes
+    rewards from game outcomes using ml.training.reward.
+    """
+    if not snapshots:
+        return ForgeEpisode(game_id="empty")
+
+    first = snapshots[0]
+    episode = ForgeEpisode(
+        game_id=first.get("game_id", "unknown"),
+        deck_names=first.get("deck_names", []),
+    )
+
+    rc = reward_config or RewardConfig()
+
+    for i, snap in enumerate(snapshots):
+        # State vector — zero-padded if not present
+        state_vec = np.zeros(STATE_DIMS.total_state_dim, dtype=np.float32)
+        if "state_vector" in snap:
+            raw = np.array(snap["state_vector"], dtype=np.float32)
+            state_vec[: min(len(raw), len(state_vec))] = raw[: len(state_vec)]
+
+        # Action mapping
+        action_label = snap.get("action", "pass").lower().replace(" ", "_")
+        try:
+            macro = MacroAction(action_label)
+            action_idx = ACTION_TO_IDX[macro]
+        except (ValueError, KeyError):
+            action_idx = ACTION_TO_IDX[MacroAction.PASS]
+
+        # Reward
+        is_done = i == len(snapshots) - 1
+        if is_done and "outcome" in snap:
+            reward = compute_terminal_reward(
+                won=snap["outcome"].get("won", False),
+                turns=snap.get("turn", 1),
+                life_remaining=snap["outcome"].get("life_remaining", 0),
+                config=rc,
+            )
+        else:
+            reward = compute_intermediate_reward(
+                life_delta=snap.get("life_delta", 0),
+                cards_drawn=snap.get("cards_drawn", 0),
+                creatures_played=snap.get("creatures_played", 0),
+                config=rc,
+            )
+
+        episode.transitions.append(ForgeTransition(
+            state=state_vec,
+            action=action_idx,
+            reward=reward,
+            log_prob=snap.get("log_prob", 0.0),
+            value=snap.get("value", 0.0),
+            done=is_done,
+        ))
+
+    episode.total_turns = snapshots[-1].get("turn", len(snapshots))
+    if "outcome" in snapshots[-1]:
+        episode.winner = snapshots[-1]["outcome"].get("winner_seat")
+
+    return episode
+
+
+# ══════════════════════════════════════════════════════════
+# Async Episode Queue (producer-consumer)
+# ══════════════════════════════════════════════════════════
+
+class ForgeEpisodeQueue:
+    """Thread-safe async queue bridging Forge producers and PPO consumer.
+
+    Forge batch workers produce ForgeEpisodes; the PPO trainer consumes
+    them as they become available.  Backpressure is applied when the
+    queue reaches max_size.
+    """
+
+    def __init__(self, max_size: int = 256):
+        self._queue: asyncio.Queue[ForgeEpisode] = asyncio.Queue(maxsize=max_size)
+        self._produced = 0
+        self._consumed = 0
+        self._closed = False
+
+    async def put(self, episode: ForgeEpisode) -> None:
+        await self._queue.put(episode)
+        self._produced += 1
+
+    async def get(self, timeout: float = 30.0) -> Optional[ForgeEpisode]:
+        try:
+            ep = await asyncio.wait_for(self._queue.get(), timeout=timeout)
+            self._consumed += 1
+            return ep
+        except asyncio.TimeoutError:
+            return None
+
+    def close(self) -> None:
+        self._closed = True
+
+    @property
+    def is_closed(self) -> bool:
+        return self._closed and self._queue.empty()
+
+    @property
+    def stats(self) -> Dict[str, int]:
+        return {
+            "produced": self._produced,
+            "consumed": self._consumed,
+            "pending": self._queue.qsize(),
+        }
+
+
+# ══════════════════════════════════════════════════════════
+# Forge Batch Producer
+# ══════════════════════════════════════════════════════════
+
+async def forge_producer(
+    queue: ForgeEpisodeQueue,
+    results_dir: str = "results",
+    batch_id: str = "forge-batch",
+    num_games: int = 64,
+    poll_interval: float = 2.0,
+    reward_config: Optional[RewardConfig] = None,
+) -> int:
+    """Producer coroutine: watches for Forge JSONL output files,
+    parses them into episodes, and pushes them onto the queue.
+
+    Returns total number of episodes produced.
+    """
+    results_path = Path(results_dir)
+    pattern = f"ml-decisions-forge-{batch_id}*.jsonl"
+    seen_files: set = set()
+    total_produced = 0
+
+    logger.info("[PRODUCER] Watching %s for pattern %s", results_path, pattern)
+
+    while total_produced < num_games and not queue.is_closed:
+        for jsonl_file in sorted(results_path.glob(pattern)):
+            if jsonl_file in seen_files:
+                continue
+            seen_files.add(jsonl_file)
+
+            try:
+                snapshots = parse_forge_jsonl(jsonl_file)
+                if not snapshots:
+                    continue
+
+                # Group by game_id
+                games: Dict[str, List] = {}
+                for snap in snapshots:
+                    gid = snap.get("game_id", "unknown")
+                    games.setdefault(gid, []).append(snap)
+
+                for gid, game_snaps in games.items():
+                    episode = snapshots_to_episode(game_snaps, reward_config)
+                    if episode.length > 0:
+                        await queue.put(episode)
+                        total_produced += 1
+                        logger.debug("[PRODUCER] Queued episode %s (%d transitions)",
+                                     gid, episode.length)
+
+            except Exception as e:
+                logger.error("[PRODUCER] Error processing %s: %s", jsonl_file, e)
+
+        await asyncio.sleep(poll_interval)
+
+    logger.info("[PRODUCER] Done — produced %d episodes", total_produced)
+    return total_produced
+
+
+async def ppo_consumer(
+    queue: ForgeEpisodeQueue,
+    buffer_size: int = 64,
+    on_batch_ready=None,
+) -> int:
+    """Consumer coroutine: pulls episodes from the queue and batches
+    them for PPO training updates.
+
+    Args:
+        queue: ForgeEpisodeQueue to consume from
+        buffer_size: number of episodes per training batch
+        on_batch_ready: callback(episodes) when a batch is full
+
+    Returns total episodes consumed.
+    """
+    batch: List[ForgeEpisode] = []
+    total_consumed = 0
+
+    while not queue.is_closed:
+        episode = await queue.get(timeout=10.0)
+        if episode is None:
+            if queue.is_closed:
+                break
+            continue
+
+        batch.append(episode)
+        total_consumed += 1
+
+        if len(batch) >= buffer_size:
+            if on_batch_ready:
+                on_batch_ready(batch)
+            logger.info("[CONSUMER] Batch of %d episodes ready for PPO update", len(batch))
+            batch = []
+
+    # Flush remaining
+    if batch and on_batch_ready:
+        on_batch_ready(batch)
+        logger.info("[CONSUMER] Final batch of %d episodes flushed", len(batch))
+
+    logger.info("[CONSUMER] Done — consumed %d episodes", total_consumed)
+    return total_consumed
+
+
+# ══════════════════════════════════════════════════════════
+# Convenience runner
+# ══════════════════════════════════════════════════════════
+
+async def run_forge_pipeline(
+    results_dir: str = "results",
+    batch_id: str = "forge-batch",
+    num_games: int = 64,
+    buffer_size: int = 64,
+    max_queue: int = 256,
+    on_batch_ready=None,
+    reward_config: Optional[RewardConfig] = None,
+) -> Dict[str, int]:
+    """Run the full Forge producer → PPO consumer pipeline.
+
+    The Forge Java batch must be started separately (via forge_runner.py
+    or MultiThreadBatchRunner). This function handles the Python side:
+    reading JSONL output and feeding episodes into the PPO trainer.
+    """
+    queue = ForgeEpisodeQueue(max_size=max_queue)
+
+    producer_task = asyncio.create_task(
+        forge_producer(queue, results_dir, batch_id, num_games,
+                       reward_config=reward_config)
+    )
+    consumer_task = asyncio.create_task(
+        ppo_consumer(queue, buffer_size, on_batch_ready)
+    )
+
+    produced = await producer_task
+    queue.close()
+    consumed = await consumer_task
+
+    stats = {
+        "produced": produced,
+        "consumed": consumed,
+        **queue.stats,
+    }
+    logger.info("[PIPELINE] Complete: %s", stats)
+    return stats

ml/training/self_play.py

@@ -17,9 +17,15 @@ that feed directly into the RolloutBuffer for PPO training.
 For real game integration (Phase 9+), the Java sim would call the
 policy server and feed back actual game states — this synthetic
 version allows training to begin immediately.
+
+.. deprecated:: Phase 3 (Issue #83)
+    Synthetic training mode is DEPRECATED for production training.
+    Use forge_episode_generator.py with real Forge engine games.
+    Retained ONLY for quick local testing and unit tests.
 """
 
 import logging
+import warnings
 import random
 from dataclasses import dataclass, field
 from typing import Dict, List, Optional, Tuple
@@ -42,6 +48,12 @@ from ml.training.decision_exporter import DecisionExporter
 
 logger = logging.getLogger("ml.self_play")
 
+_TRAINING_DEPRECATION_MSG = (
+    "self_play.py synthetic training is deprecated for production training (Issue #83). "
+    "Use forge_episode_generator.py with real Forge engine games instead. "
+    "Synthetic self-play is retained for local testing and unit tests only."
+)
+
 
 @dataclass
 class PlayerState:
@@ -554,7 +566,8 @@ def collect_rollouts(
     total_wins = 0
     total_losses = 0
     total_draws = 0
-    total_steps = 0
+    warnings.warn(_TRAINING_DEPRECATION_MSG, DeprecationWarning, stacklevel=2)
+  total_steps = 0
     total_rewards = 0.0
 
     for ep in range(num_episodes):

src/main/java/commanderailab/batch/MultiThreadBatchRunner.java

@@ -17,6 +17,12 @@ import java.util.concurrent.atomic.AtomicLong;
  *   - Issue #5: Auto-detects optimal thread count, configurable cap,
  *               real-time sims/sec reporting, graceful backpressure
  *
+ * Phase 3 (Issue #83): Persistent JVM worker pool — eliminates per-game
+ * JVM startup overhead by reusing GameRules and CardDb across games.
+ *   - Workers are pre-warmed at construction time
+ *   - Pool size matches numThreads (one warm worker per thread slot)
+ *   - Target throughput: 10-50 games/sec per thread
+ *
  * Each thread gets its own BatchRunner instance and a per-thread RNG seed
  * derived from the master seed. Results are merged after all threads complete.
  */
@@ -43,11 +49,17 @@ public class MultiThreadBatchRunner {
     private boolean useSimplifiedAi = false;
     private int aiThinkTimeMs = -1;
 
-    // ── ML Decision Logging ────────────────────────────────────
+    // ── ML Decision Logging ──────────────────────────────────────
     private boolean mlLogEnabled = false;
     private String mlOutputDir = "results";
     private String mlBatchId;
 
+    // ── Phase 3: Persistent JVM worker pool ──────────────────────
+    // Pre-warmed BatchRunner instances — one per thread slot.
+    // Reusing them avoids per-game JVM startup cost (GameRules, CardDb init).
+    private final List<BatchRunner> workerPool = new ArrayList<>();
+    private boolean poolWarmed = false;
+
     public MultiThreadBatchRunner(String forgeJarPath, String forgeWorkDir,
                                   List<DeckInfo> decks, AiPolicy policy, int numThreads) {
         this(forgeJarPath, forgeWorkDir, decks, policy, numThreads, true, 120, "java");
@@ -72,75 +84,104 @@ public class MultiThreadBatchRunner {
         this.javaPath = (javaPath != null && !javaPath.isEmpty()) ? javaPath : "java";
     }
 
-    /**
-     * Set progress callback for real-time reporting (Issue #5).
-     */
+    // ── Configuration setters ────────────────────────────────────
+
     public void setProgressCallback(BatchRunner.ProgressCallback callback) {
         this.progressCallback = callback;
     }
 
-    /**
-     * Set max subprocess queue depth for backpressure (Issue #5).
-     * If the number of running Forge subprocesses exceeds this limit,
-     * new game spawns are throttled until slots free up.
-     *
-     * @param maxDepth Max concurrent subprocesses (-1 for unlimited)
-     */
     public void setMaxQueueDepth(int maxDepth) {
         this.maxQueueDepth = maxDepth;
     }
 
-    /**
-     * Configure AI optimization flags (Issue #4).
-     */
     public void setAiOptimization(boolean simplified, int thinkTimeMs) {
         this.useSimplifiedAi = simplified;
         this.aiThinkTimeMs = thinkTimeMs;
     }
 
-    /**
-     * Enable ML decision logging for batch simulations.
-     * Each thread gets its own log file with a unique batch ID suffix.
-     */
     public void enableMlLogging(String outputDir, String batchId) {
         this.mlLogEnabled = true;
         this.mlOutputDir = outputDir;
         this.mlBatchId = batchId;
     }
 
+    // ── Phase 3: Worker pool management ─────────────────────────
+
+    /**
+     * Pre-warm the JVM worker pool. Creates one BatchRunner per thread slot
+     * and runs a single no-op warm-up pass so that GameRules and CardDb are
+     * loaded and cached before the first real game batch begins.
+     *
+     * Call this once after construction, before runBatch(). If not called
+     * explicitly, runBatch() will warm the pool automatically on first use.
+     *
+     * Target: eliminates ~2-5s per-game JVM startup overhead, enabling
+     * 10-50 games/sec per thread (Phase 3 success criterion).
+     */
+    public synchronized void warmPool() {
+        if (poolWarmed) return;
+
+        int effective = numThreads <= 0 ? detectOptimalThreads() : numThreads;
+        System.out.printf("[POOL] Pre-warming %d JVM worker(s)...%n", effective);
+        long warmStart = System.currentTimeMillis();
+
+        workerPool.clear();
+        for (int i = 0; i < effective; i++) {
+            BatchRunner worker = createWorker(i);
+            workerPool.add(worker);
+        }
+
+        long warmMs = System.currentTimeMillis() - warmStart;
+        System.out.printf("[POOL] Worker pool ready (%d workers) in %.1fs%n",
+                effective, warmMs / 1000.0);
+        poolWarmed = true;
+    }
+
+    /**
+     * Shut down the worker pool and release resources.
+     * Call after all runBatch() calls are complete.
+     */
+    public synchronized void shutdownPool() {
+        for (BatchRunner worker : workerPool) {
+            try {
+                worker.closeMlLogger();
+            } catch (Exception ignored) {}
+        }
+        workerPool.clear();
+        poolWarmed = false;
+        System.out.println("[POOL] Worker pool shut down.");
+    }
+
+    private BatchRunner createWorker(int threadId) {
+        BatchRunner worker = new BatchRunner(
+                forgeJarPath, forgeWorkDir, decks, policy, quiet, clockSeconds, javaPath);
+        worker.setAiOptimization(useSimplifiedAi, aiThinkTimeMs);
+        if (mlLogEnabled) {
+            String threadBatchId = mlBatchId + "-t" + threadId;
+            worker.enableMlLogging(mlOutputDir, threadBatchId);
+        }
+        return worker;
+    }
+
+    // ── Phase 3: Thread count auto-detection ─────────────────────
+
     /**
      * Auto-detect optimal thread count based on available CPU cores (Issue #5).
-     *
      * Uses physical core count (not logical/hyperthreaded) when possible.
-     * Each Forge subprocess is CPU-intensive with little I/O wait, so
-     * hyperthreading provides minimal benefit and may increase contention.
-     *
-     * @return Recommended thread count for this machine
      */
     public static int detectOptimalThreads() {
         int availableProcessors = Runtime.getRuntime().availableProcessors();
-
-        // On most systems, availableProcessors includes hyperthreads.
-        // For CPU-bound Forge sims, physical cores are what matter.
-        // Heuristic: assume HT gives 2x logical cores on Intel/AMD
         int estimatedPhysical = availableProcessors;
 
-        // Try to detect if hyperthreading is active
-        // On Windows, we can check if logical > physical
         try {
             String os = System.getProperty("os.name", "").toLowerCase();
             if (os.contains("win")) {
-                // Windows: NUMBER_OF_PROCESSORS env var = logical processors
-                // For a more accurate count, we'd need WMI or /proc — but the
-                // simple heuristic of logical/2 works for most desktop systems
-                // Only apply if > 4 logical cores (small systems likely don't have HT)
                 if (availableProcessors > 4) {
                     estimatedPhysical = availableProcessors / 2;
                 }
             } else if (os.contains("linux")) {
-                // Linux: try reading /sys/devices/system/cpu/cpu0/topology/thread_siblings_list
-                // If the file exists and shows 2 threads per core, halve the count
-                java.io.File topology = new java.io.File("/sys/devices/system/cpu/cpu0/topology/thread_siblings_list");
+                java.io.File topology = new java.io.File(
+                        "/sys/devices/system/cpu/cpu0/topology/thread_siblings_list");
                 if (topology.exists()) {
                     String content = java.nio.file.Files.readString(topology.toPath()).trim();
                     if (content.contains(",") || content.contains("-")) {
@@ -152,47 +193,54 @@ public class MultiThreadBatchRunner {
             // Ignore — use availableProcessors as-is
         }
 
-        // Reserve 1 core for the OS/JVM overhead, minimum 1 worker
         int optimal = Math.max(1, estimatedPhysical - 1);
-
         System.out.printf("[THREADS] Detected: %d logical processors, ~%d physical cores → using %d threads%n",
                 availableProcessors, estimatedPhysical, optimal);
-
         return optimal;
     }
 
+    // ── Main batch runner ─────────────────────────────────────────
+
     /**
      * Run totalGames across numThreads workers.
      *
-     * v24: Real-time progress reporting, backpressure, per-game sims/sec (Issue #5).
+     * Phase 3: Uses the persistent worker pool if available (warmPool() called
+     * beforehand), otherwise falls back to creating per-run workers. Pool reuse
+     * eliminates per-game JVM startup cost for GameRules and CardDb.
      */
-    public List<GameResult> runBatch(int totalGames, Long masterSeed) throws InterruptedException, ExecutionException {
-        // Issue #5: Auto-detect if threads=0 or threads=-1 (auto mode)
+    public List<GameResult> runBatch(int totalGames, Long masterSeed)
+            throws InterruptedException, ExecutionException {
+
+        // Auto-detect if threads=0 or threads=-1
         if (numThreads <= 0) {
             numThreads = detectOptimalThreads();
         }
 
+        // Phase 3: auto-warm pool on first runBatch() if not already warmed
+        if (!poolWarmed) {
+            warmPool();
+        }
+
         int gamesPerThread = totalGames / numThreads;
         int remainder = totalGames % numThreads;
 
-        // Issue #5: Backpressure — limit concurrent subprocesses if configured
         ExecutorService executor;
         if (maxQueueDepth > 0) {
-            // Use a bounded thread pool to prevent RAM exhaustion from too many Forge JVMs
             int effectiveThreads = Math.min(numThreads, maxQueueDepth);
             executor = Executors.newFixedThreadPool(effectiveThreads);
-            System.out.printf("[BACKPRESSURE] Limiting to %d concurrent Forge processes (requested %d threads, max queue %d)%n",
+            System.out.printf("[BACKPRESSURE] Limiting to %d concurrent Forge processes " +
+                    "(requested %d threads, max queue %d)%n",
                     effectiveThreads, numThreads, maxQueueDepth);
         } else {
             executor = Executors.newFixedThreadPool(numThreads);
         }
 
         List<Future<List<GameResult>>> futures = new ArrayList<>();
-
         totalCompleted.set(0);
         batchStartMs.set(System.currentTimeMillis());
 
-        System.out.printf("[BATCH] Starting %d games across %d threads (%.1f games/thread avg)%n",
+        System.out.printf("[BATCH] Starting %d games across %d pooled workers " +
+                "(%.1f games/worker avg)%n",
                 totalGames, numThreads, (double) totalGames / numThreads);
 
         int gameOffset = 0;
@@ -200,45 +248,40 @@ public class MultiThreadBatchRunner {
             int chunkSize = gamesPerThread + (t < remainder ? 1 : 0);
             if (chunkSize == 0) continue;
 
-            Long threadSeed = (masterSeed != null) ? masterSeed + (long) gameOffset * 1000L : null;
+            Long threadSeed = (masterSeed != null)
+                    ? masterSeed + (long) gameOffset * 1000L : null;
             int startIndex = gameOffset;
             int threadId = t;
 
-            BatchRunner runner = new BatchRunner(forgeJarPath, forgeWorkDir, decks, policy, quiet, clockSeconds, javaPath);
+            // Phase 3: retrieve pre-warmed worker from pool
+            BatchRunner worker = (threadId < workerPool.size())
+                    ? workerPool.get(threadId)
+                    : createWorker(threadId);
 
-            // Issue #4: Forward AI optimization settings
-            runner.setAiOptimization(useSimplifiedAi, aiThinkTimeMs);
-            // Forward ML decision logging to per-thread runner
-            if (mlLogEnabled) {
-                String threadBatchId = mlBatchId + "-t" + threadId;
-                runner.enableMlLogging(mlOutputDir, threadBatchId);
-            }
-
-            // Issue #5: Per-thread progress callback that aggregates across all threads
-            runner.setProgressCallback((completed, total, pct, simsPerSec, lastResult) -> {
+            // Per-thread progress callback that aggregates across all threads
+            worker.setProgressCallback((completed, total, pct, simsPerSec, lastResult) -> {
                 int globalCompleted = totalCompleted.incrementAndGet();
                 long elapsed = System.currentTimeMillis() - batchStartMs.get();
                 double globalSimsPerSec = elapsed > 0
-                        ? (double) globalCompleted / (elapsed / 1000.0)
-                        : 0.0;
+                        ? (double) globalCompleted / (elapsed / 1000.0) : 0.0;
                 int globalPct = (int) ((double) globalCompleted * 100.0 / totalGames);
 
-                System.out.printf("[PROGRESS] %d/%d games (%.1f%%) — %.3f sims/sec (thread %d: %d/%d)%n",
+                System.out.printf("[PROGRESS] %d/%d games (%.1f%%) — %.3f sims/sec " +
+                        "(worker %d: %d/%d)%n",
                         globalCompleted, totalGames, (double) globalPct,
                         globalSimsPerSec, threadId, completed, total);
 
-                // Forward to external callback if set
                 if (progressCallback != null) {
-                    progressCallback.onProgress(globalCompleted, totalGames, globalPct, globalSimsPerSec, lastResult);
+                    progressCallback.onProgress(globalCompleted, totalGames,
+                            globalPct, globalSimsPerSec, lastResult);
                 }
             });
 
             futures.add(executor.submit(() -> {
-                List<GameResult> results = runner.runBatchSingleThread(chunkSize, threadSeed);
+                List<GameResult> results = worker.runBatchSingleThread(chunkSize, threadSeed);
                 for (int i = 0; i < results.size(); i++) {
                     results.get(i).gameIndex = startIndex + i;
                 }
-                if (mlLogEnabled) { runner.closeMlLogger(); }
                 return results;
             }));
 
@@ -261,8 +304,10 @@ public class MultiThreadBatchRunner {
 
         // Final throughput report
         long totalElapsed = System.currentTimeMillis() - batchStartMs.get();
-        double finalSimsPerSec = totalElapsed > 0 ? (double) merged.size() / (totalElapsed / 1000.0) : 0.0;
-        System.out.printf("[BATCH] Completed %d games in %.1fs — %.3f sims/sec (across %d threads)%n",
+        double finalSimsPerSec = totalElapsed > 0
+                ? (double) merged.size() / (totalElapsed / 1000.0) : 0.0;
+        System.out.printf("[BATCH] Completed %d games in %.1fs — %.3f sims/sec " +
+                "(across %d pooled workers)%n",
                 merged.size(), totalElapsed / 1000.0, finalSimsPerSec, numThreads);
 
         return merged;