diff --git a/app/agent/session.py b/app/agent/session.py index 5df47f3..2671cb8 100644 --- a/app/agent/session.py +++ b/app/agent/session.py @@ -184,6 +184,18 @@ class MatchmakingAgentSession: search.run_sweep(prefs, fresh=fresh), ["Scanning live roles across the boards…", "Pulling the latest Naukri + LinkedIn + Foundit roles…", "Gathering fresh postings for you…"]) + if sweep.get("all_boards_failed"): + 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.", + "failed_boards": sweep.get("failed_boards", 0), + }, + ) + return new_count = await _repo.pool_save(self.user_id, sig, sweep["opportunities"]) # tapped out? a sweep that added almost nothing new → flag it so we stop block-fetching dupes. if new_count < s.POOL_MIN_NEW_PER_REFILL: @@ -261,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 @@ -275,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 d9b30d7..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__) @@ -17,14 +18,37 @@ logger = logging.getLogger(__name__) # fields a caller may set on an opportunity's state (status + the generated document ids + activity flags) _STATE_FIELDS = ("status", "tailored_resume_id", "tailored_version_id", "cover_letter_id", "seen", "viewed", "saved") -# the search-defining pref fields — same values → same signature → cursor advances (new jobs); -# change any → new signature → cursor resets to page 1. -_SIG_FIELDS = ("title", "role", "location", "industry", "experience", "years", "work_mode", "remote") +def _effective_title(prefs: dict) -> str: + """Match the board adapter's title precedence for cache identity.""" + target_titles = prefs.get("targetTitles") or [] + if target_titles and str(target_titles[0]).strip(): + return str(target_titles[0]).strip() + title = str(prefs.get("title") or "").strip() + if title: + return title + for role in prefs.get("role") or []: + if role and role != "Open to any": + return str(role) + return "" def search_signature(prefs: dict) -> str: - """Stable short hash of the search-defining prefs (ignores transient flags / user_context).""" - payload = {k: prefs.get(k) for k in _SIG_FIELDS} + """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 = { + "title": C.title_of(prefs), + "locations": C.parsed_locations(prefs), + "workMode": prefs.get("workMode"), + "industry": prefs.get("industry"), + "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 94fd675..5c02fed 100644 --- a/app/engine/search.py +++ b/app/engine/search.py @@ -116,27 +116,34 @@ 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) merged: list[dict] = [] seen: set[str] = set() sources: dict[str, int] = {} + failed_boards = 0 for r in results: if isinstance(r, Exception): + 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() @@ -146,7 +153,14 @@ async def run_sweep(prefs: dict, *, fresh: bool = False) -> dict: merged.append(j) # No ranking here — the engine (app/engine/rank.py) scores + selects after the sweep. - return {"opportunities": merged, "sources": sources} + return { + "opportunities": merged, + "sources": sources, + "all_boards_failed": not enabled or failed_boards == len(enabled), + "failed_boards": failed_boards, + "enabled_boards": len(enabled), + "configured_boards": configured, + } def has_mvq(prefs: dict) -> bool: diff --git a/docker-compose.staging.yml b/docker-compose.staging.yml index 355f8ef..3975d9e 100644 --- a/docker-compose.staging.yml +++ b/docker-compose.staging.yml @@ -35,7 +35,7 @@ services: condition: service_healthy redis: condition: service_started - command: uvicorn app.main:app --host 0.0.0.0 --port 8000 + command: sh -c "alembic upgrade head && exec uvicorn app.main:app --host 0.0.0.0 --port 8000" restart: unless-stopped postgres: diff --git a/docker-compose.yml b/docker-compose.yml index b15bf86..e809011 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -10,6 +10,7 @@ services: environment: # inside the compose network the DB host is `postgres` (overrides the host-dev .env value) DATABASE_URL: postgresql+asyncpg://matchmaking:matchmaking@postgres:5432/matchmaking + command: sh -c "alembic upgrade head && exec uvicorn app.main:app --host 0.0.0.0 --port 8000" depends_on: postgres: condition: service_healthy diff --git a/tests/test_pool.py b/tests/test_pool.py index 121607f..d293053 100644 --- a/tests/test_pool.py +++ b/tests/test_pool.py @@ -1,5 +1,13 @@ """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 + +from app.engine import search from app.engine.search import pool_decision +from app.db.repo import search_signature FLOOR, REORDER = 90, 100 # the configured thresholds (POOL_SAFETY_FLOOR / POOL_REORDER_AT) @@ -39,3 +47,218 @@ def test_floor_never_above_reorder_invariant(): from app.config import get_settings s = get_settings() assert s.POOL_SAFETY_FLOOR <= s.POOL_REORDER_AT + + +def test_target_title_changes_search_signature(): + base = { + "title": "Product Manager", + "role": ["Product"], + "targetTitles": ["Backend Engineer"], + "location": ["Bangalore · India"], + "workMode": ["Hybrid"], + } + changed = {**base, "targetTitles": ["Data Analyst"]} + + assert search_signature(base) != search_signature(changed) + + +def test_work_mode_changes_search_signature_using_dashboard_contract(): + base = {"title": "Data Analyst", "location": ["Bangalore · India"], "workMode": ["Hybrid"]} + changed = {**base, "workMode": ["Remote"]} + + assert search_signature(base) != search_signature(changed) + + +def test_legacy_work_mode_does_not_change_adapter_cache_identity(): + common = {"title": "Data Analyst", "location": ["Bangalore · India"]} + + assert search_signature(common) == search_signature({**common, "work_mode": ["Hybrid"]}) + + +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") + + monkeypatch.setattr(search, "_fetch_board", fail_board) + monkeypatch.setattr(search, "get_settings", lambda: SimpleNamespace( + BOARDS_ENABLED="naukri,foundit", + 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["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