fix: classify cache hits by current CV, not the legacy assessment column

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.

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,

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

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