feat: Introduce HiveMind Ensemble Subsystem for Swarm/Fusion Orchestration

docs/HiveMind Task.md

@@ -0,0 +1,93 @@
+# HiveMind (Swarm/Fusion) Implementation
+
+## Phase 1: Core Infrastructure
+- [/] Design and Plan
+    - [x] Explore codebase
+    - [x] Create comprehensive implementation plan
+- [ ] Create `src/rotator_library/ensemble_manager.py`
+    - [ ] Define `EnsembleManager` class skeleton
+    - [ ] Implement config loading and validation
+    - [ ] Implement `is_ensemble()` detection
+    - [ ] Implement conflict resolution for naming
+- [ ] Modify `src/rotator_library/client.py`
+    - [ ] Initialize `EnsembleManager` in `__init__`
+    - [ ] Integrate into `acompletion()` dispatcher
+    - [ ] Add logging for HiveMind operations
+- [ ] Create `ensemble_config.json`
+    - [ ] Define schema for Fusions
+    - [ ] Define schema for Swarm defaults
+    - [ ] Define arbitration strategies
+
+## Phase 2: Basic Swarm Mode
+- [ ] Implement Swarm Features
+    - [ ] `_prepare_drones()` - basic cloning
+    - [ ] `_execute_parallel()` - asyncio.gather
+    - [ ] `_format_for_arbiter()` - response aggregation
+    - [ ] `_build_arbiter_prompt()` - synthesis strategy
+    - [ ] `_call_arbiter()` - judge execution
+- [ ] Testing
+    - [ ] Test basic 3-drone swarm
+    - [ ] Test arbiter synthesis
+    - [ ] Test partial failures
+
+## Phase 3: Advanced Swarm Features
+- [ ] Temperature Jitter
+    - [ ] Implement jitter logic
+    - [ ] Test randomness and clamping
+- [ ] Adversarial Mode
+    - [ ] Implement adversarial prompt injection
+    - [ ] Test with configurable count
+- [ ] Blind Switch
+    - [ ] Implement response anonymization
+    - [ ] Test with blind=true/false
+- [ ] Confidence Scoring
+    - [ ] Implement score extraction
+    - [ ] Add logging for scores
+
+## Phase 4: Fusion Mode
+- [ ] Implement Fusion Features
+    - [ ] `_prepare_models()` - multi-model setup
+    - [ ] Role assignment and prompts
+    - [ ] Role context for Arbiter
+    - [ ] Weight system (future)
+- [ ] Testing
+    - [ ] Test 2-model fusion
+    - [ ] Test role context injection
+    - [ ] Test specialist descriptions
+
+## Phase 5: Recursive/Reflective Mode
+- [ ] Implement Recursion
+    - [ ] Consensus check logic
+    - [ ] Conflict extraction
+    - [ ] `_trigger_round_2()` implementation
+    - [ ] Max rounds enforcement
+- [ ] Testing
+    - [ ] Test low-confidence trigger
+    - [ ] Test Round 2 critique
+    - [ ] Test final re-synthesis
+
+## Phase 6: Polish & Edge Cases
+- [ ] Error Handling
+    - [ ] Partial failure handling
+    - [ ] Arbiter failure fallback
+    - [ ] Infinite recursion prevention
+- [ ] Performance
+    - [ ] Latency logging
+    - [ ] Token usage tracking
+    - [ ] Rate limit mitigation
+- [ ] Documentation
+    - [ ] User guide
+    - [ ] Example configs
+    - [ ] API reference
+
+## Verification
+- [ ] Automated Tests
+    - [ ] test_ensemble_manager.py (all 8 test cases)
+    - [ ] test_swarm_logic.py
+    - [ ] test_fusion_logic.py
+    - [ ] test_recursion.py
+- [ ] Manual Tests
+    - [ ] Scenario 1: Simple Swarm
+    - [ ] Scenario 2: Adversarial Swarm
+    - [ ] Scenario 3: Fusion with Roles
+    - [ ] Scenario 4: Recursive Refinement

src/rotator_library/client.py

@@ -33,6 +33,7 @@
 from .credential_manager import CredentialManager
 from .background_refresher import BackgroundRefresher
 from .model_definitions import ModelDefinitions
+from .ensemble import EnsembleManager
 
 
 class StreamedAPIError(Exception):
@@ -128,6 +129,10 @@ def __init__(
             if max_val < 1:
                 lib_logger.warning(f"Invalid max_concurrent for '{provider}': {max_val}. Setting to 1.")
                 self.max_concurrent_requests_per_key[provider] = 1
+
+        # Initialize HiveMind ensemble manager
+        self.ensemble_manager = EnsembleManager(rotating_client=self)
+        lib_logger.info("HiveMind ensemble manager initialized")
 
     def _is_model_ignored(self, provider: str, model_id: str) -> bool:
         """
@@ -1606,8 +1611,15 @@ def acompletion(
         Returns:
             The completion response object, or an async generator for streaming responses, or None if all retries fail.
         """
-        # Handle iflow provider: remove stream_options to avoid HTTP 406
         model = kwargs.get("model", "")
+
+        # Check if this is an ensemble request (HiveMind)
+        if model and self.ensemble_manager.is_ensemble(model):
+            lib_logger.debug(f"[HiveMind] Detected ensemble request: {model}")
+            # Delegate to ensemble manager
+            return self.ensemble_manager.handle_request(request=request, **kwargs)
+
+        # Handle iflow provider: remove stream_options to avoid HTTP 406
         provider = model.split("/")[0] if "/" in model else ""
 
         if provider == "iflow" and "stream_options" in kwargs:

src/rotator_library/ensemble/__init__.py

@@ -0,0 +1,9 @@
+"""
+HiveMind Ensemble Module
+
+This module provides parallel model execution (Swarm/Fusion) with intelligent arbitration.
+"""
+
+from .manager import EnsembleManager
+
+__all__ = ['EnsembleManager']

src/rotator_library/ensemble/config_loader.py

@@ -0,0 +1,209 @@
+"""
+Configuration loader for HiveMind ensemble configs.
+
+Loads and validates configurations from the ensemble_configs directory structure.
+"""
+
+import os
+import json
+import logging
+from pathlib import Path
+from typing import Dict, List, Any, Optional
+
+lib_logger = logging.getLogger("rotator_library.ensemble")
+
+
+class ConfigLoader:
+    """Loads and manages ensemble configurations from folder structure."""
+
+    def __init__(self, config_dir: str):
+        """
+        Initialize the config loader.
+
+        Args:
+            config_dir: Path to ensemble_configs directory (relative to rotator_library)
+        """
+        self.config_dir = Path(config_dir)
+        self.swarms_dir = self.config_dir / "swarms"
+        self.fusions_dir = self.config_dir / "fusions"
+        self.strategies_dir = self.config_dir / "strategies"
+
+        # Loaded configurations
+        self.swarm_default: Optional[Dict[str, Any]] = None
+        self.swarm_configs: Dict[str, Dict[str, Any]] = {}
+        self.fusion_configs: Dict[str, Dict[str, Any]] = {}
+        self.strategies: Dict[str, str] = {}
+
+    def load_all(self) -> None:
+        """Load all configurations from the directory structure."""
+        lib_logger.info("[HiveMind] Loading ensemble configurations...")
+
+        # Create directories if they don't exist
+        self._ensure_directories()
+
+        # Load swarm configurations
+        self._load_swarm_configs()
+
+        # Load fusion configurations
+        self._load_fusion_configs()
+
+        # Load strategy templates
+        self._load_strategies()
+
+        lib_logger.info(
+            f"[HiveMind] Loaded {len(self.swarm_configs)} swarm configs, "
+            f"{len(self.fusion_configs)} fusion configs, "
+            f"{len(self.strategies)} strategies"
+        )
+
+    def _ensure_directories(self) -> None:
+        """Create config directories if they don't exist."""
+        for directory in [self.swarms_dir, self.fusions_dir, self.strategies_dir]:
+            directory.mkdir(parents=True, exist_ok=True)
+
+    def _load_swarm_configs(self) -> None:
+        """Load swarm configurations from swarms/ directory."""
+        if not self.swarms_dir.exists():
+            lib_logger.warning(f"[HiveMind] Swarms directory not found: {self.swarms_dir}")
+            return
+
+        # Load default.json first
+        default_path = self.swarms_dir / "default.json"
+        if default_path.exists():
+            try:
+                with open(default_path, 'r', encoding='utf-8') as f:
+                    self.swarm_default = json.load(f)
+                lib_logger.debug("[HiveMind] Loaded default swarm config")
+            except Exception as e:
+                lib_logger.error(f"[HiveMind] Failed to load default swarm config: {e}")
+        else:
+            lib_logger.warning("[HiveMind] No default swarm config found")
+
+        # Load model-specific configs
+        for config_file in self.swarms_dir.glob("*.json"):
+            if config_file.name == "default.json":
+                continue
+
+            try:
+                with open(config_file, 'r', encoding='utf-8') as f:
+                    config = json.load(f)
+
+                # Extract model name from config
+                model_name = config.get("model")
+                if model_name:
+                    self.swarm_configs[model_name] = config
+                    lib_logger.debug(f"[HiveMind] Loaded swarm config for '{model_name}'")
+                else:
+                    lib_logger.warning(
+                        f"[HiveMind] Swarm config '{config_file.name}' missing 'model' field"
+                    )
+            except Exception as e:
+                lib_logger.error(f"[HiveMind] Failed to load swarm config '{config_file.name}': {e}")
+
+    def _load_fusion_configs(self) -> None:
+        """Load fusion configurations from fusions/ directory."""
+        if not self.fusions_dir.exists():
+            lib_logger.warning(f"[HiveMind] Fusions directory not found: {self.fusions_dir}")
+            return
+
+        for config_file in self.fusions_dir.glob("*.json"):
+            try:
+                with open(config_file, 'r', encoding='utf-8') as f:
+                    config = json.load(f)
+
+                fusion_id = config.get("id")
+                if not fusion_id:
+                    lib_logger.warning(
+                        f"[HiveMind] Fusion config '{config_file.name}' missing 'id' field"
+                    )
+                    continue
+
+                # Check for duplicate IDs
+                if fusion_id in self.fusion_configs:
+                    lib_logger.warning(
+                        f"[HiveMind] Duplicate fusion ID '{fusion_id}'. "
+                        f"Config from '{config_file.name}' will override previous."
+                    )
+
+                self.fusion_configs[fusion_id] = config
+                lib_logger.debug(f"[HiveMind] Loaded fusion config '{fusion_id}'")
+
+            except Exception as e:
+                lib_logger.error(f"[HiveMind] Failed to load fusion config '{config_file.name}': {e}")
+
+    def _load_strategies(self) -> None:
+        """Load strategy templates from strategies/ directory."""
+        if not self.strategies_dir.exists():
+            lib_logger.warning(f"[HiveMind] Strategies directory not found: {self.strategies_dir}")
+            return
+
+        for strategy_file in self.strategies_dir.glob("*.txt"):
+            try:
+                with open(strategy_file, 'r', encoding='utf-8') as f:
+                    content = f.read()
+
+                strategy_name = strategy_file.stem
+                self.strategies[strategy_name] = content
+                lib_logger.debug(f"[HiveMind] Loaded strategy '{strategy_name}'")
+
+            except Exception as e:
+                lib_logger.error(
+                    f"[HiveMind] Failed to load strategy '{strategy_file.name}': {e}"
+                )
+
+    def get_swarm_config(self, model: str) -> Dict[str, Any]:
+        """
+        Get swarm configuration for a specific model.
+
+        Merges default config with model-specific overrides.
+
+        Args:
+            model: Base model name (without [swarm] suffix)
+
+        Returns:
+            Merged configuration dictionary
+        """
+        # Start with default
+        config = self.swarm_default.copy() if self.swarm_default else {}
+
+        # Apply model-specific overrides
+        if model in self.swarm_configs:
+            model_config = self.swarm_configs[model]
+            # Deep merge
+            for key, value in model_config.items():
+                if key == "model":
+                    continue  # Don't copy the model name
+                if isinstance(value, dict) and key in config:
+                    config[key] = {**config[key], **value}
+                else:
+                    config[key] = value
+
+        return config
+
+    def get_fusion_config(self, fusion_id: str) -> Optional[Dict[str, Any]]:
+        """
+        Get fusion configuration by ID.
+
+        Args:
+            fusion_id: Fusion identifier
+
+        Returns:
+            Fusion configuration or None if not found
+        """
+        return self.fusion_configs.get(fusion_id)
+
+    def get_strategy(self, strategy_name: str) -> Optional[str]:
+        """
+        Get strategy template by name.
+
+        Args:
+            strategy_name: Strategy identifier
+
+        Returns:
+            Strategy template string or None if not found
+        """
+        return self.strategies.get(strategy_name)
+
+    def get_all_fusion_ids(self) -> List[str]:
+        """Get list of all fusion IDs."""
+        return list(self.fusion_configs.keys())