Fix blocking async server handlers

src/openjarvis/server/agent_manager_routes.py

@@ -2263,14 +2263,14 @@ async def sendblue_verify(request: Request):
         import httpx
 
         try:
-            resp = httpx.get(
-                "https://api.sendblue.co/api/lines",
-                headers={
-                    "sb-api-key-id": api_key_id,
-                    "sb-api-secret-key": api_secret_key,
-                },
-                timeout=15.0,
-            )
+            async with httpx.AsyncClient(timeout=15.0) as client:
+                resp = await client.get(
+                    "https://api.sendblue.co/api/lines",
+                    headers={
+                        "sb-api-key-id": api_key_id,
+                        "sb-api-secret-key": api_secret_key,
+                    },
+                )
             if resp.status_code == 401:
                 raise HTTPException(
                     status_code=401,
@@ -2290,12 +2290,16 @@ async def sendblue_verify(request: Request):
             )
             numbers = []
             for line in lines:
-                num = (
-                    line.get("number")
-                    or line.get("phone_number")
-                    or line.get("from_number")
-                    or (line if isinstance(line, str) else "")
-                )
+                if isinstance(line, str):
+                    num = line
+                elif isinstance(line, dict):
+                    num = (
+                        line.get("number")
+                        or line.get("phone_number")
+                        or line.get("from_number")
+                    )
+                else:
+                    num = None
                 if num:
                     numbers.append(num)
             return {
@@ -2327,18 +2331,18 @@ async def sendblue_register_webhook(request: Request):
         import httpx
 
         try:
-            resp = httpx.post(
-                "https://api.sendblue.co/api/account/webhooks",
-                headers={
-                    "sb-api-key-id": api_key_id,
-                    "sb-api-secret-key": api_secret_key,
-                    "Content-Type": "application/json",
-                },
-                json={
-                    "receive": webhook_url,
-                },
-                timeout=15.0,
-            )
+            async with httpx.AsyncClient(timeout=15.0) as client:
+                resp = await client.post(
+                    "https://api.sendblue.co/api/account/webhooks",
+                    headers={
+                        "sb-api-key-id": api_key_id,
+                        "sb-api-secret-key": api_secret_key,
+                        "Content-Type": "application/json",
+                    },
+                    json={
+                        "receive": webhook_url,
+                    },
+                )
             return {
                 "registered": resp.status_code < 300,
                 "status": resp.status_code,
@@ -2378,16 +2382,16 @@ async def sendblue_test(request: Request):
             if from_number:
                 payload["from_number"] = from_number
 
-            resp = httpx.post(
-                "https://api.sendblue.co/api/send-message",
-                headers={
-                    "sb-api-key-id": api_key_id,
-                    "sb-api-secret-key": api_secret_key,
-                    "Content-Type": "application/json",
-                },
-                json=payload,
-                timeout=15.0,
-            )
+            async with httpx.AsyncClient(timeout=15.0) as client:
+                resp = await client.post(
+                    "https://api.sendblue.co/api/send-message",
+                    headers={
+                        "sb-api-key-id": api_key_id,
+                        "sb-api-secret-key": api_secret_key,
+                        "Content-Type": "application/json",
+                    },
+                    json=payload,
+                )
             return {
                 "sent": resp.status_code < 300,
                 "status": resp.status_code,

src/openjarvis/server/api_routes.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import asyncio
 import inspect
 import json
 import logging
@@ -894,7 +895,12 @@ async def transcribe_speech(request: Request):
     ext = filename.rsplit(".", 1)[-1] if "." in filename else "wav"
 
     try:
-        result = backend.transcribe(audio_bytes, format=ext, language=language or None)
+        result = await asyncio.to_thread(
+            backend.transcribe,
+            audio_bytes,
+            format=ext,
+            language=language or None,
+        )
     except Exception as exc:
         logger.exception("Speech transcription failed")
         raise HTTPException(

src/openjarvis/server/routes.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import asyncio
 import logging
 import uuid
 from typing import Any
@@ -835,7 +836,7 @@ async def list_models(request: Request) -> ModelListResponse:
     # Filter out any cloud model IDs that may appear via MultiEngine.
     # Fall back to direct Ollama query only when the engine returns nothing.
     engine = request.app.state.engine
-    all_ids = engine.list_models()
+    all_ids = await asyncio.to_thread(engine.list_models)
     model_ids = [m for m in all_ids if not is_cloud_model(m)]
     if not model_ids:
         model_ids = await list_local_models()
@@ -865,12 +866,12 @@ async def pull_model(request: Request):
     import httpx as _httpx
 
     host = getattr(engine, "_host", "http://localhost:11434")
-    client = _httpx.Client(base_url=host, timeout=600.0)
     try:
-        resp = client.post(
-            "/api/pull",
-            json={"name": model_name, "stream": False},
-        )
+        async with _httpx.AsyncClient(base_url=host, timeout=600.0) as client:
+            resp = await client.post(
+                "/api/pull",
+                json={"name": model_name, "stream": False},
+            )
         resp.raise_for_status()
     except (_httpx.ConnectError, _httpx.TimeoutException) as exc:
         raise HTTPException(status_code=502, detail=f"Ollama unreachable: {exc}")
@@ -879,8 +880,6 @@ async def pull_model(request: Request):
             status_code=exc.response.status_code,
             detail=f"Ollama error: {exc.response.text[:300]}",
         )
-    finally:
-        client.close()
 
     return {"status": "ok", "model": model_name}
 
@@ -896,13 +895,13 @@ async def delete_model(model_name: str, request: Request):
     import httpx as _httpx
 
     host = getattr(engine, "_host", "http://localhost:11434")
-    client = _httpx.Client(base_url=host, timeout=30.0)
     try:
-        resp = client.request(
-            "DELETE",
-            "/api/delete",
-            json={"name": model_name},
-        )
+        async with _httpx.AsyncClient(base_url=host, timeout=30.0) as client:
+            resp = await client.request(
+                "DELETE",
+                "/api/delete",
+                json={"name": model_name},
+            )
         resp.raise_for_status()
     except (_httpx.ConnectError, _httpx.TimeoutException) as exc:
         raise HTTPException(status_code=502, detail=f"Ollama unreachable: {exc}")
@@ -911,8 +910,6 @@ async def delete_model(model_name: str, request: Request):
             status_code=exc.response.status_code,
             detail=f"Ollama error: {exc.response.text[:300]}",
         )
-    finally:
-        client.close()
 
     return {"status": "deleted", "model": model_name}
 

src/openjarvis/telemetry/aggregator.py

@@ -90,6 +90,9 @@ class TelemetryAggregator:
     def __init__(self, db_path: str | Path) -> None:
         self._db_path = str(db_path)
         self._conn = sqlite3.connect(self._db_path, check_same_thread=False)
+        self._conn.execute("PRAGMA journal_mode=WAL")
+        self._conn.execute("PRAGMA synchronous=NORMAL")
+        self._conn.execute("PRAGMA busy_timeout=5000")
         self._conn.row_factory = sqlite3.Row
 
     def _time_filter(

src/openjarvis/telemetry/store.py

@@ -5,6 +5,7 @@
 import json
 import logging
 import sqlite3
+import threading
 import time
 from pathlib import Path
 from typing import Any
@@ -148,6 +149,10 @@ class TelemetryStore:
     def __init__(self, db_path: str | Path) -> None:
         self._db_path = str(db_path)
         self._conn = sqlite3.connect(self._db_path, check_same_thread=False)
+        self._lock = threading.Lock()
+        self._conn.execute("PRAGMA journal_mode=WAL")
+        self._conn.execute("PRAGMA synchronous=NORMAL")
+        self._conn.execute("PRAGMA busy_timeout=5000")
         self._conn.execute(_CREATE_TABLE)
         self._conn.execute(_CREATE_MINING_STATS_TABLE)
         self._conn.commit()
@@ -166,82 +171,84 @@ def _migrate_schema(self) -> None:
 
     def record(self, rec: TelemetryRecord) -> None:
         """Persist a single telemetry record."""
-        self._conn.execute(
-            _INSERT,
-            (
-                rec.timestamp,
-                rec.model_id,
-                rec.engine,
-                rec.agent,
-                rec.prompt_tokens,
-                rec.prompt_tokens_evaluated,
-                rec.completion_tokens,
-                rec.total_tokens,
-                rec.latency_seconds,
-                rec.ttft,
-                rec.cost_usd,
-                rec.energy_joules,
-                rec.power_watts,
-                rec.gpu_utilization_pct,
-                rec.gpu_memory_used_gb,
-                rec.gpu_temperature_c,
-                rec.throughput_tok_per_sec,
-                rec.prefill_latency_seconds,
-                rec.decode_latency_seconds,
-                rec.energy_method,
-                rec.energy_vendor,
-                rec.batch_id,
-                1 if rec.is_warmup else 0,
-                rec.cpu_energy_joules,
-                rec.gpu_energy_joules,
-                rec.dram_energy_joules,
-                rec.tokens_per_joule,
-                rec.energy_per_output_token_joules,
-                rec.throughput_per_watt,
-                rec.prefill_energy_joules,
-                rec.decode_energy_joules,
-                rec.mean_itl_ms,
-                rec.median_itl_ms,
-                rec.p90_itl_ms,
-                rec.p95_itl_ms,
-                rec.p99_itl_ms,
-                rec.std_itl_ms,
-                1 if rec.is_streaming else 0,
-                rec.token_counting_version,
-                rec.mining_session_id,
-                json.dumps(rec.metadata),
-            ),
-        )
-        self._conn.commit()
+        with self._lock:
+            self._conn.execute(
+                _INSERT,
+                (
+                    rec.timestamp,
+                    rec.model_id,
+                    rec.engine,
+                    rec.agent,
+                    rec.prompt_tokens,
+                    rec.prompt_tokens_evaluated,
+                    rec.completion_tokens,
+                    rec.total_tokens,
+                    rec.latency_seconds,
+                    rec.ttft,
+                    rec.cost_usd,
+                    rec.energy_joules,
+                    rec.power_watts,
+                    rec.gpu_utilization_pct,
+                    rec.gpu_memory_used_gb,
+                    rec.gpu_temperature_c,
+                    rec.throughput_tok_per_sec,
+                    rec.prefill_latency_seconds,
+                    rec.decode_latency_seconds,
+                    rec.energy_method,
+                    rec.energy_vendor,
+                    rec.batch_id,
+                    1 if rec.is_warmup else 0,
+                    rec.cpu_energy_joules,
+                    rec.gpu_energy_joules,
+                    rec.dram_energy_joules,
+                    rec.tokens_per_joule,
+                    rec.energy_per_output_token_joules,
+                    rec.throughput_per_watt,
+                    rec.prefill_energy_joules,
+                    rec.decode_energy_joules,
+                    rec.mean_itl_ms,
+                    rec.median_itl_ms,
+                    rec.p90_itl_ms,
+                    rec.p95_itl_ms,
+                    rec.p99_itl_ms,
+                    rec.std_itl_ms,
+                    1 if rec.is_streaming else 0,
+                    rec.token_counting_version,
+                    rec.mining_session_id,
+                    json.dumps(rec.metadata),
+                ),
+            )
+            self._conn.commit()
 
     def record_mining_stats(self, stats: Any) -> None:
         """Persist one mining stats snapshot.
 
         ``stats`` is duck-typed to keep telemetry usable without importing the
         optional mining package at module import time.
         """
-        self._conn.execute(
-            """\
+        with self._lock:
+            self._conn.execute(
+                """\
 INSERT INTO mining_stats (
     recorded_at, provider_id, shares_submitted, shares_accepted, blocks_found,
     hashrate, uptime_seconds, last_share_at, last_error, payout_target, fees_owed
 ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
 """,
-            (
-                time.time(),
-                stats.provider_id,
-                stats.shares_submitted,
-                stats.shares_accepted,
-                stats.blocks_found,
-                stats.hashrate,
-                stats.uptime_seconds,
-                stats.last_share_at,
-                stats.last_error,
-                stats.payout_target,
-                stats.fees_owed,
-            ),
-        )
-        self._conn.commit()
+                (
+                    time.time(),
+                    stats.provider_id,
+                    stats.shares_submitted,
+                    stats.shares_accepted,
+                    stats.blocks_found,
+                    stats.hashrate,
+                    stats.uptime_seconds,
+                    stats.last_share_at,
+                    stats.last_error,
+                    stats.payout_target,
+                    stats.fees_owed,
+                ),
+            )
+            self._conn.commit()
 
     def list_recent(self, limit: int = 50) -> list[dict[str, Any]]:
         """Return recent telemetry rows as dictionaries."""