fix: select jobs for assessment by current CV match, not legacy column

CHANGELOG.md

@@ -2,6 +2,13 @@
 
 ## [Unreleased]
 
+### Bug Fixes
+
+- **`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.
+  `assess` now writes `job_matches` through the shared evaluation path (`evaluate_cached_jobs`) rather than only refreshing the legacy column, which keeps repeated runs idempotent: previously the new selection rule would have re-queued the same jobs on every invocation.
+
 ### Documentation
 
 - **Release highlights in Chinese, English and Spanish** (`docs/release-highlights.*.md` / `README*.md`)

docs/RUNBOOK.md

@@ -294,7 +294,7 @@ jobradar_email_sync_runs_total{status="failed",reason="auth"}
 | Indeed 抓取结果为 0 | JobSpy 被 Indeed 限流（常见，非故障） | 等几小时重试；查日志中 JobSpy WARNING |
 | Adzuna 返回 429 | 速率限制 | 调大 `_MIN_INTERVAL`（当前 1.2s）；核对 `.env` 中 `ADZUNA_APP_ID` / `ADZUNA_APP_KEY` |
 | LLM 评估全部拒绝 | `cv_summary` / `cv_skills` 提取失败 | 检查 CV 解析结果；`uv run jobradar assess` 补跑评估 |
-| 职位无模型评分 | 来自旧缓存（assessment 为 NULL） | `uv run jobradar assess` |
+| 职位无模型评分 | 当前 `cv_hash` 下尚无 `job_matches` 记录（换 CV 后常见） | `uv run jobradar assess` 为当前 CV 补算 |
 
 ### 安全注意
 

jobradar/cache.py

@@ -981,16 +981,27 @@ def update_job_assessment(dedup_key: str, assessment: JobAssessment) -> None:
 
 
 def get_unassessed_jobs(limit: int = 200) -> list[JobResult]:
-    """返回尚无现代匹配结果且缺少 legacy assessment 的未过期职位。"""
+    """返回当前 CV 下尚无匹配结果的未过期职位。
+
+    只按现代 ``job_matches`` 判定：legacy ``assessment`` 列没有 cv_hash 归属，
+    用它筛选会把换 CV 后本该重评的职位挡在门外。
+    """
     with _conn() as con:
-        rows = con.execute(
-            "SELECT * FROM job_cache WHERE assessment IS NULL ORDER BY fetched_at DESC LIMIT ?",
-            (limit,),
-        ).fetchall()
-    jobs = [_row_to_job(r) for r in rows]
-    for job in jobs:
+        rows = con.execute("SELECT * FROM job_cache ORDER BY fetched_at DESC").fetchall()
+
+    unassessed: list[JobResult] = []
+    for row in rows:
+        job = _row_to_job(row)
+        if job.is_expired:
+            continue
+        # 先过滤过期再挂载 match，避免为已过期职位做多余的查询。
         _attach_latest_match(job)
-    return [j for j in jobs if not j.is_expired and job.match_score is None]
+        if job.match_score is not None:
+            continue
+        unassessed.append(job)
+        if len(unassessed) >= limit:
+            break
+    return unassessed
 
 
 # ─── 流式搜索候选缓存 ─────────────────────────────────────────────────────────

jobradar/cli.py

@@ -1,6 +1,7 @@
 """CLI 入口：Typer 命令定义。"""
 from __future__ import annotations
 
+import hashlib
 import json
 import os
 import webbrowser
@@ -14,7 +15,7 @@
 
 from jobradar import cache
 from jobradar.agent import run_search
-from jobradar.assessment import batch_assess_jds
+from jobradar.assessment import gate_worker_count
 from jobradar.cv_extractor import extract_cv_profile
 from jobradar.cv_reader import read_cv
 from jobradar.display import (
@@ -43,6 +44,7 @@
     get_saved_defaults,
     save_env_key,
 )
+from jobradar.search_assessment_stage import evaluate_cached_jobs
 from jobradar.telemetry import telemetry
 from jobradar.tools import verify_job_active
 
@@ -447,11 +449,13 @@ def assess(
     llm = LLMConfig(provider=effective_provider, model=effective_model)
     console.print(f"[dim]使用模型：{llm.provider} / {llm.model}[/dim]")
 
-    # Step 1：获取 CVProfile
+    # Step 1：获取 CVProfile 及其 cv_hash（匹配结果按 cv_hash 归属，必须一并确定）
     profile = None
+    cv_hash = ""
     if cv_path is not None:
         try:
             cv_text = read_cv(cv_path)
+            cv_hash = hashlib.sha256(cv_text.encode()).hexdigest()
             with Progress(SpinnerColumn(), TextColumn("{task.description}"), transient=True) as p:
                 p.add_task("解析 CV 信息...", total=None)
                 profile = extract_cv_profile(cv_text, llm=llm)
@@ -463,26 +467,34 @@ def assess(
         if profile is None:
             console.print("[red]缓存中没有 CVProfile，请提供 CV 文件路径。[/red]")
             raise typer.Exit(1)
+        cv_hash = cache.get_latest_cv_hash()
         console.print(f"[dim]使用缓存的 CVProfile：{profile.summary[:40]}（{profile.seniority_display}）[/dim]")
 
-    # Step 2：加载未评估的 JD 并批量评估
+    if not cv_hash:
+        console.print("[red]无法确定 CV 版本（cv_hash），请提供 CV 文件路径。[/red]")
+        raise typer.Exit(1)
+
+    # Step 2：为当前 CV 下尚无匹配结果的 JD 补算评分
     unassessed = cache.get_unassessed_jobs(limit=limit)
     if unassessed:
-        console.print(f"\n[bold]待评估 JD：{len(unassessed)} 条[/bold]")
-
-        job_inputs = [(j.title, j.description_snippet or "") for j in unassessed]
+        console.print(f"\n[bold]待评估 JD：{len(unassessed)} 条[/bold]（CV 版本 {cv_hash[:8]}）")
 
         with Progress(SpinnerColumn(), TextColumn("{task.description}"), transient=True) as p:
-            p.add_task(f"批量评估（每批 8 条，共 {len(unassessed)} 条）...", total=None)
-            with telemetry.timer("JD 批量评估"):
-                assessments = batch_assess_jds(job_inputs, profile, llm)
-
-        for job, jda in zip(unassessed, assessments):
-            cache.update_job_assessment(job.dedup_key, jda.to_job_assessment())
+            p.add_task(f"评估中（共 {len(unassessed)} 条）...", total=None)
+            with telemetry.timer("JD 评估"):
+                succeeded, failed = evaluate_cached_jobs(
+                    unassessed,
+                    profile=profile,
+                    llm=llm,
+                    cv_hash=cv_hash,
+                    workers=gate_worker_count(llm.provider),
+                )
 
-        console.print(f"[green]已更新 {len(unassessed)} 条评估结果。[/green]")
+        console.print(f"[green]已更新 {succeeded} 条评估结果。[/green]")
+        if failed:
+            console.print(f"[yellow]{failed} 条评估失败，详见 logs/jobradar.log。[/yellow]")
     else:
-        console.print("[dim]所有 JD 已有评估，跳过评估步骤。[/dim]")
+        console.print("[dim]当前 CV 下所有 JD 均已评估，跳过评估步骤。[/dim]")
 
     # Step 3：加载全部已评估 JD 排序展示
     all_jobs = cache.get_recent_jobs(limit)

jobradar/search_assessment_stage.py

@@ -187,6 +187,51 @@ def _evaluate_jobs(
     return completed
 
 
+def evaluate_cached_jobs(
+    jobs: list[JobResult],
+    *,
+    profile: CVProfile,
+    llm,
+    cv_hash: str,
+    language: str = "zh",
+    workers: int = 1,
+) -> tuple[int, int]:
+    """为已缓存职位补算 JD profile 与 CV 匹配，返回 (成功, 失败) 条数。
+
+    ``assess`` 命令用此入口补全 ``job_matches``。写入 legacy ``assessment`` 列是不够的：
+    待评估判定看的是当前 cv_hash 下有无匹配结果，只写旧列会让同一批职位被反复捞出。
+    """
+    if not jobs:
+        return 0, 0
+    tasks = [
+        JobEvaluationTask(
+            key=job.dedup_key,
+            job=job,
+            full_jd=job.description_snippet,
+            kind="assess",
+            source=job.sources[0] if job.sources else "unknown",
+        )
+        for job in jobs
+    ]
+    metrics = AssessmentConcurrencyMetrics(workers=workers)
+    results = _evaluate_jobs(
+        tasks,
+        profile=profile,
+        llm=llm,
+        cv_hash=cv_hash,
+        language=language,
+        executor=None,
+        workers=workers,
+        metrics=metrics,
+    )
+    failed = 0
+    for task, error in results:
+        if error is not None:
+            failed += 1
+            logger.warning("Assessment failed for %s: %s", task.key, error)
+    return len(results) - failed, failed
+
+
 def flush_assessments(
     pf: PrefilterResult,
     job_all_sources: dict[str, list[dict]],

tests/test_cache.py

@@ -11,8 +11,10 @@
     CoarseFilterResult,
     CoverLetter,
     CVOptimization,
+    CVProfile,
     InterviewPrep,
     JDProfile,
+    JobAssessment,
     JobResult,
     JobSummary,
     MatchScore,
@@ -477,3 +479,169 @@ def test_delete_jobs_removes_cv_optimization(self, temp_db):
 
         temp_db.delete_jobs([job.dedup_key])
         assert temp_db.get_cv_optimization(job.dedup_key, "cv123", "desc") is None
+
+
+class TestUnassessedJobs:
+    """get_unassessed_jobs 只按现代 job_matches 判定，不看 legacy assessment 列。"""
+
+    CV_HASH = "cv-current"
+
+    def _seed_cv(self, temp_db):
+        temp_db.save_cv_profile(
+            self.CV_HASH,
+            CVProfile(summary="Backend engineer", skills=["Python"]),
+        )
+
+    def _save_match(self, temp_db, job, cv_hash=None):
+        from jobradar.matching import match_prompt_version
+
+        match = MatchScore(
+            job_id=job.dedup_key,
+            cv_hash=cv_hash or self.CV_HASH,
+            overall_score=82,
+            title_score=80,
+            seniority_score=70,
+            must_have_score=90,
+            nice_to_have_score=60,
+            domain_score=75,
+            location_score=85,
+            language_score=100,
+            risk_penalty=3,
+            recommendation="apply",
+        )
+        temp_db.save_job_match(
+            match,
+            job.description_snippet,
+            prompt_version=match_prompt_version("zh"),
+        )
+
+    def test_legacy_assessment_does_not_hide_unmatched_job(self, temp_db):
+        """核心修复：旧评分存在不代表当前 CV 评过，该职位仍需重评。"""
+        self._seed_cv(temp_db)
+        job = make_job(
+            description_snippet="Build Python APIs.",
+            assessment=JobAssessment(score=7, is_relevant=True),
+        )
+        temp_db.save_job(job)
+
+        result = temp_db.get_unassessed_jobs()
+        assert [j.dedup_key for j in result] == [job.dedup_key]
+
+    def test_rejected_legacy_assessment_still_returned(self, temp_db):
+        """旧 CV 判定的 is_relevant=False 不应永久排除该职位。"""
+        self._seed_cv(temp_db)
+        job = make_job(
+            description_snippet="Build Python APIs.",
+            assessment=JobAssessment(score=2, is_relevant=False),
+        )
+        temp_db.save_job(job)
+
+        assert [j.dedup_key for j in temp_db.get_unassessed_jobs()] == [job.dedup_key]
+
+    def test_job_matched_under_current_cv_is_excluded(self, temp_db):
+        self._seed_cv(temp_db)
+        job = make_job(description_snippet="Build Python APIs.")
+        temp_db.save_job(job)
+        self._save_match(temp_db, job)
+
+        assert temp_db.get_unassessed_jobs() == []
+
+    def test_match_from_another_cv_does_not_count(self, temp_db):
+        """换 CV 后旧 cv_hash 的匹配结果不算已评估。"""
+        self._seed_cv(temp_db)
+        job = make_job(description_snippet="Build Python APIs.")
+        temp_db.save_job(job)
+        self._save_match(temp_db, job, cv_hash="cv-outdated")
+
+        assert [j.dedup_key for j in temp_db.get_unassessed_jobs()] == [job.dedup_key]
+
+    def test_each_job_filtered_independently(self, temp_db):
+        """回归：过滤条件曾误用循环残留变量，导致整批结果由最后一条决定。"""
+        self._seed_cv(temp_db)
+        unmatched = [
+            make_job(company=f"Company{i}", url=f"http://example.com/{i}", description_snippet="desc")
+            for i in range(3)
+        ]
+        for job in unmatched:
+            temp_db.save_job(job)
+
+        # 最后写入的职位已有当前 CV 的匹配结果，不应影响前面三条的判定。
+        matched = make_job(company="Matched", url="http://example.com/matched", description_snippet="desc")
+        temp_db.save_job(matched)
+        self._save_match(temp_db, matched)
+
+        result = {j.dedup_key for j in temp_db.get_unassessed_jobs()}
+        assert result == {j.dedup_key for j in unmatched}
+
+    def test_expired_jobs_excluded(self, temp_db):
+        self._seed_cv(temp_db)
+        expired = make_job(
+            company="Expired",
+            url="http://example.com/expired",
+            description_snippet="desc",
+            expires_at=datetime.utcnow() - timedelta(days=1),
+        )
+        live = make_job(company="Live", url="http://example.com/live", description_snippet="desc")
+        temp_db.save_job(expired)
+        temp_db.save_job(live)
+
+        assert [j.dedup_key for j in temp_db.get_unassessed_jobs()] == [live.dedup_key]
+
+    def test_limit_counts_returned_jobs(self, temp_db):
+        """limit 作用于实际返回条数，而非过滤前的扫描量。"""
+        self._seed_cv(temp_db)
+        for i in range(5):
+            temp_db.save_job(
+                make_job(company=f"Company{i}", url=f"http://example.com/{i}", description_snippet="desc")
+            )
+
+        assert len(temp_db.get_unassessed_jobs(limit=2)) == 2
+
+    def test_assess_cycle_is_idempotent(self, temp_db, monkeypatch):
+        """回归：补跑评估必须写入 job_matches，否则同一批职位会被反复捞出。"""
+        from jobradar import search_assessment_stage as stage
+        from jobradar.llm_backend import LLMConfig
+
+        self._seed_cv(temp_db)
+        for i in range(3):
+            temp_db.save_job(
+                make_job(
+                    company=f"Company{i}",
+                    url=f"http://example.com/{i}",
+                    description_snippet="Build Python APIs.",
+                )
+            )
+
+        def fake_evaluate(job, profile, llm, cv_hash="", language="zh"):
+            jd_profile = JDProfile(job_id=job.dedup_key, title=job.title, company=job.company)
+            match = 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="apply",
+            )
+            return jd_profile, match
+
+        monkeypatch.setattr(stage, "evaluate_job_once", fake_evaluate)
+
+        pending = temp_db.get_unassessed_jobs()
+        assert len(pending) == 3
+
+        succeeded, failed = stage.evaluate_cached_jobs(
+            pending,
+            profile=CVProfile(summary="Backend engineer", skills=["Python"]),
+            llm=LLMConfig(provider="gemini", model="test-model"),
+            cv_hash=self.CV_HASH,
+        )
+
+        assert (succeeded, failed) == (3, 0)
+        # 第二次调用必须为空，否则 assess 会陷入每次重评同一批职位的循环。
+        assert temp_db.get_unassessed_jobs() == []