fix(question_gen): resolve pipeline integration issues from final review

1. Apply postprocess shuffle result (pp.options, pp.answer) to final
   GeneratedQuestion output instead of using original candidate values.

2. Record dedup rejection in store via new mark_item_rejected() method,
   preventing items from staying as 'accepted' after dedup rejects them.

3. Add .flatten() to embed_fn outputs in _is_duplicate and embed_pool
   append to handle 2D (1,D) arrays from embedding implementations.

4. Validate exactly 4 options in _validate_parsed_fields (was >= 2),
   matching the A-D answer constraint.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

app/question_gen/generator_v2.py

@@ -252,9 +252,9 @@ def _validate_parsed_fields(data: dict) -> _ValidatedFields:
     answer = str(data["answer"]).strip().upper()
     difficulty = str(data["difficulty"]).strip().lower()
 
-    # 校验 options
+    # 校验 options（answer 约束为 A-D，因此必须恰好 4 个选项）
-    if not isinstance(options_raw, list) or len(options_raw) < 2:
+    if not isinstance(options_raw, list) or len(options_raw) != 4:
-        msg = f"options 字段必须是至少 2 个选项的列表，实际: {options_raw}"
+        msg = f"options 字段必须恰好包含 4 个选项，实际数量: {len(options_raw) if isinstance(options_raw, list) else type(options_raw).__name__}"
         raise ValueError(msg)
 
     # 校验 answer

app/question_gen/pipeline_v2.py

@@ -245,7 +245,7 @@ def _is_duplicate(
     if not embed_pool:
         return False
 
-    query_vec = embed_fn(question_text)
+    query_vec = embed_fn(question_text).flatten()
     query_norm = np.linalg.norm(query_vec)
     if query_norm == 0:
         return False
@@ -266,11 +266,18 @@ def _is_duplicate(
 # ---------------------------------------------------------------------------
 
 
-def _to_generated_question(candidate: CandidateQuestion) -> GeneratedQuestion:
+def _to_generated_question(
+    candidate: CandidateQuestion,
+    *,
+    options: tuple[str, ...] | None = None,
+    answer: str | None = None,
+) -> GeneratedQuestion:
     """将 CandidateQuestion 转换为 GeneratedQuestion。
 
     参数:
         candidate: 门控通过的候选题目。
+        options: 洗牌后的选项元组（若为 None 则使用 candidate 原始选项）。
+        answer: 重映射后的答案字母（若为 None 则使用 candidate 原始答案）。
 
     返回:
         GeneratedQuestion 实例（difficulty_steps 初始为 None）。
@@ -280,8 +287,8 @@ def _to_generated_question(candidate: CandidateQuestion) -> GeneratedQuestion:
         video_id=candidate.video_id,
         task_type=candidate.task_type,
         question=candidate.question,
-        options=candidate.options,
+        options=options if options is not None else candidate.options,
-        answer=candidate.answer,
+        answer=answer if answer is not None else candidate.answer,
         source_nodes=candidate.source_nodes,
         difficulty=candidate.difficulty,
         skill_target=candidate.skill_target,
@@ -464,6 +471,7 @@ async def _process_one_slot(
             # Phase 7: 去重检测
             if _is_duplicate(candidate.question, embed_pool, embed_fn, config.dedup_threshold):
                 prev_reason = "duplicate detected by embedding similarity"
+                store.mark_item_rejected(item_id, prev_reason)
                 logger.info(
                     "slot {} 重复题被拒绝 (attempt {}/{})",
                     slot.slot_id,
@@ -472,10 +480,10 @@ async def _process_one_slot(
                 )
                 continue
 
-            # Phase 8: 通过全部检查 → 接受
+            # Phase 8: 通过全部检查 → 接受（使用洗牌后的选项和答案）
-            result = _to_generated_question(candidate)
+            result = _to_generated_question(candidate, options=pp.options, answer=pp.answer)
-            # 将题目 embedding 加入池
+            # 将题目 embedding 加入池（flatten 确保 1D）
-            embed_pool.append(embed_fn(candidate.question))
+            embed_pool.append(embed_fn(candidate.question).flatten())
             # 标记使用的节点
             used_node_ids.update(candidate.source_nodes)
 

app/question_gen/run_store.py

@@ -330,6 +330,25 @@ class QuestionGenStore:
         if cursor.rowcount == 0:
             raise ValueError(f"item_id 不存在: {item_id}")
 
+    def mark_item_rejected(self, item_id: str, reason: str) -> None:
+        """将已记录的 item 标记为 rejected（用于门控外的拒绝场景，如去重）。
+
+        Parameters
+        ----------
+        item_id : str
+            题目唯一 ID。
+        reason : str
+            拒绝原因描述。
+        """
+        cursor = self._conn.execute(
+            "UPDATE question_gen_items SET final_status='rejected', gate_reject_reason=? "
+            "WHERE item_id=?",
+            (reason, item_id),
+        )
+        self._conn.commit()
+        if cursor.rowcount == 0:
+            raise ValueError(f"item_id 不存在: {item_id}")
+
     def update_difficulty(self, item_id: str, difficulty_steps: int) -> None:
         """更新重量抽检产出的 Agent 步数。
 

tests/unit/test_run_store.py

@@ -247,9 +247,7 @@ class TestQuestionGenStore:
         """对不存在的 run_id 调用 record_run_end 应报错。"""
         stats = RunStats(total_slots=10, accepted=5, rejected=3, heavy_sampled=2)
         with pytest.raises(ValueError, match="run_id"):
-            store.record_run_end(
+            store.record_run_end(run_id="ghost-run", status="completed", stats=stats)
-                run_id="ghost-run", status="completed", stats=stats
-            )
 
     def test_update_gates_missing_item_raises(self, store: QuestionGenStore) -> None:
         """对不存在的 item_id 调用 update_gates 应报错。"""
@@ -262,6 +260,53 @@ class TestQuestionGenStore:
         with pytest.raises(ValueError, match="item_id"):
             store.update_gates(item_id="ghost-item", report=report)
 
+    def test_mark_item_rejected(self, store: QuestionGenStore) -> None:
+        """mark_item_rejected 将 final_status 设为 rejected 并记录原因。"""
+        run_id = "run-mark-rej"
+        store.record_run_start(run_id=run_id, git_sha="ddd444", config_snapshot="{}")
+        item_id = "item-mark-rej-1"
+        store.record_item(
+            item_id=item_id,
+            run_id=run_id,
+            slot_id="slot-mark",
+            video_id="v005",
+            family="retrieval",
+            task_type="Action Recognition",
+            skill_target="M1",
+            attempt=1,
+            question_text="这是什么？",
+        )
+
+        # 先模拟门控通过（将 final_status 设为 accepted）
+        report_pass = _MockGateReport(
+            key_verify=_GateResult(_Verdict.PASS, "ok"),
+            blind_answer=_GateResult(_Verdict.PASS, "ok"),
+            multi_true=_GateResult(_Verdict.PASS, "ok"),
+            leak_test=_GateResult(_Verdict.PASS, "ok"),
+        )
+        store.update_gates(item_id=item_id, report=report_pass)
+
+        # 然后因去重被拒绝
+        store.mark_item_rejected(item_id, "duplicate detected by embedding similarity")
+
+        import sqlite3
+
+        conn = sqlite3.connect(str(store._db_path))
+        conn.row_factory = sqlite3.Row
+        row = conn.execute(
+            "SELECT final_status, gate_reject_reason FROM question_gen_items WHERE item_id=?",
+            (item_id,),
+        ).fetchone()
+        conn.close()
+
+        assert row["final_status"] == "rejected"
+        assert row["gate_reject_reason"] == "duplicate detected by embedding similarity"
+
+    def test_mark_item_rejected_missing_item_raises(self, store: QuestionGenStore) -> None:
+        """对不存在的 item_id 调用 mark_item_rejected 应报错。"""
+        with pytest.raises(ValueError, match="item_id"):
+            store.mark_item_rejected(item_id="ghost-item", reason="duplicate")
+
     def test_update_difficulty_missing_item_raises(self, store: QuestionGenStore) -> None:
         """对不存在的 item_id 调用 update_difficulty 应报错。"""
         with pytest.raises(ValueError, match="item_id"):