fix: isolate selector VLM exceptions at slot level (no batch crash)

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)

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 == []