fix(scoring): 百分位精选改为名次截断，消除 off-by-one 与并列整块放行

backend/app/services/scoring_engine.py

@@ -450,25 +450,40 @@ def score_items(items: list[ScoringInput]) -> list[tuple[ScoreBreakdown, Scoring
 
     # Phase 5: Determine threshold and mark selected
     if cfg["curation_mode"] == "percentile":
-        # Use final_score ranking: top (100 - percentile)% are selected
-        # e.g. curation_percentile=70 -> top 30% selected, bounded by base quality.
         # P70 门槛只由真实 LLM 评分决定：local_fallback 的确定性假分
         # （curation 收敛 61-64 窄带）不得污染百分位门槛、挤掉真实内容。
         # 降级项仍参与 selected 判定与展示（前端带「本地速览」标记），
         # 由 requeue job 尽快换回真实分析（#90）。全候选皆降级时改用全量分数：
         # 空列表会被 _compute_percentile_threshold 回退到全局默认阈值（55），
         # 而降级内容的 final_score 普遍低于它，那样整批一条都选不出来。
-        real_scores = [bd.final_score for bd, item in results if item.summary_source != "local_fallback"]
-        final_scores = real_scores or [bd.final_score for bd, _ in results]
-        actual_threshold = _compute_percentile_threshold(
-            final_scores,
-            cfg["curation_percentile"],
-        )
-    else:
-        actual_threshold = cfg["curation_threshold"]
-
+        # 名次截断（2026-10-09 精选率层 2 修复）：按 final_score 降序取
+        # top-(100-percentile)%，取代旧「nearest-rank 门槛 + >= 比较」。
+        # 旧口径放行 n-rank+1 条（n=100 的 P70 下 31%>目标 30%），且门槛值
+        # 上的并列块被整块放行（假分钉门槛时精选率达 55% 的机制之一）；
+        # 纯 > 比较则会把小批次与均匀并列批次打到一条不选。名次截断在
+        # 并列块处按展示序截断（不放大也不全灭），小批次至少保底 1 条
+        # （须过下方质量 gate——保底不是无条件下限）。
+        ordered = sorted(results, key=lambda pair: pair[0].final_score, reverse=True)
+        n = len(ordered)
+        rank = min(math.ceil(cfg["curation_percentile"] / 100.0 * n), n)
+        keep = max(1, n - rank) if n else 0
+        cutoff_score = ordered[keep - 1][0].final_score if keep else 0.0
+
+        for position, (bd, _item) in enumerate(ordered):
+            bd.threshold_used = round(cutoff_score, 2)
+            bd.selected = (
+                position < keep
+                and bd.base_score >= cfg["min_selected_base_score"]
+                and bd.quality_factor > cfg["quality_gate_floor"]
+                and bd.risk_factor > cfg["risk_soft_floor"]
+            )
+        return results
+
+    actual_threshold = cfg["curation_threshold"]
     for bd, _item in results:
         bd.threshold_used = round(actual_threshold, 2)
+        # fixed 模式是绝对质量线：>= 语义正确（达到线即入选），无百分位
+        # rank 的 off-by-one 问题。
         bd.selected = (
             bd.final_score >= actual_threshold
             and bd.base_score >= cfg["min_selected_base_score"]

backend/tests/test_content_event_consumers.py

@@ -312,12 +312,18 @@ def _pick_row(content_id: int) -> dict:
 async def test_today_picks_hides_all_event_members_and_exposes_reason(
     monkeypatch,
 ):
-    rows = [
-        _pick_row(1),
-        _pick_row(2),
-        _pick_row(3),
-        _pick_row(4),
-    ]
+    rows = [_pick_row(i) for i in (1, 2, 3, 4)]
+    # 名次截断口径（2026-10-09）下 n=4 的 keep=1 装不下「canonical+独立内容」
+    # 两条意图目标：id=1/4 拉到最高两档、补 3 条低分陪跑撑到 n=7（keep=2）。
+    rows[0]["curation_score"] = 94.0
+    rows[3]["curation_score"] = 95.0
+    # 事件成员（id2/3）在 scoring 前被剔除出候选：陪跑补足 5 条，
+    # 保证剔除后候选仍 n=7（keep=2），canonical(id1) 与独立内容(id4)稳进。
+    for i in (5, 6, 7, 8, 9):
+        r = _pick_row(i)
+        r["curation_score"] = 50.0
+        r["quality_score"] = 40.0
+        rows.append(r)
     assignments = {
         1: EventAssignment(1, 10, 1, True),
         2: EventAssignment(2, 10, 1, False, "corroboration", 0.93),
@@ -387,9 +393,11 @@ async def load_groups(
         limit=None,
     )
 
-    assert [item["id"] for item in payload["items"]] == [1, 4]
+    assert {item["id"] for item in payload["items"]} == {1, 4}
     assert payload["event_members_hidden"] == 2
-    normalization = payload["items"][0]["normalization"]
+    # 名次截断后 items 按分数序（id4 在前），canonical 的归一化信息按 id 定位
+    canonical = next(item for item in payload["items"] if item["id"] == 1)
+    normalization = canonical["normalization"]
     assert normalization["member_count"] == 2
     assert normalization["has_more"] is True
     assert normalization["members"][0]["relation_type"] == "corroboration"

backend/tests/test_scoring_engine.py

@@ -217,44 +217,28 @@ def test_diversity_promotes_other_source_over_same_source_run():
     assert final_ids.index(99) < final_ids.index(2)
 
 
-def test_p70_threshold_excludes_local_fallback_fake_scores():
-    """#90：local_fallback 的确定性假分不得污染 P70 门槛。
-
-    三组对照（自证明区分度——守门断言保证若删掉排除逻辑本测试必红）：
-    - real_only：3 条真实分（90/80/70），门槛 T1 只由真实分决定；
-    - mixed：再混入 2 条假分（85/75，标记 local_fallback）——真实项门槛
-      仍应为 T1（排除生效）；
-    - full：同样的 2 条 85/75 但标记为真实分析——门槛 T2 应不同于 T1，
-      证明该分数组合确实会移动门槛（若排除逻辑失效，mixed 会退化成
-      full 的门槛，测试转红）。
+def test_local_fallback_fake_scores_do_not_displace_real_selections():
+    """#90（2026-10-09 名次截断语义）：假分参与排序与展示，但中低分假分
+    不得挤占真实内容的入选名额——真实内容的入选集合与无假分批次一致。
     """
 
     def _scored(content_id: int, curation: int, **extra) -> ScoringInput:
         return _item(content_id, curation_score=curation, **extra)
 
-    real_items = [_scored(1, 90), _scored(2, 80), _scored(3, 70)]
-    fallback_items = [
+    real = [_scored(1, 90), _scored(2, 80), _scored(3, 70)]
+    fake = [
         _scored(98, 85, summary_source="local_fallback"),
         _scored(99, 75, summary_source="local_fallback"),
     ]
-    same_scores_as_real = [_scored(98, 85), _scored(99, 75)]  # 不带标记
 
-    baseline = score_items(list(real_items))
-    mixed = score_items(real_items + fallback_items)
-    full = score_items(real_items + same_scores_as_real)
+    baseline_selected = {bd.content_id for bd, _ in score_items(list(real)) if bd.selected}
+    mixed = score_items(real + fake)
+    mixed_real_selected = {bd.content_id for bd, it in mixed if bd.selected and it.summary_source != "local_fallback"}
 
-    t_baseline = {bd.threshold_used for bd, _ in baseline}
-    t_mixed_real = {bd.threshold_used for bd, item in mixed if item.content_id in (1, 2, 3)}
-    t_full = {bd.threshold_used for bd, _ in full}
-
-    assert len(t_baseline) == 1 and len(t_full) == 1
-    assert t_mixed_real == t_baseline, "混入 fallback 假分后真实项门槛不应改变"
-    assert t_full != t_baseline, "守门断言：85/75 混入真实批应移动门槛——本断言失败说明对照构造失去区分度"
-
-    # fallback 项自身仍参与判定与展示（不静默消失），且用同一门槛判定
-    fallback_results = {item.content_id: bd for bd, item in mixed if item.content_id in (98, 99)}
-    assert len(fallback_results) == 2
-    assert {bd.threshold_used for bd in fallback_results.values()} == t_baseline
+    assert baseline_selected == {1}
+    assert mixed_real_selected == {1}, "85/75 的假分不得挤掉真实头名的入选名额"
+    # 假分按分数参与排序（85 排第二），但 keep=1 截断不入——宁缺毋滥
+    assert all(not bd.selected for bd, it in mixed if it.summary_source == "local_fallback")
 
 
 def test_all_fallback_batch_falls_back_to_full_scores():
@@ -285,3 +269,36 @@ def test_scoring_input_defaults_summary_source_none():
     """未传 summary_source 的旧构造路径默认 None（表示真实 LLM 分析）。"""
     item = _item(1)
     assert item.summary_source is None
+
+
+def test_percentile_selection_is_exact_top_percentile():
+    """2026-10-09 精选率层 2 修复：放行恰为 top-(100-percentile)%，无 off-by-one。
+
+    >= 语义放行 n-rank+1 条（n=10 的 P70 下 4 条=40%）；> 语义恰 3 条=30%。
+    """
+    items = [
+        _item(i, curation_score=70 + i * 3, creator_score=70, viral_score=70, quality_score=70)
+        for i in range(10)  # 70..97 十档不并列
+    ]
+    scored = score_items(items)
+    selected = [bd for bd, _ in scored if bd.selected]
+    assert len(selected) == 3, f"P70 下 10 条应恰选 3 条（30%），实际 {len(selected)}"
+
+
+def test_percentile_tie_block_at_threshold_is_rejected_not_admitted():
+    """并列块不再被门槛整块放行：>= 曾把 7 条并列 60 分全部放行（70%+）。
+
+    名次截断下按 final_score 排序取 top-3，7 条并列 60 分一条不占名额。
+    """
+    items = [
+        _item(i, curation_score=60, creator_score=70, viral_score=70, quality_score=70)
+        for i in range(1, 8)  # 7 条并列 60
+    ] + [
+        _item(8, curation_score=80, creator_score=75, viral_score=75, quality_score=75),
+        _item(9, curation_score=90, creator_score=80, viral_score=80, quality_score=80),
+        _item(10, curation_score=95, creator_score=85, viral_score=85, quality_score=85),
+    ]
+    scored = score_items(items)
+    selected = [bd for bd, _ in scored if bd.selected]
+    assert len(selected) == 3, f"门槛值并列块应整块不选，实际选了 {len(selected)}"
+    assert {bd.content_id for bd in selected} == {8, 9, 10}

backend/tests/test_today_picks.py

@@ -791,7 +791,7 @@ async def test_fallback_risk_threshold_aligned_to_82(monkeypatch):
     ids = await _seed_oltp_fixture(
         engine,
         extra_content=[
-            {"title": "中风险样本", "url": "https://example.com/mid-risk", "risk_score": 60.0},
+            {"title": "中风险样本", "url": "https://example.com/mid-risk", "risk_score": 60.0, "curation_score": 95.0},
             {
                 "title": "低质量1",
                 "url": "https://example.com/low-1",
@@ -820,6 +820,20 @@ async def test_fallback_risk_threshold_aligned_to_82(monkeypatch):
                 "curation_score": 50.0,
                 "risk_score": 10.0,
             },
+            {
+                "title": "低质量5",
+                "url": "https://example.com/low-5",
+                "quality_score": 40.0,
+                "curation_score": 50.0,
+                "risk_score": 10.0,
+            },
+            {
+                "title": "低质量6",
+                "url": "https://example.com/low-6",
+                "quality_score": 40.0,
+                "curation_score": 50.0,
+                "risk_score": 10.0,
+            },
         ],
     )
     mid_risk_id = ids[1]