chore(devops): refine entrypoint, configure offline mode, and fix test mock casing

.dockerignore

@@ -26,6 +26,7 @@ __pycache__/
 htmlcov/
 
 # Local output / temp
+output/
 output_test/
 *.tmp
 *.bak

.pre-commit-config.yaml

@@ -13,7 +13,7 @@ repos:
     hooks:
       - id: mypy
         name: mypy
-        entry: mypy
+        entry: python -m mypy
         language: system
         types: [python]
 

Dockerfile

@@ -30,12 +30,7 @@ RUN uv pip install llama-cpp-python \
 
 # Step 4: Pre-cache sentence-transformer model weights at build time
 # Prevents runtime download within the 10-minute container limit
-RUN python -c "
-from sentence_transformers import SentenceTransformer
-print('Pre-caching all-MiniLM-L6-v2...')
-SentenceTransformer('all-MiniLM-L6-v2')
-print('Model cached successfully.')
-"
+RUN python -c "from sentence_transformers import SentenceTransformer; print('Pre-caching...'); SentenceTransformer('all-MiniLM-L6-v2'); print('Model cached successfully.')"
 
 # Bundle GGUF model weights (~986 MB)
 # Make sure to run scripts/download_model.sh before building
@@ -61,6 +56,8 @@ ENV PYTHONUNBUFFERED=1 \
     LOCAL_MODEL_PATH=/app/models/qwen2.5-3b-instruct-q4_k_m.gguf \
     LOCAL_N_GPU_LAYERS=0 \
     LOCAL_N_THREADS=2 \
-    LOCAL_N_CTX=2048
+    LOCAL_N_CTX=2048 \
+    HF_HUB_OFFLINE=1 \
+    TRANSFORMERS_OFFLINE=1
 
-CMD ["python", "/app/main.py"]
+ENTRYPOINT ["/app/entrypoint.sh"]

agent/classifier.py

@@ -1,8 +1,8 @@
 """
 agent/classifier.py — Supervised Neural Network Classifier
 =========================================================
-Uses sentence-transformers (all-MiniLM-L6-v2) to encode prompts into dense 
-384-dimensional embeddings, then passes them through a PyTorch classification 
+Uses sentence-transformers (all-MiniLM-L6-v2) to encode prompts into dense
+384-dimensional embeddings, then passes them through a PyTorch classification
 head to predict the optimal routing category.
 
 Routes (constants unchanged):
@@ -12,8 +12,8 @@
 
 import logging
 import os
-import re
-from typing import Optional
+from typing import Any, Optional
+
 import torch
 import torch.nn as nn
 
@@ -43,25 +43,23 @@
 _LONG_CONTEXT_THRESHOLD = 6000  # chars; above this → API_LONG to avoid CPU OOM
 WEIGHTS_PATH = os.path.join(os.path.dirname(__file__), "supervised_model.pt")
 
+
 class LinearClassifier(nn.Module):
     def __init__(self, input_dim: int, num_classes: int):
         super().__init__()
         # Simple MLP head: 384 -> 64 -> 6 classes
-        self.net = nn.Sequential(
-            nn.Linear(input_dim, 64),
-            nn.ReLU(),
-            nn.Dropout(0.1),
-            nn.Linear(64, num_classes)
-        )
-
-    def forward(self, x):
+        self.net = nn.Sequential(nn.Linear(input_dim, 64), nn.ReLU(), nn.Dropout(0.1), nn.Linear(64, num_classes))
+
+    def forward(self, x: Any) -> Any:
         return self.net(x)
 
+
 class SemanticClassifier:
     """
     Loads all-MiniLM-L6-v2 and the trained PyTorch classification head.
     Performs fast, offline inference by passing the prompt embedding through the MLP.
     """
+
     _instance: Optional["SemanticClassifier"] = None
 
     @classmethod
@@ -75,12 +73,11 @@ def __init__(self) -> None:
         try:
             from sentence_transformers import SentenceTransformer
         except ImportError as exc:
-            raise ImportError("sentence-transformers is required. "
-                             "Run: pip install sentence-transformers") from exc
+            raise ImportError("sentence-transformers is required. " "Run: pip install sentence-transformers") from exc
 
         self.encoder = SentenceTransformer(_MODEL_NAME)
         self.model = LinearClassifier(input_dim=384, num_classes=len(ROUTES_MAP))
-        
+
         if os.path.exists(WEIGHTS_PATH):
             logger.info(f"Loading supervised weights from {WEIGHTS_PATH}")
             self.model.load_state_dict(torch.load(WEIGHTS_PATH, map_location="cpu"))
@@ -92,33 +89,34 @@ def __init__(self) -> None:
                 self._train_from_data(task_path)
             else:
                 logger.error(f"Supervised weights not found at {WEIGHTS_PATH} and no training data found. Using untrained weights.")
-                
+
         self.model.eval()
 
     def _train_from_data(self, data_path: str) -> None:
-        import torch.optim as optim
         import json
-
+
+        import torch.optim as optim
+
         with open(data_path, encoding="utf-8") as f:
             tasks = json.load(f)
-            
+
         prompts = [t["prompt"] for t in tasks]
         labels = [ROUTES_MAP.index(t["expected_route"]) for t in tasks]
-        
+
         embeddings = self.encoder.encode(prompts, convert_to_tensor=True).cpu()
         labels_tensor = torch.tensor(labels, dtype=torch.long)
-        
+
         criterion = nn.CrossEntropyLoss()
         optimizer = optim.AdamW(self.model.parameters(), lr=0.01, weight_decay=0.01)
-        
+
         self.model.train()
-        for epoch in range(150):
+        for _ in range(150):
             optimizer.zero_grad()
             outputs = self.model(embeddings)
             loss = criterion(outputs, labels_tensor)
             loss.backward()
             optimizer.step()
-            
+
         torch.save(self.model.state_dict(), WEIGHTS_PATH)
         logger.info(f"Successfully trained and saved new classifier weights to {WEIGHTS_PATH}!")
 
@@ -129,6 +127,7 @@ def classify(self, prompt: str) -> str:
             pred_idx = int(logits.argmax(dim=1).item())
         return ROUTES_MAP[pred_idx]
 
+
 # ---------------------------------------------------------------------------
 # Public API — drop-in replacement for the old classify()
 # ---------------------------------------------------------------------------
@@ -142,7 +141,7 @@ def classify(prompt: str) -> str:
         return ROUTE_API_LONG
 
     classifier = SemanticClassifier.get_instance()
-    
+
     # --- Execute Supervised PyTorch Classification ---
     route = classifier.classify(prompt)
     logger.debug("Classifier: supervised PyTorch → %s", route)

engines/remote_llm.py

@@ -65,7 +65,7 @@ def __contains__(self, item: object) -> bool:
 _OUTPUT_SUFFIX: dict[str, str] = {
     "API_CODE": " Return ONLY raw code. No markdown, no explanation.",
     "API_MATH": "",  # Handled natively in CoT system prompt
-    "API_LOGIC": "", # Handled natively in Direct system prompt
+    "API_LOGIC": "",  # Handled natively in Direct system prompt
     "API_LONG_CONTEXT": " Summarize in 3 sentences max.",
 }
 

handlers/code_utils.py

@@ -10,6 +10,7 @@
 import ast
 import logging
 import re
+from typing import Any
 
 logger = logging.getLogger(__name__)
 
@@ -29,11 +30,26 @@ def classify_code_difficulty(prompt: str) -> str:
     length = len(prompt)
 
     hard_kw = (
-        "dynamic programming", "dijkstra", "graph", "tree", "trie",
-        "heap", "linked list", "lru", "backtracking", "topological",
-        "n-queens", "knapsack", "memoization", "bfs", "dfs",
-        "binary search tree", "cycle detection", "floyd",
-        "strongly connected", "minimax",
+        "dynamic programming",
+        "dijkstra",
+        "graph",
+        "tree",
+        "trie",
+        "heap",
+        "linked list",
+        "lru",
+        "backtracking",
+        "topological",
+        "n-queens",
+        "knapsack",
+        "memoization",
+        "bfs",
+        "dfs",
+        "binary search tree",
+        "cycle detection",
+        "floyd",
+        "strongly connected",
+        "minimax",
     )
     if any(kw in low for kw in hard_kw):
         return HARD
@@ -58,22 +74,19 @@ def extract_code(text: str) -> str:
     text = text.strip()
     matches = _FENCE_RE.findall(text)
     if matches:
-        return max(matches, key=len).strip()
+        return str(max(matches, key=len)).strip()
 
     # No fences — strip leading prose until we hit a code-like line.
     lines = text.splitlines()
     for i, line in enumerate(lines):
         s = line.strip()
-        if s and (
-            s.startswith(("def ", "class ", "import ", "from ", "if __", "#"))
-            or re.match(r"^[a-zA-Z_]\w*\s*=", s)
-        ):
+        if s and (s.startswith(("def ", "class ", "import ", "from ", "if __", "#")) or re.match(r"^[a-zA-Z_]\w*\s*=", s)):
             return "\n".join(lines[i:]).strip()
 
     return text
 
 
-def validate_code(code: str) -> dict:
+def validate_code(code: str) -> dict[str, Any]:
     """Run syntax + completeness checks. Returns {is_valid, reason}."""
     if not code or not code.strip():
         return {"is_valid": False, "reason": "empty"}

handlers/logic.py

@@ -11,30 +11,30 @@
 def extract_logic(text: str) -> str:
     """Cleans up the final logic answer based on strict format."""
     text = text.strip()
-    m = re.search(r'ANSWER:\s*(.+?)(?:\n|$)', text, re.IGNORECASE)
+    m = re.search(r"ANSWER:\s*(.+?)(?:\n|$)", text, re.IGNORECASE)
     if m:
-        return m.group(1).strip().rstrip('.')
+        return m.group(1).strip().rstrip(".")
     if len(text.split()) <= 3:
         return text
-        
-    bold = re.findall(r'\*\*([^*]+?)\*\*', text)
+
+    bold = re.findall(r"\*\*([^*]+?)\*\*", text)
     if bold:
-        return bold[-1].strip()
-        
-    lines = text.strip().split('\n')
+        return str(bold[-1]).strip()
+
+    lines = text.strip().split("\n")
     last = lines[-1].strip()
-    if re.match(r'^(yes|no)\.?$', last, re.IGNORECASE):
-        return last.rstrip('.')
-        
-    if re.search(r'\byes\b', text, re.I) and not re.search(r'\bno\b', text, re.I):
+    if re.match(r"^(yes|no)\.?$", last, re.IGNORECASE):
+        return last.rstrip(".")
+
+    if re.search(r"\byes\b", text, re.I) and not re.search(r"\bno\b", text, re.I):
         return "Yes"
-    if re.search(r'\bno\b', text, re.I) and not re.search(r'\byes\b', text, re.I):
+    if re.search(r"\bno\b", text, re.I) and not re.search(r"\byes\b", text, re.I):
         return "No"
-        
+
     ans = re.findall(r'(?:answer|conclusion|therefore)\s*(?:is|:)\s*["\']?(.+?)(?:["\']?\s*(?:\.|$|!))', text, re.I)
     if ans and len(ans[-1].split()) <= 6:
-        return ans[-1].strip()
-        
+        return str(ans[-1]).strip()
+
     return last if len(last.split()) <= 6 else text[:60]
 
 

handlers/math_handler.py

@@ -11,26 +11,26 @@
 def extract_number(text: str) -> str:
     """Extracts numeric answer from a verbose Chain-of-Thought response."""
     text = text.strip()
-    if re.match(r'^-?\d+(?:/\d+)?(?:\.\d+)?$', text):
+    if re.match(r"^-?\d+(?:/\d+)?(?:\.\d+)?$", text):
         return text
-    
+
     # Try different pattern fallbacks
     patterns = [
-        r'ANSWER:\s*\$?\s*(-?\d+(?:/\d+)?(?:\.\d+)?)',
-        r'(?:answer|result|total|sum|area|speed|price|cost|value|average|remain|left|volume|profit|capacity|units|pieces|girls|seconds|original|radius)\s*(?:is|=|:)\s*\$?\s*(-?\d+(?:/\d+)?(?:\.\d+)?)',
-        r'\*\*(-?\d+(?:/\d+)?(?:\.\d+)?)\*\*', 
-        r'\\boxed\{(-?\d+(?:/\d+)?(?:\.\d+)?)\}', 
-        r'`(-?\d+(?:/\d+)?(?:\.\d+)?)`'
+        r"ANSWER:\s*\$?\s*(-?\d+(?:/\d+)?(?:\.\d+)?)",
+        r"(?:answer|result|total|sum|area|speed|price|cost|value|average|remain|left|volume|profit|capacity|units|pieces|girls|seconds|original|radius)\s*(?:is|=|:)\s*\$?\s*(-?\d+(?:/\d+)?(?:\.\d+)?)",
+        r"\*\*(-?\d+(?:/\d+)?(?:\.\d+)?)\*\*",
+        r"\\boxed\{(-?\d+(?:/\d+)?(?:\.\d+)?)\}",
+        r"`(-?\d+(?:/\d+)?(?:\.\d+)?)`",
     ]
     for p in patterns:
         m = re.findall(p, text, re.IGNORECASE)
         if m:
             val = m[-1] if isinstance(m[-1], str) else [x for x in m[-1] if x][-1]
-            return val
-            
+            return str(val)
+
     # Final fallback: last number in text
-    nums = re.findall(r'-?\d+(?:/\d+)?(?:\.\d+)?', text)
-    return nums[-1] if nums else text
+    nums = re.findall(r"-?\d+(?:/\d+)?(?:\.\d+)?", text)
+    return str(nums[-1]) if nums else text
 
 
 class MathHandler: