optimization 404

.dockerignore

@@ -0,0 +1,54 @@
+# Git
+.git
+.gitignore
+*.md
+
+# Secrets (never ship)
+.env
+.env.local
+.env.*
+!.env.example
+google_oauth_test_creds.postman_environment.json
+
+# Python / build
+__pycache__/
+*.py[cod]
+*.egg-info/
+.venv/
+venv/
+.pytest_cache/
+.ruff_cache/
+htmlcov/
+.tox/
+.nox/
+build/
+dist/
+
+# Local / tooling artifacts
+.mimocode
+.devenv/
+
+# Runtime data
+data/
+persistence/
+logs/
+*.log
+*.db
+*.sqlite3
+
+# Large repo blobs not needed by the image
+generated-python-sdk/
+ddl/
+searxng/
+scripts/
+*.json
+
+# NOTE: `lua/` (Redis Lua scripts) is loaded at runtime and MUST stay in the image.
+
+# Tests / deploy scripts (kept out of the image)
+tests/
+test_integration.py
+local_deploy.py
+deploy_sdk.py
+deploy_hf.py
+server.log

.gitignore

@@ -81,7 +81,7 @@ celerybeat.pid
 
 *.sage.py
 
-.env
+.env*
 .venv
 env/
 venv/

app/api/deps.py

@@ -4,44 +4,28 @@ from typing import Any, Dict, Optional, Tuple
 
 from fastapi import Request
 
-from app.services.auth_service import AuthService
 from app.services.converter_service import ConverterService
-from app.services.database_service import DatabaseService
 from app.services.embeddings_service import EmbeddingService
 from app.services.extraction_service import ExtractionService
-from app.services.ocr_service import OCRService
-from app.services.sql_validator_service import SqlValidatorService
 from app.services.text_cleaner_service import TextCleanerService
 from app.services.scheduler_service import SchedulerService
 from app.services.vector_store_service import VectorStoreService
 
-
-def get_sql_validator_service() -> SqlValidatorService:
-    return SqlValidatorService()
+_converter_service = ConverterService()
+_extraction_service = ExtractionService()
+_text_cleaner_service = TextCleanerService()
 
 
 def get_text_cleaner_service() -> TextCleanerService:
-    return TextCleanerService()
-
-
-def get_auth_service() -> AuthService:
-    return AuthService()
+    return _text_cleaner_service
 
 
 def get_converter_service() -> ConverterService:
-    return ConverterService()
-
-
-def get_ocr_service() -> OCRService:
-    return OCRService()
+    return _converter_service
 
 
 def get_extraction_service() -> ExtractionService:
-    return ExtractionService()
-
-
-def get_database_service() -> DatabaseService:
-    return DatabaseService()
+    return _extraction_service
 
 
 def get_embeddings_service() -> EmbeddingService:

app/api/server.py

@@ -3,9 +3,13 @@ from __future__ import annotations
 import asyncio
 from contextlib import asynccontextmanager
 
-from fastapi import FastAPI, Request
+from fastapi import Depends, FastAPI, Request
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.middleware.gzip import GZipMiddleware
+from fastapi.responses import JSONResponse
+from slowapi import Limiter
+from slowapi.errors import RateLimitExceeded
+from slowapi.util import get_remote_address
 
 from app.api.v1.router import api_v1_router
 from app.api.v1.system import is_maintenance
@@ -18,6 +22,7 @@ from app.core.scripts import load_scripts
 from app.services.embeddings_service import EmbeddingService
 from app.services.scheduler_service import SchedulerService
 from app.services.vector_store_service import VectorStoreService
+from app.utils.http_utils import SharedAsyncClient
 
 _logger = get_logger(__name__)
 _settings = get_settings()
@@ -40,19 +45,34 @@ def _is_public_path(path: str) -> bool:
 _embedding_service: EmbeddingService = EmbeddingService()
 _vector_store_service: VectorStoreService = VectorStoreService(_embedding_service)
 _scheduler_service: SchedulerService = SchedulerService()
+_ping_client = SharedAsyncClient(timeout=30.0)
+
+
+# Global per-client-IP rate limit applied to every /api/v1 route.
+# Configure via RATE_LIMIT_PER_MINUTE (0 or empty disables).
+limiter = Limiter(key_func=get_remote_address, storage_uri="memory://", headers_enabled=False)
+
+
+async def _api_rate_limit(request: Request) -> None:
+    """Apply the global per-client-IP rate limit to all /api/v1 routes."""
+    return None
+
+
+_RATE_LIMIT_PER_MINUTE = _settings.rate_limit_per_minute
+if _RATE_LIMIT_PER_MINUTE and _RATE_LIMIT_PER_MINUTE > 0:
+    _api_rate_limit = limiter.limit(f"{_RATE_LIMIT_PER_MINUTE}/minute")(_api_rate_limit)
 
 
 async def _self_ping():
-    import httpx
     health_url = _settings.self_ping_url
     while True:
         try:
-            async with httpx.AsyncClient(timeout=30.0) as client:
-                response = await client.get(health_url)
-                if response.status_code == 200:
-                    _logger.info("Self-ping successful: %s", health_url)
-                else:
-                    _logger.warning("Self-ping returned: %s - %s", health_url, response.status_code)
+            client = await _ping_client.get()
+            response = await client.get(health_url)
+            if response.status_code == 200:
+                _logger.info("Self-ping successful: %s", health_url)
+            else:
+                _logger.warning("Self-ping returned: %s - %s", health_url, response.status_code)
         except Exception as exc:
             _logger.error("Self-ping error: %s", exc)
         await asyncio.sleep(900)
@@ -127,6 +147,7 @@ async def lifespan(app: FastAPI):
     await close_storage_service()
     from app.utils.http_utils import close_shared_aiohttp_sessions
     await close_shared_aiohttp_sessions()
+    await _ping_client.close()
     from app.services.supabase import get_supabase_client
     client = get_supabase_client()
     if client:
@@ -162,6 +183,17 @@ def create_application() -> FastAPI:
         allow_headers=["*"],
     )
 
+    app.state.limiter = limiter
+
+    @app.exception_handler(RateLimitExceeded)
+    async def _rate_limit_exceeded_handler(request: Request, exc: RateLimitExceeded):
+        retry_after = getattr(exc, "retry_after", 60) or 60
+        return JSONResponse(
+            status_code=429,
+            content={"success": False, "detail": "Rate limit exceeded. Please retry later."},
+            headers={"Retry-After": str(int(retry_after))},
+        )
+
     @app.middleware("http")
     async def maintenance_middleware(request: Request, call_next):
         if is_maintenance():
@@ -202,7 +234,11 @@ def create_application() -> FastAPI:
                 )
         return await call_next(request)
 
-    app.include_router(api_v1_router, prefix="/api/v1")
+    app.include_router(
+        api_v1_router,
+        prefix="/api/v1",
+        dependencies=[Depends(_api_rate_limit)],
+    )
 
     @app.get("/", include_in_schema=False)
     async def root(request: Request):

app/api/v1/batch.py

@@ -6,7 +6,6 @@ import time
 from typing import Annotated, List, Optional
 from urllib.parse import urlparse
 
-import httpx
 from fastapi import APIRouter, Depends, File, Form, HTTPException, Query, UploadFile
 
 from app.api.deps import (
@@ -26,6 +25,7 @@ from app.models.schemas import (
 from app.services.converter_service import ConverterService
 from app.services.extraction_service import ExtractionService
 from app.services.text_cleaner_service import TextCleanerService
+from app.utils.http_utils import DownloadError, download_url
 
 router = APIRouter()
 _logger = get_logger(__name__)
@@ -156,15 +156,9 @@ async def batch_urls(
 
         if body.return_json:
             try:
-                async with httpx.AsyncClient(timeout=30.0, follow_redirects=True) as client:
-                    resp = await client.get(url)
-                    resp.raise_for_status()
-                raw_data = resp.content
-                if len(raw_data) > _MAX_UPLOAD_BYTES:
-                    return BatchFileResult(
-                        filename=filename, success=False, time_ms=0,
-                        error=f"File exceeds {_settings.max_upload_mb} MB limit.",
-                    )
+                raw_data, _ = await download_url(
+                    url, timeout_seconds=30.0, max_size_bytes=_MAX_UPLOAD_BYTES
+                )
                 loop = asyncio.get_running_loop()
                 outcome = await loop.run_in_executor(_thread_pool, converter_service.convert_stream, raw_data, filename)
                 if isinstance(outcome, ConversionError):
@@ -185,10 +179,13 @@ async def batch_urls(
                 result.json_content = json_result if "error" not in json_result else None
                 result.error = json_result.get("error") if "error" in json_result else None
                 return result
-            except httpx.HTTPError as exc:
+            except DownloadError as exc:
+                if exc.is_size_error:
+                    error = f"File exceeds {_settings.max_upload_mb} MB limit."
+                else:
+                    error = f"Failed to fetch URL: {exc}"
                 return BatchFileResult(
-                    filename=filename, success=False, time_ms=0,
-                    error=f"Failed to fetch URL: {exc}",
+                    filename=filename, success=False, time_ms=0, error=error,
                 )
 
         loop = asyncio.get_running_loop()

app/api/v1/chat.py

@@ -5,7 +5,7 @@ import logging
 import time
 from typing import Any, AsyncGenerator, Dict, List, Optional
 
-from fastapi import APIRouter, Depends, HTTPException, Request
+from fastapi import APIRouter, HTTPException, Request
 from fastapi.responses import StreamingResponse
 
 from app.api.deps import get_redis_scripts

app/api/v1/convert.py

@@ -6,7 +6,6 @@ import json as json_mod
 from typing import Annotated, Any, Dict, Optional
 from urllib.parse import urlparse
 
-import httpx
 from fastapi import (
     APIRouter,
     Depends,
@@ -32,6 +31,7 @@ from app.models.schemas import ConversionMetadata, ConversionResponse, UrlReques
 from app.services.converter_service import ConverterService
 from app.services.extraction_service import ExtractionService
 from app.services.text_cleaner_service import TextCleanerService
+from app.utils.http_utils import DownloadError, download_url
 
 router = APIRouter()
 _logger = get_logger(__name__)
@@ -225,12 +225,9 @@ async def convert_url(
     if body.return_json:
         try:
             _logger.info("Fetching URL for JSON extraction: %s", body.url)
-            async with httpx.AsyncClient(timeout=30.0, follow_redirects=True) as client:
-                resp = await client.get(body.url)
-                resp.raise_for_status()
-            raw_data = resp.content
-            if len(raw_data) > _MAX_UPLOAD_BYTES:
-                raise HTTPException(status_code=413, detail={"success": False, "message": f"File exceeds {_settings.max_upload_mb} MB limit."})
+            raw_data, _ = await download_url(
+                body.url, timeout_seconds=30.0, max_size_bytes=_MAX_UPLOAD_BYTES
+            )
 
             loop = asyncio.get_running_loop()
             outcome = await loop.run_in_executor(_thread_pool, converter_service.convert_stream, raw_data, filename)
@@ -253,9 +250,11 @@ async def convert_url(
             )
             _logger.info("Request completed for URL %s", body.url)
             return response
-        except httpx.HTTPError as exc:
+        except DownloadError as exc:
             _logger.error("Failed to fetch URL %s: %s", body.url, exc)
-            raise HTTPException(status_code=400, detail={"success": False, "message": f"Failed to fetch URL: {exc}"})
+            status_code = 413 if exc.is_size_error else 400
+            message = f"File exceeds {_settings.max_upload_mb} MB limit." if exc.is_size_error else f"Failed to fetch URL: {exc}"
+            raise HTTPException(status_code=status_code, detail={"success": False, "message": message})
 
     loop = asyncio.get_running_loop()
     outcome = await loop.run_in_executor(_thread_pool, converter_service.convert_url, body.url)

app/api/v1/json_extract.py

@@ -7,6 +7,7 @@ from fastapi import APIRouter, HTTPException
 from pydantic import BaseModel, Field
 
 from app.core.logger import get_logger
+from app.core.thread_pool import run_in_executor
 from app.services.json_service import extract_json
 
 logger = get_logger(__name__)
@@ -75,7 +76,7 @@ async def extract_json_endpoint(
 
     effective_limit = 1 if body.mode == "first" else body.limit
 
-    result = extract_json(body.content, limit=effective_limit)
+    result = await run_in_executor(extract_json, body.content, limit=effective_limit)
 
     elapsed = round((time.perf_counter() - start) * 1000, 3)
 

app/api/v1/reconcile.py

@@ -13,6 +13,7 @@ from pydantic import BaseModel
 
 from app.config import get_settings
 from app.core.logger import get_logger
+from app.core.thread_pool import run_in_executor
 from app.services.reconciliation_service import (
     SUPPORTED_EXTENSIONS,
     analyze_duplicates,
@@ -144,17 +145,14 @@ def _parse_column_mapping_model(mappings: Optional[List[ColumnMapping]]) -> Opti
     return {m.destination: m.source for m in mappings}
 
 
-async def _process_file_pair(
+def _process_pair_sync(
     source_file: bytes,
     dest_file: bytes,
     source_ext: str,
     dest_ext: str,
-    column_mapping: Optional[Dict[str, str]] = None,
-    pair_index: int = 0,
+    column_mapping: Optional[Dict[str, str]],
+    pair_index: int,
 ) -> Dict[str, Any]:
-    if source_ext != dest_ext:
-        return _failed_result(pair_index, f"File type mismatch. Source: {source_ext}, Destination: {dest_ext}")
-
     try:
         df_src = read_to_dataframe(source_file, source_ext, f"pair-{pair_index}")
         df_dst = read_to_dataframe(dest_file, dest_ext, f"pair-{pair_index}")
@@ -201,6 +199,28 @@ async def _process_file_pair(
         return _failed_result(pair_index, str(e))
 
 
+async def _process_file_pair(
+    source_file: bytes,
+    dest_file: bytes,
+    source_ext: str,
+    dest_ext: str,
+    column_mapping: Optional[Dict[str, str]] = None,
+    pair_index: int = 0,
+) -> Dict[str, Any]:
+    if source_ext != dest_ext:
+        return _failed_result(pair_index, f"File type mismatch. Source: {source_ext}, Destination: {dest_ext}")
+
+    return await run_in_executor(
+        _process_pair_sync,
+        source_file,
+        dest_file,
+        source_ext,
+        dest_ext,
+        column_mapping,
+        pair_index,
+    )
+
+
 @router.post(
     "/reconcile/files",
     response_model=ReconciliationResponse,

app/api/v1/semantic_router.py

@@ -5,6 +5,7 @@ import time
 from fastapi import APIRouter, Depends
 
 from app.api.deps import get_embeddings_service
+from app.core.thread_pool import run_in_executor
 from app.models.schemas import SemanticRouterRequest, SemanticRouterResponse
 from app.services.embeddings_service import EmbeddingService
 from app.services.semantic_router_service import SemanticRouterService
@@ -24,7 +25,7 @@ async def route_query(
     start = time.perf_counter()
     svc = SemanticRouterService(embedding_service)
     routes_dict = [r.model_dump() for r in body.routes]
-    result = svc.route(body.query, routes_dict, body.threshold)
+    result = await run_in_executor(svc.route, body.query, routes_dict, body.threshold)
     elapsed_ms = round((time.perf_counter() - start) * 1000, 3)
     return SemanticRouterResponse(
         success=result["success"],

app/api/v1/sql_validator.py

@@ -4,6 +4,7 @@ import time
 
 from fastapi import APIRouter
 
+from app.core.thread_pool import run_in_executor
 from app.models.schemas import SqlValidationRequest, SqlValidationResponse
 from app.services.sql_validator_service import SqlValidatorService
 
@@ -20,7 +21,7 @@ async def validate_sql(
     body: SqlValidationRequest,
 ):
     start = time.perf_counter()
-    result = _service.validate(body.query, body.dialect)
+    result = await run_in_executor(_service.validate, body.query, body.dialect)
     elapsed_ms = round((time.perf_counter() - start) * 1000, 3)
     return SqlValidationResponse(
         success=True,

app/api/v1/system.py

@@ -24,6 +24,7 @@ from app.core.constants import (
     WEB_EXTENSIONS,
 )
 from app.core.logger import get_logger
+from app.core.thread_pool import run_in_executor
 from app.models.schemas import (
     HealthResponse,
     InfoResponse,
@@ -136,6 +137,16 @@ def _build_archive(data_dir: Path) -> io.BytesIO:
     return buf
 
 
+def _extract_archive(content: bytes, data_dir: Path) -> int:
+    restored = 0
+    with tarfile.open(fileobj=io.BytesIO(content), mode="r:gz") as tar:
+        for member in tar.getmembers():
+            if member.isfile():
+                tar.extract(member, path=data_dir.parent)
+                restored += 1
+    return restored
+
+
 def _get_data_dir() -> Path:
     raw = _settings.data_dir
     p = Path(raw)
@@ -150,7 +161,7 @@ async def download_backup():
     if not data_dir.is_dir():
         raise HTTPException(status_code=404, detail="Data directory not found")
     try:
-        archive = _build_archive(data_dir)
+        archive = await run_in_executor(_build_archive, data_dir)
     except Exception as exc:
         _logger.error("Backup creation failed: %s", exc)
         raise HTTPException(status_code=500, detail=f"Backup failed: {exc}")
@@ -179,11 +190,7 @@ async def upload_and_restore(
 
     restored = 0
     try:
-        with tarfile.open(fileobj=io.BytesIO(content), mode="r:gz") as tar:
-            for member in tar.getmembers():
-                if member.isfile():
-                    tar.extract(member, path=data_dir.parent)
-                    restored += 1
+        restored = await run_in_executor(_extract_archive, content, data_dir)
     except tarfile.TarError as exc:
         raise HTTPException(status_code=400, detail=f"Invalid archive: {exc}")
 

app/api/v1/token_counter.py

@@ -4,6 +4,7 @@ import time
 
 from fastapi import APIRouter
 
+from app.core.thread_pool import run_in_executor
 from app.models.domain import count_tokens
 from app.models.schemas import TokenCountRequest, TokenCountResponse
 
@@ -20,7 +21,7 @@ async def count_text_tokens(
 ) -> TokenCountResponse:
     start = time.perf_counter()
     try:
-        token_count = count_tokens(body.text, body.encoding)
+        token_count = await run_in_executor(count_tokens, body.text, body.encoding)
         elapsed_ms = round((time.perf_counter() - start) * 1000, 3)
         return TokenCountResponse(
             success=True,

app/api/v1/vector_stores.py

@@ -3,7 +3,6 @@ from __future__ import annotations
 import asyncio
 import time
 
-import httpx
 from fastapi import (
     APIRouter,
     Depends,
@@ -33,6 +32,7 @@ from app.models.schemas import (
 from app.services.converter_service import ConverterService
 from app.services.text_cleaner_service import TextCleanerService
 from app.services.vector_store_service import VectorStoreService
+from app.utils.http_utils import DownloadError, download_url
 
 router = APIRouter()
 logger = get_logger(__name__)
@@ -285,20 +285,8 @@ async def ingest_pdf_url(
         raise HTTPException(status_code=404, detail=f"Vector store {store_id} not found")
 
     try:
-        async with httpx.AsyncClient(timeout=30.0, follow_redirects=True) as client:
-            resp = await client.get(body.url)
-            resp.raise_for_status()
-            raw = resp.content
-    except httpx.HTTPStatusError as exc:
-        return DocumentIngestResponse(
-            success=False,
-            vector_store_id=store_id,
-            doc_id=body.doc_id,
-            chunks_ingested=0,
-            time_ms=0,
-            error=f"Failed to fetch PDF from URL: HTTP {exc.response.status_code}",
-        )
-    except httpx.RequestError as exc:
+        raw, _ = await download_url(body.url, timeout_seconds=30.0)
+    except DownloadError as exc:
         return DocumentIngestResponse(
             success=False,
             vector_store_id=store_id,

app/api/verify.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 from fastapi import APIRouter, HTTPException, Query
 
+from app.core.thread_pool import run_in_executor
 from app.services.verify_service import VerifyService
 
 router = APIRouter(tags=["Verify"])
@@ -13,7 +14,7 @@ async def verify_phone(
     number: str = Query(..., description="Phone number (E.164 format like +14155552671, or local with country_code)"),
     country_code: str | None = Query(None, description="ISO 3166-1 alpha-2 country code (e.g. US, IN, GB)"),
 ):
-    result = _service.verify_phone(number, country_code)
+    result = await run_in_executor(_service.verify_phone, number, country_code)
     if not result.valid:
         raise HTTPException(status_code=400, detail=result.error)
     return result.dict()

app/config.py

@@ -16,6 +16,9 @@ class Settings(BaseSettings):
 
     app_name: str = "All API Collection"
     app_version: str = "1.0.0"
+
+    # Global per-client-IP rate limit (requests / minute). Set to 0 to disable.
+    rate_limit_per_minute: int = 60
     environment: str = "production"
     host: str = "0.0.0.0"
     port: int = 7860
@@ -197,6 +200,29 @@ DEFAULT_TOP_P = 0.9
 DEFAULT_STREAM = False
 
 
+_INSECURE_SECRETS = frozenset({"changeme", "changeme-jwt-secret", "dev-secret-change-me"})
+
+_PRODUCTION_ENVIRONMENTS = frozenset({"production", "prod"})
+
+_REQUIRED_SECRETS = {
+    "api_key": "API_KEY",
+    "jwt_secret_key": "JWT_SECRET_KEY",
+}
+
+
 @lru_cache(maxsize=1)
 def get_settings() -> Settings:
-    return Settings()
+    settings = Settings()
+    if settings.environment.lower() in _PRODUCTION_ENVIRONMENTS:
+        missing = [
+            env_name
+            for field, env_name in _REQUIRED_SECRETS.items()
+            if not getattr(settings, field) or getattr(settings, field) in _INSECURE_SECRETS
+        ]
+        if missing:
+            raise RuntimeError(
+                "Refusing to start in production with insecure/default secrets: "
+                + ", ".join(missing)
+                + ". Set strong values in your environment (see .env.example)."
+            )
+    return settings

app/core/auth/deps.py

@@ -1,18 +1,16 @@
 from __future__ import annotations
 
 import logging
-from datetime import datetime
-from typing import Annotated, List, Optional
+from typing import Annotated, List
 
 import jwt
 from argon2 import PasswordHasher
 from fastapi import Depends, HTTPException, Request, status
 
 from app.config import get_settings
-from app.core.auth.models import RefreshSession, User, _uuid
+from app.core.auth.models import User, _uuid
 from app.core.auth.models import _utcnow as _now
 from app.services.supabase import AuthRepository, SupabaseClient, get_supabase_client
-from app.services.supabase.repositories import _row_to_user
 from app.services.supabase.client import create_supabase_client
 
 logger = logging.getLogger("auth")

app/core/banner.py

@@ -1,26 +0,0 @@
-from __future__ import annotations
-
-import pyfiglet
-
-from app.config import get_settings
-
-
-def get_banner() -> str:
-    settings = get_settings()
-    banner = pyfiglet.figlet_format("AgentDeck", font="slant")
-    lines = [
-        banner,
-        f"  {settings.app_name} v{settings.app_version}",
-        f"  Environment: {settings.environment or 'production'}",
-    ]
-    if settings.supabase_url and settings.supabase_service_role_key:
-        lines.append("  Supabase: connected")
-    else:
-        lines.append("  Supabase: not configured")
-    if settings.redis_url:
-        lines.append("  Redis: connected")
-    else:
-        lines.append("  Redis: not configured (degraded mode)")
-    lines.append("")
-    lines.append("=" * 78)
-    return "\n".join(lines)

app/core/security.py

@@ -1,33 +0,0 @@
-from __future__ import annotations
-
-from fastapi import HTTPException, Security, status
-from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
-
-from app.config import get_settings
-from app.core.logger import get_logger
-
-_logger = get_logger(__name__)
-_security = HTTPBearer(auto_error=False)
-
-
-async def require_api_key(
-    credentials: HTTPAuthorizationCredentials | None = Security(_security),
-) -> str:
-    settings = get_settings()
-    token = settings.api_key
-
-    if not credentials:
-        _logger.warning("Missing Authorization header")
-        raise HTTPException(
-            status_code=status.HTTP_401_UNAUTHORIZED,
-            detail="Missing Authorization header",
-        )
-
-    if credentials.credentials != token:
-        _logger.warning("Invalid API key provided")
-        raise HTTPException(
-            status_code=status.HTTP_401_UNAUTHORIZED,
-            detail="Invalid API key",
-        )
-
-    return credentials.credentials

app/core/thread_pool.py

@@ -1,10 +1,26 @@
 from __future__ import annotations
 
+import asyncio
 import concurrent.futures
 import os
+from typing import Any, Callable
 
 _MAX_WORKERS = min(32, (os.cpu_count() or 1) + 4)
 
 thread_pool = concurrent.futures.ThreadPoolExecutor(
     max_workers=_MAX_WORKERS, thread_name_prefix="shared"
 )
+
+
+async def run_in_executor(fn: Callable[..., Any], *args: Any, **kwargs: Any) -> Any:
+    """Run a blocking/cpu-bound callable on the shared thread pool.
+
+    Centralises the repeated ``loop.run_in_executor(thread_pool, ...)`` pattern so
+    async routes never block the event loop on sync I/O or CPU work. Uses the
+    shared pool so a bounded number of threads is reused across the application.
+    """
+    if kwargs:
+        return await asyncio.get_running_loop().run_in_executor(
+            thread_pool, lambda: fn(*args, **kwargs)
+        )
+    return await asyncio.get_running_loop().run_in_executor(thread_pool, fn, *args)

app/core/vector_store/deps.py

@@ -1,9 +1,7 @@
 from __future__ import annotations
 
 import logging
-from typing import Optional
 
-from app.config import get_settings
 from app.services.supabase import SupabaseClient, get_supabase_client
 
 logger = logging.getLogger(__name__)

app/services/auth_service.py

@@ -23,6 +23,7 @@ from app.core.auth.schemas import (
     UpdateProfileSchema,
     UserProfile,
 )
+from app.core.thread_pool import run_in_executor
 from app.services.supabase import AuthRepository, SupabaseClient
 
 logger = logging.getLogger("auth_service")
@@ -34,6 +35,10 @@ def _token_key(raw: str) -> str:
     return hashlib.sha256(raw.encode("utf-8")).hexdigest()
 
 
+async def _hash_password(password: str) -> str:
+    return await run_in_executor(ph.hash, password)
+
+
 class AuthService:
 
     @staticmethod
@@ -56,7 +61,7 @@ class AuthService:
             email=schema.email,
             username=schema.username,
             full_name=schema.full_name,
-            password_hash=ph.hash(schema.password),
+            password_hash=await _hash_password(schema.password),
             created_at=now,
             updated_at=now,
         )
@@ -82,7 +87,7 @@ class AuthService:
                 detail=f"Account locked until {user.locked_until.isoformat()}",
             )
 
-        if not AuthService._verify_password(schema.password, user.password_hash):
+        if not await AuthService._verify_password_async(schema.password, user.password_hash):
             user.failed_login_attempts += 1
             if user.failed_login_attempts >= _settings.max_login_attempts:
                 user.locked_until = _now() + timedelta(minutes=_settings.lockout_minutes)
@@ -99,8 +104,8 @@ class AuthService:
             "last_login": now,
         })
 
-        access_token = AuthService._create_access_token(user.id)
-        raw_refresh, refresh_hash, token_key, expires_at = AuthService._create_refresh_token()
+        access_token = await AuthService._create_access_token_async(user.id)
+        raw_refresh, refresh_hash, token_key, expires_at = await AuthService._create_refresh_token_async()
 
         session_id = _uuid()
         session = RefreshSession(
@@ -134,7 +139,7 @@ class AuthService:
                 )
             raise HTTPException(status_code=401, detail="Invalid or expired refresh token")
 
-        if not AuthService._verify_token(raw_refresh_token, session.token_hash):
+        if not await AuthService._verify_token_async(raw_refresh_token, session.token_hash):
             raise HTTPException(status_code=401, detail="Invalid refresh token")
 
         user = await repo.find_user_by_id(session.user_id)
@@ -144,8 +149,8 @@ class AuthService:
         now = _now()
         await repo.revoke_session(session.id)
 
-        new_access = AuthService._create_access_token(user.id)
-        new_raw_refresh, new_hash, new_key, new_expires = AuthService._create_refresh_token()
+        new_access = await AuthService._create_access_token_async(user.id)
+        new_raw_refresh, new_hash, new_key, new_expires = await AuthService._create_refresh_token_async()
 
         new_session_id = _uuid()
         new_session = RefreshSession(
@@ -181,12 +186,12 @@ class AuthService:
 
     @staticmethod
     async def change_password(db: SupabaseClient, user: User, schema: ChangePasswordSchema):
-        if not AuthService._verify_password(schema.current_password, user.password_hash):
+        if not await AuthService._verify_password_async(schema.current_password, user.password_hash):
             raise HTTPException(status_code=400, detail="Incorrect current password")
         repo = AuthRepository(db)
         now = _now()
         await repo.update_user(user.id, {
-            "password_hash": ph.hash(schema.new_password),
+            "password_hash": await _hash_password(schema.new_password),
             "password_changed_at": now,
         })
 
@@ -195,7 +200,8 @@ class AuthService:
         repo = AuthRepository(db)
         user = await repo.find_user_by_email(schema.email)
         if user:
-            reset_token = jwt.encode(
+            reset_token = await run_in_executor(
+                jwt.encode,
                 {
                     "sub": user.id,
                     "type": "reset_password",
@@ -209,7 +215,8 @@ class AuthService:
     @staticmethod
     async def reset_password(db: SupabaseClient, schema: ResetPasswordSchema):
         try:
-            payload = jwt.decode(
+            payload = await run_in_executor(
+                jwt.decode,
                 schema.token,
                 _settings.jwt_secret_key,
                 algorithms=[_settings.jwt_algorithm],
@@ -227,7 +234,7 @@ class AuthService:
 
         now = _now()
         await repo.update_user(user_id, {
-            "password_hash": ph.hash(schema.new_password),
+            "password_hash": await _hash_password(schema.new_password),
             "password_changed_at": now,
         })
         await repo.revoke_all_user_sessions(user_id)
@@ -278,6 +285,10 @@ class AuthService:
             if count:
                 logger.info("Cleaned up %s expired sessions", count)
 
+    @staticmethod
+    async def _create_access_token_async(user_id: str) -> str:
+        return await run_in_executor(AuthService._create_access_token, user_id)
+
     @staticmethod
     def _create_access_token(user_id: str) -> str:
         now = _now()
@@ -290,6 +301,10 @@ class AuthService:
         }
         return jwt.encode(payload, _settings.jwt_secret_key, algorithm=_settings.jwt_algorithm)
 
+    @staticmethod
+    async def _create_refresh_token_async() -> tuple[str, str, str, datetime]:
+        return await run_in_executor(AuthService._create_refresh_token)
+
     @staticmethod
     def _create_refresh_token() -> tuple[str, str, str, datetime]:
         token = secrets.token_urlsafe(64)
@@ -298,6 +313,10 @@ class AuthService:
         expires_at = _now() + timedelta(days=_settings.refresh_token_expire_days)
         return token, token_hash, token_key, expires_at
 
+    @staticmethod
+    async def _verify_password_async(plain: str, hashed: str) -> bool:
+        return await run_in_executor(AuthService._verify_password, plain, hashed)
+
     @staticmethod
     def _verify_password(plain: str, hashed: str) -> bool:
         try:
@@ -306,6 +325,10 @@ class AuthService:
         except VerifyMismatchError:
             return False
 
+    @staticmethod
+    async def _verify_token_async(raw: str, hashed: str) -> bool:
+        return await run_in_executor(AuthService._verify_token, raw, hashed)
+
     @staticmethod
     def _verify_token(raw: str, hashed: str) -> bool:
         try:

app/services/code_executor_service.py

@@ -168,17 +168,9 @@ class CodeExecutorService:
     ) -> dict:
         return run_subprocess(cmd, timeout, self._max_output_bytes)
 
-    async def check_runtimes(self) -> dict[str, str]:
-        status = {}
-        for lang, runtime in [("python", "python3"), ("javascript", "node")]:
-            path = shutil.which(runtime)
-            status[lang] = f"found at {path}" if path else "missing"
-        return status
-
     @staticmethod
     def _filename_for(language: str) -> str:
-        return {"python": "code.py", "javascript": "code.js"
-               }[language]
+        return {"python": "code.py", "javascript": "code.js"}[language]
 
     @staticmethod
     def _build_command(language: str, run_dir: Path, filename: str) -> list[str]:

app/services/csv_analysis_service.py

@@ -10,7 +10,6 @@ import time
 from pathlib import Path
 from typing import Any, Dict, List, Optional, Tuple, Union
 
-from app.services.code_executor_service import CodeSanitizer
 from app.services.dataset_metadata_service import extract_metadata
 from app.utils.http_utils import download_url
 from app.utils.subprocess_utils import run_subprocess
@@ -184,85 +183,3 @@ async def execute_csv_chat_blocks(
 
 async def get_dataset_info(source: Union[str, bytes, Any]) -> Dict[str, Any]:
     return await extract_metadata(source)
-
-
-async def analyze_csv_dataset(
-    source: Union[str, bytes],
-    code: str,
-    timeout: int = 30,
-) -> Dict[str, Any]:
-    data, _ = await _resolve_source(source)
-
-    if not data:
-        return {"success": False, "output": "", "error": "No data provided", "execution_time_ms": None}
-
-    sanitized, err = CodeSanitizer.sanitize(code, "python")
-    if not sanitized:
-        return {"success": False, "output": "", "error": err, "execution_time_ms": None}
-
-    async with _semaphore:
-        run_dir = None
-        start = time.monotonic()
-        try:
-            run_dir = Path(tempfile.mkdtemp())
-            csv_path = run_dir / "data.csv"
-            csv_path.write_bytes(data)
-
-            loader = (
-                "import pandas as pd, numpy as np\n"
-                f"df = pd.read_csv(r'{csv_path}')\n"
-            )
-            full_code = loader + code
-
-            code_path = run_dir / "analysis.py"
-            code_path.write_text(full_code, encoding="utf-8")
-
-            cmd = [_PYTHON, str(code_path)]
-            result = await asyncio.to_thread(_run_subprocess, cmd, timeout, _MAX_OUTPUT_BYTES)
-
-            elapsed_ms = (time.monotonic() - start) * 1000
-
-            return {
-                "success": result["exit_code"] == 0 and not result["timed_out"],
-                "output": result["stdout"],
-                "error": result["stderr"] or None,
-                "execution_time_ms": round(elapsed_ms, 2),
-                "timed_out": result["timed_out"],
-            }
-
-        except FileNotFoundError:
-            elapsed_ms = (time.monotonic() - start) * 1000
-            return {"success": False, "output": "", "error": f"Python runtime ({_PYTHON}) not found", "execution_time_ms": round(elapsed_ms, 2)}
-        except Exception as exc:
-            elapsed_ms = (time.monotonic() - start) * 1000
-            logger.exception("CSV analysis execution error")
-            return {"success": False, "output": "", "error": f"Execution error: {exc}", "execution_time_ms": round(elapsed_ms, 2)}
-        finally:
-            if run_dir and run_dir.exists():
-                shutil.rmtree(run_dir, ignore_errors=True)
-
-
-async def create_csv_chart(
-    source: Union[str, bytes],
-    code: str,
-    timeout: int = 30,
-) -> Dict[str, Any]:
-    preamble = (
-        "import matplotlib\nmatplotlib.use('Agg')\n"
-        "import matplotlib.pyplot as plt\nimport seaborn as sns\nimport pandas as pd\n"
-    )
-    postamble = (
-        "\n\nfrom io import BytesIO\nimport base64\n"
-        "_buf = BytesIO()\n"
-        "plt.savefig(_buf, format='png', bbox_inches='tight', dpi=150)\n"
-        "_buf.seek(0)\n"
-        "print(base64.b64encode(_buf.read()).decode(), end='')\n"
-        "plt.close('all')\n"
-    )
-
-    full_code = preamble + code + postamble
-    result = await analyze_csv_dataset(source, full_code, timeout)
-
-    if result["success"]:
-        return {"success": True, "image_base64": result.get("output", "") or "", "error": None, "execution_time_ms": result["execution_time_ms"]}
-    return {"success": False, "image_base64": None, "error": result.get("error") or "Chart generation failed or produced no output", "execution_time_ms": result["execution_time_ms"]}

app/services/embeddings_service.py

@@ -48,10 +48,6 @@ class EmbeddingService:
         self._loaded_dimensions.append(dimension)
         _logger.info("Loaded embedding model dim=%s (device=%s)", dimension, self._device)
 
-    def load_all_models(self) -> None:
-        for dim in _MODEL_MAP:
-            self.load_model(dim)
-
     def generate_embedding(self, text: List[str], dimension: int) -> List[List[float]]:
         if dimension not in self._models:
             raise ValueError(f"Model for dimension {dimension} not loaded")

app/services/gcs_service.py

@@ -15,6 +15,7 @@ from pydantic import BaseModel, ValidationError
 
 from app.config import get_settings
 from app.core.logger import get_logger
+from app.core.thread_pool import run_in_executor
 from app.utils.http_utils import SharedAsyncClient
 
 _logger = get_logger(__name__)
@@ -236,7 +237,7 @@ class GCSService:
             return token
 
     async def _fetch_access_token(self, creds: GCSCredentials) -> Tuple[str, int]:
-        assertion = self._build_signed_jwt(creds)
+        assertion = await run_in_executor(self._build_signed_jwt, creds)
         body = urlencode({
             "grant_type": "urn:ietf:params:oauth:grant-type:jwt-bearer",
             "assertion": assertion,
@@ -1039,8 +1040,12 @@ class GCSService:
         response_content_type: Optional[str] = None,
         response_disposition: Optional[str] = None,
     ) -> Tuple[str, int]:
-        return self.sign_url_v4(
-            creds, "GET", bucket, name,
+        return await run_in_executor(
+            self.sign_url_v4,
+            creds,
+            "GET",
+            bucket,
+            name,
             expires_in_seconds=expires_in_seconds,
             response_content_type=response_content_type,
             response_disposition=response_disposition,
@@ -1055,8 +1060,12 @@ class GCSService:
         expires_in_seconds: Optional[int] = None,
         content_type: Optional[str] = None,
     ) -> Tuple[str, int]:
-        return self.sign_url_v4(
-            creds, "PUT", bucket, name,
+        return await run_in_executor(
+            self.sign_url_v4,
+            creds,
+            "PUT",
+            bucket,
+            name,
             expires_in_seconds=expires_in_seconds,
             content_type=content_type,
         )

app/services/google_oauth_service.py

@@ -10,6 +10,7 @@ import jwt
 
 from app.config import get_settings
 from app.core.logger import get_logger
+from app.core.thread_pool import run_in_executor
 from app.models.schemas import GoogleOAuthUserInfo
 from app.utils.http_utils import SharedAsyncClient
 
@@ -318,7 +319,8 @@ class GoogleOAuthService:
             raise
 
         try:
-            payload = jwt.decode(
+            payload = await run_in_executor(
+                jwt.decode,
                 id_token,
                 key=public_key,
                 algorithms=["RS256"],
@@ -352,29 +354,6 @@ class GoogleOAuthService:
     # Userinfo (fallback profile source)
     # ------------------------------------------------------------------
 
-    async def fetch_userinfo(self, access_token: str) -> GoogleOAuthUserInfo:
-        client = await self._get_client()
-        try:
-            response = await client.get(
-                _settings.google_oauth_userinfo_url,
-                headers={"Authorization": f"Bearer {access_token}"},
-            )
-            response.raise_for_status()
-        except httpx.HTTPStatusError as exc:
-            if exc.response.status_code in (401, 403):
-                raise GoogleOAuthError(
-                    "Access token is invalid or has expired.", status_code=401
-                ) from exc
-            raise GoogleOAuthError(
-                "Failed to fetch Google user profile.", status_code=502
-            ) from exc
-        except httpx.RequestError as exc:
-            raise GoogleOAuthError(
-                "Failed to reach Google userinfo endpoint.", status_code=502
-            ) from exc
-
-        return self._user_info_from_claims(response.json())
-
     @staticmethod
     def _user_info_from_claims(payload: Dict[str, Any]) -> GoogleOAuthUserInfo:
         return GoogleOAuthUserInfo(

app/services/google_scope_service.py

@@ -73,9 +73,6 @@ class GoogleScopeService:
     async def load_all_plans(self) -> Dict[str, Dict[str, Any]]:
         return self._plans
 
-    async def load_default_state(self) -> Dict[str, List[str]]:
-        return {pid: _default_scopes(plan) for pid, plan in self._plans.items()}
-
     async def load_state(self) -> Dict[str, List[str]]:
         return self._state()
 

app/services/ocr_service.py

@@ -119,9 +119,3 @@ def ocr_pdf(source: Union[str, bytes], *, dpi: int = 150) -> str:
 class OCRService:
     def __init__(self) -> None:
         self._engine = _get_engine()
-
-    def image_to_text(self, source, text_score: float = 0.5) -> str:
-        return ocr_image(source, text_score=text_score)
-
-    def pdf_to_text(self, source: Union[str, bytes], dpi: int = 150) -> str:
-        return ocr_pdf(source, dpi=dpi)

app/services/reconciliation_service.py

@@ -4,10 +4,8 @@ import asyncio
 import io
 import os
 import re
-import time
 import unicodedata
 from collections import Counter
-from datetime import datetime
 from typing import Any, Dict, List, Optional, Set, Tuple
 
 import aiohttp
@@ -23,7 +21,6 @@ DOWNLOAD_TIMEOUT = 60
 DOWNLOAD_MAX_RETRIES = 3
 DOWNLOAD_BACKOFF_FACTOR = 0.5
 SUPPORTED_EXTENSIONS: Set[str] = {"csv", "xlsx", "xls", "tsv", "parquet"}
-_MAX_FILE_SIZE = _settings.max_upload_bytes
 
 DEFAULT_NUMERIC_KEYWORDS: Set[str] = {
     "amount", "total", "debit", "credit", "tax",
@@ -37,15 +34,6 @@ class ReconciliationError(Exception):
 class DownloadError(ReconciliationError):
     pass
 
-class FileTypeMismatchError(ReconciliationError):
-    pass
-
-class SchemaMismatchError(ReconciliationError):
-    pass
-
-class EmptyDatasetError(ReconciliationError):
-    pass
-
 class UnsupportedFormatError(ReconciliationError):
     pass
 
@@ -253,168 +241,3 @@ def analyze_missing_data(df_src: pd.DataFrame, df_dst: pd.DataFrame, common_cols
             "difference": abs(src_nulls - dst_nulls)
         })
     return results
-
-
-def _build_failed_pair_result(job_id: str, src_url: str, dst_url: str, started_at: str, status: str, errors: List[str], schema: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
-    return {
-        "job_id": job_id,
-        "source_file": src_url,
-        "destination_file": dst_url,
-        "started_at": started_at,
-        "completed_at": datetime.utcnow().isoformat(),
-        "processing_time_ms": 0,
-        "status": status,
-        "summary": {"overall_status": status, "overall_match_percentage": 0.0},
-        "schema": schema or {"fully_match": False, "source_columns": 0, "destination_columns": 0},
-        "columns": [], "numeric_columns": [], "date_columns": [],
-        "duplicates": {}, "missing_values": [],
-        "errors": errors,
-        "warnings": []
-    }
-
-
-async def process_pair(
-    src_url: str,
-    dst_url: str,
-    src_data: bytes,
-    dst_data: bytes,
-    ext: str,
-    job_id: str,
-    column_mapping: Optional[Dict[str, str]] = None,
-    numeric_keywords: Optional[Set[str]] = None,
-) -> Dict[str, Any]:
-    start_time = time.time()
-    started_at = datetime.utcnow().isoformat()
-    errors = []
-    warnings = []
-    status = "MATCH"
-
-    try:
-        df_src = read_to_dataframe(src_data, ext, job_id)
-        df_dst = read_to_dataframe(dst_data, ext, job_id)
-
-        if df_src.empty or df_dst.empty:
-            raise EmptyDatasetError("One or both datasets are empty.")
-
-        df_src = normalize_dataframe(df_src.copy())
-        df_dst = normalize_dataframe(df_dst.copy())
-        df_src.infer_objects()
-        df_dst.infer_objects()
-
-        if column_mapping:
-            df_dst = apply_column_mapping(df_dst, column_mapping)
-
-        schema_report = compare_schemas(df_src, df_dst)
-        if not schema_report["fully_match"]:
-            status = "SCHEMA_MISMATCH"
-            warnings.append("Schema mismatch detected.")
-
-        common_cols = df_src.columns.intersection(df_dst.columns).tolist()
-
-        identical_rows, _, missing_rows, extra_rows = compare_rows(df_src, df_dst, common_cols)
-
-        if missing_rows > 0 or extra_rows > 0:
-            status = "PARTIAL_MATCH" if status == "MATCH" else status
-
-        col_reports = reconcile_columns(df_src, df_dst, common_cols)
-        num_reports = reconcile_numeric_columns(df_src, df_dst, common_cols, numeric_keywords)
-        date_reports = reconcile_date_columns(df_src, df_dst, common_cols)
-
-        dup_report = analyze_duplicates(df_src, df_dst)
-        missing_report = analyze_missing_data(df_src, df_dst, common_cols)
-
-        full_cols = sum(1 for c in col_reports if c["status"] == "fully_match")
-        part_cols = sum(1 for c in col_reports if c["status"] == "partial_match")
-
-        if part_cols > 0 and status == "MATCH":
-            status = "PARTIAL_MATCH"
-
-        total_max_rows = max(len(df_src), len(df_dst))
-        match_pct = (identical_rows / total_max_rows * 100) if total_max_rows > 0 else 100.0
-
-        summary = {
-            "total_columns": len(common_cols),
-            "fully_matched_columns": full_cols,
-            "partially_matched_columns": part_cols,
-            "unmatched_columns": len(col_reports) - full_cols - part_cols,
-            "total_rows_source": len(df_src),
-            "total_rows_destination": len(df_dst),
-            "matching_rows": identical_rows,
-            "partial_rows": 0,
-            "unmatched_rows": missing_rows + extra_rows,
-            "duplicate_rows": dup_report["duplicate_difference"],
-            "overall_match_percentage": round(match_pct, 2),
-            "overall_status": status
-        }
-    except Exception as e:
-        _logger.exception(f"Job {job_id} failed during processing.", extra={"correlation_id": job_id})
-        status = "FAILED"
-        errors.append(str(e))
-        summary = {"overall_status": "FAILED", "overall_match_percentage": 0.0}
-        schema_report, col_reports, num_reports, date_reports, dup_report, missing_report = {}, [], [], [], {}, []
-
-    return {
-        "job_id": job_id,
-        "source_file": src_url,
-        "destination_file": dst_url,
-        "started_at": started_at,
-        "completed_at": datetime.utcnow().isoformat(),
-        "processing_time_ms": round((time.time() - start_time) * 1000, 2),
-        "status": status,
-        "summary": summary,
-        "schema": schema_report,
-        "columns": col_reports,
-        "numeric_columns": num_reports,
-        "date_columns": date_reports,
-        "duplicates": dup_report,
-        "missing_values": missing_report,
-        "errors": errors,
-        "warnings": warnings
-    }
-
-
-async def reconcile_pair(
-    pair: Dict[str, Any],
-    job_id: str,
-    numeric_keywords: Optional[Set[str]] = None,
-) -> Dict[str, Any]:
-    src_url = pair["source"]
-    dst_url = pair["destination"]
-    started_at = datetime.utcnow().isoformat()
-
-    src_ext = extract_extension(src_url)
-    dst_ext = extract_extension(dst_url)
-
-    if src_ext not in SUPPORTED_EXTENSIONS or dst_ext not in SUPPORTED_EXTENSIONS:
-        return _build_failed_pair_result(
-            job_id, src_url, dst_url, started_at, "FAILED",
-            [f"Unsupported format. Source: {src_ext}, Dest: {dst_ext}"]
-        )
-
-    if src_ext != dst_ext:
-        return _build_failed_pair_result(
-            job_id, src_url, dst_url, started_at, "FILE_TYPE_MISMATCH",
-            [f"File type mismatch. Source: {src_ext}, Dest: {dst_ext}"]
-        )
-
-    column_mapping: Optional[Dict[str, str]] = pair.get("column_mapping")
-
-    async with aiohttp.ClientSession() as session:
-        try:
-            src_data, dst_data = await asyncio.gather(
-                download_file_with_retry(session, src_url, job_id),
-                download_file_with_retry(session, dst_url, job_id)
-            )
-        except DownloadError as e:
-            _logger.error(f"Download failed for job {job_id}: {str(e)}", extra={"correlation_id": job_id})
-            return _build_failed_pair_result(
-                job_id, src_url, dst_url, started_at, "FAILED", [str(e)]
-            )
-
-    if len(src_data) > _MAX_FILE_SIZE or len(dst_data) > _MAX_FILE_SIZE:
-        return _build_failed_pair_result(
-            job_id, src_url, dst_url, started_at, "FAILED",
-            ["File size exceeds maximum allowed limit"]
-        )
-
-    return await process_pair(src_url, dst_url, src_data, dst_data, src_ext, job_id, column_mapping, numeric_keywords)

app/services/scheduler_service.py

@@ -1,8 +1,6 @@
 from __future__ import annotations
 
 import asyncio
-import hashlib
-import json
 import re
 import secrets
 import uuid
@@ -456,7 +454,7 @@ class HttpExecutionEngine:
                 error_message=None if success else f"HTTP {response.status_code}",
             )
 
-        except httpx.TimeoutException as exc:
+        except httpx.TimeoutException:
             duration_ms = (datetime.now(UTC) - started).total_seconds() * 1000
             return HttpExecutionResult(
                 success=False, duration_ms=duration_ms,
@@ -955,10 +953,6 @@ class SchedulerService:
         if result.exception_type in ("TimeoutException", "ExecutionTimeout"):
             exec_status = ExecutionStatus.TIMEOUT
 
-        headers_hash = hashlib.sha256(
-            json.dumps(job.get("headers") or {}, sort_keys=True).encode()
-        ).hexdigest() if job.get("headers") else ""
-
         history_data = {
             "id": str(uuid.uuid4()),
             "job_id": job["id"],

app/services/supabase/client.py

@@ -96,14 +96,6 @@ class SupabaseClient:
             return result.data[0]
         return None
 
-    async def execute_sql(self, sql: str) -> Any:
-        result = await asyncio.to_thread(
-            self.client.rpc,
-            "exec_sql",
-            {"sql": sql},
-        )
-        return result
-
     async def find_one(
         self, table: str, column: str, value: Any,
         columns: str = "*",

app/services/supabase/repositories.py

@@ -1,11 +1,10 @@
 from __future__ import annotations
 
-import json
 import logging
-from datetime import datetime, timezone
+from datetime import datetime
 from typing import Any, Dict, List, Optional
 
-from app.core.auth.models import RefreshSession, User, _uuid
+from app.core.auth.models import RefreshSession, User
 from app.core.auth.models import _utcnow as _now
 from app.core.vector_store.models import VectorStoreIndex
 from app.services.supabase.client import SupabaseClient
@@ -132,15 +131,6 @@ class AuthRepository:
         )
         return [r["name"] for r in rows if "name" in r]
 
-    async def get_user_permissions(self, user_id: str) -> List[str]:
-        rows = await self._client.select_in(
-            "permissions",
-            "id",
-            [],
-            columns="code",
-        )
-        return [r["code"] for r in rows if "code" in r]
-
     async def create_refresh_session(self, session: RefreshSession) -> None:
         data = {
             "id": session.id,
@@ -226,12 +216,6 @@ class VectorStoreRepository:
         rows = await self._client.select("vector_store_index", order=("created_at", False))
         return [VectorStoreIndex.from_dict(r) for r in rows]
 
-    async def find_by_id(self, store_id: str) -> Optional[VectorStoreIndex]:
-        row = await self._client.find_one("vector_store_index", "store_id", store_id)
-        if row is None:
-            return None
-        return VectorStoreIndex.from_dict(row)
-
     async def upsert(self, index: VectorStoreIndex) -> None:
         data = {
             "store_id": index.store_id,

app/services/url_shortener_service.py

@@ -575,7 +575,7 @@ class Storage:
 
     def get_campaign_total_clicks(self, campaign_id: str) -> int:
         links = _run_async(self._get_client().select("url_shortener_links", eq=("campaign_id", campaign_id)))
-        return sum(l.get("click_count") or 0 for l in links)
+        return sum(link.get("click_count") or 0 for link in links)
 
     def get_campaign_analytics(self, campaign_id: str) -> Dict[str, Any]:
         client = self._get_client()
@@ -607,8 +607,8 @@ class Storage:
         client = self._get_client()
         links = _run_async(client.select("url_shortener_links", eq=("owner_id", owner_id)))
         link_count = len(links)
-        total_clicks = sum(l.get("click_count") or 0 for l in links)
-        active = sum(1 for l in links if l.get("is_active"))
+        total_clicks = sum(link.get("click_count") or 0 for link in links)
+        active = sum(1 for link in links if link.get("is_active"))
         return {
             "owner_id": owner_id,
             "name": row["name"],

app/services/vector_store_service.py

@@ -25,10 +25,6 @@ _EMBEDDING_DIM = 384
 _MAX_WORKERS = min(16, (os.cpu_count() or 1) + 4)
 
 
-def _run_sync(fn, *args, **kwargs):
-    return fn(*args, **kwargs)
-
-
 class VectorStoreRecord:
     def __init__(
         self,
@@ -258,44 +254,6 @@ class VectorStoreService:
             items.append(item)
         return items
 
-    def _fetch_documents_sync(self, store_id: str, ids: List[str]) -> Dict[str, Any]:
-        col = self._get_collection(store_id)
-        if col is None:
-            record = self._stores.get(store_id)
-            if record is None:
-                raise ValueError(f"Vector store {store_id} not found")
-            col = self._open_or_create_collection_sync(store_id, record.path)
-
-        internal_ids = []
-        for doc_id in ids:
-            fetched = col.fetch(ids=[doc_id])
-            if doc_id in fetched:
-                internal_ids.append(doc_id)
-                continue
-            for i in range(0, 1024):
-                chunk_id = f"{doc_id}_{i}"
-                fetched = col.fetch(ids=[chunk_id])
-                if chunk_id in fetched:
-                    internal_ids.append(chunk_id)
-                else:
-                    break
-            break
-
-        if not internal_ids:
-            return {}
-
-        fetched = col.fetch(ids=internal_ids)
-        result = {}
-        for k, v in fetched.items():
-            result[k] = {
-                "id": v.id,
-                "text": v.field("text") if hasattr(v, "field") else "",
-                "doc_id": v.field("doc_id") if hasattr(v, "field") else "",
-                "chunk_index": v.field("chunk_index") if hasattr(v, "field") else 0,
-                "source": v.field("source") if hasattr(v, "field") else "",
-            }
-        return result
-
     def _delete_documents_sync(
         self,
         store_id: str,
@@ -478,11 +436,6 @@ class VectorStoreService:
         elapsed = (time.perf_counter() - start) * 1000
         return items, elapsed
 
-    async def fetch_documents(self, store_id: str, ids: List[str]) -> Dict[str, Any]:
-        return await self._run_sync_fn(
-            lambda: self._fetch_documents_sync(store_id, ids)
-        )
-
     async def delete_documents(
         self,
         store_id: str,

app/services/webhook_socket_service.py

@@ -160,9 +160,3 @@ def get_manager() -> ChannelManager:
     if _manager is None:
         _manager = ChannelManager()
     return _manager
-
-
-def init_manager(default_buffer: int = 0) -> ChannelManager:
-    global _manager
-    _manager = ChannelManager(default_buffer=default_buffer)
-    return _manager

app/utils/http_utils.py

@@ -155,28 +155,28 @@ async def download_url(
     try:
         session = await get_shared_aiohttp_session()
         async with session.get(url, timeout=timeout) as resp:
-                if resp.status != 200:
-                    raise DownloadError(f"HTTP {resp.status} when fetching {url}")
-
-                if max_size_bytes and resp.content_length and resp.content_length > max_size_bytes:
+            if resp.status != 200:
+                raise DownloadError(f"HTTP {resp.status} when fetching {url}")
+
+            if max_size_bytes and resp.content_length and resp.content_length > max_size_bytes:
+                raise DownloadError(
+                    f"Remote file advertises {resp.content_length} bytes, "
+                    f"limit is {max_size_bytes}",
+                    is_size_error=True,
+                )
+
+            chunks: List[bytes] = []
+            total = 0
+            async for chunk in resp.content.iter_chunked(chunk_size):
+                total += len(chunk)
+                if max_size_bytes and total > max_size_bytes:
                     raise DownloadError(
-                        f"Remote file advertises {resp.content_length} bytes, "
-                        f"limit is {max_size_bytes}",
+                        f"Download exceeded {max_size_bytes} bytes",
                         is_size_error=True,
                     )
+                chunks.append(chunk)
 
-                chunks: List[bytes] = []
-                total = 0
-                async for chunk in resp.content.iter_chunked(chunk_size):
-                    total += len(chunk)
-                    if max_size_bytes and total > max_size_bytes:
-                        raise DownloadError(
-                            f"Download exceeded {max_size_bytes} bytes",
-                            is_size_error=True,
-                        )
-                    chunks.append(chunk)
-
-                data = b"".join(chunks)
+            data = b"".join(chunks)
     except (aiohttp.ClientError, asyncio.TimeoutError) as exc:
         raise DownloadError(f"Download failed for {url}: {exc}") from exc
 

requirements.txt

@@ -1,76 +1,66 @@
-zvec>=0.4.0
-markitdown[all]>=0.1.5
-fastapi>=0.111.0
-uvicorn[standard]>=0.30.0
-pydantic>=2.7.0
-pydantic-settings>=2.0.0
-python-multipart>=0.0.9
-httpx>=0.27.0
-numpy>=1.26.0
-rapidocr-onnxruntime>=1.4.4
-onnxruntime>=1.18.0
-pillow>=10.0.0
+# --- Core framework ---
+fastapi==0.115.5
+uvicorn[standard]==0.34.0
+pydantic==2.13.4
+pydantic-settings==2.6.1
+python-multipart==0.0.20
+httpx==0.28.1
+aiohttp==3.11.13
 
-# QR / image processing
-opencv-python-headless>=4.9.0
-pypdfium2>=4.30.0
-pandas>=2.0.0
-matplotlib>=3.8.0
-seaborn>=0.13.0
-sentence-transformers==5.6.0
+# --- Media / file processing ---
+numpy==1.26.4
+pillow==10.3.0
+opencv-python-headless==4.12.0.88
+pypdfium2==4.30.0
+rapidocr-onnxruntime==1.4.4
+onnxruntime==1.20.1
+markitdown[all]==0.1.5
 
-aiohttp>=3.9.0
-aiomysql>=0.2.0
+# --- Data / analysis ---
+pandas==3.0.1
+matplotlib==3.9.2
+seaborn==0.13.2
 
-# FIXED: Downgraded to 4.49.0. Transformers 5.x breaks Nomic Vision's trust_remote_code architecture.
-transformers==4.49.0
+# --- ML / embeddings / OCR models ---
+zvec==0.4.0
+sentence-transformers==3.4.1
+transformers==4.50.2
+torch==2.5.1
+einops==0.8.2
 
-torch==2.12.1
-# torchvision==0.27.1
-einops
-spacy>=3.7.0
-phonenumbers>=8.13.0
-sqlglot>=20.0.0
-tiktoken>=0.9.0
-PyJWT>=2.9.0
-cryptography>=42.0.0
+# --- NLP ---
+spacy==3.8.3
+phonenumbers==9.0.33
+sqlglot==25.31.4
+tiktoken==0.7.0
+clean-text==0.7.1
+text-unidecode==1.3
+Unidecode==1.4.0
+chardet==5.2.0
 
-# Authentication
-argon2-cffi>=23.1.0
-email-validator>=2.1.0
-slowapi>=0.1.9
+# --- Web scraping / search ---
+scrapling[all]==0.4.7
 
-# Async database drivers
-supabase>=2.0.0
-asyncpg>=0.31.0
-motor>=3.7.1
+# --- Auth / security ---
+PyJWT==2.10.1
+cryptography==43.0.3
+argon2-cffi==25.1.0
+email-validator==2.3.0
+slowapi==0.1.9
 
-# ASCII banner for startup
-pyfiglet>=1.0.2
+# --- Storage / cache / DB clients ---
+supabase==2.13.0
+redis==5.2.1
 
-# Web scraping
-scrapling[all]>=0.4.0
+# --- Validation / data access ---
+jsonschema==4.23.0
+jsonpath-ng==1.8.0
+qrcode[pil]==8.2
 
-# Chat / AI
-redis>=5.0.0
-defusedxml>=0.7.1
-chardet>=5.2.0
+# --- Scheduler ---
+apscheduler==3.11.0
+croniter==6.2.4
+pytz==2024.2
 
-# Text processing
-clean-text>=0.6.0
-text-unidecode>=1.3
-Unidecode>=1.3.8
-
-# JSON Schema validation
-jsonschema>=4.21.0
-
-# JSONPath extraction
-jsonpath-ng>=1.7.0
-
-# QR code generation (testing)
-qrcode[pil]>=8.0
-
-# Scheduler
-apscheduler>=3.10.4
-croniter>=2.0.0
-pytz>=2024.1
+# --- Misc ---
+pyfiglet==1.0.4