diff --git a/app/question_gen/pipeline_v2.py b/app/question_gen/pipeline_v2.py index 7ff2669..4ff006a 100644 --- a/app/question_gen/pipeline_v2.py +++ b/app/question_gen/pipeline_v2.py @@ -404,9 +404,12 @@ async def _apply_grounded_selector( config=selector_cfg, session_id=session_id, ) - except (ValueError, FileNotFoundError) as e: - # 异常路径无 outcome/observation 可落,但仍须 mark_item_rejected, - # 与 hard-fail 路径落库风格一致(该 attempt 的 item 是死记录,重出新建 item_id) + except (ValueError, FileNotFoundError, OSError, Exception) as e: # noqa: BLE001 + # slot 级隔离:selector 会多次调 VLM,任何异常(含 VLM 网络/超时/熔断 + # CircuitOpenError/StreamLivenessTimeout 等非 ValueError 类型)都降级为 + # selector_error 重出该 slot,绝不穿透 asyncio.gather 崩掉整批生成。 + # 与相邻 Phase 2 生成失败的宽捕获口径一致;异常路径无 observation 可落, + # 但仍 mark_item_rejected(该 attempt 的 item 是死记录,重出新建 item_id)。 reason = f"selector_error: {e}" logger.warning("slot {} selector 异常 (attempt {}): {}", slot_id, attempt, e) store.mark_item_rejected(item_id, reason) diff --git a/tests/unit/test_pipeline_selector_wiring.py b/tests/unit/test_pipeline_selector_wiring.py index e7e2846..2ab6e2e 100644 --- a/tests/unit/test_pipeline_selector_wiring.py +++ b/tests/unit/test_pipeline_selector_wiring.py @@ -120,3 +120,67 @@ async def test_apply_grounded_selector_marks_rejected_on_error(monkeypatch): assert store.rejected == [("item-1", reason)] # 异常路径无 observation,故不落 selector_scores assert store.selector_scores == [] + + +@pytest.mark.parametrize( + "exc", + [ + RuntimeError("vlm 500"), + TimeoutError("watchdog total timeout"), + ConnectionError("network reset"), + ], +) +async def test_apply_grounded_selector_isolates_vlm_runtime_exceptions(monkeypatch, exc): + """C3: selector 内 VLM 网络/超时/熔断(非 ValueError)异常须被隔离为 selector_error 重出, + + 不得穿透 _process_one_slot → asyncio.gather 崩掉整批生成。 + """ + 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 exc + + 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)] + assert store.selector_scores == []