feat(plugins): introduce middleware token proxy plugin suite (reorder, dedup, lookup, skill index)

.env.example

@@ -0,0 +1,4 @@
+OPENAI_API_KEY=your_openai_api_key_here
+DEEPSEEK_API_KEY=your_deepseek_api_key_here
+# Optional overrides
+# BASE_URL=https://api.openai.com/v1

.github/workflows/test.yml

@@ -26,7 +26,7 @@ jobs:
         run: |
           # Skip GPU tests as GitHub Actions runners don't have CUDA
           # To run GPU tests locally: pytest tests/ -v -m "gpu"
-          pytest tests/ -v --tb=short -m "not slow and not gpu and not integration"
+          python -m pytest tests/ -v --tb=short -m "not slow and not gpu and not integration"
 
   lint:
     runs-on: ubuntu-latest

.gitignore

@@ -8,3 +8,19 @@ dist/
 */.DS_Store
 *.DS_Store
 
+# Evaluation/Profiling ignores
+*.prof
+evaluation/sandbox/results/*
+!evaluation/sandbox/results/.gitkeep
+!evaluation/sandbox/results/run_local_eval.py
+
+# Sandbox SIF images
+*.sif
+
+# Sandbox Cache
+evaluation/sandbox/cache/
+
+# Slurm and Apptainer temporary files
+*.err
+*.out
+build-temp-*/

contextpilot/context_index/compute_distance_cpu.py

@@ -272,7 +272,7 @@ def compute_distance_matrix_cpu_optimized(contexts: List[List[int]],
     start = time.time()
     chunk_ids, original_positions, lengths, offsets = prepare_contexts_for_cpu(contexts)
     prep_time = time.time() - start
-    print(f"✓ Prepared in {prep_time:.1f}s")
+    print(f"+ Prepared in {prep_time:.1f}s")
 
     # Generate batches of pair indices
     print(f"\nGenerating pair batches...")
@@ -290,7 +290,7 @@ def compute_distance_matrix_cpu_optimized(contexts: List[List[int]],
     if current_batch:
         batches.append(current_batch)
 
-    print(f"✓ Generated {len(batches):,} batches")
+    print(f"+ Generated {len(batches):,} batches")
 
     # Prepare arguments for workers
     worker_args = [
@@ -306,13 +306,13 @@ def compute_distance_matrix_cpu_optimized(contexts: List[List[int]],
     start_time = time.time()
     processed = 0
 
-    with Pool(num_workers) as pool:
-        for batch_results in pool.imap_unordered(compute_batch_worker, worker_args):
+    if num_workers == 1:
+        # Bypass multiprocessing Pool entirely to save initialization overhead
+        for args in worker_args:
+            batch_results = compute_batch_worker(args)
             for i, j, dist in batch_results:
-                # Convert (i, j) to condensed index
                 condensed_idx = n * i - i * (i + 1) // 2 + j - i - 1
                 condensed_distances[condensed_idx] = dist
-
                 processed += 1
 
                 # Progress update
@@ -326,6 +326,27 @@ def compute_distance_matrix_cpu_optimized(contexts: List[List[int]],
                           f"Rate: {rate:,.0f} pairs/sec | "
                           f"Elapsed: {elapsed:.1f}s | "
                           f"ETA: {eta:.1f}s ({eta/60:.1f} min)")
+    else:
+        with Pool(num_workers) as pool:
+            for batch_results in pool.imap_unordered(compute_batch_worker, worker_args):
+                for i, j, dist in batch_results:
+                    # Convert (i, j) to condensed index
+                    condensed_idx = n * i - i * (i + 1) // 2 + j - i - 1
+                    condensed_distances[condensed_idx] = dist
+
+                    processed += 1
+
+                    # Progress update
+                    if processed % 100000 == 0 or processed == num_pairs:
+                        elapsed = time.time() - start_time
+                        rate = processed / elapsed if elapsed > 0 else 0
+                        eta = (num_pairs - processed) / rate if rate > 0 else 0
+                        progress_pct = processed / num_pairs * 100
+
+                        print(f"  {processed:,}/{num_pairs:,} ({progress_pct:.1f}%) | "
+                              f"Rate: {rate:,.0f} pairs/sec | "
+                              f"Elapsed: {elapsed:.1f}s | "
+                              f"ETA: {eta:.1f}s ({eta/60:.1f} min)")
 
     compute_time = time.time() - start_time
     total_time = compute_time + prep_time

contextpilot/server/http_server.py

@@ -229,6 +229,9 @@ class DeduplicateRequest(BaseModel):
 async def lifespan(app: FastAPI):
     """Lifespan context manager for startup/shutdown."""
     global _aiohttp_session
+    global _total_prompt_cache_hit_tokens
+
+    _total_prompt_cache_hit_tokens = 0
 
     # Initialize config from environment variables
     _init_config()
@@ -239,11 +242,15 @@ async def lifespan(app: FastAPI):
         logger.info(f"  max_tokens: {_max_tokens}")
     logger.info(f"  infer_api_url: {_infer_api_url}")
 
-    _aiohttp_session = aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=3600))
+    _aiohttp_session = aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=3600), trust_env=True)
     yield
     if _aiohttp_session:
         await _aiohttp_session.close()
     logger.info("ContextPilot Index Server shutting down...")
+
+    print("\n=== Final Telemetry Summary ===")
+    print(f"Total Prompt Cache Hit Tokens: {_total_prompt_cache_hit_tokens}")
+    # (Note: chars_saved_percentage and tools_filtered_percentage will be output here when plugins are fully integrated)
 
 
 app = FastAPI(
@@ -930,21 +937,36 @@ async def proxy_completions(request: Request):
             _index.track_request(request_id)
 
         # Pass request_id to inference engine so it can use the same ID for request tracking
-        # Engine will notify ContextPilot via /evict callback when this request is evicted
+        # We don't inject request_id into body anymore to avoid breaking strict APIs.
         if request_id:
-            body["rid"] = request_id          # SGLang
-            body["request_id"] = request_id   # vLLM
-            logger.info(f"Proxy: forwarding request with request_id={request_id}")
+            logger.info(f"Proxy: tracking request with request_id={request_id}")
         else:
             logger.info("Proxy: forwarding request without rid (no ContextPilot tracking)")
 
         # Forward to inference engine
         api_url = f"{infer_api_url}/v1/completions"
         logger.debug(f"Proxying to {api_url}")
 
-        async with _aiohttp_session.post(api_url, json=body) as response:
+        # Extract headers to forward
+        headers = dict(request.headers)
+        headers.pop("host", None)
+        headers.pop("content-length", None)
+        headers["accept-encoding"] = "gzip, deflate"
+
+        async with _aiohttp_session.post(api_url, json=body, headers=headers) as response:
             result = await response.json()
 
+            global _total_prompt_cache_hit_tokens
+            if response.status == 200 and isinstance(result, dict):
+                usage = result.get("usage", {})
+                if isinstance(usage, dict):
+                    # Legacy DeepSeek V3 format
+                    if "prompt_cache_hit_tokens" in usage:
+                        _total_prompt_cache_hit_tokens += int(usage["prompt_cache_hit_tokens"])
+                    # DeepSeek V4 / OpenAI standard format
+                    elif "prompt_tokens_details" in usage and isinstance(usage["prompt_tokens_details"], dict):
+                        _total_prompt_cache_hit_tokens += int(usage["prompt_tokens_details"].get("cached_tokens", 0))
+
             # Token tracking is handled by the inference engine via CONTEXTPILOT_INDEX_URL
             # The engine calls /evict after its internal cache eviction
 
@@ -999,17 +1021,39 @@ async def proxy_engine(path: str, request: Request):
 
             # Inject rid for SGLang cache tracking (same logic as proxy_completions)
             request_id = body.pop("request_id", None) or body.get("rid", None)
+
+            # Pop custom proxy parameters so upstream OpenAI doesn't reject them
+            body.pop("user_id", None)
+            body.pop("parent_id", None)
+            body.pop("_required_skills", None)
+
             if not request_id:
                 request_id = f"req-{uuid.uuid4().hex[:12]}"
                 logger.debug(f"Auto-assigned request_id={request_id}")
             if _index:
                 _index.track_request(request_id)
-            if request_id:
-                body["rid"] = request_id
-                body["request_id"] = request_id
 
-            async with _aiohttp_session.post(target_url, json=body) as response:
+            # Extract headers to forward (excluding hop-by-hop headers that aiohttp manages)
+            headers = dict(request.headers)
+            # Remove proxy-specific headers
+            headers.pop("host", None)
+            headers.pop("content-length", None)
+            headers["accept-encoding"] = "gzip, deflate"
+
+            async with _aiohttp_session.post(target_url, json=body, headers=headers) as response:
                 result = await response.json()
+
+                global _total_prompt_cache_hit_tokens
+                if response.status == 200 and isinstance(result, dict):
+                    usage = result.get("usage", {})
+                    if isinstance(usage, dict):
+                        # Legacy DeepSeek V3 format
+                        if "prompt_cache_hit_tokens" in usage:
+                            _total_prompt_cache_hit_tokens += int(usage["prompt_cache_hit_tokens"])
+                        # DeepSeek V4 / OpenAI standard format
+                        elif "prompt_tokens_details" in usage and isinstance(usage["prompt_tokens_details"], dict):
+                            _total_prompt_cache_hit_tokens += int(usage["prompt_tokens_details"].get("cached_tokens", 0))
+
                 return JSONResponse(content=result, status_code=response.status)
 
     except aiohttp.ClientError as e:

contextpilot/server/live_index.py

@@ -172,15 +172,15 @@ def build_and_schedule(self, contexts: List[List[int]],
         print("\n1. Building static index...")
         self.initial_result = self.fit_transform(contexts)
 
-        print(f"   ✓ Built tree with {self.initial_result.stats['total_nodes']} nodes")
-        print(f"   ✓ Leaf nodes: {self.initial_result.stats['leaf_nodes']}")
+        print(f"   + Built tree with {self.initial_result.stats['total_nodes']} nodes")
+        print(f"   + Leaf nodes: {self.initial_result.stats['leaf_nodes']}")
 
         # Step 2: Inter-context scheduling
         print("\n2. Scheduling contexts for optimal execution...")
         scheduled_reordered, scheduled_originals, final_mapping, groups = \
             self.inter_scheduler.schedule_contexts(self.initial_result)
 
-        print(f"   ✓ Created {len(groups)} execution groups")
+        print(f"   + Created {len(groups)} execution groups")
 
         self.scheduled_result = {
             'reordered_contexts': scheduled_reordered,
@@ -197,8 +197,8 @@ def build_and_schedule(self, contexts: List[List[int]],
             num_input_contexts=len(contexts)
         )
 
-        print(f"   ✓ Initialized {len(self.metadata)} nodes with metadata")
-        print(f"   ✓ Auto-assigned {len(request_id_mapping)} request IDs")
+        print(f"   + Initialized {len(self.metadata)} nodes with metadata")
+        print(f"   + Auto-assigned {len(request_id_mapping)} request IDs")
 
         # Add request_id mapping to result (dict and ordered list)
         self.scheduled_result['request_id_mapping'] = request_id_mapping
@@ -208,7 +208,7 @@ def build_and_schedule(self, contexts: List[List[int]],
         self.is_live = True
 
         print("\n" + "=" * 80)
-        print("✓ INDEX IS NOW LIVE - Ready for dynamic operations")
+        print("+ INDEX IS NOW LIVE - Ready for dynamic operations")
         print("=" * 80 + "\n")
 
         return self.scheduled_result
@@ -534,8 +534,8 @@ def build_incremental(self, contexts: List[List[int]],
                 # No match - will build new index for these
                 unmatched_contexts.append((i, context))
 
-        print(f"   ✓ Found {len(matched_contexts)} contexts with matches")
-        print(f"   ✓ Found {len(unmatched_contexts)} contexts without matches")
+        print(f"   + Found {len(matched_contexts)} contexts with matches")
+        print(f"   + Found {len(unmatched_contexts)} contexts without matches")
 
         # Prepare result arrays (will fill in order)
         request_ids = [None] * len(contexts)
@@ -585,7 +585,7 @@ def build_incremental(self, contexts: List[List[int]],
             )
             temp_result = temp_index.fit_transform(unmatched_only)
 
-            print(f"   ✓ Built temp index with {temp_result.stats['total_nodes']} nodes")
+            print(f"   + Built temp index with {temp_result.stats['total_nodes']} nodes")
 
             # Step 4: Merge temp index into global index
             print("\n4. Merging temp index into global index...")
@@ -606,16 +606,16 @@ def build_incremental(self, contexts: List[List[int]],
                 context_info.append((orig_idx, merged_request_ids[i], merged_search_paths[i]))
 
             merged_count = len(unmatched_contexts)
-            print(f"   ✓ Merged {merged_count} new subtrees under global root")
+            print(f"   + Merged {merged_count} new subtrees under global root")
 
         # Step 5: Schedule execution order
         print("\n5. Scheduling execution order for cache reuse...")
         scheduled_order = self._schedule_incremental(context_info)
         groups = self._group_by_path_prefix(context_info)
-        print(f"   ✓ Scheduled {len(scheduled_order)} contexts into {len(groups)} groups")
+        print(f"   + Scheduled {len(scheduled_order)} contexts into {len(groups)} groups")
 
         print("\n" + "=" * 80)
-        print(f"✓ INCREMENTAL BUILD COMPLETE")
+        print(f"+ INCREMENTAL BUILD COMPLETE")
         print(f"   Matched & inserted: {len(matched_contexts)}")
         print(f"   Built & merged: {merged_count}")
         print("=" * 80 + "\n")
@@ -906,15 +906,15 @@ def schedule_only(self, contexts: List[List[int]]) -> Dict:
         print("\n1. Building static index...")
         result = self.fit_transform(contexts)
 
-        print(f"   ✓ Built tree with {result.stats['total_nodes']} nodes")
-        print(f"   ✓ Leaf nodes: {result.stats['leaf_nodes']}")
+        print(f"   + Built tree with {result.stats['total_nodes']} nodes")
+        print(f"   + Leaf nodes: {result.stats['leaf_nodes']}")
 
         # Step 2: Inter-context scheduling
         print("\n2. Scheduling contexts for optimal execution...")
         scheduled_reordered, scheduled_originals, final_mapping, groups = \
             self.inter_scheduler.schedule_contexts(result)
 
-        print(f"   ✓ Created {len(groups)} execution groups")
+        print(f"   + Created {len(groups)} execution groups")
 
         # Return results without going live (stateless)
         scheduled_result = {
@@ -931,7 +931,7 @@ def schedule_only(self, contexts: List[List[int]]) -> Dict:
         }
 
         print("\n" + "=" * 80)
-        print("✓ BATCH SCHEDULED (Stateless - no cache tracking)")
+        print("+ BATCH SCHEDULED (Stateless - no cache tracking)")
         print("=" * 80 + "\n")
 
         return scheduled_result

docs/guides/multi_turn.md

@@ -256,10 +256,10 @@ print(f"New docs:    {result['new_docs']}")           # [2]
 
 | Operation | `/reorder` | `/deduplicate` |
 |-----------|----------|----------------|
-| Index build | ✓ | ✗ |
-| Clustering | ✓ | ✗ |
-| Search | ✓ | ✗ |
-| Deduplication | ✓ | ✓ |
+| Index build | + | ✗ |
+| Clustering | + | ✗ |
+| Search | + | ✗ |
+| Deduplication | + | + |
 | **Latency** | ~50-200ms | ~1-5ms |
 
 For multi-turn conversations, Turn 2+ typically doesn't need index operations — just deduplication against conversation history. The `/deduplicate` endpoint is **10-100x faster**.

eval_ablation_dynamic_pruning.sh

@@ -0,0 +1,53 @@
+#!/bin/bash
+export PYTHONPATH="$(pwd):$PYTHONPATH"
+export NO_PROXY="localhost,127.0.0.1,::1"
+export no_proxy="localhost,127.0.0.1,::1"
+
+LIMIT=0
+while [[ $# -gt 0 ]]; do
+  case $1 in
+    --limit) LIMIT="$2"; shift 2 ;;
+    *) echo "Unknown parameter: $1"; exit 1 ;;
+  esac
+done
+
+echo "============================================="
+echo "  Dynamic Pruning Hyperparameter Ablation"
+echo "  Limit: $LIMIT"
+echo "============================================="
+
+echo "[1/3] Booting ContextPilot Proxy Server..."
+python -m contextpilot.server.http_server --port 8000 --infer-api-url "https://api.openai.com" > proxy_ablation.log 2>&1 &
+PROXY_PID=$!
+sleep 5
+
+THRESHOLDS=(0.1 0.2 0.4 0.5)
+
+echo "[2/3] Running Python Ablation Iterations..."
+for THRESHOLD in "${THRESHOLDS[@]}"; do
+    echo "--- Running Threshold: $THRESHOLD ---"
+    python evaluation/benchmarks/run_bigcodebench.py \
+        --model gpt-5.5 \
+        --api_base "https://api.openai.com/v1" \
+        --concurrency 5 \
+        --limit "$LIMIT" \
+        --plugins "all" \
+        --eval_mode with_plugin \
+        --threshold "$THRESHOLD"
+
+    mv evaluation/benchmarks/results_with_plugin_gpt-5.5.jsonl evaluation/benchmarks/results_ablation_${THRESHOLD}_with_plugin_gpt-5.5.jsonl
+done
+
+echo "Shutting down Proxy Server to flush telemetry..."
+kill -INT $PROXY_PID
+sleep 2
+cat proxy_ablation.log
+
+echo "[3/3] Sandbox Evaluation..."
+for THRESHOLD in "${THRESHOLDS[@]}"; do
+    echo "--- Evaluating Sandbox for Threshold: $THRESHOLD ---"
+    cp evaluation/benchmarks/results_ablation_${THRESHOLD}_with_plugin_gpt-5.5.jsonl evaluation/benchmarks/elm_samples_full.jsonl
+    cd evaluation/benchmarks && bash run_sandbox_eval_full.sh && cd ../..
+done
+
+echo "Pipeline Complete!"