diff --git a/src/openjarvis/server/agent_manager_routes.py b/src/openjarvis/server/agent_manager_routes.py index d2825ec31..44bd97d55 100644 --- a/src/openjarvis/server/agent_manager_routes.py +++ b/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, diff --git a/src/openjarvis/server/api_routes.py b/src/openjarvis/server/api_routes.py index 3f046346d..b022ba583 100644 --- a/src/openjarvis/server/api_routes.py +++ b/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( diff --git a/src/openjarvis/server/routes.py b/src/openjarvis/server/routes.py index 956bc3c4c..cda60f567 100644 --- a/src/openjarvis/server/routes.py +++ b/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} diff --git a/src/openjarvis/telemetry/aggregator.py b/src/openjarvis/telemetry/aggregator.py index 79ac6fec2..b6bb95a9e 100644 --- a/src/openjarvis/telemetry/aggregator.py +++ b/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( diff --git a/src/openjarvis/telemetry/store.py b/src/openjarvis/telemetry/store.py index ab151b683..630912c49 100644 --- a/src/openjarvis/telemetry/store.py +++ b/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,53 +171,54 @@ 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. @@ -220,28 +226,29 @@ def record_mining_stats(self, stats: Any) -> None: ``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.""" diff --git a/tests/server/test_agent_manager_routes.py b/tests/server/test_agent_manager_routes.py index 4902563c7..6dda4fbcc 100644 --- a/tests/server/test_agent_manager_routes.py +++ b/tests/server/test_agent_manager_routes.py @@ -6,7 +6,7 @@ import tempfile from pathlib import Path from types import SimpleNamespace -from unittest.mock import MagicMock +from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -48,6 +48,71 @@ def test_list_agents_empty(self, client): assert resp.status_code == 200 assert resp.json()["agents"] == [] + def test_sendblue_verify_uses_async_http_client(self, client): + mock_resp = MagicMock() + mock_resp.status_code = 200 + mock_resp.json.return_value = [ + "+15551234567", + {"phone_number": "+15557654321"}, + None, + ] + + with patch("httpx.AsyncClient") as mock_client_cls: + instance = mock_client_cls.return_value.__aenter__.return_value + instance.get = AsyncMock(return_value=mock_resp) + resp = client.post( + "/v1/channels/sendblue/verify", + json={"api_key_id": "key-id", "api_secret_key": "secret"}, + ) + + assert resp.status_code == 200 + assert resp.json()["valid"] is True + assert resp.json()["numbers"] == ["+15551234567", "+15557654321"] + instance.get.assert_awaited_once() + + def test_sendblue_register_webhook_uses_async_http_client(self, client): + mock_resp = MagicMock() + mock_resp.status_code = 200 + mock_resp.json.return_value = {"ok": True} + + with patch("httpx.AsyncClient") as mock_client_cls: + instance = mock_client_cls.return_value.__aenter__.return_value + instance.post = AsyncMock(return_value=mock_resp) + resp = client.post( + "/v1/channels/sendblue/register-webhook", + json={ + "api_key_id": "key-id", + "api_secret_key": "secret", + "webhook_url": "https://example.com/webhooks/sendblue", + }, + ) + + assert resp.status_code == 200 + assert resp.json()["registered"] is True + instance.post.assert_awaited_once() + + def test_sendblue_test_message_uses_async_http_client(self, client): + mock_resp = MagicMock() + mock_resp.status_code = 200 + mock_resp.json.return_value = {"ok": True} + + with patch("httpx.AsyncClient") as mock_client_cls: + instance = mock_client_cls.return_value.__aenter__.return_value + instance.post = AsyncMock(return_value=mock_resp) + resp = client.post( + "/v1/channels/sendblue/test", + json={ + "api_key_id": "key-id", + "api_secret_key": "secret", + "from_number": "+15550000000", + "to_number": "+15551234567", + }, + ) + + assert resp.status_code == 200 + assert resp.json()["sent"] is True + instance.post.assert_awaited_once() + def test_create_agent(self, client): resp = client.post( "/v1/managed-agents", diff --git a/tests/server/test_model_management.py b/tests/server/test_model_management.py index 7e300135f..61f0440a6 100644 --- a/tests/server/test_model_management.py +++ b/tests/server/test_model_management.py @@ -3,7 +3,7 @@ from __future__ import annotations import json -from unittest.mock import MagicMock, patch +from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -75,10 +75,9 @@ def test_pull_success(self): mock_resp.status_code = 200 mock_resp.raise_for_status = MagicMock() - with patch("httpx.Client") as MockClient: - instance = MockClient.return_value - instance.post.return_value = mock_resp - instance.close = MagicMock() + with patch("httpx.AsyncClient") as MockClient: + instance = MockClient.return_value.__aenter__.return_value + instance.post = AsyncMock(return_value=mock_resp) resp = client.post("/v1/models/pull", json={"model": "qwen3.5:4b"}) @@ -86,6 +85,10 @@ def test_pull_success(self): data = resp.json() assert data["status"] == "ok" assert data["model"] == "qwen3.5:4b" + instance.post.assert_awaited_once_with( + "/api/pull", + json={"name": "qwen3.5:4b", "stream": False}, + ) def test_pull_ollama_unreachable(self): engine = _make_ollama_engine() @@ -93,10 +96,9 @@ def test_pull_ollama_unreachable(self): import httpx - with patch("httpx.Client") as MockClient: - instance = MockClient.return_value - instance.post.side_effect = httpx.ConnectError("refused") - instance.close = MagicMock() + with patch("httpx.AsyncClient") as MockClient: + instance = MockClient.return_value.__aenter__.return_value + instance.post = AsyncMock(side_effect=httpx.ConnectError("refused")) resp = client.post("/v1/models/pull", json={"model": "foo"}) @@ -123,10 +125,9 @@ def test_delete_success(self): mock_resp.status_code = 200 mock_resp.raise_for_status = MagicMock() - with patch("httpx.Client") as MockClient: - instance = MockClient.return_value - instance.request.return_value = mock_resp - instance.close = MagicMock() + with patch("httpx.AsyncClient") as MockClient: + instance = MockClient.return_value.__aenter__.return_value + instance.request = AsyncMock(return_value=mock_resp) resp = client.delete("/v1/models/qwen3:0.6b") @@ -134,6 +135,11 @@ def test_delete_success(self): data = resp.json() assert data["status"] == "deleted" assert data["model"] == "qwen3:0.6b" + instance.request.assert_awaited_once_with( + "DELETE", + "/api/delete", + json={"name": "qwen3:0.6b"}, + ) # --------------------------------------------------------------------------- @@ -268,3 +274,19 @@ def test_models_empty_engine(self): assert resp.status_code == 200 # The endpoint returns whatever list_models() gives assert resp.json()["object"] == "list" + + def test_models_list_offloads_engine_list_models(self): + engine = _make_engine(models=["qwen3.5:4b"]) + app = create_app(engine, "qwen3.5:4b") + client = TestClient(app) + + with patch( + "openjarvis.server.routes.asyncio.to_thread", + new_callable=AsyncMock, + ) as mock_to_thread: + mock_to_thread.return_value = ["qwen3.5:4b"] + resp = client.get("/v1/models") + + assert resp.status_code == 200 + assert [m["id"] for m in resp.json()["data"]] == ["qwen3.5:4b"] + mock_to_thread.assert_awaited_once_with(engine.list_models) diff --git a/tests/server/test_speech_routes.py b/tests/server/test_speech_routes.py index 8e0dd3115..d5390367d 100644 --- a/tests/server/test_speech_routes.py +++ b/tests/server/test_speech_routes.py @@ -1,6 +1,6 @@ """Tests for speech API endpoints.""" -from unittest.mock import MagicMock +from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -56,6 +56,33 @@ def test_transcribe_endpoint(client, mock_speech_backend): assert data["duration_seconds"] == 1.5 +def test_transcribe_endpoint_offloads_backend_work(client, mock_speech_backend): + expected = TranscriptionResult( + text="Offloaded", + language="en", + confidence=0.9, + duration_seconds=1.0, + segments=[], + ) + + with patch( + "openjarvis.server.api_routes.asyncio.to_thread", + new_callable=AsyncMock, + ) as mock_to_thread: + mock_to_thread.return_value = expected + response = client.post( + "/v1/speech/transcribe", + files={"file": ("test.wav", b"fake audio data", "audio/wav")}, + ) + + assert response.status_code == 200 + mock_to_thread.assert_awaited_once() + args, kwargs = mock_to_thread.await_args + assert args == (mock_speech_backend.transcribe, b"fake audio data") + assert kwargs == {"format": "wav", "language": None} + assert response.json()["text"] == "Offloaded" + + def test_transcribe_endpoint_surfaces_backend_error(client, mock_speech_backend): mock_speech_backend.transcribe.side_effect = RuntimeError("missing cublas64_12.dll") diff --git a/tests/telemetry/test_aggregator.py b/tests/telemetry/test_aggregator.py index e386854c3..832875482 100644 --- a/tests/telemetry/test_aggregator.py +++ b/tests/telemetry/test_aggregator.py @@ -49,6 +49,21 @@ def _setup(tmp_path: Path, records: list[TelemetryRecord] | None = None): class TestTelemetryAggregator: + def test_uses_wal_with_normal_synchronous_and_busy_timeout( + self, + tmp_path: Path, + ) -> None: + agg = _setup(tmp_path) + + journal_mode = agg._conn.execute("PRAGMA journal_mode").fetchone()[0] + synchronous = agg._conn.execute("PRAGMA synchronous").fetchone()[0] + busy_timeout = agg._conn.execute("PRAGMA busy_timeout").fetchone()[0] + + assert journal_mode.lower() == "wal" + assert synchronous == 1 + assert busy_timeout == 5000 + agg.close() + def test_empty_db_summary(self, tmp_path: Path) -> None: agg = _setup(tmp_path) s = agg.summary() diff --git a/tests/telemetry/test_store.py b/tests/telemetry/test_store.py index c91ce264d..578f11512 100644 --- a/tests/telemetry/test_store.py +++ b/tests/telemetry/test_store.py @@ -3,6 +3,7 @@ from __future__ import annotations import time +from concurrent.futures import ThreadPoolExecutor from pathlib import Path from openjarvis.core.events import EventBus, EventType @@ -17,6 +18,35 @@ def test_creates_table(self, tmp_path: Path) -> None: assert rows == [] store.close() + def test_uses_wal_with_normal_synchronous(self, tmp_path: Path) -> None: + store = TelemetryStore(tmp_path / "test.db") + journal_mode = store._conn.execute("PRAGMA journal_mode").fetchone()[0] + synchronous = store._conn.execute("PRAGMA synchronous").fetchone()[0] + busy_timeout = store._conn.execute("PRAGMA busy_timeout").fetchone()[0] + + assert journal_mode.lower() == "wal" + assert synchronous == 1 + assert busy_timeout == 5000 + store.close() + + def test_concurrent_record_writes_are_serialized(self, tmp_path: Path) -> None: + store = TelemetryStore(tmp_path / "test.db") + + def write_one(i: int) -> None: + store.record( + TelemetryRecord( + timestamp=time.time(), + model_id=f"model-{i}", + engine="test", + ) + ) + + with ThreadPoolExecutor(max_workers=8) as pool: + list(pool.map(write_one, range(32))) + + assert len(store._fetchall()) == 32 + store.close() + def test_record_values(self, tmp_path: Path) -> None: store = TelemetryStore(tmp_path / "test.db") rec = TelemetryRecord(