chore: fix ruff lint findings, restore green CI lint gate

api/main.py

@@ -1,18 +1,15 @@
+import logging
 import os
 
 from fastapi import FastAPI, Request
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import JSONResponse
 
-import logging
-
-from api.routes import rag
 from api.auth import validate_auth_config
-
-from index.vector_store import VectorStore
+from api.routes import rag
 from index.graph_store import GraphStore
 from index.plugin_index import PluginIndex
-
+from index.vector_store import VectorStore
 from rag.control_plane import CONTROL_PLANE
 
 logger = logging.getLogger(__name__)
@@ -153,11 +150,10 @@ def status():
 @app.middleware("http")
 async def guard_requests(request: Request, call_next):
 
-    if request.url.path.startswith("/rag"):
-        if CONTROL_PLANE.status()["status"] != "READY":
-            return JSONResponse(
-                status_code=503,
-                content={"detail": "Control plane not ready"}
-            )
+    if request.url.path.startswith("/rag") and CONTROL_PLANE.status()["status"] != "READY":
+        return JSONResponse(
+            status_code=503,
+            content={"detail": "Control plane not ready"}
+        )
 
     return await call_next(request)

api/middleware/trace.py

@@ -1,10 +1,10 @@
-import uuid
 import time
-from typing import Dict
+import uuid
+
 
 class TraceContext:
     def __init__(self):
-        self.store: Dict[str, dict] = {}
+        self.store: dict[str, dict] = {}
 
     def start(self, query: str):
         trace_id = str(uuid.uuid4())

api/routes/rag.py

@@ -1,13 +1,13 @@
-from fastapi import APIRouter, Depends, HTTPException
-from fastapi.responses import StreamingResponse
-from pydantic import BaseModel
-from typing import Optional
 import json
 import os
 import traceback
 
-from rag.control_plane import CONTROL_PLANE
+from fastapi import APIRouter, Depends, HTTPException
+from fastapi.responses import StreamingResponse
+from pydantic import BaseModel
+
 from api.auth import require_auth
+from rag.control_plane import CONTROL_PLANE
 
 router = APIRouter()
 
@@ -24,8 +24,8 @@ def _debug_tracebacks_enabled() -> bool:
 
 class QueryRequest(BaseModel):
     query: str
-    repo: Optional[str] = None
-    bundle: Optional[str] = None
+    repo: str | None = None
+    bundle: str | None = None
 
 
 # =========================================================
@@ -45,8 +45,8 @@ def get_engine():
 
         return engine
 
-    except Exception as e:
-        raise RuntimeError(f"Engine access failed: {str(e)}")
+    except Exception as e:  # noqa: BLE001 -- boundary: any engine-access failure becomes a clean RuntimeError for the caller
+        raise RuntimeError(f"Engine access failed: {e!s}")
 
 
 # =========================================================
@@ -67,7 +67,7 @@ def query(req: QueryRequest, actor: str = Depends(require_auth)):
             "api_version": "v6"
         }
 
-    except Exception as e:
+    except Exception as e:  # noqa: BLE001 -- top-level HTTP boundary: must catch anything to return a clean 500 instead of an unhandled crash
         detail = {"error": str(e)}
         # Stack traces can leak file paths, internals, and other sensitive
         # detail into the HTTP response -- only include one when a deployer
@@ -106,7 +106,7 @@ def event_stream():
 
             yield f"data: {json.dumps({'type': 'done'})}\n\n"
 
-        except Exception as e:
+        except Exception as e:  # noqa: BLE001 -- SSE generator boundary: must catch anything to emit an error event instead of killing the stream
             yield f"data: {json.dumps({'type': 'error', 'message': str(e)})}\n\n"
 
     return StreamingResponse(

embeddings/embedder.py

@@ -1,7 +1,6 @@
-from sentence_transformers import SentenceTransformer
-from typing import List, Union
-import numpy as np
 
+import numpy as np
+from sentence_transformers import SentenceTransformer
 
 # =========================================================
 # EMBEDDER V4 - OMNIBIOAI
@@ -25,7 +24,7 @@ def __init__(self, model_name: str = "all-MiniLM-L6-v2"):
     # =========================================================
     # MAIN ENCODE FUNCTION
     # =========================================================
-    def encode(self, texts: Union[str, List[str]], batch_size: int = 32) -> List[List[float]]:
+    def encode(self, texts: str | list[str], batch_size: int = 32) -> list[list[float]]:
         """
         Convert text → embeddings
         """
@@ -67,7 +66,7 @@ def _normalize(self, embeddings: np.ndarray) -> np.ndarray:
     # =========================================================
     # UTILITY: SINGLE EMBEDDING
     # =========================================================
-    def encode_single(self, text: str) -> List[float]:
+    def encode_single(self, text: str) -> list[float]:
         """
         Convenience method for single query embedding
         """

index/graph_store.py

@@ -1,6 +1,5 @@
 from collections import defaultdict, deque
-from typing import Dict, List, Tuple, Set, Any
-
+from typing import Any
 
 # =========================================================
 # GRAPH STORE V4 - OMNIBIOAI
@@ -21,7 +20,7 @@ class GraphStore:
 
     def __init__(self):
         # adjacency list: node -> [(neighbor, relation)]
-        self.edges: Dict[str, List[Tuple[str, str]]] = defaultdict(list)
+        self.edges: dict[str, list[tuple[str, str]]] = defaultdict(list)
 
     # =========================================================
     # ADD EDGE
@@ -39,7 +38,7 @@ def add_edge(self, a: str, b: str, relation: str = "related"):
     # =========================================================
     # MAIN SEARCH ENTRY
     # =========================================================
-    def search(self, query: str, max_depth: int = 2) -> List[Dict[str, Any]]:
+    def search(self, query: str, max_depth: int = 2) -> list[dict[str, Any]]:
         """
         Semantic graph expansion search (V4)
 
@@ -52,8 +51,8 @@ def search(self, query: str, max_depth: int = 2) -> List[Dict[str, Any]]:
         if not query:
             return []
 
-        visited: Set[str] = set()
-        results: List[Dict[str, Any]] = []
+        visited: set[str] = set()
+        results: list[dict[str, Any]] = []
 
         seed_nodes = self._find_seed_nodes(query)
 
@@ -70,13 +69,13 @@ def search(self, query: str, max_depth: int = 2) -> List[Dict[str, Any]]:
     # =========================================================
     # SEED NODE DETECTION
     # =========================================================
-    def _find_seed_nodes(self, query: str) -> List[str]:
+    def _find_seed_nodes(self, query: str) -> list[str]:
 
         seeds = []
 
         q_tokens = set(query.split())
 
-        for node in self.edges.keys():
+        for node in self.edges:
 
             node_l = node.lower()
             node_tokens = set(node_l.split())
@@ -100,8 +99,8 @@ def _bfs_expand(
         start: str,
         query: str,
         max_depth: int,
-        visited: Set[str]
-    ) -> List[Dict[str, Any]]:
+        visited: set[str]
+    ) -> list[dict[str, Any]]:
 
         queue = deque([(start, 0)])
         results = []
@@ -153,7 +152,7 @@ def _score_match(self, query: str, a: str, b: str) -> float:
     # =========================================================
     # GRAPH STATS (FOR UI)
     # =========================================================
-    def size(self) -> Dict[str, int]:
+    def size(self) -> dict[str, int]:
 
         node_count = len(self.edges)
         edge_count = sum(len(v) for v in self.edges.values())
@@ -166,7 +165,7 @@ def size(self) -> Dict[str, int]:
     # =========================================================
     # DEBUG EXPORT (UI GRAPH VISUALIZATION READY)
     # =========================================================
-    def export(self) -> Dict[str, Any]:
+    def export(self) -> dict[str, Any]:
 
         nodes = list(self.edges.keys())
         edges = []

index/vector_store.py

@@ -1,8 +1,9 @@
+import logging
 import os
 import pickle
-import numpy as np
+
 import faiss
-import logging
+import numpy as np
 
 logger = logging.getLogger(__name__)
 
@@ -103,7 +104,7 @@ def search(self, query_vec, top_k: int = 5):
 
         return results
 
-    def filter_search(self, query_vec, top_k: int = 5, field: str = None, value: str = None):
+    def filter_search(self, query_vec, top_k: int = 5, field: str | None = None, value: str | None = None):
         """FAISS search with post-filtering on a metadata field.
 
         Retrieves top_k * 3 candidates from FAISS then keeps only those whose

ingestion/doc_loader.py

@@ -41,7 +41,7 @@ def load_documents(repo_paths):
                                 "text": text,
                                 "source": fpath
                             })
-                    except Exception as e:
+                    except Exception as e:  # noqa: BLE001 -- one unreadable file must not abort the whole indexing walk
                         print(f"⚠️  Could not read {fpath}: {e}")
     print(f"📄 Loaded {len(docs)} documents")
     return docs

processing/chunker.py

@@ -1,13 +1,12 @@
 import re
-from typing import List, Optional
 
 MAX_CHARS = 2000
 
 _FENCE_RE = re.compile(r'```.*?```', re.DOTALL)
 _HEADER_RE = re.compile(r'^(#{1,3})\s+.+$', re.MULTILINE)
 
 
-def _split_at_word_boundary(text: str, max_chars: int) -> List[str]:
+def _split_at_word_boundary(text: str, max_chars: int) -> list[str]:
     """Split text at word boundaries, keeping each piece ≤ max_chars."""
     if len(text) <= max_chars:
         return [text]
@@ -23,16 +22,16 @@ def _split_at_word_boundary(text: str, max_chars: int) -> List[str]:
     return chunks
 
 
-def _split_at_paragraphs(text: str, max_chars: int) -> List[str]:
+def _split_at_paragraphs(text: str, max_chars: int) -> list[str]:
     """Split at blank-line boundaries; fall back to word-boundary for oversize paragraphs."""
     if max_chars <= 0:
         return _split_at_word_boundary(text, MAX_CHARS) if text.strip() else []
     if len(text) <= max_chars:
         return [text] if text.strip() else []
 
     paragraphs = [p for p in re.split(r'\n\n+', text) if p.strip()]
-    result: List[str] = []
-    current_parts: List[str] = []
+    result: list[str] = []
+    current_parts: list[str] = []
     current_len = 0
 
     for para in paragraphs:
@@ -58,7 +57,7 @@ def _split_at_paragraphs(text: str, max_chars: int) -> List[str]:
     return result
 
 
-def chunk_text(text: str, chunk_size: int = 500) -> List[str]:
+def chunk_text(text: str, chunk_size: int = 500) -> list[str]:
     """
     Markdown-structure-aware chunker. chunk_size is kept for API compatibility.
 
@@ -73,7 +72,7 @@ def chunk_text(text: str, chunk_size: int = 500) -> List[str]:
         return []
 
     # Protect fenced code blocks — replace with non-splitting placeholders
-    fences: List[str] = []
+    fences: list[str] = []
 
     def _stash(m: re.Match) -> str:
         fences.append(m.group(0))
@@ -82,9 +81,9 @@ def _stash(m: re.Match) -> str:
     protected = _FENCE_RE.sub(_stash, text)
 
     # Split into sections: list of (header_line | None, body_str, header_level)
-    sections: List[tuple] = []
+    sections: list[tuple] = []
     last_end = 0
-    current_header: Optional[str] = None
+    current_header: str | None = None
     current_level = 0
 
     for m in _HEADER_RE.finditer(protected):
@@ -100,8 +99,8 @@ def _stash(m: re.Match) -> str:
         sections.append((current_header, tail, current_level))
 
     # Build chunks, maintaining a header breadcrumb stack
-    header_stack: List[str] = []
-    all_chunks: List[str] = []
+    header_stack: list[str] = []
+    all_chunks: list[str] = []
 
     for header_line, body, level in sections:
         if header_line is not None:
@@ -133,7 +132,7 @@ def _stash(m: re.Match) -> str:
             all_chunks.append(full)  # fallback: emit as-is
 
     # Restore fenced code blocks in every chunk
-    restored: List[str] = []
+    restored: list[str] = []
     for chunk in all_chunks:
         for i, fence in enumerate(fences):
             chunk = chunk.replace(f'\x00FENCE{i}\x00', fence)

rag/control_plane.py

@@ -1,8 +1,7 @@
-import time
 import threading
+import time
 from dataclasses import dataclass, field
-from typing import Optional, Dict, Any
-
+from typing import Any
 
 # =========================================================
 # CONTROL PLANE STATE
@@ -12,8 +11,8 @@
 class ControlPlaneState:
     status: str = "INIT"   # INIT | BUILDING | READY | FAILED
     started_at: float = field(default_factory=time.time)
-    last_error: Optional[str] = None
-    metrics: Dict[str, Any] = field(default_factory=dict)
+    last_error: str | None = None
+    metrics: dict[str, Any] = field(default_factory=dict)
 
 
 # =========================================================