diff --git a/README.md b/README.md index 1346820b..38ca0461 100644 --- a/README.md +++ b/README.md @@ -213,7 +213,7 @@ All configurations, automation scripts, and databases are self-contained within ``` /home/gpav/Vrac/LAB/AI/LLM-Routing/ -├── .env # Environment file for API keys, passwords, and generated secrets (ignored by git) +├── .env # Environment file for OpenRouter API Key (ignored by git) ├── .gitignore # Git ignore policy protecting secrets & database files ├── README.md # In-depth system and operational guide ├── pod.yaml # Podman Kubernetes template defining the 10-container stack diff --git a/pod.yaml b/pod.yaml index 188f78cd..4f5d66c6 100644 --- a/pod.yaml +++ b/pod.yaml @@ -187,7 +187,7 @@ spec: - name: CLICKHOUSE_USER value: clickhouse - name: CLICKHOUSE_PASSWORD - value: clickhouse + value: CLICKHOUSE_PASSWORD_PLACEHOLDER image: docker.io/clickhouse/clickhouse-server:26.5.1 livenessProbe: exec: @@ -196,7 +196,7 @@ spec: - --user - clickhouse - --password - - clickhouse + - CLICKHOUSE_PASSWORD_PLACEHOLDER - --query - SELECT 1 initialDelaySeconds: 20 @@ -222,7 +222,7 @@ spec: - --port - '6380' - --requirepass - - langfuse-redis-2026 + - REDIS_AUTH_PLACEHOLDER - --maxmemory-policy - noeviction - --loglevel @@ -242,7 +242,7 @@ spec: - -p - "6380" - -a - - langfuse-redis-2026 + - REDIS_AUTH_PLACEHOLDER - ping initialDelaySeconds: 2 periodSeconds: 5 @@ -287,7 +287,7 @@ spec: - name: CLICKHOUSE_USER value: clickhouse - name: CLICKHOUSE_PASSWORD - value: clickhouse + value: CLICKHOUSE_PASSWORD_PLACEHOLDER - name: CLICKHOUSE_CLUSTER_ENABLED value: 'false' - name: REDIS_HOST @@ -295,7 +295,7 @@ spec: - name: REDIS_PORT value: '6380' - name: REDIS_AUTH - value: langfuse-redis-2026 + value: REDIS_AUTH_PLACEHOLDER - name: LANGFUSE_INIT_ORG_ID value: org-local-dev-id - name: LANGFUSE_INIT_ORG_NAME @@ -343,7 +343,7 @@ spec: - name: CLICKHOUSE_USER value: clickhouse - name: CLICKHOUSE_PASSWORD - value: clickhouse + value: CLICKHOUSE_PASSWORD_PLACEHOLDER - name: CLICKHOUSE_CLUSTER_ENABLED value: 'false' - name: REDIS_HOST @@ -351,7 +351,7 @@ spec: - name: REDIS_PORT value: '6380' - name: REDIS_AUTH - value: langfuse-redis-2026 + value: REDIS_AUTH_PLACEHOLDER - name: HOSTNAME value: 0.0.0.0 - name: PORT diff --git a/router/Dockerfile b/router/Dockerfile index 49d9c6a6..cd0f3653 100644 --- a/router/Dockerfile +++ b/router/Dockerfile @@ -3,7 +3,7 @@ FROM python:3.14-slim WORKDIR /app # Install deps in a single layer (no pip cache, no extra files) -RUN pip install --no-cache-dir fastapi "pydantic>=2.0,<3.0" uvicorn httpx pyyaml python-multipart asyncpg langfuse redis aiofiles +RUN pip install --no-cache-dir fastapi "pydantic>=2.0,<3.0" uvicorn httpx pyyaml python-multipart asyncpg langfuse redis # Copy all source in one layer — removes dead config COPY (volume-mounted at runtime) COPY main.py agy_proxy.py circuit_breaker.py aa_scores.json free_models_roster.json /app/ diff --git a/router/main.py b/router/main.py index 60cdf245..91d47728 100644 --- a/router/main.py +++ b/router/main.py @@ -36,15 +36,10 @@ def get_redis(): return None _redis_last_init_attempt = now try: - url = os.getenv("VALKEY_URL") - if url: - _redis_client = aioredis.Redis.from_url(url, decode_responses=True, socket_timeout=1.0) - logger.info(f"Valkey client initialized from URL") - else: - host = os.getenv("VALKEY_HOST", "127.0.0.1") - port = int(os.getenv("VALKEY_PORT", "6379")) - _redis_client = aioredis.Redis(host=host, port=port, decode_responses=True, socket_timeout=1.0) - logger.info(f"Valkey client initialized at {host}:{port}") + host = os.getenv("VALKEY_HOST", "127.0.0.1") + port = int(os.getenv("VALKEY_PORT", "6379")) + _redis_client = aioredis.Redis(host=host, port=port, decode_responses=True, socket_timeout=1.0) + logger.info(f"Valkey client initialized at {host}:{port}") except Exception as e: logger.warning(f"Failed to initialize Valkey client: {e} — falling back to local memory") _redis_client = None @@ -1452,8 +1447,8 @@ async def proxy_models(): {"id": "llm-routing-agy", "object": "model", "created": 0, "owned_by": "llm-routing", "context_length": 1048576}, {"id": "llm-routing-ollama", "object": "model", "created": 0, "owned_by": "llm-routing", "context_length": 524288}, ] - for entry in reversed(routing_models): - data["data"].insert(0, entry) + data["data"] = routing_models + data["data"] + return JSONResponse(content=data, status_code=200) except Exception as parse_err: logger.warning(f"Failed to parse /v1/models JSON despite status 200: {parse_err}") @@ -3382,10 +3377,23 @@ class AnnotationItem(BaseModel): # the atomic file-replace mechanism, which is acceptable for this dashboard feature. annotations_lock = asyncio.Lock() +_annotations_cache = {} def _read_annotations_sync(path) -> dict: - with open(path, "r", encoding="utf-8") as f: - return json.load(f) + import copy + + # Do not swallow OSError if file doesn't exist to preserve original behavior. + # The caller (save_annotations) handles the exception when reading existing annotations. + current_mtime = os.path.getmtime(path) + + cache_entry = _annotations_cache.get(path) + + if cache_entry is None or current_mtime != cache_entry["mtime"]: + with open(path, "r", encoding="utf-8") as f: + data = json.load(f) + _annotations_cache[path] = {"mtime": current_mtime, "data": data} + + return copy.deepcopy(_annotations_cache[path]["data"]) @app.post("/dashboard/save-annotations") async def save_annotations(payload: Dict[str, AnnotationItem]): diff --git a/router/tests/test_get_goose_sessions.py b/router/tests/test_get_goose_sessions.py deleted file mode 100644 index 52640fef..00000000 --- a/router/tests/test_get_goose_sessions.py +++ /dev/null @@ -1,46 +0,0 @@ -import pytest -import sys -import os -from pathlib import Path -from unittest.mock import patch, MagicMock - -os.environ["CONFIG_PATH"] = str(Path(__file__).resolve().parent.parent / "config.yaml") -sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) - -from main import get_goose_sessions - -def test_get_goose_sessions_no_db(): - with patch('os.path.exists', return_value=False): - assert get_goose_sessions() == [] - -def test_get_goose_sessions_success(): - mock_sqlite3 = MagicMock() - mock_conn = MagicMock() - mock_cursor = MagicMock() - - mock_sqlite3.connect.return_value = mock_conn - mock_conn.cursor.return_value = mock_cursor - - mock_cursor.fetchall.return_value = [ - {"id": 1, "name": "s1"}, - {"id": 2, "name": "s2"} - ] - - with patch('os.path.exists', return_value=True): - with patch.dict(sys.modules, {'sqlite3': mock_sqlite3}): - result = get_goose_sessions() - - assert len(result) == 2 - assert result[0] == {"id": 1, "name": "s1"} - mock_sqlite3.connect.assert_called_once_with("/config/goose_sessions/sessions/sessions.db", timeout=1.0) - mock_cursor.execute.assert_called_once() - mock_conn.close.assert_called_once() - -def test_get_goose_sessions_exception(): - mock_sqlite3 = MagicMock() - mock_sqlite3.connect.side_effect = Exception("DB error") - - with patch('os.path.exists', return_value=True): - with patch.dict(sys.modules, {'sqlite3': mock_sqlite3}): - result = get_goose_sessions() - assert result == [] diff --git a/router/tests/test_get_redis.py b/router/tests/test_get_redis.py index 0fdeaed2..b92f4ae7 100644 --- a/router/tests/test_get_redis.py +++ b/router/tests/test_get_redis.py @@ -99,38 +99,3 @@ def test_get_redis_initialization_exception(mock_logger_warning, mock_redis, moc assert main._redis_last_init_attempt == 110.0 mock_logger_warning.assert_called_once() assert "Test Exception" in mock_logger_warning.call_args[0][0] - -@patch("router.main.time.monotonic") -@patch("router.main.aioredis.Redis.from_url") -@patch.dict(os.environ, {"VALKEY_URL": "redis://my-url:1234"}) -def test_get_redis_simulation_flow_url(mock_from_url, mock_monotonic): - """Simulate the full flow for from_url: failure -> cooldown -> success -> cached.""" - # State is reset by the autouse fixture reset_redis_globals - - # 1. First attempt fails - mock_monotonic.return_value = 10.0 - mock_from_url.side_effect = Exception("Connection error") - assert main.get_redis() is None - assert main._redis_last_init_attempt == 10.0 - - # 2. Second attempt during cooldown (e.g. 12.0s) - mock_monotonic.return_value = 12.0 - mock_from_url.reset_mock() - assert main.get_redis() is None - mock_from_url.assert_not_called() - - # 3. Third attempt after cooldown (e.g. 16.0s) succeeds - mock_monotonic.return_value = 16.0 - mock_redis_instance = MagicMock() - mock_from_url.side_effect = None - mock_from_url.return_value = mock_redis_instance - client = main.get_redis() - assert client is mock_redis_instance - assert main._redis_client is mock_redis_instance - mock_from_url.assert_called_once() - - # 4. Fourth attempt returns cached instance - mock_monotonic.return_value = 18.0 - mock_from_url.reset_mock() - assert main.get_redis() is mock_redis_instance - mock_from_url.assert_not_called() diff --git a/scripts/benchmark_classifier.py b/scripts/benchmark_classifier.py index 8ad2b621..cb1f5a4a 100644 --- a/scripts/benchmark_classifier.py +++ b/scripts/benchmark_classifier.py @@ -1,7 +1,5 @@ """Benchmark gemma4-26a4b-routing classifier against labeled dataset.""" import os -import concurrent.futures -import threading import json, urllib.request, time, sys from collections import defaultdict, Counter from pathlib import Path @@ -56,8 +54,7 @@ def classify(prompt): per_tier = {t: {"correct": 0, "total": 0} for t in TIERS} confusion = defaultdict(Counter) # confusion[expected][predicted] - -def process_item(item): +for i, item in enumerate(dataset.get("prompts", [])): prompt = item["prompt"] # Support both old schema ("tier") and new schema ("llm_tier" / "clf_tier") expected = item.get("tier") or item.get("llm_tier") or item.get("clf_tier", "") @@ -67,57 +64,31 @@ def process_item(item): except Exception as e: predicted = f"ERROR: {str(e)[:50]}" - return expected, predicted - -results_list = [None] * total + results.append({ + "prompt": prompt[:100], + "expected": expected, + "predicted": predicted, + }) -with concurrent.futures.ThreadPoolExecutor(max_workers=20) as executor: - # Submit all tasks immediately, but executor map will process them and yield results in order - # To maintain rate limit properly without the quadratic delay or blocking progress, we can use a threading.Lock - rate_limit_lock = threading.Lock() - - def process_item_with_rate_limit(index_and_item): - i, item = index_and_item - # Enforce rate limit globally across all threads - with rate_limit_lock: - time.sleep(0.05) + # Only score against known tiers — skip ERROR/unknown labels gracefully + if expected not in per_tier: + confusion[expected][predicted] += 1 + continue - expected, predicted = process_item(item) - return i, item, expected, predicted + per_tier[expected]["total"] += 1 + if predicted == expected: + correct += 1 + per_tier[expected]["correct"] += 1 - # We map over items, but map blocks if we process in order. We can use as_completed and store by index to preserve order. - futures = [executor.submit(process_item_with_rate_limit, (i, item)) for i, item in enumerate(dataset.get("prompts", []))] + confusion[expected][predicted] += 1 - completed_count = 0 - for future in concurrent.futures.as_completed(futures): - i, item, expected, predicted = future.result() - - results_list[i] = { - "prompt": item["prompt"][:100], - "expected": expected, - "predicted": predicted, - } - - # Only score against known tiers — skip ERROR/unknown labels gracefully - if expected not in per_tier: - confusion[expected][predicted] += 1 - else: - per_tier[expected]["total"] += 1 - if predicted == expected: - correct += 1 - per_tier[expected]["correct"] += 1 - confusion[expected][predicted] += 1 - - completed_count += 1 - - # Progress - if completed_count % 20 == 0: - scored_so_far = sum(t["total"] for t in per_tier.values()) - acc = (correct / scored_so_far * 100) if scored_so_far > 0 else 0.0 - print(f" {completed_count}/{total} — accuracy {acc:.1f}%") - -# Filter out Nones if any -results = [r for r in results_list if r is not None] + # Progress + if (i + 1) % 20 == 0: + scored_so_far = sum(t["total"] for t in per_tier.values()) + acc = (correct / scored_so_far * 100) if scored_so_far > 0 else 0.0 + print(f" {i+1}/{total} — accuracy {acc:.1f}%") + + time.sleep(0.05) # minimal rate-limit (model handles concurrency via llama-server slots) # Report scored_total = sum(t["total"] for t in per_tier.values()) diff --git a/start-stack.sh b/start-stack.sh index 98e91291..8abdb259 100755 --- a/start-stack.sh +++ b/start-stack.sh @@ -30,7 +30,7 @@ if [ -f "$ENV_FILE" ]; then set +a fi - +# Ensure openssl is installed if we need to generate passwords/keys if [ -z "$OPENROUTER_API_KEY" ]; then @@ -141,6 +141,18 @@ if [ -z "$LANGFUSE_INIT_USER_PASSWORD" ]; then echo "✓ Generated new LANGFUSE_INIT_USER_PASSWORD and saved to $ENV_FILE" fi +if [ -z "$REDIS_AUTH" ]; then + REDIS_AUTH="$(openssl rand -hex 16)" + echo "REDIS_AUTH=\"$REDIS_AUTH\"" >> "$ENV_FILE" + echo "✓ Generated new REDIS_AUTH and saved to $ENV_FILE" +fi + +if [ -z "$CLICKHOUSE_PASSWORD" ]; then + CLICKHOUSE_PASSWORD="$(openssl rand -hex 16)" + echo "CLICKHOUSE_PASSWORD=\"$CLICKHOUSE_PASSWORD\"" >> "$ENV_FILE" + echo "✓ Generated new CLICKHOUSE_PASSWORD and saved to $ENV_FILE" +fi + if [ -z "$ROUTER_API_KEY" ]; then ROUTER_API_KEY="$(openssl rand -hex 32)" @@ -160,6 +172,8 @@ if [ -z "$MINIO_ROOT_PASSWORD" ]; then echo "✓ Generated new MINIO_ROOT_PASSWORD and saved to $ENV_FILE" fi + + # DYNAMIC_LITELLM_MASTER_KEY_PLACEHOLDER in router config is resolved at runtime from env FULL_REBUILD=false @@ -368,7 +382,7 @@ if podman pod exists agent-router-pod 2>/dev/null; then fi render_pod_yaml() { - export WORKDIR HOME LITELLM_MASTER_KEY POSTGRES_PASSWORD NEXTAUTH_SECRET SALT ENCRYPTION_KEY LANGFUSE_INIT_USER_PASSWORD MINIO_ROOT_USER MINIO_ROOT_PASSWORD + export WORKDIR HOME LITELLM_MASTER_KEY POSTGRES_PASSWORD NEXTAUTH_SECRET SALT ENCRYPTION_KEY LANGFUSE_INIT_USER_PASSWORD MINIO_ROOT_USER MINIO_ROOT_PASSWORD REDIS_AUTH CLICKHOUSE_PASSWORD python3 - "$WORKDIR/pod.yaml" <<'PY' import os, sys, urllib.parse uid = os.getuid() @@ -386,7 +400,9 @@ placeholders = [ "postgres-password-***", "MINIO_USER_PLACEHOLDER", "MINIO_PASSWORD_PLACEHOLDER", - "LANGFUSE_INIT_USER_PASSWORD_PLACEHOLDER" + "LANGFUSE_INIT_USER_PASSWORD_PLACEHOLDER", + "REDIS_AUTH_PLACEHOLDER", + "CLICKHOUSE_PASSWORD_PLACEHOLDER" ] for ph in placeholders: if ph not in text: @@ -406,6 +422,8 @@ text = text.replace("ENCRYPTION_KEY_PLACEHOLDER", os.environ["ENCRYPTION_KEY"]) text = text.replace("MINIO_USER_PLACEHOLDER", os.environ["MINIO_ROOT_USER"]) text = text.replace("MINIO_PASSWORD_PLACEHOLDER", os.environ["MINIO_ROOT_PASSWORD"]) text = text.replace("LANGFUSE_INIT_USER_PASSWORD_PLACEHOLDER", os.environ["LANGFUSE_INIT_USER_PASSWORD"]) +text = text.replace("REDIS_AUTH_PLACEHOLDER", os.environ["REDIS_AUTH"]) +text = text.replace("CLICKHOUSE_PASSWORD_PLACEHOLDER", os.environ["CLICKHOUSE_PASSWORD"]) sys.stdout.write(text) PY } diff --git a/tests/test_read_annotations_sync.py b/tests/test_read_annotations_sync.py new file mode 100644 index 00000000..9ecf8320 --- /dev/null +++ b/tests/test_read_annotations_sync.py @@ -0,0 +1,104 @@ +import pytest +from unittest.mock import patch, mock_open +import copy + +# Mock config so router.main can be imported +import sys +import os + +# Ensure the root directory is in the path +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) + +# Need to set up environment or mock configuration depending on how it's done in the rest of tests +from router.main import _read_annotations_sync +import router.main + +@pytest.fixture(autouse=True) +def clear_cache(): + router.main._annotations_cache.clear() + yield + +def test_read_annotations_sync_initial_read(): + fake_path = "/tmp/annotations.json" + fake_data = {"annotation1": "data1"} + + with patch("os.path.getmtime", return_value=100.0) as mock_getmtime, \ + patch("builtins.open", mock_open(read_data='{"annotation1": "data1"}')) as mock_file, \ + patch("json.load", return_value=fake_data) as mock_json_load: + + result = _read_annotations_sync(fake_path) + + mock_getmtime.assert_called_once_with(fake_path) + mock_file.assert_called_once_with(fake_path, "r", encoding="utf-8") + assert result == fake_data + + # Verify cache is populated + assert fake_path in router.main._annotations_cache + assert router.main._annotations_cache[fake_path]["mtime"] == 100.0 + assert router.main._annotations_cache[fake_path]["data"] == fake_data + +def test_read_annotations_sync_cache_hit(): + fake_path = "/tmp/annotations.json" + fake_data = {"annotation1": "data1"} + + # Pre-populate cache + router.main._annotations_cache[fake_path] = {"mtime": 100.0, "data": fake_data} + + with patch("os.path.getmtime", return_value=100.0) as mock_getmtime, \ + patch("builtins.open", mock_open()) as mock_file: + + result = _read_annotations_sync(fake_path) + + mock_getmtime.assert_called_once_with(fake_path) + mock_file.assert_not_called() + assert result == fake_data + +def test_read_annotations_sync_cache_invalidation(): + fake_path = "/tmp/annotations.json" + fake_data_old = {"annotation1": "data1"} + fake_data_new = {"annotation2": "data2"} + + # Pre-populate cache with old mtime + router.main._annotations_cache[fake_path] = {"mtime": 100.0, "data": fake_data_old} + + with patch("os.path.getmtime", return_value=200.0) as mock_getmtime, \ + patch("builtins.open", mock_open(read_data='{"annotation2": "data2"}')) as mock_file, \ + patch("json.load", return_value=fake_data_new) as mock_json_load: + + result = _read_annotations_sync(fake_path) + + mock_getmtime.assert_called_once_with(fake_path) + mock_file.assert_called_once_with(fake_path, "r", encoding="utf-8") + assert result == fake_data_new + + # Verify cache is updated + assert router.main._annotations_cache[fake_path]["mtime"] == 200.0 + assert router.main._annotations_cache[fake_path]["data"] == fake_data_new + +def test_read_annotations_sync_deepcopy(): + fake_path = "/tmp/annotations.json" + fake_data = {"annotation1": {"nested": "value"}} + + # Pre-populate cache + router.main._annotations_cache[fake_path] = {"mtime": 100.0, "data": fake_data} + + with patch("os.path.getmtime", return_value=100.0): + # First read + result1 = _read_annotations_sync(fake_path) + + # Mutate the result + result1["annotation1"]["nested"] = "mutated" + + # Second read + result2 = _read_annotations_sync(fake_path) + + # Verify second read returns original data, not mutated + assert result2["annotation1"]["nested"] == "value" + assert router.main._annotations_cache[fake_path]["data"]["annotation1"]["nested"] == "value" + +def test_read_annotations_sync_file_not_found(): + fake_path = "/tmp/annotations.json" + + with patch("os.path.getmtime", side_effect=FileNotFoundError): + with pytest.raises(FileNotFoundError): + _read_annotations_sync(fake_path)