feat(quality-ledger): 生成成色与提示词版本落库

backend/packages/ai_engine/src/windup_ai_engine/impl/character_generator.py

@@ -27,7 +27,13 @@
     ProgressPort,
 )
 from windup_ai_engine.postprocess import align_bottom_center, frame_durations
-from windup_ai_engine.slicing import dead_frame_indices, loop_seam, motion_scale
+from windup_ai_engine.prompt import PROMPT_VERSION
+from windup_ai_engine.slicing import (
+    dead_frame_indices,
+    loop_seam,
+    motion_scale,
+    subject_blobs,
+)
 from windup_ai_engine.strategy.base import (
     ROUTE_MATRIX,
     DerivationStrategy,
@@ -149,6 +155,7 @@ def generate(
             frames=[_png(im) for im in aligned],
             durations=frame_durations(action.action.value, len(aligned)),
             quality=quality,
+            prompt_version=PROMPT_VERSION,
         )
 
     def _assess(self, frames: list[Image.Image], action: ActionSpec) -> ActionQuality:
@@ -165,6 +172,7 @@ def _assess(self, frames: list[Image.Image], action: ActionSpec) -> ActionQualit
             motion_scale=motion_scale(frames),
             dead_frames=dead_frame_indices(frames),
             loop_seam=loop_seam(frames) if is_cyclic(action) else None,
+            subject_blobs=subject_blobs(frames),
         )
 
     def _lastmile(

backend/packages/ai_engine/src/windup_ai_engine/ports/__init__.py

@@ -63,7 +63,7 @@ class ActionQuality:
     一段**每帧都一样**的 walk 和一段步态干净的 walk,帧数、时长、fps 完全相同,
     调用方分辨不出 —— 本仓吃过四次的正是这类"看起来成功的错结果"。
 
-    三个字段各自不可由其他两个推导(下面逐条说明必要性)。刻意**没有**的字段:
+    四个字段各自不可由其余三个推导(下面逐条说明必要性)。刻意**没有**的字段:
       - 糊帧率(``slicing.quality.blur_ratio``):2026-08-05 实测 6 段真 i2v
         **没有一帧糊帧**,加进来是个恒等于 1 的常数,上层拿它做不了任何决定。
         真出现糊帧再加,那时才有阈值可依。
@@ -103,10 +103,20 @@ class ActionQuality:
     0.0 会被读成"完美闭环",正是本仓忌讳的"貌似合理的默认值"。
     """
 
+    subject_blobs: tuple[int, ...]
+    """逐帧的"够大"连通块数(alpha>128,4-邻域;见 ``slicing.quality.subject_blobs``)。
+
+    上层拿它做的决定:全程恒为 2(或更多)→ 母版/提示词让引擎画出了第二个角色,
+    提示重试或换母版,这类病 ``motion_scale``/``dead_frames``/``loop_seam`` 全部
+    测不出——三者都只看"帧与帧之间变了多少",一个稳定存在的额外主体不影响它们
+    任何一个读数。只在中段冒出的 2 是另一类病(挥动的肢体/道具被抠断),修法是
+    调抠图阈值而非换母版,与前者必须分开看,故给逐帧序列而非一个均值。
+    """
+
 
 @dataclass
 class GeneratedAction:
-    """一个动作的生成产物:对齐后的原地序列帧 + 逐帧时长 + 成色。
+    """一个动作的生成产物:对齐后的原地序列帧 + 逐帧时长 + 成色 + 提示词版本。
 
     frames / durations **等长**;server 侧把每帧上传对象存储得 URL,组成
     ``CharacterActionOutput.frames[{index, image_url, duration_ms}]`` 回填 character_data。
@@ -122,6 +132,9 @@ class GeneratedAction:
     # 给个 None 缺省的话,漏测与"测出来没问题"在调用方看来一模一样,而这个出参的
     # 全部意义就是把这两者分开。
     quality: ActionQuality = field(kw_only=True)
+    # 同一条理由:不给缺省,逼调用方显式带出当下的 ``windup_ai_engine.prompt.PROMPT_VERSION``。
+    # 改了提示词模板而没带上新版本号,这批产出与改动前的产出在账本里就再也分不清。
+    prompt_version: str = field(kw_only=True)
 
 
 # ---- ai_engine 暴露给 server(server 调用的唯一入口)----

backend/packages/ai_engine/src/windup_ai_engine/prompt/__init__.py

@@ -5,6 +5,11 @@
 from .jump import JUMP_PHASES, build_jump_prompt
 from .walk import build_walk_prompt
 
+# 改动本包任何一个 build_*_prompt 的输出(包括 prompts/*.md 模板)都必须连带把这个
+# 常量加一:落库的 GeneratedAction.prompt_version 就靠它,分不清新旧模板的产出，
+# 改完提示词也没法与改前的成色对比。
+PROMPT_VERSION = "v1"
+
 __all__ = [
     "build_walk_prompt",
     "JUMP_PHASES",
@@ -13,4 +18,5 @@
     "build_attack_prompt",
     "build_custom_prompt",
     "MAX_ACTION_CHARS",
+    "PROMPT_VERSION",
 ]

backend/packages/ai_engine/src/windup_ai_engine/slicing/__init__.py

@@ -4,7 +4,8 @@
 loop),一次性动作裁动作区间。像素化 / 对齐 / 打包在 :mod:`..postprocess`。
 
 :mod:`.quality` 原本纯做诊断,现在还兼一份出参职责:交付帧的成色读数
-(``motion_scale`` / ``dead_frame_indices`` / ``loop_seam``)汇成 ``ports.ActionQuality``。
+(``motion_scale`` / ``dead_frame_indices`` / ``loop_seam`` / ``subject_blobs``)汇成
+``ports.ActionQuality``。
 注意它**仍然不参与选帧** —— 那条消融结论没变,见 :func:`.loop.pick_cycle`。
 """
 
@@ -17,17 +18,18 @@
     pick_oneshot,
     split_jump_phases,
 )
-from .quality import dead_frame_indices, loop_seam, motion_scale
+from .quality import dead_frame_indices, loop_seam, motion_scale, subject_blobs
 
 __all__ = [
     "extract_frames_bytes",
     "extract_all_frames_bytes",
     "find_period",
     "pick_cycle",
-    # 交付成色的三个读数(汇成 ports.ActionQuality;其余 quality.* 仍是内部诊断)
+    # 交付成色的四个读数(汇成 ports.ActionQuality;其余 quality.* 仍是内部诊断)
     "dead_frame_indices",
     "loop_seam",
     "motion_scale",
+    "subject_blobs",
     "find_motion_span",
     "first_action_end",
     "foot_line_series",

backend/packages/ai_engine/src/windup_ai_engine/slicing/quality.py

@@ -13,7 +13,7 @@
 from ._frames import gray as _gray
 
 __all__ = ["active_span", "blur_ratio", "dead_frame_indices", "dead_frame_mask",
-           "frame_deltas", "loop_seam", "motion_scale"]
+           "frame_deltas", "loop_seam", "motion_scale", "subject_blobs"]
 
 
 def frame_deltas(frames) -> np.ndarray:
@@ -116,6 +116,79 @@ def active_span(frames, floor: float = 0.25, min_run: int = 3) -> tuple[int, int
     return s, e
 
 
+def _row_runs(row: np.ndarray) -> list[tuple[int, int]]:
+    """一行内为真的连续区间 ``[start, end)``。差分找边沿,不逐像素判断。"""
+    padded = np.concatenate(([False], row, [False]))
+    edges = np.flatnonzero(padded[1:] != padded[:-1])
+    return [(int(edges[i]), int(edges[i + 1])) for i in range(0, len(edges), 2)]
+
+
+def _count_blobs(mask: np.ndarray, min_area_ratio: float) -> int:
+    """4-连通域计数(游程并查集,不依赖 scipy)。
+
+    按行取真值游程,相邻两行的游程只要列区间有重叠就判定竖直相连——同一游程内的像素
+    horizontal 方向本就连续,故这一条合并规则等价于逐像素 4-邻域标记,但只需在"游程"
+    这个粗粒度上做并查集,免去逐像素扫描。
+    """
+    parent: list[int] = []
+
+    def find(a: int) -> int:
+        while parent[a] != a:
+            parent[a] = parent[parent[a]]
+            a = parent[a]
+        return a
+
+    def union(a: int, b: int) -> None:
+        ra, rb = find(a), find(b)
+        if ra != rb:
+            parent[max(ra, rb)] = min(ra, rb)
+
+    prev_runs: list[tuple[int, int, int]] = []  # (start, end, label)
+    areas: dict[int, int] = {}
+    for row in mask:
+        cur_runs = []
+        for start, end in _row_runs(row):
+            label = len(parent)
+            parent.append(label)
+            areas[label] = end - start
+            for ps, pe, plabel in prev_runs:
+                if ps < end and start < pe:      # 列区间重叠 → 与上一行竖直相连
+                    union(label, plabel)
+            cur_runs.append((start, end, label))
+        prev_runs = cur_runs
+
+    if not areas:
+        return 0
+    totals: dict[int, int] = {}
+    for label, area in areas.items():
+        root = find(label)
+        totals[root] = totals.get(root, 0) + area
+    max_area = max(totals.values())
+    # 阈值语义:比全帧最大块小的块,只有达到该块 min_area_ratio 的面积才算数——
+    # 目的只是滤掉"主体+噪点"里的噪点(面积占比通常 <1%),不是要卡死一个精确的
+    # "第二主体"下限;真出现被这条误伤/漏判的样本,回头拿那批样本重新校这个数。
+    return sum(1 for a in totals.values() if a >= min_area_ratio * max_area)
+
+
+def subject_blobs(frames, *, min_area_ratio: float = 0.15) -> tuple[int, ...]:
+    """逐帧统计画面里有几个"够大"的连通块(alpha>128,4-邻域)。
+
+    **返回逐帧计数,不是均值** —— 与 :func:`dead_frame_indices` 给下标同一个理由:
+    分布形态对应不同的病,修法不同。全程恒为 2 = 真出了第二个角色(母版/提示词问题);
+    只有中段冒出 2 = 挥动的手臂或手持物被抠断成两截(抠图/对齐问题)。压成一个均值,
+    这两种病看起来一样。
+
+    单人持长条物(如剑)只要与身体像素相连,就与身体同属一个连通块,不会被数成 2 ——
+    这条计数器的价值就在于分得清"真第二主体"与"伸出去的长条肢体/道具"，
+    见校准测试 ``test_subject_blobs.py``。
+    """
+    out = []
+    for f in frames:
+        alpha = np.asarray(f.convert("RGBA"))[:, :, 3]
+        out.append(_count_blobs(alpha > 128, min_area_ratio))
+    return tuple(out)
+
+
 def blur_ratio(frames, ps: int = 32) -> np.ndarray:
     """逐帧「静止区清晰度 / 前后帧同区清晰度」。<1 = 这帧自己糊了，与动作快慢无关。"""
     def _pm(a):

backend/packages/app/src/windup_app/server/orchestrator/executor.py

@@ -13,6 +13,7 @@
 
 from __future__ import annotations
 
+import dataclasses
 import logging
 import threading
 from collections.abc import Callable
@@ -292,7 +293,15 @@ def _produce_action(self, input: CharacterActionInput, cons: ProjectConstraints)
              "duration_ms": dur}
             for i, (png, dur) in enumerate(zip(generated.frames, generated.durations))
         ]
-        return {"type": "character_action", "action_type": input.action_type.value, "frames": frames}
+        # quality / prompt_version 只落库记账,不在此处据成色改判决:交付/重试是产品
+        # 决策,该由读这本账的下游按阈值决定,任务状态仍只反映"生成流程是否跑完"。
+        return {
+            "type": "character_action",
+            "action_type": input.action_type.value,
+            "frames": frames,
+            "quality": dataclasses.asdict(generated.quality),
+            "prompt_version": generated.prompt_version,
+        }
 
     def _get_generator(self, video_model: str | None = None) -> CharacterGeneratorPort:
         """懒装配 CharacterGenerator,按模型名分桶。

backend/packages/app/src/windup_app/server/orchestrator/model.py

@@ -111,11 +111,17 @@ class CharacterActionOutput:
     前端拿到后写入 ``character_data.outfits[].actions[]``：
     ``action_type`` → ``CharacterAction.type``，
     ``frames`` → ``CharacterAction.frames[]``。
+
+    ``quality`` / ``prompt_version`` 是引擎产出成色的账本(``ai_engine.ports.ActionQuality``
+    的原样转录 + 提示词版本),不参与前端回填、只落库供后续对比——本层不据此判成败,
+    见 executor 里"只记账不判决"的说明。
     """
 
     type: str = "character_action"
     action_type: str = ""
     frames: list[CharacterActionFrame] = field(default_factory=list)
+    quality: dict | None = None
+    prompt_version: str | None = None
 
 
 # -- 任务记录 ------------------------------------------------------------

backend/packages/app/src/windup_app/server/orchestrator/task_repo.py

@@ -227,5 +227,7 @@ def _deserialize_result(
             type=raw.get("type", "character_action"),
             action_type=raw.get("action_type", ""),
             frames=frames,
+            quality=raw.get("quality"),
+            prompt_version=raw.get("prompt_version"),
         )
     return None

backend/tests/test_generation_orchestration.py

@@ -119,6 +119,46 @@ def _upload(png: bytes) -> str:
         assert frame.duration_ms is not None
 
 
+def test_quality_and_prompt_version_reach_the_persisted_result(session_factory, monkeypatch):
+    """成色从生成到落库这条链路必须闭合,否则线上永远答不出"改完提示词到底有没有
+    变好"(见 executor 里"只记账不判决"的说明)。
+
+    落库后必须能读到 motion_scale / dead_frames / subject_blobs 三个成色读数,
+    以及 prompt_version。
+    """
+    service = AiGenerationService()
+    executor = ActionTaskExecutor(
+        generator=_real_offline_generator(monkeypatch),
+        upload=lambda png: "https://cdn.example.com/f.png",
+        fetch_master=lambda _input: _tiny_png(),
+        session_factory=session_factory,
+    )
+    action_input = CharacterActionInput(
+        character_id=1, action_type=ActionType.WALK, num_frames=6,
+    )
+    with session_factory() as s:
+        task = service.generate_character_action(s, user_id=1, input=action_input)
+        s.commit()
+        task_id = task.id
+
+    executor.run_action_task(task_id, action_input)
+
+    with session_factory() as s:
+        done = service.get_task(s, project_id=1, task_id=task_id)
+    assert done.status is TaskStatus.COMPLETED
+    quality = done.result.quality
+    assert quality is not None, "quality 被丢在了 executor 到落库之间的某一步"
+    assert isinstance(quality["motion_scale"], float)
+    assert "dead_frames" in quality
+    assert "subject_blobs" in quality and len(quality["subject_blobs"]) == len(
+        done.result.frames
+    )
+    assert done.result.prompt_version, "prompt_version 没有随成色一起落库"
+
+    # 本步只记账,不判决:即便 motion_scale 恰好是 0 这种"典型坏产出"信号,
+    # 任务仍然是 COMPLETED —— 交付/重试是产品决策,不该由这一步替调用方做。
+
+
 def _png_of(w: int, h: int) -> bytes:
     """指定尺寸的一张带主体的 PNG。"""
     img = Image.new("RGBA", (w, h), (0, 0, 0, 0))
@@ -160,7 +200,10 @@ def generate(self, card, action, master, progress, canvas=None):
         return GeneratedAction(
             frames=[_png_of(*size)],
             durations=[100],
-            quality=ActionQuality(motion_scale=1.0, dead_frames=[], loop_seam=None),
+            quality=ActionQuality(
+                motion_scale=1.0, dead_frames=[], loop_seam=None, subject_blobs=(1,)
+            ),
+            prompt_version="test-v0",
         )
 
 

backend/tests/test_master_check_and_quality.py

@@ -232,5 +232,6 @@ def test_loop_seam_measures_the_gap_between_last_and_first():
 def test_quality_fields_are_independent():
     """三个字段互不可推导：全同帧的 motion_scale=0 而 dead_frames 为空，
     两者若能互推，这一组断言不可能同时成立。"""
-    q = ActionQuality(motion_scale=0.0, dead_frames=(), loop_seam=None)
+    q = ActionQuality(motion_scale=0.0, dead_frames=(), loop_seam=None, subject_blobs=(1,))
     assert q.motion_scale == 0.0 and q.dead_frames == () and q.loop_seam is None
+    assert q.subject_blobs == (1,)