[VERSION] - This is TokenGate beta, partial is now used instead of lambda for executions, concurrency boosted, types hinted.

Author: TavariAgent

DOCS/SETUP.md

@@ -24,7 +24,7 @@ class OperationsCoordinator: # Under this class are where you find configuration
             self, 
             workers_per_core: int = 4, # Controls number of workers and may notgo under 2 workers.
             enable_convergence: bool = True, # Controls dynamic worker counts.
-            convergence_verbose: bool = False,
+            convergence_verbose: bool = False, # DO NOT ENABLE INSIDE A REPL!!!
             base_memory_budget_mb: int = 45, # This is a control for operation memory (adjustable).
             num_executors: int = 8, # This controls number of parallel executors.
             auto_block_dangerous: bool = False, # Blocks explosive tasks with possibly dangerous inputs.

allocation_optimizer.py

@@ -407,108 +407,4 @@ def print_portfolio(self):
                     print(f"  Success Rate: {alloc.get_success_rate():.1f}%")
 
             print()
-            print("=" * 80)
-
-
-# ============================================================================
-# TESTING
-# ============================================================================
-
-if __name__ == '__main__':
-    from .code_inspector import CodeInspector
-
-    print("=" * 70)
-    print("ALLOCATION OPTIMIZER TEST")
-    print("=" * 70)
-    print()
-
-    # Create optimizer
-    optimizer = AllocationOptimizer(base_budget_mb=50)
-
-
-    # Test functions with different complexity
-    def trivial_task(x):
-        return x * 2
-
-
-    def moderate_task(data):
-        result = []
-        for item in data:
-            if item > 0:
-                result.append(item * 2)
-        return result
-
-
-    def extreme_task(size):
-        matrix = [[i * j for j in range(size)] for i in range(size)]
-        return sum(sum(row) for row in matrix)
-
-
-    # Analyze and initialize operations
-    test_ops = [
-        ('trivial_op', trivial_task),
-        ('moderate_op', moderate_task),
-        ('extreme_op', extreme_task)
-    ]
-
-    for op_name, func in test_ops:
-        print("-" * 70)
-        print(f"Initializing: {op_name}")
-        print("-" * 70)
-
-        metrics = CodeInspector.analyze(func)
-        optimizer.initialize_operation(op_name, metrics)
-        print()
-
-    # Simulate optimization cycles
-    print("=" * 70)
-    print("SIMULATING OPTIMIZATION CYCLES")
-    print("=" * 70)
-    print()
-
-    # Trivial operation - should optimize aggressively
-    print("-" * 70)
-    print("TRIVIAL OPERATION - Aggressive Optimization (5% rate)")
-    print("-" * 70)
-
-    for cycle in range(5):
-        print(f"\nCycle {cycle + 1}:")
-
-        # Start at baseline and increment
-        confidence = 85.0 + (cycle * 1.5)
-
-        new_alloc = optimizer.attempt_optimization('trivial_op', confidence)
-
-        if new_alloc:
-            # Simulate successful execution with confidence gain
-            success_confidence = confidence + 1.5
-            optimizer.record_execution_result('trivial_op', True, success_confidence)
-        else:
-            print("  Optimization blocked")
-
-    # Extreme operation - should optimize conservatively
-    print()
-    print("-" * 70)
-    print("EXTREME OPERATION - Conservative Optimization (2% rate)")
-    print("-" * 70)
-
-    for cycle in range(5):
-        print(f"\nCycle {cycle + 1}:")
-
-        # Start at baseline and increment
-        confidence = 83.7 + (cycle * 1.0)
-
-        new_alloc = optimizer.attempt_optimization('extreme_op', confidence)
-
-        if new_alloc:
-            # Simulate with smaller confidence gains
-            success_confidence = confidence + 1.0
-            optimizer.record_execution_result('extreme_op', True, success_confidence)
-        else:
-            print("  Optimization blocked")
-
-    # Show final portfolio
-    optimizer.print_portfolio()
-
-    print()
-    print("Test complete!")
+            print("=" * 80)

code_inspector.py

@@ -359,105 +359,4 @@ def predict_initial_allocation(metrics: CodeMetrics, base_budget_mb: int = 50) -
             'confidence': metrics.confidence,
             'reasoning': reasoning,
             'complexity_score': metrics.complexity_score
-        }
-
-
-# ============================================================================
-# TESTING & EXAMPLES
-# ============================================================================
-
-if __name__ == '__main__':
-
-    print("=" * 70)
-    print("CODE INSPECTOR TEST")
-    print("=" * 70)
-    print()
-
-
-    # Test functions with varying complexity
-
-    def trivial_function(x):
-        """Trivial: just return input."""
-        return x
-
-
-    def simple_function(a, b):
-        """Simple: basic arithmetic."""
-        return a + b * 2
-
-
-    def moderate_function(data):
-        """Moderate: some loops and logic."""
-        result = []
-        for item in data:
-            if item > 0:
-                result.append(item * 2)
-        return result
-
-
-    def complex_function(size):
-        """Complex: nested loops."""
-        matrix = [[0 for _ in range(size)] for _ in range(size)]
-        for i in range(size):
-            for j in range(size):
-                matrix[i][j] = i * j
-        return matrix
-
-
-    def extreme_function(n):
-        """Extreme: recursive + heavy computation."""
-        if n <= 1:
-            return 1
-        result = extreme_function(n - 1) + extreme_function(n - 2)
-        data = [i ** 2 for i in range(100)]
-        return result + sum(data)
-
-
-    # Analyze each function
-    test_functions = [
-        trivial_function,
-        simple_function,
-        moderate_function,
-        complex_function,
-        extreme_function
-    ]
-
-    for func in test_functions:
-        print("-" * 70)
-        print(f"Analyzing: {func.__name__}")
-        print("-" * 70)
-
-        metrics = CodeInspector.analyze(func)
-
-        print(f"  Bytecode length:  {metrics.bytecode_length}")
-        print(f"  Stack size:       {metrics.stack_size}")
-        print(f"  Local vars:       {metrics.local_vars}")
-        print(f"  Constants:        {metrics.const_count} ({metrics.const_footprint} bytes)")
-        print(f"  External calls:   {len(metrics.external_calls)}")
-        if metrics.external_calls:
-            print(f"    Calls: {', '.join(metrics.external_calls[:5])}")
-        print(f"  Loops:            {metrics.loop_count}")
-        print(f"  Branches:         {metrics.branch_count}")
-        print()
-        print(f"  Complexity Score: {metrics.complexity_score:.1f}")
-        print(f"  Complexity Level: {metrics.complexity_level.name}")
-        print(f"  Confidence:       {metrics.confidence}%")
-        print()
-
-        # Get allocation prediction
-        allocation = CodeInspector.predict_initial_allocation(metrics)
-        print(f"  Recommended Allocation:")
-        print(f"    Memory:   {allocation['memory_mb']} MB")
-        print(f"    Timeout:  {allocation['timeout_seconds']:.0f} seconds")
-        print(f"    Reasoning: {allocation['reasoning']}")
-        print()
-
-    print("=" * 70)
-    print("INSPECTOR TEST COMPLETE")
-    print("=" * 70)
-    print()
-    print("Next Steps:")
-    print("  1. Integrate with overflow_guard.py")
-    print("  2. Add token-level tracking")
-    print("  3. Implement retry logic with allocation bumping")
-    print("  4. Build confidence-weighted optimization (2% drops)")
+        }

core_affinity_queue.py

@@ -18,8 +18,6 @@
 from queue import Queue
 from typing import List
 
-from .token_system import TaskToken
-
 
 class TaskWeight(Enum):
     """Routing weight classes used by the affinity policy."""
@@ -190,33 +188,6 @@ def __init__(self, topology, workers_per_core: int = 4):
         self.routing_failures = 0
         self._routing_lock = threading.Lock()
 
-    def classify_token_weight(self, token: TaskToken) -> TaskWeight:
-        """Infer routing weight from token tags or operation type.
-
-        Tag-based weight takes precedence. If no explicit weight tag is present,
-        the operation type is inspected for heavy/light hints. Medium is the
-        fallback class.
-        """
-        # Check tags first
-        if 'weight' in token.metadata.tags:
-            weight_str = token.metadata.tags['weight'].lower()
-            if weight_str == 'heavy':
-                return TaskWeight.HEAVY
-            elif weight_str == 'light':
-                return TaskWeight.LIGHT
-            else:
-                return TaskWeight.MEDIUM
-
-        # Check operation_type suffix
-        op_type = token.metadata.operation_type.lower()
-        if op_type.endswith('_heavy') or 'heavy' in op_type:
-            return TaskWeight.HEAVY
-        elif op_type.endswith('_light') or 'light' in op_type:
-            return TaskWeight.LIGHT
-
-        # Default to medium
-        return TaskWeight.MEDIUM
-
     def get_valid_cores_for_weight(self, weight: TaskWeight) -> List[int]:
         """Return the allowed core chain for the given weight."""
         return self.policy.get_preference_chain(weight)

core_pinned_staggered_queue.py

@@ -15,21 +15,11 @@
 """
 
 import asyncio
-import os
-import sys
 import time
+from functools import partial
 from typing import Dict, List, Tuple, Any
 
 from .threading_metrics import get_metrics
-
-# Add project paths for imports
-current_dir = os.path.dirname(os.path.abspath(__file__))
-if current_dir not in sys.path:
-    sys.path.insert(0, current_dir)
-parent_dir = os.path.dirname(current_dir)
-if parent_dir not in sys.path:
-    sys.path.insert(0, parent_dir)
-
 from .token_system import TaskToken, TokenState
 from .admission_gate import WorkerTaskQueue
 from .core_affinity_queue import TaskWeight
@@ -58,6 +48,7 @@ def __init__(self, num_cores: int, workers_per_core: int, coordinator: Any):
                 retry, overflow, and Guard House callbacks.
         """
         super().__init__()
+        self.result_verbose = False
         self.coordinator = coordinator
         self.num_cores = num_cores
         self.workers_per_core = workers_per_core
@@ -79,7 +70,7 @@ def __init__(self, num_cores: int, workers_per_core: int, coordinator: Any):
         self.core_busy: Dict[int, int] = {c: 0 for c in range(1, self.num_cores + 1)}
 
         # Capped mailbox length to prevent runaway memory (DOS safety)
-        self.MAILBOX_MAX = 100 # Max tokens per worker mailbox
+        self.MAILBOX_MAX = 75 # Max tokens per worker mailbox
 
         self.core_patterns: Dict[int, int] = {}
         for core_id in range(1, num_cores + 1):
@@ -176,12 +167,9 @@ async def _execute_token(self, token: TaskToken, worker_id: str, core_id: int):
         success = False
 
         try:
-            # Execute function
-            result = await asyncio.get_event_loop().run_in_executor(
-                None,
-                lambda: token.func(*token.args, **token.kwargs))
-
-            # Success!
+            loop = asyncio.get_running_loop()
+            bound_func = partial(token.func, *token.args, **token.kwargs)
+            result = await loop.run_in_executor(None, bound_func)
             token.set_result(result)
             self.total_executed += 1
             success = True
@@ -192,8 +180,9 @@ async def _execute_token(self, token: TaskToken, worker_id: str, core_id: int):
             # Failed!
             token.set_error(e)
             self.total_failed += 1
+            if self.result_verbose:
+                print(f"[{worker_id.upper()}] ✗ Failed {token.token_id}: {e}")
 
-            print(f"[{worker_id.upper()}] ✗ Failed {token.token_id}: {e}")
         finally:
             execution_duration = time.time() - start_time
 
@@ -276,13 +265,6 @@ async def _execute_token_with_metrics(self, token: "TaskToken", worker_id: str,
         """Execute one token while updating worker-state and outcome metrics."""
         op_type = token.metadata.tags.get("operation_type", "unknown")
 
-        # Busy/idle update (pattern aware)
-        active_workers = self.core_patterns.get(core_id, self.workers_per_core)
-        self.core_busy[core_id] = min(active_workers, self.core_busy.get(core_id, 0) + 1)
-        busy = self.core_busy[core_id]
-        idle = max(0, active_workers - busy)
-        self.metrics.update_worker_state(core_id, busy, idle)
-
         t0 = time.perf_counter()
         try:
             await self._execute_token(token, worker_id, core_id)
@@ -291,12 +273,6 @@ async def _execute_token_with_metrics(self, token: "TaskToken", worker_id: str,
         except Exception:
             self.metrics.record_task_failure(op_type, core_id)
             raise
-        finally:
-            # Decrement busy and update again
-            self.core_busy[core_id] = max(0, self.core_busy.get(core_id, 0) - 1)
-            busy = self.core_busy[core_id]
-            idle = max(0, active_workers - busy)
-            self.metrics.update_worker_state(core_id, busy, idle)
 
     def get_core_for_weight(self, weight: TaskWeight) -> List[int]:
         """Return the eligible core range for a routing weight.
@@ -327,7 +303,8 @@ def get_core_for_weight(self, weight: TaskWeight) -> List[int]:
                 # Fallback for single-core
                 return [1]
 
-    def classify_token_weight(self, token: TaskToken) -> TaskWeight:
+    @staticmethod
+    def classify_token_weight(token: TaskToken) -> TaskWeight:
         """Infer routing weight from token tags or operation-type naming.
 
         Explicit weight tags take precedence over operation-type heuristics.
@@ -382,7 +359,7 @@ def assign_worker_positions(self, worker_id: str, worker_index: int, core_id: in
 
         print(f"[CORE_PINNED] Worker {worker_id} (Core {core_id}): {positions[:5]}... (every {self.total_workers})")
 
-    def assign_position_for_token(self, token: TaskToken, verbose = True) -> int:
+    def assign_position_for_token(self, token: TaskToken) -> int:
         """Assign a staggered global route position for a token.
 
         The assigned position respects token weight, valid-core range, current
@@ -412,7 +389,7 @@ def assign_position_for_token(self, token: TaskToken, verbose = True) -> int:
         # Increment counter
         self.core_position_counters[chosen_core] += 1
 
-        if verbose:
+        if self.result_verbose:
             print(f"[ROUTING] Token {token.token_id} ({weight.value}) → Pos {position} (Core {chosen_core}, Pattern {active_workers})")
         return position