diff --git a/backend/app/adapters/anthropic/llm.py b/backend/app/adapters/anthropic/llm.py index 98f5c5e..b5c4c05 100644 --- a/backend/app/adapters/anthropic/llm.py +++ b/backend/app/adapters/anthropic/llm.py @@ -2,7 +2,7 @@ import json import re -from collections.abc import AsyncIterator +from collections.abc import AsyncGenerator from typing import Literal, cast from anthropic import AsyncAnthropic @@ -124,7 +124,7 @@ class AnthropicRecommender: history: tuple[ConversationTurn, ...], selection_count: int, correction: str | None = None, - ) -> AsyncIterator[RerankSelection]: + ) -> AsyncGenerator[RerankSelection]: """Yield each complete valid selection while the JSON is streaming.""" schema = transform_schema(RerankOutput.model_json_schema()) parser = _RecommendationObjectParser() diff --git a/backend/app/pipeline/event.py b/backend/app/pipeline/event.py new file mode 100644 index 0000000..e383246 --- /dev/null +++ b/backend/app/pipeline/event.py @@ -0,0 +1,62 @@ +"""Application-owned events emitted by the recommendation pipeline.""" + +from dataclasses import dataclass, field +from typing import Literal + +from app.domain.models import Track + + +@dataclass(frozen=True) +class PipelineMetadataEvent: + """Describe the interpreted request before track results.""" + + request_id: str + intent_summary: str + candidate_count: int + type: Literal["metadata"] = field(default="metadata", init=False) + + +@dataclass(frozen=True) +class PipelineTrackEvent: + """Carry one ranked, grounded recommendation.""" + + rank: int + track: Track + justification: str + type: Literal["track"] = field(default="track", init=False) + + +@dataclass(frozen=True) +class PipelineWarningEvent: + """Report a non-terminal degradation.""" + + code: str + message: str + type: Literal["warning"] = field(default="warning", init=False) + + +@dataclass(frozen=True) +class PipelineErrorEvent: + """Report a terminal recommendation failure.""" + + code: str + message: str + type: Literal["error"] = field(default="error", init=False) + + +@dataclass(frozen=True) +class PipelineDoneEvent: + """Report final track count and elapsed time.""" + + track_count: int + total_ms: int + type: Literal["done"] = field(default="done", init=False) + + +type PipelineEvent = ( + PipelineMetadataEvent + | PipelineTrackEvent + | PipelineWarningEvent + | PipelineErrorEvent + | PipelineDoneEvent +) diff --git a/backend/app/pipeline/grounding.py b/backend/app/pipeline/grounding.py new file mode 100644 index 0000000..32ea2b8 --- /dev/null +++ b/backend/app/pipeline/grounding.py @@ -0,0 +1,339 @@ +"""Resolve proposed tracks with bounded concurrency and conservative matching.""" + +import asyncio +import time +from collections import OrderedDict +from collections.abc import Callable +from dataclasses import dataclass +from enum import StrEnum + +import structlog + +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 +from app.observability.timing import increment_cache_hits +from app.ports.protocols import CatalogQuotaExhaustedError, MusicCatalog + + +class ResolutionStatus(StrEnum): + """Terminal outcome of one candidate resolution attempt.""" + + RESOLVED = "resolved" + MISS = "miss" + MISMATCH = "mismatch" + QUOTA = "quota" + + +@dataclass(frozen=True) +class GroundingMetrics: + """Separate resolver outcomes for operational visibility.""" + + attempted_count: int + miss_count: int + mismatch_guard_count: int + cache_hit_count: int + did_reach_deadline: bool + did_exhaust_quota: bool + + @property + def miss_rate(self) -> float: + """Return the share of attempted candidates with no search result.""" + return self.miss_count / self.attempted_count if self.attempted_count else 0.0 + + @property + def mismatch_guard_rate(self) -> float: + """Return the share rejected by client-side identity checks.""" + return self.mismatch_guard_count / self.attempted_count if self.attempted_count else 0.0 + + +@dataclass(frozen=True) +class GroundingResult: + """Resolved pool and its operational metrics.""" + + tracks: tuple[Track, ...] + metrics: GroundingMetrics + + +@dataclass(frozen=True) +class _ResolutionAttempt: + index: int + status: ResolutionStatus + track: Track | None = None + is_cache_hit: bool = False + + +class ResolutionCache: + """Bound successful name resolutions by age and least-recent use.""" + + def __init__( + self, + ttl_seconds: float, + max_entries: int, + clock: Callable[[], float] = time.monotonic, + ) -> None: + """Create an empty successful-resolution cache.""" + self.ttl_seconds = ttl_seconds + self.max_entries = max_entries + self.clock = clock + self._entries: OrderedDict[str, tuple[float, Track]] = OrderedDict() + + def get(self, key: str) -> Track | None: + """Return a fresh cached track and refresh its recency.""" + entry = self._entries.get(key) + if entry is None: + return None + created_at, track = entry + if self.clock() - created_at >= self.ttl_seconds: + del self._entries[key] + return None + self._entries.move_to_end(key) + increment_cache_hits() + return track + + def put(self, key: str, track: Track) -> None: + """Cache a successful unambiguous resolution.""" + if self.max_entries <= 0 or self.ttl_seconds <= 0: + return + self._entries[key] = (self.clock(), track) + self._entries.move_to_end(key) + while len(self._entries) > self.max_entries: + self._entries.popitem(last=False) + + +class Grounder: + """Build a safe Spotify pool with early stop and bounded fan-out.""" + + def __init__(self, settings: Settings, cache: ResolutionCache | None = None) -> None: + """Bind resolver settings and a process-local resolution cache.""" + self.settings = settings + self.cache = cache or ResolutionCache( + settings.resolution_cache_ttl_seconds, + settings.resolution_cache_max_entries, + ) + + async def ground( + self, + catalog: MusicCatalog, + candidates: tuple[TrackCandidate, ...], + known_track_ids: frozenset[str], + familiarity: Familiarity, + pool_target: int, + ) -> GroundingResult: + """Resolve candidates until the target, deadline, or quota boundary.""" + accepted: dict[int, Track] = {} + seen_ids: set[str] = set() + seen_keys: set[str] = set() + 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 + next_index = self._launch_tasks( + catalog, + candidates, + pending, + next_index, + semaphore, + ) + if not pending: + break + remaining_seconds = deadline_at - time.monotonic() + if remaining_seconds <= 0: + metrics.did_reach_deadline = True + break + done, _ = await asyncio.wait( + pending, + timeout=remaining_seconds, + return_when=asyncio.FIRST_COMPLETED, + ) + if not done: + metrics.did_reach_deadline = True + break + self._collect_done( + done, + pending, + accepted, + seen_ids, + seen_keys, + known_track_ids, + familiarity, + pool_target, + metrics, + ) + if metrics.did_exhaust_quota: + break + finally: + await _cancel_tasks(tuple(pending)) + + tracks = tuple(track for _, track in sorted(accepted.items())) + if familiarity is Familiarity.FAMILIAR: + tracks = tuple(sorted(tracks, key=lambda track: track.id not in known_track_ids)) + result = GroundingResult(tracks=tracks, metrics=metrics.freeze()) + _log_grounding(result) + return result + + def _launch_tasks( + self, + catalog: MusicCatalog, + 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) + ) + pending[task] = next_index + next_index += 1 + return next_index + + def _collect_done( + self, + done: set[asyncio.Task[_ResolutionAttempt]], + pending: dict[asyncio.Task[_ResolutionAttempt], int], + accepted: dict[int, Track], + seen_ids: set[str], + seen_keys: set[str], + known_track_ids: frozenset[str], + familiarity: Familiarity, + pool_target: int, + metrics: "_MutableMetrics", + ) -> None: + for task in sorted(done, key=pending.__getitem__): + del pending[task] + attempt = task.result() + metrics.record(attempt) + if attempt.status is ResolutionStatus.QUOTA: + continue + track = attempt.track + if track is None or len(accepted) >= pool_target: + continue + if familiarity is Familiarity.NEW and track.id in known_track_ids: + continue + normalized_key = track_key(track) + if track.id in seen_ids or normalized_key in seen_keys: + continue + seen_ids.add(track.id) + seen_keys.add(normalized_key) + accepted[attempt.index] = track + + async def _resolve( + self, + catalog: MusicCatalog, + index: int, + candidate: TrackCandidate, + semaphore: asyncio.Semaphore, + ) -> _ResolutionAttempt: + async with semaphore: + return await self._resolve_with_slot(catalog, index, candidate) + + async def _resolve_with_slot( + self, + catalog: MusicCatalog, + index: int, + candidate: TrackCandidate, + ) -> _ResolutionAttempt: + key = candidate_key(candidate) + cached_track = self.cache.get(key) + if cached_track is not None: + return _ResolutionAttempt( + index=index, + status=ResolutionStatus.RESOLVED, + track=cached_track, + is_cache_hit=True, + ) + + try: + field_results = await catalog.search_tracks(_field_query(candidate)) + matched_track = _best_match( + candidate, field_results, self.settings.title_similarity_threshold + ) + if matched_track is None: + bare_results = await catalog.search_tracks(f"{candidate.title} {candidate.artist}") + matched_track = _best_match( + candidate, + bare_results, + self.settings.title_similarity_threshold, + ) + else: + bare_results = [] + except CatalogQuotaExhaustedError: + return _ResolutionAttempt(index=index, status=ResolutionStatus.QUOTA) + + if matched_track is not None: + self.cache.put(key, matched_track) + return _ResolutionAttempt( + index=index, status=ResolutionStatus.RESOLVED, track=matched_track + ) + status = ( + ResolutionStatus.MISMATCH if field_results or bare_results else ResolutionStatus.MISS + ) + return _ResolutionAttempt(index=index, status=status) + + +@dataclass +class _MutableMetrics: + attempted_count: int = 0 + miss_count: int = 0 + mismatch_guard_count: int = 0 + cache_hit_count: int = 0 + did_reach_deadline: bool = False + did_exhaust_quota: bool = False + + def record(self, attempt: _ResolutionAttempt) -> None: + self.attempted_count += 1 + self.miss_count += attempt.status is ResolutionStatus.MISS + self.mismatch_guard_count += attempt.status is ResolutionStatus.MISMATCH + self.cache_hit_count += attempt.is_cache_hit + self.did_exhaust_quota = self.did_exhaust_quota or attempt.status is ResolutionStatus.QUOTA + + def freeze(self) -> GroundingMetrics: + return GroundingMetrics( + attempted_count=self.attempted_count, + miss_count=self.miss_count, + mismatch_guard_count=self.mismatch_guard_count, + cache_hit_count=self.cache_hit_count, + did_reach_deadline=self.did_reach_deadline, + did_exhaust_quota=self.did_exhaust_quota, + ) + + +def _best_match(candidate: TrackCandidate, tracks: list[Track], threshold: float) -> Track | None: + verdicts = ((judge_candidate_match(candidate, track, threshold), track) for track in tracks) + accepted = [ + (verdict.title_similarity, track) for verdict, track in verdicts if verdict.is_match + ] + return max(accepted, key=lambda item: item[0])[1] if accepted else None + + +def _field_query(candidate: TrackCandidate) -> str: + title = candidate.title.replace('"', " ") + artist = candidate.artist.replace('"', " ") + return f'track:"{title}" artist:"{artist}"' + + +async def _cancel_tasks(tasks: tuple[asyncio.Task[_ResolutionAttempt], ...]) -> None: + for task in tasks: + task.cancel() + if tasks: + await asyncio.gather(*tasks, return_exceptions=True) + + +def _log_grounding(result: GroundingResult) -> None: + metrics = result.metrics + structlog.get_logger().info( + "grounding_complete", + track_count=len(result.tracks), + attempted_count=metrics.attempted_count, + miss_rate=metrics.miss_rate, + mismatch_guard_rate=metrics.mismatch_guard_rate, + 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 new file mode 100644 index 0000000..bfc1cba --- /dev/null +++ b/backend/app/pipeline/orchestrator.py @@ -0,0 +1,255 @@ +"""Compose taste, intent, grounding, and reranking into streamed events.""" + +import asyncio +import time +from collections.abc import AsyncGenerator +from contextlib import aclosing +from dataclasses import dataclass + +import structlog + +from app.config import Settings +from app.domain.models import ( + CompressedTasteProfile, + ConversationTurn, + Intent, + PreviousRecommendation, + TasteProfile, + Track, +) +from app.domain.profile import compress_taste_profile +from app.observability.timing import increment_cache_hits +from app.pipeline.event import ( + PipelineDoneEvent, + PipelineErrorEvent, + PipelineEvent, + PipelineMetadataEvent, + PipelineTrackEvent, + PipelineWarningEvent, +) +from app.pipeline.grounding import Grounder +from app.ports.protocols import MusicCatalog, Recommender, RecommenderOutputError + +RERANK_FALLBACK_CODE = "rerank_fallback" +RERANK_FALLBACK_MESSAGE = "Ranking output was invalid, so grounded results are shown instead." +RERANK_FALLBACK_JUSTIFICATION = "Selected as a grounded match for your request." + + +@dataclass +class _RankingState: + selected_ids: set[str] + selected_tracks: list[Track] + + @property + def track_count(self) -> int: + return len(self.selected_tracks) + + +class RecommendationPipeline: + """Orchestrate the code-defined recommendation stages.""" + + def __init__( + self, + recommender: Recommender, + settings: Settings, + grounder: Grounder | None = None, + ) -> None: + """Create process-local caches around the provided service ports.""" + self.recommender = recommender + self.settings = settings + self.grounder = grounder or Grounder(settings) + self.taste_cache = _TasteProfileCache(settings) + self.last_pools: dict[str, tuple[Track, ...]] = {} + + async def stream( + self, + session_id: str, + request_id: str, + catalog: MusicCatalog, + query: str, + history: tuple[ConversationTurn, ...], + previous_recommendations: tuple[PreviousRecommendation, ...], + ) -> 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, + ) + yield PipelineMetadataEvent( + request_id=request_id, + intent_summary=intent.intent_summary, + candidate_count=len(intent.candidates), + ) + + pool = self.last_pools.get(session_id) if intent.is_refinement else None + if pool is None: + result = await self.grounder.ground( + catalog, + intent.candidates, + taste.known_track_ids, + intent.familiarity, + self.settings.rerank_count + self.settings.rerank_pool_buffer, + ) + pool = result.tracks + if pool: + self.last_pools[session_id] = pool + + if len(pool) < self.settings.grounding_floor: + yield PipelineErrorEvent( + code="insufficient_grounding", + message="Not enough requested tracks could be verified safely.", + ) + return + + state = _RankingState(selected_ids=set(), selected_tracks=[]) + async for event in self._stream_ranking(intent, pool, taste.text, history, state): + yield event + + new_track_count = sum( + track.id not in taste.known_track_ids for track in state.selected_tracks + ) + structlog.get_logger().info( + "recommendations_complete", + track_count=state.track_count, + new_track_count=new_track_count, + ) + yield PipelineDoneEvent( + track_count=state.track_count, + total_ms=round((time.monotonic() - started_at) * 1000), + ) + + async def _stream_ranking( + self, + intent: Intent, + pool: tuple[Track, ...], + taste_summary: str, + history: tuple[ConversationTurn, ...], + state: _RankingState, + ) -> AsyncGenerator[PipelineTrackEvent | PipelineWarningEvent]: + correction: str | None = None + for _attempt_number in range(2): + remaining_count = self.settings.rerank_count - state.track_count + if remaining_count <= 0: + return + try: + async for event in self._validated_rerank( + intent, + pool, + taste_summary, + history, + remaining_count, + correction, + state, + ): + yield event + return + except RecommenderOutputError as error: + emitted_ids = ", ".join(sorted(state.selected_ids)) or "none" + correction = ( + f"Validation failed: {error}. Already emitted track ids: {emitted_ids}." + ) + + yield PipelineWarningEvent(code=RERANK_FALLBACK_CODE, message=RERANK_FALLBACK_MESSAGE) + for track in pool: + if state.track_count >= self.settings.rerank_count: + break + if track.id in state.selected_ids: + continue + state.selected_ids.add(track.id) + state.selected_tracks.append(track) + yield PipelineTrackEvent( + rank=state.track_count, + track=track, + justification=RERANK_FALLBACK_JUSTIFICATION, + ) + + async def _validated_rerank( + self, + intent: Intent, + pool: tuple[Track, ...], + taste_summary: str, + history: tuple[ConversationTurn, ...], + selection_count: int, + correction: str | None, + state: _RankingState, + ) -> AsyncGenerator[PipelineTrackEvent]: + tracks_by_id = {track.id: track for track in pool} + stream = self.recommender.stream_rerank( + intent, + pool, + taste_summary, + history, + selection_count, + correction, + ) + async with aclosing(stream) as selections: + async for selection in selections: + if state.track_count >= self.settings.rerank_count: + raise RecommenderOutputError("Rerank returned too many track ids") + track = tracks_by_id.get(selection.track_id) + if track is None: + raise RecommenderOutputError("Rerank selected an out-of-pool track id") + if track.id in state.selected_ids: + raise RecommenderOutputError("Rerank selected a duplicate track id") + state.selected_ids.add(track.id) + state.selected_tracks.append(track) + yield PipelineTrackEvent( + rank=state.track_count, + track=track, + justification=selection.justification, + ) + + +class _TasteProfileCache: + def __init__(self, settings: Settings) -> None: + self.settings = settings + self._entries: dict[str, tuple[float, CompressedTasteProfile]] = {} + self._locks: dict[str, asyncio.Lock] = {} + + async def get(self, session_id: str, catalog: MusicCatalog) -> CompressedTasteProfile: + cached = self._fresh_entry(session_id) + if cached is not None: + increment_cache_hits() + return cached + lock = self._locks.setdefault(session_id, asyncio.Lock()) + async with lock: + cached = self._fresh_entry(session_id) + if cached is not None: + increment_cache_hits() + return cached + compressed = await self._fetch(catalog) + self._entries[session_id] = (time.monotonic(), compressed) + return compressed + + def _fresh_entry(self, session_id: str) -> CompressedTasteProfile | None: + entry = self._entries.get(session_id) + if entry is None: + return None + created_at, profile = entry + if time.monotonic() - created_at >= self.settings.taste_profile_ttl_seconds: + del self._entries[session_id] + return None + 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), + ) + 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), + ) + ) diff --git a/backend/app/ports/protocols.py b/backend/app/ports/protocols.py index dc40d3d..0dbf7ff 100644 --- a/backend/app/ports/protocols.py +++ b/backend/app/ports/protocols.py @@ -1,6 +1,6 @@ """Structural ports implemented by external service adapters.""" -from collections.abc import AsyncIterator +from collections.abc import AsyncGenerator from typing import Literal, Protocol from app.domain.models import ( @@ -65,7 +65,7 @@ class Recommender(Protocol): history: tuple[ConversationTurn, ...], selection_count: int, correction: str | None = None, - ) -> AsyncIterator[RerankSelection]: + ) -> AsyncGenerator[RerankSelection]: """Stream validated selections from the grounded pool.""" ... diff --git a/backend/tests/test_grounding.py b/backend/tests/test_grounding.py new file mode 100644 index 0000000..e5b896e --- /dev/null +++ b/backend/tests/test_grounding.py @@ -0,0 +1,153 @@ +"""Deterministic tests for bounded Spotify grounding.""" + +import asyncio +from collections.abc import Awaitable, Callable + +from app.config import Settings +from app.domain.models import Familiarity, Track, TrackCandidate +from app.pipeline.grounding import Grounder +from app.ports.protocols import TimeRange + +SearchHandler = Callable[[str], Awaitable[list[Track]]] + + +class FakeCatalog: + """Expose a programmable search surface for grounding tests.""" + + def __init__(self, search_handler: SearchHandler) -> None: + """Store the search behavior and call trace.""" + self.search_handler = search_handler + self.search_queries: list[str] = [] + + async def search_tracks(self, query: str, limit: int = 10) -> list[Track]: + """Record and delegate one fake search.""" + self.search_queries.append(query) + return await self.search_handler(query) + + async def fetch_top_artists(self, time_range: TimeRange, limit: int) -> list[str]: + """Return no top artists.""" + return [] + + async def fetch_top_tracks(self, time_range: TimeRange, limit: int) -> list[Track]: + """Return no top tracks.""" + return [] + + async def fetch_saved_tracks(self, limit: int) -> list[Track]: + """Return no saved tracks.""" + return [] + + +def test_early_stop_honors_pool_target() -> None: + async def run() -> None: + async def search(query: str) -> list[Track]: + 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"track-{index}", "Artist") for index in range(6)) + + result = await grounder.ground(catalog, candidates, frozenset(), Familiarity.MIX, 2) + + assert len(result.tracks) == 2 + assert len(catalog.search_queries) == 2 + + asyncio.run(run()) + + +def test_miss_and_mismatch_are_counted_separately() -> None: + async def run() -> None: + async def search(query: str) -> list[Track]: + if "Missing" in query: + return [] + return [_track("wrong", "Different Song", "Different Artist")] + + catalog = FakeCatalog(search) + candidates = ( + TrackCandidate("Missing", "Artist"), + TrackCandidate("Rejected", "Artist"), + ) + + result = await Grounder(_settings()).ground( + catalog, + candidates, + frozenset(), + Familiarity.MIX, + 2, + ) + + assert result.metrics.miss_count == 1 + assert result.metrics.mismatch_guard_count == 1 + assert result.metrics.attempted_count == 2 + + asyncio.run(run()) + + +def test_resolution_cache_hit_skips_catalog() -> None: + async def run() -> None: + async def search(query: str) -> list[Track]: + return [_track("cached", "Cached Song", "Artist")] + + catalog = FakeCatalog(search) + grounder = Grounder(_settings()) + candidates = (TrackCandidate("Cached Song", "Artist"),) + + await grounder.ground(catalog, candidates, frozenset(), Familiarity.MIX, 1) + first_call_count = len(catalog.search_queries) + second = await grounder.ground(catalog, candidates, frozenset(), Familiarity.MIX, 1) + + assert len(catalog.search_queries) == first_call_count + assert second.metrics.cache_hit_count == 1 + + asyncio.run(run()) + + +def test_deadline_returns_resolved_partial_pool() -> None: + async def run() -> None: + never_finishes = asyncio.Event() + + async def search(query: str) -> list[Track]: + if "Slow Song" in query: + await never_finishes.wait() + return [_track("fast", "Fast Song", "Artist")] + + catalog = FakeCatalog(search) + settings = _settings(grounding_concurrency=2, request_deadline_seconds=0.02) + candidates = ( + TrackCandidate("Fast Song", "Artist"), + TrackCandidate("Slow Song", "Artist"), + ) + + result = await Grounder(settings).ground( + catalog, + candidates, + frozenset(), + Familiarity.MIX, + 2, + ) + + assert [track.id for track in result.tracks] == ["fast"] + assert result.metrics.did_reach_deadline + + asyncio.run(run()) + + +def _settings(**overrides: object) -> Settings: + values: dict[str, object] = { + "grounding_concurrency": 1, + "request_deadline_seconds": 1.0, + } + values.update(overrides) + return Settings.model_validate(values) + + +def _track(track_id: str, title: str, artist: str) -> Track: + return Track( + id=track_id, + uri=f"spotify:track:{track_id}", + title=title, + artists=(artist,), + album_name="Album", + album_art_url=None, + external_url=None, + ) diff --git a/backend/tests/test_orchestrator.py b/backend/tests/test_orchestrator.py new file mode 100644 index 0000000..5dd0189 --- /dev/null +++ b/backend/tests/test_orchestrator.py @@ -0,0 +1,245 @@ +"""End-to-end pipeline tests using deterministic service fakes.""" + +import asyncio +from collections.abc import AsyncGenerator + +from app.config import Settings +from app.domain.models import ( + ConversationTurn, + Familiarity, + Intent, + PreviousRecommendation, + RerankSelection, + Track, + TrackCandidate, +) +from app.pipeline.event import PipelineEvent, PipelineTrackEvent +from app.pipeline.orchestrator import RecommendationPipeline +from app.ports.protocols import RecommenderOutputError, TimeRange + + +class FakeCatalog: + """Return exact tracks and a configurable known-track sample.""" + + def __init__(self, tracks: tuple[Track, ...], known_tracks: tuple[Track, ...] = ()) -> None: + """Index tracks by title and expose taste-call counters.""" + self.tracks_by_title = {track.title: track for track in tracks} + self.known_tracks = known_tracks + self.search_call_count = 0 + self.taste_call_count = 0 + + async def search_tracks(self, query: str, limit: int = 10) -> list[Track]: + """Resolve an exact fielded title and miss bare fallbacks.""" + self.search_call_count += 1 + if 'track:"' not in query: + return [] + title = query.split('track:"', 1)[1].split('"', 1)[0] + track = self.tracks_by_title.get(title) + return [track] if track is not None else [] + + async def fetch_top_artists(self, time_range: TimeRange, limit: int) -> list[str]: + """Return one stable taste artist.""" + self.taste_call_count += 1 + return ["Taste Artist"] + + async def fetch_top_tracks(self, time_range: TimeRange, limit: int) -> list[Track]: + """Return no top tracks.""" + self.taste_call_count += 1 + return [] + + async def fetch_saved_tracks(self, limit: int) -> list[Track]: + """Return the configured known tracks.""" + self.taste_call_count += 1 + return list(self.known_tracks) + + +class FakeRecommender: + """Return fixed intents and either selections or structured failures.""" + + def __init__( + self, + intents: list[Intent], + selection_ids: tuple[str, ...] = (), + failure_count: int = 0, + ) -> None: + """Store deterministic outputs for successive calls.""" + self.intents = intents + self.selection_ids = selection_ids + self.failure_count = failure_count + self.rerank_call_count = 0 + + async def create_intent( + self, + query: str, + history: tuple[ConversationTurn, ...], + previous_recommendations: tuple[PreviousRecommendation, ...], + taste_summary: str, + candidate_count: int, + ) -> Intent: + """Return the next fixed intent.""" + return self.intents.pop(0) + + 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]: + """Stream configured ids or fail before yielding.""" + self.rerank_call_count += 1 + if self.failure_count: + self.failure_count -= 1 + raise RecommenderOutputError("invalid test output") + selected_ids = self.selection_ids or tuple(track.id for track in grounded_tracks) + for track_id in selected_ids[:selection_count]: + yield RerankSelection(track_id, f"Reason for {track_id}") + + +def test_event_order_and_rerank_ids_stay_inside_grounded_pool() -> None: + async def run() -> None: + first = _track("first", "First Song") + second = _track("second", "Second Song") + catalog = FakeCatalog((first, second)) + recommender = FakeRecommender([_intent(first, second)], ("second", "first")) + + events = await _run_pipeline(catalog, recommender) + + assert [event.type for event in events] == ["metadata", "track", "track", "done"] + track_events = [event for event in events if isinstance(event, PipelineTrackEvent)] + assert [event.track.id for event in track_events] == ["second", "first"] + assert {event.track.id for event in track_events} <= {"first", "second"} + + asyncio.run(run()) + + +def test_rerank_fallback_warns_then_streams_grounded_order() -> None: + async def run() -> None: + first = _track("first", "First Song") + second = _track("second", "Second Song") + catalog = FakeCatalog((first, second)) + recommender = FakeRecommender([_intent(first, second)], failure_count=2) + + events = await _run_pipeline(catalog, recommender) + + assert [event.type for event in events] == [ + "metadata", + "warning", + "track", + "track", + "done", + ] + assert recommender.rerank_call_count == 2 + track_events = [event for event in events if isinstance(event, PipelineTrackEvent)] + assert [event.track.id for event in track_events] == ["first", "second"] + + asyncio.run(run()) + + +def test_new_familiarity_excludes_known_track_ids() -> None: + async def run() -> None: + known = _track("known", "Known Song") + new = _track("new", "New Song") + catalog = FakeCatalog((known, new), known_tracks=(known,)) + intent = _intent(known, new, familiarity=Familiarity.NEW) + recommender = FakeRecommender([intent]) + + events = await _run_pipeline(catalog, recommender, rerank_count=1) + + track_events = [event for event in events if isinstance(event, PipelineTrackEvent)] + assert [event.track.id for event in track_events] == ["new"] + + asyncio.run(run()) + + +def test_refinement_reuses_last_grounded_pool_and_cached_taste() -> None: + async def run() -> None: + track = _track("first", "First Song") + initial = _intent(track) + refinement = _intent(track, is_refinement=True) + catalog = FakeCatalog((track,)) + recommender = FakeRecommender([initial, refinement]) + pipeline = _pipeline(recommender, rerank_count=1) + + await _collect(pipeline, catalog, "first request") + initial_search_calls = catalog.search_call_count + initial_taste_calls = catalog.taste_call_count + await _collect(pipeline, catalog, "refine it") + + assert catalog.search_call_count == initial_search_calls + assert catalog.taste_call_count == initial_taste_calls + + asyncio.run(run()) + + +async def _run_pipeline( + catalog: FakeCatalog, + recommender: FakeRecommender, + rerank_count: int = 2, +) -> list[PipelineEvent]: + return await _collect(_pipeline(recommender, rerank_count=rerank_count), catalog, "query") + + +def _pipeline(recommender: FakeRecommender, rerank_count: int) -> RecommendationPipeline: + return RecommendationPipeline( + recommender, + Settings( + rerank_count=rerank_count, + rerank_pool_buffer=0, + grounding_floor=1, + grounding_concurrency=2, + request_deadline_seconds=1.0, + ), + ) + + +async def _collect( + pipeline: RecommendationPipeline, + catalog: FakeCatalog, + query: str, +) -> list[PipelineEvent]: + return [ + event + async for event in pipeline.stream( + "session", + "request", + catalog, + query, + (), + (), + ) + ] + + +def _intent( + *tracks: Track, + familiarity: Familiarity = Familiarity.MIX, + is_refinement: bool = False, +) -> Intent: + return Intent( + mood=("focused",), + activity=None, + era=(), + languages=(), + genres=("electronic",), + familiarity=familiarity, + is_refinement=is_refinement, + intent_summary="Focused electronic discovery.", + candidates=tuple( + TrackCandidate(title=track.title, artist=track.artists[0]) for track in tracks + ), + ) + + +def _track(track_id: str, title: str) -> Track: + return Track( + id=track_id, + uri=f"spotify:track:{track_id}", + title=title, + artists=("Artist",), + album_name="Album", + album_art_url=None, + external_url=None, + ) diff --git a/docs/logboek.md b/docs/logboek.md index d52d76c..fc8bcfb 100644 --- a/docs/logboek.md +++ b/docs/logboek.md @@ -2,8 +2,8 @@ Bijgehouden tijdens de bouw. Per blok: wat ik deed, waarom, wat ik heb laten vallen. - -## Opzet (avond dag 1) +## Dag 1 - korte sessie in de avond +### Opzet Wat ik deed: @@ -34,7 +34,8 @@ Wat ik heb laten vallen of uitgesteld: waar ze horen. Ik probeer op die manier bewust vroeg drift en dode code te voorkomen. - Geen apart beslisdocument. De motivering staat in de README en hier. -## Spotify-koppeling (ochtend dag 2) +## Dag 2 +### Spotify-koppeling Wat ik deed: @@ -69,3 +70,39 @@ Wat ik heb laten vallen of uitgesteld: pipeline-stap; daar bestaat het ontwerp pas echt. - OpenAPI-codegen voor het contract overwogen en afgewezen: de kern van dit contract is de event-stream en die modelleert OpenAPI niet. + +### Pipeline + +Wat ik deed: + +- Domeinbasis: titel/artiest-matching (normalisatie, met tolerantie voor + Spotify's versie-suffixen zoals "- Remaster 2023"), compressie van het + Spotify-smaakprofiel naar prompttekst plus een set bekende track-ids, + prompts als data in een eigen module, en alle instelbare waarden + gesectioneerd in de config met per waarde het waarom. +- Twee LLM-aanroepen achter een eigen interface: aanroep 1 interpreteert de + vraag (stemming, activiteit, taal, bekendheid) en stelt 30-40 echte + nummers voor als gestructureerde output; aanroep 2 herordent uitsluitend + geverifieerde nummers en streamt per nummer een eerlijke onderbouwing. + Elke output wordt gevalideerd, met hooguit 1 herstelpoging. +- Grounding: begrensde parallelle zoekslag met vroege stop, een deadline, + een naam-naar-id cache en twee aparte metrieken: niet gevonden versus + wel gevonden maar afgekeurd door de controle. Een track-id dat niet in + de geverifieerde pool zit kan nooit bij de gebruiker terechtkomen. + +Waarom: + +- De LLM is hier de aanbeveler, maar mag alleen creatief zijn tussen twee + deterministische muren: alles wat hij ziet is echte data, alles wat de + gebruiker ziet is geverifieerd op Spotify. Een verzonnen nummer valt + stilletjes af en verschijnt nooit. +- Zoeken geeft maximaal 10 resultaten per aanroep, dus resolutie is per + definitie een fan-out; liever een kandidaat laten vallen dan het + verkeerde nummer aanbevelen. + +Wat ik heb laten vallen of uitgesteld: + +- Verfijnvragen doen geen nieuwe zoekslag: turn 2 herordent de bestaande + geverifieerde pool. Sneller en consistent, maar een verfijning haalt + geen nieuwe nummers op. Dit is een bewuste afweging, mocht er tijd over + zijn is dit 1 van de uitbreidingen die ik op zou kunnen pakken.