diff --git a/experiments/active/2026-07-14-cold-start-candidate-filter.md b/experiments/active/2026-07-14-cold-start-candidate-filter.md new file mode 100644 index 00000000..ca091ca7 --- /dev/null +++ b/experiments/active/2026-07-14-cold-start-candidate-filter.md @@ -0,0 +1,96 @@ +# Experiment: Cold Start Candidate Filter + +Created: 2026-07-14 +Status: active - implementation in PR +Owner: ceo +Implementer: CTO +Implementation Issue: FFM-1882 +Verification Issue: FFM-1883 +Deployed: pending +Measure after: 2026-07-28 + +## Hypothesis + +True-new users are not primarily failing on the first meme. The July 13 first-10 +readout showed first-meme continuation at 87.5%, but only 58.9% of true-new +users reached five sends and 26.8% reached ten. If the cold-start candidate +pools exclude weak source slices for positions 2-10, reached-10 and first-10 +continuation should improve without changing the mature-user feed. + +## Changes Planned + +- Scope the treatment to true-new cold-start users only. +- Add a feature-flagged treatment path for cold-start candidate selection. +- Start with source-level guardrails from the July 13 readout: + `https://vk.com/dfzwe4`, `https://vk.com/eternalclassic`, + `https://t.me/ukrmemesmineproblemes`, and `https://t.me/hindi_jokes_desi_memes`. +- Avoid a broad language reweight in v1. EN/RU mix should be measured as a + guardrail, not forced from a 56-user sample. +- Preserve existing fallbacks so empty pools route to the current + `text_light_lr_smoothed` / `best_uploaded_memes` behavior rather than + starving users. +- Keep `recently_liked_blender_v2` untouched because it is a mature-user + experiment with its own 2026-07-22 checkpoint. + +## Implementation Notes + +- Rollback: set `COLD_START_CANDIDATE_GUARDRAILS_ENABLED=false`. +- Treatment attribution: guarded candidates are persisted as + `cold_start_explore_guarded` or `cold_start_adapt_guarded` in + `user_meme_reaction.recommended_by`. +- Fallback rate: use recommendation diagnostics `fallback_used`, grouped by + `cold_start_candidate_guardrails_applied`. + +## Metrics Before + +2026-07-13 readout, 14-day true-new cohort: + +- Cohort users: 56. +- First-meme LR: 41.4%; first-meme continuation: 87.5%. +- First-10 LR: 39.5%; first-10 continuation: 69.8%; first-10 quality score: -0.179. +- Reached 5 users: 33/56 (58.9%). +- Reached 10 users: 15/56 (26.8%). +- Second-session users: 31/56 (55.4%). +- Position 2 continuation: 62.0%. +- Positions 6-7 continuation: 51.9-52.2%. +- 2026-07-14 health: North Star 21.0 vs previous 21.0, no severity incident; + `recently_liked_blender_v2` sample gate remains unmet at 357 control / 400 treatment. + +## Success Criteria + +- Treatment reached-10 improves from 26.8% to at least 35% after the + 2026-07-28 checkpoint, or reaches a directionally positive lift with at least + 30 treatment users. +- Positions 2-5 continuation improves by at least 5pp versus baseline/control. +- First-meme continuation does not fall below 84%. +- First-10 LR does not fall by more than 3pp. +- No product-health incident: North Star does not drop by more than 10% week + over week, stats remain fresh, and candidate pool fallback rate does not spike. + +## Failure Criteria + +- Treatment reached-10 is flat or worse after the checkpoint with enough users. +- First-meme continuation drops below 84% or first-10 LR drops by more than 3pp. +- Cold-start pool exhaustion causes fallback-heavy recommendations or visible + delivery issues. +- The implementation changes mature-user routing, `recently_liked_blender_v2`, + moderator low-sent quota, or unrelated engines. + +## Guardrails + +- Feature flag or equivalent fast rollback path required before production exposure. +- Report source/language mix, fallback rate, first-10 positions, reached-5, + reached-10, second-session users, first-meme continuation, first-10 LR, and + first-10 continuation. +- Preferred Analyst checkpoint artifact: + `experiments/reports/2026-07-28-cold-start-candidate-filter-readout.md`. +- No Comms post until the experiment produces a user-facing result, not just an + internal filter launch. + +## Metrics After + +Pending 2026-07-28 readout. + +## Conclusion + +Pending. diff --git a/src/config.py b/src/config.py index 76dbd8b7..5697dc11 100644 --- a/src/config.py +++ b/src/config.py @@ -55,6 +55,9 @@ class Config(BaseSettings): # users (nsessions <= 1). Dormant returners with nmemes_sent < 30 but # multiple sessions fall through to the growing-user blender instead. COLD_START_NSESSIONS_GATE_ENABLED: bool = True + # FFM-1882: narrow true-new cold-start positions 2-10 experiment. + # Roll back by setting this to false in production env. + COLD_START_CANDIDATE_GUARDRAILS_ENABLED: bool = True # Recommendation batch diagnostics are realtime operational data, not # durable product facts. Keep compact logs/spans always on and sample full diff --git a/src/recommendations/candidates.py b/src/recommendations/candidates.py index 8ff5544d..3dd9b058 100644 --- a/src/recommendations/candidates.py +++ b/src/recommendations/candidates.py @@ -16,6 +16,16 @@ GOAT_MIN_AGE_DAYS = 3 # days since created_at — must have aged enough to accumulate reactions GOAT_RECENTLY_SENT_WINDOW_DAYS = 30 TEXT_LIGHT_MAX_OCR_WORDS = 30 +COLD_START_GUARDRAIL_SOURCE_URLS = ( + "https://vk.com/dfzwe4", + "https://vk.com/eternalclassic", + "https://t.me/ukrmemesmineproblemes", + "https://t.me/hindi_jokes_desi_memes", +) +COLD_START_EXPLORE_RECOMMENDED_BY = "cold_start_explore" +COLD_START_ADAPT_RECOMMENDED_BY = "cold_start_adapt" +COLD_START_EXPLORE_GUARDED_RECOMMENDED_BY = "cold_start_explore_guarded" +COLD_START_ADAPT_GUARDED_RECOMMENDED_BY = "cold_start_adapt_guarded" _OCR_TEXT_SQL = "trim(coalesce(M.ocr_result->>'text', M.ocr_result->'raw_result'->>'ocr_text', ''))" TEXT_LIGHT_OCR_FILTER_SQL = f""" @@ -26,6 +36,9 @@ END ) <= :text_light_max_ocr_words """ +COLD_START_GUARDRAIL_SOURCE_FILTER_SQL = """ + AND NOT (S.url = ANY(:cold_start_guardrail_source_urls)) +""" def _build_params( @@ -41,6 +54,38 @@ def _build_params( return params +def _cold_start_recommended_by(engine: str, candidate_guardrails_enabled: bool) -> str: + if engine == COLD_START_EXPLORE_RECOMMENDED_BY and candidate_guardrails_enabled: + return COLD_START_EXPLORE_GUARDED_RECOMMENDED_BY + if engine == COLD_START_ADAPT_RECOMMENDED_BY and candidate_guardrails_enabled: + return COLD_START_ADAPT_GUARDED_RECOMMENDED_BY + return engine + + +def _cold_start_guardrail_source_filter(candidate_guardrails_enabled: bool) -> str: + return COLD_START_GUARDRAIL_SOURCE_FILTER_SQL if candidate_guardrails_enabled else "" + + +def _cold_start_params( + user_id: int, + limit: int, + exclude_meme_ids: list[int], + *, + engine: str, + candidate_guardrails_enabled: bool, +) -> dict[str, Any]: + params = _build_params( + user_id, + limit, + exclude_meme_ids, + recommended_by=_cold_start_recommended_by(engine, candidate_guardrails_enabled), + text_light_max_ocr_words=TEXT_LIGHT_MAX_OCR_WORDS, + ) + if candidate_guardrails_enabled: + params["cold_start_guardrail_source_urls"] = list(COLD_START_GUARDRAIL_SOURCE_URLS) + return params + + async def best_uploaded_memes( user_id: int, limit: int = 10, @@ -413,6 +458,7 @@ async def cold_start_explore( user_id: int, limit: int = 5, exclude_meme_ids: list[int] = [], + candidate_guardrails_enabled: bool = False, ): """Phase 1 cold start: quality-first selection for new user first impression. @@ -436,12 +482,14 @@ async def cold_start_explore( SELECT M.id , M.type, M.telegram_file_id, M.caption - , 'cold_start_explore' AS recommended_by + , :recommended_by AS recommended_by , COALESCE(MS.nlikes, 0) AS nlikes FROM meme M INNER JOIN meme_stats MS ON MS.meme_id = M.id + INNER JOIN meme_source S + ON S.id = M.meme_source_id INNER JOIN user_language L ON L.language_code = M.language_code AND L.user_id = :user_id @@ -455,6 +503,7 @@ async def cold_start_explore( AND (MS.nlikes + MS.ndislikes) >= 20 AND MS.lr_smoothed >= 0.10 {TEXT_LIGHT_OCR_FILTER_SQL} + {_cold_start_guardrail_source_filter(candidate_guardrails_enabled)} {exclude_meme_ids_sql_filter(exclude_meme_ids)} ORDER BY (MS.nlikes::float / NULLIF(MS.nlikes + MS.ndislikes, 0)) DESC, @@ -463,11 +512,12 @@ async def cold_start_explore( """ return await fetch_all( text(query), - _build_params( + _cold_start_params( user_id, limit, exclude_meme_ids, - text_light_max_ocr_words=TEXT_LIGHT_MAX_OCR_WORDS, + engine=COLD_START_EXPLORE_RECOMMENDED_BY, + candidate_guardrails_enabled=candidate_guardrails_enabled, ), ) @@ -476,6 +526,7 @@ async def cold_start_adapt( user_id: int, limit: int = 15, exclude_meme_ids: list[int] = [], + candidate_guardrails_enabled: bool = False, ): """Phase 2 cold start: adapt to user's reactions in real-time. @@ -506,12 +557,14 @@ async def cold_start_adapt( SELECT M.id , M.type, M.telegram_file_id, M.caption - , 'cold_start_adapt' AS recommended_by + , :recommended_by AS recommended_by , COALESCE(MS.nlikes, 0) AS nlikes FROM meme M INNER JOIN meme_stats MS ON MS.meme_id = M.id + INNER JOIN meme_source S + ON S.id = M.meme_source_id INNER JOIN user_language L ON L.language_code = M.language_code @@ -530,6 +583,7 @@ async def cold_start_adapt( AND MS.nlikes > 1 AND MS.nmemes_sent >= 10 {TEXT_LIGHT_OCR_FILTER_SQL} + {_cold_start_guardrail_source_filter(candidate_guardrails_enabled)} {exclude_meme_ids_sql_filter(exclude_meme_ids)} ORDER BY -1 @@ -539,11 +593,12 @@ async def cold_start_adapt( """ return await fetch_all( text(query), - _build_params( + _cold_start_params( user_id, limit, exclude_meme_ids, - text_light_max_ocr_words=TEXT_LIGHT_MAX_OCR_WORDS, + engine=COLD_START_ADAPT_RECOMMENDED_BY, + candidate_guardrails_enabled=candidate_guardrails_enabled, ), ) diff --git a/src/recommendations/meme_queue.py b/src/recommendations/meme_queue.py index 9dcea924..213835c5 100644 --- a/src/recommendations/meme_queue.py +++ b/src/recommendations/meme_queue.py @@ -25,7 +25,20 @@ from src.tgbot.constants import UserType from src.tgbot.user_info import get_user_info -COLD_START_RECOMMENDED_BY = frozenset({"cold_start_explore", "cold_start_adapt"}) +COLD_START_RECOMMENDED_BY = frozenset( + { + "cold_start_explore", + "cold_start_adapt", + "cold_start_explore_guarded", + "cold_start_adapt_guarded", + } +) +COLD_START_GUARDED_RECOMMENDED_BY = frozenset( + { + "cold_start_explore_guarded", + "cold_start_adapt_guarded", + } +) async def get_next_meme_for_user(user_id: int) -> MemeData | None: @@ -70,6 +83,17 @@ async def _queued_meme_is_sendable( meme_id: int, recommended_by: str | None = None, ) -> bool: + if ( + _is_cold_start_guarded_engine(recommended_by) + and not settings.COLD_START_CANDIDATE_GUARDRAILS_ENABLED + ): + logging.info( + "discarding guarded cold_start queued meme after rollback for user_id=%s meme_id=%s", + user_id, + meme_id, + ) + return False + row = await fetch_one( text( """ @@ -108,6 +132,10 @@ def _is_cold_start_engine(recommended_by: str | None) -> bool: return recommended_by in COLD_START_RECOMMENDED_BY +def _is_cold_start_guarded_engine(recommended_by: str | None) -> bool: + return recommended_by in COLD_START_GUARDED_RECOMMENDED_BY + + def _cold_start_allowed_by_realtime_state(state: dict[str, Any] | None) -> bool: if state is None: return False @@ -288,6 +316,9 @@ async def generate_recommendations( meme_ids_in_queue=meme_ids_in_queue, random_seed=random_seed, cold_start_nsessions_gate_enabled=settings.COLD_START_NSESSIONS_GATE_ENABLED, + cold_start_candidate_guardrails_enabled=( + settings.COLD_START_CANDIDATE_GUARDRAILS_ENABLED + ), # FFM-1357: stop new exposure until this overlay has a CEO-owned # active experiment record. Existing assignment rows remain readable. text_light_blender_v1_enabled=False, diff --git a/src/recommendations/pipeline.py b/src/recommendations/pipeline.py index 28c5888b..6ecd71e5 100644 --- a/src/recommendations/pipeline.py +++ b/src/recommendations/pipeline.py @@ -14,6 +14,8 @@ from src.config import settings from src.feed_turn.planner import ( + COLD_START_1, + COLD_START_2, MATURE, CandidateSelectionPlan, low_sent_quota, @@ -76,6 +78,7 @@ class RecommendationBatchRequest: meme_ids_in_queue: Sequence[int] = field(default_factory=tuple) random_seed: int | None = None cold_start_nsessions_gate_enabled: bool = False + cold_start_candidate_guardrails_enabled: bool = False text_light_blender_v1_enabled: bool = False source_diversity_enabled: bool = False shadow_scoring_enabled: bool = True @@ -139,6 +142,8 @@ class RecommendationBatchDiagnostics: queue_len_before: int exclude_count: int cold_start_nsessions_gate_enabled: bool = False + cold_start_candidate_guardrails_enabled: bool = False + cold_start_candidate_guardrails_applied: bool = False diagnostics_sample_rate: float = 0.01 maturity_stage: str | None = None duration_ms: int = 0 @@ -183,6 +188,12 @@ def compact(self) -> dict[str, Any]: "account_age_days": self.account_age_days, "cold_start_account_too_old": self.cold_start_account_too_old, "cold_start_nsessions_gate_enabled": self.cold_start_nsessions_gate_enabled, + "cold_start_candidate_guardrails_enabled": ( + self.cold_start_candidate_guardrails_enabled + ), + "cold_start_candidate_guardrails_applied": ( + self.cold_start_candidate_guardrails_applied + ), "queue_len_before": self.queue_len_before, "exclude_count": self.exclude_count, "selected_count": self.selected_count, @@ -295,6 +306,9 @@ async def run(self, request: RecommendationBatchRequest) -> RecommendationBatchR queue_len_before=len(request.meme_ids_in_queue), exclude_count=len(request.meme_ids_in_queue), cold_start_nsessions_gate_enabled=request.cold_start_nsessions_gate_enabled, + cold_start_candidate_guardrails_enabled=( + request.cold_start_candidate_guardrails_enabled + ), diagnostics_sample_rate=request.diagnostics_sample_rate, source_diversity_enabled=request.source_diversity_enabled, shadow_scoring_enabled=request.shadow_scoring_enabled, @@ -405,12 +419,13 @@ async def _select_by_plan( diagnostics.maturity_stage = plan.maturity_stage if plan.primary_engine: - candidates = await self._fetch_engine( + candidates = await self._fetch_primary_engine( plan.primary_engine, - request.user_id, + request, + plan, + diagnostics, limit, exclude_ids, - diagnostics, ) if candidates: return candidates @@ -441,6 +456,103 @@ async def _select_by_plan( return await self._run_fallbacks(plan, request, limit, exclude_ids, diagnostics) + async def _fetch_primary_engine( + self, + engine: str, + request: RecommendationBatchRequest, + plan: CandidateSelectionPlan, + diagnostics: RecommendationBatchDiagnostics, + limit: int, + exclude_ids: list[int], + ) -> list[Candidate]: + selected: list[Candidate] = [] + segment_exclude_ids = list(exclude_ids) + segments = self._primary_engine_fetch_segments( + engine, + request, + plan, + limit, + diagnostics, + queued_or_selected_ahead=len(exclude_ids), + ) + + for segment_limit, kwargs in segments: + if segment_limit <= 0: + continue + + candidates = await self._fetch_engine( + engine, + request.user_id, + segment_limit, + segment_exclude_ids, + diagnostics, + **kwargs, + ) + excluded = set(segment_exclude_ids) + candidates = [ + candidate for candidate in candidates if candidate.get("id") not in excluded + ] + if candidates: + selected.extend(candidates) + segment_exclude_ids.extend(_candidate_ids(candidates)) + if ( + not kwargs.get("candidate_guardrails_enabled") + or len(candidates) >= segment_limit + ): + continue + + if kwargs.get("candidate_guardrails_enabled"): + fallback_limit = max(0, limit - len(selected)) + if fallback_limit > 0: + selected.extend( + await self._run_fallbacks( + plan, + request, + fallback_limit, + segment_exclude_ids, + diagnostics, + ) + ) + return selected + + return selected + + def _primary_engine_fetch_segments( + self, + engine: str, + request: RecommendationBatchRequest, + plan: CandidateSelectionPlan, + limit: int, + diagnostics: RecommendationBatchDiagnostics, + *, + queued_or_selected_ahead: int, + ) -> list[tuple[int, dict[str, Any]]]: + default_segments = [(limit, {})] + if engine not in {"cold_start_explore", "cold_start_adapt"}: + return default_segments + if not request.cold_start_candidate_guardrails_enabled: + return default_segments + if plan.maturity_stage not in {COLD_START_1, COLD_START_2}: + return default_segments + if (request.nsessions or 0) > 1 or request.cold_start_account_too_old: + return default_segments + + first_position = request.nmemes_sent + queued_or_selected_ahead + 1 + last_position = request.nmemes_sent + queued_or_selected_ahead + limit + guarded_start = max(first_position, 2) + guarded_end = min(last_position, 10) + if guarded_start > guarded_end: + return default_segments + + diagnostics.cold_start_candidate_guardrails_applied = True + segments: list[tuple[int, dict[str, Any]]] = [] + if guarded_start > first_position: + segments.append((guarded_start - first_position, {})) + segments.append((guarded_end - guarded_start + 1, {"candidate_guardrails_enabled": True})) + if last_position > guarded_end: + segments.append((last_position - guarded_end, {})) + return segments + async def _blend_weights( self, plan: CandidateSelectionPlan, @@ -469,6 +581,8 @@ async def _run_fallbacks( diagnostics: RecommendationBatchDiagnostics, ) -> list[Candidate]: for fallback in plan.fallback_engines: + if not self._retriever_supports_engine(fallback.engine): + continue candidates = await self._fetch_engine( fallback.engine, request.user_id, @@ -482,6 +596,14 @@ async def _run_fallbacks( return candidates return [] + def _retriever_supports_engine(self, engine: str) -> bool: + engine_map = getattr(self.retriever, "engine_map", None) + if engine_map is None: + return True + if type(self.retriever).get_candidates is not CandidatesRetriever.get_candidates: + return True + return engine in engine_map + async def _fetch_candidates_dict( self, engines: list[str], diff --git a/src/stats/cold_start_quality.py b/src/stats/cold_start_quality.py index 2c38114e..ad5334d4 100644 --- a/src/stats/cold_start_quality.py +++ b/src/stats/cold_start_quality.py @@ -40,7 +40,12 @@ sent_at AS first_send_at FROM first_sends WHERE sent_at >= NOW() - (:lookback_days * INTERVAL '1 day') - AND recommended_by IN ('cold_start_explore', 'cold_start_adapt') + AND recommended_by IN ( + 'cold_start_explore', + 'cold_start_adapt', + 'cold_start_explore_guarded', + 'cold_start_adapt_guarded' + ) ), candidate_sends AS ( diff --git a/tests/recommendations/test_cold_start_candidate_guardrails.py b/tests/recommendations/test_cold_start_candidate_guardrails.py new file mode 100644 index 00000000..d8a876af --- /dev/null +++ b/tests/recommendations/test_cold_start_candidate_guardrails.py @@ -0,0 +1,64 @@ +from unittest.mock import AsyncMock, patch + +import pytest + +from src.recommendations.candidates import ( + COLD_START_ADAPT_GUARDED_RECOMMENDED_BY, + COLD_START_ADAPT_RECOMMENDED_BY, + COLD_START_EXPLORE_GUARDED_RECOMMENDED_BY, + COLD_START_GUARDRAIL_SOURCE_URLS, + cold_start_adapt, + cold_start_explore, +) + + +def _fetch_call_sql_and_params(fetch_all: AsyncMock) -> tuple[str, dict]: + query, params = fetch_all.call_args.args + return str(query), params + + +@pytest.mark.asyncio +async def test_cold_start_explore_guardrails_exclude_weak_sources_and_mark_treatment(): + with patch( + "src.recommendations.candidates.fetch_all", + new_callable=AsyncMock, + return_value=[], + ) as fetch_all: + await cold_start_explore(123, limit=5, candidate_guardrails_enabled=True) + + query_sql, params = _fetch_call_sql_and_params(fetch_all) + assert "INNER JOIN meme_source S" in query_sql + assert "S.url = ANY(:cold_start_guardrail_source_urls)" in query_sql + assert params["recommended_by"] == COLD_START_EXPLORE_GUARDED_RECOMMENDED_BY + assert params["cold_start_guardrail_source_urls"] == list(COLD_START_GUARDRAIL_SOURCE_URLS) + + +@pytest.mark.asyncio +async def test_cold_start_adapt_guardrails_exclude_weak_sources_and_mark_treatment(): + with patch( + "src.recommendations.candidates.fetch_all", + new_callable=AsyncMock, + return_value=[], + ) as fetch_all: + await cold_start_adapt(123, limit=5, candidate_guardrails_enabled=True) + + query_sql, params = _fetch_call_sql_and_params(fetch_all) + assert "INNER JOIN meme_source S" in query_sql + assert "S.url = ANY(:cold_start_guardrail_source_urls)" in query_sql + assert params["recommended_by"] == COLD_START_ADAPT_GUARDED_RECOMMENDED_BY + assert params["cold_start_guardrail_source_urls"] == list(COLD_START_GUARDRAIL_SOURCE_URLS) + + +@pytest.mark.asyncio +async def test_cold_start_guardrails_default_off_preserves_control_query(): + with patch( + "src.recommendations.candidates.fetch_all", + new_callable=AsyncMock, + return_value=[], + ) as fetch_all: + await cold_start_adapt(123, limit=5) + + query_sql, params = _fetch_call_sql_and_params(fetch_all) + assert "cold_start_guardrail_source_urls" not in query_sql + assert "cold_start_guardrail_source_urls" not in params + assert params["recommended_by"] == COLD_START_ADAPT_RECOMMENDED_BY diff --git a/tests/recommendations/test_meme_queue.py b/tests/recommendations/test_meme_queue.py index 59549efb..f6cd510b 100644 --- a/tests/recommendations/test_meme_queue.py +++ b/tests/recommendations/test_meme_queue.py @@ -248,6 +248,54 @@ async def pop_queue(_queue_key): realtime_state.assert_awaited_once_with(TEST_USER_ID) +@pytest.mark.asyncio +async def test_get_next_meme_for_user_discards_guarded_payload_after_flag_rollback(): + queued_payloads = [ + { + "id": 101, + "type": "image", + "telegram_file_id": "guarded-cold-start-file-id", + "caption": None, + "recommended_by": "cold_start_explore_guarded", + }, + { + "id": 102, + "type": "image", + "telegram_file_id": "control-file-id", + "caption": None, + "recommended_by": "lr_smoothed", + }, + ] + + async def pop_queue(_queue_key): + return queued_payloads.pop(0) if queued_payloads else None + + with ( + patch("src.recommendations.meme_queue.settings") as settings, + patch( + "src.recommendations.meme_queue.redis.pop_meme_from_queue_by_key", + new_callable=AsyncMock, + side_effect=pop_queue, + ), + patch( + "src.recommendations.meme_queue.fetch_one", + new_callable=AsyncMock, + return_value={"id": 102}, + ) as fetch_one, + patch( + "src.recommendations.meme_queue._get_realtime_cold_start_routing_state", + new_callable=AsyncMock, + ) as realtime_state, + ): + settings.COLD_START_CANDIDATE_GUARDRAILS_ENABLED = False + meme = await get_next_meme_for_user(TEST_USER_ID) + + assert meme is not None + assert meme.id == 102 + fetch_one.assert_awaited_once() + realtime_state.assert_not_awaited() + + @pytest.mark.asyncio async def test_cold_start_phase1_uses_explore(): """Phase 1 (<6 memes): uses cold_start_explore engine""" @@ -272,9 +320,16 @@ class TestRetriever(CandidatesRetriever): "best_uploaded_memes": best_uploaded, } - candidates = await generate_recommendations( - TEST_USER_ID, 10, nmemes_sent=3, retriever=TestRetriever() - ) + with patch("src.recommendations.meme_queue.settings") as settings: + settings.COLD_START_NSESSIONS_GATE_ENABLED = False + settings.COLD_START_CANDIDATE_GUARDRAILS_ENABLED = False + settings.RECOMMENDATION_SOURCE_DIVERSITY_ENABLED = False + settings.RECOMMENDATION_SHADOW_SCORING_ENABLED = False + settings.RECOMMENDATION_DIAGNOSTICS_SAMPLE_RATE = 0.0 + candidates = await generate_recommendations( + TEST_USER_ID, 10, nmemes_sent=3, retriever=TestRetriever() + ) + assert len(candidates) == 3 assert candidates[0]["id"] == 101 @@ -908,7 +963,6 @@ async def test_gate_on_first_session_routes_to_cold_start_explore(): candidates = await generate_recommendations( TEST_USER_ID, 10, nmemes_sent=0, retriever=retriever ) - assert len(candidates) == 1 assert candidates[0]["recommended_by"] == "cold_start_explore" diff --git a/tests/recommendations/test_pipeline.py b/tests/recommendations/test_pipeline.py index ef43679a..ba1a2473 100644 --- a/tests/recommendations/test_pipeline.py +++ b/tests/recommendations/test_pipeline.py @@ -3,6 +3,7 @@ import pytest +from src.recommendations.candidates import CandidatesRetriever from src.recommendations.pipeline import ( RecommendationBatchPipeline, RecommendationBatchRequest, @@ -197,6 +198,235 @@ async def test_fallback_diagnostics_record_engine_that_supplied_candidates(): assert result.diagnostics.fallback_used == "best_uploaded_memes" +@pytest.mark.asyncio +async def test_cold_start_guardrails_apply_to_true_new_positions_2_to_10(): + retriever = FakeRetriever( + { + "cold_start_explore": [_meme(101, "cold_start_explore_guarded")], + "text_light_lr_smoothed": [_meme(301, "text_light_lr_smoothed")], + } + ) + + result = await _pipeline(retriever).run( + _request( + limit=1, + nmemes_sent=1, + nsessions=1, + cold_start_candidate_guardrails_enabled=True, + ) + ) + + assert _ids(result.selected) == [101] + assert retriever.calls[0]["engine"] == "cold_start_explore" + assert retriever.calls[0]["kwargs"] == {"candidate_guardrails_enabled": True} + assert result.diagnostics.cold_start_candidate_guardrails_enabled is True + assert result.diagnostics.cold_start_candidate_guardrails_applied is True + + +@pytest.mark.asyncio +async def test_cold_start_guardrails_keep_first_position_control_in_mixed_batch(): + retriever = FakeRetriever( + { + "cold_start_explore": [ + _meme(101, "cold_start_explore"), + _meme(102, "cold_start_explore_guarded"), + ] + } + ) + + result = await _pipeline(retriever).run( + _request( + nmemes_sent=0, + nsessions=1, + cold_start_candidate_guardrails_enabled=True, + ) + ) + + assert _ids(result.selected) == [101, 102] + assert retriever.calls[0]["kwargs"] == {} + assert retriever.calls[1]["kwargs"] == {"candidate_guardrails_enabled": True} + assert result.diagnostics.cold_start_candidate_guardrails_applied is True + + +@pytest.mark.asyncio +async def test_cold_start_guardrails_preserve_existing_fallbacks(): + retriever = FakeRetriever( + { + "cold_start_adapt": [], + "text_light_lr_smoothed": [_meme(301, "text_light_lr_smoothed")], + "best_uploaded_memes": [_meme(401, "best_uploaded_memes")], + } + ) + + result = await _pipeline(retriever).run( + _request( + nmemes_sent=8, + nsessions=1, + cold_start_candidate_guardrails_enabled=True, + ) + ) + + assert _ids(result.selected) == [301] + assert [call["engine"] for call in retriever.calls] == [ + "cold_start_adapt", + "text_light_lr_smoothed", + ] + assert retriever.calls[0]["kwargs"] == {"candidate_guardrails_enabled": True} + assert retriever.calls[1]["kwargs"] == {"min_sends": 10} + assert result.diagnostics.fallback_used == "text_light_lr_smoothed" + + +@pytest.mark.asyncio +async def test_cold_start_guardrails_skip_unsupported_injected_fallback_engines(): + calls = [] + + async def cold_start_adapt( + user_id: int, + limit: int = 10, + exclude_meme_ids: list[int] | None = None, + **kwargs: Any, + ) -> list[dict[str, Any]]: + calls.append({"engine": "cold_start_adapt", "kwargs": kwargs}) + return [] + + async def best_uploaded_memes( + user_id: int, + limit: int = 10, + exclude_meme_ids: list[int] | None = None, + **kwargs: Any, + ) -> list[dict[str, Any]]: + calls.append({"engine": "best_uploaded_memes", "kwargs": kwargs}) + return [_meme(401, "best_uploaded_memes")] + + class LimitedFallbackRetriever(CandidatesRetriever): + engine_map = { + "cold_start_adapt": cold_start_adapt, + "best_uploaded_memes": best_uploaded_memes, + } + + result = await _pipeline(LimitedFallbackRetriever()).run( + _request( + nmemes_sent=8, + nsessions=1, + cold_start_candidate_guardrails_enabled=True, + ) + ) + + assert _ids(result.selected) == [401] + assert [call["engine"] for call in calls] == [ + "cold_start_adapt", + "best_uploaded_memes", + ] + assert result.diagnostics.fallback_used == "best_uploaded_memes" + + +@pytest.mark.asyncio +async def test_cold_start_guardrails_use_queued_items_for_true_new_positions(): + retriever = FakeRetriever( + { + "cold_start_adapt": [ + _meme(101, "cold_start_adapt"), + _meme(102, "cold_start_adapt"), + ] + } + ) + + result = await _pipeline(retriever).run( + _request( + nmemes_sent=7, + nsessions=1, + meme_ids_in_queue=list(range(901, 909)), + cold_start_candidate_guardrails_enabled=True, + ) + ) + + assert _ids(result.selected) == [101, 102] + assert [call["engine"] for call in retriever.calls] == ["cold_start_adapt"] + assert retriever.calls[0]["kwargs"] == {} + assert result.diagnostics.cold_start_candidate_guardrails_applied is False + + +@pytest.mark.asyncio +async def test_partial_cold_start_guarded_segment_falls_back_before_unguarded_tail(): + retriever = FakeRetriever( + { + "cold_start_explore": [ + _meme(101, "cold_start_explore"), + _meme(102, "cold_start_explore_guarded"), + _meme(103, "cold_start_explore_guarded"), + _meme(104, "cold_start_explore_guarded"), + _meme(105, "cold_start_explore_guarded"), + ], + "text_light_lr_smoothed": [ + _meme(301, "text_light_lr_smoothed"), + _meme(302, "text_light_lr_smoothed"), + ], + } + ) + + result = await _pipeline(retriever).run( + _request( + limit=12, + nmemes_sent=0, + nsessions=1, + cold_start_candidate_guardrails_enabled=True, + ) + ) + + assert _ids(result.selected) == [101, 102, 103, 104, 105, 301, 302] + assert [call["engine"] for call in retriever.calls] == [ + "cold_start_explore", + "cold_start_explore", + "text_light_lr_smoothed", + ] + assert retriever.calls[0]["kwargs"] == {} + assert retriever.calls[1]["kwargs"] == {"candidate_guardrails_enabled": True} + assert retriever.calls[2]["kwargs"] == {"min_sends": 10} + assert result.diagnostics.fallback_used == "text_light_lr_smoothed" + + +@pytest.mark.asyncio +async def test_cold_start_guardrails_do_not_apply_to_dormant_or_mature_users(): + dormant_retriever = FakeRetriever( + { + "lr_smoothed": [_meme(101, "lr_smoothed")], + "recently_liked": [_meme(102, "recently_liked")], + } + ) + + dormant_result = await _pipeline(dormant_retriever).run( + _request( + nmemes_sent=8, + nsessions=3, + cold_start_nsessions_gate_enabled=True, + cold_start_candidate_guardrails_enabled=True, + ) + ) + + assert _ids(dormant_result.selected) == [101, 102] + assert all(call["kwargs"] == {} for call in dormant_retriever.calls) + assert dormant_result.diagnostics.cold_start_candidate_guardrails_applied is False + + mature_retriever = FakeRetriever( + { + "lr_smoothed": [_meme(201, "lr_smoothed")], + "recently_liked": [_meme(202, "recently_liked")], + } + ) + + mature_result = await _pipeline(mature_retriever).run( + _request( + nmemes_sent=50, + nsessions=5, + cold_start_candidate_guardrails_enabled=True, + ) + ) + + assert _ids(mature_result.selected) == [201, 202] + assert all(call["kwargs"] == {} for call in mature_retriever.calls) + assert mature_result.diagnostics.cold_start_candidate_guardrails_applied is False + + @pytest.mark.asyncio async def test_blend_engine_failure_is_recorded_but_other_engines_continue(): retriever = FakeRetriever( diff --git a/tests/stats/test_cold_start_quality.py b/tests/stats/test_cold_start_quality.py index c3a0cc9f..79fc650e 100644 --- a/tests/stats/test_cold_start_quality.py +++ b/tests/stats/test_cold_start_quality.py @@ -150,6 +150,13 @@ def test_readout_script_prefers_analyst_database_url(monkeypatch): assert os.environ["DATABASE_URL"] == expected_url +def test_readout_includes_guarded_cold_start_treatment_engines(): + query_sql = str(build_cold_start_first10_quality_query("summary")) + + assert "cold_start_explore_guarded" in query_sql + assert "cold_start_adapt_guarded" in query_sql + + @pytest.mark.asyncio async def test_summary_uses_true_new_cold_start_cohort(cold_start_readout_data): rows = await _fetch_section("summary")