chore: retire the legacy assessment column from the scoring path

CHANGELOG.md

@@ -2,6 +2,13 @@
 
 ## [Unreleased]
 
+### Changes
+
+- **Scores now come from `job_matches` alone** (`schemas.py` / `display.py` / `cli.py` / `search_assessment_stage.py` / `search_prefilter.py` / `cache.py`)
+  The legacy `job_cache.assessment` column is no longer written or read for scoring. It recorded no `cv_hash`, and its 0–10 scale was being mixed with `MatchScore`'s 0–100 inside the same `effective_score` property, so cached legacy scores and modern ones sorted against each other on incompatible scales. The `effective_*` properties, both `display.py` render paths, and the `assess` listing filter now depend only on `match_score`; the search pipeline stops populating the column, and `cache.update_job_assessment` is removed.
+  `classify_cache_hit` no longer consults the column when no `cv_hash` is available and returns `reuse` instead — returning `reassess` there would strand those jobs, since `flush_assessments` skips `patch_pending` when `has_cv` is false.
+  The column, the `JobAssessment` model, and `JobResult.assessment` are kept so historical rows stay readable — `model_quality_audit` exports them. Existing data is untouched, and `_merge_job` still preserves it.
+
 ### Bug Fixes
 
 - **Cache hits ignored which CV produced the score** (`search_prefilter.py` / `agent.py`)

jobradar/cache.py

@@ -971,15 +971,6 @@ def get_latest_cv_hash() -> str:
     return row["cv_hash"] if row is not None else ""
 
 
-def update_job_assessment(dedup_key: str, assessment: JobAssessment) -> None:
-    """单独更新某条 JD 的 assessment（独立评估命令使用）。"""
-    with _conn() as con:
-        con.execute(
-            "UPDATE job_cache SET assessment = ? WHERE dedup_key = ?",
-            (assessment.model_dump_json(), dedup_key),
-        )
-
-
 def get_unassessed_jobs(limit: int = 200) -> list[JobResult]:
     """返回当前 CV 下尚无匹配结果的未过期职位。
 

jobradar/cli.py

@@ -498,7 +498,7 @@ def assess(
 
     # Step 3：加载全部已评估 JD 排序展示
     all_jobs = cache.get_recent_jobs(limit)
-    all_jobs = [j for j in all_jobs if j.match_score is not None or j.assessment is not None]
+    all_jobs = [j for j in all_jobs if j.match_score is not None]
     all_jobs.sort(key=lambda j: (j.effective_score if j.effective_score is not None else -1), reverse=True)
 
     if not all_jobs:

jobradar/display.py

@@ -114,12 +114,6 @@ def show_job_detail(job: JobResult) -> None:
         )
         if job.effective_keywords:
             lines.append(f"[bold cyan]匹配关键词：[/bold cyan]{', '.join(job.effective_keywords)}")
-    elif job.assessment:
-        a = job.assessment
-        bar = "█" * a.score + "░" * (10 - a.score)
-        lines.append(f"\n[bold cyan]匹配分：[/bold cyan]{a.score}/10  {bar}")
-        if a.matched_keywords:
-            lines.append(f"[bold cyan]匹配关键词：[/bold cyan]{', '.join(a.matched_keywords)}")
 
     lines.append("")
     lines.append(job.description_snippet or "（无摘要）")
@@ -155,22 +149,8 @@ def _job_to_markdown(job: JobResult) -> str:
         lines.append("**劣势 / 差距**")
         for w in m.weaknesses + m.risks:
             lines.append(f"- {w}")
-    elif job.assessment:
-        a = job.assessment
-        bar = "█" * a.score + "░" * (10 - a.score)
-        lines.append(f"**整体匹配分**：{a.score}/10  `{bar}`")
-        if a.matched_keywords:
-            lines.append(f"\n**匹配关键词**：{', '.join(a.matched_keywords)}")
-        lines.append("")
-        lines.append("**优势**")
-        for s in a.strengths:
-            lines.append(f"- {s}")
-        lines.append("")
-        lines.append("**劣势 / 差距**")
-        for w in a.weaknesses:
-            lines.append(f"- {w}")
     else:
-        lines.append("_本职位未进行 CV 匹配评估（无 CV 数据或评估被跳过）。_")
+        lines.append("_本职位在当前 CV 下尚无匹配结果，可运行 `jobradar assess` 补算。_")
 
     return "\n".join(lines)
 

jobradar/schemas.py

@@ -548,44 +548,32 @@ def is_possibly_closed(self) -> bool:
             return False
         return bool(_CLOSED_PATTERN.search(self.description_snippet))
 
+    # 以下属性只认现代 match_score。legacy ``assessment`` 已退出评分口径：
+    # 它不记录 cv_hash，且用的是 0~10 分制，与 match_score 的 0~100 混在一起排序会失真。
+    # 该字段仍保留用于读取历史数据（见 model_quality_audit）。
+
     @property
     def effective_score(self) -> float | None:
-        if self.match_score is not None:
-            return self.match_score.overall_score
-        if self.assessment is not None:
-            return float(self.assessment.score)
-        return None
+        return self.match_score.overall_score if self.match_score is not None else None
 
     @property
     def effective_strengths(self) -> list[str]:
-        if self.match_score is not None:
-            return self.match_score.strengths
-        if self.assessment is not None:
-            return self.assessment.strengths
-        return []
+        return self.match_score.strengths if self.match_score is not None else []
 
     @property
     def effective_weaknesses(self) -> list[str]:
-        if self.match_score is not None:
-            return self.match_score.weaknesses
-        if self.assessment is not None:
-            return self.assessment.weaknesses
-        return []
+        return self.match_score.weaknesses if self.match_score is not None else []
 
     @property
     def effective_keywords(self) -> list[str]:
         if self.match_score is not None and (self.jd_profile is not None or self.job_summary is not None):
             return self.match_score.matched_keywords[:6]
-        if self.assessment is not None:
-            return self.assessment.matched_keywords[:6]
         return []
 
     @property
     def is_effectively_relevant(self) -> bool:
         if self.match_score is not None:
             return self.match_score.recommendation != "skip"
-        if self.assessment is not None:
-            return self.assessment.is_relevant
         return True
 
 

jobradar/search_assessment_stage.py

@@ -332,7 +332,6 @@ def _is_visible_job(job_obj) -> bool:
                     "expires_at": cached_job.expires_at,
                     "is_complete": cached_job.is_complete,
                     "coarse_filter": cached_job.coarse_filter,
-                    "assessment": assessment.to_job_assessment(),
                 }
             )
             if assessment.relevant:
@@ -421,7 +420,6 @@ def _is_visible_job(job_obj) -> bool:
             else:
                 logger.debug("LLM assess matched: %s | score=%d", title, assessment.score)
 
-            job_assessment = assessment.to_job_assessment() if has_cv else None
             dedup_key = make_dedup_key(job.get("company", ""), title)
             raw_sources = job_all_sources.get(
                 dedup_key,
@@ -438,7 +436,6 @@ def _is_visible_job(job_obj) -> bool:
                     "expires_at": expires_at,
                     "is_complete": job.get("is_complete", True),
                     "coarse_filter": job.get("coarse_filter"),
-                    "assessment": job_assessment,
                     "sources": [entry["source"] for entry in raw_sources],
                     "raw_sources": raw_sources,
                 }
@@ -456,7 +453,6 @@ def _is_visible_job(job_obj) -> bool:
                     expires_at=expires_at,
                     is_complete=job.get("is_complete", True),
                     coarse_filter=job.get("coarse_filter"),
-                    assessment=job_assessment,
                 )
                 if llm is not None:
                     pending_tasks.append(

jobradar/search_prefilter.py

@@ -80,24 +80,23 @@ def classify_cache_hit(cached_job: JobResult, cv_hash: str, language: str = "zh"
     """判断缓存命中的职位在当前 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"
+    if not cv_hash:
+        # 无 CV 时评分无从谈起，直接复用缓存内容。
+        # 不能返回 reassess：该场景下 flush_assessments 的 has_cv 为假，
+        # patch_pending 分支会被整个跳过，这些职位将从结果中消失。
+        return "reuse"
 
-    # 无从判断 CV 版本时退回 legacy 列，保持既有行为。
-    if cached_job.assessment is not None:
-        return "reuse" if cached_job.assessment.is_relevant else "skip"
-    return "reassess"
+    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"
 
 
 @dataclass

tests/test_new_features.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import io
+import os
 import tempfile
 from pathlib import Path
 from unittest.mock import patch
@@ -695,7 +696,8 @@ def test_sr_dot_filtered(self):
 
 
 class TestJobResultCompatibility:
-    def test_job_result_without_match_uses_legacy_assessment(self):
+    def test_legacy_assessment_no_longer_feeds_effective_score(self):
+        """legacy assessment 已退出评分口径：它不记录 cv_hash，且分制与 match_score 不同。"""
         job = JobResult(
             title="Backend Engineer",
             company="Acme",
@@ -709,8 +711,23 @@ def test_job_result_without_match_uses_legacy_assessment(self):
             ),
         )
 
-        assert job.effective_score == 8.0
-        assert job.effective_keywords == ["Python", "SQL"]
+        assert job.effective_score is None
+        assert job.effective_strengths == []
+        assert job.effective_weaknesses == []
+        assert job.effective_keywords == []
+        # 无匹配结果时默认可见，由后续评估决定去留。
+        assert job.is_effectively_relevant is True
+        # 字段本身保留，历史数据仍可读取（model_quality_audit 依赖）。
+        assert job.assessment is not None and job.assessment.score == 8
+
+    def test_legacy_rejection_no_longer_hides_job(self):
+        job = JobResult(
+            title="Backend Engineer",
+            company="Acme",
+            url="https://example.com/job",
+            assessment=JobAssessment(score=1, is_relevant=False),
+        )
+
         assert job.is_effectively_relevant is True
 
 
@@ -786,6 +803,83 @@ def fake_match_job_to_cv(
         assert rejected == 0
         assert saved == 0
 
+    def test_pipeline_no_longer_writes_the_legacy_assessment_column(self, db, monkeypatch: pytest.MonkeyPatch):
+        """评分只经 job_matches 落库；legacy 列不再由管道写入。"""
+        import sqlite3
+
+        import jobradar.search_assessment_stage as stage
+        from jobradar.assessment import JDAssessment
+        from jobradar.search_prefilter import PrefilterResult
+
+        scraped = {
+            "title": "Backend Engineer",
+            "company": "Acme",
+            "location": "Dublin",
+            "url": "https://example.com/backend",
+            "source": "indeed.ie",
+            "description_snippet": "Build Python APIs.",
+            "is_complete": True,
+        }
+
+        monkeypatch.setattr(
+            stage,
+            "batch_assess_jds",
+            lambda jobs, profile, llm, language="zh": [
+                JDAssessment(
+                    relevant=True,
+                    reason="match",
+                    score=9,
+                    strengths=["Python"],
+                    weaknesses=["Cloud"],
+                    matched_keywords=["Python"],
+                )
+                for _ in jobs
+            ],
+        )
+
+        def fake_evaluate(job, profile, llm, cv_hash="", language="zh"):
+            jd_profile = JDProfile(job_id=job.dedup_key, title=job.title, company=job.company)
+            return jd_profile, MatchScore(
+                job_id=job.dedup_key,
+                cv_hash=cv_hash,
+                overall_score=78,
+                title_score=80,
+                seniority_score=80,
+                must_have_score=75,
+                nice_to_have_score=70,
+                domain_score=75,
+                location_score=100,
+                language_score=100,
+                risk_penalty=5,
+                recommendation="apply",
+            )
+
+        monkeypatch.setattr(stage.cache, "get_jd_profile", lambda *args, **kwargs: None)
+        monkeypatch.setattr(stage, "evaluate_job_once", fake_evaluate)
+
+        keys, _, saved = stage.flush_assessments(
+            PrefilterResult(pending=[(scraped, scraped["description_snippet"], None)]),
+            job_all_sources={},
+            profile=_make_profile(),
+            llm=_make_llm(),
+            cv_hash="current-cv",
+            cb=lambda msg: None,
+            on_job=None,
+            language="zh",
+        )
+
+        assert saved == 1
+        # 评分落在 job_matches，按 (job_id, cv_hash) 归属。
+        persisted = db.get_job_match(keys[0], "current-cv", scraped["description_snippet"])
+        assert persisted is not None and persisted.overall_score == 78
+
+        con = sqlite3.connect(os.environ["CACHE_DB_PATH"])
+        raw = con.execute(
+            "SELECT assessment FROM job_cache WHERE dedup_key = ?", (keys[0],)
+        ).fetchone()
+        con.close()
+        assert raw[0] is None
+
 
 class TestPipelineStats:
     def test_write_report_accepts_explicit_directory(self, tmp_path: Path):

tests/test_search_prefilter.py

@@ -133,8 +133,13 @@ def test_stale_prompt_version_triggers_reassessment(self, temp_db):
 
         assert classify_cache_hit(job, CURRENT_CV) == "reassess"
 
-    def test_without_cv_hash_falls_back_to_legacy_column(self, temp_db):
-        """无从判断 CV 版本时保持既有行为，避免无 CV 场景下丢结果。"""
+    def test_without_cv_hash_everything_is_reused(self, temp_db):
+        """无 CV 时评分无从谈起，一律复用缓存内容。
+
+        不能返回 reassess：该场景下 flush_assessments 的 has_cv 为假，
+        patch_pending 分支会被整个跳过，这些职位将从结果中消失。
+        legacy assessment 也不再参与判定——它不记录 cv_hash。
+        """
         from jobradar.search_prefilter import classify_cache_hit
 
         relevant = make_job(assessment=JobAssessment(score=8, is_relevant=True))
@@ -146,8 +151,8 @@ def test_without_cv_hash_falls_back_to_legacy_column(self, temp_db):
         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"
+        assert classify_cache_hit(rejected, "") == "reuse"
+        assert classify_cache_hit(unassessed, "") == "reuse"
 
 
 class TestPrefilterCacheHit: