From 5d17cfbf704e025758082738dce3936293a15f4c Mon Sep 17 00:00:00 2001 From: -Puter <22245429+puterhimself@users.noreply.github.com> Date: Fri, 17 Jul 2026 19:24:58 +0530 Subject: [PATCH] fix matchmaking outage and cache regressions --- app/agent/session.py | 12 ++- app/db/repo.py | 25 +++--- app/engine/search.py | 16 ++-- tests/test_pool.py | 183 +++++++++++++++++++++++++++++++++++++++++-- 4 files changed, 212 insertions(+), 24 deletions(-) diff --git a/app/agent/session.py b/app/agent/session.py index c002781..2671cb8 100644 --- a/app/agent/session.py +++ b/app/agent/session.py @@ -188,6 +188,7 @@ class MatchmakingAgentSession: await self.push( "agent_error", action="search_failed", + message="All enabled job boards are temporarily unavailable.", data={ "code": "job_boards_unavailable", "message": "All enabled job boards are temporarily unavailable.", @@ -272,10 +273,12 @@ class MatchmakingAgentSession: dedupe_key=f"matchmaking:generated:{self.user_id}:{sig}", ) - # The user already has their deck — if the pool dipped below the reorder mark, top it up OUT OF - # BAND so the next search stays free and the shelf never runs thin. (No await: fire-and-forget.) + # The persisted feed intentionally drops resume context, but a refill must use the same + # retrieval inputs as the blocking sweep whose signature owns the pool. Preserve only + # the adapter's top-skill input; never retain the rest of the resume context. if decision == "pool+refill": - asyncio.create_task(self._background_refill(clean_prefs, sig)) + refill_prefs = {**clean_prefs, "_skills": (user_context or {}).get("skills") or prefs.get("_skills") or []} + asyncio.create_task(self._background_refill(refill_prefs, sig)) async def _background_refill(self, prefs: dict, sig: str): """Proactive pool top-up: re-fetch the freshest jobs in the background + bank the new ones, so the @@ -286,6 +289,9 @@ class MatchmakingAgentSession: try: from app.config import get_settings sweep = await search.run_sweep(prefs, fresh=True) + if sweep.get("all_boards_failed"): + logger.warning("pool refill skipped: all configured job boards failed, sig=%s", sig[:8]) + return new_count = await _repo.pool_save(self.user_id, sig, sweep["opportunities"]) if new_count < get_settings().POOL_MIN_NEW_PER_REFILL: # tapped out → relax the floor await _repo.mark_pool_exhausted(self.user_id, sig) diff --git a/app/db/repo.py b/app/db/repo.py index 4be34a9..f16678c 100644 --- a/app/db/repo.py +++ b/app/db/repo.py @@ -10,6 +10,7 @@ from sqlalchemy import func, select from sqlalchemy.dialects.postgresql import insert as pg_insert from app.db.models import Base, OpportunityState, UserFeed, UserJobPool +from app.engine.board_adapters import coerce as C from app.db.session import engine, session_factory logger = logging.getLogger(__name__) @@ -32,19 +33,21 @@ def _effective_title(prefs: dict) -> str: def search_signature(prefs: dict) -> str: - """Stable short hash of the search-defining prefs (ignores transient flags / user_context).""" - # Canonicalize the fields exactly as the board adapters consume them. The - # dashboard contract is camelCase; work_mode remains a legacy fallback. - work_mode = prefs.get("workMode") if "workMode" in prefs else prefs.get("work_mode") + """Stable short hash of every preference that changes raw board retrieval.""" + # The pool stores unranked jobs. Ranking and curation rerun on every request, + # so only adapter inputs belong in this identity. Canonicalize through the + # same helpers used by the board adapters to avoid cache-only semantics. + user_context = prefs.get("user_context") or {} + skills = user_context.get("skills") or prefs.get("_skills") or [] + retrieval_skill = str(skills[0]).strip().lower() if skills else "" payload = { - "effective_title": _effective_title(prefs), - "role": prefs.get("role"), - "location": prefs.get("location"), + "title": C.title_of(prefs), + "locations": C.parsed_locations(prefs), + "workMode": prefs.get("workMode"), "industry": prefs.get("industry"), - "experience": prefs.get("experience"), - "years": prefs.get("years"), - "workMode": work_mode, - "remote": prefs.get("remote"), + "experience": C.seniority_band(prefs), + "years": C.years_of(prefs), + "retrieval_skill": retrieval_skill, } blob = json.dumps(payload, sort_keys=True, default=str) return hashlib.sha1(blob.encode()).hexdigest()[:16] diff --git a/app/engine/search.py b/app/engine/search.py index a292260..5c02fed 100644 --- a/app/engine/search.py +++ b/app/engine/search.py @@ -116,16 +116,18 @@ async def run_sweep(prefs: dict, *, fresh: bool = False) -> dict: # fresh=True forces a live fetch + cache overwrite (dev "fresh search" toggle); # otherwise honor the configured cache mode (dev replays cached results for $0). cache_mode = "refresh" if fresh else None - enabled = [b.strip() for b in get_settings().BOARDS_ENABLED.split(",") if b.strip() in BOARDS] + configured = [b.strip() for b in get_settings().BOARDS_ENABLED.split(",") if b.strip()] + enabled = [b for b in configured if b in BOARDS] cap = get_settings().BOARD_TIMEOUT_S async def _timed(k: str): # A single slow/hung board must NOT hold the whole search hostage — cap it and degrade. try: - return await asyncio.wait_for(_fetch_board(k, prefs, cache_mode), timeout=cap) + key, jobs = await asyncio.wait_for(_fetch_board(k, prefs, cache_mode), timeout=cap) + return key, jobs, False except asyncio.TimeoutError: logger.warning("board %s exceeded %ss — skipped (search stays fast)", k, cap) - return k, [] + return k, [], True results = await asyncio.gather(*(_timed(k) for k in enabled), return_exceptions=True) @@ -138,7 +140,10 @@ async def run_sweep(prefs: dict, *, fresh: bool = False) -> dict: failed_boards += 1 logger.warning("board fetch failed: %s", r) continue - key, jobs = r + key, jobs, failed = r + if failed: + failed_boards += 1 + continue sources[key] = len(jobs) for j in jobs: dk = (j.get("apply_url") or "").strip() or f"{j['organization']}|{j['title']}|{j.get('location_city')}".lower() @@ -151,9 +156,10 @@ async def run_sweep(prefs: dict, *, fresh: bool = False) -> dict: return { "opportunities": merged, "sources": sources, - "all_boards_failed": bool(enabled) and failed_boards == len(enabled), + "all_boards_failed": not enabled or failed_boards == len(enabled), "failed_boards": failed_boards, "enabled_boards": len(enabled), + "configured_boards": configured, } diff --git a/tests/test_pool.py b/tests/test_pool.py index b3b1b83..d293053 100644 --- a/tests/test_pool.py +++ b/tests/test_pool.py @@ -1,5 +1,7 @@ """Warm-pool gate — the reorder/floor decision that keeps the shelf stocked without over-fetching.""" +import asyncio from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock import pytest @@ -67,14 +69,51 @@ def test_work_mode_changes_search_signature_using_dashboard_contract(): assert search_signature(base) != search_signature(changed) -def test_legacy_and_camel_case_work_mode_have_same_signature(): +def test_legacy_work_mode_does_not_change_adapter_cache_identity(): common = {"title": "Data Analyst", "location": ["Bangalore · India"]} - assert search_signature({**common, "workMode": ["Hybrid"]}) == search_signature({**common, "work_mode": ["Hybrid"]}) + assert search_signature(common) == search_signature({**common, "work_mode": ["Hybrid"]}) -@pytest.mark.asyncio -async def test_sweep_marks_total_board_failure(monkeypatch): +def test_retrieval_inputs_change_search_signature(): + base = { + "title": "Data Analyst", + "location": ["Bangalore · India"], + "workMode": ["Hybrid"], + "industry": ["Fintech"], + "experience": ["Mid-level"], + "years": 5, + "user_context": {"skills": ["Python", "SQL"]}, + } + changes = [ + {"title": "Product Manager"}, + {"location": ["Mumbai · India"]}, + {"workMode": ["Remote"]}, + {"industry": ["Healthcare"]}, + {"experience": ["Senior"]}, + {"years": 8}, + {"user_context": {"skills": ["Rust", "SQL"]}}, + ] + + for change in changes: + assert search_signature(base) != search_signature({**base, **change}) + + +def test_rank_only_preferences_do_not_cold_start_raw_pool(): + base = {"title": "Data Analyst", "location": ["Bangalore · India"]} + changed = {**base, "targetComp": "₹22–30L", "dealBreakers": ["Travel"], "stretch": "reach"} + + assert search_signature(base) == search_signature(changed) + + +def test_search_signature_ignores_unrelated_user_context(): + base = {"title": "Data Analyst", "location": ["Bangalore · India"], "user_context": {"skills": ["Python"]}} + changed = {**base, "user_context": {"skills": ["Python"], "education": ["MBA"]}} + + assert search_signature(base) == search_signature(changed) + + +def test_sweep_marks_total_board_exception_failure(monkeypatch): async def fail_board(*_args, **_kwargs): raise RuntimeError("provider blocked") @@ -84,8 +123,142 @@ async def test_sweep_marks_total_board_failure(monkeypatch): BOARD_TIMEOUT_S=1, )) - result = await search.run_sweep({"title": "Data Analyst", "location": ["Bangalore · India"]}) + result = asyncio.run(search.run_sweep({"title": "Data Analyst", "location": ["Bangalore · India"]})) assert result["all_boards_failed"] is True assert result["failed_boards"] == 2 assert result["sources"] == {} + + +def test_sweep_marks_total_board_timeout_failure(monkeypatch): + async def timeout_board(*_args, **_kwargs): + await asyncio.sleep(0.05) + return "naukri", [] + + monkeypatch.setattr(search, "_fetch_board", timeout_board) + monkeypatch.setattr(search, "get_settings", lambda: SimpleNamespace( + BOARDS_ENABLED="naukri,foundit", + BOARD_TIMEOUT_S=0.001, + )) + result = asyncio.run(search.run_sweep({"title": "Data Analyst", "location": ["Bangalore · India"]})) + + assert result["all_boards_failed"] is True + assert result["failed_boards"] == 2 + + +def test_sweep_marks_empty_or_unknown_board_config_as_failure(monkeypatch): + monkeypatch.setattr(search, "get_settings", lambda: SimpleNamespace( + BOARDS_ENABLED="unknown-board", + BOARD_TIMEOUT_S=1, + )) + + result = asyncio.run(search.run_sweep({"title": "Data Analyst", "location": ["Bangalore · India"]})) + + assert result["all_boards_failed"] is True + assert result["enabled_boards"] == 0 + assert result["configured_boards"] == ["unknown-board"] + + +def test_total_outage_emits_frontend_readable_agent_error(monkeypatch): + from app.agent.session import MatchmakingAgentSession + + ws = MagicMock() + ws.send_json = AsyncMock() + session = MatchmakingAgentSession(ws, "user-1") + async def outage_heartbeat(awaitable, _messages): + awaitable.close() + return { + "opportunities": [], + "sources": {}, + "all_boards_failed": True, + "failed_boards": 2, + } + + session._with_heartbeat = outage_heartbeat + monkeypatch.setattr("app.agent.session.get_settings", lambda: SimpleNamespace( + POOL_ENABLED=False, + POOL_SAFETY_FLOOR=90, + POOL_REORDER_AT=100, + ), raising=False) + monkeypatch.setattr("app.config.get_settings", lambda: SimpleNamespace( + POOL_ENABLED=False, + POOL_SAFETY_FLOOR=90, + POOL_REORDER_AT=100, + )) + monkeypatch.setattr("app.db.repo.get_seen_ids", AsyncMock(return_value=set())) + + asyncio.run(session.handle_run_search({"title": "Data Analyst", "location": ["Bangalore · India"]})) + + message = ws.send_json.await_args.args[0] + assert message["type"] == "agent_error" + assert message["message"] == "All enabled job boards are temporarily unavailable." + assert message["data"]["code"] == "job_boards_unavailable" + + +def test_background_refill_does_not_mark_outage_as_exhausted(monkeypatch): + from app.agent.session import MatchmakingAgentSession + + session = MatchmakingAgentSession(MagicMock(), "user-1") + monkeypatch.setattr("app.engine.search.run_sweep", AsyncMock(return_value={ + "opportunities": [], + "sources": {}, + "all_boards_failed": True, + })) + save = AsyncMock() + exhausted = AsyncMock() + monkeypatch.setattr("app.db.repo.pool_save", save) + monkeypatch.setattr("app.db.repo.mark_pool_exhausted", exhausted) + + asyncio.run(session._background_refill({"title": "Data Analyst"}, "signature")) + + save.assert_not_awaited() + exhausted.assert_not_awaited() + + +def test_background_refill_preserves_signature_retrieval_skill(monkeypatch): + from app.agent.session import MatchmakingAgentSession + ws = MagicMock() + ws.send_json = AsyncMock() + session = MatchmakingAgentSession(ws, "user-1") + prefs = { + "title": "Data Analyst", + "location": ["Bangalore · India"], + "user_context": {"skills": ["Python", "SQL"], "education": ["MBA"]}, + } + sig = search_signature(prefs) + captured = {} + + monkeypatch.setattr("app.config.get_settings", lambda: SimpleNamespace( + POOL_ENABLED=True, + POOL_SAFETY_FLOOR=90, + POOL_REORDER_AT=100, + )) + monkeypatch.setattr("app.db.repo.get_seen_ids", AsyncMock(return_value=set())) + monkeypatch.setattr("app.db.repo.pool_load_fresh_unseen", AsyncMock(return_value=[{"id": f"job-{i}"} for i in range(95)])) + monkeypatch.setattr("app.db.repo.is_pool_exhausted", AsyncMock(return_value=False)) + monkeypatch.setattr("app.db.repo.save_feed", AsyncMock()) + monkeypatch.setattr("app.db.repo.update_salary_band", AsyncMock()) + monkeypatch.setattr("app.db.repo.mark_seen", AsyncMock()) + monkeypatch.setattr("app.db.repo.get_opportunity_states", AsyncMock(return_value={})) + monkeypatch.setattr("app.agent.session.emit_grow_event", AsyncMock()) + monkeypatch.setattr("app.engine.sift.sift", lambda *_args: ([{"id": "job-1"}], {"mode": "test", "scored": 1})) + monkeypatch.setattr("app.engine.curate.curate", lambda *_args: [{"id": "job-1"}]) + + async def capture_refill(refill_prefs, refill_sig): + captured["prefs"] = refill_prefs + captured["sig"] = refill_sig + + async def direct_heartbeat(awaitable, _messages): + return await awaitable + + session._background_refill = capture_refill + session._with_heartbeat = direct_heartbeat + session._grow_activity_fields = AsyncMock(return_value={}) + + asyncio.run(session.handle_run_search(prefs)) + asyncio.run(asyncio.sleep(0)) + + assert captured["sig"] == sig + assert captured["prefs"]["_skills"] == ["Python", "SQL"] + assert "user_context" not in captured["prefs"] + assert search_signature(captured["prefs"]) == sig