fix: mark item rejected on selector_error for consistent bookkeeping

selector_error 分支（捕获 ValueError/FileNotFoundError）此前只返回 reason，
未 mark_item_rejected，导致 Phase 3 已 record 的 pending attempt 行永远停在
pending；而 hard-fail 分支会标记 rejected。两条失败路径落库风格现统一为
mark_item_rejected（异常路径无 outcome/observation，故不写 selector_scores）。

补单测 test_apply_grounded_selector_marks_rejected_on_error 守卫该路径。

app/question_gen/pipeline_v2.py

@@ -403,8 +403,12 @@ async def _apply_grounded_selector(
             session_id=session_id,
         )
     except (ValueError, FileNotFoundError) as e:
+        # 异常路径无 outcome/observation 可落，但仍须 mark_item_rejected，
+        # 与 hard-fail 路径落库风格一致（该 attempt 的 item 是死记录，重出新建 item_id）
+        reason = f"selector_error: {e}"
         logger.warning("slot {} selector 异常 (attempt {}): {}", slot_id, attempt, e)
-        return None, f"selector_error: {e}"
+        store.mark_item_rejected(item_id, reason)
+        return None, reason
 
     # observation 始终落库（含 hard-fail），供 EOB 退化观测与调参
     store.update_selector_scores(item_id, json.dumps(outcome.observation, ensure_ascii=False))

tests/unit/test_pipeline_selector_wiring.py

@@ -1,5 +1,7 @@
 """selector 织入辅助：正解文本提取 + 分流。"""
 
+from pathlib import Path
+
 import pytest
 
 from app.question_gen.pipeline_v2 import _extract_correct_text
@@ -40,3 +42,81 @@ def test_replace_candidate_options():
     assert new.options == ("A. 蒸", "B. 炒", "C. 煮", "D. 炸")
     assert new.question == "?"  # 其余字段不变
     assert new.source_nodes == ("n1",)
+
+
+class _RecordingStore:
+    """记录 store 调用的假实现，用于断言落库一致性。"""
+
+    def __init__(self) -> None:
+        self.rejected: list[tuple[str, str]] = []
+        self.selector_scores: list[tuple[str, str]] = []
+
+    def mark_item_rejected(self, item_id: str, reason: str) -> None:
+        self.rejected.append((item_id, reason))
+
+    def update_selector_scores(self, item_id: str, selector_scores_json: str) -> None:
+        self.selector_scores.append((item_id, selector_scores_json))
+
+
+class _ARStrategy:
+    """最小策略替身：仅暴露 uses_grounded_selector=True。"""
+
+    uses_grounded_selector = True
+
+
+class _DummyMaterial:
+    """占位素材：异常在 build_grounded_options 内抛出前从不被读取。"""
+
+
+async def test_apply_grounded_selector_marks_rejected_on_error(monkeypatch):
+    """selector 异常路径须 mark_item_rejected（与 hard-fail 落库一致），但不写 selector_scores。"""
+    import app.question_gen.distractor_selector as ds
+    from app.question_gen.generator_v2 import CandidateQuestion
+    from app.question_gen.pipeline_v2 import PipelineConfig, _apply_grounded_selector
+
+    async def _raise(**_kwargs):
+        raise ValueError("boom")
+
+    monkeypatch.setattr(ds, "build_grounded_options", _raise)
+
+    candidate = CandidateQuestion(
+        question_id="q",
+        video_id="v",
+        task_type="Action Recognition",
+        skill_target="M1_AR",
+        question="?",
+        options=("A. a", "B. b", "C. c", "D. d"),
+        answer="A",
+        source_nodes=("n1",),
+        difficulty="hard",
+    )
+    config = PipelineConfig(
+        per_type=1,
+        retry_limit=1,
+        heavy_sample_rate=0.0,
+        dedup_threshold=0.85,
+        concurrency=1,
+        seed=0,
+        output_dir=Path("."),
+    )
+    store = _RecordingStore()
+
+    result_candidate, reason = await _apply_grounded_selector(
+        candidate,
+        _ARStrategy(),
+        _DummyMaterial(),
+        vlm=None,
+        config=config,
+        store=store,
+        item_id="item-1",
+        slot_id="slot-1",
+        attempt=1,
+        session_id="s",
+    )
+
+    assert result_candidate is None
+    assert reason is not None
+    assert reason.startswith("selector_error: ")
+    assert store.rejected == [("item-1", reason)]
+    # 异常路径无 observation，故不落 selector_scores
+    assert store.selector_scores == []