fix: harden canonical_answer_text and align decision-core signatures to spec

app/question_gen/adversarial_filter.py

@@ -12,10 +12,8 @@ from __future__ import annotations
 import enum
 import hashlib
 import json
-from typing import TYPE_CHECKING
 
-if TYPE_CHECKING:
-    from core.types import GeneratedQuestion
+_VALID_LETTERS = ("A", "B", "C", "D")
 
 
 class FlipDecision(enum.Enum):
@@ -26,20 +24,22 @@ class FlipDecision(enum.Enum):
     FLIP_SKIPPED = "flip_skipped"
 
 
-def question_hash(question: GeneratedQuestion) -> str:
+def question_hash(question: str, options: tuple[str, ...], answer: str) -> str:
     """题 payload（question+options+answer）的稳定 hash，防 JSON 变动误用旧 verdict。
 
     参数:
-        question: 题目。
+        question: 题目文本。
+        options: 选项元组。
+        answer: 正确答案字母。
 
     返回:
         16 位十六进制摘要。
     """
     payload = json.dumps(
         {
-            "question": question.question,
-            "options": list(question.options),
-            "answer": question.answer,
+            "question": question,
+            "options": list(options),
+            "answer": answer,
         },
         ensure_ascii=False,
         sort_keys=True,
@@ -53,25 +53,29 @@ def agent_config_fingerprint(*, skill_mode: str, max_steps: int, model: str) ->
     return hashlib.sha256(raw.encode("utf-8")).hexdigest()[:16]
 
 
-def canonical_answer_text(question: GeneratedQuestion, letter: str | None) -> str | None:
+def canonical_answer_text(options: tuple[str, ...], letter: str | None) -> str | None:
     """把 agent 预测的选项字母映射为选项规范化文本；非法/越界返回 None。
 
-    镜像题选项会重洗牌，字母无语义，必须按选项文本比较。
+    镜像题选项会重洗牌，字母无语义，必须按选项文本比较。防御性：空串、多字符、
+    非 A-D 字母、越界一律返回 None，下游据此保守 flip_skipped，绝不崩溃、绝不误杀。
 
     参数:
-        question: 题目（提供 options）。
-        letter: agent 预测字母（大小写不敏感），None/空/越界视为无效。
+        options: 选项元组。
+        letter: agent 预测字母（大小写不敏感，须为单个 A-D），其余一律无效。
 
     返回:
         去掉 "X. " 前缀的选项文本；无效时 None。
     """
-    if not letter or not isinstance(letter, str):
+    if not isinstance(letter, str):
         return None
-    idx = ord(letter.strip().upper()) - ord("A")
-    if not 0 <= idx < len(question.options):
+    s = letter.strip().upper()
+    if s not in _VALID_LETTERS:
         return None
-    opt = question.options[idx]
-    prefix = f"{letter.strip().upper()}. "
+    idx = ord(s) - ord("A")
+    if idx >= len(options):
+        return None
+    opt = options[idx]
+    prefix = f"{s}. "
     return opt[len(prefix):] if opt.startswith(prefix) else opt
 
 

tests/unit/test_adversarial_filter_core.py

@@ -7,26 +7,20 @@ from app.question_gen.adversarial_filter import (
     judge_flip,
     question_hash,
 )
-from core.types import GeneratedQuestion
 
-
-def _q(qid="q1", options=("A. 蒸", "B. 炒", "C. 煮", "D. 炸"), answer="A"):
-    return GeneratedQuestion(
-        question_id=qid, video_id="v1", task_type="Action Recognition",
-        question="?", options=options, answer=answer,
-        source_nodes=("n1",), difficulty="hard",
-        sub_pattern="temporal_reasoning_failure",
-    )
+_OPTIONS = ("A. 蒸", "B. 炒", "C. 煮", "D. 炸")
 
 
 def test_question_hash_stable_and_payload_sensitive():
-    h1 = question_hash(_q())
-    h2 = question_hash(_q())
+    h1 = question_hash("?", _OPTIONS, "A")
+    h2 = question_hash("?", _OPTIONS, "A")
     assert h1 == h2
-    h3 = question_hash(_q(answer="B"))  # answer 变 → hash 变
+    h3 = question_hash("?", _OPTIONS, "B")  # answer 变 → hash 变
     assert h1 != h3
-    h4 = question_hash(_q(options=("A. 蒸", "B. 炒", "C. 煮", "D. 烤")))  # option 变 → 变
+    h4 = question_hash("?", ("A. 蒸", "B. 炒", "C. 煮", "D. 烤"), "A")  # option 变 → 变
     assert h1 != h4
+    h5 = question_hash("!", _OPTIONS, "A")  # question 变 → 变
+    assert h1 != h5
 
 
 def test_agent_config_fingerprint_changes_with_inputs():
@@ -37,14 +31,20 @@ def test_agent_config_fingerprint_changes_with_inputs():
 
 
 def test_canonical_answer_text_maps_letter_to_option_text():
-    assert canonical_answer_text(_q(), "C") == "煮"
-    assert canonical_answer_text(_q(), "c") == "煮"
+    assert canonical_answer_text(_OPTIONS, "C") == "煮"
+    assert canonical_answer_text(_OPTIONS, "c") == "煮"
 
 
 def test_canonical_answer_text_invalid_returns_none():
-    assert canonical_answer_text(_q(), "Z") is None
-    assert canonical_answer_text(_q(), "") is None
-    assert canonical_answer_text(_q(), None) is None
+    assert canonical_answer_text(_OPTIONS, "Z") is None
+    assert canonical_answer_text(_OPTIONS, "") is None
+    assert canonical_answer_text(_OPTIONS, None) is None
+
+
+def test_canonical_answer_text_invalid_inputs_return_none():
+    for bad in ["", "AB", "E", "1", " ", "AA"]:
+        assert canonical_answer_text(_OPTIONS, bad) is None, bad
+    assert canonical_answer_text(_OPTIONS, "b") == "炒"  # 合法仍工作（去前缀、大小写不敏感）
 
 
 def test_judge_flip_different_answers_passed():