diff --git a/backend/app/adapters/anthropic/llm.py b/backend/app/adapters/anthropic/llm.py index 6225e03..b4558eb 100644 --- a/backend/app/adapters/anthropic/llm.py +++ b/backend/app/adapters/anthropic/llm.py @@ -221,7 +221,8 @@ class _RecommendationObjectParser: return None def _finish_object(self) -> RerankSelectionOutput: - assert self._object_start is not None + if self._object_start is None: + raise RecommenderOutputError("Rerank parser lost the object start position") object_text = self.complete_text[self._object_start : self._scan_index + 1] self._object_start = None try: diff --git a/backend/app/adapters/spotify/client.py b/backend/app/adapters/spotify/client.py index 5aee722..9791540 100644 --- a/backend/app/adapters/spotify/client.py +++ b/backend/app/adapters/spotify/client.py @@ -7,6 +7,7 @@ import httpx2 from app.adapters.spotify.auth import refresh_access_token from app.adapters.spotify.errors import ( SpotifyAuthenticationError, + SpotifyQuotaExhaustedError, SpotifyRateLimitedError, SpotifyRequestError, SpotifyUnavailableError, @@ -119,7 +120,7 @@ class SpotifyClient: params: dict[str, str | int] | None = None, json: dict[str, object] | None = None, ) -> httpx2.Response: - access_token = await self._access_token() + access_token, refresh_generation = await self._access_token() response = await self._send( method, path, @@ -127,18 +128,27 @@ class SpotifyClient: params=params, json=json, ) - response = await self._retry_once_if_unauthorized( + response, refresh_generation = await self._retry_once_if_unauthorized( response, method, path, - access_token, + refresh_generation, params=params, json=json, ) - response = await self._retry_once_if_rate_limited( + response, refresh_generation = await self._retry_once_if_rate_limited( response, method, path, + refresh_generation, + params=params, + json=json, + ) + response, _ = await self._retry_once_if_unauthorized( + response, + method, + path, + refresh_generation, params=params, json=json, ) @@ -154,34 +164,41 @@ class SpotifyClient: json: dict[str, object] | None, ) -> httpx2.Response: increment_spotify_calls() - return await self.http.request( - method, - f"{self.settings.spotify_api_base_url.rstrip('/')}{path}", - params=params, - json=json, - headers={"Authorization": f"Bearer {access_token}"}, - ) + try: + return await self.http.request( + method, + f"{self.settings.spotify_api_base_url.rstrip('/')}{path}", + params=params, + json=json, + headers={"Authorization": f"Bearer {access_token}"}, + ) + except httpx2.HTTPError as error: + raise SpotifyUnavailableError(504, "Spotify request failed") from error async def _retry_once_if_unauthorized( self, response: httpx2.Response, method: str, path: str, - access_token: str, + refresh_generation: int, *, params: dict[str, str | int] | None, json: dict[str, object] | None, - ) -> httpx2.Response: + ) -> tuple[httpx2.Response, int]: if response.status_code != 401: - return response + return response, refresh_generation - await self._refresh_if_current(access_token) - return await self._send( - method, - path, - self.session.tokens.access_token, - params=params, - json=json, + await self._refresh_if_current(refresh_generation) + access_token, retry_generation = self._token_snapshot() + return ( + await self._send( + method, + path, + access_token, + params=params, + json=json, + ), + retry_generation, ) async def _retry_once_if_rate_limited( @@ -189,12 +206,13 @@ class SpotifyClient: response: httpx2.Response, method: str, path: str, + refresh_generation: int, *, params: dict[str, str | int] | None, json: dict[str, object] | None, - ) -> httpx2.Response: + ) -> tuple[httpx2.Response, int]: if response.status_code != 429: - return response + return response, refresh_generation retry_after_seconds = _parse_retry_after(response) _, reason = _parse_error_details(response) @@ -204,15 +222,19 @@ class SpotifyClient: or retry_after_seconds is None or retry_after_seconds > self.settings.spotify_retry_after_cap_seconds ): - return response + return response, refresh_generation await asyncio.sleep(retry_after_seconds) - return await self._send( - method, - path, - self.session.tokens.access_token, - params=params, - json=json, + access_token, retry_generation = self._token_snapshot() + return ( + await self._send( + method, + path, + access_token, + params=params, + json=json, + ), + retry_generation, ) def _raise_for_error(self, response: httpx2.Response) -> httpx2.Response: @@ -223,26 +245,33 @@ class SpotifyClient: if response.status_code == 401: raise SpotifyAuthenticationError("Spotify rejected refreshed authentication") if response.status_code == 429: + if reason == "QUOTA_EXCEEDED": + raise SpotifyQuotaExhaustedError(_parse_retry_after(response), reason) raise SpotifyRateLimitedError(_parse_retry_after(response), reason) if response.status_code >= 500: raise SpotifyUnavailableError(response.status_code, message) raise SpotifyRequestError(response.status_code, message) - async def _access_token(self) -> str: - access_token = self.session.tokens.access_token - if self.session.tokens.is_expired: - await self._refresh_if_current(access_token) - return self.session.tokens.access_token + async def _access_token(self) -> tuple[str, int]: + tokens = self.session.tokens + refresh_generation = self.session.refresh_generation + if tokens.is_expired: + await self._refresh_if_current(refresh_generation) + return self._token_snapshot() - async def _refresh_if_current(self, access_token: str) -> None: + async def _refresh_if_current(self, refresh_generation: int) -> None: async with self.session.refresh_lock: - if self.session.tokens.access_token != access_token: + if self.session.refresh_generation != refresh_generation: return self.session.tokens = await refresh_access_token( self.http, client_id=self.settings.spotify_client_id, tokens=self.session.tokens, ) + self.session.refresh_generation += 1 + + def _token_snapshot(self) -> tuple[str, int]: + return self.session.tokens.access_token, self.session.refresh_generation def _parse_retry_after(response: httpx2.Response) -> float | None: diff --git a/backend/app/adapters/spotify/errors.py b/backend/app/adapters/spotify/errors.py index 9665573..74468e0 100644 --- a/backend/app/adapters/spotify/errors.py +++ b/backend/app/adapters/spotify/errors.py @@ -11,7 +11,7 @@ class SpotifyAuthenticationError(SpotifyError): """Spotify rejected authentication or token refresh.""" -class SpotifyRateLimitedError(SpotifyError, CatalogQuotaExhaustedError): +class SpotifyRateLimitedError(SpotifyError): """Spotify rate limited a request that could not be retried.""" def __init__(self, retry_after_seconds: float | None, reason: str | None = None) -> None: @@ -21,6 +21,10 @@ class SpotifyRateLimitedError(SpotifyError, CatalogQuotaExhaustedError): self.reason = reason +class SpotifyQuotaExhaustedError(SpotifyRateLimitedError, CatalogQuotaExhaustedError): + """Spotify rejected a request because the application quota is exhausted.""" + + class SpotifyUnavailableError(SpotifyError): """Spotify returned a server-side failure.""" diff --git a/backend/app/adapters/spotify/session.py b/backend/app/adapters/spotify/session.py index 613933b..e9d12a0 100644 --- a/backend/app/adapters/spotify/session.py +++ b/backend/app/adapters/spotify/session.py @@ -17,6 +17,7 @@ class SpotifySession: tokens: TokenSet account_id: str display_name: str + refresh_generation: int = 0 refresh_lock: asyncio.Lock = field(default_factory=asyncio.Lock) @@ -51,7 +52,15 @@ class PendingLogins: def add(self, state: str, code_verifier: str) -> None: """Store a PKCE verifier for a newly issued OAuth state.""" - self._entries[state] = (code_verifier, time.monotonic()) + now = time.monotonic() + expired_states = ( + pending_state + for pending_state, (_, created_at) in self._entries.items() + if now - created_at >= PENDING_LOGIN_LIFETIME_SECONDS + ) + for expired_state in tuple(expired_states): + del self._entries[expired_state] + self._entries[state] = (code_verifier, now) def pop(self, state: str) -> str | None: """Consume a verifier unless its OAuth state is unknown or expired.""" diff --git a/backend/app/api/recommendations.py b/backend/app/api/recommendations.py index 4b36afb..218328c 100644 --- a/backend/app/api/recommendations.py +++ b/backend/app/api/recommendations.py @@ -8,7 +8,7 @@ import structlog from fastapi import APIRouter, HTTPException, Request from fastapi.responses import StreamingResponse -from app.adapters.spotify.errors import SpotifyError +from app.adapters.spotify.errors import SpotifyAuthenticationError, SpotifyError from app.adapters.spotify.session import SpotifySession from app.api.routes import resolve_session from app.api.schemas import ( @@ -69,7 +69,6 @@ async def recommendations( return StreamingResponse( _stream_lines( - request, pipeline, resolved.session_id, request_id, @@ -101,6 +100,9 @@ async def create_playlist( "Music discovery selected by the listener.", ) await writer.add_tracks_to_playlist(playlist.id, payload.track_uris) + except SpotifyAuthenticationError as error: + structlog.get_logger().warning("playlist_write_failed", error_type=type(error).__name__) + raise HTTPException(status_code=401, detail="Spotify authentication expired") from error except (SpotifyError, ValueError) as error: structlog.get_logger().warning("playlist_write_failed", error_type=type(error).__name__) raise HTTPException(status_code=502, detail="Spotify playlist creation failed") from error @@ -108,7 +110,6 @@ async def create_playlist( async def _stream_lines( - request: Request, pipeline: RecommendationPipeline, session_id: str, request_id: str, @@ -127,8 +128,6 @@ async def _stream_lines( ) try: async for event in event_stream: - if await request.is_disconnected(): - return yield f"{_to_wire_event(event).model_dump_json()}\n" except asyncio.CancelledError: raise @@ -137,12 +136,11 @@ async def _stream_lines( "recommendation_stream_failed", error_type=type(error).__name__, ) - if not await request.is_disconnected(): - failure = ErrorEvent( - code="recommendation_failed", - message="Recommendation could not be completed.", - ) - yield f"{failure.model_dump_json()}\n" + failure = ErrorEvent( + code="recommendation_failed", + message="Recommendation could not be completed.", + ) + yield f"{failure.model_dump_json()}\n" finally: await event_stream.aclose() diff --git a/backend/app/config.py b/backend/app/config.py index 7f02ee5..a560614 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -41,6 +41,8 @@ class Settings(BaseSettings): llm_model: str = "claude-sonnet-5" intent_effort: str = "low" rerank_effort: str = "medium" + # Bound provider calls independently from the grounding deadline. + llm_timeout_seconds: float = 120.0 # Ceilings include adaptive thinking tokens, which is why they sit far # above the size of the structured output itself. intent_max_tokens: int = 16384 diff --git a/backend/app/main.py b/backend/app/main.py index ee1ea42..b5cc8dc 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -5,6 +5,7 @@ from contextlib import asynccontextmanager from pathlib import Path import httpx2 +import structlog from anthropic import AsyncAnthropic from fastapi import FastAPI from fastapi.staticfiles import StaticFiles @@ -12,6 +13,7 @@ from fastapi.staticfiles import StaticFiles from app.adapters.anthropic.llm import AnthropicRecommender from app.adapters.spotify.auth import TokenSet, refresh_access_token from app.adapters.spotify.client import SpotifyClient +from app.adapters.spotify.errors import SpotifyError from app.adapters.spotify.session import PendingLogins, SessionStore, SpotifySession from app.api.recommendations import router as recommendations_router from app.api.routes import router @@ -44,7 +46,8 @@ def create_app( application.state.settings = active_settings application.state.seed_session_id = None anthropic_client = AsyncAnthropic( - api_key=active_settings.anthropic_api_key or "unused-demo-key" + api_key=active_settings.anthropic_api_key or "unused-demo-key", + timeout=active_settings.llm_timeout_seconds, ) application.state.anthropic = anthropic_client application.state.recommendation_pipeline = RecommendationPipeline( @@ -60,11 +63,17 @@ def create_app( active_settings.app_mode is AppMode.LIVE and active_settings.spotify_seed_refresh_token ): - application.state.seed_session_id = await _install_seed_session( - http, - application.state.session_store, - active_settings, - ) + try: + application.state.seed_session_id = await _install_seed_session( + http, + application.state.session_store, + active_settings, + ) + except (SpotifyError, ValueError) as error: + structlog.get_logger().warning( + "seed_session_install_failed", + error_type=type(error).__name__, + ) try: yield finally: diff --git a/backend/app/pipeline/grounding.py b/backend/app/pipeline/grounding.py index 7ecfba6..0ad7163 100644 --- a/backend/app/pipeline/grounding.py +++ b/backend/app/pipeline/grounding.py @@ -7,8 +7,10 @@ from collections.abc import Callable from dataclasses import dataclass from enum import StrEnum +import httpx2 import structlog +from app.adapters.spotify.errors import SpotifyError from app.config import Settings from app.domain.matching import candidate_key, judge_candidate_match, track_key from app.domain.models import Familiarity, Track, TrackCandidate @@ -22,6 +24,7 @@ class ResolutionStatus(StrEnum): RESOLVED = "resolved" MISS = "miss" MISMATCH = "mismatch" + FAILED = "failed" QUOTA = "quota" @@ -32,6 +35,7 @@ class GroundingMetrics: attempted_count: int miss_count: int mismatch_guard_count: int + failed_count: int cache_hit_count: int did_reach_deadline: bool did_exhaust_quota: bool @@ -111,6 +115,7 @@ class Grounder: settings.resolution_cache_ttl_seconds, settings.resolution_cache_max_entries, ) + self._semaphore = asyncio.Semaphore(settings.grounding_concurrency) async def ground( self, @@ -119,6 +124,7 @@ class Grounder: known_track_ids: frozenset[str], familiarity: Familiarity, pool_target: int, + deadline_at: float, ) -> GroundingResult: """Resolve candidates until the target, deadline, or quota boundary.""" accepted: dict[int, Track] = {} @@ -127,18 +133,18 @@ class Grounder: metrics = _MutableMetrics() pending: dict[asyncio.Task[_ResolutionAttempt], int] = {} next_index = 0 - deadline_at = time.monotonic() + self.settings.request_deadline_seconds - semaphore = asyncio.Semaphore(self.settings.grounding_concurrency) try: while next_index < len(candidates) or pending: if len(accepted) >= pool_target: break + if time.monotonic() >= deadline_at: + metrics.did_reach_deadline = True + break next_index = self._launch_tasks( catalog, candidates, pending, next_index, - semaphore, ) if not pending: break @@ -183,12 +189,9 @@ class Grounder: candidates: tuple[TrackCandidate, ...], pending: dict[asyncio.Task[_ResolutionAttempt], int], next_index: int, - semaphore: asyncio.Semaphore, ) -> int: while next_index < len(candidates) and len(pending) < self.settings.grounding_concurrency: - task = asyncio.create_task( - self._resolve(catalog, next_index, candidates[next_index], semaphore) - ) + task = asyncio.create_task(self._resolve(catalog, next_index, candidates[next_index])) pending[task] = next_index next_index += 1 return next_index @@ -228,9 +231,8 @@ class Grounder: catalog: MusicCatalog, index: int, candidate: TrackCandidate, - semaphore: asyncio.Semaphore, ) -> _ResolutionAttempt: - async with semaphore: + async with self._semaphore: return await self._resolve_with_slot(catalog, index, candidate) async def _resolve_with_slot( @@ -265,6 +267,15 @@ class Grounder: bare_results = [] except CatalogQuotaExhaustedError: return _ResolutionAttempt(index=index, status=ResolutionStatus.QUOTA) + except (SpotifyError, httpx2.HTTPError) as error: + structlog.get_logger().info( + "candidate_unresolved", + title=candidate.title, + artist=candidate.artist, + status=ResolutionStatus.FAILED, + error_type=type(error).__name__, + ) + return _ResolutionAttempt(index=index, status=ResolutionStatus.FAILED) if matched_track is not None: self.cache.put(key, matched_track) @@ -289,6 +300,7 @@ class _MutableMetrics: attempted_count: int = 0 miss_count: int = 0 mismatch_guard_count: int = 0 + failed_count: int = 0 cache_hit_count: int = 0 did_reach_deadline: bool = False did_exhaust_quota: bool = False @@ -297,6 +309,7 @@ class _MutableMetrics: self.attempted_count += 1 self.miss_count += attempt.status is ResolutionStatus.MISS self.mismatch_guard_count += attempt.status is ResolutionStatus.MISMATCH + self.failed_count += attempt.status is ResolutionStatus.FAILED self.cache_hit_count += attempt.is_cache_hit self.did_exhaust_quota = self.did_exhaust_quota or attempt.status is ResolutionStatus.QUOTA @@ -305,6 +318,7 @@ class _MutableMetrics: attempted_count=self.attempted_count, miss_count=self.miss_count, mismatch_guard_count=self.mismatch_guard_count, + failed_count=self.failed_count, cache_hit_count=self.cache_hit_count, did_reach_deadline=self.did_reach_deadline, did_exhaust_quota=self.did_exhaust_quota, @@ -340,6 +354,7 @@ def _log_grounding(result: GroundingResult) -> None: attempted_count=metrics.attempted_count, miss_rate=metrics.miss_rate, mismatch_guard_rate=metrics.mismatch_guard_rate, + failed_count=metrics.failed_count, cache_hits=metrics.cache_hit_count, deadline_reached=metrics.did_reach_deadline, quota_exhausted=metrics.did_exhaust_quota, diff --git a/backend/app/pipeline/orchestrator.py b/backend/app/pipeline/orchestrator.py index f9380b4..fc20eac 100644 --- a/backend/app/pipeline/orchestrator.py +++ b/backend/app/pipeline/orchestrator.py @@ -7,6 +7,7 @@ from contextlib import aclosing import structlog +from app.adapters.spotify.errors import SpotifyError from app.config import Settings from app.domain.models import ( CompressedTasteProfile, @@ -27,7 +28,12 @@ from app.pipeline.event import ( PipelineWarningEvent, ) from app.pipeline.grounding import Grounder -from app.ports.protocols import MusicCatalog, Recommender, RecommenderOutputError +from app.ports.protocols import ( + CatalogQuotaExhaustedError, + MusicCatalog, + Recommender, + RecommenderOutputError, +) RERANK_FALLBACK_CODE = "rerank_fallback" RERANK_FALLBACK_MESSAGE = "Ranking output was invalid, so grounded results are shown instead." @@ -115,21 +121,43 @@ class RecommendationPipeline: ) -> AsyncGenerator[PipelineEvent]: """Yield ordered events for one recommendation request.""" started_at = time.monotonic() - taste = await self.taste_cache.get(session_id, catalog) - intent = await self.recommender.create_intent( - query, - history, - previous_recommendations, - taste.text, - self.settings.candidate_count, - ) + deadline_at = started_at + self.settings.request_deadline_seconds + try: + taste = await self.taste_cache.get(session_id, catalog) + intent = await self.recommender.create_intent( + query, + history, + previous_recommendations, + taste.text, + self.settings.candidate_count, + ) + except RecommenderOutputError: + yield PipelineErrorEvent( + code="intent_failed", + message="Recommendation intent could not be generated from the model response.", + ) + return + except CatalogQuotaExhaustedError: + yield PipelineErrorEvent( + code="quota_exhausted", + message=( + "Spotify request quota was exhausted before recommendations could be prepared." + ), + ) + return + except SpotifyError: + yield PipelineErrorEvent( + code="spotify_unavailable", + message="Spotify was unavailable while preparing recommendations.", + ) + return yield PipelineMetadataEvent( request_id=request_id, intent_summary=intent.intent_summary, candidate_count=len(intent.candidates), ) - pool = await self._grounded_pool(session_id, catalog, intent, taste) + pool = await self._grounded_pool(session_id, catalog, intent, taste, deadline_at) if not pool: yield PipelineErrorEvent( code="no_grounded_results", @@ -160,6 +188,7 @@ class RecommendationPipeline: catalog: MusicCatalog, intent: Intent, taste: CompressedTasteProfile, + deadline_at: float, ) -> tuple[Track, ...]: """Reuse the session's pool on refinement, otherwise ground anew.""" if intent.is_refinement: @@ -172,6 +201,7 @@ class RecommendationPipeline: taste.known_track_ids, intent.familiarity, self.settings.rerank_count + self.settings.rerank_pool_buffer, + deadline_at, ) if result.tracks: self.last_pools[session_id] = result.tracks @@ -290,19 +320,31 @@ class _TasteProfileCache: return profile async def _fetch(self, catalog: MusicCatalog) -> CompressedTasteProfile: - short_artists, long_artists, short_tracks, long_tracks, saved_tracks = await asyncio.gather( - catalog.fetch_top_artists("short_term", self.settings.top_items_limit), - catalog.fetch_top_artists("long_term", self.settings.top_items_limit), - catalog.fetch_top_tracks("short_term", self.settings.top_items_limit), - catalog.fetch_top_tracks("long_term", self.settings.top_items_limit), - catalog.fetch_saved_tracks(self.settings.saved_tracks_limit), - ) + try: + async with asyncio.TaskGroup() as task_group: + short_artists_task = task_group.create_task( + catalog.fetch_top_artists("short_term", self.settings.top_items_limit) + ) + long_artists_task = task_group.create_task( + catalog.fetch_top_artists("long_term", self.settings.top_items_limit) + ) + short_tracks_task = task_group.create_task( + catalog.fetch_top_tracks("short_term", self.settings.top_items_limit) + ) + long_tracks_task = task_group.create_task( + catalog.fetch_top_tracks("long_term", self.settings.top_items_limit) + ) + saved_tracks_task = task_group.create_task( + catalog.fetch_saved_tracks(self.settings.saved_tracks_limit) + ) + except ExceptionGroup as errors: + raise errors.exceptions[0] from None return compress_taste_profile( TasteProfile( - short_term_artists=tuple(short_artists), - long_term_artists=tuple(long_artists), - short_term_tracks=tuple(short_tracks), - long_term_tracks=tuple(long_tracks), - saved_tracks=tuple(saved_tracks), + short_term_artists=tuple(short_artists_task.result()), + long_term_artists=tuple(long_artists_task.result()), + short_term_tracks=tuple(short_tracks_task.result()), + long_term_tracks=tuple(long_tracks_task.result()), + saved_tracks=tuple(saved_tracks_task.result()), ) ) diff --git a/backend/tests/test_anthropic_llm.py b/backend/tests/test_anthropic_llm.py new file mode 100644 index 0000000..27d3a61 --- /dev/null +++ b/backend/tests/test_anthropic_llm.py @@ -0,0 +1,13 @@ +"""Focused tests for Anthropic response parsing.""" + +import pytest + +from app.adapters.anthropic.llm import _RecommendationObjectParser +from app.ports.protocols import RecommenderOutputError + + +def test_parser_missing_object_start_raises_typed_output_error() -> None: + parser = _RecommendationObjectParser() + + with pytest.raises(RecommenderOutputError, match="object start position"): + parser._finish_object() diff --git a/backend/tests/test_auth_routes.py b/backend/tests/test_auth_routes.py index c1c067f..9484f37 100644 --- a/backend/tests/test_auth_routes.py +++ b/backend/tests/test_auth_routes.py @@ -113,6 +113,29 @@ def test_seed_session_authenticates_requests_without_a_cookie() -> None: assert response.json() == {"display_name": "Seed Listener"} +def test_seed_session_failure_keeps_application_serving() -> None: + async def spotify_handler(request: httpx2.Request) -> httpx2.Response: + return httpx2.Response(400) + + app = create_app( + application_settings=Settings( + app_mode=AppMode.LIVE, + spotify_client_id="client-id", + anthropic_api_key="test-key", + spotify_seed_refresh_token="seed-refresh", + ), + http_transport=httpx2.MockTransport(spotify_handler), + ) + with TestClient(app, follow_redirects=False) as client: + health_response = client.get("/api/health") + login_response = client.get("/api/auth/login") + + assert app.state.seed_session_id is None + + assert health_response.status_code == 200 + assert login_response.status_code == 307 + + def _live_settings() -> Settings: return Settings( app_mode=AppMode.LIVE, diff --git a/backend/tests/test_grounding.py b/backend/tests/test_grounding.py index e5b896e..757e7ea 100644 --- a/backend/tests/test_grounding.py +++ b/backend/tests/test_grounding.py @@ -1,8 +1,14 @@ """Deterministic tests for bounded Spotify grounding.""" import asyncio +import time from collections.abc import Awaitable, Callable +from app.adapters.spotify.errors import ( + SpotifyQuotaExhaustedError, + SpotifyRateLimitedError, + SpotifyUnavailableError, +) from app.config import Settings from app.domain.models import Familiarity, Track, TrackCandidate from app.pipeline.grounding import Grounder @@ -47,7 +53,14 @@ def test_early_stop_honors_pool_target() -> None: grounder = Grounder(_settings(grounding_concurrency=2)) candidates = tuple(TrackCandidate(f"track-{index}", "Artist") for index in range(6)) - result = await grounder.ground(catalog, candidates, frozenset(), Familiarity.MIX, 2) + result = await grounder.ground( + catalog, + candidates, + frozenset(), + Familiarity.MIX, + 2, + _deadline(), + ) assert len(result.tracks) == 2 assert len(catalog.search_queries) == 2 @@ -74,6 +87,7 @@ def test_miss_and_mismatch_are_counted_separately() -> None: frozenset(), Familiarity.MIX, 2, + _deadline(), ) assert result.metrics.miss_count == 1 @@ -92,9 +106,23 @@ def test_resolution_cache_hit_skips_catalog() -> None: grounder = Grounder(_settings()) candidates = (TrackCandidate("Cached Song", "Artist"),) - await grounder.ground(catalog, candidates, frozenset(), Familiarity.MIX, 1) + await grounder.ground( + catalog, + candidates, + frozenset(), + Familiarity.MIX, + 1, + _deadline(), + ) first_call_count = len(catalog.search_queries) - second = await grounder.ground(catalog, candidates, frozenset(), Familiarity.MIX, 1) + second = await grounder.ground( + catalog, + candidates, + frozenset(), + Familiarity.MIX, + 1, + _deadline(), + ) assert len(catalog.search_queries) == first_call_count assert second.metrics.cache_hit_count == 1 @@ -124,6 +152,7 @@ def test_deadline_returns_resolved_partial_pool() -> None: frozenset(), Familiarity.MIX, 2, + time.monotonic() + settings.request_deadline_seconds, ) assert [track.id for track in result.tracks] == ["fast"] @@ -132,6 +161,127 @@ def test_deadline_returns_resolved_partial_pool() -> None: asyncio.run(run()) +def test_spotify_failure_is_counted_and_remaining_candidates_continue() -> None: + async def run() -> None: + async def search(query: str) -> list[Track]: + if "Failing Song" in query: + raise SpotifyUnavailableError(503) + title = query.split('track:"', 1)[1].split('"', 1)[0] + return [_track(title.lower().replace(" ", "-"), title, "Artist")] + + catalog = FakeCatalog(search) + candidates = ( + TrackCandidate("Failing Song", "Artist"), + TrackCandidate("First Good Song", "Artist"), + TrackCandidate("Second Good Song", "Artist"), + ) + + result = await Grounder(_settings()).ground( + catalog, + candidates, + frozenset(), + Familiarity.MIX, + 2, + _deadline(), + ) + + assert [track.title for track in result.tracks] == [ + "First Good Song", + "Second Good Song", + ] + assert result.metrics.failed_count == 1 + assert result.metrics.attempted_count == 3 + assert not result.metrics.did_exhaust_quota + + asyncio.run(run()) + + +def test_plain_rate_limit_continues_but_quota_exhaustion_stops_fanout() -> None: + async def run() -> None: + async def plain_rate_limited_search(query: str) -> list[Track]: + if "Rate Limited" in query: + raise SpotifyRateLimitedError(6.0) + return [_track("found", "Found Song", "Artist")] + + plain_catalog = FakeCatalog(plain_rate_limited_search) + candidates = ( + TrackCandidate("Rate Limited", "Artist"), + TrackCandidate("Found Song", "Artist"), + ) + plain_result = await Grounder(_settings()).ground( + plain_catalog, + candidates, + frozenset(), + Familiarity.MIX, + 1, + _deadline(), + ) + + async def quota_search(query: str) -> list[Track]: + raise SpotifyQuotaExhaustedError(0.0, "QUOTA_EXCEEDED") + + quota_catalog = FakeCatalog(quota_search) + quota_result = await Grounder(_settings()).ground( + quota_catalog, + candidates, + frozenset(), + Familiarity.MIX, + 1, + _deadline(), + ) + + assert [track.id for track in plain_result.tracks] == ["found"] + assert plain_result.metrics.failed_count == 1 + assert not plain_result.metrics.did_exhaust_quota + assert quota_result.tracks == () + assert quota_result.metrics.did_exhaust_quota + assert len(quota_catalog.search_queries) == 1 + + asyncio.run(run()) + + +def test_grounding_concurrency_is_shared_across_requests() -> None: + async def run() -> None: + active_searches = 0 + maximum_active_searches = 0 + + async def search(query: str) -> list[Track]: + nonlocal active_searches, maximum_active_searches + active_searches += 1 + maximum_active_searches = max(maximum_active_searches, active_searches) + await asyncio.sleep(0.01) + active_searches -= 1 + title = query.split('track:"', 1)[1].split('"', 1)[0] + return [_track(title, title, "Artist")] + + catalog = FakeCatalog(search) + grounder = Grounder(_settings(grounding_concurrency=2)) + candidates = tuple(TrackCandidate(f"Song {index}", "Artist") for index in range(2)) + + await asyncio.gather( + grounder.ground( + catalog, + candidates, + frozenset(), + Familiarity.MIX, + 2, + _deadline(), + ), + grounder.ground( + catalog, + candidates, + frozenset(), + Familiarity.MIX, + 2, + _deadline(), + ), + ) + + assert maximum_active_searches == 2 + + asyncio.run(run()) + + def _settings(**overrides: object) -> Settings: values: dict[str, object] = { "grounding_concurrency": 1, @@ -141,6 +291,10 @@ def _settings(**overrides: object) -> Settings: return Settings.model_validate(values) +def _deadline() -> float: + return time.monotonic() + 1.0 + + def _track(track_id: str, title: str, artist: str) -> Track: return Track( id=track_id, diff --git a/backend/tests/test_health.py b/backend/tests/test_health.py index a876295..b4e9b61 100644 --- a/backend/tests/test_health.py +++ b/backend/tests/test_health.py @@ -1,7 +1,11 @@ """Smoke test for the application factory.""" +from unittest.mock import AsyncMock, Mock + +import pytest from fastapi.testclient import TestClient +from app.config import Settings from app.main import create_app @@ -10,3 +14,16 @@ def test_health_reports_mode() -> None: response = client.get("/api/health") assert response.status_code == 200 assert response.json()["mode"] in ("live", "demo") + + +def test_anthropic_client_uses_configured_timeout(monkeypatch: pytest.MonkeyPatch) -> None: + anthropic_client = Mock() + anthropic_client.close = AsyncMock() + constructor = Mock(return_value=anthropic_client) + monkeypatch.setattr("app.main.AsyncAnthropic", constructor) + + with TestClient(create_app(Settings(llm_timeout_seconds=42.0))) as client: + response = client.get("/api/health") + + assert response.status_code == 200 + constructor.assert_called_once_with(api_key="unused-demo-key", timeout=42.0) diff --git a/backend/tests/test_orchestrator.py b/backend/tests/test_orchestrator.py index 4f2fc6f..4dce350 100644 --- a/backend/tests/test_orchestrator.py +++ b/backend/tests/test_orchestrator.py @@ -3,6 +3,7 @@ import asyncio from collections.abc import AsyncGenerator +from app.adapters.spotify.errors import SpotifyQuotaExhaustedError, SpotifyUnavailableError from app.config import Settings from app.domain.models import ( ConversationTurn, @@ -13,7 +14,7 @@ from app.domain.models import ( Track, TrackCandidate, ) -from app.pipeline.event import PipelineEvent, PipelineTrackEvent +from app.pipeline.event import PipelineErrorEvent, PipelineEvent, PipelineTrackEvent from app.pipeline.orchestrator import RecommendationPipeline from app.ports.protocols import RecommenderOutputError, TimeRange @@ -61,11 +62,15 @@ class FakeRecommender: intents: list[Intent], selection_ids: tuple[str, ...] = (), failure_count: int = 0, + intent_error: Exception | None = None, + intent_delay_seconds: float = 0.0, ) -> None: """Store deterministic outputs for successive calls.""" self.intents = intents self.selection_ids = selection_ids self.failure_count = failure_count + self.intent_error = intent_error + self.intent_delay_seconds = intent_delay_seconds self.rerank_call_count = 0 async def create_intent( @@ -77,6 +82,9 @@ class FakeRecommender: candidate_count: int, ) -> Intent: """Return the next fixed intent.""" + await asyncio.sleep(self.intent_delay_seconds) + if self.intent_error is not None: + raise self.intent_error return self.intents.pop(0) async def stream_rerank( @@ -115,6 +123,108 @@ def test_event_order_and_rerank_ids_stay_inside_grounded_pool() -> None: asyncio.run(run()) +def test_intent_stage_failures_yield_one_typed_error_event() -> None: + async def run() -> None: + cases = ( + (RecommenderOutputError("invalid intent"), "intent_failed"), + ( + SpotifyQuotaExhaustedError(0.0, "QUOTA_EXCEEDED"), + "quota_exhausted", + ), + (SpotifyUnavailableError(503), "spotify_unavailable"), + ) + for error, expected_code in cases: + catalog = FakeCatalog(()) + recommender = FakeRecommender([], intent_error=error) + + events = await _run_pipeline(catalog, recommender) + + assert len(events) == 1 + assert events[0].type == "error" + assert events[0].code == expected_code + + asyncio.run(run()) + + +def test_taste_failure_cancels_sibling_fetches() -> None: + class FailingTasteCatalog(FakeCatalog): + """Fail one taste request after all sibling requests have started.""" + + def __init__(self) -> None: + super().__init__(()) + self.started_count = 0 + self.cancelled_count = 0 + self.all_started = asyncio.Event() + self.never_finishes = asyncio.Event() + + async def fetch_top_artists(self, time_range: TimeRange, limit: int) -> list[str]: + await self._wait_for_all_fetches() + await self._wait_until_cancelled() + return [] + + async def fetch_top_tracks(self, time_range: TimeRange, limit: int) -> list[Track]: + await self._wait_for_all_fetches() + await self._wait_until_cancelled() + return [] + + async def fetch_saved_tracks(self, limit: int) -> list[Track]: + await self._wait_for_all_fetches() + raise SpotifyUnavailableError(503) + + async def _wait_for_all_fetches(self) -> None: + self.started_count += 1 + if self.started_count == 5: + self.all_started.set() + await self.all_started.wait() + + async def _wait_until_cancelled(self) -> None: + try: + await self.never_finishes.wait() + except asyncio.CancelledError: + self.cancelled_count += 1 + raise + + async def run() -> None: + catalog = FailingTasteCatalog() + events = await _run_pipeline(catalog, FakeRecommender([])) + + assert len(events) == 1 + assert events[0].type == "error" + assert events[0].code == "spotify_unavailable" + assert catalog.cancelled_count == 4 + + asyncio.run(run()) + + +def test_grounding_deadline_starts_before_intent_generation() -> None: + async def run() -> None: + track = _track("found", "Found Song") + catalog = FakeCatalog((track,)) + recommender = FakeRecommender( + [_intent(track)], + intent_delay_seconds=0.02, + ) + pipeline = RecommendationPipeline( + recommender, + Settings( + rerank_count=1, + rerank_pool_buffer=0, + grounding_floor=1, + grounding_concurrency=1, + request_deadline_seconds=0.01, + ), + ) + + events = await _collect(pipeline, catalog, "query") + + assert [event.type for event in events] == ["metadata", "error"] + assert isinstance(events[-1], PipelineErrorEvent) + assert events[-1].code == "no_grounded_results" + assert catalog.search_call_count == 0 + + asyncio.run(run()) + + def test_rerank_fallback_warns_then_streams_grounded_order() -> None: async def run() -> None: first = _track("first", "First Song") diff --git a/backend/tests/test_recommendation_api.py b/backend/tests/test_recommendation_api.py index f5e83ce..1920454 100644 --- a/backend/tests/test_recommendation_api.py +++ b/backend/tests/test_recommendation_api.py @@ -3,10 +3,12 @@ import time from collections.abc import AsyncGenerator +import pytest from fastapi.testclient import TestClient from pydantic import TypeAdapter from app.adapters.spotify.auth import TokenSet +from app.adapters.spotify.errors import SpotifyAuthenticationError from app.adapters.spotify.session import SessionStore, SpotifySession from app.api.routes import SESSION_COOKIE_NAME from app.api.schemas import StreamEvent @@ -37,13 +39,16 @@ class FakePipeline: class FakePlaylistWriter: """Capture playlist writes without external calls.""" - def __init__(self) -> None: + def __init__(self, error: Exception | None = None) -> None: """Create an empty write trace.""" + self.error = error self.name: str | None = None self.track_uris: list[str] = [] async def create_playlist(self, name: str, description: str) -> CreatedPlaylist: """Record the prefixed name and return a stable playlist.""" + if self.error is not None: + raise self.error self.name = name return CreatedPlaylist("playlist", "https://open.spotify.com/playlist/playlist") @@ -52,7 +57,13 @@ class FakePlaylistWriter: self.track_uris = track_uris -def test_recommendations_stream_lines_validate_against_frozen_schemas() -> None: +def test_recommendations_stream_lines_validate_against_frozen_schemas( + monkeypatch: pytest.MonkeyPatch, +) -> None: + async def fail_if_polled(request: object) -> bool: + raise AssertionError("Request disconnect state must not be polled") + + monkeypatch.setattr("starlette.requests.Request.is_disconnected", fail_if_polled) app = create_app() with TestClient(app) as client: _authenticate(client, session_store=app.state.session_store) @@ -103,6 +114,26 @@ def test_playlist_endpoint_prefixes_name_and_adds_tracks() -> None: assert writer.track_uris == ["spotify:track:track"] +def test_playlist_authentication_failure_signals_relogin() -> None: + app = create_app() + writer = FakePlaylistWriter(SpotifyAuthenticationError("expired")) + with TestClient(app) as client: + _authenticate(client, session_store=app.state.session_store) + app.state.spotify_client_factory = lambda session: writer + + response = client.post( + "/api/playlists", + json={ + "schema_version": 1, + "name": "Night drive", + "track_uris": ["spotify:track:track"], + }, + ) + + assert response.status_code == 401 + assert response.json() == {"detail": "Spotify authentication expired"} + + def _authenticate(client: TestClient, session_store: SessionStore) -> None: session_id = session_store.create( SpotifySession( diff --git a/backend/tests/test_spotify_auth.py b/backend/tests/test_spotify_auth.py index c9f0565..8a9a360 100644 --- a/backend/tests/test_spotify_auth.py +++ b/backend/tests/test_spotify_auth.py @@ -6,12 +6,14 @@ import hashlib import time import httpx2 +import pytest from app.adapters.spotify.auth import ( TokenSet, derive_code_challenge, refresh_access_token, ) +from app.adapters.spotify.session import PendingLogins def test_code_challenge_is_unpadded_base64url_sha256() -> None: @@ -43,3 +45,18 @@ def test_refresh_keeps_existing_refresh_token_when_omitted() -> None: def test_token_expiry_uses_sixty_second_skew() -> None: assert TokenSet("access", "refresh", time.monotonic() + 59).is_expired assert not TokenSet("access", "refresh", time.monotonic() + 61).is_expired + + +def test_pending_login_add_sweeps_expired_entries(monkeypatch: pytest.MonkeyPatch) -> None: + current_time = 0.0 + monkeypatch.setattr( + "app.adapters.spotify.session.time.monotonic", + lambda: current_time, + ) + pending_logins = PendingLogins() + pending_logins.add("expired", "old-verifier") + + current_time = 601.0 + pending_logins.add("current", "new-verifier") + + assert set(pending_logins._entries) == {"current"} diff --git a/backend/tests/test_spotify_client.py b/backend/tests/test_spotify_client.py index 4840673..0d54ada 100644 --- a/backend/tests/test_spotify_client.py +++ b/backend/tests/test_spotify_client.py @@ -9,10 +9,16 @@ import pytest from app.adapters.spotify.auth import TokenSet from app.adapters.spotify.client import SpotifyClient -from app.adapters.spotify.errors import SpotifyRateLimitedError, SpotifyRequestError +from app.adapters.spotify.errors import ( + SpotifyQuotaExhaustedError, + SpotifyRateLimitedError, + SpotifyRequestError, + SpotifyUnavailableError, +) from app.adapters.spotify.session import SpotifySession from app.config import Settings from app.domain.models import Track +from app.ports.protocols import CatalogQuotaExhaustedError TransportHandler = Callable[[httpx2.Request], Coroutine[None, None, httpx2.Response]] @@ -44,19 +50,19 @@ def test_unauthorized_response_refreshes_once_and_returns_result() -> None: def test_concurrent_unauthorized_responses_share_one_refresh() -> None: async def run() -> None: token_calls = 0 - old_api_calls = 0 - both_old_requests_arrived = asyncio.Event() + api_calls = 0 + both_initial_requests_arrived = asyncio.Event() async def handler(request: httpx2.Request) -> httpx2.Response: - nonlocal old_api_calls, token_calls + nonlocal api_calls, token_calls if request.url.host == "accounts.spotify.com": token_calls += 1 - return _token_response() - if request.headers["Authorization"] == "Bearer old-access": - old_api_calls += 1 - if old_api_calls == 2: - both_old_requests_arrived.set() - await asyncio.wait_for(both_old_requests_arrived.wait(), timeout=1) + return _token_response("old-access") + api_calls += 1 + if api_calls <= 2: + if api_calls == 2: + both_initial_requests_arrived.set() + await asyncio.wait_for(both_initial_requests_arrived.wait(), timeout=1) return httpx2.Response(401) return httpx2.Response(200, json=_search_payload()) @@ -69,6 +75,7 @@ def test_concurrent_unauthorized_responses_share_one_refresh() -> None: assert first == second assert token_calls == 1 + assert client.session.refresh_generation == 1 asyncio.run(run()) @@ -105,6 +112,7 @@ def test_get_rate_limit_above_cap_raises_without_retry() -> None: await _search_with_handler(handler) assert error.value.retry_after_seconds == 6 + assert not isinstance(error.value, CatalogQuotaExhaustedError) assert api_calls == 1 asyncio.run(run()) @@ -136,10 +144,11 @@ def test_quota_exhaustion_raises_without_retry_or_sleep( ) monkeypatch.setattr("app.adapters.spotify.client.asyncio.sleep", fake_sleep) - with pytest.raises(SpotifyRateLimitedError) as error: + with pytest.raises(SpotifyQuotaExhaustedError) as error: await _search_with_handler(handler) assert error.value.reason == "QUOTA_EXCEEDED" + assert isinstance(error.value, CatalogQuotaExhaustedError) assert api_calls == 1 assert sleep_calls == 0 @@ -172,6 +181,46 @@ def test_authentication_then_rate_limit_retries_each_policy_once() -> None: asyncio.run(run()) +def test_rate_limit_retry_landing_on_unauthorized_refreshes_once() -> None: + async def run() -> None: + token_calls = 0 + api_calls = 0 + + async def handler(request: httpx2.Request) -> httpx2.Response: + nonlocal api_calls, token_calls + if request.url.host == "accounts.spotify.com": + token_calls += 1 + return _token_response() + api_calls += 1 + if api_calls == 1: + return httpx2.Response(429, headers={"Retry-After": "0"}) + if api_calls == 2: + return httpx2.Response(401) + return httpx2.Response(200, json=_search_payload()) + + tracks = await _search_with_handler(handler) + + assert len(tracks) == 1 + assert token_calls == 1 + assert api_calls == 3 + + asyncio.run(run()) + + +def test_transport_error_becomes_spotify_unavailable() -> None: + async def run() -> None: + async def handler(request: httpx2.Request) -> httpx2.Response: + raise httpx2.ConnectError("connection failed", request=request) + + with pytest.raises(SpotifyUnavailableError) as error: + await _search_with_handler(handler) + + assert error.value.status_code == 504 + assert str(error.value) == "Spotify request failed" + + asyncio.run(run()) + + def test_request_error_includes_parsed_spotify_message() -> None: async def run() -> None: async def handler(request: httpx2.Request) -> httpx2.Response: @@ -267,8 +316,8 @@ def _client(http: httpx2.AsyncClient) -> SpotifyClient: return SpotifyClient(http, session, Settings(spotify_client_id="client")) -def _token_response() -> httpx2.Response: - return httpx2.Response(200, json={"access_token": "new-access", "expires_in": 3600}) +def _token_response(access_token: str = "new-access") -> httpx2.Response: + return httpx2.Response(200, json={"access_token": access_token, "expires_in": 3600}) def _search_payload() -> dict[str, object]: diff --git a/docs/logboek.md b/docs/logboek.md index e0782b2..96858ea 100644 --- a/docs/logboek.md +++ b/docs/logboek.md @@ -318,3 +318,29 @@ Waarom: Wat ik heb laten vallen of uitgesteld: - Verdere styling; de tijd gaat naar eval, demo mode en de README. + +### Backend hardening + +Wat ik deed: + +- Een hardening pass over de backend randgevallen, deels gevonden via een + adversarial review: fouten per candidate ingedamd zodat 1 kapotte + candidate nooit de hele request breekt, quota exhaustion apart herkend + van gewone 429's, token refresh single-flight per sessie-generatie, + typed errors voor de intent stap, en de seed-sessie kan de boot niet + meer laten crashen. +- De request deadline start nu bij binnenkomst van de request en dekt + alles tot en met grounding. De gestreamde rerank heeft bewust een eigen + timeout: een totaalbudget zou een gezonde stream halverwege afkappen. +- Concurrency op de grounding fan-out is nu process-wide begrensd in + plaats van per request. + +Waarom: + +- Dit zijn precies de randgevallen die je in een demo niet wilt zien; ze + nu dichtzetten is goedkoper dan er straks 1 in een review tegenkomen. + +Wat ik heb laten vallen of uitgesteld: + +- Een totaalbudget over de hele request heen; de afweging staat hierboven + en komt ook in de README.