From 84cb2673c4bf465cf33043861636d1fcaa63f84e Mon Sep 17 00:00:00 2001 From: sangowu <34704289+sangowu@users.noreply.github.com> Date: Fri, 7 Aug 2026 22:12:19 +0100 Subject: [PATCH] fix: classify cache hits by current CV, not the legacy assessment column prefilter_jobs decided a cached job's fate from job_cache.assessment, a column that records no cv_hash. A job rejected under an earlier CV was dropped by `continue` before it could reach the pipeline, so no later CV could ever reconsider it -- 95 of 293 jobs in the local cache were blocked this way. Cache hits now go through classify_cache_hit, which looks up job_matches for the current cv_hash and match prompt version: reuse - match exists, recommendation != skip skip - match exists, recommendation == skip reassess - no match for this CV (CV changed, prompt version bumped, or never scored); re-enters assessment on cached JD content, no re-fetch Without a cv_hash the previous legacy-column behavior is kept, so CV-less runs keep working and the no-CV path does not start dropping results. Verified read-only against the real cache on jobs a previous CV had rejected: passing the current cv_hash yields skipped=0 / re-assess=5, while the legacy path on the same rows yields skipped=5 / re-assess=0. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_012DbPx4FbswofbGn41anE1q --- CHANGELOG.md | 4 + jobradar/agent.py | 3 +- jobradar/search_prefilter.py | 44 +++++-- tests/test_search_prefilter.py | 227 +++++++++++++++++++++++++++++++++ 4 files changed, 268 insertions(+), 10 deletions(-) create mode 100644 tests/test_search_prefilter.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 1a2112f..2229480 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,10 @@ ### Bug Fixes +- **Cache hits ignored which CV produced the score** (`search_prefilter.py` / `agent.py`) + A cached job was reused or discarded based on the legacy `assessment` column, which records no `cv_hash`. A job rejected under an earlier CV was dropped by `continue` before reaching the pipeline, so no later CV could ever reconsider it — 95 jobs in a local 293-job cache were blocked this way. Cache hits are now classified by `classify_cache_hit` against the current `cv_hash` and match prompt version: a match under the current CV is reused (or skipped when its recommendation is `skip`), and anything else re-enters assessment using the cached JD content, without re-fetching the page. With no `cv_hash` available the previous legacy-column behavior is kept, so CV-less runs are unaffected. + Expect higher assessment volume on the first search after changing CV: cached jobs that only carry another CV's score are now re-evaluated instead of silently reused. + - **`assess` skipped every job that carried a legacy assessment** (`cache.py` / `cli.py` / `search_assessment_stage.py`) `get_unassessed_jobs` selected on `assessment IS NULL` before consulting `job_matches`. Since the legacy `assessment` column records no `cv_hash`, any job scored under an earlier CV was treated as done forever — on a local cache of 293 jobs the command reported "all assessed" while 205 had no match under the current CV, and 95 rejected by a previous CV could never be reconsidered. Selection is now driven solely by the presence of a match for the current `cv_hash`. Its filter predicate also read `job.match_score` — a leftover loop variable — instead of `j.match_score`, so the whole batch was kept or dropped according to the last row alone. diff --git a/jobradar/agent.py b/jobradar/agent.py index 097a6db..81f533d 100644 --- a/jobradar/agent.py +++ b/jobradar/agent.py @@ -239,6 +239,7 @@ def _assessment_tasks() -> Iterator[ScheduledBatch[tuple[PrefilterResult, list[s language=language, run_id=run_id, seen_dedup_keys=seen_dedup_keys, + cv_hash=cv_hash, ) candidate_jobs = [job for job, _, _ in pf.pending] persisted_at = time.monotonic() @@ -506,7 +507,7 @@ def _write_scraped( job_all_sources = collect_all_sources(jobs) - pf = prefilter_jobs(jobs, seen_urls, cb, profile, language=language, run_id=run_id) + pf = prefilter_jobs(jobs, seen_urls, cb, profile, language=language, run_id=run_id, cv_hash=cv_hash) _profile = profile or CVProfile( summary=_cv_summary, skills=_cv_skills, diff --git a/jobradar/search_prefilter.py b/jobradar/search_prefilter.py index 88c4bfd..65cdb39 100644 --- a/jobradar/search_prefilter.py +++ b/jobradar/search_prefilter.py @@ -7,7 +7,8 @@ from jobradar import cache from jobradar.filters import infer_title_seniority, is_title_seniority_ok from jobradar.logger import get_logger -from jobradar.schemas import CVProfile, is_closed_posting, make_dedup_key +from jobradar.matching import match_prompt_version +from jobradar.schemas import CVProfile, JobResult, is_closed_posting, make_dedup_key from jobradar.tools import record_failed_url logger = get_logger(__name__) @@ -75,6 +76,30 @@ def collect_all_sources(jobs: list[dict]) -> dict[str, list[dict]]: return result +def classify_cache_hit(cached_job: JobResult, cv_hash: str, language: str = "zh") -> str: + """判断缓存命中的职位在当前 CV 下应 reuse / skip / reassess。 + + 评分是 JD × CV 的函数，因此命中与否要看当前 ``cv_hash`` 下有无匹配结果， + 而不是看该职位评过没有。legacy ``assessment`` 列不记录 cv_hash， + 用它判断会把换 CV 后本该重评的职位永久挡在管道之外。 + """ + if cv_hash: + match = cache.get_job_match( + cached_job.dedup_key, + cv_hash, + cached_job.description_snippet, + prompt_version=match_prompt_version(language), + ) + if match is None: + return "reassess" + return "skip" if match.recommendation == "skip" else "reuse" + + # 无从判断 CV 版本时退回 legacy 列，保持既有行为。 + if cached_job.assessment is not None: + return "reuse" if cached_job.assessment.is_relevant else "skip" + return "reassess" + + @dataclass class PrefilterResult: immediate_keys: list[str] = field(default_factory=list) @@ -103,8 +128,8 @@ def prefilter_jobs( language: str = "zh", run_id: str = "", seen_dedup_keys: set[str] | None = None, + cv_hash: str = "", ) -> PrefilterResult: - del language result = PrefilterResult() run_seen_dedup_keys = seen_dedup_keys if seen_dedup_keys is not None else set() title_candidates: list[tuple[dict, str, str, str, str, dict[str, int]]] = [] @@ -158,18 +183,19 @@ def prefilter_jobs( cached_job = cache.get_job_by_url(url) if cached_job is not None and not cached_job.is_expired: - if cached_job.assessment is not None: - if not cached_job.assessment.is_relevant: - logger.debug("URL cache hit (rejected), skip: %s", title) - source_stats["cache_hit"] += 1 - result.cache_hit += 1 - continue + disposition = classify_cache_hit(cached_job, cv_hash, language) + if disposition == "skip": + logger.debug("URL cache hit (rejected for current CV), skip: %s", title) + source_stats["cache_hit"] += 1 + result.cache_hit += 1 + continue + if disposition == "reuse": logger.debug("URL cache hit, skip fetch+LLM: %s", title) result.immediate_keys.append(cached_job.dedup_key) source_stats["cache_hit"] += 1 result.cache_hit += 1 continue - logger.debug("URL cache hit, pending LLM re-assess: %s", title) + logger.debug("URL cache hit without a match for current CV, re-assess: %s", title) result.patch_pending.append((cached_job, cached_job.description_snippet)) result.cache_patch += 1 continue diff --git a/tests/test_search_prefilter.py b/tests/test_search_prefilter.py new file mode 100644 index 0000000..724fcee --- /dev/null +++ b/tests/test_search_prefilter.py @@ -0,0 +1,227 @@ +"""测试 prefilter 的缓存命中判定：评分按 cv_hash 归属，换 CV 后必须重进管道。""" +from __future__ import annotations + +import importlib +import os +import tempfile + +import pytest + +from jobradar.matching import match_prompt_version +from jobradar.schemas import CVProfile, JobAssessment, JobResult, MatchScore + +CURRENT_CV = "cv-current" +OUTDATED_CV = "cv-outdated" + + +@pytest.fixture(autouse=True) +def temp_db(monkeypatch): + import jobradar.cache as cache_mod + + with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f: + db_path = f.name + + monkeypatch.setenv("CACHE_DB_PATH", db_path) + importlib.reload(cache_mod) + yield cache_mod + + try: + os.unlink(db_path) + except OSError: + pass + + +def make_job(**kwargs) -> JobResult: + defaults = dict( + title="Backend Engineer", + company="Example", + url="https://example.com/jobs/1", + description_snippet="Build Python services.", + ) + defaults.update(kwargs) + return JobResult(**defaults) + + +def make_match(job: JobResult, cv_hash: str, recommendation: str = "apply") -> MatchScore: + return MatchScore( + job_id=job.dedup_key, + cv_hash=cv_hash, + overall_score=80, + title_score=80, + seniority_score=80, + must_have_score=80, + nice_to_have_score=80, + domain_score=80, + location_score=100, + language_score=100, + risk_penalty=0, + recommendation=recommendation, + ) + + +def profile() -> CVProfile: + return CVProfile( + summary="Python backend engineer", + skills=["Python", "SQL"], + years_of_experience=2, + seniority="junior", + preferred_roles=["Backend Engineer"], + preferred_locations=["Dublin"], + ) + + +class TestClassifyCacheHit: + def test_match_under_current_cv_is_reused(self, temp_db): + from jobradar.search_prefilter import classify_cache_hit + + job = make_job() + temp_db.save_job(job) + temp_db.save_job_match( + make_match(job, CURRENT_CV), + job.description_snippet, + prompt_version=match_prompt_version("zh"), + ) + + assert classify_cache_hit(job, CURRENT_CV) == "reuse" + + def test_skip_recommendation_under_current_cv_is_skipped(self, temp_db): + from jobradar.search_prefilter import classify_cache_hit + + job = make_job() + temp_db.save_job(job) + temp_db.save_job_match( + make_match(job, CURRENT_CV, recommendation="skip"), + job.description_snippet, + prompt_version=match_prompt_version("zh"), + ) + + assert classify_cache_hit(job, CURRENT_CV) == "skip" + + def test_match_from_another_cv_triggers_reassessment(self, temp_db): + """核心：换 CV 后旧匹配结果不算数，该职位必须重进管道。""" + from jobradar.search_prefilter import classify_cache_hit + + job = make_job() + temp_db.save_job(job) + temp_db.save_job_match( + make_match(job, OUTDATED_CV), + job.description_snippet, + prompt_version=match_prompt_version("zh"), + ) + + assert classify_cache_hit(job, CURRENT_CV) == "reassess" + + def test_legacy_rejection_no_longer_blocks_forever(self, temp_db): + """核心：旧 CV 判定的 is_relevant=False 不再是永久结论。""" + from jobradar.search_prefilter import classify_cache_hit + + job = make_job(assessment=JobAssessment(score=2, is_relevant=False)) + temp_db.save_job(job) + + assert classify_cache_hit(job, CURRENT_CV) == "reassess" + + def test_stale_prompt_version_triggers_reassessment(self, temp_db): + from jobradar.search_prefilter import classify_cache_hit + + job = make_job() + temp_db.save_job(job) + temp_db.save_job_match( + make_match(job, CURRENT_CV), + job.description_snippet, + prompt_version="match_v1:zh", + ) + + assert classify_cache_hit(job, CURRENT_CV) == "reassess" + + def test_without_cv_hash_falls_back_to_legacy_column(self, temp_db): + """无从判断 CV 版本时保持既有行为，避免无 CV 场景下丢结果。""" + from jobradar.search_prefilter import classify_cache_hit + + relevant = make_job(assessment=JobAssessment(score=8, is_relevant=True)) + rejected = make_job( + company="Other", + url="https://example.com/jobs/2", + assessment=JobAssessment(score=1, is_relevant=False), + ) + unassessed = make_job(company="Third", url="https://example.com/jobs/3") + + assert classify_cache_hit(relevant, "") == "reuse" + assert classify_cache_hit(rejected, "") == "skip" + assert classify_cache_hit(unassessed, "") == "reassess" + + +class TestPrefilterCacheHit: + def _scraped(self, job: JobResult) -> dict: + return { + "title": job.title, + "company": job.company, + "location": "Dublin", + "url": job.url, + "source": "indeed.ie", + "description_snippet": job.description_snippet, + "is_complete": True, + } + + def test_cached_job_reassessed_when_cv_changed(self, temp_db): + from jobradar.search_prefilter import prefilter_jobs + + job = make_job() + temp_db.save_job(job) + temp_db.save_job_match( + make_match(job, OUTDATED_CV), + job.description_snippet, + prompt_version=match_prompt_version("zh"), + ) + + result = prefilter_jobs( + [self._scraped(job)], + set(), + lambda message: None, + profile(), + cv_hash=CURRENT_CV, + ) + + assert result.cache_patch == 1 + assert result.cache_hit == 0 + assert [cached.dedup_key for cached, _ in result.patch_pending] == [job.dedup_key] + + def test_cached_job_reused_when_cv_matches(self, temp_db): + from jobradar.search_prefilter import prefilter_jobs + + job = make_job() + temp_db.save_job(job) + temp_db.save_job_match( + make_match(job, CURRENT_CV), + job.description_snippet, + prompt_version=match_prompt_version("zh"), + ) + + result = prefilter_jobs( + [self._scraped(job)], + set(), + lambda message: None, + profile(), + cv_hash=CURRENT_CV, + ) + + assert result.immediate_keys == [job.dedup_key] + assert result.cache_hit == 1 + assert result.cache_patch == 0 + + def test_legacy_rejected_job_reenters_pipeline(self, temp_db): + """回归：旧 CV 拒绝过的职位曾被 continue 直接丢弃，永不重评。""" + from jobradar.search_prefilter import prefilter_jobs + + job = make_job(assessment=JobAssessment(score=2, is_relevant=False)) + temp_db.save_job(job) + + result = prefilter_jobs( + [self._scraped(job)], + set(), + lambda message: None, + profile(), + cv_hash=CURRENT_CV, + ) + + assert result.cache_patch == 1 + assert result.cache_hit == 0