diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..cc22299 --- /dev/null +++ b/.dockerignore @@ -0,0 +1,10 @@ +.git +**/.venv +**/__pycache__ +**/.mypy_cache +**/.pytest_cache +**/.ruff_cache +frontend/node_modules +eval/fixtures/raw +dist +**/dist diff --git a/backend/Dockerfile b/backend/Dockerfile index be491c8..cc1ba2c 100644 --- a/backend/Dockerfile +++ b/backend/Dockerfile @@ -6,11 +6,15 @@ RUN npm ci COPY frontend/ ./ RUN npm run build -FROM ghcr.io/astral-sh/uv:python3.13-bookworm-slim -WORKDIR /srv +FROM ghcr.io/astral-sh/uv:python3.13-bookworm-slim AS live +WORKDIR /srv/backend COPY backend/pyproject.toml backend/uv.lock ./ RUN uv sync --frozen --no-dev --no-install-project COPY backend/app ./app +COPY eval/scenarios.yaml ../eval/scenarios.yaml COPY --from=frontend /build/dist ./app/static EXPOSE 8000 CMD ["uv", "run", "--no-sync", "uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"] + +FROM live AS demo +COPY eval/fixtures ../eval/fixtures diff --git a/backend/app/adapters/anthropic/llm.py b/backend/app/adapters/anthropic/llm.py index b4558eb..205df35 100644 --- a/backend/app/adapters/anthropic/llm.py +++ b/backend/app/adapters/anthropic/llm.py @@ -112,7 +112,7 @@ class AnthropicRecommender: if parsed is None: raise RecommenderOutputError("Intent response contained no structured output") validated = IntentOutput.model_validate(parsed.model_dump()) - return _to_intent(validated) + return to_intent(validated) async def stream_rerank( self, @@ -125,7 +125,7 @@ class AnthropicRecommender: ) -> AsyncGenerator[RerankSelection]: """Yield each complete valid selection while the JSON is streaming.""" schema = transform_schema(RerankOutput.model_json_schema()) - parser = _RecommendationObjectParser() + parser = RecommendationObjectParser() async with self.client.messages.stream( model=self.settings.llm_model, max_tokens=self.settings.rerank_max_tokens, @@ -166,8 +166,11 @@ class AnthropicRecommender: raise RecommenderOutputError("Rerank response returned too many selections") -class _RecommendationObjectParser: +class RecommendationObjectParser: + """Incrementally extract complete recommendation objects from JSON text.""" + def __init__(self) -> None: + """Create an empty parser for one rerank response.""" self.complete_text = "" self._scan_index = 0 self._object_start: int | None = None @@ -177,6 +180,7 @@ class _RecommendationObjectParser: self._has_found_array = False def feed(self, text_delta: str) -> list[RerankSelectionOutput]: + """Consume a text delta and return newly completed selections.""" self.complete_text += text_delta if not self._has_found_array: match = _RECOMMENDATION_ARRAY.search(self.complete_text) @@ -231,7 +235,8 @@ class _RecommendationObjectParser: raise RecommenderOutputError("Rerank item failed validation") from error -def _to_intent(output: IntentOutput) -> Intent: +def to_intent(output: IntentOutput) -> Intent: + """Map validated intent output to the application domain.""" return Intent( mood=tuple(output.mood), activity=output.activity, diff --git a/backend/app/adapters/demo/cassette.py b/backend/app/adapters/demo/cassette.py new file mode 100644 index 0000000..c4e470d --- /dev/null +++ b/backend/app/adapters/demo/cassette.py @@ -0,0 +1,128 @@ +"""Decode recorded Spotify and Anthropic responses for demo replay.""" + +import base64 +import json +from dataclasses import dataclass +from functools import cache +from pathlib import Path + +from app.adapters.demo.scenario import FIXTURE_ROOT + + +@dataclass(frozen=True) +class RecordedResponse: + """One decoded response retained from a recorded HTTP interaction.""" + + method: str + url: str + status: int + response_body: bytes + response_chunks: tuple[bytes, ...] + + def json_body(self) -> object: + """Parse the decoded response body as JSON.""" + return json.loads(self.response_body) + + +@dataclass(frozen=True) +class DemoCassette: + """The recorded service responses needed by one demo scenario.""" + + key: str + spotify_responses: tuple[RecordedResponse, ...] + intent_response_bodies: tuple[bytes, ...] + rerank_response_chunks: tuple[bytes, ...] + + @property + def intent_response_body(self) -> bytes: + """Return the final intent response recorded for this scenario.""" + return self.intent_response_bodies[-1] + + @property + def spotify_search_responses(self) -> tuple[RecordedResponse, ...]: + """Return recorded Spotify search responses in capture order.""" + return tuple(response for response in self.spotify_responses if "/search?" in response.url) + + @property + def spotify_taste_responses(self) -> tuple[RecordedResponse, ...]: + """Return recorded Spotify taste responses in capture order.""" + return tuple( + response + for response in self.spotify_responses + if any( + endpoint in response.url + for endpoint in ("/me/top/artists", "/me/top/tracks", "/me/tracks?") + ) + ) + + +@cache +def load_cassette(scenario_key: str) -> DemoCassette: + """Load and decode both service cassettes for one scenario.""" + scenario_root = FIXTURE_ROOT / scenario_key + spotify = _load_responses(scenario_root / "spotify.json") + anthropic = _load_responses(scenario_root / "anthropic.json") + intent_responses = tuple( + response for response in anthropic if _is_anthropic_message(response.response_body) + ) + rerank_responses = tuple( + response for response in anthropic if not _is_anthropic_message(response.response_body) + ) + if not spotify or not intent_responses or not rerank_responses: + raise ValueError(f"Scenario cassette is incomplete: {scenario_key}") + return DemoCassette( + key=scenario_key, + spotify_responses=spotify, + intent_response_bodies=tuple(response.response_body for response in intent_responses), + rerank_response_chunks=rerank_responses[-1].response_chunks, + ) + + +def _load_responses(path: Path) -> tuple[RecordedResponse, ...]: + payload = json.loads(path.read_text(encoding="ascii")) + if not isinstance(payload, list): + raise ValueError(f"Cassette must contain a response list: {path}") + return tuple(_decode_response(entry) for entry in payload) + + +def _decode_response(payload: object) -> RecordedResponse: + if not isinstance(payload, dict): + raise ValueError("Cassette response must be a mapping") + try: + chunks_value = payload["response_chunks_base64"] + if not isinstance(chunks_value, list): + raise TypeError + chunks = tuple(_decode_base64(value) for value in chunks_value) + return RecordedResponse( + method=_string(payload["method"]), + url=_string(payload["url"]), + status=_integer(payload["status"]), + response_body=_decode_base64(payload["response_body_base64"]), + response_chunks=chunks, + ) + except (KeyError, TypeError, ValueError) as error: + raise ValueError("Cassette response is invalid") from error + + +def _decode_base64(value: object) -> bytes: + return base64.b64decode(_string(value), validate=True) + + +def _string(value: object) -> str: + if not isinstance(value, str): + raise TypeError + return value + + +def _integer(value: object) -> int: + if not isinstance(value, int): + raise TypeError + return value + + +def _is_anthropic_message(body: bytes) -> bool: + try: + payload = json.loads(body) + except (json.JSONDecodeError, UnicodeDecodeError): + return False + return isinstance(payload, dict) and payload.get("type") == "message" diff --git a/backend/app/adapters/demo/catalog.py b/backend/app/adapters/demo/catalog.py new file mode 100644 index 0000000..f07bd48 --- /dev/null +++ b/backend/app/adapters/demo/catalog.py @@ -0,0 +1,89 @@ +"""Spotify catalog adapter backed by one recorded demo cassette.""" + +from collections import defaultdict +from urllib.parse import parse_qs, urlparse + +from app.adapters.demo.cassette import DemoCassette, RecordedResponse +from app.adapters.demo.scenario import normalize_text +from app.adapters.spotify.mapping import ( + parse_saved_track_page, + parse_search_tracks, + parse_top_artists, + parse_track_page, +) +from app.domain.models import Track +from app.ports.protocols import TimeRange + + +class DemoCatalog: + """Replay recorded Spotify search and taste responses without HTTP.""" + + def __init__(self, cassette: DemoCassette) -> None: + """Index one scenario cassette for deterministic request replay.""" + self.cassette = cassette + self._search_pages = _index_search_pages(cassette.spotify_search_responses) + self._search_cursors: dict[str, int] = defaultdict(int) + + async def search_tracks(self, query: str, limit: int = 10) -> list[Track]: + """Return the next recorded search page for a normalized query.""" + normalized_query = normalize_text(query) + pages = self._search_pages.get(normalized_query, ()) + cursor = self._search_cursors[normalized_query] + if cursor >= len(pages): + return [] + self._search_cursors[normalized_query] += 1 + return parse_search_tracks(pages[cursor].json_body())[:limit] + + async def fetch_top_artists(self, time_range: TimeRange, limit: int) -> list[str]: + """Return the recorded synthetic top artists for a time range.""" + response = self._taste_response("/me/top/artists", time_range) + return parse_top_artists(response.json_body())[:limit] + + async def fetch_top_tracks(self, time_range: TimeRange, limit: int) -> list[Track]: + """Return the recorded synthetic top tracks for a time range.""" + response = self._taste_response("/me/top/tracks", time_range) + return parse_track_page(response.json_body())[:limit] + + async def fetch_saved_tracks(self, limit: int) -> list[Track]: + """Return recorded synthetic saved-track pages in offset order.""" + responses = sorted( + ( + response + for response in self.cassette.spotify_taste_responses + if urlparse(response.url).path.endswith("/me/tracks") + ), + key=_saved_track_offset, + ) + tracks: list[Track] = [] + for response in responses: + tracks.extend(parse_saved_track_page(response.json_body())) + if len(tracks) >= limit: + break + return tracks[:limit] + + def _taste_response(self, endpoint: str, time_range: TimeRange) -> RecordedResponse: + for response in self.cassette.spotify_taste_responses: + parsed_url = urlparse(response.url) + query = parse_qs(parsed_url.query) + if parsed_url.path.endswith(endpoint) and query.get("time_range") == [time_range]: + return response + raise ValueError(f"Cassette lacks {endpoint} for {time_range}") + + +def _index_search_pages( + responses: tuple[RecordedResponse, ...], +) -> dict[str, tuple[RecordedResponse, ...]]: + pages: dict[str, list[RecordedResponse]] = defaultdict(list) + for response in responses: + query = parse_qs(urlparse(response.url).query).get("q") + if query: + pages[normalize_text(query[0])].append(response) + return {key: tuple(value) for key, value in pages.items()} + + +def _saved_track_offset(response: RecordedResponse) -> int: + value = parse_qs(urlparse(response.url).query).get("offset", ["0"])[0] + try: + return int(value) + except ValueError: + return 0 diff --git a/backend/app/adapters/demo/pipeline.py b/backend/app/adapters/demo/pipeline.py new file mode 100644 index 0000000..aab0fed --- /dev/null +++ b/backend/app/adapters/demo/pipeline.py @@ -0,0 +1,78 @@ +"""Demo-only pipeline decorator for honest fuzzy replay disclosure.""" + +import time +from collections.abc import AsyncGenerator +from contextlib import aclosing + +from app.adapters.demo.catalog import DemoCatalog +from app.adapters.demo.recommender import DemoRecommender, parse_recorded_intent +from app.adapters.demo.scenario import select_replay_scenario +from app.config import Settings +from app.domain.models import ConversationTurn, PreviousRecommendation, Track +from app.pipeline.event import PipelineEvent, PipelineMetadataEvent, PipelineWarningEvent +from app.pipeline.grounding import Grounder +from app.pipeline.orchestrator import RecommendationPipeline +from app.ports.protocols import MusicCatalog + +DEMO_REPLAY_CODE = "demo_replay" + + +class DemoReplayPipeline: + """Decorate the real pipeline with a fuzzy-replay warning event.""" + + def __init__(self, settings: Settings) -> None: + """Create one cache-preserving pipeline with a replay recommender.""" + self.settings = settings + self.pipeline = RecommendationPipeline(DemoRecommender(settings), settings) + + async def stream( + self, + session_id: str, + request_id: str, + catalog: MusicCatalog, + query: str, + history: tuple[ConversationTurn, ...], + previous_recommendations: tuple[PreviousRecommendation, ...], + ) -> AsyncGenerator[PipelineEvent]: + """Stream the selected fixture and disclose non-exact selection.""" + match = select_replay_scenario(query, bool(previous_recommendations)) + seeded_pool = await self._prepare_refinement_pool(catalog) + event_stream = self.pipeline.stream( + session_id, + request_id, + catalog, + query, + history, + previous_recommendations, + seeded_pool=seeded_pool, + ) + async with aclosing(event_stream) as events: + async for event in events: + yield event + if isinstance(event, PipelineMetadataEvent) and not match.is_exact: + yield PipelineWarningEvent( + code=DEMO_REPLAY_CODE, + message=f'Demo replay is showing the recorded "{match.chip}" scenario.', + ) + + async def _prepare_refinement_pool( + self, + catalog: MusicCatalog, + ) -> tuple[Track, ...] | None: + if not isinstance(catalog, DemoCatalog): + return None + intent_bodies = catalog.cassette.intent_response_bodies + if len(intent_bodies) < 2: + return None + # Refinement cassettes record the parent intent first and refinement intent last. + parent_intent = parse_recorded_intent(intent_bodies[0]) + result = await Grounder(self.settings).ground( + catalog, + parent_intent.candidates, + # The recorded parent pool is listener-neutral, so replay has no known-track exclusions. + frozenset(), + parent_intent.familiarity, + self.settings.rerank_count + self.settings.rerank_pool_buffer, + time.monotonic() + self.settings.request_deadline_seconds, + ) + return result.tracks diff --git a/backend/app/adapters/demo/playlist.py b/backend/app/adapters/demo/playlist.py new file mode 100644 index 0000000..b630769 --- /dev/null +++ b/backend/app/adapters/demo/playlist.py @@ -0,0 +1,28 @@ +"""Playlist writer that makes demo saves explicit and network-free.""" + +import hashlib + +import structlog + +from app.domain.models import CreatedPlaylist + + +class DemoPlaylistWriter: + """Simulate playlist writes with stable Spotify-shaped URLs.""" + + async def create_playlist(self, name: str, description: str) -> CreatedPlaylist: + """Return a deterministic fake playlist without an external write.""" + digest = hashlib.sha256(f"{name}\n{description}".encode()).hexdigest()[:12] + playlist_id = f"demo-{digest}" + return CreatedPlaylist( + id=playlist_id, + url=f"https://open.spotify.com/playlist/{playlist_id}", + ) + + async def add_tracks_to_playlist(self, playlist_id: str, track_uris: list[str]) -> None: + """Log the simulated track addition without writing to Spotify.""" + structlog.get_logger().info( + "demo_playlist_simulated", + playlist_id=playlist_id, + track_count=len(track_uris), + ) diff --git a/backend/app/adapters/demo/recommender.py b/backend/app/adapters/demo/recommender.py new file mode 100644 index 0000000..dbbad20 --- /dev/null +++ b/backend/app/adapters/demo/recommender.py @@ -0,0 +1,128 @@ +"""Anthropic recommender adapter backed by recorded response chunks.""" + +import asyncio +import json +from collections.abc import AsyncGenerator +from contextvars import ContextVar + +from app.adapters.anthropic.llm import ( + IntentOutput, + RecommendationObjectParser, + RerankOutput, + to_intent, +) +from app.adapters.demo.cassette import DemoCassette, load_cassette +from app.adapters.demo.scenario import select_replay_scenario +from app.config import Settings +from app.domain.models import ( + ConversationTurn, + Intent, + PreviousRecommendation, + RerankSelection, + Track, +) +from app.ports.protocols import RecommenderOutputError + + +class DemoRecommender: + """Replay recorded intent and rerank output through live validation.""" + + def __init__(self, settings: Settings) -> None: + """Bind replay pacing and task-local scenario state.""" + self.settings = settings + self._cassette: ContextVar[DemoCassette | None] = ContextVar( + "demo_recommender_cassette", + default=None, + ) + + async def create_intent( + self, + query: str, + history: tuple[ConversationTurn, ...], + previous_recommendations: tuple[PreviousRecommendation, ...], + taste_summary: str, + candidate_count: int, + ) -> Intent: + """Parse the selected scenario's recorded structured intent.""" + match = select_replay_scenario(query, bool(previous_recommendations)) + cassette = load_cassette(match.key) + self._cassette.set(cassette) + return parse_recorded_intent(cassette.intent_response_body) + + async def stream_rerank( + self, + intent: Intent, + grounded_tracks: tuple[Track, ...], + taste_summary: str, + history: tuple[ConversationTurn, ...], + selection_count: int, + correction: str | None = None, + ) -> AsyncGenerator[RerankSelection]: + """Replay recorded SSE chunks through the live incremental parser.""" + cassette = self._cassette.get() + if cassette is None: + raise RecommenderOutputError("Demo rerank has no selected scenario") + parser = RecommendationObjectParser() + async for text_delta in _text_deltas( + cassette.rerank_response_chunks, + self.settings.demo_chunk_delay_seconds, + ): + for selection in parser.feed(text_delta): + yield RerankSelection(selection.track_id, selection.justification) + try: + validated = RerankOutput.model_validate_json(parser.complete_text) + except ValueError as error: + raise RecommenderOutputError("Recorded rerank response is invalid") from error + if len(validated.recommendations) > selection_count: + raise RecommenderOutputError("Recorded rerank returned too many selections") + + +def parse_recorded_intent(response_body: bytes) -> Intent: + """Parse an Anthropic message body through the live intent output model.""" + try: + response = json.loads(response_body) + content = response["content"] + text = content[0]["text"] + if not isinstance(text, str): + raise TypeError + return to_intent(IntentOutput.model_validate_json(text)) + except (IndexError, KeyError, TypeError, ValueError) as error: + raise RecommenderOutputError("Recorded intent response is invalid") from error + + +async def _text_deltas( + chunks: tuple[bytes, ...], + delay_seconds: float, +) -> AsyncGenerator[str]: + buffer = b"" + for chunk_index, chunk in enumerate(chunks): + if chunk_index and delay_seconds > 0: + await asyncio.sleep(delay_seconds) + buffer += chunk + buffer = buffer.replace(b"\r\n", b"\n") + while b"\n\n" in buffer: + event, buffer = buffer.split(b"\n\n", 1) + text_delta = _event_text_delta(event) + if text_delta is not None: + yield text_delta + if buffer: + text_delta = _event_text_delta(buffer) + if text_delta is not None: + yield text_delta + + +def _event_text_delta(event: bytes) -> str | None: + data_lines = [line[5:].strip() for line in event.splitlines() if line.startswith(b"data:")] + if not data_lines: + return None + try: + payload = json.loads(b"\n".join(data_lines)) + except (json.JSONDecodeError, UnicodeDecodeError): + return None + if not isinstance(payload, dict) or payload.get("type") != "content_block_delta": + return None + delta = payload.get("delta") + if not isinstance(delta, dict) or delta.get("type") != "text_delta": + return None + text = delta.get("text") + return text if isinstance(text, str) else None diff --git a/backend/app/adapters/demo/scenario.py b/backend/app/adapters/demo/scenario.py new file mode 100644 index 0000000..a523729 --- /dev/null +++ b/backend/app/adapters/demo/scenario.py @@ -0,0 +1,151 @@ +"""Select replayable demo scenarios from the shared scenario catalog.""" + +import re +import unicodedata +from dataclasses import dataclass +from difflib import SequenceMatcher +from functools import lru_cache +from pathlib import Path + +import yaml + +_REPOSITORY_ROOT = Path(__file__).parents[4] +SCENARIO_PATH = _REPOSITORY_ROOT / "eval" / "scenarios.yaml" +FIXTURE_ROOT = _REPOSITORY_ROOT / "eval" / "fixtures" +_WORD_PATTERN = re.compile(r"[a-z0-9]+") + + +@dataclass(frozen=True) +class Scenario: + """One shared suggestion and its optional parent scenario.""" + + key: str + chip: str + query: str + after: str | None = None + + @property + def is_refinement(self) -> bool: + """Return whether this scenario refines an earlier result.""" + return self.after is not None + + +@dataclass(frozen=True) +class ScenarioMatch: + """Describe which recorded scenario will answer a demo request.""" + + key: str + is_exact: bool + scenario_query: str + chip: str + + +@lru_cache(maxsize=1) +def load_scenarios() -> tuple[Scenario, ...]: + """Load and validate the shared scenario catalog once.""" + payload = yaml.safe_load(SCENARIO_PATH.read_text(encoding="ascii")) + if not isinstance(payload, dict) or payload.get("version") != 1: + raise ValueError("Scenario catalog must have version 1") + entries = payload.get("scenarios") + if not isinstance(entries, list) or not entries: + raise ValueError("Scenario catalog must contain scenarios") + scenarios = tuple(_parse_scenario(entry) for entry in entries) + keys = {scenario.key for scenario in scenarios} + if len(keys) != len(scenarios): + raise ValueError("Scenario keys must be unique") + if any(scenario.after not in keys for scenario in scenarios if scenario.after is not None): + raise ValueError("Scenario catalog contains an unknown parent") + return scenarios + + +def select_scenario(query: str) -> ScenarioMatch: + """Select an exact scenario or the nearest available recorded fixture.""" + scenarios = _replayable_scenarios() + normalized_query = normalize_text(query) + for scenario in scenarios: + if normalized_query in {normalize_text(scenario.query), normalize_text(scenario.chip)}: + return _to_match(scenario, is_exact=True) + + selected = max( + scenarios, + key=lambda scenario: max( + SequenceMatcher(None, normalized_query, normalize_text(scenario.query)).ratio(), + SequenceMatcher(None, normalized_query, normalize_text(scenario.chip)).ratio(), + ), + ) + return _to_match(selected, is_exact=False) + + +def select_replay_scenario( + query: str, + has_previous_recommendations: bool, +) -> ScenarioMatch: + """Map a request with prior results to its recorded refinement when available.""" + selected = select_scenario(query) + scenarios = _replayable_scenarios() + selected_scenario = next(scenario for scenario in scenarios if scenario.key == selected.key) + if selected_scenario.is_refinement or not has_previous_recommendations: + return selected + refinement = next( + (scenario for scenario in scenarios if scenario.after == selected_scenario.key), + None, + ) + return _to_match(refinement, is_exact=False) if refinement is not None else selected + + +def suggestion_items() -> list[dict[str, str]]: + """Return chip and query pairs for non-refinement scenarios.""" + return [ + {"chip": scenario.chip, "query": scenario.query} + for scenario in load_scenarios() + if not scenario.is_refinement + ] + + +def normalize_text(value: str) -> str: + """Normalize text for exact and similarity matching.""" + decomposed = unicodedata.normalize("NFKD", value.casefold()) + ascii_text = decomposed.encode("ascii", errors="ignore").decode("ascii") + return " ".join(_WORD_PATTERN.findall(ascii_text)) + + +@lru_cache(maxsize=1) +def _replayable_scenarios() -> tuple[Scenario, ...]: + fixture_keys = {path.name for path in FIXTURE_ROOT.iterdir() if path.is_dir()} + scenarios = tuple(scenario for scenario in load_scenarios() if scenario.key in fixture_keys) + if not scenarios: + raise ValueError("Demo mode requires at least one scenario fixture") + return scenarios + + +def _parse_scenario(payload: object) -> Scenario: + if not isinstance(payload, dict): + raise ValueError("Each scenario must be a mapping") + required = {"key", "chip", "query", "facets"} + if not required <= set(payload): + raise ValueError("Scenario is missing a required field") + after = payload.get("after") + if after is not None and not isinstance(after, str): + raise ValueError("Scenario parent must be a key") + return Scenario( + key=_required_string(payload, "key"), + chip=_required_string(payload, "chip"), + query=_required_string(payload, "query"), + after=after, + ) + + +def _required_string(payload: dict[object, object], key: str) -> str: + value = payload.get(key) + if not isinstance(value, str) or not value: + raise ValueError(f"Scenario {key} must be a non-empty string") + return value + + +def _to_match(scenario: Scenario, is_exact: bool) -> ScenarioMatch: + return ScenarioMatch( + key=scenario.key, + is_exact=is_exact, + scenario_query=scenario.query, + chip=scenario.chip, + ) diff --git a/backend/app/api/recommendations.py b/backend/app/api/recommendations.py index 218328c..3b60512 100644 --- a/backend/app/api/recommendations.py +++ b/backend/app/api/recommendations.py @@ -1,4 +1,4 @@ -"""Authenticated recommendation streaming and playlist creation routes.""" +"""Recommendation streaming and playlist creation routes.""" import asyncio from collections.abc import AsyncIterator, Callable @@ -38,6 +38,10 @@ from app.pipeline.orchestrator import RecommendationPipeline from app.ports.protocols import MusicCatalog, PlaylistWriter SpotifyClientFactory = Callable[[SpotifySession], MusicCatalog | PlaylistWriter] +MusicCatalogFactory = Callable[ + [SpotifySession, str, tuple[PreviousRecommendation, ...]], + MusicCatalog, +] router = APIRouter() @@ -47,14 +51,12 @@ async def recommendations( request: Request, payload: RecommendationRequest, ) -> StreamingResponse: - """Stream one authenticated discovery response as NDJSON.""" + """Stream one session-resolved discovery response as NDJSON.""" resolved = resolve_session(request) if resolved is None: raise HTTPException(status_code=401, detail="Not authenticated") pipeline = cast(RecommendationPipeline, request.app.state.recommendation_pipeline) - factory = cast(SpotifyClientFactory, request.app.state.spotify_client_factory) - catalog = cast(MusicCatalog, factory(resolved.session)) request_id = cast(str, request.state.request_id) history = tuple(ConversationTurn(turn.role, turn.content) for turn in payload.history) previous = tuple( @@ -66,6 +68,8 @@ async def recommendations( ) for item in payload.prior_recommendations ) + factory = cast(MusicCatalogFactory, request.app.state.music_catalog_factory) + catalog = factory(resolved.session, payload.query, previous) return StreamingResponse( _stream_lines( @@ -86,7 +90,7 @@ async def create_playlist( request: Request, payload: PlaylistCreateRequest, ) -> PlaylistCreateResponse: - """Create and fill one authenticated Spotify playlist.""" + """Create and fill one live or simulated playlist.""" resolved = resolve_session(request) if resolved is None: raise HTTPException(status_code=401, detail="Not authenticated") diff --git a/backend/app/api/routes.py b/backend/app/api/routes.py index aab0682..05128a6 100644 --- a/backend/app/api/routes.py +++ b/backend/app/api/routes.py @@ -1,4 +1,4 @@ -"""HTTP routes for Spotify login and session management.""" +"""HTTP routes for login and session management.""" from dataclasses import dataclass from typing import cast @@ -9,7 +9,7 @@ from fastapi.responses import JSONResponse, RedirectResponse, Response from app.adapters.spotify.login import begin_login, complete_login from app.adapters.spotify.session import PendingLogins, SessionStore, SpotifySession -from app.config import Settings +from app.config import AppMode, Settings SESSION_COOKIE_NAME = "discovery_session" @@ -26,8 +26,10 @@ class ResolvedSession: @router.get("/api/auth/login") def login(request: Request) -> RedirectResponse: - """Start Spotify Authorization Code with PKCE login.""" + """Start the configured live or demo login flow.""" application_settings = cast(Settings, request.app.state.settings) + if application_settings.app_mode is AppMode.DEMO: + return RedirectResponse("/?login=demo", status_code=307) pending_logins = cast(PendingLogins, request.app.state.pending_logins) authorize_url = begin_login(application_settings, pending_logins) return RedirectResponse(authorize_url, status_code=307) @@ -40,11 +42,13 @@ async def callback( state: str | None = None, error: str | None = None, ) -> RedirectResponse: - """Complete Spotify login and establish an opaque cookie session.""" + """Complete live login and establish an opaque cookie session.""" + application_settings = cast(Settings, request.app.state.settings) + if application_settings.app_mode is AppMode.DEMO: + return RedirectResponse("/?login=demo", status_code=307) if error is not None or code is None or state is None: return _login_error_redirect() - application_settings = cast(Settings, request.app.state.settings) http = cast(httpx2.AsyncClient, request.app.state.http) pending_logins = cast(PendingLogins, request.app.state.pending_logins) session_store = cast(SessionStore, request.app.state.session_store) @@ -80,6 +84,12 @@ def current_session(request: Request) -> JSONResponse: return JSONResponse({"display_name": resolved.session.display_name}) +@router.get("/api/suggestions") +def suggestions(request: Request) -> list[dict[str, str]]: + """Return shared non-refinement suggestion chips and queries.""" + return cast(list[dict[str, str]], request.app.state.suggestions) + + @router.post("/api/auth/logout", status_code=204) def logout(request: Request) -> Response: """Remove the current application session and clear its cookie.""" @@ -105,7 +115,11 @@ def _login_error_redirect() -> RedirectResponse: def resolve_session(request: Request) -> ResolvedSession | None: - """Resolve the cookie session or the installed live seed session.""" + """Resolve the stable demo identity, cookie, or live seed session.""" + application_settings = cast(Settings, request.app.state.settings) + if application_settings.app_mode is AppMode.DEMO: + demo_session = cast(SpotifySession, request.app.state.demo_session) + return ResolvedSession("demo", demo_session) session_store = cast(SessionStore, request.app.state.session_store) cookie_session_id = request.cookies.get(SESSION_COOKIE_NAME) if cookie_session_id is not None: diff --git a/backend/app/config.py b/backend/app/config.py index a560614..fcd8d3e 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -73,6 +73,9 @@ class Settings(BaseSettings): resolution_cache_max_entries: int = 2048 taste_profile_ttl_seconds: float = 900.0 + # Demo replay paces recorded stream chunks so cards appear as they did live. + demo_chunk_delay_seconds: float = 0.05 + # Taste profile fetch bounds: enough signal to describe a listener # without paging through an entire library on session start. top_items_limit: int = 50 diff --git a/backend/app/main.py b/backend/app/main.py index 4b1a37c..4bd51cf 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -12,6 +12,11 @@ from fastapi import FastAPI from fastapi.staticfiles import StaticFiles from app.adapters.anthropic.llm import AnthropicRecommender +from app.adapters.demo.cassette import load_cassette +from app.adapters.demo.catalog import DemoCatalog +from app.adapters.demo.pipeline import DemoReplayPipeline +from app.adapters.demo.playlist import DemoPlaylistWriter +from app.adapters.demo.scenario import select_replay_scenario, suggestion_items from app.adapters.spotify.auth import TokenSet, refresh_access_token from app.adapters.spotify.client import SpotifyClient from app.adapters.spotify.errors import SpotifyError @@ -19,6 +24,7 @@ from app.adapters.spotify.session import PendingLogins, SessionStore, SpotifySes from app.api.recommendations import router as recommendations_router from app.api.routes import router from app.config import AppMode, Settings, settings +from app.domain.models import PreviousRecommendation from app.observability.logging import configure_logging from app.observability.timing import RequestTimingMiddleware from app.pipeline.orchestrator import RecommendationPipeline @@ -38,17 +44,35 @@ def create_app( @asynccontextmanager async def lifespan(application: FastAPI) -> AsyncIterator[None]: _validate_live_settings(active_settings) + application.state.session_store = SessionStore() + application.state.pending_logins = PendingLogins() + application.state.settings = active_settings + application.state.seed_session_id = None + application.state.suggestions = suggestion_items() + if active_settings.app_mode is AppMode.DEMO: + application.state.demo_session = _demo_session() + application.state.recommendation_pipeline = DemoReplayPipeline(active_settings) + + def demo_catalog_factory( + session: SpotifySession, + query: str, + previous_recommendations: tuple[PreviousRecommendation, ...], + ) -> DemoCatalog: + match = select_replay_scenario(query, bool(previous_recommendations)) + return DemoCatalog(load_cassette(match.key)) + + application.state.music_catalog_factory = demo_catalog_factory + application.state.spotify_client_factory = lambda session: DemoPlaylistWriter() + yield + return + async with httpx2.AsyncClient( timeout=active_settings.spotify_timeout_seconds, transport=http_transport, ) as http: application.state.http = http - application.state.session_store = SessionStore() - application.state.pending_logins = PendingLogins() - 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, timeout=active_settings.llm_timeout_seconds, http_client=anthropic_http_client, ) @@ -61,7 +85,15 @@ def create_app( def spotify_client_factory(session: SpotifySession) -> SpotifyClient: return SpotifyClient(http, session, active_settings) + def music_catalog_factory( + session: SpotifySession, + query: str, + previous_recommendations: tuple[PreviousRecommendation, ...], + ) -> SpotifyClient: + return spotify_client_factory(session) + application.state.spotify_client_factory = spotify_client_factory + application.state.music_catalog_factory = music_catalog_factory if ( active_settings.app_mode is AppMode.LIVE and active_settings.spotify_seed_refresh_token @@ -141,3 +173,11 @@ async def _install_seed_session( display_name=current_user.display_name, ) ) + + +def _demo_session() -> SpotifySession: + return SpotifySession( + tokens=TokenSet(access_token="demo", refresh_token="demo", expires_at=float("inf")), + account_id="demo-listener", + display_name="Demo Listener", + ) diff --git a/backend/app/pipeline/orchestrator.py b/backend/app/pipeline/orchestrator.py index fc20eac..f3a35fb 100644 --- a/backend/app/pipeline/orchestrator.py +++ b/backend/app/pipeline/orchestrator.py @@ -118,6 +118,7 @@ class RecommendationPipeline: query: str, history: tuple[ConversationTurn, ...], previous_recommendations: tuple[PreviousRecommendation, ...], + seeded_pool: tuple[Track, ...] | None = None, ) -> AsyncGenerator[PipelineEvent]: """Yield ordered events for one recommendation request.""" started_at = time.monotonic() @@ -157,7 +158,9 @@ class RecommendationPipeline: candidate_count=len(intent.candidates), ) - pool = await self._grounded_pool(session_id, catalog, intent, taste, deadline_at) + pool = await self._grounded_pool( + session_id, catalog, intent, taste, deadline_at, seeded_pool + ) if not pool: yield PipelineErrorEvent( code="no_grounded_results", @@ -189,10 +192,13 @@ class RecommendationPipeline: intent: Intent, taste: CompressedTasteProfile, deadline_at: float, + seeded_pool: tuple[Track, ...] | None, ) -> tuple[Track, ...]: - """Reuse the session's pool on refinement, otherwise ground anew.""" + """Reuse the seeded or session pool on refinement, otherwise ground anew.""" if intent.is_refinement: - cached_pool = self.last_pools.get(session_id) + cached_pool = ( + seeded_pool if seeded_pool is not None else self.last_pools.get(session_id) + ) if cached_pool: return cached_pool result = await self.grounder.ground( diff --git a/backend/tests/test_anthropic_llm.py b/backend/tests/test_anthropic_llm.py index 27d3a61..c8287e3 100644 --- a/backend/tests/test_anthropic_llm.py +++ b/backend/tests/test_anthropic_llm.py @@ -2,12 +2,12 @@ import pytest -from app.adapters.anthropic.llm import _RecommendationObjectParser +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() + parser = RecommendationObjectParser() with pytest.raises(RecommenderOutputError, match="object start position"): parser._finish_object() diff --git a/backend/tests/test_demo_adapters.py b/backend/tests/test_demo_adapters.py new file mode 100644 index 0000000..ebdd20c --- /dev/null +++ b/backend/tests/test_demo_adapters.py @@ -0,0 +1,68 @@ +"""Offline replay tests for demo service adapters.""" + +import asyncio + +from app.adapters.demo.cassette import load_cassette +from app.adapters.demo.catalog import DemoCatalog +from app.adapters.demo.recommender import DemoRecommender +from app.config import Settings + + +def test_cassette_decodes_bodies_and_preserves_rerank_chunks() -> None: + cassette = load_cassette("focus-coding") + + assert cassette.intent_response_body.startswith(b'{"model"') + assert len(cassette.spotify_search_responses) > 20 + assert len(cassette.spotify_taste_responses) == 6 + assert len(cassette.rerank_response_chunks) > 1 + assert b"event: content_block_delta" in b"".join(cassette.rerank_response_chunks) + + +def test_demo_catalog_replays_search_and_synthetic_taste_pages() -> None: + async def run() -> None: + catalog = DemoCatalog(load_cassette("focus-coding")) + + tracks = await catalog.search_tracks('track:"Stay" artist:"Hybrid Minds"') + artists = await catalog.fetch_top_artists("short_term", 50) + top_tracks = await catalog.fetch_top_tracks("long_term", 50) + saved_tracks = await catalog.fetch_saved_tracks(100) + + assert tracks[0].title == "Stay" + assert artists[0] == "Synthetic Focus Artist" + assert {track.title for track in top_tracks} >= { + "Synthetic Focus Track", + "Synthetic Jazz Track", + } + assert saved_tracks[0].title == "Synthetic Saved Track" + + asyncio.run(run()) + + +def test_demo_recommender_parses_intent_and_streams_recorded_rerank() -> None: + async def run() -> None: + recommender = DemoRecommender(Settings(demo_chunk_delay_seconds=0)) + intent = await recommender.create_intent( + "Focus while coding", + (), + (), + "Synthetic taste", + 35, + ) + selections = [ + selection + async for selection in recommender.stream_rerank( + intent, + (), + "Synthetic taste", + (), + 15, + ) + ] + + assert intent.activity == "programming" + assert len(intent.candidates) == 35 + assert len(selections) == 15 + assert selections[0].track_id == "2nIixNuuV5eHCJydG5aYIB" + assert all(selection.justification for selection in selections) + + asyncio.run(run()) diff --git a/backend/tests/test_demo_app.py b/backend/tests/test_demo_app.py new file mode 100644 index 0000000..5b04d31 --- /dev/null +++ b/backend/tests/test_demo_app.py @@ -0,0 +1,208 @@ +"""Full HTTP contract tests for key-free demo mode.""" + +import asyncio + +import httpx +from fastapi.testclient import TestClient +from pydantic import TypeAdapter +from starlette.types import ASGIApp, Message, Receive, Scope, Send + +from app.adapters.demo.scenario import load_scenarios +from app.api.schemas import StreamEvent +from app.config import Settings +from app.main import create_app + + +class _MetadataBarrier: + """Pause one marked ASGI response immediately after its metadata event.""" + + def __init__(self, app: ASGIApp) -> None: + self.app = app + self.metadata_sent = asyncio.Event() + self.resume_response = asyncio.Event() + + async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: + is_refinement = dict(scope.get("headers", ())).get(b"x-demo-request") == b"refinement" + + async def send_with_barrier(message: Message) -> None: + await send(message) + if message["type"] == "http.response.body" and b'"type":"metadata"' in message.get( + "body", b"" + ): + self.metadata_sent.set() + await self.resume_response.wait() + + await self.app(scope, receive, send_with_barrier if is_refinement else send) + + +def test_demo_request_streams_valid_events_without_a_session() -> None: + app = create_app(Settings(demo_chunk_delay_seconds=0)) + with TestClient(app) as client: + response = client.post( + "/api/recommendations", + json={"schema_version": 1, "query": "Focus while coding"}, + ) + + adapter: TypeAdapter[StreamEvent] = TypeAdapter(StreamEvent) + events = [adapter.validate_json(line) for line in response.text.splitlines()] + assert response.status_code == 200 + assert events[0].type == "metadata" + assert events[-1].type == "done" + assert sum(event.type == "track" for event in events) == 15 + + +def test_unknown_demo_query_discloses_the_replayed_scenario_before_tracks() -> None: + app = create_app(Settings(demo_chunk_delay_seconds=0)) + with TestClient(app) as client: + response = client.post( + "/api/recommendations", + json={"schema_version": 1, "query": "Focus while codign"}, + ) + + adapter: TypeAdapter[StreamEvent] = TypeAdapter(StreamEvent) + events = [adapter.validate_json(line) for line in response.text.splitlines()] + warning_index = next( + index + for index, event in enumerate(events) + if event.type == "warning" and event.code == "demo_replay" + ) + first_track_index = next(index for index, event in enumerate(events) if event.type == "track") + warning = events[warning_index] + assert warning_index < first_track_index + assert warning.type == "warning" + assert "Focus while coding" in warning.message + + +def test_demo_refinement_reconstructs_the_pool_recorded_with_its_cassette() -> None: + app = create_app(Settings(demo_chunk_delay_seconds=0)) + with TestClient(app) as client: + initial_response = client.post( + "/api/recommendations", + json={"schema_version": 1, "query": "Focus while coding"}, + ) + initial_adapter: TypeAdapter[StreamEvent] = TypeAdapter(StreamEvent) + initial_events: list[StreamEvent] = [ + initial_adapter.validate_json(line) for line in initial_response.text.splitlines() + ] + prior_recommendations = [ + { + "rank": event.rank, + "track_id": event.track.id, + "title": event.track.title, + "artists": event.track.artists, + } + for event in initial_events + if event.type == "track" + ] + response = client.post( + "/api/recommendations", + json={ + "schema_version": 1, + "query": "More electronic", + "prior_recommendations": prior_recommendations, + }, + ) + + adapter: TypeAdapter[StreamEvent] = TypeAdapter(StreamEvent) + events = [adapter.validate_json(line) for line in response.text.splitlines()] + warning_codes = [event.code for event in events if event.type == "warning"] + assert "rerank_fallback" not in warning_codes + assert sum(event.type == "track" for event in events) == 15 + + +def test_concurrent_demo_refinement_uses_its_request_local_recorded_pool() -> None: + async def run() -> None: + settings = Settings(demo_chunk_delay_seconds=0) + previous = [ + { + "rank": 1, + "track_id": "previous", + "title": "Previous track", + "artists": ["Previous artist"], + } + ] + payload = { + "schema_version": 1, + "query": "More electronic", + "prior_recommendations": previous, + } + + baseline_app = create_app(settings) + baseline_transport = httpx.ASGITransport(app=baseline_app) + async with ( + baseline_app.router.lifespan_context(baseline_app), + httpx.AsyncClient( + transport=baseline_transport, + base_url="http://test", + ) as client, + ): + baseline_response = await client.post("/api/recommendations", json=payload) + baseline_events = _stream_events(baseline_response) + expected_track_ids = [event.track.id for event in baseline_events if event.type == "track"] + + concurrent_app = create_app(settings) + barrier = _MetadataBarrier(concurrent_app) + transport = httpx.ASGITransport(app=barrier) + async with ( + concurrent_app.router.lifespan_context(concurrent_app), + httpx.AsyncClient(transport=transport, base_url="http://test") as client, + ): + refinement_task = asyncio.create_task( + client.post( + "/api/recommendations", + json=payload, + headers={"x-demo-request": "refinement"}, + ) + ) + try: + await asyncio.wait_for(barrier.metadata_sent.wait(), timeout=2) + unrelated_response = await client.post( + "/api/recommendations", + json={"schema_version": 1, "query": "Energy for the gym"}, + ) + finally: + barrier.resume_response.set() + refinement_response = await refinement_task + + events = _stream_events(refinement_response) + warning_codes = [event.code for event in events if event.type == "warning"] + track_ids = [event.track.id for event in events if event.type == "track"] + assert unrelated_response.status_code == 200 + assert refinement_response.status_code == 200 + assert "rerank_fallback" not in warning_codes + assert track_ids == expected_track_ids + + asyncio.run(run()) + + +def test_demo_auth_playlist_and_suggestions_are_explicitly_simulated() -> None: + app = create_app(Settings(demo_chunk_delay_seconds=0)) + with TestClient(app, follow_redirects=False) as client: + current_user = client.get("/api/auth/me") + login = client.get("/api/auth/login") + playlist = client.post( + "/api/playlists", + json={ + "schema_version": 1, + "name": "Night drive", + "track_uris": ["spotify:track:demo"], + }, + ) + suggestions = client.get("/api/suggestions") + + expected_suggestions = [ + {"chip": scenario.chip, "query": scenario.query} + for scenario in load_scenarios() + if not scenario.is_refinement + ] + assert current_user.json() == {"display_name": "Demo Listener"} + assert login.headers["location"] == "/?login=demo" + assert playlist.json()["url"].startswith("https://open.spotify.com/playlist/demo-") + assert suggestions.json() == expected_suggestions + assert not hasattr(app.state, "http") + assert not hasattr(app.state, "anthropic") + + +def _stream_events(response: httpx.Response) -> list[StreamEvent]: + adapter: TypeAdapter[StreamEvent] = TypeAdapter(StreamEvent) + return [adapter.validate_json(line) for line in response.text.splitlines()] diff --git a/backend/tests/test_demo_scenario.py b/backend/tests/test_demo_scenario.py new file mode 100644 index 0000000..99c30d7 --- /dev/null +++ b/backend/tests/test_demo_scenario.py @@ -0,0 +1,32 @@ +"""Tests for deterministic demo scenario selection.""" + +from app.adapters.demo.scenario import select_replay_scenario, select_scenario + + +def test_scenario_selection_prefers_normalized_exact_query_or_chip() -> None: + query_match = select_scenario("something calm for while I am programming, but not boring") + chip_match = select_scenario(" FOCUS, while CODING! ") + + assert query_match.key == "focus-coding" + assert query_match.is_exact + assert chip_match.key == "focus-coding" + assert chip_match.is_exact + + +def test_scenario_selection_falls_back_to_nearest_fixture() -> None: + match = select_scenario("Focus while codign") + + assert match.key == "focus-coding" + assert not match.is_exact + + +def test_prior_results_select_a_matching_refinement_or_same_scenario() -> None: + refinement = select_replay_scenario("Focus while coding", True) + direct_refinement = select_replay_scenario("More electronic", False) + same_scenario = select_replay_scenario("Rainy Sunday", True) + + assert refinement.key == "focus-coding-refine" + assert not refinement.is_exact + assert direct_refinement.key == "focus-coding-refine" + assert direct_refinement.is_exact + assert same_scenario.key == "rainy-sunday" diff --git a/backend/tests/test_health.py b/backend/tests/test_health.py index 441bdc1..eb12b52 100644 --- a/backend/tests/test_health.py +++ b/backend/tests/test_health.py @@ -22,8 +22,14 @@ def test_anthropic_client_uses_configured_timeout(monkeypatch: pytest.MonkeyPatc constructor = Mock(return_value=anthropic_client) monkeypatch.setattr("app.main.AsyncAnthropic", constructor) - with TestClient(create_app(Settings(llm_timeout_seconds=42.0))) as client: + live_settings = Settings( + app_mode="live", + spotify_client_id="client-id", + anthropic_api_key="api-key", + llm_timeout_seconds=42.0, + ) + with TestClient(create_app(live_settings)) 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, http_client=None) + constructor.assert_called_once_with(api_key="api-key", timeout=42.0, http_client=None) diff --git a/backend/tests/test_recommendation_api.py b/backend/tests/test_recommendation_api.py index 1920454..569ad67 100644 --- a/backend/tests/test_recommendation_api.py +++ b/backend/tests/test_recommendation_api.py @@ -12,6 +12,7 @@ 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 +from app.config import AppMode, Settings from app.domain.models import ConversationTurn, CreatedPlaylist, PreviousRecommendation, Track from app.main import create_app from app.pipeline.event import PipelineDoneEvent, PipelineMetadataEvent, PipelineTrackEvent @@ -81,8 +82,13 @@ def test_recommendations_stream_lines_validate_against_frozen_schemas( assert [event.type for event in events] == ["metadata", "track", "done"] -def test_recommendations_require_a_valid_session_without_seed() -> None: - with TestClient(create_app()) as client: +def test_live_recommendations_require_a_valid_session_without_seed() -> None: + application_settings = Settings( + app_mode=AppMode.LIVE, + spotify_client_id="client-id", + anthropic_api_key="test-key", + ) + with TestClient(create_app(application_settings)) as client: response = client.post( "/api/recommendations", json={"schema_version": 1, "query": "focused electronic music"}, diff --git a/docker-compose.yml b/docker-compose.yml index 1c38478..7e487aa 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -3,6 +3,7 @@ services: build: context: . dockerfile: backend/Dockerfile + target: demo env_file: - path: .env required: false