diff --git a/GROUPBRIEF_L3_DIAGNOSTIC_FALLBACK_GUARD_TASK.md b/GROUPBRIEF_L3_DIAGNOSTIC_FALLBACK_GUARD_TASK.md
new file mode 100644
index 0000000..32a3395
--- /dev/null
+++ b/GROUPBRIEF_L3_DIAGNOSTIC_FALLBACK_GUARD_TASK.md
@@ -0,0 +1,62 @@
+# 茶馆 Level 3 诊断图禁发任务
+
+## 背景
+
+茶馆日报在 Prompt 连续校验失败后生成了 Pillow Level 3 本地信息图。该文件本应仅供诊断，但旧流程把它记录为生图成功并推进到自动发送，最终在 2026-09-03 08:36 发送。Dashboard 同时以 `object-fit: cover` 裁切了竖版预览。
+
+## 目标
+
+- 保留 PR #16 已有的按真实 `sender_id` 去重、参与者补足和越权消息 ID 校验。
+- 将 Level 3/Pillow 兜底定义为“已保存诊断产物的图片生成失败”，不得推进到发送就绪。
+- 在发送扫描、原子 claim、发送前预检三层拒绝诊断图，统一错误码 `IMAGE_FALLBACK_NOT_SENDABLE`。
+- 让 Dashboard 同时保留历史发送事实并明确展示图片失败、诊断图不可发送和完整竖图预览。
+
+## 允许修改范围
+
+- `app/image/`、`app/pipeline/`、`app/v2/`、`app/scheduler/` 中与图片状态和发送门禁直接相关的代码。
+- `app/api/v2_ui_read.py` 的 Dashboard 只读投影。
+- `frontend/src/api.ts`、`frontend/src/pages/v2/Dashboard.tsx`、`frontend/src/styles.css`。
+- 与上述行为直接相关的后端测试、前端单测和 Playwright 测试。
+- 本任务说明文件与 PR #16 描述。
+
+## 禁止修改范围
+
+- 不修改数据库结构或执行迁移。
+- 不读取或修改 `.env`、认证、Provider、登录方式及任何密钥。
+- 不修改既有 `output/`、`run.json`、PNG、发送记录或生产日志。
+- 不调用生成、恢复、发送、重发、发布、部署或服务重启接口。
+- 不改动主工作树的 `.codemap` 和复检报告。
+
+## 已确定实现要求
+
+1. Level 3/Pillow 允许落盘诊断 PNG、失败原因及 SHA-256，但任务返回失败并停在图片阶段 `FAILED`。
+2. 自动重试仍受现有预算限制；诊断 PNG 不得被“已有有效图片”逻辑误判为正常成功。后续真实图片成功时清除兜底元数据并恢复正常流程。
+3. 发送扫描、claim 和发送前预检均拒绝 `image_fallback_level >= 3` 或 `image_variant=pillow`，且不得调用文字或图片发送器。
+4. Dashboard API 返回 `image_status`、`image_fallback_level`、`image_fallback_reason`、`image_variant`、`image_delivery_eligible`。
+5. 历史 `SENT + Level 3` 保留“已发送”，同时将图片节点投影为失败，卡片显示“图片生成失败（已发送）”和“诊断图不可发送”，且不出现重发入口。
+6. Dashboard 图片预览使用完整适配；普通有效图片和关闭生图的流程保持不变。
+
+## 验收标准
+
+- PR #16 的人物去重/补足/消息归属测试继续通过。
+- Prompt、Provider、事实校验失败生成诊断图后，状态为图片失败而非待发送。
+- 构造遗留 `READY_TO_SEND + Level 3/Pillow` 时，扫描、claim、发送前预检均返回统一错误码，发送器调用数为零。
+- 真实图片成功、关闭生图、诊断后重试成功等流程通过。
+- 历史 `SENT + Level 3` 的 API 与 UI 同时呈现发送事实和图片失败，无发送按钮。
+- 后端相关测试和全量测试、Python 编译检查、前端单测、Playwright、前端构建及 `git diff --check` 全部通过。
+
+## 测试命令
+
+```powershell
+python -m pytest tests/test_v2_prompt_builder.py tests/test_v2_image_task.py tests/test_v2_pipeline.py tests/test_runtime_status.py tests/test_v2_ui_router_contract.py -q
+python -m pytest tests -q
+python -m compileall -q app scripts tests
+npm test
+npm run test:e2e
+npm run build
+git diff --check
+```
+
+## 返回格式
+
+返回修改摘要、根因闭环、测试命令与结果、实际 diff 范围、提交 SHA、远端 SHA、PR #16 地址与 CI 状态；明确说明未重发、未重启、未合并。
diff --git a/app/ai/layouts.py b/app/ai/layouts.py
index 2c1a76a..0ae57b6 100644
--- a/app/ai/layouts.py
+++ b/app/ai/layouts.py
@@ -228,6 +228,9 @@ def to_meta(self) -> dict[str, Any]:
 dialogue、reaction、close_up、punchline、insert。全部话题必须恰好出现一次于 panel_beats，
 topic_order 也必须恰好覆盖全部话题。
 
+只有 1 个入选话题时必须使用 hero_rhythm，并把该话题作为唯一 featured_topic_id；
+不得为了满足双焦点结构复制或虚构第二个话题。
+
 根据内容选择 hero_rhythm、dual_rhythm 或 ensemble_rhythm。无论哪种模式，都必须保留明显的
 大/中/小格层级，禁止等宽等高列表。指定风格只控制配色、画材、造型、纹理和光影；版式不能改写画风。
 优先选择最符合笑点节奏的分镜骨架；不得选择前一次使用的骨架，并尽量避开最近三次。"""
@@ -334,8 +337,8 @@ def build_layout_director_prompt(
 
 def _expected_topic_ids(topic_ids: Iterable[str]) -> list[str]:
     expected = [str(topic_id) for topic_id in topic_ids if str(topic_id)]
-    if not (2 <= len(expected) <= 7) or len(set(expected)) != len(expected):
-        raise LayoutPlanError("入选主题必须是 2～7 个不重复 ID")
+    if not (1 <= len(expected) <= 7) or len(set(expected)) != len(expected):
+        raise LayoutPlanError("入选主题必须是 1～7 个不重复 ID")
     return expected
 
 
diff --git a/app/ai/poster_copy.py b/app/ai/poster_copy.py
index 4a4a385..0bbb2f4 100644
--- a/app/ai/poster_copy.py
+++ b/app/ai/poster_copy.py
@@ -106,6 +106,16 @@ def _selected_topics(selection: dict, topic_order: Iterable[str]) -> list[dict[s
     return ordered
 
 
+def _participant_identity(sender_id: object, speaker: object) -> tuple[str, str]:
+    normalized_sender_id = str(sender_id or "").strip().casefold()
+    normalized_speaker = str(speaker or "").strip()
+    return (
+        ("id", normalized_sender_id)
+        if normalized_sender_id
+        else ("name", normalized_speaker)
+    )
+
+
 def build_poster_editor_source(
     selection: dict,
     layout: LayoutPlan,
@@ -142,6 +152,12 @@ def build_poster_editor_source(
             name = entry["speaker"]
             if name not in participants:
                 participants.append(name)
+        available_identities = {
+            _participant_identity(entry["sender_id"], entry["speaker"])
+            for entry in evidence_dialogue
+        }
+        participant_min = 2 if len(available_identities) >= 2 else 1
+        participant_max = min(MAX_VISIBLE_PARTICIPANTS, len(available_identities))
         beat = beats.get(topic_id)
         topics.append(
             {
@@ -150,6 +166,8 @@ def build_poster_editor_source(
                 "source_summary": str(item.get("summary") or "").strip(),
                 "source_visual_gag": str(item.get("visual_gag") or "").strip(),
                 "participant_options": participants,
+                "participant_min": participant_min,
+                "participant_max": participant_max,
                 "evidence_dialogue": evidence_dialogue,
                 "speaker_bindings": evidence_dialogue,
                 "shot_hints": [
@@ -173,7 +191,8 @@ def build_poster_editor_source(
 POSTER_EDITOR_SYSTEM = """你是「群报 GroupBrief」的漫画日报内容编辑。
 你只能根据给定的 evidence package 写一个 JSON 对象，不得输出 Markdown 或解释。
 所有事实、姓名和对白必须来自对应 topic；不得改变金额、时间、地点和人物关系。
-每个 panel 必须对应一个 topic_id，顺序不得改变。优先使用 2～4 位 participant_options；
+每个 panel 必须对应一个 topic_id，顺序不得改变。participants 数量必须严格落在对应 topic 的
+participant_min～participant_max 范围内；同一 sender_id 只能出现一次；
 每位人物都要从 speaker_bindings 选择一个真实 message_id，并写可绘制的动作、站位或反应。
 不得输出或改写人物姓名；程序会从 message_id 绑定中填入姓名。quote 只能逐字复制同一
 message_id 的完整原消息或其中连续、语义完整的片段，不得跨消息拼接，不得改写，
@@ -334,7 +353,12 @@ def _assert_text_grounded(text: str, evidence: str, label: str) -> None:
         raise PosterCopyError(f"{label}没有回收到对应真实话题")
 
 
-def parse_poster_copy(raw: str, source: dict[str, Any]) -> DailyPosterCopy:
+def parse_poster_copy(
+    raw: str,
+    source: dict[str, Any],
+    *,
+    repair_log: list[dict[str, Any]] | None = None,
+) -> DailyPosterCopy:
     try:
         payload = json.loads(_strip_json_fence(raw))
     except json.JSONDecodeError as exc:
@@ -390,9 +414,7 @@ def parse_poster_copy(raw: str, source: dict[str, Any]) -> DailyPosterCopy:
         participant_options = [str(value).strip() for value in topic.get("participant_options") or [] if str(value).strip()]
         bindings = _bindings_by_message_id(topic)
         available_identities = {
-            ("id", binding["sender_id"].casefold())
-            if binding["sender_id"]
-            else ("name", binding["speaker"])
+            _participant_identity(binding["sender_id"], binding["speaker"])
             for binding in bindings.values()
         }
         raw_participants = raw_panel.get("participants")
@@ -400,12 +422,13 @@ def parse_poster_copy(raw: str, source: dict[str, Any]) -> DailyPosterCopy:
             raise PosterCopyError(f"版面{index}缺少 participants")
         minimum = 2 if len(available_identities) >= 2 else 1
         maximum = min(MAX_VISIBLE_PARTICIPANTS, len(available_identities))
-        if not bindings or not (minimum <= len(raw_participants) <= maximum):
+        if not bindings:
             raise PosterCopyError(f"版面{index}应使用 {minimum}～{maximum} 位真实参与者")
 
-        participants: list[ParticipantCopy] = []
-        identities: set[tuple[str, str]] = set()
-        quoted_speakers: set[tuple[str, str]] = set()
+        participants_by_identity: dict[tuple[str, str], ParticipantCopy] = {}
+        identity_order: list[tuple[str, str]] = []
+        duplicate_count = 0
+        trimmed_count = 0
         for person_index, raw_person in enumerate(raw_participants, start=1):
             if not isinstance(raw_person, dict):
                 raise PosterCopyError(f"版面{index}第{person_index}位人物格式无效")
@@ -417,10 +440,9 @@ def parse_poster_copy(raw: str, source: dict[str, Any]) -> DailyPosterCopy:
                 raise PosterCopyError(f"版面{index}包含未授权的消息ID：{message_id or '空'}")
             name = binding["speaker"]
             sender_id = binding["sender_id"]
-            identity = ("id", sender_id.casefold()) if sender_id else ("name", name)
-            if name not in participant_options or identity in identities:
-                raise PosterCopyError(f"版面{index}包含未授权或重复的群友身份：{name}")
-            identities.add(identity)
+            identity = _participant_identity(sender_id, name)
+            if name not in participant_options:
+                raise PosterCopyError(f"版面{index}包含未授权的群友身份：{name}")
             action = _clean_text(raw_person.get("action"), maximum=120, label=f"版面{index}人物动作")
             quote = re.sub(r"\s+", " ", str(raw_person.get("quote") or "")).strip().strip("“”\"")
             if quote:
@@ -442,21 +464,69 @@ def parse_poster_copy(raw: str, source: dict[str, Any]) -> DailyPosterCopy:
                         source_messages,
                     ),
                 )
-                quoted_speakers.add(identity)
-            participants.append(
-                ParticipantCopy(
-                    message_id=message_id,
-                    sender_id=sender_id,
+            participant = ParticipantCopy(
+                message_id=message_id,
+                sender_id=sender_id,
+                name=name,
+                action=action,
+                quote=quote,
+            )
+            if identity in participants_by_identity:
+                duplicate_count += 1
+                if not participants_by_identity[identity].quote and participant.quote:
+                    participants_by_identity[identity] = participant
+                continue
+            if len(identity_order) >= maximum:
+                trimmed_count += 1
+                continue
+            identity_order.append(identity)
+            participants_by_identity[identity] = participant
+
+        model_identity_count = len(identity_order)
+        filled_count = 0
+        if len(identity_order) < minimum:
+            for binding in bindings.values():
+                name = binding["speaker"]
+                identity = _participant_identity(binding["sender_id"], name)
+                if identity in participants_by_identity or name not in participant_options:
+                    continue
+                identity_order.append(identity)
+                participants_by_identity[identity] = ParticipantCopy(
+                    message_id=binding["message_id"],
+                    sender_id=binding["sender_id"],
                     name=name,
-                    action=action,
-                    quote=quote,
+                    action="参与本话题讨论",
+                    quote="",
                 )
-            )
+                filled_count += 1
+                if len(identity_order) >= minimum:
+                    break
+        if not (minimum <= len(identity_order) <= maximum):
+            raise PosterCopyError(f"版面{index}应使用 {minimum}～{maximum} 位真实参与者")
 
-        required_quotes = 2 if len(available_identities) >= 2 and len(participants) >= 2 else 1
+        participants = [participants_by_identity[identity] for identity in identity_order]
+        quoted_speakers = {
+            _participant_identity(participant.sender_id, participant.name)
+            for participant in participants
+            if participant.quote
+        }
+        # 自动补齐的人物对白保持为空，不能伪造原话；模型至少仍须提供一位真实气泡。
+        required_quotes = 2 if model_identity_count >= 2 else 1
         if len(quoted_speakers) < required_quotes:
             raise PosterCopyError(f"版面{index}缺少足够的真实多人对白")
 
+        if repair_log is not None and (duplicate_count or trimmed_count or filled_count):
+            repair_log.append(
+                {
+                    "panel_index": index,
+                    "topic_id": topic_id,
+                    "deduplicated_count": duplicate_count,
+                    "trimmed_count": trimmed_count,
+                    "filled_count": filled_count,
+                    "final_participant_count": len(participants),
+                }
+            )
+
         panels.append(
             PanelCopy(
                 topic_id=topic_id,
@@ -501,7 +571,7 @@ def _overall_visual(style_text: str, *, explicit_style: bool) -> str:
             "生成一张适合微信手机端阅读的竖版漫画群报，优先采用 1024×1536、2:3 画布；其他完整可读的竖版尺寸也可以采用，不要为了匹配尺寸裁切或拉伸。",
             style_line,
             "整张图像一页热闹的群聊漫画：顶部是群名称、完整统计时间、主标题和副标题，中间由多个大小错落的话题漫画格组成，底部展示当天总结和统计数据。",
-            "每个话题都要画成一个真实的“群友讨论现场”，而不是单人物插画。每个话题优先选择 2～4 位真正参与该段聊天的群友出镜，人物旁边直接标注对应的真实群昵称。",
+            "每个话题都要画成一个真实的“群友讨论现场”。严格按对应版面已经列出的真实参与者出镜，不重复人物、不补无关群友；有多位真实参与者时通常展示 2～4 位，只有 1 位时就只画该人物。人物旁边直接标注对应的真实群昵称。",
             "不同群友使用不同动作、表情和站位，以真实聊天气泡、人物反应、道具、动作线和视觉笑点表现讨论过程。所有剧情、人物关系和聊天内容均来自当天真实群聊，不额外编造新的聊天事实。",
         )
     )
@@ -621,8 +691,8 @@ def validate_fixed_prompt_contract(
     headings = _section_headings(prompt)
     panel_headings = [heading for heading in headings if re.fullmatch(r"版面\d+", heading)]
     panel_count = len(panel_headings)
-    if not 2 <= panel_count <= 7:
-        raise PosterCopyError("最终 Prompt 必须包含连续的 2～7 个版面")
+    if not 1 <= panel_count <= 7:
+        raise PosterCopyError("最终 Prompt 必须包含连续的 1～7 个版面")
     if expected_panel_count is not None and panel_count != expected_panel_count:
         raise PosterCopyError("版面数量与本次已校验入选话题数量不一致")
     expected = [
diff --git a/app/ai/prompt_builder.py b/app/ai/prompt_builder.py
index af55bca..aa7ab29 100644
--- a/app/ai/prompt_builder.py
+++ b/app/ai/prompt_builder.py
@@ -19,6 +19,7 @@
 from copy import deepcopy
 from time import perf_counter
 from datetime import datetime
+from typing import Any
 
 from app.ai.prompt_templates import (
     ImagePromptTemplateError,
@@ -396,6 +397,10 @@ def analyze(item: tuple[int, ConversationChunk]) -> tuple[list[dict], int]:
                 selection["speaker_fingerprint"] = speaker_fingerprint
             meta["topic_selection_version"] = selection["topic_selection_version"]
             meta["topic_selection"] = selection
+            if selection.get("political_keyword_policy_version"):
+                meta["political_keyword_policy_version"] = selection[
+                    "political_keyword_policy_version"
+                ]
             selected_payload = selected_topics_json(selection)
             topic_ids = selected_topic_ids(selection)
             recent_history = tuple(data.recent_layout_history or ())[:3]
@@ -456,6 +461,7 @@ def analyze(item: tuple[int, ConversationChunk]) -> tuple[list[dict], int]:
             text = ""
             final_calls = 0
             last_violations: list[str] = []
+            participant_repairs: list[dict[str, Any]] = []
             for attempt in range(FINAL_PROMPT_MAX_ATTEMPTS):
                 prompt = final_user_prompt
                 if attempt:
@@ -473,8 +479,13 @@ def analyze(item: tuple[int, ConversationChunk]) -> tuple[list[dict], int]:
                     max_tokens=6000,
                 )
                 final_calls += 1
+                attempt_repairs: list[dict[str, Any]] = []
                 try:
-                    copy = parse_poster_copy(raw_copy, editor_source)
+                    copy = parse_poster_copy(
+                        raw_copy,
+                        editor_source,
+                        repair_log=attempt_repairs,
+                    )
                     candidate_text = render_poster_prompt(
                         copy,
                         group_name=visible_group_name,
@@ -495,11 +506,13 @@ def analyze(item: tuple[int, ConversationChunk]) -> tuple[list[dict], int]:
                     )
                     continue
                 text = candidate_text
+                participant_repairs = attempt_repairs
                 meta["poster_copy_version"] = POSTER_COPY_VERSION
                 meta["poster_topic_count"] = len(copy.panels)
                 meta["poster_visible_participant_count"] = sum(
                     len(panel.participants) for panel in copy.panels
                 )
+                meta["poster_participant_repairs"] = participant_repairs
                 break
             if not text:
                 raise ValueError("最终生图 Prompt 未通过固定漫画合同：" + "；".join(last_violations[:8]))
diff --git a/app/ai/topic_selection.py b/app/ai/topic_selection.py
index 28cb0f4..ea6881c 100644
--- a/app/ai/topic_selection.py
+++ b/app/ai/topic_selection.py
@@ -1,4 +1,4 @@
-"""有证据、喜剧优先的日报候选主题评分与 5～7 个高密度动态选题。"""
+"""有证据、喜剧优先的日报候选主题评分与 1～7 个高密度动态选题。"""
 
 from __future__ import annotations
 
@@ -14,9 +14,9 @@
 
 from app.ai.conversation_segments import ConversationChunk, PromptMessage
 
-TOPIC_SELECTION_VERSION = "6.0"
+TOPIC_SELECTION_VERSION = "7.0"
 MAX_CANDIDATES = 10
-MIN_SELECTED = 2
+MIN_SELECTED = 1
 TARGET_SELECTED = 5
 MAX_SELECTED = 7
 HIGH_VOLUME_MESSAGE_THRESHOLD = 200
@@ -26,6 +26,92 @@
 VISIBLE_PARTICIPANT_CHAR_BUDGET = 48
 UNRESOLVED_PARTICIPANT_LABEL = "群友（昵称未识别）"
 
+POLITICAL_KEYWORD_POLICY_VERSION = "political-keywords-v1"
+# 仅包含明确政治名词，不维护人物姓名，也不使用容易误伤日常聊天的宽泛词。
+POLITICAL_TOPIC_KEYWORDS: tuple[str, ...] = (
+    "地缘政治",
+    "外交制裁",
+    "主权争议",
+    "人大常委会",
+    "全国人大",
+    "人大代表",
+    "全国政协",
+    "政协委员",
+    "国家主席",
+    "副总统",
+    "执政党",
+    "在野党",
+    "党代会",
+    "总书记",
+    "参议院",
+    "众议院",
+    "政治",
+    "政党",
+    "两会",
+    "选举",
+    "大选",
+    "竞选",
+    "公投",
+    "总统",
+    "总理",
+    "首相",
+    "国会",
+    "议会",
+    "内阁",
+    "弹劾",
+    "政变",
+    "示威",
+    "抗议",
+    "台独",
+    "港独",
+    "藏独",
+    "疆独",
+    "standing committee of the national people's congress",
+    "chinese people's political consultative conference",
+    "national people's congress",
+    "house of representatives",
+    "hong kong independence",
+    "taiwan independence",
+    "tibetan independence",
+    "xinjiang independence",
+    "presidential election",
+    "diplomatic sanctions",
+    "sovereignty dispute",
+    "opposition party",
+    "political party",
+    "political campaign",
+    "party congress",
+    "general election",
+    "general secretary",
+    "vice president",
+    "prime minister",
+    "ruling party",
+    "cppcc member",
+    "npc deputy",
+    "head of state",
+    "military coup",
+    "political protest",
+    "political demonstration",
+    "geopolitics",
+    "geopolitical",
+    "referendum",
+    "impeachment",
+    "parliament",
+    "president",
+    "premier",
+    "election",
+    "campaign",
+    "congress",
+    "senate",
+    "cabinet",
+    "protest",
+    "demonstration",
+    "politics",
+    "political",
+    "cppcc",
+    "coup",
+)
+
 SCORE_WEIGHTS = {
     "comedy": 40.0,
     "group_recognition": 20.0,
@@ -344,6 +430,45 @@ def _log_normalized(value: float, maximum: float, weight: float) -> float:
     return round(weight * math.log1p(value) / math.log1p(maximum), 1)
 
 
+def _political_keyword_matches(
+    candidate: dict[str, Any], evidence_dialogue: Iterable[dict[str, Any]]
+) -> list[str]:
+    evidence_text = " ".join(
+        str(entry.get("original_text") or entry.get("text") or "")
+        for entry in evidence_dialogue
+        if isinstance(entry, dict)
+    )
+    combined = " ".join(
+        (
+            str(candidate.get("title") or ""),
+            str(candidate.get("summary") or ""),
+            str(candidate.get("visual_gag") or ""),
+            evidence_text,
+        )
+    )
+    normalized = unicodedata.normalize("NFKC", combined).casefold()
+    matches: list[str] = []
+    normalized_matches: list[str] = []
+    for keyword in POLITICAL_TOPIC_KEYWORDS:
+        normalized_keyword = unicodedata.normalize("NFKC", keyword).casefold()
+        if normalized_keyword.isascii():
+            matched = bool(
+                re.search(
+                    rf"(?<![a-z0-9_]){re.escape(normalized_keyword)}(?![a-z0-9_])",
+                    normalized,
+                )
+            )
+        else:
+            matched = normalized_keyword in normalized
+        if not matched or any(
+            normalized_keyword in existing for existing in normalized_matches
+        ):
+            continue
+        matches.append(keyword)
+        normalized_matches.append(normalized_keyword)
+    return matches
+
+
 def score_and_select_topics(
     candidates: list[dict[str, Any]], messages: Iterable[PromptMessage]
 ) -> dict[str, Any]:
@@ -352,7 +477,7 @@ def score_and_select_topics(
     if len(candidates) < MIN_SELECTED:
         raise TopicSelectionError(
             "TOPIC_CANDIDATES_INSUFFICIENT",
-            "无法从真实消息中取得至少两个拥有独立证据的主题",
+            "无法从真实消息中取得至少一个拥有独立证据的主题",
         )
 
     metrics: list[dict[str, Any]] = []
@@ -377,6 +502,9 @@ def score_and_select_topics(
                 f"候选主题“{candidate['title']}”没有可回查的原话证据",
             )
         evidence_dialogue = _evidence_dialogue(items)
+        political_keyword_matches = _political_keyword_matches(
+            candidate, evidence_dialogue
+        )
         dialogue_speaker_count = len(
             {
                 str(entry.get("speaker") or "").strip()
@@ -391,6 +519,11 @@ def score_and_select_topics(
                 "people": attribution["participants"],
                 "quotes": verified_quotes,
                 "evidence_dialogue": evidence_dialogue,
+                "image_eligible": not political_keyword_matches,
+                "political_keyword_matches": political_keyword_matches,
+                "selection_exclusion_reason": (
+                    "POLITICAL_KEYWORD_MATCH" if political_keyword_matches else ""
+                ),
                 "dialogue_speaker_count": dialogue_speaker_count,
                 "evidence_message_count": len(ids),
                 **attribution,
@@ -418,13 +551,30 @@ def score_and_select_topics(
         scores["total"] = round(sum(scores.values()), 1)
         scored.append({**item, "scores": scores})
 
-    scored.sort(key=lambda item: (-item["scores"]["total"], item["start_time"] or "", item["topic_id"]))
+    scored.sort(
+        key=lambda item: (
+            -item["scores"]["total"],
+            item["start_time"] or "",
+            item["topic_id"],
+        )
+    )
+    for rank, item in enumerate(scored, start=1):
+        item["rank"] = rank
+        item["eligible_rank"] = None
+        item["selected"] = False
+
+    eligible_scored = [item for item in scored if item["image_eligible"]]
     selected_ids: list[str] = []
     previous_total: float | None = None
-    guaranteed_selected = min(TARGET_SELECTED, len(scored))
-    for rank, item in enumerate(scored, start=1):
-        selected = rank <= guaranteed_selected
-        if guaranteed_selected < rank <= MAX_SELECTED:
+    guaranteed_selected = min(TARGET_SELECTED, len(eligible_scored))
+    additional_selection_open = True
+    for eligible_rank, item in enumerate(eligible_scored, start=1):
+        item["eligible_rank"] = eligible_rank
+        selected = eligible_rank <= guaranteed_selected
+        if (
+            guaranteed_selected < eligible_rank <= MAX_SELECTED
+            and additional_selection_open
+        ):
             total = item["scores"]["total"]
             gap = (previous_total - total) if previous_total is not None else 0.0
             selected = (
@@ -434,19 +584,14 @@ def score_and_select_topics(
                 and item["dialogue_speaker_count"] >= 2
             )
             if not selected:
-                # 分数降序；第一个不满足后，后续候选也不再入选。
-                for tail in scored[rank - 1 :]:
-                    tail["selected"] = False
-                break
-        item["rank"] = rank
+                # 合格候选按分数降序；第一个不满足后，后续候选也不再入选。
+                additional_selection_open = False
         item["selected"] = selected
         if selected:
             selected_ids.append(item["topic_id"])
         previous_total = item["scores"]["total"]
 
     for rank, item in enumerate(scored, start=1):
-        item.setdefault("rank", rank)
-        item.setdefault("selected", False)
         item.pop("interestingness_score", None)
         item.pop("comedy_score", None)
         item.pop("group_recognition_score", None)
@@ -454,6 +599,7 @@ def score_and_select_topics(
 
     return {
         "topic_selection_version": TOPIC_SELECTION_VERSION,
+        "political_keyword_policy_version": POLITICAL_KEYWORD_POLICY_VERSION,
         "weights": SCORE_WEIGHTS,
         "thresholds": {
             "max_candidates": MAX_CANDIDATES,
@@ -465,6 +611,20 @@ def score_and_select_topics(
             "additional_max_gap": SELECTION_MAX_GAP,
         },
         "candidate_count": len(scored),
+        "safe_candidate_count": len(eligible_scored),
+        "blocked_candidate_count": len(scored) - len(eligible_scored),
+        "blocked_topic_ids": [
+            item["topic_id"] for item in scored if not item["image_eligible"]
+        ],
+        "political_keyword_hits": [
+            {
+                "topic_id": item["topic_id"],
+                "title": item["title"],
+                "matched_keywords": item["political_keyword_matches"],
+            }
+            for item in scored
+            if not item["image_eligible"]
+        ],
         "selected_count": len(selected_ids),
         "selected_topic_ids": selected_ids,
         "candidates": scored,
@@ -473,6 +633,13 @@ def score_and_select_topics(
 
 def selected_topics_json(selection: dict[str, Any]) -> str:
     selected = [item for item in selection.get("candidates", []) if item.get("selected")]
+    if not selected and selection.get("blocked_candidate_count"):
+        raise TopicSelectionError(
+            "TOPIC_CANDIDATES_POLITICAL",
+            "全部候选主题均命中政治关键词，已停止外部生图内容整理",
+        )
     if not (MIN_SELECTED <= len(selected) <= MAX_SELECTED):
-        raise TopicSelectionError("TOPIC_CANDIDATES_INSUFFICIENT", "最终入选主题数量不在 2～7 范围")
+        raise TopicSelectionError(
+            "TOPIC_CANDIDATES_INSUFFICIENT", "最终入选主题数量不在 1～7 范围"
+        )
     return json.dumps({"selected_topics": selected}, ensure_ascii=False, separators=(",", ":"))
diff --git a/app/api/v2_ui_read.py b/app/api/v2_ui_read.py
index 074037c..a9c4fab 100644
--- a/app/api/v2_ui_read.py
+++ b/app/api/v2_ui_read.py
@@ -20,6 +20,7 @@
 )
 from app.config.settings import Settings, get_settings
 from app.db import repository as repo
+from app.image.delivery_guard import image_delivery_eligible, image_fallback_level
 from app.scheduler.period import PeriodResolver
 from app.scheduler.runtime_status import build_daily_status
 from app.services.runtime_logs import read_runtime_logs
@@ -60,6 +61,14 @@ def dashboard(
         runtime_run.setdefault("group_name", name)
         runtime_runs.append(runtime_run)
         status = run.get("status", "PENDING")
+        fallback_level = image_fallback_level(run)
+        image_variant = str(run.get("image_variant") or "normal")
+        image_can_deliver = image_delivery_eligible(run)
+        image_status = (
+            "failed"
+            if not image_can_deliver
+            else str(run.get("image_status") or "")
+        )
         image_path = store.image_path(name, selected_run_date)
         image_url = ""
         if image_path.exists() and Path(image_path).stat().st_size > 0:
@@ -115,11 +124,17 @@ def dashboard(
                 "message_count": run.get("message_count", 0),
                 "speaker_count": run.get("speaker_count", 0),
                 "image_url": image_url,
+                "image_status": image_status,
+                "image_fallback_level": fallback_level,
+                "image_fallback_reason": str(run.get("image_fallback_reason") or ""),
+                "image_variant": image_variant,
+                "image_delivery_eligible": image_can_deliver,
                 "ranking_preview": ranking_preview,
                 "ranking_error": ranking_error,
                 "error": (
                     run.get("error")
                     or run.get("image_error")
+                    or run.get("prompt_original_error")
                     or run.get("send_error")
                     or run.get("error_type")
                     or ""
@@ -155,6 +170,7 @@ def dashboard(
         and not card["sent_at"]
         and card["wechat_send_enabled"]
         and not card["send_hold"]
+        and card["image_delivery_eligible"]
         for card in cards
     ):
         next_send = f"{settings.schedule_send_time}（按群 ID 串行批次）"
diff --git a/app/image/delivery_guard.py b/app/image/delivery_guard.py
new file mode 100644
index 0000000..a47a882
--- /dev/null
+++ b/app/image/delivery_guard.py
@@ -0,0 +1,23 @@
+"""图片产物来源门禁：诊断兜底永远不能进入外部发送。"""
+
+from __future__ import annotations
+
+from typing import Any, Mapping
+
+
+def image_fallback_level(metadata: Mapping[str, Any] | None) -> int:
+    """读取兜底等级；非空脏值按 Level 3 处理，保持 fail-closed。"""
+    metadata = metadata if isinstance(metadata, Mapping) else {}
+    raw_level = metadata.get("image_fallback_level", 0)
+    try:
+        return int(raw_level or 0)
+    except (TypeError, ValueError):
+        return 3 if str(raw_level or "").strip() else 0
+
+
+def image_delivery_eligible(metadata: Mapping[str, Any] | None) -> bool:
+    """只有真实或安全化生成图可以发送；Level 3/Pillow 仅供诊断。"""
+    metadata = metadata if isinstance(metadata, Mapping) else {}
+    fallback_level = image_fallback_level(metadata)
+    image_variant = str(metadata.get("image_variant") or "").strip().lower()
+    return fallback_level < 3 and image_variant != "pillow"
diff --git a/app/image/image_task.py b/app/image/image_task.py
index fffd9f5..52566a3 100644
--- a/app/image/image_task.py
+++ b/app/image/image_task.py
@@ -24,6 +24,7 @@
     IMAGE_FILE_MISSING,
     PROMPT_FAILED,
 )
+from app.image.delivery_guard import image_delivery_eligible
 from app.image.fallback import (
     image_failure_code,
     image_result_is_unknown,
@@ -206,19 +207,36 @@ def _local_fallback(self, failure_class: str) -> dict:
                     self.output_path.unlink()
                 raise ValueError(verification_detail)
             detail["fact_verification"] = verification_detail
+        error_type = (
+            failure_class
+            if failure_class
+            in {
+                PROMPT_FAILED,
+                IMAGE_CONTENT_VERIFICATION_FAILED,
+                IMAGE_FILE_MISSING,
+            }
+            else IMAGE_GENERATION_FAILED
+        )
         return {
             "group_name": self.group_name,
-            "status": "success",
-            "success": True,
-            "detail": f"外部生图失败，已生成本地简化信息图：{self.output_path}",
-            "error_type": "",
+            "status": "diagnostic_fallback",
+            "success": False,
+            "detail": f"图片生成失败，已保留不可发送的本地诊断图：{self.output_path}",
+            "error_type": error_type,
             "generator_detail": detail,
         }
 
     def run(self) -> dict:
         """执行生图并验证落盘。返回结构化结果。"""
-        # 已存在有效图片且非 force：跳过，不重复生成
-        if not self.force:
+        try:
+            run_state = json.loads(
+                (self.output_path.parent / "run.json").read_text(encoding="utf-8")
+            )
+        except (OSError, UnicodeError, json.JSONDecodeError):
+            run_state = {}
+        # 已存在的 Level 3/Pillow 文件虽然是合法 PNG，但只是诊断产物，不能
+        # 被重试流程误认成真实成功。只有来源可发送的旧图才允许跳过。
+        if not self.force and image_delivery_eligible(run_state):
             ok, _ = verify_image_contract(self.prompt_file, self.output_path)
             if ok:
                 return {
@@ -228,12 +246,6 @@ def run(self) -> dict:
                     "detail": "图片已存在，跳过生成",
                     "error_type": "",
                 }
-        try:
-            run_state = json.loads(
-                (self.output_path.parent / "run.json").read_text(encoding="utf-8")
-            )
-        except (OSError, UnicodeError, json.JSONDecodeError):
-            run_state = {}
         if isinstance(run_state, dict) and run_state.get("image_force_local_fallback"):
             try:
                 return self._local_fallback(
diff --git a/app/pipeline/daily_pipeline.py b/app/pipeline/daily_pipeline.py
index 437a29b..3711f2e 100644
--- a/app/pipeline/daily_pipeline.py
+++ b/app/pipeline/daily_pipeline.py
@@ -38,6 +38,7 @@
 from app.db.models import Group
 from app.db.resilience import run_with_sqlite_retry
 from app.image.codex_generator import CodexImageGenerator
+from app.image.delivery_guard import image_delivery_eligible
 from app.image.image_task import ImageJob
 from app.pipeline.delivery_stages import DeliveryStages
 from app.pipeline.generation_stages import GenerationStages
@@ -62,6 +63,7 @@
 from app.v2.constants import (
     CORRUPT,
     FAILED,
+    IMAGE_FALLBACK_NOT_SENDABLE,
     IMAGE_GENERATION_FAILED,
     IMAGE_READY,
     PROMPT_FAILED,
@@ -628,6 +630,31 @@ def send_due_for_dates(
                     continue  # 已发送，绝不重复
                 if run.get("send_hold"):
                     continue  # unknown / 手工审核必须保持 fail-closed
+                if not image_delivery_eligible(run):
+                    detail = "Level 3/Pillow 诊断图不可发送，已在发送扫描阶段拦截"
+                    self.store.update(
+                        group_name,
+                        run_date,
+                        status=FAILED,
+                        failed_stage="image",
+                        error=detail,
+                        error_type=IMAGE_FALLBACK_NOT_SENDABLE,
+                        send_state="held",
+                        send_hold=True,
+                        send_hold_reason=IMAGE_FALLBACK_NOT_SENDABLE,
+                        needs_manual_send=False,
+                        send_error=detail,
+                        send_error_type=IMAGE_FALLBACK_NOT_SENDABLE,
+                    )
+                    results.append(
+                        {
+                            "group_name": group_name,
+                            "status": "failed",
+                            "error_type": IMAGE_FALLBACK_NOT_SENDABLE,
+                            "detail": detail,
+                        }
+                    )
+                    continue
                 send_time = parse_send_time(self.settings.schedule_send_time)
                 due_at = datetime.combine(report_date, send_time, tzinfo=now.tzinfo)
                 if now < due_at:
@@ -1465,6 +1492,12 @@ def force_send(
             return {"status": "failed", "error": f"群不存在 {group_id}"}
         group_name = group.display_name or group.wechat_group_name
         run = self.store.load_run(group_name, run_date)
+        if not image_delivery_eligible(run):
+            return {
+                "status": "failed",
+                "error_type": IMAGE_FALLBACK_NOT_SENDABLE,
+                "error": "Level 3/Pillow 诊断图不可发送",
+            }
         can_resend_review = (
             run.get("status") == SENT
             and run.get("image_regen_status") == "ready_for_review"
diff --git a/app/pipeline/delivery_stages.py b/app/pipeline/delivery_stages.py
index b7334e2..b7b3f31 100644
--- a/app/pipeline/delivery_stages.py
+++ b/app/pipeline/delivery_stages.py
@@ -10,11 +10,18 @@
 from app.config.settings import Settings
 from app.core.observability import log_event
 from app.db.models import Group
+from app.image.delivery_guard import image_delivery_eligible
 from app.image.image_task import verify_image
 from app.pipeline.stage_result import StageResult
 from app.sender.base import WechatSender
 from app.services.group_name_sync import effective_send_target, send_target_mode
-from app.v2.constants import FAILED, IMAGE_FILE_MISSING, READY_TO_SEND, SENT
+from app.v2.constants import (
+    FAILED,
+    IMAGE_FALLBACK_NOT_SENDABLE,
+    IMAGE_FILE_MISSING,
+    READY_TO_SEND,
+    SENT,
+)
 from app.v2.run_store import RunStore
 
 
@@ -118,6 +125,15 @@ def _claim(self, context: DeliveryContext) -> StageResult[DeliveryContext]:
             allow_sent=context.allow_sent,
         )
         if not claim_id:
+            if claim_reason == IMAGE_FALLBACK_NOT_SENDABLE:
+                return StageResult.stop(
+                    {
+                        "group_name": context.group_name,
+                        "status": "failed",
+                        "error_type": IMAGE_FALLBACK_NOT_SENDABLE,
+                        "detail": "Level 3/Pillow 诊断图不可发送",
+                    }
+                )
             if claim_reason == "result_unknown":
                 return StageResult.stop(
                     {
@@ -160,6 +176,31 @@ def _prepare_payload(
         self,
         context: DeliveryContext,
     ) -> StageResult[DeliveryContext]:
+        if not image_delivery_eligible(context.run):
+            detail = "Level 3/Pillow 诊断图不可发送，已在发送前预检阶段拦截"
+            self.store.finish_send_claim(
+                context.group_name,
+                context.run_date,
+                context.claim_id,
+                send_state="held",
+                send_hold=True,
+                send_hold_reason=IMAGE_FALLBACK_NOT_SENDABLE,
+                needs_manual_send=False,
+                status=FAILED,
+                failed_stage="image",
+                error=detail,
+                error_type=IMAGE_FALLBACK_NOT_SENDABLE,
+                send_error=detail,
+                send_error_type=IMAGE_FALLBACK_NOT_SENDABLE,
+            )
+            return StageResult.stop(
+                {
+                    "group_name": context.group_name,
+                    "status": "failed",
+                    "error_type": IMAGE_FALLBACK_NOT_SENDABLE,
+                    "detail": detail,
+                }
+            )
         ranking_path = self.store.ranking_txt_path(
             context.group_name,
             context.run_date,
diff --git a/app/pipeline/image_stages.py b/app/pipeline/image_stages.py
index b9f7856..67e2ee2 100644
--- a/app/pipeline/image_stages.py
+++ b/app/pipeline/image_stages.py
@@ -7,12 +7,14 @@
 from typing import Callable
 import uuid
 
+from app.image.delivery_guard import image_delivery_eligible
 from app.image.image_task import ImageJob, SerialImageQueue
 from app.image.regeneration import normalize_candidate_diagnostics
 from app.core.logging import get_logger
 from app.core.observability import log_event
 from app.v2.constants import (
     FAILED,
+    IMAGE_FALLBACK_NOT_SENDABLE,
     IMAGE_GENERATION_FAILED,
     IMAGE_READY,
     READY_TO_SEND,
@@ -93,31 +95,44 @@ def record_result(self, job: ImageJob, result: dict) -> None:
         stage_timings = dict(current.get("stage_timings") or {})
         imagegen_ms = int(result.get("imagegen_ms") or 0)
         stage_timings["imagegen_ms"] = imagegen_ms
+        generator_detail = result.get("generator_detail")
+        if not isinstance(generator_detail, dict):
+            generator_detail = {}
+        image_metadata = {
+            "image_fallback_level": generator_detail.get("fallback_level"),
+            "image_variant": generator_detail.get("image_variant"),
+        }
+        diagnostic_fallback = not image_delivery_eligible(image_metadata)
         image_size_bytes = (
             job.output_path.stat().st_size
-            if result["success"] and job.output_path.is_file()
+            if (result["success"] or diagnostic_fallback) and job.output_path.is_file()
             else 0
         )
-        generator_detail = result.get("generator_detail")
-        if not isinstance(generator_detail, dict):
-            generator_detail = {}
+        finished_at = datetime.now().astimezone().isoformat()
         image_job = current.get("image_job") if isinstance(current.get("image_job"), dict) else {}
         candidates = normalize_candidate_diagnostics(generator_detail)
+        if result["success"]:
+            image_job_status = "completed"
+        elif diagnostic_fallback:
+            image_job_status = "diagnostic_fallback"
+        elif generator_detail.get("outcome_unknown"):
+            image_job_status = "result_unknown"
+        elif generator_detail.get("stage") == "ambiguous" or candidates:
+            image_job_status = "ambiguous_result"
+        else:
+            image_job_status = "failed"
         next_job = {
             **image_job,
-            "status": "completed" if result["success"] else (
-                "result_unknown"
-                if generator_detail.get("outcome_unknown")
-                else "ambiguous_result"
-                if generator_detail.get("stage") == "ambiguous" or candidates
-                else "failed"
-            ),
-            "finished_at": datetime.now().astimezone().isoformat(),
+            "status": image_job_status,
+            "finished_at": finished_at,
             "receipt": {
                 "job_id": job.job_id,
                 "revision": job.revision,
                 "prompt_sha256": job.prompt_sha256,
                 "image_path": str(job.output_path.resolve()) if result["success"] else "",
+                "diagnostic_path": (
+                    str(job.output_path.resolve()) if diagnostic_fallback else ""
+                ),
                 "sha256": str(generator_detail.get("sha256") or ""),
                 "source": str(generator_detail.get("receipt_source") or ""),
             },
@@ -136,10 +151,17 @@ def record_result(self, job: ImageJob, result: dict) -> None:
             stage_timings=stage_timings,
             imagegen_ms=imagegen_ms,
             image_generated_at=(
-                datetime.now().astimezone().isoformat()
+                finished_at
                 if result["success"]
                 else current.get("image_generated_at")
             ),
+            image_diagnostic_generated_at=(
+                finished_at
+                if diagnostic_fallback
+                else ""
+                if result["success"]
+                else current.get("image_diagnostic_generated_at", "")
+            ),
             image_size_bytes=image_size_bytes,
             image_attempt_count=int(generator_detail.get("attempt_count") or 0),
             image_recovery_status=str(generator_detail.get("recovery_status") or ""),
@@ -185,6 +207,17 @@ def record_result(self, job: ImageJob, result: dict) -> None:
     def advance_ready(self, job: ImageJob, run_date: str) -> None:
         run = self.store.load_run(job.group_name, run_date)
         if run.get("status") == IMAGE_READY:
+            if not image_delivery_eligible(run):
+                self.store.update(
+                    job.group_name,
+                    run_date,
+                    status=FAILED,
+                    failed_stage="image",
+                    error="Level 3/Pillow 诊断图不可进入发送流程",
+                    image_error="Level 3/Pillow 诊断图不可进入发送流程",
+                    error_type=IMAGE_FALLBACK_NOT_SENDABLE,
+                )
+                return
             self.store.update(job.group_name, run_date, status=READY_TO_SEND)
 
     def run_jobs(
diff --git a/app/scheduler/runtime_status.py b/app/scheduler/runtime_status.py
index b7f87e1..829b0a1 100644
--- a/app/scheduler/runtime_status.py
+++ b/app/scheduler/runtime_status.py
@@ -9,6 +9,7 @@
 from typing import Iterable
 from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
 
+from app.image.delivery_guard import image_delivery_eligible, image_fallback_level
 from app.v2.constants import (
     CORRUPT,
     EXECUTION_ACTIVE,
@@ -170,6 +171,9 @@ def _group_node_status(
 ) -> str:
     if node_id == "scheduler":
         return "success" if scheduler_started else "pending"
+    if node_id == "image" and not image_delivery_eligible(run):
+        # 历史 SENT 仍保持发送成功，但图片节点必须呈现诊断失败事实。
+        return "failed"
 
     status = _step_status(run, required_checkpoint, stage)
     if status != "pending":
@@ -333,7 +337,10 @@ def _group_snapshot(
             if isinstance(run.get("image_job"), dict)
             else "",
             "attempts": int(run.get("image_attempt_count") or 0),
-            "fallback_level": int(run.get("image_fallback_level") or 0),
+            "fallback_level": image_fallback_level(run),
+            "fallback_reason": str(run.get("image_fallback_reason") or ""),
+            "variant": str(run.get("image_variant") or "normal"),
+            "delivery_eligible": image_delivery_eligible(run),
         },
         "send": {
             "status": send_status,
@@ -354,6 +361,7 @@ def _group_snapshot(
         "last_error_summary": str(
             run.get("last_error_summary")
             or run.get("error")
+            or run.get("prompt_original_error")
             or run.get("send_error")
             or run.get("prompt_operation_error")
             or run.get("prompt_hold_reason")
diff --git a/app/v2/constants.py b/app/v2/constants.py
index 6040b8a..9275b4e 100644
--- a/app/v2/constants.py
+++ b/app/v2/constants.py
@@ -64,6 +64,7 @@
 IMAGE_GENERATION_FAILED = "IMAGE_GENERATION_FAILED"
 IMAGE_FILE_MISSING = "IMAGE_FILE_MISSING"
 IMAGE_CONTENT_VERIFICATION_FAILED = "IMAGE_CONTENT_VERIFICATION_FAILED"
+IMAGE_FALLBACK_NOT_SENDABLE = "IMAGE_FALLBACK_NOT_SENDABLE"
 WECHAT_OFFLINE = "WECHAT_OFFLINE"
 SEND_TEXT_FAILED = "SEND_TEXT_FAILED"
 SEND_IMAGE_FAILED = "SEND_IMAGE_FAILED"
diff --git a/app/v2/reliability.py b/app/v2/reliability.py
index 95ac608..b6b0fb8 100644
--- a/app/v2/reliability.py
+++ b/app/v2/reliability.py
@@ -21,6 +21,8 @@
     EXECUTION_WAIT_RETRY,
     FAILED,
     IMAGE_FILE_MISSING,
+    IMAGE_CONTENT_VERIFICATION_FAILED,
+    IMAGE_FALLBACK_NOT_SENDABLE,
     IMAGE_GENERATION_FAILED,
     IMAGE_READY,
     MESSAGE_FETCH_FAILED,
@@ -64,6 +66,7 @@
         PROMPT_FAILED,
         IMAGE_GENERATION_FAILED,
         IMAGE_FILE_MISSING,
+        IMAGE_CONTENT_VERIFICATION_FAILED,
         "DEEPSEEK_FAILED",
         "UNEXPECTED_GENERATION_ERROR",
         "SEND_TEXT_FAILED",
@@ -83,6 +86,7 @@
         "MESSAGE_SNAPSHOT_INVALID",
         "MISSED_SEND_WINDOW",
         "GROUP_TARGET_MISMATCH",
+        IMAGE_FALLBACK_NOT_SENDABLE,
     }
 )
 
diff --git a/app/v2/run_store.py b/app/v2/run_store.py
index dd14bbf..f962ad6 100644
--- a/app/v2/run_store.py
+++ b/app/v2/run_store.py
@@ -27,6 +27,7 @@
     validate_iso_date,
     validate_path_label,
 )
+from app.image.delivery_guard import image_delivery_eligible
 from app.services.handoff_service import safe_dir_name
 from app.v2.constants import (
     CORRUPT,
@@ -39,6 +40,7 @@
     FILE_RANKING_JSON,
     FILE_RANKING_TXT,
     FILE_RUN,
+    IMAGE_FALLBACK_NOT_SENDABLE,
     IMAGE_READY,
     PENDING,
     PROMPT_READY,
@@ -651,6 +653,8 @@ def claim_send(
                 return None, data, "failed_final"
             if data.get("send_hold") and not allow_hold:
                 return None, data, "send_hold"
+            if not image_delivery_eligible(data):
+                return None, data, IMAGE_FALLBACK_NOT_SENDABLE
             prompt_meta = (
                 data.get("prompt_meta")
                 if isinstance(data.get("prompt_meta"), dict)
diff --git a/frontend/e2e/dashboard.spec.ts b/frontend/e2e/dashboard.spec.ts
index 733528d..c8d56d6 100644
--- a/frontend/e2e/dashboard.spec.ts
+++ b/frontend/e2e/dashboard.spec.ts
@@ -70,6 +70,11 @@ const dashboard = {
       message_count: 12,
       speaker_count: 3,
       image_url: "",
+      image_status: "",
+      image_fallback_level: 0,
+      image_fallback_reason: "",
+      image_variant: "normal",
+      image_delivery_eligible: true,
       ranking_preview: [{ rank: 1, name: "成员甲", count: 8 }],
       ranking_error: "",
       error: "",
@@ -94,7 +99,7 @@ async function json(route: Route, body: unknown) {
   });
 }
 
-async function installFakeApi(page: Page, held = false) {
+async function installFakeApi(page: Page, held = false, diagnostic = false) {
   const calls: { path: string; search: string; body: unknown }[] = [];
   await page.route("**/api/**", async (route) => {
     const request = route.request();
@@ -103,11 +108,33 @@ async function installFakeApi(page: Page, held = false) {
     const body = request.postDataJSON?.() ?? null;
     calls.push({ path, search: url.search, body });
 
-    if (path === "/api/v2/dashboard") return json(route, held ? {
-      ...dashboard,
-      counts: { ...dashboard.counts, generated: 0, held: 1 },
-      cards: dashboard.cards.map((card) => ({ ...card, send_hold: true, send_state: "unknown", send_hold_reason: "SEND_RESULT_UNKNOWN", error: "发送结果需要人工核对" })),
-    } : dashboard);
+    if (path === "/api/v2/dashboard") {
+      if (diagnostic) {
+        return json(route, {
+          ...dashboard,
+          counts: { ...dashboard.counts, generated: 0, sent: 1 },
+          next_send: "",
+          cards: dashboard.cards.map((card) => ({
+            ...card,
+            image_enabled: true,
+            status: "SENT",
+            sent_at: "2026-08-25T08:36:00+08:00",
+            image_url: "data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNk+A8AAQUBAScY42YAAAAASUVORK5CYII=",
+            image_status: "failed",
+            image_fallback_level: 3,
+            image_fallback_reason: "PROMPT_FAILED",
+            image_variant: "pillow",
+            image_delivery_eligible: false,
+            error: "Prompt 连续校验失败",
+          })),
+        });
+      }
+      return json(route, held ? {
+        ...dashboard,
+        counts: { ...dashboard.counts, generated: 0, held: 1 },
+        cards: dashboard.cards.map((card) => ({ ...card, send_hold: true, send_state: "unknown", send_hold_reason: "SEND_RESULT_UNKNOWN", error: "发送结果需要人工核对" })),
+      } : dashboard);
+    }
     if (path === "/api/v2/runtime/logs") {
       return json(route, {
         run_date: runDate,
@@ -212,6 +239,21 @@ test("Dashboard 人工核对只写状态，不调用发送接口", async ({ page
   expect(calls.some((call) => call.path === "/api/v2/pipeline/send")).toBe(false);
 });
 
+test("Dashboard 对历史已发送诊断图同时显示失败事实且不提供重发", async ({ page }) => {
+  const calls = await installFakeApi(page, false, true);
+  await page.goto("/#/dashboard");
+
+  await expect(page.getByText("已发送", { exact: true })).toBeVisible();
+  await expect(page.getByText("图片生成失败（已发送）", { exact: true })).toBeVisible();
+  await expect(page.getByText("诊断图不可发送", { exact: true })).toBeVisible();
+  const image = page.getByAltText("测试群 不可发送诊断图");
+  await expect(image).toBeVisible();
+  expect(await image.evaluate((element) => getComputedStyle(element).objectFit)).toBe("contain");
+  await expect(page.getByRole("button", { name: "立即发送" })).toHaveCount(0);
+  await expect(page.getByRole("button", { name: "人工核对" })).toHaveCount(0);
+  expect(calls.some((call) => call.path === "/api/v2/pipeline/send")).toBe(false);
+});
+
 test.describe("减少动态效果", () => {
   test.use({ reducedMotion: "reduce" });
   test("页面转场降级后仍到达最终可访问状态", async ({ page }) => {
diff --git a/frontend/src/api.ts b/frontend/src/api.ts
index 34ec3ab..212ec81 100644
--- a/frontend/src/api.ts
+++ b/frontend/src/api.ts
@@ -134,6 +134,11 @@ export interface DashboardCard {
   message_count: number;
   speaker_count: number;
   image_url: string;
+  image_status: string;
+  image_fallback_level: number;
+  image_fallback_reason: string;
+  image_variant: string;
+  image_delivery_eligible: boolean;
   ranking_preview: {
     rank: number;
     name: string;
diff --git a/frontend/src/pages/v2/Dashboard.tsx b/frontend/src/pages/v2/Dashboard.tsx
index 74480b3..5200f5b 100644
--- a/frontend/src/pages/v2/Dashboard.tsx
+++ b/frontend/src/pages/v2/Dashboard.tsx
@@ -75,19 +75,23 @@ interface ViewerImage {
 
 function ImagePreview({ card, onOpen }: { card: DashboardCard; onOpen: (image: ViewerImage) => void }) {
   const [imageBroken, setImageBroken] = useState(false);
+  const diagnostic = card.image_delivery_eligible === false;
 
   if (card.image_url) {
-    const alt = `${card.group_name} 日报图片`;
+    const alt = diagnostic ? `${card.group_name} 不可发送诊断图` : `${card.group_name} 日报图片`;
     if (!imageBroken) {
       return (
-        <ImagePreviewTrigger
-          src={card.image_url}
-          alt={alt}
-          imageClassName="dashboard-task-image"
-          className="dashboard-task-image-trigger"
-          onError={() => setImageBroken(true)}
-          onOpen={() => onOpen({ src: card.image_url, alt, filename: "daily_image.png", title: alt })}
-        />
+        <div className={`dashboard-image-frame${diagnostic ? " is-diagnostic" : ""}`}>
+          {diagnostic && <StatusBadge tone="danger">诊断图不可发送</StatusBadge>}
+          <ImagePreviewTrigger
+            src={card.image_url}
+            alt={alt}
+            imageClassName="dashboard-task-image"
+            className="dashboard-task-image-trigger"
+            onError={() => setImageBroken(true)}
+            onOpen={() => onOpen({ src: card.image_url, alt, filename: "daily_image.png", title: alt })}
+          />
+        </div>
       );
     }
   }
@@ -124,7 +128,10 @@ function TaskActions({
 }: TaskActionsProps) {
   const canGenerate = card.status !== "SENT" && !card.prompt_hold;
   const canResolvePrompt = card.prompt_hold && card.prompt_hold_reason === "PROMPT_RESULT_UNKNOWN";
-  const canSend = ["IMAGE_READY", "READY_TO_SEND"].includes(card.status) && !card.sent_at && !card.send_hold;
+  const canSend = ["IMAGE_READY", "READY_TO_SEND"].includes(card.status)
+    && !card.sent_at
+    && !card.send_hold
+    && card.image_delivery_eligible !== false;
 
   return (
     <div className="dashboard-task-actions">
@@ -171,7 +178,7 @@ function TaskActions({
           微信发送未启用
         </Button>
       )}
-      {card.send_hold && (
+      {card.send_hold && card.image_delivery_eligible !== false && (
         <Button tone="secondary" className="ui-button-compact" onClick={onResolve}>
           <WarningCircle size={16} aria-hidden="true" />
           人工核对
@@ -555,7 +562,12 @@ export default function Dashboard() {
                     <h3>{card.group_name}</h3>
                     <span>{formatPeriod(card)} · 更新 {formatDateTime(card.updated_at)}</span>
                   </div>
-                  {card.prompt_hold ? <StatusBadge tone="warning">暂停待核对</StatusBadge> : <StatusPill status={card.status} />}
+                  <div className="dashboard-task-statuses">
+                    {card.prompt_hold ? <StatusBadge tone="warning">暂停待核对</StatusBadge> : <StatusPill status={card.status} />}
+                    {card.image_enabled && card.image_delivery_eligible === false && (
+                      <StatusBadge tone="danger">{card.sent_at || card.status === "SENT" ? "图片生成失败（已发送）" : "图片生成失败"}</StatusBadge>
+                    )}
+                  </div>
                 </div>
                 <div className="dashboard-task-content">
                   <section className="dashboard-ranking-preview" aria-label={`${card.group_name} Top 5 排行榜`}>
diff --git a/frontend/src/styles.css b/frontend/src/styles.css
index ee77650..fee82ee 100644
--- a/frontend/src/styles.css
+++ b/frontend/src/styles.css
@@ -1927,6 +1927,15 @@ textarea:focus-visible {
   min-width: 0;
 }
 
+.dashboard-task-statuses {
+  display: flex;
+  flex: 0 0 auto;
+  align-items: center;
+  justify-content: flex-end;
+  flex-wrap: wrap;
+  gap: 6px;
+}
+
 .dashboard-task-card h3 {
   overflow: hidden;
   color: var(--text);
@@ -1952,10 +1961,31 @@ textarea:focus-visible {
 
 .dashboard-task-image {
   display: block;
-  object-fit: cover;
+  object-fit: contain;
+  background: #edf1f6;
+}
+
+.dashboard-image-frame {
+  position: relative;
+  height: 142px;
+  overflow: hidden;
+  border-radius: 10px;
   background: #edf1f6;
 }
 
+.dashboard-image-frame.is-diagnostic {
+  outline: 2px solid color-mix(in srgb, var(--danger) 55%, transparent);
+  outline-offset: -2px;
+}
+
+.dashboard-image-frame > .ui-status {
+  position: absolute;
+  z-index: 1;
+  top: 8px;
+  left: 8px;
+  pointer-events: none;
+}
+
 .dashboard-task-image-trigger {
   height: 142px;
   border-radius: 10px;
diff --git a/tests/test_generation_concurrency.py b/tests/test_generation_concurrency.py
index 48a5302..ed034d0 100644
--- a/tests/test_generation_concurrency.py
+++ b/tests/test_generation_concurrency.py
@@ -7,6 +7,7 @@
 from datetime import datetime
 
 import pytest
+from PIL import Image
 from sqlmodel import Session, SQLModel, create_engine
 
 from app.ai.concurrency import bounded_slot
@@ -170,10 +171,7 @@ def generate(self, prompt_file, output_path):
         try:
             self.barrier.wait(timeout=5)
             output_path.parent.mkdir(parents=True, exist_ok=True)
-            output_path.write_bytes(bytes.fromhex(
-                "89504e470d0a1a0a0000000d49484452000000010000000108060000001f15c489"
-                "0000000d4944415478da63f8cfc0f80100050001fff83f240000000049454e44ae426082"
-            ))
+            Image.new("RGBA", (1, 1), (0, 0, 0, 0)).save(output_path, format="PNG")
         finally:
             with self.lock:
                 self.active -= 1
diff --git a/tests/test_image_layouts.py b/tests/test_image_layouts.py
index 79b5269..f095917 100644
--- a/tests/test_image_layouts.py
+++ b/tests/test_image_layouts.py
@@ -62,8 +62,8 @@ def test_catalog_contains_distinct_comic_panel_grammars():
         assert any(word in combined for word in ("大格", "宽格", "竖格", "特写", "跨格", "尺寸"))
 
 
-@pytest.mark.parametrize("count", [5, 6, 7])
-def test_parse_accepts_five_to_seven_topics_and_expands_panel_count(count: int):
+@pytest.mark.parametrize("count", [1, 5, 6, 7])
+def test_parse_accepts_one_to_seven_topics_and_expands_panel_count(count: int):
     plan = parse_layout_plan(_raw_plan(count), _topic_ids(count))
     assert list(plan.topic_order) == _topic_ids(count)
     assert plan.panel_count == count + 1
@@ -131,6 +131,15 @@ def test_fallback_avoids_recent_three_and_keeps_unequal_storyboard():
     assert "topic-" not in instruction
 
 
+def test_single_topic_fallback_uses_one_featured_topic_without_padding():
+    plan = fallback_layout_plan(_topic_ids(1), seed_text="group|2026-09-02")
+
+    assert plan.topic_order == ("topic-01",)
+    assert plan.featured_topic_ids == ("topic-01",)
+    assert plan.structure_mode == "hero_rhythm"
+    assert plan.panel_count == 2
+
+
 def test_only_current_catalog_with_valid_panel_beats_is_restored():
     plan = parse_layout_plan(_raw_plan(5, layout_id="split_focus"), _topic_ids(5))
     meta = plan.to_meta()
@@ -147,5 +156,5 @@ def test_only_current_catalog_with_valid_panel_beats_is_restored():
 
 
 def test_more_than_seven_topics_is_rejected():
-    with pytest.raises(LayoutPlanError, match="2～7"):
+    with pytest.raises(LayoutPlanError, match="1～7"):
         fallback_layout_plan(_topic_ids(8))
diff --git a/tests/test_runtime_status.py b/tests/test_runtime_status.py
index 3f844f4..8f26804 100644
--- a/tests/test_runtime_status.py
+++ b/tests/test_runtime_status.py
@@ -216,3 +216,40 @@ def test_build_daily_status_keeps_queued_image_pending_instead_of_claiming_runni
     assert group["image"]["job_status"] == "queued"
     assert group["image"]["status"] == "pending"
     assert group["node_status"] == "pending"
+
+
+def test_runtime_status_preserves_sent_truth_but_marks_diagnostic_image_failed(tmp_path):
+    store = RunStore(tmp_path / "output")
+    store.save_run(
+        "历史诊断图群",
+        "2026-09-03",
+        {
+            "group_id": "23",
+            "status": SENT,
+            "sent_at": "2026-09-03T08:36:00+08:00",
+            "image_status": "success",
+            "image_fallback_level": 3,
+            "image_fallback_reason": "PROMPT_FAILED",
+            "image_variant": "pillow",
+            "prompt_original_error": "Prompt 连续校验失败",
+        },
+    )
+
+    payload = build_daily_status(store, "2026-09-03")
+    group = payload["groups"][0]
+    by_node = {item["id"]: item for item in group["nodes"]}
+
+    assert group["run_status"] == SENT
+    assert group["node_status"] == "success"
+    assert group["send"]["status"] == "success"
+    assert by_node["image"]["status"] == "failed"
+    assert group["image"] == {
+        "status": "failed",
+        "job_status": "",
+        "attempts": 0,
+        "fallback_level": 3,
+        "fallback_reason": "PROMPT_FAILED",
+        "variant": "pillow",
+        "delivery_eligible": False,
+    }
+    assert group["last_error_summary"] == "Prompt 连续校验失败"
diff --git a/tests/test_topic_selection.py b/tests/test_topic_selection.py
index abd4e59..1cae905 100644
--- a/tests/test_topic_selection.py
+++ b/tests/test_topic_selection.py
@@ -116,14 +116,114 @@ def test_high_volume_chat_does_not_reduce_topic_density():
     assert selection["thresholds"]["high_volume_message_threshold"] == 200
 
 
-def test_single_candidate_fails_instead_of_splitting_one_topic_to_fill_quota():
-    with pytest.raises(TopicSelectionError, match="TOPIC_CANDIDATES_INSUFFICIENT"):
-        score_and_select_topics([_candidate(1, ["m0", "m1"])], _messages(2))
+def test_single_candidate_is_selected_without_splitting_or_padding():
+    selection = score_and_select_topics(
+        [_candidate(1, ["m0", "m1"])], _messages(2)
+    )
+
+    assert selection["selected_count"] == 1
+    assert selection["selected_topic_ids"] == ["topic-01"]
+
+
+def test_single_message_keeps_one_real_topic_without_fabricating_second_topic():
+    selection = score_and_select_topics([_candidate(1, ["m0"])], _messages(1))
+
+    assert selection["selected_count"] == 1
+    assert selection["candidates"][0]["evidence_message_count"] == 1
+
+
+def test_high_scoring_political_topic_is_blocked_and_safe_topic_is_promoted():
+    political = _candidate(1, ["m0", "m1"], comedy=40, visual=20, recognition=20)
+    political.update(title="总统大选", summary="围绕总统大选展开讨论。")
+    safe = _candidate(2, ["m2", "m3"], comedy=20, visual=10, recognition=10)
+
+    selection = score_and_select_topics([political, safe], _messages(4))
+    by_id = {item["topic_id"]: item for item in selection["candidates"]}
+
+    assert selection["selected_topic_ids"] == ["topic-02"]
+    assert selection["safe_candidate_count"] == 1
+    assert selection["blocked_topic_ids"] == ["topic-01"]
+    assert selection["political_keyword_hits"] == [
+        {
+            "topic_id": "topic-01",
+            "title": "总统大选",
+            "matched_keywords": ["大选", "总统"],
+        }
+    ]
+    assert by_id["topic-01"]["rank"] == 1
+    assert by_id["topic-01"]["selected"] is False
+    assert by_id["topic-01"]["image_eligible"] is False
+    assert set(by_id["topic-01"]["political_keyword_matches"]) == {"总统", "大选"}
+    assert by_id["topic-02"]["eligible_rank"] == 1
+
+
+@pytest.mark.parametrize(
+    ("title", "summary"),
+    [
+        ("实弹卡弹成真伪证明", "射击体验中遇到卡弹。"),
+        ("拿破仑轶事", "群友分享一个只含人物姓名的历史故事。"),
+        ("公司主席投票", "公司内部投票讨论由谁主持会议。"),
+        ("周末 party", "群友商量周末聚会安排。"),
+        ("company party", "The company party starts after work."),
+    ],
+)
+def test_non_political_or_broad_terms_are_not_blocked(title: str, summary: str):
+    candidate = _candidate(1, ["m0"])
+    candidate.update(title=title, summary=summary)
+
+    selection = score_and_select_topics([candidate], _messages(1))
+
+    assert selection["selected_topic_ids"] == ["topic-01"]
+    assert selection["blocked_topic_ids"] == []
+    assert selection["candidates"][0]["image_eligible"] is True
+
+
+def test_explicit_english_political_term_is_blocked():
+    candidate = _candidate(1, ["m0"])
+    candidate.update(
+        title="Ruling party election",
+        summary="The ruling party prepared for a general election.",
+    )
+
+    selection = score_and_select_topics([candidate], _messages(1))
+
+    assert selection["selected_count"] == 0
+    assert set(selection["candidates"][0]["political_keyword_matches"]) == {
+        "ruling party",
+        "general election",
+    }
+
+
+def test_verified_evidence_text_is_checked_for_political_keywords():
+    messages = _messages(1)
+    messages[0] = PromptMessage(
+        messages[0].message_id,
+        messages[0].timestamp,
+        messages[0].sender_name,
+        "讨论全国人大相关消息",
+        messages[0].sender_id,
+    )
+
+    selection = score_and_select_topics([_candidate(1, ["m0"])], messages)
+
+    assert selection["selected_count"] == 0
+    assert selection["candidates"][0]["political_keyword_matches"] == ["全国人大"]
+
+
+def test_all_political_candidates_keep_audit_data_and_stop_prompt_selection():
+    first = _candidate(1, ["m0"])
+    first.update(title="总统竞选", summary="总统竞选相关讨论。")
+    second = _candidate(2, ["m1"])
+    second.update(title="议会选举", summary="议会选举相关讨论。")
 
+    selection = score_and_select_topics([first, second], _messages(2))
 
-def test_single_message_fails_without_fabricating_second_topic():
-    with pytest.raises(TopicSelectionError, match="TOPIC_CANDIDATES_INSUFFICIENT"):
-        score_and_select_topics([_candidate(1, ["m0"])], _messages(1))
+    assert selection["candidate_count"] == 2
+    assert selection["safe_candidate_count"] == 0
+    assert selection["blocked_candidate_count"] == 2
+    assert selection["selected_count"] == 0
+    with pytest.raises(TopicSelectionError, match="TOPIC_CANDIDATES_POLITICAL"):
+        selected_topics_json(selection)
 
 
 def test_parse_filters_duplicate_and_invalid_message_ids():
diff --git a/tests/test_v2_image_task.py b/tests/test_v2_image_task.py
index 83399b7..d5a5fc1 100644
--- a/tests/test_v2_image_task.py
+++ b/tests/test_v2_image_task.py
@@ -160,9 +160,14 @@ def test_single_failure_does_not_block_others(tmp_path):
         _job(tmp_path, "群3", FakeGenerator(fail=True)),
     ]
     results = queue.run_all(jobs)
-    assert [r["status"] for r in results] == ["success", "success", "success"]
+    assert [r["status"] for r in results] == [
+        "diagnostic_fallback",
+        "success",
+        "diagnostic_fallback",
+    ]
+    assert [r["success"] for r in results] == [False, True, False]
     assert results[0]["generator_detail"]["fallback_level"] == 3
-    # 外部失败的群得到本地信息图，其他群也不受影响。
+    # 外部失败的群保留不可发送诊断图，其他群仍独立成功。
     for job in jobs:
         ok, _ = verify_image(job.output_path)
         assert ok is True
@@ -237,7 +242,7 @@ def generate(self, prompt_file, output_path):
     assert not job.output_path.exists()
 
 
-def test_strict_verification_known_failure_uses_local_fallback(tmp_path, monkeypatch):
+def test_strict_verification_known_failure_keeps_diagnostic_fallback_failed(tmp_path, monkeypatch):
     from app.image.image_task import ImageTaskResult
 
     class QualityRetryFailsKnown:
@@ -273,21 +278,46 @@ def generate(self, _prompt_file, output_path, **_kwargs):
         lambda self, reason: fallback_calls.append(reason)
         or {
             "group_name": self.group_name,
-            "status": "success",
-            "success": True,
+            "status": "diagnostic_fallback",
+            "success": False,
             "detail": "local fallback",
-            "error_type": "",
-            "generator_detail": {"fallback_level": 3},
+            "error_type": "IMAGE_CONTENT_VERIFICATION_FAILED",
+            "generator_detail": {"fallback_level": 3, "image_variant": "pillow"},
         },
     )
 
     result = job.run()
 
-    assert result["status"] == "success"
+    assert result["status"] == "diagnostic_fallback"
+    assert result["success"] is False
+    assert result["error_type"] == "IMAGE_CONTENT_VERIFICATION_FAILED"
     assert generator.calls == 2
     assert fallback_calls == ["IMAGE_CONTENT_VERIFICATION_FAILED"]
 
 
+def test_retry_does_not_treat_existing_diagnostic_png_as_success(tmp_path):
+    generator = FakeGenerator()
+    job = _job(tmp_path, "诊断图重试群", generator)
+    job.output_path.parent.mkdir(parents=True, exist_ok=True)
+    job.output_path.write_bytes(_PNG_1PX)
+    (job.output_path.parent / "run.json").write_text(
+        json.dumps(
+            {
+                "image_fallback_level": 3,
+                "image_variant": "pillow",
+                "image_force_local_fallback": False,
+            },
+            ensure_ascii=False,
+        ),
+        encoding="utf-8",
+    )
+
+    result = job.run()
+
+    assert result["status"] == "success"
+    assert len(generator.calls) == 1
+
+
 def test_strict_verification_unknown_retry_stays_failed_closed(tmp_path, monkeypatch):
     from app.image.image_task import ImageTaskResult
 
diff --git a/tests/test_v2_pipeline.py b/tests/test_v2_pipeline.py
index f90aa47..c427841 100644
--- a/tests/test_v2_pipeline.py
+++ b/tests/test_v2_pipeline.py
@@ -29,6 +29,7 @@
 from app.pipeline.daily_pipeline import DailyPipeline
 from app.v2.constants import (
     FAILED,
+    IMAGE_FALLBACK_NOT_SENDABLE,
     IMAGE_FILE_MISSING,
     IMAGE_GENERATION_FAILED,
     IMAGE_READY,
@@ -107,9 +108,10 @@ def fetch_messages(self, group_id, start_time, end_time) -> FetchResult:
 
 
 class FakePrompt:
-    def __init__(self, fail=False, failure_meta=None):
+    def __init__(self, fail=False, failure_meta=None, topic_selection=None):
         self.fail = fail
         self.failure_meta = failure_meta
+        self.topic_selection = topic_selection
         self.inputs = []
 
     def build(self, data):
@@ -123,7 +125,7 @@ def build(self, data):
                 error="DeepSeek 失败",
                 meta=self.failure_meta,
             )
-        topic_selection = data.persisted_topic_selection or {
+        topic_selection = data.persisted_topic_selection or self.topic_selection or {
             "topic_selection_version": "4.0",
             "selected_topic_ids": ["topic-01", "topic-02"],
             "selected_count": 2,
@@ -146,6 +148,8 @@ def build(self, data):
         topic_selection = dict(topic_selection)
         topic_selection.setdefault("message_snapshot_sha256", snapshot_hash)
         topic_selection.setdefault("speaker_fingerprint", speaker_fingerprint)
+        selected_topic_ids = list(topic_selection.get("selected_topic_ids") or [])
+        single_topic = len(selected_topic_ids) == 1
         return PromptOutput(
             True,
             "【任务】\n生成图片\n【主标题】今天热聊",
@@ -156,13 +160,16 @@ def build(self, data):
                 "speaker_fingerprint": speaker_fingerprint,
                 "speaker_bindings": [],
                 "layout_catalog_version": "comic-panels-v3",
-                "layout_id": "split_focus",
-                "structure_mode": "dual_rhythm",
-                "featured_topic_ids": ["topic-01", "topic-02"],
-                "topic_order": ["topic-01", "topic-02"],
+                "layout_id": "hero_with_insets" if single_topic else "split_focus",
+                "structure_mode": "hero_rhythm" if single_topic else "dual_rhythm",
+                "featured_topic_ids": selected_topic_ids[:1] if single_topic else selected_topic_ids[:2],
+                "topic_order": selected_topic_ids,
                 "panel_beats": [
-                    {"topic_id": "topic-01", "shots": ["establishing", "reaction"]},
-                    {"topic_id": "topic-02", "shots": ["dialogue"]},
+                    {
+                        "topic_id": topic_id,
+                        "shots": ["establishing", "reaction"] if index == 0 else ["dialogue"],
+                    }
+                    for index, topic_id in enumerate(selected_topic_ids)
                 ],
             },
         )
@@ -746,15 +753,23 @@ def test_force_generate_blocks_corrupt_state_before_name_sync(tmp_path, monkeypa
     assert run_path.read_bytes() == original
 
 
-def test_force_generate_image_failure_uses_local_fallback(tmp_path):
+def test_force_generate_image_failure_keeps_diagnostic_fallback_failed(tmp_path):
     gen = FakeGenerator(fail=True)
     pipeline, group = _make_pipeline(tmp_path, gen=gen)
     result = pipeline.force_generate(group.id, "2026-08-18")
-    assert result["status"] == "ready_to_send"
+    assert result["status"] == "failed"
     run = pipeline.store.load_run("测试群", "2026-08-18")
-    assert run["status"] == READY_TO_SEND
+    assert run["status"] == FAILED
+    assert run["failed_stage"] == "image"
+    assert run["error_type"] == IMAGE_GENERATION_FAILED
+    assert run["image_status"] == "diagnostic_fallback"
     assert run["image_fallback_level"] == 3
     assert run["image_variant"] == "pillow"
+    assert run["image_job"]["status"] == "diagnostic_fallback"
+    assert run["image_job"]["receipt"]["image_path"] == ""
+    assert run["image_job"]["receipt"]["diagnostic_path"]
+    assert len(run["image_job"]["receipt"]["sha256"]) == 64
+    assert pipeline.store.image_path("测试群", "2026-08-18").is_file()
 
 
 def test_image_success_clears_stale_failure_fields(tmp_path):
@@ -768,6 +783,9 @@ def test_image_success_clears_stale_failure_fields(tmp_path):
         error="旧生图失败",
         error_type=IMAGE_GENERATION_FAILED,
         image_error="旧生图失败",
+        image_fallback_level=3,
+        image_fallback_reason="PROMPT_FAILED",
+        image_variant="pillow",
     )
     job = pipeline._make_image_job(group, "2026-08-18", force=True)
 
@@ -789,6 +807,9 @@ def test_image_success_clears_stale_failure_fields(tmp_path):
     assert run["error"] is None
     assert run["error_type"] is None
     assert run["image_error"] is None
+    assert run["image_fallback_level"] == 0
+    assert run["image_fallback_reason"] == ""
+    assert run["image_variant"] == "normal"
 
 
 def test_image_job_refuses_stale_prompt_contract(tmp_path):
@@ -816,9 +837,15 @@ def test_generate_data_failure_marks_failed(tmp_path):
     assert run["failed_stage"] == "data"
 
 
-def test_generate_prompt_failure_uses_local_infographic(tmp_path):
+def test_generate_prompt_failure_keeps_local_infographic_as_failed_diagnostic(tmp_path):
     preserved_meta = {
-        "topic_selection": {"selected_topic_ids": ["topic-01"]},
+        "topic_selection": {
+            "political_keyword_policy_version": "political-keywords-v1",
+            "selected_topic_ids": [],
+            "safe_candidate_count": 0,
+            "blocked_candidate_count": 2,
+            "blocked_topic_ids": ["topic-01", "topic-02"],
+        },
         "layout_id": "split_focus",
     }
     pipeline, group = _make_pipeline(
@@ -826,9 +853,12 @@ def test_generate_prompt_failure_uses_local_infographic(tmp_path):
         prompt=FakePrompt(fail=True, failure_meta=preserved_meta),
     )
     results = pipeline.generate_all(run_date="2026-08-18")
-    assert results[0]["status"] == "ready_to_send"
+    assert results[0]["status"] == "failed"
+    assert results[0]["error_type"] == "PROMPT_FAILED"
     run = pipeline.store.load_run("测试群", "2026-08-18")
-    assert run["status"] == READY_TO_SEND
+    assert run["status"] == FAILED
+    assert run["failed_stage"] == "image"
+    assert run["image_status"] == "diagnostic_fallback"
     assert run["prompt_fallback_level"] == 3
     assert run["image_fallback_level"] == 3
     assert run["image_variant"] == "pillow"
@@ -838,6 +868,52 @@ def test_generate_prompt_failure_uses_local_infographic(tmp_path):
     assert pipeline.store.image_path("测试群", "2026-08-18").is_file()
 
 
+def test_generate_one_safe_topic_still_calls_normal_image_generator(tmp_path):
+    topic_selection = {
+        "topic_selection_version": "7.0",
+        "political_keyword_policy_version": "political-keywords-v1",
+        "selected_topic_ids": ["topic-02"],
+        "selected_count": 1,
+        "safe_candidate_count": 1,
+        "blocked_candidate_count": 1,
+        "blocked_topic_ids": ["topic-01"],
+        "candidates": [
+            {
+                "topic_id": "topic-01",
+                "selected": False,
+                "image_eligible": False,
+                "political_keyword_matches": ["总统"],
+            },
+            {
+                "topic_id": "topic-02",
+                "selected": True,
+                "image_eligible": True,
+                "title": "票房",
+                "summary": "张三聊票房",
+                "message_ids": ["m1"],
+                "quotes": ["今天聊了票房"],
+                "visible_participants": ["张三"],
+            },
+        ],
+    }
+    generator = FakeGenerator()
+    pipeline, _ = _make_pipeline(
+        tmp_path,
+        prompt=FakePrompt(topic_selection=topic_selection),
+        gen=generator,
+    )
+
+    results = pipeline.generate_all(run_date="2026-08-18")
+
+    assert results[0]["status"] == "ready_to_send"
+    assert len(generator.calls) == 1
+    run = pipeline.store.load_run("测试群", "2026-08-18")
+    assert run["prompt_fallback_level"] == 0
+    assert run["image_fallback_level"] == 0
+    assert run["prompt_meta"]["topic_selection"]["selected_topic_ids"] == ["topic-02"]
+    assert run["prompt_meta"]["structure_mode"] == "hero_rhythm"
+
+
 def test_image_disabled_goes_ready_to_send_without_image(tmp_path):
     pipeline, group = _make_pipeline(tmp_path, image_enabled=False)
     results = pipeline.generate_all(run_date="2026-08-18")
@@ -1016,6 +1092,121 @@ def test_send_due_sends_text_then_image(tmp_path):
     assert len(sender.image_calls) == 1
 
 
+@pytest.mark.parametrize(
+    ("fallback_level", "image_variant"),
+    [(3, "normal"), (0, "pillow")],
+)
+def test_send_scan_rejects_diagnostic_fallback_without_sender_calls(
+    tmp_path, fallback_level, image_variant
+):
+    pipeline = _ready_to_send(tmp_path)
+    pipeline.store.update(
+        "测试群",
+        "2026-08-18",
+        status=READY_TO_SEND,
+        image_fallback_level=fallback_level,
+        image_variant=image_variant,
+        image_fallback_reason="PROMPT_FAILED",
+    )
+
+    result = pipeline.send_due(now=datetime(2026, 8, 18, 9, 0, 0))[0]
+    run = pipeline.store.load_run("测试群", "2026-08-18")
+
+    assert result["status"] == "failed"
+    assert result["error_type"] == IMAGE_FALLBACK_NOT_SENDABLE
+    assert run["status"] == FAILED
+    assert run["failed_stage"] == "image"
+    assert run["send_hold_reason"] == IMAGE_FALLBACK_NOT_SENDABLE
+    assert pipeline.sender.text_calls == []
+    assert pipeline.sender.image_calls == []
+
+
+def test_run_store_send_claim_rejects_diagnostic_fallback(tmp_path):
+    store = RunStore(tmp_path / "output")
+    store.save_run(
+        "测试群",
+        "2026-08-18",
+        {
+            **_ready_run_contract(image_enabled=True),
+            "image_fallback_level": 3,
+            "image_variant": "pillow",
+        },
+    )
+
+    claim_id, _, reason = store.claim_send(
+        "测试群",
+        "2026-08-18",
+        now=datetime(2026, 8, 18, 8, 30, 0),
+        lease_seconds=60,
+    )
+
+    assert claim_id is None
+    assert reason == IMAGE_FALLBACK_NOT_SENDABLE
+
+
+def test_force_send_rejects_diagnostic_fallback_before_late_send_checks(tmp_path):
+    pipeline, group = _make_pipeline(tmp_path)
+    pipeline.generate_all(run_date="2026-08-18")
+    pipeline.store.update(
+        "测试群",
+        "2026-08-18",
+        status=READY_TO_SEND,
+        image_fallback_level=3,
+        image_variant="pillow",
+    )
+
+    result = pipeline.force_send(group.id, "2026-08-18")
+
+    assert result["status"] == "failed"
+    assert result["error_type"] == IMAGE_FALLBACK_NOT_SENDABLE
+    assert pipeline.sender.text_calls == []
+    assert pipeline.sender.image_calls == []
+
+
+def test_send_preflight_rejects_diagnostic_fallback_without_sender_calls(
+    tmp_path, monkeypatch
+):
+    pipeline, group = _make_pipeline(tmp_path)
+    pipeline.generate_all(run_date="2026-08-18")
+    now = datetime(2026, 8, 18, 8, 30, 0)
+    claim_id, _, reason = pipeline.store.claim_send(
+        "测试群",
+        "2026-08-18",
+        now=now,
+        lease_seconds=60,
+    )
+    assert claim_id and reason == "claimed"
+    diagnostic_run = pipeline.store.update(
+        "测试群",
+        "2026-08-18",
+        image_fallback_level=3,
+        image_variant="pillow",
+        image_fallback_reason="IMAGE_GENERATION_FAILED",
+    )
+    monkeypatch.setattr(
+        pipeline.store,
+        "claim_send",
+        lambda *_args, **_kwargs: (claim_id, diagnostic_run, "claimed"),
+    )
+
+    result = pipeline._send_one(
+        group,
+        "测试群",
+        diagnostic_run,
+        "2026-08-18",
+        now,
+    )
+    run = pipeline.store.load_run("测试群", "2026-08-18")
+
+    assert result["status"] == "failed"
+    assert result["error_type"] == IMAGE_FALLBACK_NOT_SENDABLE
+    assert run["status"] == FAILED
+    assert run["failed_stage"] == "image"
+    assert run["send_hold_reason"] == IMAGE_FALLBACK_NOT_SENDABLE
+    assert pipeline.sender.text_calls == []
+    assert pipeline.sender.image_calls == []
+
+
 def test_explicit_text_failures_use_backoff_and_stop_after_send_retry_budget(tmp_path):
     sender = FakeSender(fail_text=True)
     pipeline, _ = _make_pipeline(tmp_path, sender=sender, image_enabled=False)
diff --git a/tests/test_v2_prompt_builder.py b/tests/test_v2_prompt_builder.py
index 75531a1..14630c5 100644
--- a/tests/test_v2_prompt_builder.py
+++ b/tests/test_v2_prompt_builder.py
@@ -167,6 +167,90 @@ def _chat(self, messages: list[dict], **kwargs) -> str:
         return json.dumps({"events": []}, ensure_ascii=False)
 
 
+class DuplicateIdentitySummaryProvider(FakeSummaryProvider):
+    """复现三个候选人物条目实际只对应两个 sender_id 的 Eason 结构。"""
+
+    def _chat(self, messages: list[dict], **kwargs) -> str:
+        user = messages[1]["content"]
+        if '"copy_version":"fixed-chat-comic-v2"' in user:
+            self.calls.append((messages[0]["content"], user))
+            source = _raw_json_from_user(user)
+            topic = source["topics"][0]
+            bindings = topic["speaker_bindings"]
+            return json.dumps(
+                {
+                    "title": "票房火箭",
+                    "subtitle": "票房讨论接成火箭",
+                    "panels": [
+                        {
+                            "topic_id": topic["topic_id"],
+                            "title": "票房火箭",
+                            "event_summary": topic["source_summary"],
+                            "composition": "两位群友站在火箭走势图两侧接话",
+                            "participants": [
+                                {
+                                    "message_id": binding["message_id"],
+                                    "action": "站在走势图旁接话",
+                                    "quote": binding["text"],
+                                }
+                                for binding in bindings
+                            ],
+                            "visual_gag": topic["source_visual_gag"],
+                            "fact_line": topic["source_summary"],
+                        }
+                    ],
+                    "footer_summary": "票房讨论接成火箭",
+                },
+                ensure_ascii=False,
+            )
+        if "可选分镜骨架" in user and "已入选主题" in user:
+            self.calls.append((messages[0]["content"], user))
+            topic_id = re.search(r'"topic_id":"(topic-[^"]+)"', user).group(1)
+            return json.dumps(
+                {
+                    "layout_id": "hero_with_insets",
+                    "structure_mode": "hero_rhythm",
+                    "featured_topic_ids": [topic_id],
+                    "topic_order": [topic_id],
+                    "panel_beats": [
+                        {"topic_id": topic_id, "shots": ["dialogue", "reaction"]}
+                    ],
+                    "comedy_device": "接话反差",
+                    "layout_reason": "单一真实话题使用头条节奏",
+                },
+                ensure_ascii=False,
+            )
+        if '"candidates"' in user:
+            self.calls.append((messages[0]["content"], user))
+            ids = list(
+                dict.fromkeys(
+                    re.findall(r"消息ID:([^\]#]+)(?:#片段\d+/\d+)?\]", user)
+                )
+            )
+            return json.dumps(
+                {
+                    "candidates": [
+                        {
+                            "topic_id": "topic-01",
+                            "title": "票房火箭",
+                            "summary": "票房讨论中，两位群友把走势接成火箭笑点。",
+                            "start_time": "2026-08-17 10:30",
+                            "end_time": "2026-08-17 10:30",
+                            "message_ids": ids[:3],
+                            "comedy_score": 35,
+                            "group_recognition_score": 18,
+                            "visual_score": 18,
+                            "comedy_angle": "真实接话形成反差",
+                            "visual_gag": "把票房走势画成火箭",
+                            "score_reason": "同一人连续发言后由另一人接话",
+                        }
+                    ]
+                },
+                ensure_ascii=False,
+            )
+        return super()._chat(messages, **kwargs)
+
+
 def _builder(provider: FakeSummaryProvider | None = None, tmp_path=None):
     return DeepSeekImagePromptBuilder(
         provider=provider or FakeSummaryProvider(),
@@ -208,6 +292,27 @@ def test_build_renders_only_fixed_sections_and_real_multi_person_dialogue():
         assert hidden not in output.prompt
 
 
+def test_builder_repairs_duplicate_sender_identity_and_records_prompt_meta():
+    output = _builder(DuplicateIdentitySummaryProvider()).build(_input())
+
+    assert output.success
+    assert validate_fixed_prompt_contract(output.prompt, expected_panel_count=1) == 1
+    assert output.meta["political_keyword_policy_version"] == "political-keywords-v1"
+    assert output.meta["poster_visible_participant_count"] == 2
+    assert output.meta["poster_participant_repairs"] == [
+        {
+            "panel_index": 1,
+            "topic_id": "topic-01",
+            "deduplicated_count": 1,
+            "trimmed_count": 0,
+            "filled_count": 0,
+            "final_participant_count": 2,
+        }
+    ]
+    assert output.prompt.count("人物旁清晰标注“张三”") == 1
+    assert output.prompt.count("人物旁清晰标注“李四”") == 1
+
+
 def test_summary_and_prompt_phases_use_independent_provider_instances():
     summary = FakeSummaryProvider()
     prompt = FakeSummaryProvider()
@@ -462,6 +567,105 @@ def test_poster_name_is_derived_from_message_id_when_quotes_are_identical():
     ]
 
 
+def _participant_repair_fixture() -> tuple[dict, dict]:
+    source = {
+        "topics": [
+            {
+                "topic_id": "topic-repair",
+                "source_title": "同一身份连续接话",
+                "source_summary": "甲和乙参与讨论，甲连续说了两句。",
+                "source_visual_gag": "两人站在同一张桌边接话",
+                "participant_options": ["甲", "乙"],
+                "participant_min": 2,
+                "participant_max": 2,
+                "evidence_dialogue": [
+                    {"message_id": "m-a1", "sender_id": "WXID-A", "speaker": "甲", "text": "第一句"},
+                    {"message_id": "m-a2", "sender_id": "wxid-a", "speaker": "甲", "text": "第二句"},
+                    {"message_id": "m-b", "sender_id": "wxid-b", "speaker": "乙", "text": "收到"},
+                ],
+                "speaker_bindings": [
+                    {"message_id": "m-a1", "sender_id": "WXID-A", "speaker": "甲", "text": "第一句"},
+                    {"message_id": "m-a2", "sender_id": "wxid-a", "speaker": "甲", "text": "第二句"},
+                    {"message_id": "m-b", "sender_id": "wxid-b", "speaker": "乙", "text": "收到"},
+                ],
+                "shot_hints": ["对白"],
+            }
+        ]
+    }
+    payload = {
+        "title": "同一身份连续接话",
+        "subtitle": "甲和乙参与讨论",
+        "panels": [
+            {
+                "topic_id": "topic-repair",
+                "title": "同一身份连续接话",
+                "event_summary": "甲和乙参与讨论，甲连续说了两句。",
+                "composition": "两人站在同一张桌边接话",
+                "participants": [
+                    {"message_id": "m-a1", "action": "站在左侧发言", "quote": "第一句"},
+                    {"message_id": "m-a2", "action": "继续站在左侧", "quote": "第二句"},
+                    {"message_id": "m-b", "action": "站在右侧回应", "quote": "收到"},
+                ],
+                "visual_gag": "两人站在同一张桌边接话",
+                "fact_line": "甲和乙参与讨论，甲连续说了两句。",
+            }
+        ],
+        "footer_summary": "甲和乙参与讨论",
+    }
+    return source, payload
+
+
+def test_poster_deduplicates_three_bindings_to_two_real_sender_identities():
+    source, payload = _participant_repair_fixture()
+    repairs: list[dict] = []
+
+    copy = parse_poster_copy(
+        json.dumps(payload, ensure_ascii=False), source, repair_log=repairs
+    )
+
+    assert [(item.message_id, item.name) for item in copy.panels[0].participants] == [
+        ("m-a1", "甲"),
+        ("m-b", "乙"),
+    ]
+    assert repairs == [
+        {
+            "panel_index": 1,
+            "topic_id": "topic-repair",
+            "deduplicated_count": 1,
+            "trimmed_count": 0,
+            "filled_count": 0,
+            "final_participant_count": 2,
+        }
+    ]
+
+
+def test_poster_fills_missing_identity_from_same_topic_without_inventing_quote():
+    source, payload = _participant_repair_fixture()
+    payload["panels"][0]["participants"] = payload["panels"][0]["participants"][:1]
+    repairs: list[dict] = []
+
+    copy = parse_poster_copy(
+        json.dumps(payload, ensure_ascii=False), source, repair_log=repairs
+    )
+
+    added = copy.panels[0].participants[1]
+    assert (added.message_id, added.name, added.action, added.quote) == (
+        "m-b",
+        "乙",
+        "参与本话题讨论",
+        "",
+    )
+    assert repairs[0]["filled_count"] == 1
+
+
+def test_poster_repair_still_rejects_message_id_outside_topic_bindings():
+    source, payload = _participant_repair_fixture()
+    payload["panels"][0]["participants"][0]["message_id"] = "m-unauthorized"
+
+    with pytest.raises(PosterCopyError, match="未授权的消息ID"):
+        parse_poster_copy(json.dumps(payload, ensure_ascii=False), source)
+
+
 def test_poster_source_keeps_each_message_scoped_name_for_same_sender():
     selection = {
         "candidates": [
@@ -506,6 +710,8 @@ def test_poster_source_keeps_each_message_scoped_name_for_same_sender():
     source = build_poster_editor_source(selection, layout)
     bindings = source["topics"][0]["speaker_bindings"]
 
+    assert source["topics"][0]["participant_min"] == 1
+    assert source["topics"][0]["participant_max"] == 1
     assert [(item["message_id"], item["speaker"]) for item in bindings] == [
         ("m-old", "早些时候的名字"),
         ("m-new", "后来改的名字"),
@@ -542,10 +748,10 @@ def test_default_template_file_and_builtin_are_synchronized():
     assert file_body.count("【漫画分镜】") == 1
 
 
-def test_fixed_storyboard_contract_supports_two_five_and_seven_panels():
+def test_fixed_storyboard_contract_supports_one_two_five_and_seven_panels():
     output = _builder().build(_input())
     assert output.success
-    for panel_count in (2, 5, 7):
+    for panel_count in (1, 2, 5, 7):
         panels = "\n\n".join(
             f"【版面{index}】\n真实话题{index}"
             for index in range(1, panel_count + 1)
diff --git a/tests/test_v2_ui_router_contract.py b/tests/test_v2_ui_router_contract.py
index cedf392..477a48d 100644
--- a/tests/test_v2_ui_router_contract.py
+++ b/tests/test_v2_ui_router_contract.py
@@ -172,6 +172,67 @@ def ranking_json_path(self, _group_name, _run_date):
     assert result["cards"][0]["prompt_operation_id"] == "operation-123"
 
 
+def test_dashboard_preserves_sent_truth_and_exposes_diagnostic_image_failure(
+    tmp_path, monkeypatch
+) -> None:
+    group = SimpleNamespace(
+        id=23,
+        display_name="历史诊断图群",
+        wechat_group_name="历史诊断图群",
+        send_time="08:30",
+        schedule_rule="daily",
+        image_enabled=True,
+        wechat_send_enabled=True,
+        ranking_template="",
+        image_prompt_template="",
+    )
+    image_path = Path(tmp_path) / "daily_image.png"
+    image_path.write_bytes(b"diagnostic-png-placeholder")
+
+    class FakeStore:
+        root = Path(tmp_path)
+
+        def load_run(self, _group_name, _run_date):
+            return {
+                "status": "SENT",
+                "sent_at": "2026-09-03T08:36:00+08:00",
+                "image_status": "success",
+                "image_fallback_level": 3,
+                "image_fallback_reason": "PROMPT_FAILED",
+                "image_variant": "pillow",
+                "prompt_original_error": "Prompt 连续校验失败",
+            }
+
+        def image_path(self, _group_name, _run_date):
+            return image_path
+
+        def ranking_json_path(self, _group_name, _run_date):
+            return Path(tmp_path) / "missing-ranking.json"
+
+    monkeypatch.setattr(v2_ui_read, "_store", lambda _settings: FakeStore())
+    monkeypatch.setattr(v2_ui_read.repo, "list_groups", lambda *_args, **_kwargs: [group])
+
+    result = v2_ui_read.dashboard(
+        session=object(),
+        settings=Settings(_env_file=None, output_dir=tmp_path),
+        run_date="2026-09-03",
+    )
+    card = result["cards"][0]
+    runtime_group = result["runtime"]["groups"][0]
+
+    assert result["counts"]["sent"] == 1
+    assert card["status"] == "SENT"
+    assert card["image_status"] == "failed"
+    assert card["image_fallback_level"] == 3
+    assert card["image_fallback_reason"] == "PROMPT_FAILED"
+    assert card["image_variant"] == "pillow"
+    assert card["image_delivery_eligible"] is False
+    assert card["error"] == "Prompt 连续校验失败"
+    assert runtime_group["run_status"] == "SENT"
+    assert runtime_group["image"]["status"] == "failed"
+    assert runtime_group["send"]["status"] == "success"
+
+
 def test_dashboard_accepts_run_date_and_returns_top_five_ranking_preview(
     tmp_path, monkeypatch
 ) -> None: