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:
         """
@@ -636,6 +641,15 @@ async def _execute_with_retry(
         kwargs = self._convert_model_params(**kwargs)
 
         # The main rotation loop. It continues as long as there are untried credentials and the global deadline has not been exceeded.
+
+        # Resolve model ID early, before any credential operations
+        # This ensures consistent model ID usage for acquisition, release, and tracking
+        resolved_model = self._resolve_model_id(model, provider)
+        if resolved_model != model:
+            lib_logger.info(f"Resolved model '{model}' to '{resolved_model}'")
+            model = resolved_model
+            kwargs["model"] = model  # Ensure kwargs has the resolved model for litellm
+
         while (
             len(tried_creds) < len(credentials_for_provider) and time.time() < deadline
         ):
@@ -689,13 +703,8 @@ async def _execute_with_retry(
 
                 provider_plugin = self._get_provider_instance(provider)
 
-                # Convert model name to ID if custom mapping exists
-                resolved_model = self._resolve_model_id(model, provider)
-                if resolved_model != model:
-                    lib_logger.info(f"Resolved model '{model}' to '{resolved_model}'")
-                    litellm_kwargs["model"] = resolved_model
-                    # Update the model variable for subsequent logging
-                    model = resolved_model
+                # Model ID is already resolved before the loop, and kwargs['model'] is updated.
+                # No further resolution needed here.
 
                 # Apply model-specific options for custom providers
                 if provider_plugin and hasattr(provider_plugin, "get_model_options"):
@@ -996,6 +1005,14 @@ async def _streaming_acompletion_with_retry(
 
         consecutive_quota_failures = 0
 
+        # Resolve model ID early, before any credential operations
+        # This ensures consistent model ID usage for acquisition, release, and tracking
+        resolved_model = self._resolve_model_id(model, provider)
+        if resolved_model != model:
+            lib_logger.info(f"Resolved model '{model}' to '{resolved_model}'")
+            model = resolved_model
+            kwargs["model"] = model  # Ensure kwargs has the resolved model for litellm
+
         try:
             while (
                 len(tried_creds) < len(credentials_for_provider)
@@ -1071,13 +1088,8 @@ async def _streaming_acompletion_with_retry(
 
                     provider_plugin = self._get_provider_instance(provider)
 
-                    # Convert model name to ID if custom mapping exists
-                    resolved_model = self._resolve_model_id(model, provider)
-                    if resolved_model != model:
-                        lib_logger.info(f"Resolved model '{model}' to '{resolved_model}'")
-                        litellm_kwargs["model"] = resolved_model
-                        # Update the model variable for subsequent logging
-                        model = resolved_model
+                    # Model ID is already resolved before the loop, and kwargs['model'] is updated.
+                    # No further resolution needed here.
 
                     # Apply model-specific options for custom providers
                     if provider_plugin and hasattr(
@@ -1606,8 +1618,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']