local-llm-router

name	local-llm-router
description	Route AI coding queries to local LLMs in air-gapped networks. Integrates Serena MCP for semantic code understanding. Use when working offline, with local models (Ollama, LM Studio, Jan, OpenWebUI), or in secure/closed environments. Triggers on local LLM, Ollama, LM Studio, Jan, air-gapped, offline AI, Serena, local inference, closed network, model routing, defense network, secure coding.

Local LLM Router for Air-Gapped Networks

Intelligent routing of AI coding queries to local LLMs with Serena LSP integration for secure, offline-capable development environments.

Prerequisites (CRITICAL)

Before using this skill, ensure:

1. Serena MCP Server installed and running (PRIMARY TOOL)
2. At least one local LLM service running (Ollama, LM Studio, Jan, etc.)

# Install Serena (required)
pip install serena
# Or via uvx
uvx --from git+https://github.com/oraios/serena serena start-mcp-server

# Verify local LLM service
curl http://localhost:11434/api/version  # Ollama
curl http://localhost:1234/v1/models     # LM Studio
curl http://localhost:1337/v1/models     # Jan

Quick Start

import httpx
import asyncio
from dataclasses import dataclass
from enum import Enum
from typing import Optional

class TaskCategory(Enum):
    CODING = "coding"
    REASONING = "reasoning"
    ANALYSIS = "analysis"
    DOCUMENTATION = "documentation"

@dataclass
class RouterConfig:
    """Local LLM Router configuration."""
    ollama_url: str = "http://localhost:11434"
    lmstudio_url: str = "http://localhost:1234"
    jan_url: str = "http://localhost:1337"
    serena_enabled: bool = True
    timeout: int = 30

async def quick_route(query: str, config: RouterConfig = RouterConfig()):
    """Quick routing example - detects services and routes query."""

    # 1. Detect available services
    services = await discover_services(config)
    if not services:
        raise RuntimeError("No local LLM services available")

    # 2. Classify task
    category = classify_task(query)

    # 3. Select best model for task
    model = select_model(category, services)

    # 4. Execute query
    return await execute_query(query, model, services[0])

# Example usage
async def main():
    response = await quick_route("Write a function to parse JSON safely")
    print(response)

asyncio.run(main())

Serena Integration (PRIMARY TOOL)

CRITICAL: Serena MCP MUST be invoked FIRST for all code-related tasks. This provides semantic understanding of the codebase before routing to an LLM.

Why Serena First?

1. Token Efficiency: Serena extracts only relevant code context
2. Accuracy: Symbol-level operations vs grep-style searches
3. Codebase Awareness: Understands types, references, call hierarchies
4. Edit Precision: Applies changes at symbol level, not string matching

Serena MCP Setup

import subprocess
import json
from typing import Any

class SerenaMCP:
    """Serena MCP client for code intelligence."""

    def __init__(self, workspace_root: str):
        self.workspace = workspace_root
        self.process = None

    async def start(self):
        """Start Serena MCP server."""
        self.process = subprocess.Popen(
            ["serena", "start-mcp-server", "--workspace", self.workspace],
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE
        )

    async def call(self, method: str, params: dict) -> Any:
        """Call Serena MCP method."""
        request = {
            "jsonrpc": "2.0",
            "id": 1,
            "method": method,
            "params": params
        }
        self.process.stdin.write(json.dumps(request).encode() + b"\n")
        self.process.stdin.flush()
        response = self.process.stdout.readline()
        return json.loads(response)

      () -> :
        
          .call(, {: name})

      () -> :
        
          .call(, {
            : file,
            : line,
            : char
        })

      () -> :
        
          .call(, {
            : file,
            : line,
            : char
        })

      () -> :
        
          .call(, {: file})

      () -> :
        
          .call(, {: file, : edits})


SERENA_TOOLS = {
    
    : {: , : [, ]},
    : {: , : [, ]},
    : {: , : [, ]},

    
    : {: , : []},
    : {: , : []},
    : {: , : []},

    
    : {: , : []},
    : {: , : []},
    : {: , : []},

    
    : {: , : []},
    : {: , : []},

    
    : {: , : [, ]},
    : {: , : []},
}

Serena-First Request Handler

async def handle_code_request(
    query: str,
    file_context: Optional[dict] = None,
    serena: SerenaMCP = None,
    router: "LLMRouter" = None
):
    """
    Handle code request with Serena-first pattern.

    CRITICAL: Serena is ALWAYS invoked first for code tasks.
    """

    # Step 1: Classify the task
    category = classify_task(query)

    # Step 2: ALWAYS use Serena for code context (if available)
    serena_context = {}
    if serena and file_context:
        # Gather semantic context from Serena
        if file_context.get("file") and file_context.get("position"):
            file = file_context["file"]
            line = file_context["position"]["line"]
            char = file_context["position"]["character"]

            # Get hover info (type, docs)
            serena_context["hover"] = await serena.get_hover_info(file, line, char)

            # For refactoring/analysis, get references
            if category in [TaskCategory.ANALYSIS, TaskCategory.CODING]:
                if "refactor" in query.lower() or "rename" in query.lower():
                    serena_context["references"] = await serena.get_references(
                        file, line, char
                    )

            # Always get diagnostics for the file
            serena_context[] =  serena.get_diagnostics(file)

    
    enriched_query = build_enriched_query(query, serena_context)

    
    model = router.select_model(category)
    response =  router.execute(enriched_query, model)

    
     serena  contains_code_edit(response):
        edits = parse_code_edits(response)
         serena.apply_edit(file_context[], edits)

     response

 () -> :
    
    parts = [query]

     serena_context.get():
        hover = serena_context[]
        parts.append()

     serena_context.get():
        refs = serena_context[]
        parts.append()
         ref  refs[:]:  
            parts.append()

     serena_context.get():
        diags = serena_context[]
         diags:
            parts.append()
             diag  diags[:]:
                parts.append()

     .join(parts)

Service Discovery

Supported Services

Service	Default Endpoint	Health Check	Models Endpoint	Chat Endpoint	API Style
Ollama	localhost:11434	/api/version	/api/tags	/api/chat	Native
LM Studio	localhost:1234	/v1/models	/v1/models	/v1/chat/completions	OpenAI
Jan	localhost:1337	/v1/models	/v1/models	/v1/chat/completions	OpenAI
OpenWebUI	localhost:3000	/api/health	/api/models	/api/chat	Custom
LocalAI	localhost:8080	/readyz	/v1/models	/v1/chat/completions	OpenAI
vLLM	localhost:8000	/health	/v1/models	/v1/chat/completions	OpenAI
llama.cpp	localhost:8080	/health	/v1/models	/v1/chat/completions	OpenAI
Kobold.cpp	localhost:5001	/api/v1/info	/api/v1/models	/api/v1/generate	Custom
GPT4All	localhost:4891	/v1/models	/v1/models	/v1/chat/completions	OpenAI
text-generation-webui

OS Detection

import sys
import os
import platform
from dataclasses import dataclass

@dataclass
class OSInfo:
    platform: str      # 'windows', 'linux', 'darwin'
    release: str
    arch: str          # 'x64', 'arm64'
    is_wsl: bool
    is_container: bool

def detect_os() -> OSInfo:
    """Detect operating system and environment."""
    plat = sys.platform

    # Normalize platform name
    if plat == 'win32':
        plat = 'windows'
    elif plat == 'darwin':
        plat = 'darwin'
    else:
        plat = 'linux'

    # WSL detection
    is_wsl = False
    if plat == 'linux':
        try:
            with open('/proc/version', 'r') as f:
                is_wsl = 'microsoft' in f.read().lower()
        except FileNotFoundError:
            pass
        is_wsl = is_wsl or os.environ.get('WSL_DISTRO_NAME') is not None

    # Container detection
    is_container = (
        os.path.exists() 
        os.environ.get()   
    )
      is_container  plat == :
        :
             (, )  f:
                is_container =   f.read()    f.read()
         FileNotFoundError:
            

     OSInfo(
        platform=plat,
        release=platform.release(),
        arch=platform.machine(),
        is_wsl=is_wsl,
        is_container=is_container
    )

 () -> :
    
     os_info.is_wsl  os_info.is_container:
        
         endpoint.replace(, )
     endpoint

Service Discovery Implementation

import httpx
import asyncio
from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional

@dataclass
class DiscoveredModel:
    id: str
    name: str
    size: int = 0
    family: Optional[str] = None
    context_length: int = 4096
    quantization: Optional[str] = None

@dataclass
class LLMService:
    name: str
    type: str  # 'ollama', 'lmstudio', 'jan', 'openwebui', 'custom'
    endpoint: str
    status: str = 'unknown'  # 'online', 'offline', 'unknown'
    models: list = field(default_factory=list)
    last_checked: datetime = None
    api_style: str = 'openai'  # 'openai', 'native'

    # Endpoint paths
    health_path: str = '/v1/models'
    models_path: str = '/v1/models'
    chat_path: str = '/v1/chat/completions'

# Default service configurations
SERVICE_DEFAULTS = {
    : LLMService(
        name=,
        =,
        endpoint=,
        health_path=,
        models_path=,
        chat_path=,
        api_style=
    ),
    : LLMService(
        name=,
        =,
        endpoint=,
        health_path=,
        models_path=,
        chat_path=,
        api_style=
    ),
    : LLMService(
        name=,
        =,
        endpoint=,
        health_path=,
        models_path=,
        chat_path=,
        api_style=
    ),
    : LLMService(
        name=,
        =,
        endpoint=,
        health_path=,
        models_path=,
        chat_path=,
        api_style=
    ),
    : LLMService(
        name=,
        =,
        endpoint=,
        health_path=,
        models_path=,
        chat_path=,
        api_style=
    ),
    : LLMService(
        name=,
        =,
        endpoint=,
        health_path=,
        models_path=,
        chat_path=,
        api_style=
    ),
    : LLMService(
        name=,
        =,
        endpoint=,
        health_path=,
        models_path=,
        chat_path=,
        api_style=
    ),
    : LLMService(
        name=,
        =,
        endpoint=,
        health_path=,
        models_path=,
        chat_path=,
        api_style=
    ),
    : LLMService(
        name=,
        =,
        endpoint=,
        health_path=,
        models_path=,
        chat_path=,
        api_style=
    ),
}

 :
    

     ():
        .services: [, LLMService] = {}
        .os_info = detect_os()
        .custom_endpoints = custom_endpoints  []
        ._client = httpx.AsyncClient(timeout=)

      () -> [LLMService]:
        
        discovered = []

        
        tasks = []
         key, default  SERVICE_DEFAULTS.items():
            service = LLMService(
                name=default.name,
                =default.,
                endpoint=adjust_endpoint_for_os(default.endpoint, .os_info),
                health_path=default.health_path,
                models_path=default.models_path,
                chat_path=default.chat_path,
                api_style=default.api_style
            )
            tasks.append(._check_service(service))

        
         custom  .custom_endpoints:
            service = LLMService(
                name=custom.get(, ),
                =,
                endpoint=custom[],
                health_path=custom.get(, ),
                models_path=custom.get(, ),
                chat_path=custom.get(, ),
                api_style=custom.get(, )
            )
            tasks.append(._check_service(service))

        results =  asyncio.gather(*tasks, return_exceptions=)

         result  results:
             (result, LLMService)  result.status == :
                discovered.append(result)
                .services[result.] = result

         discovered

      () -> LLMService:
        
        :
            
            response =  ._client.get(
                
            )

             response.status_code == :
                service.status = 
                service.last_checked = datetime.now()

                
                service.models =  ._discover_models(service)
            :
                service.status = 

         (httpx.ConnectError, httpx.TimeoutException):
            service.status = 

         service

      () -> [DiscoveredModel]:
        
        :
            response =  ._client.get(
                
            )
            data = response.json()

            
             service. == :
                 [
                    DiscoveredModel(
                        =m[],
                        name=m[],
                        size=m.get(, ),
                        family=m.get(, {}).get(),
                        context_length=._infer_context_length(m[])
                    )
                     m  data.get(, [])
                ]
            :  
                 [
                    DiscoveredModel(
                        =m[],
                        name=m[],
                        context_length=m.get(, )
                    )
                     m  data.get(, [])
                ]
         Exception:
             []

     () -> :
        
        name_lower = model_name.lower()

        
           name_lower    name_lower:
             
           name_lower:
             
           name_lower:
             
           name_lower:
             

        
           name_lower:
               
           name_lower:
             
           name_lower    name_lower:
             
           name_lower:
             
           name_lower:
             

           

Task Classification

Classification System

import re
from enum import Enum
from dataclasses import dataclass

class TaskCategory(Enum):
    CODING = "coding"
    REASONING = "reasoning"
    ANALYSIS = "analysis"
    DOCUMENTATION = "documentation"

@dataclass
class ClassificationResult:
    category: TaskCategory
    confidence: float  # 0.0 - 1.0
    requires_serena: bool
    keywords_matched: list[str]

# Task patterns (regex)
TASK_PATTERNS = {
    TaskCategory.CODING: [
        r"(?:write|create|implement|code|generate)\s+(?:a\s+)?(?:function|class|method|component)",
        r"(?:fix|debug|solve)\s+(?:this|the)\s+(?:bug|error|issue)",
        r"refactor\s+(?:this|the)",
        r"add\s+(?:error\s+handling|validation|logging|tests?)",
        r"complete\s+(?:this|the)\s+code",
        r"(?:convert|translate)\s+(?:this|the)\s+code",
        r"(?:optimize|improve)\s+(?:this|the)\s+(?:function|code|performance)",
    ],
    TaskCategory.REASONING: [
        r"(?:design|architect|plan)\s+(?:a|the)\s+(?:system|architecture|solution)",
        r"how\s+should\s+(?:I|we)\s+(?:approach|structure|implement)",
        r"what\s+(?:is|would\s+be)\s+the\s+best\s+(?:way|approach|pattern)",
        r"explain\s+the\s+(?:logic|reasoning|algorithm)",
        r"compare\s+(?:and\s+contrast|between)",
        r"(?:recommend|suggest)\s+(?:an?\s+)?(?:approach|solution|pattern)",
        r"trade-?offs?\s+(?:between|of)",
    ],
    TaskCategory.ANALYSIS: [
        r"(?:review|analyze|audit)\s+(?:this|the)\s+code",
        ,
        ,
        ,
        ,
        ,
    ],
    TaskCategory.DOCUMENTATION: [
        ,
        ,
        ,
        ,
        ,
        ,
    ],
}


KEYWORD_WEIGHTS = {
    
    : (TaskCategory.CODING, ),
    : (TaskCategory.CODING, ),
    : (TaskCategory.CODING, ),
    : (TaskCategory.CODING, ),
    : (TaskCategory.CODING, ),
    : (TaskCategory.CODING, ),
    : (TaskCategory.CODING, ),
    : (TaskCategory.CODING, ),

    
    : (TaskCategory.REASONING, ),
    : (TaskCategory.REASONING, ),
    : (TaskCategory.REASONING, ),
    : (TaskCategory.REASONING, ),
    : (TaskCategory.REASONING, ),
    : (TaskCategory.REASONING, ),
    : (TaskCategory.REASONING, ),

    
    : (TaskCategory.ANALYSIS, ),
    : (TaskCategory.ANALYSIS, ),
    : (TaskCategory.ANALYSIS, ),
    : (TaskCategory.ANALYSIS, ),
    : (TaskCategory.ANALYSIS, ),
    : (TaskCategory.ANALYSIS, ),

    
    : (TaskCategory.DOCUMENTATION, ),
    : (TaskCategory.DOCUMENTATION, ),
    : (TaskCategory.DOCUMENTATION, ),
    : (TaskCategory.DOCUMENTATION, ),
    : (TaskCategory.DOCUMENTATION, ),
}

 () -> ClassificationResult:
    
    query_lower = query.lower()
    scores = {cat:   cat  TaskCategory}
    matched_keywords = []

    
     category, patterns  TASK_PATTERNS.items():
         pattern  patterns:
             re.search(pattern, query_lower):
                scores[category] += 

    
    words = re.findall(, query_lower)
     word  words:
         word  KEYWORD_WEIGHTS:
            category, weight = KEYWORD_WEIGHTS[word]
            scores[category] += weight * 
            matched_keywords.append(word)

    
    best_category = (scores, key=scores.get)
    confidence = (scores[best_category], )

    
     confidence < :
        best_category = TaskCategory.CODING
        confidence = 

    
    requires_serena = (
        best_category == TaskCategory.ANALYSIS 
        (kw  query_lower  kw  [
            , , , ,
            , , , 
        ])
    )

     ClassificationResult(
        category=best_category,
        confidence=confidence,
        requires_serena=requires_serena,
        keywords_matched=matched_keywords
    )

Model Selection

Model Capability Matrix

from dataclasses import dataclass
from typing import Optional

@dataclass
class ModelCapability:
    id: str
    family: str
    context_window: int
    vram_gb: float
    categories: list[TaskCategory]
    performance_scores: dict[TaskCategory, int]  # 0-100
    tier: int  # 1=best, 2=good, 3=basic
    quantization: Optional[str] = None

# Comprehensive model database (40+ models) - Updated January 2025
MODEL_DATABASE: dict[str, ModelCapability] = {
    # === CODING SPECIALISTS (Tier 1) ===
    "deepseek-v3": ModelCapability(
        id="deepseek-v3",
        family="deepseek",
        context_window=128000,
        vram_gb=48,  # MoE: 685B total, 37B active
        categories=[TaskCategory.CODING, TaskCategory.REASONING, TaskCategory.ANALYSIS],
        performance_scores={
            TaskCategory.CODING: 99,
            TaskCategory.REASONING: 97,
            TaskCategory.ANALYSIS: 96,
            TaskCategory.DOCUMENTATION: 92
        },
        tier=1
    ),
    "qwen2.5-coder-32b": ModelCapability(
        id="qwen2.5-coder-32b",
        family="qwen",
        context_window=131072,
        vram_gb=,
        categories=[TaskCategory.CODING, TaskCategory.ANALYSIS],
        performance_scores={
            TaskCategory.CODING: ,
            TaskCategory.REASONING: ,
            TaskCategory.ANALYSIS: ,
            TaskCategory.DOCUMENTATION: 
        },
        tier=
    ),
    : ModelCapability(
        =,
        family=,
        context_window=,
        vram_gb=,  
        categories=[TaskCategory.CODING, TaskCategory.ANALYSIS, TaskCategory.REASONING],
        performance_scores={
            TaskCategory.CODING: ,
            TaskCategory.REASONING: ,
            TaskCategory.ANALYSIS: ,
            TaskCategory.DOCUMENTATION: 
        },
        tier=
    ),
    : ModelCapability(
        =,
        family=,
        context_window=,
        vram_gb=,
        categories=[TaskCategory.CODING],
        performance_scores={
            TaskCategory.CODING: ,
            TaskCategory.REASONING: ,
            TaskCategory.ANALYSIS: ,
            TaskCategory.DOCUMENTATION: 
        },
        tier=
    ),
    : ModelCapability(
        =,
        family=,
        context_window=,
        vram_gb=,
        categories=[TaskCategory.CODING],
        performance_scores={
            TaskCategory.CODING: ,
            TaskCategory.REASONING: ,
            TaskCategory.ANALYSIS: ,
            TaskCategory.DOCUMENTATION: 
        },
        tier=
    ),
    : ModelCapability(
        =,
        family=,
        context_window=,
        vram_gb=,
        categories=[TaskCategory.CODING],
        performance_scores={
            TaskCategory.CODING: ,
            TaskCategory.REASONING: ,
            TaskCategory.ANALYSIS: ,
            TaskCategory.DOCUMENTATION: 
        },
        tier=
    ),
    : ModelCapability(
        =,
        family=,
        context_window=,
        vram_gb=,
        categories=[TaskCategory.CODING],
        performance_scores={
            TaskCategory.CODING: ,
            TaskCategory.REASONING: ,
            TaskCategory.ANALYSIS: ,
            TaskCategory.DOCUMENTATION: 
        },
        tier=
    ),
    : ModelCapability(
        =,
        family=,
        context_window=,
        vram_gb=,
        categories=[TaskCategory.CODING],
        performance_scores={
            TaskCategory.CODING: ,
            TaskCategory.REASONING: ,
            TaskCategory.ANALYSIS: ,
            TaskCategory.DOCUMENTATION: 
        },
        tier=
    ),
    : ModelCapability(
        =,
        family=,
        context_window=,
        vram_gb=,
        categories=[TaskCategory.CODING],
        performance_scores={
            TaskCategory.CODING: ,
            TaskCategory.REASONING: ,
            TaskCategory.ANALYSIS: ,
            TaskCategory.DOCUMENTATION: 
        },
        tier=
    ),

    
    : ModelCapability(
        =,
        family=,
        context_window=,
        vram_gb=,  
        categories=[TaskCategory.REASONING, TaskCategory.CODING],
        performance_scores={
            TaskCategory.CODING: ,
            TaskCategory.REASONING: ,
            TaskCategory.ANALYSIS: ,
            TaskCategory.DOCUMENTATION: 
        },
        tier=
    ),
    : ModelCapability(
        =,
        family=,
        context_window=,
        vram_gb=,
        categories=[TaskCategory.REASONING, TaskCategory.CODING],
        performance_scores={
            TaskCategory.CODING: ,
            TaskCategory.REASONING: ,
            TaskCategory.ANALYSIS: ,
            TaskCategory.DOCUMENTATION: 
        },
        tier=
    ),
    : ModelCapability(
        =,
        family=,
        context_window=,
        vram_gb=,
        categories=[TaskCategory.REASONING, TaskCategory.CODING],
        performance_scores={
            TaskCategory.CODING: ,
            TaskCategory.REASONING: ,
            TaskCategory.ANALYSIS: ,
            TaskCategory.DOCUMENTATION: 
        },
        tier=
    ),
    : ModelCapability(
        =,
        family=,
        context_window=,
        vram_gb=,
        categories=[TaskCategory.REASONING, TaskCategory.CODING],
        performance_scores={
            TaskCategory.CODING: ,
            TaskCategory.REASONING: ,
            TaskCategory.ANALYSIS: ,
            TaskCategory.DOCUMENTATION: 
        },
        tier=
    ),
    : ModelCapability(
        =,
        family=,
        context_window=,
        vram_gb=,
        categories=[TaskCategory.REASONING, TaskCategory.CODING],
        performance_scores={
            TaskCategory.CODING: ,
            TaskCategory.REASONING: ,
            TaskCategory.ANALYSIS: ,
            TaskCategory.DOCUMENTATION: 
        },
        tier=
    ),
    : ModelCapability(
        =,
        family=,
        context_window=,
        vram_gb=,
        categories=[TaskCategory.REASONING, TaskCategory.CODING],
        performance_scores={
            TaskCategory.CODING: ,
            TaskCategory.REASONING: ,
            TaskCategory.ANALYSIS: ,
            TaskCategory.DOCUMENTATION: 
        },
        tier=
    ),
    : ModelCapability(
        =,
        family=,
        context_window=,
        vram_gb=,
        categories=[TaskCategory.REASONING, TaskCategory.DOCUMENTATION],
        performance_scores={
            TaskCategory.CODING: ,
            TaskCategory.REASONING: ,
            TaskCategory.ANALYSIS: ,
            TaskCategory.DOCUMENTATION: 
        },
        tier=
    ),
    : ModelCapability(
        =,
        family=,
        context_window=,
        vram_gb=,
        categories=[TaskCategory.REASONING, TaskCategory.CODING],
        performance_scores={
            TaskCategory.CODING: ,
            TaskCategory.REASONING: ,
            TaskCategory.ANALYSIS: ,
            TaskCategory.DOCUMENTATION: 
        },
        tier=
    ),
    : ModelCapability(
        =,
        family=,
        context_window=,
        vram_gb=,
        categories=[TaskCategory.REASONING],
        performance_scores={
            TaskCategory.CODING: ,
            TaskCategory.REASONING: ,
            TaskCategory.ANALYSIS: ,
            TaskCategory.DOCUMENTATION: 
        },
        tier=
    ),
    : ModelCapability(
        =,
        family=,
        context_window=,
        vram_gb=,
        categories=[TaskCategory.REASONING, TaskCategory.DOCUMENTATION],
        performance_scores={
            TaskCategory.CODING: ,
            TaskCategory.REASONING: ,
            TaskCategory.ANALYSIS: ,
            TaskCategory.DOCUMENTATION: 
        },
        tier=
    ),
    : ModelCapability(
        =,
        family=,
        context_window=,
        vram_gb=,
        categories=[TaskCategory.REASONING, TaskCategory.CODING],
        performance_scores={
            TaskCategory.CODING: ,
            TaskCategory.REASONING: ,
            TaskCategory.ANALYSIS: ,
            TaskCategory.DOCUMENTATION: 
        },
        tier=
    ),
    : ModelCapability(
        =,
        family=,
        context_window=,
        vram_gb=,
        categories=[TaskCategory.REASONING],
        performance_scores={
            TaskCategory.CODING: ,
            TaskCategory.REASONING: ,
            TaskCategory.ANALYSIS: ,
            TaskCategory.DOCUMENTATION: 
        },
        tier=
    ),
    : ModelCapability(
        =,
        family=,
        context_window=,
        vram_gb=,
        categories=[TaskCategory.REASONING, TaskCategory.CODING],
        performance_scores={
            TaskCategory.CODING: ,
            TaskCategory.REASONING: ,
            TaskCategory.ANALYSIS: ,
            TaskCategory.DOCUMENTATION: 
        },
        tier=
    ),
    : ModelCapability(
        =,
        family=,
        context_window=,
        vram_gb=,
        categories=[TaskCategory.REASONING],
        performance_scores={
            TaskCategory.CODING: ,
            TaskCategory.REASONING: ,
            TaskCategory.ANALYSIS: ,
            TaskCategory.DOCUMENTATION: 
        },
        tier=
    ),

    
    : ModelCapability(
        =,
        family=,
        context_window=,
        vram_gb=,
        categories=[TaskCategory.ANALYSIS],
        performance_scores={
            TaskCategory.CODING: ,
            TaskCategory.REASONING: ,
            TaskCategory.ANALYSIS: ,
            TaskCategory.DOCUMENTATION: 
        },
        tier=
    ),

    
    : ModelCapability(
        =,
        family=,
        context_window=,
        vram_gb=,
        categories=[TaskCategory.DOCUMENTATION],
        performance_scores={
            TaskCategory.CODING: ,
            TaskCategory.REASONING: ,
            TaskCategory.ANALYSIS: ,
            TaskCategory.DOCUMENTATION: 
        },
        tier=
    ),
    : ModelCapability(
        =,
        family=,
        context_window=,
        vram_gb=,
        categories=[TaskCategory.DOCUMENTATION],
        performance_scores={
            TaskCategory.CODING: ,
            TaskCategory.REASONING: ,
            TaskCategory.ANALYSIS: ,
            TaskCategory.DOCUMENTATION: 
        },
        tier=
    ),

    
    : ModelCapability(
        =,
        family=,
        context_window=,
        vram_gb=,
        categories=[TaskCategory.CODING, TaskCategory.REASONING],
        performance_scores={
            TaskCategory.CODING: ,
            TaskCategory.REASONING: ,
            TaskCategory.ANALYSIS: ,
            TaskCategory.DOCUMENTATION: 
        },
        tier=
    ),
    : ModelCapability(
        =,
        family=,
        context_window=,
        vram_gb=,
        categories=[TaskCategory.CODING, TaskCategory.REASONING],
        performance_scores={
            TaskCategory.CODING: ,
            TaskCategory.REASONING: ,
            TaskCategory.ANALYSIS: ,
            TaskCategory.DOCUMENTATION: 
        },
        tier=
    ),
    : ModelCapability(
        =,
        family=,
        context_window=,
        vram_gb=,
        categories=[TaskCategory.REASONING, TaskCategory.DOCUMENTATION],
        performance_scores={
            TaskCategory.CODING: ,
            TaskCategory.REASONING: ,
            TaskCategory.ANALYSIS: ,
            TaskCategory.DOCUMENTATION: 
        },
        tier=
    ),
    : ModelCapability(
        =,
        family=,
        context_window=,
        vram_gb=,
        categories=[TaskCategory.REASONING, TaskCategory.DOCUMENTATION],
        performance_scores={
            TaskCategory.CODING: ,
            TaskCategory.REASONING: ,
            TaskCategory.ANALYSIS: ,
            TaskCategory.DOCUMENTATION: 
        },
        tier=
    ),
    : ModelCapability(
        =,
        family=,
        context_window=,
        vram_gb=,
        categories=[TaskCategory.CODING],
        performance_scores={
            TaskCategory.CODING: ,
            TaskCategory.REASONING: ,
            TaskCategory.ANALYSIS: ,
            TaskCategory.DOCUMENTATION: 
        },
        tier=
    ),
    : ModelCapability(
        =,
        family=,
        context_window=,
        vram_gb=,
        categories=[TaskCategory.CODING],
        performance_scores={
            TaskCategory.CODING: ,
            TaskCategory.REASONING: ,
            TaskCategory.ANALYSIS: ,
            TaskCategory.DOCUMENTATION: 
        },
        tier=
    ),
    : ModelCapability(
        =,
        family=,
        context_window=,
        vram_gb=,
        categories=[TaskCategory.CODING, TaskCategory.REASONING],
        performance_scores={
            TaskCategory.CODING: ,
            TaskCategory.REASONING: ,
            TaskCategory.ANALYSIS: ,
            TaskCategory.DOCUMENTATION: 
        },
        tier=
    ),
    : ModelCapability(
        =,
        family=,
        context_window=,
        vram_gb=,
        categories=[TaskCategory.REASONING],
        performance_scores={
            TaskCategory.CODING: ,
            TaskCategory.REASONING: ,
            TaskCategory.ANALYSIS: ,
            TaskCategory.DOCUMENTATION: 
        },
        tier=
    ),
    : ModelCapability(
        =,
        family=,
        context_window=,
        vram_gb=,
        categories=[TaskCategory.REASONING, TaskCategory.DOCUMENTATION],
        performance_scores={
            TaskCategory.CODING: ,
            TaskCategory.REASONING: ,
            TaskCategory.ANALYSIS: ,
            TaskCategory.DOCUMENTATION: 
        },
        tier=
    ),
}


TASK_MODEL_PRIORITY = {
    TaskCategory.CODING: [
        
        , , ,
        
        , , ,
        , ,
        
        , , 
    ],
    TaskCategory.REASONING: [
        
        , , ,
        , ,
        
        , , ,
        , ,
        
        , , 
    ],
    TaskCategory.ANALYSIS: [
        
        , , ,
        , 
    ],
    TaskCategory.DOCUMENTATION: [
        , , ,
        , , 
    ],
}

Model Selection Logic

from typing import Optional

class ModelSelector:
    """Select optimal model for task based on availability and requirements."""

    def __init__(self, available_models: list[str]):
        self.available = set(m.lower() for m in available_models)

    def select(
        self,
        category: TaskCategory,
        required_context: int = 0,
        max_vram_gb: Optional[float] = None
    ) -> Optional[str]:
        """Select best available model for task category."""

        # Get priority list for category
        priority_list = TASK_MODEL_PRIORITY.get(category, [])

        for model_id in priority_list:
            # Check if model is available
            if not self._is_available(model_id):
                continue

            # Check model capability
            capability = MODEL_DATABASE.get(model_id)
            if not capability:
                continue

            # Check context window requirement
            if required_context > 0 and capability.context_window < required_context:
                continue

            # Check VRAM constraint
             max_vram_gb  capability.vram_gb > max_vram_gb:
                

             model_id

        
         model_id, capability  MODEL_DATABASE.items():
             ._is_available(model_id):
                 model_id

         

     () -> :
        
        model_lower = model_id.lower()

        
         model_lower  .available:
             

        
         avail  .available:
             model_lower  avail  avail  model_lower:
                 

         

     () -> []:
        
        priority_list = TASK_MODEL_PRIORITY.get(category, [])

        available_in_priority = [
            m  m  priority_list  ._is_available(m)
        ]

        
        fallbacks = []
         model_id  available_in_priority:
            capability = MODEL_DATABASE.get(model_id)
             capability  capability.tier >= :
                fallbacks.append(model_id)

         fallbacks

Context Management

Token Counting

from abc import ABC, abstractmethod
import re

class TokenCounter(ABC):
    """Base class for token counting."""

    @abstractmethod
    def count(self, text: str) -> int:
        pass

class EstimationCounter(TokenCounter):
    """Estimation-based token counter (no external dependencies)."""

    def __init__(self, chars_per_token: float = 4.0):
        self.chars_per_token = chars_per_token

    def count(self, text: str) -> int:
        return int(len(text) / self.chars_per_token)

class QwenCounter(TokenCounter):
    """Token counter for Qwen models."""

    def count(self, text: str) -> int:
        # Qwen uses slightly different tokenization
        return int(len(text) / 3.5)

class LlamaCounter():
    

     () -> :
        
         ((text) / )


TOKEN_COUNTERS = {
    : QwenCounter(),
    : EstimationCounter(),
    : LlamaCounter(),
    : EstimationCounter(),
    : EstimationCounter(),
    : EstimationCounter(),
}

 () -> TokenCounter:
    
    capability = MODEL_DATABASE.get(model_id)
     capability:
         TOKEN_COUNTERS.get(capability.family, TOKEN_COUNTERS[])
     TOKEN_COUNTERS[]

Context Manager

from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional

@dataclass
class Message:
    role: str  # 'system', 'user', 'assistant', 'tool'
    content: str
    timestamp: datetime = field(default_factory=datetime.now)
    token_count: int = 0
    metadata: dict = field(default_factory=dict)

@dataclass
class ConversationContext:
    session_id: str
    messages: list[Message] = field(default_factory=list)
    total_tokens: int = 0
    system_prompt: str = ""
    system_prompt_tokens: int = 0
    active_model: str = ""
    model_history: list[str] = field(default_factory=list)
    compaction_count: int = 0

class ContextManager:
    """Manage conversation context with compaction support."""

    def __init__(
        self,
        session_id: str,
        system_prompt: str = "",
        compaction_threshold: float = 0.8,  # 80% of context window
        compaction_target:  = ,      
        preserve_recent:  =             
    ):
        .context = ConversationContext(
            session_id=session_id,
            system_prompt=system_prompt
        )
        .compaction_threshold = compaction_threshold
        .compaction_target = compaction_target
        .preserve_recent = preserve_recent
        ._counter: [TokenCounter] = 

     ():
        
         .context.active_model:
            .context.model_history.append(.context.active_model)
        .context.active_model = model_id
        ._counter = get_token_counter(model_id)

        
        ._recount_tokens()

     ():
        
        token_count = ._counter.count(content)  ._counter  

        message = Message(
            role=role,
            content=content,
            token_count=token_count,
            metadata=metadata  {}
        )

        .context.messages.append(message)
        .context.total_tokens += token_count

     () -> :
        
        threshold = (max_tokens * .compaction_threshold)

         .context.total_tokens > threshold:
            ._compact(max_tokens)
             
         

     ():
        
        target = (max_tokens * .compaction_target)

        
         msg  .context.messages:
             msg.role ==   msg.token_count > :
                original = msg.token_count
                msg.content = 
                msg.token_count = ._counter.count(msg.content)
                msg.metadata[] = 
                msg.metadata[] = original

        ._recalculate_total()

         .context.total_tokens <= target:
            

        
         (.context.messages) > .preserve_recent:
            older = .context.messages[:-.preserve_recent]
            recent = .context.messages[-.preserve_recent:]

            
            summary = ._create_summary(older)
            summary_msg = Message(
                role=,
                content=,
                token_count=._counter.count(summary),
                metadata={: }
            )

            .context.messages = [summary_msg] + recent
            .context.compaction_count += 

        ._recalculate_total()

     () -> :
        
        
        key_points = []

         msg  messages:
             msg.role == :
                
                first_sentence = msg.content.split()[][:]
                key_points.append()
             msg.role ==   (key_points) < :
                
                   msg.content.lower()    msg.content.lower():
                    first_sentence = msg.content.split()[][:]
                    key_points.append()

         .join(key_points[:])

     ():
        
          ._counter:
            

        .context.system_prompt_tokens = ._counter.count(.context.system_prompt)
         msg  .context.messages:
            msg.token_count = ._counter.count(msg.content)
        ._recalculate_total()

     ():
        
        .context.total_tokens = (
            .context.system_prompt_tokens +
            (m.token_count  m  .context.messages)
        )

     () -> []:
        
        messages = []

         .context.system_prompt:
            messages.append({
                : ,
                : .context.system_prompt
            })

         msg  .context.messages:
            messages.append({
                : msg.role,
                : msg.content
            })

         messages

     () -> :
        
        .set_model(new_model)
         

Configuration

Inline Configuration Schema

from dataclasses import dataclass, field
from typing import Optional

@dataclass
class ServiceConfig:
    """Configuration for a single LLM service."""
    enabled: bool = True
    endpoint: str = ""
    priority: int = 1
    timeout: int = 30000
    max_retries: int = 3
    api_style: str = "openai"

@dataclass
class TaskRoutingConfig:
    """Configuration for task routing."""
    primary_models: list[str] = field(default_factory=list)
    fallback_models: list[str] = field(default_factory=list)
    min_context: int = 8192
    require_serena: bool = False

@dataclass
class SecurityConfig:
    """Security configuration for air-gapped networks."""
    allow_external: bool = False
    allowed_hosts: list[str] = field(default_factory=lambda: [
        "localhost", "127.0.0.1", "host.docker.internal"
    ])
    allowed_cidrs: list[] = field(default_factory=: [
        , , 
    ])
    audit_enabled:  = 
    audit_log_path:  = 
    log_queries:  = 
    log_responses:  =   
    verify_checksums:  = 


 :
    
    compaction_threshold:  = 
    compaction_target:  = 
    preserve_recent_messages:  = 
    preserve_recent_tool_calls:  = 
    max_tool_output_tokens:  = 


 :
    
    
    ollama: ServiceConfig = field(default_factory=: ServiceConfig(
        endpoint=,
        priority=
    ))
    lmstudio: ServiceConfig = field(default_factory=: ServiceConfig(
        endpoint=,
        priority=
    ))
    jan: ServiceConfig = field(default_factory=: ServiceConfig(
        endpoint=,
        priority=
    ))
    custom_endpoints: [] = field(default_factory=)

    
    coding: TaskRoutingConfig = field(default_factory=: TaskRoutingConfig(
        primary_models=[, , ],
        fallback_models=[, , ],
        min_context=
    ))
    reasoning: TaskRoutingConfig = field(default_factory=: TaskRoutingConfig(
        primary_models=[, , ],
        fallback_models=[, ],
        min_context=
    ))
    analysis: TaskRoutingConfig = field(default_factory=: TaskRoutingConfig(
        primary_models=[, ],
        fallback_models=[, ],
        min_context=,
        require_serena=
    ))
    documentation: TaskRoutingConfig = field(default_factory=: TaskRoutingConfig(
        primary_models=[, ],
        fallback_models=[, ],
        min_context=
    ))

    
    serena_enabled:  = 
    serena_priority:  = 

    
    context: ContextConfig = field(default_factory=ContextConfig)

    
    security: SecurityConfig = field(default_factory=SecurityConfig)


DEFAULT_CONFIG = RouterConfig()

 () -> RouterConfig:
    
    config = RouterConfig()

    
       data:
         service_name, service_data  data[].items():
             (config, service_name):
                (config, service_name, ServiceConfig(**service_data))

    
     category  [, , , ]:
         category  data.get(, {}):
            (config, category, TaskRoutingConfig(**data[][category]))

    
       data:
        config.security = SecurityConfig(**data[])

     config

Example YAML Configuration (for reference)

# local-llm-router.yaml
# Copy this to your project and customize

version: "1.0"
environment: "air-gapped"

services:
  ollama:
    enabled: true
    endpoint: "http://localhost:11434"
    priority: 1
    timeout: 30000

  lmstudio:
    enabled: true
    endpoint: "http://localhost:1234"
    priority: 2

  jan:
    enabled: false
    endpoint: "http://localhost:1337"
    priority: 3

  custom_endpoints:
    - name: "internal-gpu-server"
      endpoint: "http://192.168.1.100:8000"
      priority: 0
      api_style: "openai"

task_routing:
  coding:
    primary_models:
      - "deepseek-v3"
      - "qwen2.5-coder-32b"
      - "deepseek-coder-v2"
    fallback_models:
      - "codellama-34b"
      - 
       
     

  
    
       
       
       
    
       
       
     

  
    
       
       
     

  
    
       
       
    
       


   
   
   


   
   


   
  
     
     
     
   
   

Fallback Strategy

Graceful Degradation

from enum import IntEnum
from dataclasses import dataclass
from typing import Optional, Any

class FallbackLevel(IntEnum):
    PRIMARY = 0
    FALLBACK_MODELS = 1
    REDUCED_CONTEXT = 2
    SMALLEST_MODEL = 3
    FAILED = 4

@dataclass
class ExecutionResult:
    success: bool
    model: Optional[str] = None
    service: Optional[str] = None
    response: Any = None
    fallback_level: FallbackLevel = FallbackLevel.PRIMARY
    error: Optional[str] = None

class FallbackExecutor:
    """Execute queries with multi-level fallback."""

    def __init__(
        self,
        discovery: ServiceDiscovery,
        context_manager: ContextManager,
        config: RouterConfig
    ):
        self.discovery = discovery
        self.context = context_manager
        self.config = config

    async def execute_with_fallback(
        self,
        query: str,
        category: TaskCategory
    ) -> ExecutionResult:
        """Execute query with fallback strategy."""

        
        task_config = (.config, category.value)
        primary_models = task_config.primary_models
        fallback_models = task_config.fallback_models

        
         model  primary_models:
            result =  ._try_model(model, query)
             result.success:
                result.fallback_level = FallbackLevel.PRIMARY
                 result

        
         model  fallback_models:
            result =  ._try_model(model, query)
             result.success:
                result.fallback_level = FallbackLevel.FALLBACK_MODELS
                 result

        
        .context._compact(task_config.min_context)
         model  primary_models + fallback_models:
            result =  ._try_model(model, query)
             result.success:
                result.fallback_level = FallbackLevel.REDUCED_CONTEXT
                 result

        
        smallest =  ._find_smallest_model()
         smallest:
            result =  ._try_model(smallest, query)
             result.success:
                result.fallback_level = FallbackLevel.SMALLEST_MODEL
                 result

        
         ExecutionResult(
            success=,
            fallback_level=FallbackLevel.FAILED,
            error=
        )

      () -> ExecutionResult:
        
        
        service =  ._find_service_with_model(model_id)
          service:
             ExecutionResult(
                success=,
                error=
            )

        :
            response =  ._execute_on_service(service, model_id, query)
             ExecutionResult(
                success=,
                model=model_id,
                service=service.name,
                response=response
            )
         Exception  e:
             ExecutionResult(
                success=,
                error=(e)
            )

      () -> [LLMService]:
        
        services = (.discovery.services.values())

        
        services.sort(key= s: (.config, s., ServiceConfig()).priority)

         service  services:
             model  service.models:
                 model_id.lower()  model..lower()  model..lower()  model_id.lower():
                     service

         

      () -> []:
        
        smallest = 
        smallest_vram = ()

         service  .discovery.services.values():
             model  service.models:
                capability = MODEL_DATABASE.get(model.)
                 capability  capability.vram_gb < smallest_vram:
                    smallest = model.
                    smallest_vram = capability.vram_gb

         smallest

      () -> :
        
         httpx

        messages = .context.export_for_api()
        messages.append({: , : query})

          httpx.AsyncClient()  client:
             service.api_style ==   service. == :
                
                response =  client.post(
                    ,
                    json={
                        : model_id,
                        : messages,
                        : 
                    },
                    timeout=.config.ollama.timeout / 
                )
                data = response.json()
                 data.get(, {}).get(, )

            :
                
                response =  client.post(
                    ,
                    json={
                        : model_id,
                        : messages,
                        : 
                    },
                    timeout=
                )
                data = response.json()
                 data.get(, [{}])[].get(, {}).get(, )

Security (Air-Gapped)

Network Isolation

import hashlib
import json
from datetime import datetime
from dataclasses import dataclass
from typing import Optional
import ipaddress
import logging

@dataclass
class AuditLogEntry:
    timestamp: str
    event_type: str
    session_id: Optional[str] = None
    model: Optional[str] = None
    service: Optional[str] = None
    query_hash: Optional[str] = None  # Hashed, not plaintext
    tokens_in: int = 0
    tokens_out: int = 0
    success: bool = True
    error: Optional[str] = None

class SecurityModule:
    """Security enforcement for air-gapped networks."""

    def __init__(self, config: SecurityConfig):
        self.config = config
        self._allowed_ips = self._parse_allowed_networks()
        self._logger = self._setup_audit_logger()

     () -> :
        
        networks = []

         host  .config.allowed_hosts:
               host:
                
                networks.append(ipaddress.ip_network(host, strict=))
            :
                
                :
                    ip = ipaddress.ip_address(host)
                    networks.append(ipaddress.ip_network())
                 ValueError:
                    
                     host == :
                        networks.append(ipaddress.ip_network())
                     host == :
                        
                        networks.append(ipaddress.ip_network())

         cidr  .config.allowed_cidrs:
            networks.append(ipaddress.ip_network(cidr, strict=))

         networks

     () -> logging.Logger:
        
        logger = logging.getLogger()
        logger.setLevel(logging.INFO)

         .config.audit_enabled:
            handler = logging.FileHandler(.config.audit_log_path)
            handler.setFormatter(logging.Formatter())
            logger.addHandler(handler)

         logger

     () -> :
        
         .config.allow_external:
             

        :
             urllib.parse  urlparse
            parsed = urlparse(url)
            host = parsed.hostname

            
             host  [, , ]:
                 

            
            :
                ip = ipaddress.ip_address(host)
                 network  ._allowed_ips:
                     ip  network:
                         
             ValueError:
                
                 host  [, ]

             

         Exception:
             

     ():
        
          .config.audit_enabled:
            

        entry = AuditLogEntry(
            timestamp=datetime.now().isoformat(),
            event_type=,
            session_id=session_id,
            model=model,
            service=service,
            query_hash=._hash_content(query)  .config.log_queries  ,
            tokens_in=tokens_in,
            tokens_out=tokens_out,
            success=success,
            error=error
        )

        ._logger.info(json.dumps(entry.__dict__))

     ():
        
          .config.audit_enabled:
            

        entry = {
            : datetime.now().isoformat(),
            : ,
            **details
        }

        ._logger.warning(json.dumps(entry))

     () -> :
        
         hashlib.sha256(content.encode()).hexdigest()[:]


AIR_GAPPED_CHECKLIST = 

Coding Agent Detection

Detect Active Coding Agent

import os
import sys
from dataclasses import dataclass
from typing import Optional

@dataclass
class CodingAgentInfo:
    name: str
    type: str
    version: Optional[str] = None
    config_path: Optional[str] = None

# Environment variable markers for different agents
AGENT_ENV_MARKERS = {
    # CLI-based agents
    'QWEN_CLI_VERSION': ('qwen-cli', 'cli'),
    'OPENCODE_SESSION': ('opencode', 'cli'),
    'AIDER_SESSION': ('aider', 'cli'),
    'CODEX_SESSION': ('codex', 'cli'),
    'GEMINI_CLI_SESSION': ('gemini-cli', 'cli'),

    # IDE extensions
    'CONTINUE_SESSION': ('continue', 'ide'),
    'CLINE_SESSION': ('cline', 'ide'),
    'ROO_CODE_SESSION': ('roo-code', 'ide'),
    'CURSOR_SESSION': ('cursor', 'ide'),

    # Local GUI apps
    'OPENWEBUI_SESSION': (, ),
    : (, ),
    : (, ),

    
    : (, ),
}

 () -> CodingAgentInfo:
    

    
     env_var, (name, agent_type)  AGENT_ENV_MARKERS.items():
        value = os.environ.get(env_var)
         value:
             CodingAgentInfo(
                name=name,
                =agent_type,
                version=value  value !=   
            )

    
    :
         psutil
        parent = psutil.Process(os.getppid())
        parent_name = parent.name().lower()

        agent_process_names = {
            : ,
            : ,
            : ,
            : ,
            : ,
        }

         proc_name, agent_name  agent_process_names.items():
             proc_name  parent_name:
                 CodingAgentInfo(name=agent_name, =)

     ImportError:
          

    
     os.environ.get():
         CodingAgentInfo(
            name=os.environ.get(, ),
            =
        )

    
     CodingAgentInfo(name=, =)

 () -> :
    

    configs = {
        : {
            : ,
            : ,
        },
        : {
            : ,
            : ,
        },
        : {
            : ,
            : ,
        },
        : {
            : ,
            : ,
        },
    }

     configs.get(agent.name, {})

Complete Router Implementation

class LocalLLMRouter:
    """
    Complete Local LLM Router with Serena integration.

    Usage:
        router = LocalLLMRouter(workspace="/path/to/project")
        await router.initialize()

        response = await router.route("Implement a binary search function")
        print(response)
    """

    def __init__(
        self,
        workspace: str,
        config: RouterConfig = None,
        session_id: str = None
    ):
        self.workspace = workspace
        self.config = config or DEFAULT_CONFIG
        self.session_id = session_id or self._generate_session_id()

        # Components
        self.serena: Optional[SerenaMCP] = None
        self.discovery: Optional[ServiceDiscovery] = None
        self.context: Optional[ContextManager] = None
        self.security: Optional[SecurityModule] = None
        self.selector: Optional[ModelSelector] = None
        self.fallback: Optional[FallbackExecutor] = None

        # State
        self.os_info = detect_os()
        self.coding_agent = detect_coding_agent()
        self._initialized = False

    async  ():
        

        
        .security = SecurityModule(.config.security)

        
        .discovery = ServiceDiscovery(.config.custom_endpoints)
        services =  .discovery.discover_all()

          services:
             RuntimeError()

        
        all_models = []
         service  services:
            all_models.extend(m.  m  service.models)
        .selector = ModelSelector(all_models)

        
        .context = ContextManager(
            session_id=.session_id,
            system_prompt=._build_system_prompt(),
            compaction_threshold=.config.context.compaction_threshold,
            compaction_target=.config.context.compaction_target,
            preserve_recent=.config.context.preserve_recent_messages
        )

        
         .config.serena_enabled:
            .serena = SerenaMCP(.workspace)
            :
                 .serena.start()
             Exception  e:
                logging.warning()
                .serena = 

        
        .fallback = FallbackExecutor(
            .discovery,
            .context,
            .config
        )

        ._initialized = 

      () -> :
        
          ._initialized:
             .initialize()

        
        classification = classify_task(query)

        
        serena_context = {}
         .serena  (classification.requires_serena  file_context):
            serena_context =  ._gather_serena_context(
                query, file_context, classification
            )

        
        enriched_query = ._build_enriched_query(query, serena_context)

        
        model = .selector.select(
            classification.category,
            required_context=.context.context.total_tokens + (query) // 
        )

          model:
             RuntimeError()

        
        .context.set_model(model)

        
        model_capability = MODEL_DATABASE.get(model)
         model_capability:
            .context.check_and_compact(model_capability.context_window)

        
        result =  .fallback.execute_with_fallback(
            enriched_query,
            classification.category
        )

        
        .security.log_query(
            session_id=.session_id,
            model=result.model  model,
            service=result.service  ,
            query=query,
            tokens_in=(query) // ,
            tokens_out=(result.response  ) // ,
            success=result.success,
            error=result.error
        )

          result.success:
             RuntimeError()

        
        .context.add_message(, query)
        .context.add_message(, result.response)

        
         .serena  file_context  contains_code_edit(result.response):
             ._apply_serena_edits(result.response, file_context)

         result.response

      () -> :
        
        context = {}

          file_context:
             context

        file = file_context.get()
        position = file_context.get(, {})
        line = position.get(, )
        char = position.get(, )

        :
            
            context[] =  .serena.get_hover_info(file, line, char)

            
               query.lower()    query.lower():
                context[] =  .serena.get_references(file, line, char)

            
             classification.category == TaskCategory.ANALYSIS:
                context[] =  .serena.get_diagnostics(file)

         Exception  e:
            logging.warning()

         context

     () -> :
        
         build_enriched_query(query, serena_context)

      ():
        
        edits = parse_code_edits(response)
         edits:
             .serena.apply_edit(file_context[], edits)

     () -> :
        
         

     () -> :
        
         uuid
         (uuid.uuid4())[:]


 () -> :
    
    markers = [, , , , , , ]
     (marker  response  marker  markers)

 () -> :
    
    
     re
    code_blocks = re.findall(, response, re.DOTALL)
     [{: block.strip()}  block  code_blocks]

Resources

• Serena MCP: https://github.com/oraios/serena
• Serena Documentation: https://github.com/oraios/serena#user-guide
• Ollama API: https://github.com/ollama/ollama/blob/main/docs/api.md
• LM Studio: https://lmstudio.ai/docs/developer
• Jan AI: https://jan.ai/docs/desktop/api-server
• OpenWebUI: https://docs.openwebui.com/