"""p11_video_render.py — Phase 11: video render with parallel shot fan-out. Reference port of the Node.js Step 10 video half + Step 11 video render. Per CONTEXT D-35-04 this module is PURE ORCHESTRATION — no LLM calls, no prompt templates, no business logic. All render work is delegated to the ``visual_executor`` (animator sub-step) and ``audio_pipeline`` (lip_sync sub-step) movie-expert skills via ``delegate_task``. This is the ONLY phase that uses ``parallel_shots`` (D-36-08). The run() signature extends the standard 5-arg Phase 35 contract with an extra keyword-only ``parallel_shots: int = 4`` parameter. The phase fans out shot-level ``delegate_task`` calls concurrently via ``concurrent.futures.ThreadPoolExecutor(max_workers=parallel_shots)``, then aggregates results into single consolidated asset-bus writes. Behavioral contract (mirrors ``lib/phases/index.js`` video-render handler but ports the parallel shot dispatch natively in Python per D-36-08). Per-shot work (CF-36-03, Pattern 3): visual_executor animator + audio_pipeline lip_sync collaborate in a SINGLE delegate_task call per shot — the subagent orchestrates the 2-expert collaboration internally for that shot. Inputs (asset bus READ): - ``shot-list`` — output of p09 (one entry per shot) - ``scene-images`` — output of p07 (5-view per scene) - ``character-assets`` — output of p04 (L1-L4 character asset manifest) - ``voice-clips`` — output of p10 (TTS audio per shot — drives the ``ttsAudio`` multipart field injection into the H3 unified pipeline API) Outputs (asset bus WRITE): - ``video-clips`` — aggregated rendered video clips (one entry per shot) - ``lip-sync-reports`` — aggregated lip-sync alignment reports per shot Gate triggered (when ``trigger_gate`` is provided): - Gate 7 ``render-preview`` — operator confirms video clips meet quality bar (per references/review-gates.md). """ from __future__ import annotations import json import logging import os import re import subprocess import sys import time from concurrent.futures import ThreadPoolExecutor from datetime import datetime, timezone from pathlib import Path from typing import Any, Callable from pipeline.structured_output import parse_delegate_output as _parse_expert_output # json_repair engine from pipeline.prompt_ledger import get_ledger from ._gate_guard import raise_if_rejected from ._prompt_template import GoalBuilder from ._video_prompt import get_video_prompt as _get_vp, set_video_prompt as _set_vp from ..ratelimit import glm_concurrent, MAX_CONCURRENT_GL from plugins.kais_aigc.video_engine import ( # P1-2: in-process H3 render engine-adapter VideoEngineError, select_video_engine, ) # SHARED-01 / SHARED-02 (Phase 10-01): shared/ resource layer lives at the # skill root, NOT under pipeline/. Resolve the skill root from this file's # location (pipeline/phases/p11_video_render.py → skill root 2 levels up) # and insert it on sys.path so `from shared.resource_loader import ...` works # without depending on the runner's sys.path setup. _SKILL_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")) if _SKILL_ROOT not in sys.path: sys.path.insert(0, _SKILL_ROOT) from shared.resource_loader import load_shared_yaml # noqa: E402 logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # # Form-factor-aware render arguments (FORM-01 v14.0) # --------------------------------------------------------------------------- # # P11 builds the H3 API call hint with form-factor-correct resolution. H3 is # the ONLY local video engine (2026-08-05) — no LTX fallback path. Resolution # constraint: 32-multiples (portrait=768×1344 / landscape=1344×768). # --------------------------------------------------------------------------- # # H3 提示词增强 (video-engine-adapter skill, 纯 Python 零 KAP 依赖) # --------------------------------------------------------------------------- # # 裸 prompt 直进 H3 是历史缺陷:短需求缺乏 H3 协议要求的 [Shot N] 分镜、 # 参考标签、对白 结构。这里在构造 H3 调用前先增强 prompt。 # # 实现:直接 import video-engine-adapter skill 的 h3_prompt_enhancer 模块, # 不依赖 KAP HTTP 服务,不需要 KAP 持有 LLM key。 # - 本地模板模式(默认):纯规则拼装协议骨架,零 LLM 调用,零成本。 # - API增强模式:通过 Hermes hermes_llm 工具调系统提示词(由 agent 执行时 # 走 hermes 工具链,不走 KAP)。 # # 降级纪律:增强是锦上添花,不是渲染前置条件。任何异常一律回落原始 prompt。 # 开关:shot 级 h3_enhance > episode_config 级 h3_enhance > 默认 true。 # --------------------------------------------------------------------------- # video-engine-adapter skill 路径 (kais-hermes-skills 同级目录)。 _ADAPTER_SKILL_PATH = Path(__file__).resolve().parents[3] / "movie-experts" / "video-engine-adapter" / "scripts" if str(_ADAPTER_SKILL_PATH) not in sys.path: sys.path.insert(0, str(_ADAPTER_SKILL_PATH)) try: from h3_prompt_enhancer import ( # type: ignore[import-not-found] H3_MODE_TO_TASK as _H3_MODE_TO_TASK, build_local_template as _h3_build_local_template, ) _H3_ENHANCER_AVAILABLE = True except ImportError: _H3_ENHANCER_AVAILABLE = False _H3_MODE_TO_TASK = {"ref2va": "全参考模式(Reference to Video)", "i2va": "首帧图生视频(I2VA)", "t2va": "文生视频(T2VA)"} def _h3_build_local_template(**_kwargs: object) -> str: # type: ignore[no-redef] raise RuntimeError("h3_prompt_enhancer not available") # --------------------------------------------------------------------------- # # Hell Grind prompt anchors — identity locks (P0-2, 2026-08-13) # --------------------------------------------------------------------------- # # Top-tier AI films (Higgsfield "Hell Grind") anchor every character with an # explicit visual lock declaration so the model treats identity as a hard # constraint. KMC's identity mechanism is the P04 turnaround reference sheet; # these functions emit a textual ``[IDENTITY LOCKS]`` preamble that (a) uses # any structured visual meta already present (identity_lock / visual_identity # from P04 character_bible) and (b) otherwise anchors identity to the # turnaround reference path. Downward compatible + never blocks: returns "" # when the shot has no lockable character info. def _dispatch_with_attempt_redundancy( dispatch_fn: Callable[[int, dict], dict], idx: int, shot: dict, *, pre: int, phase_id: str, meta_out: list, ) -> dict: """(27-03) per-shot dispatch with attempt-redundancy (first-success wins). pre>1 = 失败才补渲下一变体（checkpoint 缓存令补渲近似零成本），而非 并行 N× 渲染（GPU 成本线性放大 + 缓存去重 = 最差组合）。每 shot 的尝试 记录 append 进 meta_out（成功/失败都留痕）。 """ _last_exc: Exception | None = None for variant in range(1, max(1, pre) + 1): try: result = dispatch_fn(idx, shot) meta_out.append({ "shot_index": idx, "shot_id": shot.get("shot_id") if isinstance(shot, dict) else None, "requested_pre": pre, "variant_attempted": variant, "selected_variant": variant, "succeeded": True, }) return result except Exception as exc: # noqa: BLE001 — 下一变体接管 _last_exc = exc logger.warning( "%s: shot %s variant %d/%d failed — %s", phase_id, shot.get("shot_id", f"shot_{idx}") if isinstance(shot, dict) else idx, variant, pre, exc, ) meta_out.append({ "shot_index": idx, "shot_id": shot.get("shot_id") if isinstance(shot, dict) else None, "requested_pre": pre, "variant_attempted": max(1, pre), "selected_variant": None, "succeeded": False, }) raise _last_exc # type: ignore[misc] def _lock_str(val: object) -> str | None: """Return a non-empty stripped string, or None.""" if isinstance(val, str) and val.strip(): return val.strip() return None def _derive_lock_from_meta(meta: dict) -> str | None: """Derive a 'Lock: ...' anchor from a character meta dict. Priority (first non-empty wins): 1. explicit ``identity_lock`` (forward-compatible — P04/P09 may emit it) 2. structured ``visual_identity`` (P04 character_bible shape) — harvest head/eyes/mouth/skin_texture/body_type + distinguishing_features 3. top-level distinguishing_features / clothing / accessories 4. free-text visual_description / appearance / description / l1_anchor Returns None when nothing lockable is present. """ if not isinstance(meta, dict) or not meta: return None explicit = _lock_str(meta.get("identity_lock")) if explicit: return explicit parts: list[str] = [] vi = meta.get("visual_identity") if isinstance(vi, dict): for k in ("body_type", "head", "eyes", "mouth", "skin_texture", "size", "hair"): v = _lock_str(vi.get(k)) if v: parts.append(f"{k}: {v}") df = vi.get("distinguishing_features") if isinstance(df, list): feats = [_lock_str(x) for x in df if _lock_str(x)] if feats: parts.append("distinguishing: " + ", ".join(feats)) for k in ("distinguishing_features", "clothing", "accessories"): v = meta.get(k) if isinstance(v, list): feats = [_lock_str(x) for x in v if _lock_str(x)] if feats: parts.append(f"{k}: " + ", ".join(feats)) elif _lock_str(v): parts.append(f"{k}: {_lock_str(v)}") if parts: return "Lock: " + "; ".join(parts) for k in ("visual_description", "appearance", "description", "l1_anchor"): v = meta.get(k) if _lock_str(v): return f"Lock: {_lock_str(v)}" if isinstance(v, dict): # l1_anchor is a multi-view text dict txt = ", ".join(_lock_str(x) for x in v.values() if _lock_str(x)) if txt: return f"Lock: {txt}" return None def _inject_identity_locks(shot: dict, character_assets: dict | None) -> str: """Build a ``[IDENTITY LOCKS]`` prompt block for the shot's characters. Returns "" (empty) when the shot has no lockable character info — never raises, never blocks rendering. Sourced from ``shot.character_refs`` joined with ``character-assets.l1_anchors`` (matched by name) for structured meta. """ if not isinstance(shot, dict): return "" refs = shot.get("character_refs") if not isinstance(refs, list) or not refs: return "" ca = character_assets if isinstance(character_assets, dict) else {} anchors_by_name: dict[str, dict] = {} for a in (ca.get("l1_anchors") or []): if isinstance(a, dict): nm = _lock_str(a.get("name")) if nm: anchors_by_name.setdefault(nm, a) lines: list[str] = [] for ref in refs: if not isinstance(ref, dict): continue name = _lock_str(ref.get("name")) or _lock_str(ref.get("character")) if not name: continue # Merge: anchor meta (richer) base, ref fields overlay (ref carries the # resolved, scene-correct turnaround_path that must win on conflict). meta = {**anchors_by_name.get(name, {}), **ref} lock = _derive_lock_from_meta(meta) if lock: lines.append(f"[{name}] {lock}") continue # Baseline: KMC's identity mechanism IS the turnaround reference sheet. # Anchor identity to it so the model treats the reference as a hard # identity constraint (Hell Grind "Lock" concept, KMC-native). tp = _lock_str(ref.get("turnaround_path")) or _lock_str(meta.get("turnaround_sheet")) if tp: lines.append( f"[{name}] Lock: face, body build, costume and species MUST match " f"the turnaround reference ({tp}) exactly — no identity drift" ) else: lines.append( f"[{name}] Lock: keep character identity consistent across the shot" ) if not lines: return "" return "[IDENTITY LOCKS]\n" + "\n".join(lines) + "\n\n" # --------------------------------------------------------------------------- # # Hell Grind prompt anchors — HARD LOCK constraints (P1-1, 2026-08-13) # --------------------------------------------------------------------------- # # Hell Grind: 9.6% of shots carry explicit hard constraints that pin a single # physical dimension (camera / emotion / action / speed / position) so the # model does not improvise. ``LOCK_TEMPLATES`` maps each lock kind to its # verbatim constraint text; ``_inject_hard_locks`` emits the templates whose # trigger is detectable on the shot. Selective (only when a clear trigger is # present), not universal — never blocks. LOCK_TEMPLATES: dict[str, str] = { "static": "⚠️ CAMERA LOCKED-OFF. No camera movement. Subtle handheld breathing only.", "dialogue": "⚠️ EMOTION LOCK: Character expression must match dialogue emotion. No ad-lib gestures.", "action": "⚠️ ACTION LOCK: Execute ONLY the described action sequence. No extra movements.", "slow_motion": "⚠️ REAL-TIME SPEED. No slow-motion, no speed ramps.", "multi_char": "⚠️ POSITION LOCK: Each character stays at their assigned screen position. No drifting.", } # Slow-motion intent markers (speed-ramp effect, NOT slow pacing). 升格 is the # Chinese cinematography term for the speed-ramp-up slow-motion effect. _SLOW_MOTION_KEYWORDS = ("慢动作", "slow motion", "slow-motion", "slow mo", "升格") def _inject_hard_locks(shot: dict, raw_prompt: str = "") -> str: """Build a ``[HARD LOCK]`` constraint block from detectable shot properties. Returns "" when no clear trigger is present — selective, never blocks. ``raw_prompt`` (the clean pre-anchor prompt) is used only for the slow-motion keyword scan so prepended anchors cannot self-trigger it. """ if not isinstance(shot, dict): return "" triggered: list[str] = [] # static camera -> locked-off if _lock_str(shot.get("camera_movement")) == "固定": triggered.append(LOCK_TEMPLATES["static"]) # spoken dialogue present -> emotion lock if _lock_str(shot.get("dialogue_text")): triggered.append(LOCK_TEMPLATES["dialogue"]) # multi-character shot -> position lock refs = shot.get("character_refs") if isinstance(refs, list) and len(refs) > 1: triggered.append(LOCK_TEMPLATES["multi_char"]) # complex multi-beat action (P0-1 action_timeline) -> action lock atl = shot.get("action_timeline") if isinstance(atl, list) and len(atl) >= 3: triggered.append(LOCK_TEMPLATES["action"]) # slow-motion intent (speed-ramp effect) -> real-time speed lock haystack = " ".join(filter(None, [ raw_prompt or "", _lock_str(shot.get("action")) or "", _lock_str(shot.get("intent")) or "", ])).lower() if any(kw in haystack for kw in _SLOW_MOTION_KEYWORDS): triggered.append(LOCK_TEMPLATES["slow_motion"]) if not triggered: return "" # de-dup while preserving order (a lock could theoretically trigger twice) seen: set[str] = set() uniq = [t for t in triggered if not (t in seen or seen.add(t))] return "[HARD LOCK]\n" + "\n".join(uniq) + "\n\n" # --------------------------------------------------------------------------- # # N2 (repro-experiment §6.3): SUBJECT STATE — 状态级外观快照 # --------------------------------------------------------------------------- # # 样本 2（job 2530884c）12.6% 字符是「状态级角色外观」（战损账本/武器化/义肢）。 # shot_intent.continuity.already_happened 承接「本镜头时刻角色身上已发生的事」， # character-bible 的 visual_identity 战损/武器化键承接外观快照——两侧此前都 # 无 prompt 消费者。插入位：identity locks 之后、hard locks 之前（外观状态 # 紧跟身份锁）。选择性注入 + 永不阻塞。 # bible visual_identity 中承载「当前外观状态」的子键（自由文本，N2 键集）。 _N2_BIBLE_STATE_KEYS = ("damage_state", "battle_damage", "weapons", "weaponized", "equipment", "gear", "prosthetics", "outfit_state") def _inject_subject_state(shot: dict, character_assets: dict | None) -> str: """[SUBJECT STATE] 块 — 逐角色渲染「本镜头时刻的外观状态快照」。 双源: (a) shot_intent.continuity.already_happened（本镜头开始前角色身上 已发生的事 → 外观必须体现）; (b) character-bible 对应角色 visual_identity 中的战损/武器化/装备状态文本。character_assets 为 _h3_enhance_prompt 的 现有参数——其中 character-bible 侧数据若由调用方注入（characters 列表） 则按 name 匹配; None/无命中 → 只渲染 already_happened 侧。 全缺 → ""（不注入）。永不抛错。 """ if not isinstance(shot, dict): return "" lines: list[str] = [] # (a) already_happened — continuity 四栏之一，此前无消费者。 si = shot.get("shot_intent") if isinstance(si, str): si = {} si = si if isinstance(si, dict) else {} cont = si.get("continuity") cont = cont if isinstance(cont, dict) else {} already = cont.get("already_happened") if isinstance(already, list): items = [str(x).strip() for x in already if str(x).strip()] if items: lines.append("本镜头开始前已发生（外观必须体现）：" + "；".join(items)) # (b) character-bible visual_identity 战损/武器化键。 # character_assets 兼容两种形状: {characters: [...]}（character-bible # 直通）或顶层 name → visual_identity 映射（tests/调用方注入）。 bible_chars: list[dict] = [] if isinstance(character_assets, dict): raw_chars = character_assets.get("characters") if isinstance(raw_chars, list): bible_chars = [c for c in raw_chars if isinstance(c, dict)] elif all(isinstance(v, dict) for v in character_assets.values()) and not any( k in character_assets for k in ("l1_anchors", "scene_costume_turnarounds") ): bible_chars = [ {"name": k, "visual_identity": v} for k, v in character_assets.items() ] if bible_chars: refs = shot.get("character_refs") names_in_shot = { _lock_str(r.get("name")) for r in refs if isinstance(r, dict) and _lock_str(r.get("name")) } if isinstance(refs, list) else set() for c in bible_chars: nm = _lock_str(c.get("name")) or _lock_str(c.get("character_id")) if not nm or (names_in_shot and nm not in names_in_shot): continue vi = c.get("visual_identity") if not isinstance(vi, dict): continue segs: list[str] = [] for k in _N2_BIBLE_STATE_KEYS: v = vi.get(k) if _lock_str(v): segs.append(_lock_str(v)) elif isinstance(v, list): segs.extend(x for x in (_lock_str(i) for i in v) if x) if segs: lines.append(f"{nm} 当前外观状态：" + "；".join(segs)) if not lines: return "" return "[SUBJECT STATE]\n" + "\n".join(lines) + "\n\n" # --------------------------------------------------------------------------- # # #4 (2026-08-15, hellgrind-iframe-optimization §3.3.2): ATMOSPHERE LOCK # --------------------------------------------------------------------------- # # HG 统一大气指令机制：同一段大气块逐字出现在场景的每个镜头 prompt 里 + # 显式一致性声明（意译会漂移）。P09 后处理已把同场景统一 atmosphere_lock # 盖章到每个 shot；这里逐字段渲染 + consistency_decl。与 shot 级 # color_guidance 冲突时锁赢（场景级硬约束）。选择性注入 + 永不阻塞。 _ATMOSPHERE_RENDER_KEYS = ("color_temp_k", "weather", "visibility", "palette_ratio", "emotion_palette", "lighting") _ATMOSPHERE_LABELS = { "color_temp_k": "色温", "weather": "天气", "visibility": "能见度", "palette_ratio": "调色板比例", "emotion_palette": "情绪组合", "lighting": "光照", } def _inject_atmosphere_lock(shot: dict) -> str: """[#4] [ATMOSPHERE LOCK] 块 — shot.atmosphere_lock 逐字段渲染。 所有 shot 都注入（不只 chained）——HG 的统一块覆盖场景内全部镜头。 consistency_decl 逐字渲染（显式一致性声明）。缺失/空 → ""。永不抛错。 """ if not isinstance(shot, dict): return "" lock = shot.get("atmosphere_lock") if not isinstance(lock, dict) or not lock: return "" lines: list[str] = [] for k in _ATMOSPHERE_RENDER_KEYS: v = lock.get(k) if v is None or (isinstance(v, str) and not v.strip()): continue label = _ATMOSPHERE_LABELS.get(k, k) lines.append(f"{label}：{v}") decl = lock.get("consistency_decl") if isinstance(decl, str) and decl.strip(): lines.append(decl.strip()) if not lines: return "" return "[ATMOSPHERE LOCK]\n" + "\n".join(lines) + "\n\n" # --------------------------------------------------------------------------- # # N3 (repro-experiment §6.3): crowd directives — 群众指令通道 # --------------------------------------------------------------------------- # # 实验 §5 D 行：群众行为指令（无尽援军/混乱冲锋/人墙呼吸/反应尖峰）此前无 # 字段无注入——D 模块最后 ~25% 缺口。P09 新 optional 字段 crowd_directives # （schema 已随 #8 落地）；这里做 H3 渲染侧消费。仅 prompt_richness == # director 或字段非空时渲染。 def _inject_crowd(shot: dict) -> str: """[N3] 群众指令块 — shot.crowd_directives 单行渲染。 仅当字段非空时注入（字段非空本身即渲染条件——「或字段非空」的另一支 在帧模板侧，语义同为：有内容就渲染，director 只是让 LLM 更倾向写它）。 缺失/空 → ""。永不抛错。 """ if not isinstance(shot, dict): return "" crowd = shot.get("crowd_directives") if not (isinstance(crowd, str) and crowd.strip()): return "" return f"[CROWD]\n{crowd.strip()}\n\n" # --------------------------------------------------------------------------- # # #6 (2026-08-15, hellgrind-iframe-optimization §4.3(a)): [FRAME-0 ANCHOR] # --------------------------------------------------------------------------- # # HG 无条件帧时的 prompt 级帧锚定（0s 起始锁定 + 逐秒时间线 + 状态延续声明） # 填补无条件帧 shot 的起始状态控制缺口。仅当 shot 无 iframe（iframe_mode # 缺失/single_ref）时注入；有条件帧（i2va/paired/chained）的 shot 不注入—— # 首帧图像已是更强的锚，避免 prompt 与图像打架。判断复用 _load_iframe_manifest # 已加载的 iframe_mode。anchor 与 video_prompt 动作描述冲突时以 anchor 的 # 已完成时态为准（prompt 构造统一时态）。 def _inject_frame_anchor(shot: dict, iframe_manifest_entry: dict | None = None) -> str: """[#6] [FRAME-0 ANCHOR] 块 — ref2va/无 iframe shot 的 0 秒起始锚定。 注入条件（任一）： - ``iframe_manifest_entry`` 为 None（manifest 缺该 shot / 无 manifest）； - entry 的 ``iframe_mode`` 缺失或为 ``"single_ref"``。 i2va/paired/chained（有条件帧）→ 不注入。 渲染：start_frame_description 重写为已完成时态 + 显式声明「该状态在 镜头开启前已经存在并持续（NOT the beginning of the action）」+ action_timeline 逐秒一行（"0-2s: …; 2-4s: …"）。action_timeline 缺失 → 降级到描述-only。start_frame_description 也缺 → ""。永不抛错。 """ if not isinstance(shot, dict): return "" if isinstance(iframe_manifest_entry, dict): mode = iframe_manifest_entry.get("iframe_mode") if mode in ("i2va", "paired", "chained"): return "" # 有条件帧——首帧图像已是更强的锚，不注入 desc = shot.get("start_frame_description") desc = desc.strip() if isinstance(desc, str) else "" if not desc: return "" lines = [ f"视频第 0 秒画面：{desc}（该状态为已完成时态——动作在镜头开启前已在进行）。", "该状态在镜头开启前已经存在并持续（NOT the beginning of the action）。", ] # anchor 与 video_prompt 动作描述冲突时以 anchor 的已完成时态为准 # （时态冲突是 R11 physics 检查的既有模式，这里在构造期统一）。 atl = shot.get("action_timeline") if isinstance(atl, list) and atl: beats = [] for beat in atl: if not isinstance(beat, dict): continue tr = beat.get("time_range") action = beat.get("action") if isinstance(action, str) and action.strip(): label = action.strip() if isinstance(tr, str) and tr.strip(): label = f"{tr.strip()}: {label}" beats.append(label) if beats: lines.append("逐秒：" + "; ".join(beats)) return "[FRAME-0 ANCHOR]\n" + "\n".join(lines) + "\n\n" # --------------------------------------------------------------------------- # # #5 (2026-08-15, hellgrind-iframe-optimization §3.3.2 + repro-experiment # §6.2 Q3 修正): [CONTINUITY-IN] chained shot 注入 # --------------------------------------------------------------------------- # # HG 跨镜头视觉交接声明（"EXACT visual handoff from 69B.15's final frame"） # 的 KMC 落地。适用范围（实验 Q3 修正，双触发）： # a) frame_chain == continuous 的 shot（chain copy 的 prompt 侧补全—— # 像素级拼缝零跳变，但渲染 prompt 对「这是上一镜头延续」零表达）； # b) transitiondesign 有明确 visual_hook 的硬切——数据源 # shot_intent.continuity.already_happened 有交接语义 # （如 "EXACT visual handoff from 69B.15"）也注入。 # tidx 复用 gen_iframes._build_transition_index 的构造逻辑（transition-design # slot）。transition_index=None（默认）→ 不注入，旧调用方零改动。 # already_happened 交接语义信号（中英双语）。 _HANDOFF_SIGNALS = ("handoff", "visual handoff", "exactly match", "延续", "交接", "承接上", "上一镜头") # FC-V21 (2026-08-17): _inject_chain_continuity 触发判定的三通道本地常量—— # 忠实复刻 gen_iframes.derive_frame_chain 的 R1/R2/R3（此前实现是「任何非 # hard_cut method 都触发」，比 gen_iframes 宽松，与本函数 docstring 声称的 # 复刻语义不符）。侵入最小原则：本地复刻，不 import templates/（pipeline # 包不依赖模板脚本）——三处常量与 gen_iframes 侧同名常量人工同步。 # p11b camera_continuity 枚举恰好是 {none,pan_to,zoom_to,dolly_through, # whip_pan,match_cut_on_motion}，即集合成员判定 ≡ 旧 `!= "none"` 判定的收紧。 _CHAIN_CONTINUOUS_CAMERA = { "pan_to", "zoom_to", "dolly_through", "whip_pan", "match_cut_on_motion", } # KEEP IN SYNC with gen_iframes.CONTINUOUS_CAMERA # FC-V21: gen_iframes.CONTINUOUS_TRANSITIONS 2026-08-17 收紧 6→2 值（感知 # 关联类 dissolve/match_cut_shared_object/match_cut_shared_landscape/ # color_match 不再触发帧连续），此处同步收紧。 _CHAIN_CONTINUOUS_METHODS = { "motion_vector_match", "whip_pan", } # KEEP IN SYNC with gen_iframes.CONTINUOUS_TRANSITIONS _CHAIN_HARD_CUT_HOOK_KEYWORDS = ( "承接", "顺接", "接续", "继续", "接着", "紧接着", "紧接", "顺势", "顺着", "延续", "接上", "跟上", "惯性", ) # KEEP IN SYNC with gen_iframes._HARD_CUT_HOOK_KEYWORDS # FC-V21 (2026-08-17, Phase 17 BOUND-02): boundary_continuity 显式字段枚举 # 本地复刻常量——shot 级 OPTIONAL enum（p09 字段 14），有效值（strip+lower # 归一化后）赢过 R1/R2/R3 三通道复刻；缺失/非 str/未知值 → 复刻判定兜底 # （fail-open，与 gen_iframes 侧 T-17-01 同语义，坏值不崩溃不翻转全量）。 _BOUNDARY_CONTINUOUS = frozenset({"frame_identical", "action_linked"}) _BOUNDARY_INDEPENDENT = frozenset({"contrast", "unrelated"}) # KEEP IN SYNC with gen_iframes._BOUNDARY_CONTINUOUS / _BOUNDARY_INDEPENDENT def _inject_chain_continuity(shot: dict, tidx: dict | None) -> str: """[#5] [CONTINUITY-IN] 块 — chained/硬切交接 shot 的连续性声明。 触发条件（任一）： 1. ``tidx`` 中指向本 shot 的 transition 满足 gen_iframes 的 continuous 判定三通道（R1 camera_continuity ∈ _CHAIN_CONTINUOUS_CAMERA / R2 transition_method ∈ _CHAIN_CONTINUOUS_METHODS / R3 硬切+钩子词， FC-V21 2026-08-17 起忠实复刻，不再对任意非 hard_cut method 放行）； shot 级 boundary_continuity 显式声明优先（2026-08-17 FC-V21 起， Phase 17）——有效值赢过三通道复刻：frame_identical/action_linked → 触发，contrast/unrelated → 分支 1 不触发（直接落分支 2 检查）， 缺失/未知 → 复刻判定兜底； 2. ``shot_intent.continuity.already_happened`` 含交接语义 （EXACT visual handoff 类）——实验 Q3 修正：硬切交接无 chain copy， 纯靠 prompt 声明（本分支不受 boundary_continuity 字段影响）。 tidx 为 None / transition-design 缺失且无交接语义 → ""（现状降级）。 """ if not isinstance(shot, dict): return "" sid = shot.get("shot_id") or shot.get("id") prev_sid: str | None = None triggered = False if isinstance(tidx, dict) and sid: # 构造与 gen_iframes._build_transition_index 相同的查询： # tidx[tgt] = (src, next_shot_link)。复用其 frame_chain 判定 # （FC-V21: R1→R2→R3 三通道，与 derive_frame_chain 语义一致）。 entry = tidx.get(sid) if isinstance(entry, tuple) and len(entry) == 2: prev_sid, link = entry # FC-V21 (2026-08-17, Phase 17): shot 级显式边界声明优先——与 # gen_iframes.derive_frame_chain 的 boundary 优先级同步（KEEP IN # SYNC with gen_iframes.derive_frame_chain boundary precedence）。 # 归一化 strip+lower；非 str / 纯空白 / 未知值 → 按缺失回退复刻。 bc = shot.get("boundary_continuity") bc = bc.strip().lower() if isinstance(bc, str) else None if bc in _BOUNDARY_CONTINUOUS: triggered = True elif bc not in _BOUNDARY_INDEPENDENT: # 缺失/未知 → 原三通道复刻判定原样执行（contrast/unrelated # 跳过复刻，triggered 保持 False 直接落到分支 2 检查）。 camera = (link or {}).get("camera_continuity", "none") method = (link or {}).get("transition_method", "hard_cut") if camera in _CHAIN_CONTINUOUS_CAMERA \ or method in _CHAIN_CONTINUOUS_METHODS \ or _has_handoff_hook(link or {}): triggered = True if not triggered: # 分支 2: already_happened 交接语义（无 transition-design 也可触发）。 si = shot.get("shot_intent") si = si if isinstance(si, dict) else {} cont = si.get("continuity") cont = cont if isinstance(cont, dict) else {} already = cont.get("already_happened") if isinstance(already, list): blob = " ".join(str(x) for x in already).lower() if any(sig.lower() in blob for sig in _HANDOFF_SIGNALS): triggered = True if not triggered: return "" src_note = f"<{prev_sid}>" if prev_sid else "<上一镜头>" return ( "[CONTINUITY-IN]\n" f"本镜头是 {src_note} 的直接延续。起始画面（已由首帧图像锁定 = " "上一镜头尾帧）的机位、构图与角色姿态必须严格保持；本镜头的运动" "从该状态继续，绝不重置。\n\n" ) def _has_handoff_hook(link: dict) -> bool: """硬切 + visual_hook 含承接语义 → 视为 continuous（gen_iframes 同款词表）。 FC-V21 (2026-08-17): 词表 4→13 词对齐 gen_iframes._HARD_CUT_HOOK_KEYWORDS （CHAIN-AUDIT 扩表后 p11b 此前未同步）；签名与返回语义不变。 """ hook = link.get("visual_hook") or "" return bool(link.get("transition_method") == "hard_cut" and isinstance(hook, str) and any(kw in hook for kw in _CHAIN_HARD_CUT_HOOK_KEYWORDS)) def _build_tidx_from_transition_design(transition_design: Any) -> dict: """#5: 从 transition-design 数据构造 {target_shot_id: (prev, link)} 映射。 与 gen_iframes._build_transition_index 同一构造逻辑（target 视角：指向 本 shot 的 next_shot_link 所在 entry，其 shot_id 即前镜）。侵入最小实现： 本地复刻（避免 pipeline 包 import templates/ 脚本）；两层任何形状不对 → 空 dict（调用方不注入）。 """ data = transition_design # asset-bus envelope ({"value": [...]}) 兼容 if isinstance(data, dict) and isinstance(data.get("value"), list): data = data["value"] if not isinstance(data, list): return {} idx: dict[str, tuple[str | None, dict]] = {} for entry in data: if not isinstance(entry, dict): continue src = entry.get("shot_id") link = entry.get("next_shot_link") if isinstance(link, dict): tgt = link.get("shot_id") if tgt: idx[tgt] = (src if isinstance(src, str) else None, link) return idx # --------------------------------------------------------------------------- # # N1 (repro-experiment §6.3): PHYSICS — continuity.locks 正向 diegetic 注入 # --------------------------------------------------------------------------- # # 实验 §5 E 行: 物理规则现状「不触达任何引擎」——negative_prompt 预算 10-30 # token 装不下 HG 的硬物理约束全集，H3 CFG=1.0 下负面词又无效。修复: # continuity.locks[]（可承接 HG 硬物理约束全集）+ frame_physics_rule（#3 新 # 字段，单条核心约束）渲染为正向 diegetic 约束块。改写纪律照抄 # _generate_targeted_avoid（HG-11）: 否定词必须正向化。插入位: _anchors 链 # hard locks 之后、targeted AVOID 之前。 # locks/负面向 → 正向 diegetic 的改写表（负向词 → 期望的正向结果表述）。 # 未命中的条目按「保持 <原文描述的物理事实>」兜底（原文已是正向写法时直接用）。 _PHYSICS_POSITIVE_REWRITES: list[tuple[str, str]] = [ ("不刺穿", "仅与表面发生刮擦接触（never pierce — surface scratch contact only）"), ("不穿透", "接触停留在表面（contact stays on the surface）"), ("零火花", "金属接触无火花（metal contact stays spark-free）"), ("无火花", "金属接触无火花（metal contact stays spark-free）"), ("不发光", "保持哑光不发光（stays matte, no glow）"), ("零出血", "伤口无出血（wounds stay bloodless — 黑灰烬烟替代）"), ("不流血", "伤口无出血（wounds stay bloodless）"), ("不消失", "全程持续存在（present throughout the whole shot）"), ("不改变", "全程保持不变（held constant throughout the shot）"), ("不移", "位置全程保持（position held throughout the shot）"), ("不断裂", "保持完整结构（structure stays intact）"), ] def _positivize_lock(lock: str) -> str: """把一条 lock 里最常见的否定表述正向化（N1 改写纪律）。 未匹配任何否定模式 → 原文返回（多数 lock 本就是正向物理事实陈述）。 """ text = lock.strip() for neg, pos in _PHYSICS_POSITIVE_REWRITES: if neg in text: return text.replace(neg, pos) return text def _inject_physics_locks(shot: dict) -> str: """[PHYSICS] 块 — 把 shot 的物理约束渲染为正向 diegetic 约束块。 H3 CFG=1.0 → negative_prompt 无效，否定词必须正向化（改写纪律照抄 _generate_targeted_avoid）。数据源优先级: shot_intent.continuity.locks[] 全量 > frame_physics_rule 单条 > 空串（不注入）。 """ if not isinstance(shot, dict): return "" rules: list[str] = [] si = shot.get("shot_intent") if isinstance(si, str): si = {} si = si if isinstance(si, dict) else {} cont = si.get("continuity") cont = cont if isinstance(cont, dict) else {} locks = cont.get("locks") if isinstance(locks, list): rules.extend(_positivize_lock(str(x)) for x in locks if str(x).strip()) if not rules: fpr = _lock_str(shot.get("frame_physics_rule")) if fpr: rules.append(_positivize_lock(fpr)) if not rules: return "" body = "\n".join(f"- {r}" for r in rules) return "[PHYSICS]\n" + body + "\n\n" # --------------------------------------------------------------------------- # # Hell Grind §7.1 — targeted AVOID (HG-11, 2026-08-13) # --------------------------------------------------------------------------- # # Hell Grind replaces the generic ban list with a PER-SHOT targeted AVOID # paragraph: the 3-5 artifact concerns most likely to fail for THIS shot. # CRITICAL: H3 runs at CFG=1.0 where negative_prompt is INEFFECTIVE (see # _H3_PROMPT_RULES / "NO NEGATIVE-PROMPT RELIANCE"). So every concern is # phrased as the DESIRED POSITIVE diegetic outcome, never a ban word — that # is the only content-control channel H3 honors. ``_generate_targeted_avoid`` # auto-selects modes from shot characteristics and honors any the performer # flagged explicitly (``anti_ai_artifact``, HG-10). Selective + never blocks. AVOID_POSITIVE: dict[str, str] = { "identity_drift": ( "each character keeps a consistent, distinct identity (face, hair, " "costume) throughout the shot" ), "robotic_lipsync": ( "mouths articulate visibly and naturally on every spoken line" ), "frozen_face": ( "faces stay alive with continuous micro-acting (eyes, brows, mouth) — " "never a static mask or blank stare" ), "body_morphing": ( "bodies hold stable geometry throughout — no limb or torso morphing" ), "position_shift": ( "characters hold their blocking positions between cuts — no unexplained relocation" ), } def _generate_targeted_avoid(shot: dict) -> str: """Build a ``[TARGETED CONSISTENCY]`` block of POSITIVE diegetic language. Hell Grind §7.1: instead of a generic ban list, select the artifact concerns most likely to fail for THIS shot and phrase them as the desired positive outcome. H3's CFG=1.0 makes negative_prompt ineffective, so positive diegetic phrasing is the only content-control channel. Auto-detected triggers: - ≥2 characters (``blocking`` HG-1 / ``character_refs``) → identity_drift - spoken dialogue (``dialogue_text``) → robotic_lipsync - expression change (``micro_acting`` HG-9 / ``action_timeline`` emotion) → frozen_face - multi-beat / multi-cut action (``action_timeline`` ≥3) → position_shift - significant motion (multi-char OR multi-beat) → body_morphing Plus any mode the performer flagged explicitly in ``anti_ai_artifact`` (HG-10) — authoritative, added even if not auto-detected. Returns "" when nothing triggers — selective, never blocks, never raises. """ if not isinstance(shot, dict): return "" modes: list[str] = [] # Character count: blocking (HG-1) is the structured source; fall back to # character_refs. ≥2 characters → identity drift is the top failure mode. blocking = shot.get("blocking") refs = shot.get("character_refs") n_chars = ( len(blocking) if isinstance(blocking, list) else (len(refs) if isinstance(refs, list) else 0) ) if n_chars >= 2: modes.append("identity_drift") # Spoken dialogue → robotic lip-sync is the model-specific failure mode. if _lock_str(shot.get("dialogue_text")): modes.append("robotic_lipsync") # Expression change → frozen mask-face. micro_acting (HG-9) is the direct # signal; an action_timeline beat carrying emotion also implies it. atl = shot.get("action_timeline") micro = shot.get("micro_acting") has_expression = isinstance(micro, dict) and any( _lock_str(micro.get(k)) for k in ("eyes", "mouth", "brows", "body", "beat_chain") ) if not has_expression and isinstance(atl, list): has_expression = any( isinstance(b, dict) and _lock_str(b.get("emotion")) for b in atl ) if has_expression: modes.append("frozen_face") # Multi-beat / multi-cut action → position shift between cuts. if isinstance(atl, list) and len(atl) >= 3: modes.append("position_shift") # Significant motion (multi-character OR multi-beat action) → body morphing. if n_chars >= 2 or (isinstance(atl, list) and len(atl) >= 3): modes.append("body_morphing") # Honor explicit performer-flagged concerns (anti_ai_artifact, HG-10). aaa = shot.get("anti_ai_artifact") if isinstance(aaa, list): for concern in aaa: if isinstance(concern, dict): mode = _lock_str(concern.get("mode")) if mode and mode in AVOID_POSITIVE and mode not in modes: modes.append(mode) if not modes: return "" # de-dup preserving order; render each as its positive diegetic phrase. seen: set[str] = set() uniq_modes = [m for m in modes if not (m in seen or seen.add(m))] lines = [AVOID_POSITIVE[m] for m in uniq_modes] return "[TARGETED CONSISTENCY]\n" + "; ".join(lines) + ".\n\n" # --------------------------------------------------------------------------- # # Hell Grind §3.1 — intra-shot multi-CUT editing (HG-12, 2026-08-13) # --------------------------------------------------------------------------- # # Hell Grind packs 2-3 edited segments into a SINGLE 15s H3 generation with # explicit ``[CUT N]`` headers + HARD CUT markers, so H3 emits an already-edited # segment (no post stitch). ``_inject_multi_cut`` turns the shot's ``multi_cut`` # plan (HG-12 P09 field) into that prompt block. Selective + never blocks. def _inject_multi_cut(shot: dict) -> str: """Build a ``[MULTI-CUT]`` intra-shot editing block from ``shot.multi_cut``. Each CUT becomes a ``[CUT-N — ~s]`` header followed by its lens / camera-position / action lines and a ``HARD CUT`` marker. Returns "" when ``multi_cut`` is absent/empty — selective, never blocks, never raises. """ if not isinstance(shot, dict): return "" cuts = shot.get("multi_cut") if not isinstance(cuts, list) or not cuts: return "" lines: list[str] = [] for i, cut in enumerate(cuts, 1): if not isinstance(cut, dict): continue cid = _lock_str(cut.get("cut_id")) or f"CUT-{i}" scale = _lock_str(cut.get("shot_scale")) dur = cut.get("duration_sec") dur_s = f"~{dur}s" if isinstance(dur, (int, float)) else "" header = f"[{cid}" suffix = " — ".join(p for p in (scale, dur_s) if p) header += f" — {suffix}]" if suffix else "]" lines.append(header) lens = _lock_str(cut.get("lens_mm")) if lens: lines.append(f"Lens: {lens}.") cam = _lock_str(cut.get("camera_position")) if cam: lines.append(f"Camera position: {cam}.") action = _lock_str(cut.get("action")) if action: lines.append(f"Action: {action}.") lines.append("HARD CUT.") if not lines: return "" return "[MULTI-CUT]\n" + "\n".join(lines) + "\n\n" def _h3_enhance_prompt(shot: dict, episode_config: dict | None = None, character_assets: dict | None = None, transition_index: dict | None = None, iframe_manifest_entry: dict | None = None) -> str: """增强 shot 的提示词为 H3 协议格式。 输入: shot 的 ``video_prompt`` (引擎无关镜头视频描述; ``ltx_prompt`` 为 legacy alias, 经 _video_prompt helper 降级读取) / ``prompt`` (裸短需求)。 输出: 符合 H3 协议的增强中文提示词;任何失败/关闭都回落原始 prompt, 绝不抛异常、绝不阻塞渲染。 本地模板模式(默认)零 LLM 成本,纯规则拼装协议骨架。 API增强模式在 agent 执行环境中由调用方通过 hermes_llm 工具调 h3_prompt_enhancer.get_system_prompt() + build_user_request()。 ``transition_index``（#5, optional，默认 None）: gen_iframes ``_build_transition_index`` 产出的 {target_shot_id: (prev_shot_id, next_shot_link)} 映射。非 None 且本 shot 命中连续性/硬切交接判定 → 注入 [CONTINUITY-IN] 块（ATMOSPHERE 之后、PHYSICS 之前）。None → 不注入，旧调用方零改动。 ``iframe_manifest_entry``（#6, optional，默认 None）: 该 shot 在 iframe-manifest.json 的 entry（gen_iframes 产出）。None 或 iframe_mode 缺失/single_ref → 注入 [FRAME-0 ANCHOR]（SUBJECT STATE 之后、hard locks 之前）；i2va/paired/chained → 不注入。 """ raw_prompt = _get_vp(shot) shot_id = shot.get("shot_id") or shot.get("id", "unknown") # P0-2 + P1-1 (Hell Grind): identity locks + hard constraint locks as a # prompt preamble. Applied BEFORE the enhance switch so anchors reach the # engine even when template enhancement is disabled/unavailable. Never # blocks (empty → no-op). character_assets defaults None → legacy callers # unchanged. The hard-lock slow-motion scan uses the clean raw_prompt so a # prepended anchor cannot self-trigger it. # N2 (repro-experiment §6.3): [SUBJECT STATE] 状态级外观快照 — 紧跟身份锁 # （外观状态是身份的一部分）; #6 (2026-08-15): [FRAME-0 ANCHOR] 0 秒起始 # 锚定 — SUBJECT STATE 之后、hard locks 之前（仅无条件帧 shot 注入）; # #4 (2026-08-15): [ATMOSPHERE LOCK] 场景级大气锁 — hard locks 之后、 # CONTINUITY/PHYSICS 之前; #5 (2026-08-15): [CONTINUITY-IN] chained/硬切 # 交接声明 — ATMOSPHERE 之后、PHYSICS 之前。 # 注入链最终顺序: identity → SUBJECT STATE → FRAME-0 ANCHOR → hard → ATMOSPHERE → CONTINUITY → CROWD → PHYSICS → AVOID → multi-cut。 _anchors = ( _inject_identity_locks(shot, character_assets) + _inject_subject_state(shot, character_assets) + _inject_frame_anchor(shot, iframe_manifest_entry) + _inject_hard_locks(shot, raw_prompt) + _inject_atmosphere_lock(shot) + _inject_chain_continuity(shot, transition_index) + _inject_crowd(shot) ) # #8 (hellgrind-iframe-optimization §5.3(b)): director shot 的 enhance # 建议记录（仅日志建议，不强制改 API 调用路径——本地模板仍是管线运行时 # 唯一确定性路径；API 增强 by design 由 agent 经 hermes_llm 执行）。 if isinstance(shot, dict) and str(shot.get("prompt_richness") or "").strip().lower() == "director": logger.info( "%s: shot %s prompt_richness=director — 建议对该 shot 采用 API增强 " "(h3_enhance_mode) 做精修（成本集中到关键镜头）", PHASE_ID, shot.get("shot_id") or shot.get("id", "unknown"), ) if _anchors.strip(): raw_prompt = _anchors + raw_prompt # HG-11 (Hell Grind §7.1): targeted AVOID — per-shot POSITIVE diegetic # consistency language (H3 CFG=1.0 → negative_prompt ineffective). Appended # AFTER anchors so it lands in the prompt body the local template consumes. # Never blocks (empty → no-op). # N1 (repro-experiment §6.3): [PHYSICS] 先于 AVOID 注入 prompt body —— # 物理约束是本镜头的帧级核心规则，优先级高于 artifact 一致性语言。 _physics = _inject_physics_locks(shot) if _physics.strip(): raw_prompt = raw_prompt.rstrip() + "\n" + _physics _targeted_avoid = _generate_targeted_avoid(shot) if _targeted_avoid.strip(): raw_prompt = raw_prompt.rstrip() + "\n" + _targeted_avoid # HG-12 (Hell Grind §3.1): intra-shot multi-CUT — pack 2-3 edited segments # into one generation with [CUT N] + HARD CUT markers. Appended to the # prompt body; never blocks (empty → no-op, single continuous shot). _multi_cut = _inject_multi_cut(shot) if _multi_cut.strip(): raw_prompt = raw_prompt.rstrip() + "\n" + _multi_cut cfg = episode_config or {} # 开关: shot 级 > episode_config 级 > 默认 true。 enhance_enabled = shot.get("h3_enhance") if enhance_enabled is None: enhance_enabled = cfg.get("h3_enhance", True) if not enhance_enabled: logger.info("%s: shot %s h3_enhance=false, 跳过增强", PHASE_ID, shot_id) return raw_prompt # enhancer 不可用 → 直接回落(hard-fail-safe)。 if not _H3_ENHANCER_AVAILABLE: logger.warning("%s: shot %s h3_prompt_enhancer 模块不可用, 回落原始 prompt", PHASE_ID, shot_id) return raw_prompt # taskMode 映射: shot.mode > episode_config.h3_mode > 默认 ref2va。 mode = shot.get("mode") or cfg.get("h3_mode") or "ref2va" task_mode = _H3_MODE_TO_TASK.get(mode, _H3_MODE_TO_TASK["ref2va"]) # referenceLabels: ref2va/i2va 及未列出 mode 都引用 ; # 仅 t2va(文生视频,无参考图)不带。 reference_labels = None if mode == "t2va" else "" duration = int(float(shot.get("duration_sec", 8) or 8)) # enhanceMode: shot 级 > episode_config 级 > 默认本地模板。 enhance_mode = shot.get("h3_enhance_mode") or cfg.get("h3_enhance_mode") or "本地模板" if enhance_mode not in ("本地模板", "API增强"): enhance_mode = "本地模板" try: if enhance_mode == "API增强": # API 增强由 agent 通过 hermes_llm 工具执行(非管线运行时直调)。 # 管线运行时降级为本地模板(保证不阻塞)。 logger.info( "%s: shot %s API增强模式在管线运行时降级为本地模板 " "(API增强由 agent 通过 hermes_llm 执行)", PHASE_ID, shot_id, ) enhanced = _h3_build_local_template( task_mode=task_mode, prompt=raw_prompt, duration=duration, reference_labels=reference_labels, ) logger.info( "%s: shot %s H3 提示词增强成功 (mode=%s, taskMode=%s, enhancer=local-template)", PHASE_ID, shot_id, mode, task_mode, ) return enhanced except Exception as exc: logger.warning( "%s: shot %s 提示词增强失败 (%s: %s), 回落原始 prompt", PHASE_ID, shot_id, type(exc).__name__, exc, ) return raw_prompt def _form_factor_render_args_h3(shot: dict, episode_config: dict | None = None) -> str: """Build the H3 API call hint with form-factor-correct resolution. H3 通过 KAP 统一管线 API (multipart/form-data) 调用,不走 batch render script。分辨率约束:32 倍数 (portrait=768×1344 / landscape=1344×768)。 帧数对齐:n%17==5 (由 H3DurationAdapter 计算,与 adapter skill 一致)。 返回的是可由 subagent 直接执行的 curl 模板 (ref2va 默认模式)。 """ ff = (episode_config or {}).get("form_factor") or shot.get("form_factor", "portrait") if ff == "landscape": w, h = 1344, 768 else: w, h = 768, 1344 # H3 useCase (2026-08-17 KAP API 精简): final-shot = native-sage 36 步 + # 完整 Foley + TTS 混音 (KAP 侧预射, 取代旧 preview/production profile 平铺)。 # 优先级: shot 级 h3_useCase > episode_config 级 > 默认 final-shot。 h3_usecase = shot.get("h3_useCase") or (episode_config or {}).get("h3_useCase", "final-shot") if h3_usecase not in ("preview-lock", "final-shot"): h3_usecase = "final-shot" shot_id = shot.get("shot_id") or shot.get("id", "unknown") # H3 提示词增强:裸需求先进 video-engine-adapter skill 的 # h3_prompt_enhancer 模块(纯 Python,零 KAP 依赖),失败自动回落 # 原始 prompt(见 _h3_enhance_prompt 的降级纪律)。curl 值压平换行并 # escape 单引号 (' → '\''),保证 multipart 单行安全、不被 shell 截断。 prompt = _h3_enhance_prompt(shot, episode_config).replace("\n", " ").replace("'", "'\\''") # H3 帧数:duration × 24fps, snap 到最近 n%17==5 网格点 (与 adapter 一致)。 # P0-4: 抽取到 _expected_frame_count 以复用于帧数==meta 硬校验。 length = _expected_frame_count(shot) return ( f"curl -X POST {H3_API_URL} \\\n" f" -F 'projectId=1' \\\n" f" -F 'prompt={prompt}' \\\n" f" -F 'mode=ref2va' \\\n" f" -F 'useCase={h3_usecase}' \\\n" f" -F 'width={w}' \\\n" f" -F 'height={h}' \\\n" f" -F 'length={length}' \\\n" f" -F 'filenamePrefix={shot_id}' \\\n" f" # ref2va: attach refImages[]=@ + @ (≤9)\\n" f" # i2va (首帧锁定): replace mode + attach image=@\\n" f" # ttsAudio (TTS 对白): attach ttsAudio=@ (可选)" ) # --------------------------------------------------------------------------- # Phase metadata — referenced by runner.py + tests # --------------------------------------------------------------------------- PHASE_ID = "p11b_final_render" EXPERT = "visual_executor" # INPUT_SLOTS aligned to actual runtime asset_bus_read() calls in _run_body: # P11b reads shot-list/scene-images/character-assets/voice-clips directly, # NOT preview-clips (rapid-preview-clips from P11a is operator-facing only). # depends_on=["p11a_preview_clips"] provides ordering; data dependency is direct. INPUT_SLOTS = ["shot-list", "scene-images", "character-assets", "voice-clips"] OUTPUT_SLOTS = ["video-clips", "take-log"] GATE_ID = "p11b-gate" # Gate 7 per V8.6 SSoT (derive_gate_id('p11b_final_render'); # 2026-08-18 full-token derivation — legacy "p11-gate" aliased in # gate_config._LEGACY_GATE_ID_TO_PHASE_ID) # Fail-closed ffprobe validation: when ffprobe is missing/times out in # production we treat the media as INVALID (return False) rather than letting # corrupt/truncated clips pass. Set KMP_SKIP_FFPROBE=1 to restore the old # permissive skip behaviour in dev/test (returns True). _SKIP_FFPROBE = os.environ.get("KMP_SKIP_FFPROBE", "") == "1" # --------------------------------------------------------------------------- # DEFAULT_VIDEO_ENGINE — H3 (MiniMax H3) is the ONLY video generation engine # for P11 (2026-08-05). H3 通过 KAP 统一管线 API 调用 # (POST /api/production/minimax-h3/generate):原生音频、CFG=1.0(负面词无效)、 # 无需 trim,输入适配由 video-engine-adapter skill 完成。 # 不设 LTX fallback:H3 出问题先修复再尝试,不自动回退。 # # H3 batch entry: KAP API /api/production/minimax-h3/generate (via :10588) # H3 adapter: skills/movie-experts/video-engine-adapter/ # --------------------------------------------------------------------------- DEFAULT_VIDEO_ENGINE = "h3" # H3 统一管线端点 (KAP)。 H3_API_URL = "http://localhost:10588/api/production/minimax-h3/generate" # --------------------------------------------------------------------------- # Take Log — structured retake tracking (Seedance retake-protocol inspired) # # Each generated take appends one line to the take_log JSONL. This creates # an auditable iteration history: what changed, what seed, what verdict. # The five-verdict triage (Keep / Fix in post / Edit / Re-roll / Rewrite) # surfaces systematic failures — same flaw twice = rewrite, not third re-roll. # # Per Seedance retake-protocol: "Two takes in the log with the same flaw # is a rewrite, by rule — no third attempt on luck." # # Take log entry schema: # {take_n, shot_id, changed_variable, seed, verdict, evidence, timestamp} # verdict enum: keep | fix_in_post | edit | re_roll | rewrite # --------------------------------------------------------------------------- # --------------------------------------------------------------------------- # Reference Role Map — structured asset role assignment # # Per Seedance reference-workflow: every reference asset gets ONE primary role. # Prevents accidental identity/motion/style contamination. # P09 already emits reference_images[{slot, image, description}]; we upgrade # description → role enum at P11 read time for engine-agnostic consumption. # # Role enum: identity | scene | pose | motion | camera | style | audio | first_frame | last_frame # --------------------------------------------------------------------------- # --------------------------------------------------------------------------- # R11 per-shot pre-check (2026-06-29 Volvo S3-2 lessons) # # Lightweight inline scan of shot['video_prompt'] for the two failure modes # that cost 12 wasted renders on Volvo S3-2. The authoritative full-version # lives in scripts/ltx_batch_submit.py (check_prompt_physics + # detect_bgm_triggers); P11 only needs the subset subagents realistically # trip on — full word lists would crowd this orchestration module. # # KEEP IN SYNC with ltx_batch_submit.py when adding new conflict words. # --------------------------------------------------------------------------- _R11_PHYSICS_CONFLICTS = { "driving-straight": ( "turning", "turn a corner", "turn-signal", "turn signal", "cornering", "转弯", "转向灯", "拐弯", ), "driving-turn": ( "straight road", "driving straight", "直行", "笔直", ), "stationary": ( "driving", "in motion", "moving forward", "行驶", "运动", ), } # --------------------------------------------------------------------------- # SHARED-01 (Phase 10-01): BGM word list migrated from the inline tuple to the # shared/bgm_blacklist.yaml resource. P09 (producer) and P11 (consumer) now # read from one version-controlled source of truth. The inline # `_R11_BGM_TRIGGERS` tuple is REMOVED — tests assert it no longer exists. # # Lazy load + module-level cache: load once, reuse. Tests can reset the cache # by setting `p11._BGM_CACHE = None`. # --------------------------------------------------------------------------- _BGM_CACHE: dict | None = None def _get_bgm_blacklist() -> dict: """Return the BGM blacklist dict from shared/bgm_blacklist.yaml (cached). Shape: {"bgm_triggers": [...], "negative_prompt_phrases": [...]}. """ global _BGM_CACHE if _BGM_CACHE is None: _BGM_CACHE = load_shared_yaml("bgm_blacklist") return _BGM_CACHE # --------------------------------------------------------------------------- # FIX-1 (2026-07-31): 逐镜头断点续传 checkpoint # --------------------------------------------------------------------------- # 问题: P11 有最多 26 个并行 subagent,全部完成才写 phase checkpoint。 # 如果中途崩溃或被 GLM 限流,已渲染好的镜头会全部丢失,重跑浪费大量 token/GPU。 # # 解决方案: 每完成一个镜头就即时写入 p11_progress.json,记录 completed_shots。 # P11 重新进入时先读取该文件,跳过已完成镜头。配合 ffprobe 校验确保 .mp4 有效 # (文件存在不代表渲染成功 —— 半写/截断的 mp4 必须被重新渲染)。 # # 进度文件位置: /.pipeline-assets/p11_progress.json # 格式: {"completed_shots": ["S1_1", ...], "failed_shots": [...], "last_updated": "ISO"} # # 注意: 这与现有 R-40-01 的 _try_cache_hit 文件存在性检查是互补关系 —— # _try_cache_hit 只看 .mp4+.meta 文件是否存在(快速路径),而 checkpoint # 记录的是「渲染流程已经确认完成」的语义状态(慢速路径)。两者都会跳过已 # 完成镜头:cache_hit 优先(更快),progress 文件作为跨崩溃的补充记录。 # --------------------------------------------------------------------------- # 进度文件名(相对于 run_dir,放在 .pipeline-assets/ 下与 shot-list.json 同目录) _P11_PROGRESS_FILENAME = "p11_progress.json" def _p11_progress_path(output_dir: str) -> str: """返回 p11_progress.json 的绝对路径。 output_dir 通常是 ``assets/P11`` (相对 cwd);进度文件写在 run_dir 下 的 ``.pipeline-assets/`` 里 (与 shot-list.json 同目录),保持所有管线 元数据集中。从 ``assets/P11`` 向上两层即可到 run_dir。 """ # assets/P11 → run_dir; .pipeline-assets/ 与 assets/ 平级 run_dir = os.path.dirname(os.path.dirname(os.path.abspath(output_dir))) return os.path.join(run_dir, ".pipeline-assets", _P11_PROGRESS_FILENAME) # --------------------------------------------------------------------------- # # TTS duration measurement — inlined ffprobe logic (formerly P10a shot_timeline) # --------------------------------------------------------------------------- # # P10a shot_timeline was removed 2026-08-09: voice clips and the per-shot # video render are inherently 1:1, so measuring the real TTS audio duration # belongs here (P11b already reads ``voice-clips``) rather than in a # dedicated upstream phase. #: Duration clamp bounds (seconds). P11b never emits a duration outside [MIN, MAX]. MIN_DURATION_SEC = 1.5 MAX_DURATION_SEC = 15.0 #: Tail buffer added to the measured TTS duration so the rendered clip isn't #: cut exactly at the speech end. _TTS_TAIL_BUFFER_SEC = 0.5 #: Minimum duration for a shot without a usable voice clip. _SILENT_MIN_SEC = 3.0 def _ffprobe_audio_duration(audio_path: str) -> float | None: """Measure the actual TTS audio duration of ``audio_path`` via ffprobe. Returns the duration in seconds (float), or ``None`` when ffprobe is unavailable, the path is empty/missing, or the output is unparseable. """ if not audio_path or not isinstance(audio_path, str): return None try: r = subprocess.run( ["ffprobe", "-v", "quiet", "-show_entries", "format=duration", "-of", "csv=p=0", audio_path], capture_output=True, text=True, timeout=10, ) except (FileNotFoundError, subprocess.TimeoutExpired, OSError): return None if r.returncode != 0: return None try: dur = float((r.stdout or "").strip()) except ValueError: return None return dur if dur > 0 else None def _clamp_duration(val: float) -> float: """Clamp ``val`` into [MIN_DURATION_SEC, MAX_DURATION_SEC].""" return max(MIN_DURATION_SEC, min(MAX_DURATION_SEC, val)) def _safe_duration_sec(val: Any) -> float: """Coerce ``val`` to a non-negative float, falling back to 0.0.""" try: f = float(val) except (TypeError, ValueError): return 0.0 return f if f >= 0 else 0.0 def _norm_shot_id(shot_id: str) -> str: """Normalize shot_id so P09 shot-list and P10 voice-clips match across zero-padding disagreement (``S01_B01`` → ``S1_B1``). Any digit run with leading zeros (len > 1) is stripped of its leading zeros, segment by segment. Replaces the per-block ``_norm_sid`` lambdas that lived inside the former shot-timeline override blocks. """ if not isinstance(shot_id, str) or not shot_id: return "" def _strip_zeros(m: re.Match[str]) -> str: digits = m.group(0) if len(digits) > 1 and digits[0] == "0": return digits.lstrip("0") or "0" return digits return re.sub(r"\d+", _strip_zeros, shot_id) def _ffprobe_mp4_valid(mp4_path: str) -> bool: """用 ffprobe 校验 mp4 文件是否完整可读 (有 duration 才算有效)。 文件存在 ≠ 渲染成功:崩溃时可能留下截断/半写的 mp4。ffprobe 能读取 duration 说明 moov atom 完整、解码器可解析。 ffprobe 缺失或超时时 **fail-closed** (返回 False),避免让损坏/静默的 媒体静默通过校验。仅当设置环境变量 ``KMP_SKIP_FFPROBE=1`` (dev/test) 时才退化为「文件存在且非空即通过」的弱校验,不阻断管线。 """ if not os.path.isfile(mp4_path) or os.path.getsize(mp4_path) == 0: return False try: # -v error: 只输出错误; -show_entries format=duration: 取时长; # 超时 10s 防止 ffprobe 挂死在损坏文件上。 r = subprocess.run( ["ffprobe", "-v", "error", "-show_entries", "format=duration", "-of", "default=noprint_wrappers=1:nokey=1", mp4_path], capture_output=True, text=True, timeout=10, ) if r.returncode != 0: return False dur = (r.stdout or "").strip() # 能解析出正数 duration 即视为有效 return bool(dur) and float(dur) > 0 except (FileNotFoundError, OSError): if _SKIP_FFPROBE: logger.warning( "%s: ffprobe not found, skipping media validation " "(KMP_SKIP_FFPROBE=1)", PHASE_ID) return True # dev mode skip logger.error( "%s: ffprobe not found — cannot validate media. " "Set KMP_SKIP_FFPROBE=1 to bypass in dev.", PHASE_ID) return False # FAIL CLOSED in production except (ValueError, subprocess.TimeoutExpired): if _SKIP_FFPROBE: return True logger.error( "%s: ffprobe validation failed/timed out — treating media as INVALID", PHASE_ID) return False # FAIL CLOSED in production # P0-4: 帧数==meta 硬校验 —— 闭合 trim-bypass。 _FRAME_TOLERANCE = 1 # 编码器 padding 容差；>1 视为 trim_bypass def _expected_frame_count(shot: dict) -> int: """shot.duration_sec → H3 合法帧数 (与 _form_factor_render_args_h3 一致)。 duration × 24fps, snap 到最近 n%17==5 网格点 (由 H3DurationAdapter 计算)。 adapter 不可用时回落到 H3 默认 124 (124 % 17 == 5)。 """ try: from ..engine_adapters import get_adapter as _get_adapter duration_sec = float((shot or {}).get("duration_sec", 5.0) or 5.0) return _get_adapter("h3").render_to_frame_count(duration_sec, fps=24) except Exception: return 124 # H3 默认 (124 % 17 == 5) def _ffprobe_frame_count(mp4_path: str) -> int | None: """读 mp4 帧数。优先 nb_frames (header, 快)；N/A 时回退 count_frames (全解码, 慢)。 ffprobe 缺失/超时/解析失败一律返回 None (fail-closed)，由调用方决定处置。 """ if not os.path.isfile(mp4_path): return None try: # 快路径：读 header 的 nb_frames。 r = subprocess.run( ["ffprobe", "-v", "error", "-select_streams", "v:0", "-show_entries", "stream=nb_frames", "-of", "default=noprint_wrappers=1:nokey=1", mp4_path], capture_output=True, text=True, timeout=10, ) nb = (r.stdout or "").strip() if nb and nb != "N/A": return int(nb) # 慢路径：全解码计数（header 无 nb_frames 时）。 r2 = subprocess.run( ["ffprobe", "-v", "error", "-count_frames", "-select_streams", "v:0", "-show_entries", "stream=nb_read_frames", "-of", "default=noprint_wrappers=1:nokey=1", mp4_path], capture_output=True, text=True, timeout=60, ) nb2 = (r2.stdout or "").strip() return int(nb2) if nb2 and nb2 != "N/A" else None except (FileNotFoundError, ValueError, subprocess.TimeoutExpired, OSError): return None def _mp4_valid_with_frames(mp4_path: str, shot: dict) -> tuple[bool, str]: """_ffprobe_mp4_valid + 帧数==期望 硬校验。 duration 校验失败 → 无效；帧数与期望 (容差 _FRAME_TOLERANCE) 不符 → trim_bypass → 无效。dev 模式 (_SKIP_FFPROBE) 帧探针失败时退化为仅 duration 校验。 """ if not _ffprobe_mp4_valid(mp4_path): return False, "duration probe failed" expected = _expected_frame_count(shot) actual = _ffprobe_frame_count(mp4_path) if actual is None: if _SKIP_FFPROBE: return True, "frame-count skipped (dev)" # dev：仅 duration return False, "frame count unreadable (fail-closed)" if abs(actual - expected) > _FRAME_TOLERANCE: return False, f"trim_bypass: actual={actual} expected={expected}" return True, f"ok (frames={actual})" def _requeue_ids_from_qc(asset_bus_read) -> set: """2026-08-18 冗余审核体系 (T10): 需强制重渲的镜头集合。 来源并集： - ``fix-required`` slot（p11c 硬门 reject 打回——findings 携带 shot_ids， rollback_to 匹配 p11b） - ``failed-shots`` slot（p11c 写入的 qwen fail 镜头；operator 在 Gate 15 approve 放行的带 waived 标记，不再重渲） 读 slot 失败返回空集（普通重跑路径不受影响）。 """ requeue: set = set() try: fix_req = asset_bus_read("fix-required") or {} if isinstance(fix_req, dict) and fix_req.get("rollback_to") in ( "p11b_final_render", "p11b", ): for f in (fix_req.get("findings") or []): sid = (f or {}).get("shot_id") if isinstance(f, dict) else None if sid: requeue.add(str(sid)) except Exception: # noqa: BLE001 — 读 slot 失败不阻断（普通重跑路径） pass try: fs = asset_bus_read("failed-shots") or {} if isinstance(fs, dict): for f in (fs.get("failures") or []): if ( isinstance(f, dict) and f.get("shot_id") and not f.get("waived") ): requeue.add(str(f["shot_id"])) except Exception: # noqa: BLE001 pass # 69-03 (v3.2 WBI-03):operator 显式「重渲」指令——G15 面板经 kap # g15/ops(requeue)→ 平台 review_result.requeue_shot_ids → runner_hooks # 注入 → p11c 透传 video-qc slot。显式点名镜头无条件进重渲集(不要求 # 它在 failed 集里——重渲是主动动作,waived 与否都执行)。 try: vq = asset_bus_read("video-qc") or {} if isinstance(vq, dict): for sid in (vq.get("requeue_shot_ids") or []): if sid: requeue.add(str(sid)) except Exception: # noqa: BLE001 pass return requeue def _load_p11_progress(output_dir: str, shot_list: list | None = None) -> dict: """读取 p11_progress.json,返回 {completed_shots:set, failed_shots:set}。 文件不存在/损坏时返回空集合 (首次运行或上次崩溃未写)。读出来的 completed_shots 会进一步用 _ffprobe_mp4_valid 复核 —— 若 mp4 已损坏, 则从 completed 集合中剔除,强制重渲。 """ path = _p11_progress_path(output_dir) empty = {"completed_shots": set(), "failed_shots": set()} if not os.path.isfile(path): return empty try: with open(path, encoding="utf-8") as f: data = json.load(f) except (OSError, json.JSONDecodeError): logger.warning("%s: p11_progress.json 损坏或不可读,忽略已有进度", PHASE_ID) return empty completed = set(data.get("completed_shots") or []) failed = set(data.get("failed_shots") or []) # P0-4: 复核 completed 镜头时叠加帧数==期望校验 (闭合 trim_bypass)。 # shot 元数据不可用时退化为仅 duration 复核 (保留原防线语义)。 shot_by_sid = { (shot.get("shot_id") or shot.get("id")): shot for shot in (shot_list or []) if isinstance(shot, dict) and (shot.get("shot_id") or shot.get("id")) } still_valid = set() for sid in completed: mp4 = os.path.join(output_dir, f"{sid}.mp4") shot = shot_by_sid.get(sid) ok = ( _mp4_valid_with_frames(mp4, shot)[0] if shot is not None else _ffprobe_mp4_valid(mp4) ) if ok: still_valid.add(sid) dropped = completed - still_valid if dropped: logger.info( "%s: checkpoint 复核发现 %d 个镜头的 mp4 已失效,将重渲: %s", PHASE_ID, len(dropped), sorted(dropped), ) return {"completed_shots": still_valid, "failed_shots": failed} def _save_p11_progress(output_dir: str, completed: set, failed: set) -> None: """原子写入 p11_progress.json (先写临时文件再 rename,避免半写)。 每完成一个镜头都会调用,所以必须轻量 + 原子。写失败只警告不抛出 (进度记录是尽力而为的优化,不能因为记进度失败而中断渲染)。 """ path = _p11_progress_path(output_dir) try: os.makedirs(os.path.dirname(path), exist_ok=True) payload = { "completed_shots": sorted(completed), "failed_shots": sorted(failed), "last_updated": datetime.now(timezone.utc).isoformat(), } tmp = path + ".tmp" with open(tmp, "w", encoding="utf-8") as f: json.dump(payload, f, ensure_ascii=False, indent=2) os.replace(tmp, path) # 原子 rename except OSError: logger.warning("%s: 写入 p11_progress.json 失败 (不影响渲染)", PHASE_ID, exc_info=True) class BgmTriggerError(RuntimeError): """Raised by _r11_precheck when a shot's prompt contains BGM trigger words. SHARED-01 (Phase 10-01) upgrades BGM from warn → REJECT. The dispatch loop catches this, writes a fix-required entry, and re-raises so the runner retries. The error message lists ONLY the matched words + shot_id — NOT the full prompt (T-10-01-03 mitigation: avoid leaking unrelated prompt context). """ def __init__(self, shot_id: str, matched_words: tuple[str, ...]) -> None: self.shot_id = shot_id self.matched_words = matched_words words_str = ", ".join(repr(w) for w in matched_words) super().__init__( f"BGM TRIGGER (shot={shot_id}): prompt contains banned word(s) " f"[{words_str}]. These trigger LTX audio VAE to generate background " f"music even with negative-prompt blocking. Rewrite the prompt " f"using physical sound verbs (e.g. 'speaks softly' instead of " f"'hums a tune'). Source: shared/bgm_blacklist.yaml." ) def _r11_precheck(shot: dict) -> list[str]: """Scan shot prompt for R11 physics + BGM violations. Returns a list of warning strings for PHYSICS conflicts. Empty list means no physics warnings. RAISES ``BgmTriggerError`` on BGM trigger words (SHARED-01 Phase 10-01: BGM upgraded from warn → reject). P09 sets ``shot['physics_state']`` (e.g. 'driving-straight') so we can detect action-physics mismatch — when missing, only BGM triggers and self-contradictions are detectable. """ prompt = _get_vp(shot) if not prompt: return [] prompt_lower = prompt.lower() warnings: list[str] = [] state = (shot.get("physics_state") or "").strip().lower() if state in _R11_PHYSICS_CONFLICTS: for word in _R11_PHYSICS_CONFLICTS[state]: if word.lower() in prompt_lower: warnings.append( f"PHYSICS CONFLICT: physics_state='{state}' but prompt " f"contains '{word}'. Rewrite the prompt action to match " f"the reference image physics, OR change physics_state." ) # SHARED-01: BGM trigger words → RAISE (not warn). # Task 6c note: bgm_blacklist.yaml 只列动词形式 (humming/hums/hummed, 不含 # 词干 hum)，故 "ventilation hum"/"electrical hum" 等机械名词不会误报。子串 # 匹配有已知边界——"hummingbird" 会误命中 "humming" (罕见，prompt 回避即可)； # 不改用 \b 词边界是因为中文禁词 (旋律/哼唱) 在 Python re 下无词边界，会漏检。 bgm_data = _get_bgm_blacklist() banned_words = bgm_data.get("bgm_triggers") or [] matched: list[str] = [] for word in banned_words: if word.lower() in prompt_lower: matched.append(word) if matched: shot_id = shot.get("shot_id") or shot.get("id") or "?" raise BgmTriggerError(shot_id, tuple(matched)) return warnings # --------------------------------------------------------------------------- # Goal building blocks (H3 only — 2026-08-05: H3 是唯一本地视频引擎) # # H3 遵循 video-engine-adapter skill 的规格:CFG=1.0 → 正面 diegetic BGM ban, # 原生音频 (ttsAudio 字段注入 TTS 对白)。不设 LTX fallback。 # --------------------------------------------------------------------------- def _build_r11_safety_block() -> str: """R11 SAFETY RULES 文本 (H3 only)。 H3 用 CFG=1.0 → 负面词无效,BGM ban 靠正面 diegetic 语言 ("strictly diegetic in-world sound, unscored scene, no scored music")。 PHYSICS MATCH (动作/音频必须匹配参考图物理状态)。 """ return ( " R11 SAFETY RULES (MANDATORY — H3 engine):\n" "1. PHYSICS MATCH: The prompt's action/audio MUST match the reference image physics. " "If the reference image shows a car driving straight, write 'driving straight' — NEVER " "'turning a corner' or 'turn-signal click'. Physics mismatch causes distorted video.\n" "2. BGM BAN (POSITIVE DIEGETIC — H3-specific): H3 uses CFG=1.0, so NEGATIVE prompts DO " "NOT work. Ban background music by writing POSITIVE diegetic in-world sound language. " "Append to the POSITIVE prompt: " "\"strictly diegetic in-world sound, unscored scene, no scored music\". " "NEVER write 'no music' in the positive prompt — 实测会把所有音频 (含环境音和对白) " "压到 -57dB. Also avoid trigger verbs (humming/singing/melody/song/tune/lullaby) — " "these bias the H3 audio VAE toward music even at CFG=1.0.\n" "3. NO NEGATIVE-PROMPT RELIANCE: H3 ignores negative_prompt (CFG=1.0). All content " "control is via the POSITIVE prompt only — do not rely on a negative-prompt wordlist.\n" "4. SINGLE-PERSON: If only one character should be visible, anchor with 'sits ALONE' " "and 'NO other person' in the POSITIVE prompt.\n" "5. DIALOGUE: Speaking/whispering descriptions are safe. Humming/singing are NOT " "(biases audio toward music). Use 'speaks softly to the baby' NOT 'hums a lullaby'." ) # H3 task-step: 指示 subagent 加载 video-engine-adapter skill 做 7 步输入适配。 _H3_ENGINE_ADAPTER_INSTRUCTION = ( "ENGINE ADAPTER: Load skill video-engine-adapter (skill_view name='video-engine-adapter'). " "Follow its 7-step adaptation workflow to convert this shot's universal input " "(prompt, references, duration, audio_intent) into H3-specific API parameters. " "The adapter skill knows H3's n%17==5 frame grid, 32-multiple resolution constraint, " "diegetic positive prompt style (CFG=1.0, negative prompt ineffective), and ref2va " "reference image mapping (character turnaround + scene background → refImages[], ≤9).\n\n" "SAMPLER: P11b uses Native KSampler + SigmaShift, 15 steps (not T8 DualClock, not Turbo LoRA). " "Profile=native-sage → SageAttention global, TESpeed disabled. " "Steps=15 is the verified optimal quality/speed point for 1216×672 production renders " "(Kai A/B 2026-08-09; emitted by the in-process H3VideoEngine as profile=native-sage + " "native=true + tespeed=false + steps=15).\n\n" "KB REFERENCE (Cinema KB — field/routing cross-reference; the adapter skill above is authoritative):\n" "POST http://localhost:10588/api/v1/cinema/query\n" ' - {"category": "engine_field_spec", "key_name": "h3.generate."} → H3 API field ' "documentation (the same fields the adapter skill encodes; use for cross-reference).\n" ' - {"category": "engine_selection_rule", "key_name": ""} → engine routing rules ' "(which engine per shot-type); consult when the shot's `engine` field is ambiguous)." ) def _build_h3_audio_block( shot_id: str | None, voice_clip: dict, custom_audio_arg: str | None, ) -> str: """Build the per-shot H3 audio injection block (ttsAudio multipart 字段). H3 通过 ``ttsAudio`` 字段接收 TTS 对白音频 (API 原生支持)。决策二元: 有 TTS → 上传 ttsAudio (对白/旁白保真);无 TTS → 不传,让 H3 原生 audio VAE 生成环境音 (CFG=1.0,总是出声)。 """ clip_type = (voice_clip or {}).get("clip_type", "") vc_engine = (voice_clip or {}).get("engine", "") if custom_audio_arg: # 有真实 TTS 音频 → 作为 ttsAudio 上传 (对白/旁白保真) return ( f"\n\nH3 AUDIO (ttsAudio multipart field):\n" f"This shot has a TTS audio_path for shot {shot_id}. Upload it as the " f"``ttsAudio`` multipart field to the H3 API. H3 mixes the TTS dialogue " f"with its native ambient bed.\n" f" -F 'ttsAudio=@{custom_audio_arg}'\n" f"Voice-clip clip_type='{clip_type}', engine='{vc_engine}'.\n" f"BGM ban reminder: append POSITIVE diegetic language " f"(\"strictly diegetic in-world sound, unscored scene, no scored music\") — " f"H3 negative_prompt is ineffective (CFG=1.0)." ) # 无 TTS (sfx/ambient/needs_review/无 clip) — 不上传 ttsAudio,让 H3 原生成环境音 return ( f"\n\nH3 AUDIO (ttsAudio multipart field): shot {shot_id} has no TTS audio to " f"upload (clip_type='{clip_type}', engine='{vc_engine}'). Let H3's native audio " f"VAE generate the ambient bed. Ban BGM via POSITIVE diegetic language " f"(negative_prompt ineffective at CFG=1.0)." ) def _engine_constraints() -> list[str]: """CONSTRAINTS list (H3 only). H3: BGM ban 改为正面 diegetic 约束 + engine='h3' (CFG=1.0,无负面词依赖)。 """ return [ "MUST match prompt action/audio to reference image physics_state (R11 Physics)", "MUST NOT let positive prompt contain humming/singing/melody/song/tune/vocalizing (R11 BGM BAN — H3 trigger verbs bias audio VAE toward music)", "MUST append POSITIVE diegetic BGM-ban language: \"strictly diegetic in-world sound, unscored scene, no scored music\" (H3 CFG=1.0 → negative_prompt INEFFECTIVE; NEVER write 'no music' in the positive prompt)", "MUST anchor single-person shots with 'sits ALONE' + 'NO other person' in the POSITIVE prompt (R11 Single-person)", "MUST use speaking/whispering (safe), NEVER humming/singing (triggers BGM)", "MUST output engine ('h3') and murch_grade in video_clip for canvas structured fields", "MUST record take_log entry per shot: take_n + changed_variable + seed + verdict (keep|fix_in_post|edit|re_roll|rewrite) + evidence — enables systematic failure detection (same flaw twice = rewrite not re-roll)", "MUST record observed_end_state per shot: character_position + character_pose + camera_end_position + environment_state + open_motion_vectors — this is what the video ACTUALLY shows (observed footage overrides planned state for downstream continuity). ALSO record tts_audio_path per shot when voice-clips supplied TTS for that shot (B1 — enables downstream P12 audio audit + lipdub verification)", ] # --------------------------------------------------------------------------- # B1: per-shot TTS audio decision (H3 only) # # H3 音频决策是二元的,没有 audio_mode / dialogue_end_time / freeze chain # (那些都是已移除的 LTX 概念): # 有 TTS (clip_type in dialogue|narration + 真实 audio_path) → 上传 ttsAudio # 无 TTS (sfx/ambient/needs_review/无 clip) → 不传 # # 返回 (custom_audio_arg, None):custom_audio_arg 为 TTS 路径或 None, # 第二位恒为 None (保留 tuple 形态,无 audio_mode 语义)。 # P10 voice-clips 提供 per-shot {audio_path, clip_type, engine, duration} 驱动决策。 # --------------------------------------------------------------------------- def _decide_audio_injection(voice_clip: dict | None) -> tuple[str | None, None]: """Return ``(custom_audio_arg, None)`` for one shot's voice-clip (H3 only). H3 音频二元决策:有 TTS → 上传 ttsAudio (``custom_audio_arg``=path); 无 TTS → 不传 (``custom_audio_arg``=None)。第二位恒为 None — audio_mode / dialogue_end_time 是已移除的 LTX 概念,保留 None 占位以维持 tuple 解包形态。 needs_review engine + sfx/ambient clip_type + 无 audio_path → 均视为「无 TTS」。 """ if not isinstance(voice_clip, dict): return (None, None) # 无 voice-clip → 不传 ttsAudio,让 H3 原生生成环境音 engine = (voice_clip.get("engine") or "").strip() clip_type = (voice_clip.get("clip_type") or "").strip() audio_path = (voice_clip.get("audio_path") or "").strip() # needs_review → 不上传 (未审核 TTS 不可信,避免注入不可信对白) if engine == "needs_review": return (None, None) # 有真实 TTS (qwen_tts 等) 对白/旁白音频 → 上传 ttsAudio if audio_path and clip_type in ("dialogue", "narration"): return (audio_path, None) # 无 TTS (sfx/ambient/缺 audio_path) → 不传 return (None, None) def _case_insensitive_replace(text: str, needle: str, replacement: str) -> str: """Replace all case-insensitive occurrences of `needle` in `text`. Uses re.escape + re.IGNORECASE so the banned word is matched regardless of case (e.g. 'Hums', 'HUMS', 'hums'). Whitespace squash on adjacent spaces is handled by the caller (strip). """ import re as _re if not needle: return text pattern = _re.compile(_re.escape(needle), _re.IGNORECASE) return pattern.sub(replacement, text) def _apply_selected_variants(entry: dict, output_dir: str) -> None: """Apply p11a0_iframe_qc's per-slot variant selection onto the path fields. Phase A 选优消费 (docs/iframe-qc-design.md §3.3): 当 entry 带 ``selected_first_variant`` / ``selected_last_variant`` (p11a0 判定回写, 1-based int) 时, 把 ``first_frame_path`` / ``last_frame_path`` 重指向 ``all_first_frames[selected-1]`` / ``all_last_frames[selected-1]``。 坐标约定 (2026-08-16 审验缺陷 1): manifest 内的变体路径是相对 **episode 根**的 (如 ``assets/P11/iframes_S01_B01/first_frame_v2.png``), 而 ``output_dir`` 恒为 ``/assets/P11`` (run() 用 ``os.path.join(os.path.curdir, "assets", "P11")`` 构造)。因此相对路径的 解析基座是 ``dirname(dirname(abspath(output_dir)))`` = episode 根 — 首版误用 ``dirname(output_dir)`` (= ``assets``) 拼出 ``assets/assets/P11/...`` → isfile 永远 miss → 永远静默回退 v1, 选优功能实际失效 (防御性回退掩盖了 bug)。 防御性回退 (advisory 原则 — 选优失败不得阻塞渲染): 未选 (None) / 非法值 / 越界 / 文件不在盘 → 保持原 ``*_frame_path`` (v1) + warn。 绝对路径原样接受 (isfile 校验后直接写入)。 """ # abspath 先归一: output_dir 可能是 curdir-joined 相对串 ("assets/P11"), # dirname² 前先取绝对路径才能落到真实 episode 根 (同 _p11_progress_path)。 episode_root = os.path.dirname(os.path.dirname(os.path.abspath(output_dir))) for sel_key, all_key, path_key in ( ("selected_first_variant", "all_first_frames", "first_frame_path"), ("selected_last_variant", "all_last_frames", "last_frame_path"), ): sel = entry.get(sel_key) if not isinstance(sel, int) or isinstance(sel, bool) or sel < 1: continue # 未选 (None) / 非法 → 未评估或无 pass, 保持 v1 paths = entry.get(all_key) if not isinstance(paths, list) or sel > len(paths): logger.warning( "%s: %s=%s 越界 (all_* %s 条) — 回退 v1", PHASE_ID, sel_key, sel, len(paths) if isinstance(paths, list) else 0, ) continue chosen = paths[sel - 1] if not isinstance(chosen, str) or not chosen: logger.warning("%s: %s 指向空路径 — 回退 v1", PHASE_ID, sel_key) continue chosen_abs = ( chosen if os.path.isabs(chosen) else os.path.normpath(os.path.join(episode_root, chosen)) ) if not os.path.isfile(chosen_abs): logger.warning( "%s: 选中变体文件不在盘 (%s) — 回退 v1", PHASE_ID, chosen_abs, ) continue entry[path_key] = chosen def _load_iframe_manifest(output_dir: str) -> dict[str, dict]: """Load ``assets/P11/iframe-manifest.json`` (gen_iframes.py output) by shot_id. Returns ``{shot_id: {iframe_mode, frame_chain_from, first_frame_path, last_frame_path}}``. Empty dict when the manifest is absent (tests, first run, or single_ref episodes that never generated frames). Defensive: the on-disk shape is a bare list, but the asset-bus ``{"value": [...]}`` envelope is also tolerated. p11a0 选优消费: entry 带 ``selected_*_variant`` 时, 加载即应用选优 (路径重指向选中变体; 缺选/越界/文件缺失回退 v1 — 见 ``_apply_selected_variants``), 下游 setdefault 合并拿到的已是选后路径。 """ manifest_path = os.path.join(output_dir, "iframe-manifest.json") if not os.path.isfile(manifest_path): return {} try: with open(manifest_path, encoding="utf-8") as f: data = json.load(f) except (OSError, json.JSONDecodeError): logger.warning( "%s: iframe-manifest.json unreadable; skipping iframe field merge", PHASE_ID, ) return {} if isinstance(data, dict) and isinstance(data.get("value"), list): data = data["value"] if not isinstance(data, list): return {} index: dict[str, dict] = {} for entry in data: if isinstance(entry, dict) and entry.get("shot_id"): _apply_selected_variants(entry, output_dir) index[entry["shot_id"]] = entry return index def _resolve_path_safe(path: str, workdir: str) -> str | None: """Resolve a shot reference path to an existing absolute file, else None. Used by the upstream-asset existence check (FIX-4): a shot's ``scene_ref`` and ``character_refs[].turnaround_path`` may be absolute or relative to the run workdir, and may point at a file that was declared by an upstream phase but never materialized on disk. Returns the absolute path when the file exists, None otherwise (so callers can flag the missing reference). """ if not path or not isinstance(path, str): return None path = path.strip() if not path: return None if os.path.isabs(path): return path if os.path.isfile(path) else None for base in (workdir, os.getcwd()): cand = os.path.join(base, path) if os.path.isfile(cand): return cand return None # ── PPL FIX-3 (2026-08-25): engine-render prompt provenance ────────────── # 过程记录断层的 P11b 侧修复：每次真实 engine.render 的最终 prompt/参数/ # 引用图/产物路径落 Prompt Ledger（生成器调用层强制溯源），并把 # prompt/negative_prompt 补写回 {sid}_meta.json —— 这是手动渲染路径 # (templates/ltx_msr_single_shot_render.py) 一直有、引擎路径丢失的字段。 # 全部 degrade-tolerant：溯源是审计产物，绝不是门。 def _record_video_provenance( shot: dict, sid: str, output_path: str, final_prompt: str, ref_inputs: list[str], mode: str, engine: Any, result: Any, status: str, elapsed_sec: float, error: str = "", ) -> None: """Ledger one ``engine.render`` call (PPL FIX-3.1, degrade-tolerant). No ledger installed (tests / phase runs outside the runner) → no-op. ``record()`` already swallows write failures; the local try/except is belt-and-braces for payload assembly (铁律: 日志失败绝不阻塞管线). """ ledger = get_ledger() if ledger is None: return try: params: dict[str, Any] = { "seed": shot.get("seed") or 0, "resolution": ( shot.get("resolution_override") or shot.get("resolution") or "1280x704" ), "duration_sec": shot.get("duration_sec"), } if isinstance(result, dict) and result.get("prompt_id"): params["comfy_prompt_id"] = result["prompt_id"] if error: params["error"] = str(error)[:500] ledger.record( event="gen_call", phase_id=PHASE_ID, asset_type="video", asset_id=sid, engine=(result.get("engine") if isinstance(result, dict) else None) or getattr(engine, "ENGINE_NAME", "") or "h3", mode=mode, final_prompt=final_prompt, negative_prompt=shot.get("negative_prompt") or "", params=params, ref_inputs=list(ref_inputs), output_path=output_path, captured_via="engine_hook", status=status, duration_sec=round(elapsed_sec, 1), ) except Exception: logger.debug( "%s: prompt-ledger video record failed for %s — swallowed", PHASE_ID, sid, exc_info=True, ) def _write_render_meta( sid: str, output_path: str, *, final_prompt: str, negative_prompt: str, shot: dict, result: Any, elapsed_sec: float, ) -> None: """Merge prompt provenance into ``{sid}_meta.json`` (PPL FIX-3.2). Merge semantics: an existing (delegate-written) meta file is kept and only gains/refreshes provenance keys; a missing one is created with the standard fields the cache reader (``_try_cache_hit``) expects plus ``prompt`` / ``negative_prompt`` — restoring the field set the manual render path always wrote. Atomic (tmp + os.replace). Swallowed on any failure (audit artifact, never a gate). """ try: meta_path = os.path.join( os.path.dirname(output_path) or ".", f"{sid}_meta.json", ) meta: dict[str, Any] = {} if os.path.isfile(meta_path): try: with open(meta_path, encoding="utf-8") as f: loaded = json.load(f) if isinstance(loaded, dict): meta = loaded except Exception: logger.debug( "%s: unreadable meta %s — rebuilding provenance keys only", PHASE_ID, meta_path, exc_info=True, ) meta = {} meta.setdefault("shot_id", sid) meta.setdefault("engine", "h3") meta.setdefault("video_path", output_path) meta.setdefault("duration_sec", shot.get("duration_sec") or 0) meta.setdefault( "resolution", shot.get("resolution_override") or shot.get("resolution") or "1280x704", ) meta.setdefault("seed", shot.get("seed") or 0) if isinstance(result, dict) and result.get("frame_count"): meta.setdefault("frame_count", result["frame_count"]) meta.setdefault("render_time", round(elapsed_sec, 1)) # The two provenance keys this function exists for — same values the # ledger entry carries (同源同值). meta["prompt"] = final_prompt meta["negative_prompt"] = negative_prompt tmp_path = meta_path + ".ppl.tmp" with open(tmp_path, "w", encoding="utf-8") as f: json.dump(meta, f, indent=2, ensure_ascii=False) os.replace(tmp_path, meta_path) except Exception: logger.debug( "%s: meta provenance write failed for %s — swallowed", PHASE_ID, sid, exc_info=True, ) # --------------------------------------------------------------------------- # run() — the orchestration entry point (D-36-08 parallel_shots extension) # --------------------------------------------------------------------------- def run( episode_id: str, asset_bus_read: Callable[[str], Any], asset_bus_write: Callable[[str, dict], None], delegate_task: Callable[[str, str, list[str]], dict], trigger_gate: Callable[[str, str], dict] | None = None, sync_canvas: Callable[[str, str, dict], None] | None = None, *, parallel_shots: int = MAX_CONCURRENT_GL, # GLM rate-limit-safe (2026-07-01) episode_config: dict | None = None, sync_shot_progress: Callable[[str, str, dict], None] | None = None, ) -> dict: """Execute phase p11 (V8.6 Step 10 + Step 11 video render, parallel shots). Args: episode_id: Episode identifier. asset_bus_read: Callable(slot) -> data (injected; tests pass mock). asset_bus_write: Callable(slot, entry) -> None (injected). delegate_task: Callable(goal, context, toolsets) -> dict with a "summary" key whose value is text containing a fenced JSON block (D-35-07 contract). Called once PER SHOT in this phase. trigger_gate: Optional Callable(gate_id, episode_id) -> dict. When provided, Gate 7 "render-preview" is triggered after the asset-bus writes; when ``None`` no gate is triggered. sync_shot_progress: 逐镜增量画布同步（2026-08-26）。runner 在 on_phase_complete 注册时 **不**注入 sync_canvas（防 phase 完成 双同步），本 kwarg 是 partial 专用通道——每镜渲染+校验通过后 即刻触发，资产管理中心实时可见；缺省 None 时回落 sync_canvas （非 runner 调用场景）。 parallel_shots: Shot-level parallelism (D-36-08). Defaults to ``MAX_CONCURRENT_GL`` (= 3 unless ``GLM_MAX_CONCURRENT`` env var set), the GLM rate-limit-safe value. p11 is the ONLY phase that accepts this kwarg — fans out per-shot delegate_task calls via ThreadPoolExecutor(max_workers=...). The global GLM semaphore (``@glm_concurrent`` on ``_dispatch_shot``) caps actual concurrent heavy GLM calls regardless of this value. Tests pass ``parallel_shots=1`` for sequential determinism. Returns: ``{"phase": "p11_video_render", "outputs": {...}, "gate": {...} | None}`` where ``outputs`` carries the aggregated expert payload (video_clips + lip_sync_reports arrays across all shots). """ # 1. Gather inputs (graceful when slot empty — first run / tests). shot_list = asset_bus_read("shot-list") or [] scene_images = asset_bus_read("scene-images") or {} character_assets = asset_bus_read("character-assets") or {} voice_clips = asset_bus_read("voice-clips") or [] # #5 (2026-08-15): transition-design → tidx，驱动 [CONTINUITY-IN] 注入。 # 缺 slot / 形状不对 → 空 dict → 不注入（现状降级）。 _tidx = _build_tidx_from_transition_design(asset_bus_read("transition-design")) # Measure the real TTS audio duration per shot via ffprobe and override # each shot's duration_sec so the rendered clip matches the actual spoken # audio. Inlines the ffprobe logic formerly in the P10a shot_timeline # phase (removed 2026-08-09): the per-shot video render is 1:1 with its # voice clip, so the measurement belongs here where both are read. # P0 #4 (2026-08-14 review): a shot may carry MULTIPLE dialogue elements # — index list-per-shot, not dict-of-one. The duration override for a # multi-clip shot uses the SUM of all its clips' TTS durations (the # elements play back sequentially), so the shot is long enough for every # line. Single-clip shots are byte-identical to the previous behavior. if isinstance(voice_clips, list) and isinstance(shot_list, list): _vc_by_shot: dict[str, list[dict]] = {} for _vc in voice_clips: if not isinstance(_vc, dict): continue _vsid = _vc.get("shot_id") or _vc.get("id") if isinstance(_vsid, str) and _vsid: _vc_by_shot.setdefault(_norm_shot_id(_vsid), []).append(_vc) _measured = 0 for _shot in shot_list: if not isinstance(_shot, dict): continue _sid = _shot.get("shot_id") or _shot.get("id") _vcs = _vc_by_shot.get(_norm_shot_id(_sid)) if isinstance(_sid, str) else None _durs = [ _ffprobe_audio_duration( (_vc.get("audio_path") or _vc.get("clip_path") or ""), ) for _vc in (_vcs or []) ] _durs = [d for d in _durs if d is not None] if _durs: _tts = sum(_durs) # Voice present + ffprobe ok → TTS duration + tail buffer. _shot["duration_sec"] = _clamp_duration(_tts + _TTS_TAIL_BUFFER_SEC) # 改造 1: record per-clip TTS positioning so P12b can place each # TTS at its precise absolute master-timeline offset. When the # shot has exactly one clip this is unchanged (path + its # start_time_sec). Multi-clip shots leave tts_audio_path unset # (p11b records ONE positioning per shot; p12b then derives # placement per element from the voice-clips themselves — # see _assemble_voice_track's list-based tts_by_shot). if len(_vcs) == 1: _vc = _vcs[0] _audio = _vc.get("audio_path") or _vc.get("clip_path") or "" _shot["tts_audio_path"] = _audio _shot["tts_audio_offset_sec"] = ( _safe_duration_sec(_vc.get("start_time_sec")) if isinstance(_vc, dict) else 0.0 ) _measured += 1 else: # No voice / ffprobe failed → keep P09 duration, floored. _p09 = _safe_duration_sec(_shot.get("duration_sec")) _shot["duration_sec"] = _clamp_duration(max(_p09, _SILENT_MIN_SEC)) if _measured: logger.info( "%s: ffprobe-measured TTS duration overrode duration_sec for %d/%d shots", PHASE_ID, _measured, len(shot_list), ) # SHARED-01 (Phase 10-01): BGM reject retry-then-strip path. # When fix-required.rollback_to == PHASE_ID (a previous BgmTriggerError- # triggered retry), STRIP banned BGM words from every shot's video_prompt # BEFORE dispatch — prevents infinite loop. The runner's outer retry loop # re-invokes run() with fix-required set; we consume it here. After # stripping, the precheck passes and dispatch proceeds normally. fix_required_in = asset_bus_read("fix-required") or None is_p11_retry = bool( fix_required_in and isinstance(fix_required_in, dict) and fix_required_in.get("rollback_to") == PHASE_ID ) if is_p11_retry and isinstance(shot_list, list): try: banned = _get_bgm_blacklist().get("bgm_triggers") or [] except Exception: banned = [] stripped_count = 0 for shot in shot_list: if not isinstance(shot, dict): continue prompt = _get_vp(shot) if not prompt: continue new_prompt = prompt for word in banned: # Case-insensitive replacement — both en and zh triggers. new_prompt = _case_insensitive_replace(new_prompt, word, "") if new_prompt != prompt: _set_vp(shot, new_prompt) stripped_count += 1 if stripped_count: logger.warning( "%s: BGM retry path — stripped banned words from %d shot(s) " "to avoid infinite loop.", PHASE_ID, stripped_count, ) # Clear fix-required so downstream sees a clean state. asset_bus_write("fix-required", {}) # B1: index voice-clips by shot_id for O(1) per-shot lookup. Each entry # is {shot_id, audio_path, clip_type, engine, ...} per p10_voice schema. # The lookup drives --custom-audio + --audio-mode injection into the LTX # render command (Task A 0a0b02c freeze chain). # P0 #4 (2026-08-14 review): a shot may carry multiple dialogue elements; # the FIRST clip wins the H3 ttsAudio slot (H3 takes one audio bed per # shot). Full per-element placement happens in p12b's # _assemble_voice_track (list-based tts_by_shot), so later elements are # not lost — they are only absent from the H3 native audio bed. voice_clips_by_shot: dict[str, dict] = {} if isinstance(voice_clips, list): for vc in voice_clips: if isinstance(vc, dict) and vc.get("shot_id"): voice_clips_by_shot.setdefault(vc["shot_id"], vc) # Serialize the shared (non-shot-specific) payload once. Each per-shot # goal embeds these so the subagent has full context for that shot. si_json = json.dumps(scene_images, ensure_ascii=False) ca_json = json.dumps(character_assets, ensure_ascii=False) def _goal_for_shot(shot: dict, shot_idx: int) -> tuple[str, str, bool]: """Build (goal, context, has_frame_pair) for one shot's delegate_task call. Returns ``has_frame_pair`` (always False for H3 — H3 uses ref2va reference images, not a first/last-frame conditioning chain) so the dispatcher interface is unchanged. CRITICAL (dialogue injection fix): The shot payload may carry a ``dialogue_text`` field (from p09 shot_breakdown per CONTRACT-02) containing the pure spoken line for TTS in Chinese. The video engine generates audio + video simultaneously from the text prompt — dialogue MUST be embedded in the prompt so the model produces synchronized speech + lip sync. Two formats (per LTX validation cases in ~/文档/LTX/validition_v1/): - Format A (on-screen dialogue): embed in narration: ``老林低声说："等一下。"`` - Format B (voiceover/narration): structured tag: ``【@角色】说话，台词内容：xxx。音色：xxx。表演要求：xxx。`` If ``dialogue_text`` is absent / None / whitespace-only, skip dialogue injection. P09 also emits ``dialogue_note`` (stage direction / SFX description) which P11 IGNORES for dialogue injection — dialogue_note is never TTS-speakable text. """ # CONTRACT-03 (09-02): presence check on dialogue_text replaces the # pre-v2.0 keyword scan over the ambiguous pre-v2.0 single-cue field. P09 now # separates dialogue_text (pure spoken line) from dialogue_note # (stage direction) — humming-only dialogue_text is impossible: P09 # emits dialogue_text=None for non-speech shots (humming lives in # dialogue_note, which P11 ignores for TTS injection). dialogue_text = shot.get("dialogue_text") # B1: per-shot TTS audio decision (H3 ttsAudio multipart field). # voice_clips_by_shot is captured from run()'s asset_bus_read("voice-clips"). # H3 audio is binary: 有 TTS → 上传 ttsAudio, 无 TTS → 不传。 shot_id_for_vc = shot.get("shot_id") or shot.get("id") voice_clip_for_shot = voice_clips_by_shot.get(shot_id_for_vc, {}) custom_audio_arg, _ = _decide_audio_injection(voice_clip_for_shot) # H3 引擎: 用 ttsAudio multipart 字段注入 TTS 对白音频。 custom_audio_block = _build_h3_audio_block( shot_id_for_vc, voice_clip_for_shot, custom_audio_arg, ) # CONTRACT-03 (09-02): has_speech is now a pure presence check on # dialogue_text. T-09-02-01 mitigation — eliminates the keyword-scan # tampering vector where a dialogue_note containing Chinese keywords # (说/台词/旁白) could trigger speech injection for a silent shot. has_speech = bool(dialogue_text and dialogue_text.strip()) dialogue_instruction = ( f" DIALOGUE INJECTION (CRITICAL): This shot's dialogue_text contains " f"spoken dialogue: \"{dialogue_text}\". You MUST embed the dialogue " f"into the video prompt text using 【@角色】format or narrative " f"embedding. Include voice characteristics (age/gender/tone) and " f"performance direction. The video engine generates synchronized " f"audio + lip sync from the prompt — do NOT leave dialogue for " f"post-production." if has_speech else "" ) # R11 SAFETY INJECTION — mandatory for ALL shots (H3: positive diegetic). r11_safety = _build_r11_safety_block() # R11 PER-SHOT PRE-CHECK — concrete warnings for THIS shot's prompt. # Surfaces the exact conflict words so the subagent doesn't have to # grep its own prompt against the abstract rules above. Empty when # the prompt is clean. r11_warnings = _r11_precheck(shot) if r11_warnings: r11_warning_block = ( "\n\n⚠️⚠️⚠️ R11 PRE-CHECK WARNINGS FOR THIS SHOT — " "DO NOT SUBMIT UNTIL RESOLVED:\n" + "\n".join(f" - {w}" for w in r11_warnings) + "\nResolve by rewriting the video_prompt (preferred) OR by " "changing physics_state. Do not silently proceed.\n" ) else: r11_warning_block = "" # H3 不用首尾帧条件 (ref2va 用参考图,非首尾帧 conditioning chain)。 # has_frame_pair 始终为 False,仅保留以维持 _dispatch_shot 接口不变。 has_frame_pair = False task_steps = [ f"ENGINE: Use H3 (MiniMax) via KAP unified pipeline API :10588 as the DEFAULT video engine (DEFAULT_VIDEO_ENGINE='{DEFAULT_VIDEO_ENGINE}', endpoint POST /api/production/minimax-h3/generate)", _H3_ENGINE_ADAPTER_INSTRUCTION, "PROMPT RULES: H3 uses CFG=1.0 — negative_prompt is INEFFECTIVE. All content/BGM control is via the POSITIVE prompt (diegetic in-world sound language). See video-engine-adapter skill references/engine-specs.md + adaptation-rules.md.", "Load both experts via skill_view", "visual_executor renders video clip from scene-images + character-assets using the H3 ref2va workflow (character turnaround + scene background → refImages[])", f"{r11_warning_block}" if r11_warning_block else "No R11 pre-check warnings for this shot", f"{dialogue_instruction}" if dialogue_instruction else "No speech/dialogue injection needed", f"R11 SAFETY (mandatory):\n{r11_safety}", f"{custom_audio_block}", "SKIP LOGGING (issue-6): After all shots are attempted, write\n" "assets/P11/skip_log.json with\n" " {\"skipped_shots\": [\"S5_001\", ...],\n" " \"reason\": {\"S5_001\": \"H3 API 400 x2\"}}\n" "so P12 can identify which shots have no rendered video and skip them\n" "during final assembly instead of crashing.", ] # p11_video（iframe/video_render）generation_config knob 已于 2026-08-16 # 退役：P1-2 引擎化后子代理被禁止渲染（见下方 execution_hint），宿主 # _run_engine_render 每 shot 单次 engine.render —— 该注入指令两端均 # 无法兑现。真实现 best-of-N 需 engine-level fan-out，另行立项。 # P1-2 (engine-adapter): the actual H3 render no longer happens in the # goal — the host pipeline calls H3VideoEngine (multipart POST to # :10588/api/production/minimax-h3/generate) directly and frame-validates # the result (P0-4). The retired curl template lives on as the pure # helper ``_form_factor_render_args_h3`` (kept for tests / audit); it is # NOT emitted into the goal, so subagents no longer hand-write # render_S*.py wrappers or shell out to curl for the render. execution_hint_str = ( "RENDER IS NOW IN-PROCESS (P1-2 engine-adapter): the actual H3 video render " "no longer happens in this goal. The host pipeline calls the H3VideoEngine " "(multipart POST to :10588/api/production/minimax-h3/generate) directly, " "downloads the clip, and frame-validates it (P0-4 nb_frames == expected). " "Do NOT write or run a render_S*.py wrapper, and do NOT shell out to " "curl/ffmpeg to render the shot.\n\n" "YOUR JOB IS CREATIVE + METADATA ONLY:\n" "1. Produce the best H3-ready prompt phrasing for this shot — the host has " "already run the local-template enhancer; refine for physics/audio match.\n" "2. Record the per-shot creative metadata the engine CANNOT infer:\n" " - take_log {take_n, changed_variable, seed, verdict, evidence}\n" " - observed_end_state {character_position, character_pose, " "camera_end_position, environment_state, open_motion_vectors[]}\n" " - murch_grade (excellent|pass|weak|fail)\n" " - lip_sync_report {alignment_score, mouth_frames_analyzed, issues}\n" "3. Report video_path as assets/P11/{shot_id}.mp4 (the engine's canonical " "output name) so downstream aggregation + P14 audit stay consistent.\n" "FILENAME CONSISTENCY (P1-G1): the engine names every MP4 `{shot_id}.mp4` " "verbatim (filenamePrefix={shot_id}); scene-qualified naming breaks aggregation.\n" "META.JSON CONSISTENCY (P1-G2): every MP4 should have a sibling " "`{shot_id}_meta.json` (seed, resolution, duration, frame_count, render_time).\n" + ( f"H3 AUDIO (ttsAudio, B1): this shot has a TTS audio_path ({custom_audio_arg}) " f"— the host engine uploads it as ttsAudio (H3 mixes TTS with its native " f"ambient bed, CFG=1.0).\n" if custom_audio_arg else "H3 AUDIO (ttsAudio, B1): this shot has no TTS audio_path — the host " "engine does NOT upload ttsAudio; H3 native audio VAE generates the ambient " "bed (CFG=1.0).\n" ) + "If the H3 API returns 400 for a shot, the host engine degrades that shot " "(400 = rejected, not transient) and the batch continues; record it in " "assets/P11/skip_log.json so P12 can skip it during assembly." ) goal = ( GoalBuilder(phase_id=PHASE_ID, episode_id=episode_id) .role( expert_skills=["visual_executor", "audio_pipeline"], pipeline_position=f"Step 11 — 视频渲染 (shot {shot_idx})", operation=f"H3 video render + lip_sync alignment for shot {shot_idx}", ) .context(shot=shot, scene_images=scene_images, character_assets=character_assets) .task(steps=task_steps) .execution_hint(execution_hint_str) .schema(shape={ "video_clip": "{shot_id, video_path, duration_sec, resolution, codec, " # canvas structured fields (mandatory — consumed by infinite-canvas frontend) "engine (enum: h3|jimeng|seedance|runway|kling|veo|sora), " "murch_grade (enum: excellent|pass|weak|fail), " # Take log entry — one per generation attempt (Seedance retake-protocol) "take_log: {take_n, changed_variable (what was different from last take), " "seed, verdict (enum: keep|fix_in_post|edit|re_roll|rewrite), " "evidence (one sentence why this verdict)}, " # Observed state — what the generated video ACTUALLY shows at end # (Seedance canon rule: observed footage overrides planned state) "observed_end_state: {character_position, character_pose, " "camera_end_position, environment_state, open_motion_vectors[]}, " # First/last frame provenance (gen_iframes.py output, merged from # iframe-manifest.json at collection time — see _merge_iframe_fields). # first_frame_path/last_frame_path: relative path to the PNG, null when # single_ref mode (no frame conditioning). iframe_mode: paired|single_ref| # chained. frame_chain_from: previous shot_id whose last_frame was copied # into this shot's first_frame (chained continuity), else null. "first_frame_path (relative path to first frame PNG, null if single_ref mode), " "last_frame_path (relative path to last frame PNG, null if single_ref mode), " "iframe_mode ('paired'|'single_ref'|'chained'), " "frame_chain_from (previous shot_id if chained, else null)}", "lip_sync_report": "{shot_id, alignment_score (0-1), mouth_frames_analyzed, issues}", }) .constraints(_engine_constraints()) .quality([ "Video clip duration matches shot duration_sec ±1s?", "Lip-sync alignment_score ≥ 0.7?", "No BGM-banned words in final prompt?", "take_log entry present with verdict + evidence?", "observed_end_state records what the video actually shows (not what was planned)?", ]) .build() ) context = json.dumps( { "episode_id": episode_id, "shot_index": shot_idx, "shot": shot, "scene_images": scene_images, "character_assets": character_assets, }, ensure_ascii=False, ) return goal, context, has_frame_pair # 2. Fan out shot-level delegate_task calls concurrently (D-36-08). # ThreadPoolExecutor is appropriate because delegate_task is a blocking # sync callable (D-35-02) — threads release the GIL during the # underlying delegate dispatch, giving real concurrency in production. # Tests typically pass parallel_shots=1 for sequential determinism OR # monkeypatch ThreadPoolExecutor with a FakePool. # ── Shot-level resume: skip already-rendered shots ────────────── # When resuming after a crash (e.g. 429 rate-limit kill), P11 may have # already rendered some shots to disk (assets/P11/{shot_id}.mp4 + .json). # Without this check, ALL shots are re-rendered from scratch, wasting # GLM tokens and ComfyUI GPU time. (Root cause: phase-level checkpoint # only records phase completion; individual shot renders are invisible # to the checkpoint system.) # # R-40-01 (2026-07-09): shot-level cache-hit check. import os.path as _osp _output_dir = _osp.join(_osp.curdir, "assets", "P11") # P1-2: run workdir (run_dir) — hoisted so the engine-render closure + the # FIX-4 upstream check share one resolution base (assets/P11 → run_dir). _workdir = os.path.dirname(os.path.dirname(os.path.abspath(_output_dir))) # FIX-1 (2026-07-31): 逐镜头断点续传 checkpoint。 # 在 fan-out 之前读取 p11_progress.json,获取上次运行已完成/失败的镜头。 # 与下面的 R-40-01 _try_cache_hit 互补:progress 记录语义完成状态, # cache_hit 检查文件存在。两者都能跳过已完成镜头,互为补充。 _p11_progress = _load_p11_progress(_output_dir) _completed_shots: set = _p11_progress["completed_shots"] _failed_shots: set = _p11_progress["failed_shots"] # ── 2026-08-18 冗余审核体系 (T10): qwen fail 镜头强制重渲 ────────── # p11c 硬门 reject 打回 ∪ failed-shots slot（未 waived）→ 从 # completed_shots 剔除：mp4 在盘也视为未完成（QC 判不过≠渲染成功）， # fan-out 只重渲剔除集，其余镜头照常跳过（FIX-1 机制不变）。 _requeue_ids = _requeue_ids_from_qc(asset_bus_read) if _requeue_ids & _completed_shots: logger.info( "%s: QC-failed 镜头强制重渲（剔除 completed）: %s", PHASE_ID, sorted(_requeue_ids & _completed_shots), ) _completed_shots -= _requeue_ids _failed_shots |= _requeue_ids if _completed_shots: logger.info( "%s: checkpoint 发现 %d 个镜头已完成,将跳过 (completed=%s)", PHASE_ID, len(_completed_shots), sorted(_completed_shots), ) # gen_iframes.py writes assets/P11/iframe-manifest.json (first/last-frame # provenance per shot). Index by shot_id so each collected video_clip can # merge its iframe fields (first/last_frame_path, iframe_mode, frame_chain_from). iframe_manifest_by_shot = _load_iframe_manifest(_output_dir) def _try_cache_hit(shot_idx: int, shot: dict) -> dict | None: """Return a cached video_clip dict if the mp4 + meta already exist.""" sid = shot.get("shot_id", shot.get("id", f"shot_{shot_idx}")) mp4 = _osp.join(_output_dir, f"{sid}.mp4") meta = _osp.join(_output_dir, f"{sid}_meta.json") if not _osp.isfile(mp4) or not _osp.isfile(meta): return None try: import json as _json with open(meta, encoding="utf-8") as f: m = _json.load(f) return { "video_clip": { "shot_id": sid, "shot_index": shot_idx, "video_path": mp4, "filePath": mp4, "duration_sec": float(m.get("duration_sec", 0)), "resolution": m.get("resolution", "1280x704"), "codec": m.get("codec", "h264"), "engine": m.get("engine", "h3"), "murch_grade": "pass", "cached": True, }, "lip_sync_report": { "shot_id": sid, "shot_index": shot_idx, "alignment_score": 1.0, "mouth_frames_analyzed": 0, "issues": "skipped (cache hit)", }, } except Exception: return None # ── P1-2: in-process H3 video engine (engine-adapter) ──────────────── # Retires the per-shot hand-written render_S*.py wrappers: the actual # render is a deterministic multipart POST done by H3VideoEngine, not a # subagent freelancing curl. ONE engine instance is shared across the # fan-out worker threads; its internal ``_lock`` serializes the POST # (WR-01 — httpx.Client connection-pool thread-safety). Constructed once # here and closed after the pool joins below (the fan-out block catches # every per-shot exception, so an unconditional close after it is the # ``with``-equivalent and avoids re-indenting the whole closure). _engine_outcomes: dict[str, str] = {} _engine_clips: dict[str, str] = {} engine = select_video_engine() def _run_engine_render(shot: dict, shot_idx: int) -> None: """Deterministic H3 render for one shot (P1-2). Resolves the reference assets that ACTUALLY exist on disk and asks the engine to render the shot to ``assets/P11/{shot_id}.mp4``. Gated on real references (P1-1 pattern): no turnaround/scene_ref on disk → skip (tests / dev / missing upstream), and the delegate-reported result flows through unchanged. When the engine succeeds, the authoritative clip is recorded in ``_engine_clips`` and overrides the delegate-reported path at collection time. Degrades are counted in ``_engine_outcomes`` for the all-degraded fail-loud check after the loop. """ _sid = shot.get("shot_id") or shot.get("id") if not _sid: return turnaround_paths: list[str] = [] for _cr in (shot.get("character_refs") or []): if isinstance(_cr, dict): _tp = _resolve_path_safe((_cr.get("turnaround_path") or "").strip(), _workdir) if _tp: turnaround_paths.append(_tp) scene_ref_path = _resolve_path_safe((shot.get("scene_ref") or "").strip(), _workdir) _vc = voice_clips_by_shot.get(_sid, {}) _vc_audio, _ = _decide_audio_injection(_vc) voice_clip_path = ( _resolve_path_safe(_vc_audio.strip(), _workdir) if _vc_audio else None ) # Gate: no reference assets on disk → engine not runnable (tests/dev). if not turnaround_paths and not scene_ref_path: return output_path = os.path.join(_output_dir, f"{_sid}.mp4") # SPEC-firstframe 注入 (2026-08-22): assets/P11/spec_firstframes/{sid}.png # 存在时作为 H3 first-frame lock (multipart `image` 字段) 传入。根因: # ref2va 的图像先验(参考图构图)会覆盖 prompt 里的景别/运镜文本指令, # p11c 判 6/7 fail 均为"渲成参考图视角的偏近固定镜头"。首帧锁把景别 # 控制从文本层挪到图像层。缺图 → None → 行为与改造前完全一致 # (WARN 提示而非静默回落——否则"画质回去了"无从排查)。 _spec_ff = os.path.join(_output_dir, "spec_firstframes", f"{_sid}.png") first_frame = _spec_ff if os.path.isfile(_spec_ff) else None if first_frame is None: logger.warning( "%s: shot %s spec 首帧缺失, 回落 ref2va 参考图直出 " "(scripts/gen_spec_firstframes.py 可批量生成)", PHASE_ID, _sid) # #6: 该 shot 的 iframe manifest entry（有条件帧 → 不注入 FRAME-0 # ANCHOR；single_ref/缺失 → 注入 0 秒起始锚定）。 _iframe_entry = iframe_manifest_by_shot.get(_sid) _enhanced = _h3_enhance_prompt( shot, episode_config, character_assets, _tidx, _iframe_entry) # PPL FIX-3: provenance inputs — the ACTUAL reference images handed to # the engine this call, and the taskMode resolution order mirrored # from H3VideoEngine._resolve_mode (shot > episode_config > ref2va). _ref_inputs = list(turnaround_paths) if scene_ref_path: _ref_inputs.append(scene_ref_path) if first_frame: _ref_inputs.append(first_frame) _mode = ( shot.get("mode") or (episode_config or {}).get("h3_mode") or "ref2va" ) _t_render = time.time() try: result = engine.render( shot=shot, output_path=output_path, turnaround_paths=turnaround_paths or None, scene_ref_path=scene_ref_path, voice_clip_path=voice_clip_path, first_frame=first_frame, prompt=_enhanced, episode_config=episode_config, ) except VideoEngineError as exc: # 4xx caller bug — per-shot isolation: record degraded, keep going. logger.error("%s: shot %s H3 engine client error: %s", PHASE_ID, _sid, exc) _engine_outcomes[_sid] = "degraded" _record_video_provenance( shot, _sid, output_path, _enhanced, _ref_inputs, _mode, engine, None, "degraded", time.time() - _t_render, error=f"VideoEngineError: {exc}", ) return if isinstance(result, dict) and result.get("degraded"): _engine_outcomes[_sid] = "degraded" logger.warning( "%s: shot %s H3 engine degraded: %s", PHASE_ID, _sid, result.get("reason"), ) _record_video_provenance( shot, _sid, output_path, _enhanced, _ref_inputs, _mode, engine, result, "degraded", time.time() - _t_render, error=str(result.get("reason") or ""), ) return if isinstance(result, dict) and result.get("clip_path"): _engine_outcomes[_sid] = "ok" _engine_clips[_sid] = result["clip_path"] logger.info( "%s: shot %s H3 engine rendered %s (frame_count=%s)", PHASE_ID, _sid, result["clip_path"], result.get("frame_count"), ) _t_elapsed = time.time() - _t_render _record_video_provenance( shot, _sid, output_path, _enhanced, _ref_inputs, _mode, engine, result, "ok", _t_elapsed, ) # PPL FIX-3.2: {sid}_meta.json 补写回归 (prompt/negative_prompt). _write_render_meta( _sid, output_path, final_prompt=_enhanced, negative_prompt=shot.get("negative_prompt") or "", shot=shot, result=result, elapsed_sec=_t_elapsed, ) @glm_concurrent def _dispatch_shot(shot_idx: int, shot: dict) -> dict: # P11b final render (2026-08-17 KAP API 精简迁移): useCase=final-shot → # KAP 侧预射 native-sage 36 步 @1344×768 + 完整 LTX Foley + TTS 混音 # (取代旧 profile=native-sage + steps=15 平铺参数; 音频改由 KAP 统一混音)。 # setdefault preserves an explicit operator/shot override. shot.setdefault("h3_useCase", "final-shot") # R-40-01: skip if mp4 + meta already on disk. PPL FIX-3.2 note: the # engine path now writes {sid}_meta.json too, so QC-requeued shots # (fix-required / failed-shots / video-qc.requeue_shot_ids) MUST be # excluded here — otherwise the rejected clip would cache-hit and the # requeue (T10) would silently never re-render. _sid_cache = shot.get("shot_id") or shot.get("id") cached = ( None if (_sid_cache and _sid_cache in _requeue_ids) else _try_cache_hit(shot_idx, shot) ) if cached is not None: logger.info("P11 cache hit for shot %s — skipping render", shot.get("shot_id", shot_idx)) return cached # FIX-1 (2026-07-31): 也跳过 checkpoint 已记录完成(且 mp4 仍有效)的镜头。 # _load_p11_progress 已对 completed_shots 做过 ffprobe 复核,这里只需 # 查集合;若 mp4 在复核后被删,下一轮 run() 会因复核失败而重渲。 _sid = shot.get("shot_id") or shot.get("id") if _sid and _sid in _completed_shots: logger.info( "%s: checkpoint 命中,镜头 %s 已完成 — 跳过渲染", PHASE_ID, _sid, ) return _try_cache_hit(shot_idx, shot) or {"video_clip": {"shot_id": _sid, "cached": True}, "lip_sync_report": {"shot_id": _sid}} goal, ctx, _ = _goal_for_shot(shot, shot_idx) # FIX-4 (frame-workflow): upstream asset existence check. Verify the # cross-phase references this render needs — scene_ref (P07) and each # character_refs[].turnaround_path (P04 scene_costume_turnarounds) — # actually exist on disk before dispatching. Enforces the "gray TR must # exist before costume TR; costume TR must exist before render" chain in # CODE, not just in docs. NON-BLOCKING: missing assets log a warning + # are recorded to the fix-required bus for audit/rollback, but the # render still proceeds (a degraded run beats a hard stop; the goal's # identity FALLBACK covers missing turnarounds). asset_bus_write is # atomic (tmp+os.replace), so concurrent per-shot writes cannot corrupt # state — last writer wins on the single-value slot, which is fine for # an audit record; every shot's warning is also in the log regardless. _workdir = os.path.dirname(os.path.dirname(os.path.abspath(_output_dir))) _upstream_warnings: list[str] = [] _sr = shot.get("scene_ref") if isinstance(_sr, str) and _sr.strip(): if not _resolve_path_safe(_sr.strip(), _workdir): _upstream_warnings.append(f"scene_ref not found: {_sr.strip()}") for _cr in (shot.get("character_refs") or []): if isinstance(_cr, dict): _tp = _cr.get("turnaround_path") if isinstance(_tp, str) and _tp.strip(): _cname = _cr.get("name") or _cr.get("character") or "?" if not _resolve_path_safe(_tp.strip(), _workdir): _upstream_warnings.append( f"turnaround_path not found for {_cname}: {_tp.strip()}" ) if _upstream_warnings: _shot_ref = _sid or f"shot_{shot_idx}" logger.warning( "%s: shot %s upstream assets missing: %s", PHASE_ID, _shot_ref, "; ".join(_upstream_warnings), ) try: asset_bus_write("fix-required", { "phase_id": PHASE_ID, "error_type": "upstream_assets_missing", "rollback_to": "p04_character_design", "shot_id": _shot_ref, "fix_instructions": ( "Upstream assets missing: " + "; ".join(_upstream_warnings) ), }) except Exception: logger.warning( "%s: failed to record fix-required for shot %s", PHASE_ID, _shot_ref, exc_info=True, ) delegate_result = delegate_task(goal, ctx, ["skills", "file", "terminal"]) # P1-2: deterministic in-process H3 render — retires the hand-written # render_S*.py wrappers. _run_engine_render is gated on real reference # assets; when it produces a clip, the authoritative path is recorded in # _engine_clips and overrides the delegate-reported path at collection. _run_engine_render(shot, shot_idx) return delegate_result video_clips: list[dict] = [] lip_sync_reports: list[dict] = [] take_log_entries: list[dict] = [] # (27-03 Phase 27, p11_video.video_render) 引擎产物类 pre 接线。 # 语义裁决（SUMMARY 记 deviation）：pre>1 = **尝试冗余**——失败才补渲下 # 一变体（checkpoint 缓存令补渲近似零成本），而非并行 N× 渲染（后者 # GPU 成本线性放大且被缓存去重成最差组合）。final>1 仅留档 # _render_variants_meta，不放大 video-clips slot（p12 单镜单 clip 契约）。 # qwen-eye 创意选优（select_fn）维持 2026-08-16 立项注记。 from pipeline.phases._vision_review import resolve_redundancy as _p11_resolve p11_pre, p11_final = _p11_resolve( episode_config, "p11_video", "video_render", default_pre=1, default_final=1, ) if p11_pre > 1: logger.warning( "%s: video_render pre=%d → 尝试冗余（失败补渲，非并行 N× 渲染）；" "创意变体选优见 2026-08-16 立项注记", PHASE_ID, p11_pre, ) _render_variants_meta: list[dict] = [] def _dispatch_with_variants(idx: int, shot: dict): """(27-03) per-shot dispatch with attempt-redundancy (first-success wins).""" return _dispatch_with_attempt_redundancy( _dispatch_shot, idx, shot, pre=p11_pre, phase_id=PHASE_ID, meta_out=_render_variants_meta, ) # C10: per-future error capture. A single failing shot must NOT discard # the results of its already-completed siblings. We collect failures here # and re-raise (BgmTriggerError only) AFTER the loop so the runner's outer # retry still fires — but only once every successful shot has been collected # and checkpointed. _batch_errors: list[tuple[int, str, Exception]] = [] if shot_list: with ThreadPoolExecutor(max_workers=parallel_shots) as pool: futures = [ pool.submit(_dispatch_with_variants, idx, shot) for idx, shot in enumerate(shot_list) ] for idx, future in enumerate(futures): try: delegate_result = future.result() except BgmTriggerError as bgm_exc: # SHARED-01 (Phase 10-01): write fix-required, then CAPTURE # the error and CONTINUE collecting already-completed sibling # futures. The old `raise` here exited the loop at shot K, # discarding the results of every shot with idx > K that had # already finished in the pool — completed mp4 checkpoints # for those shots were never written either. We re-raise the # first BgmTriggerError AFTER the loop (see below) so the # runner's outer retry still fires (strip banned words + # retry); on retry, completed shots are resumed/skipped. # T-10-01-04 (DoS): max 1 retry per shot — strip on retry # prevents infinite loop. words_str = ", ".join(bgm_exc.matched_words) asset_bus_write("fix-required", { "phase_id": PHASE_ID, "error_type": "bgm_trigger_words", "rollback_to": PHASE_ID, "attempt": int((fix_required_in or {}).get("attempt", 0)) + 1, "shot_id": bgm_exc.shot_id, "fix_instructions": ( f"Shot {bgm_exc.shot_id} video_prompt contains BGM " f"trigger word(s): {words_str}. These trigger LTX " f"audio VAE to generate background music. Rewrite " f"to physical sound verbs (e.g. 'speaks softly' " f"instead of 'hums'). On retry the pipeline will " f"auto-strip these words." ), }) _batch_errors.append((idx, bgm_exc.shot_id, bgm_exc)) logger.warning( "%s: shot %s failed with BgmTriggerError (%s) — " "continuing batch to preserve completed shots", PHASE_ID, bgm_exc.shot_id, words_str, ) continue except Exception as _dispatch_exc: # Any other per-shot error: capture and keep going so a single # failing shot doesn't discard its already-completed siblings. _shot_ref = ( shot_list[idx].get("shot_id", f"shot_{idx}") if isinstance(shot_list, list) and idx < len(shot_list) else f"shot_{idx}" ) logger.error( "%s: shot %s unexpected dispatch error: %s", PHASE_ID, _shot_ref, _dispatch_exc, exc_info=True, ) _batch_errors.append((idx, _shot_ref, _dispatch_exc)) continue # 2026-08-19: cache-hit 早返回的 dict 没有 summary 键；GLM 429 崩溃 # 时 bridge 回吐裸 CLI 信封（is_error/api_error_status，summary 空）。 # 两种形状都会让 parse_delegate_output 抛 "summary is empty"，然后 # 在 @glm_concurrent join 时废掉整批（10:08/12:44/16:01 三杀根因）。 # 已是合同形状（含 video_clip）→ 直通；解析失败但 mp4+meta 在盘 # （引擎/前次已产出）→ 回退 cache 形状；都没有才真失败。 if isinstance(delegate_result, dict) and "video_clip" in delegate_result: expert_output = delegate_result else: try: expert_output = _parse_expert_output(delegate_result) except ValueError as _parse_exc: # 存量 bug 修复 (2026-08-23): 本作用域循环变量是 idx # (L2570 setdefault("shot_index", idx) 为证)，旧代码引用 # 未定义的 shot_idx —— 正常路径不走此分支从未暴露， # delegate parse 失败触发 fallback 时 NameError 崩整批。 # 存量 bug 修复续 (2026-08-26): 同分支还引用了未绑定的 # `shot`（UnboundLocalError 实锤 18:11 崩 p11b 收尾期）—— # 本作用域拿 shot 只能经 shot_list[idx]。 shot = ( shot_list[idx] if isinstance(shot_list, list) and idx < len(shot_list) else {} ) _fb = _try_cache_hit(idx, shot) if _fb is None: raise logger.warning( "%s: shot %s delegate parse failed (%s) — " "mp4+meta 在盘，回退 cache 形状继续", PHASE_ID, shot.get("shot_id", f"shot_{idx}"), _parse_exc, ) expert_output = _fb # Annotate with shot_index so aggregated writes preserve order. video_clip = expert_output.get("video_clip", {}) lip_sync_report = expert_output.get("lip_sync_report", {}) video_clip.setdefault("shot_index", idx) lip_sync_report.setdefault("shot_index", idx) # Merge first/last-frame provenance from gen_iframes.py manifest # (setdefault: the subagent's own values win if it emitted them; # the manifest is the fallback source of truth for these paths). _iframe_sid = video_clip.get("shot_id") or video_clip.get("id") _iframe_entry = iframe_manifest_by_shot.get(_iframe_sid) if _iframe_sid else None if _iframe_entry: video_clip.setdefault("first_frame_path", _iframe_entry.get("first_frame_path")) video_clip.setdefault("last_frame_path", _iframe_entry.get("last_frame_path")) video_clip.setdefault("iframe_mode", _iframe_entry.get("iframe_mode")) video_clip.setdefault("frame_chain_from", _iframe_entry.get("frame_chain_from")) # 改造 1: propagate per-clip TTS positioning into the video_clip # OUTPUT so P12b can place each TTS at clip_start_sec + # tts_audio_offset_sec (measured above at ffprobe time). setdefault: # a subagent-emitted value wins; the shot's measured value is the # fallback source of truth. Enables P12b per-shot positioned mix. _tts_src_shot = ( shot_list[idx] if isinstance(shot_list, list) and idx < len(shot_list) and isinstance(shot_list[idx], dict) else {} ) video_clip.setdefault("tts_audio_path", _tts_src_shot.get("tts_audio_path")) video_clip.setdefault("tts_audio_offset_sec", _tts_src_shot.get("tts_audio_offset_sec")) # Extract take_log entry from video_clip (per-shot retake tracking) take_entry = video_clip.pop("take_log", None) if take_entry and isinstance(take_entry, dict): take_entry.setdefault("shot_index", idx) take_log_entries.append(take_entry) # P1-2: override the delegate-reported path with the REAL # in-process H3 engine clip when the engine produced one. The # engine clip is authoritative (frame-validated); the delegate's # creative metadata (murch_grade / observed_end_state / etc.) is # preserved — only the path + engine are overridden. _over_sid = video_clip.get("shot_id") or ( shot_list[idx].get("shot_id") if isinstance(shot_list, list) and idx < len(shot_list) and isinstance(shot_list[idx], dict) else None ) if _over_sid and _over_sid in _engine_clips: _eng_path = _engine_clips[_over_sid] video_clip["video_path"] = _eng_path video_clip["filePath"] = _eng_path video_clip["path"] = _eng_path video_clip["engine"] = "h3" video_clips.append(video_clip) lip_sync_reports.append(lip_sync_report) # FIX-1 (2026-07-31): 每完成一个镜头即时写 checkpoint。 # 从 delegate 输出里提取 shot_id,校验 .mp4 真实存在且 # ffprobe 可读,然后追加到 completed_shots 并原子写入进度文件。 # 这样即使后续镜头崩溃/限流,已完成镜头下次重进时会被跳过。 # 注意: 本回调循环只绑定 (idx, future),shot 需从 shot_list[idx] 取。 _orig_shot = shot_list[idx] if isinstance(shot_list, list) and idx < len(shot_list) else {} _ckpt_sid = video_clip.get("shot_id") or _orig_shot.get("shot_id") or _orig_shot.get("id") _ckpt_mp4 = None for _p in (video_clip.get("video_path"), video_clip.get("filePath"), video_clip.get("path")): if isinstance(_p, str) and _p.strip(): _ckpt_mp4 = _p.strip() break if _ckpt_sid: # 优先校验 delegate 报告的 video_path;缺失则回退到 # 规范命名 assets/P11/{shot_id}.mp4。 _ckpt_target = _ckpt_mp4 or os.path.join(_output_dir, f"{_ckpt_sid}.mp4") _ckpt_ok, _ckpt_reason = _mp4_valid_with_frames(_ckpt_target, _orig_shot) if _ckpt_ok: _completed_shots.add(_ckpt_sid) _failed_shots.discard(_ckpt_sid) _save_p11_progress(_output_dir, _completed_shots, _failed_shots) # ── 2026-08-26 逐镜增量画布同步 ────────────────── # 渲染+mp4 校验通过即刻触发一次 partial 同步：资产管理 # 中心/画布实时看到已渲镜头；mid-phase 停机（预览人工 # 门/crash/看门狗重启）不再留下同步空洞（p11b 18 镜 # ×~7min，此前只有 phase 完成才同步=数小时盲区）。 # 幂等性：canvas_sync 2a/2b 磁盘并入保证每次 partial # 重建的都是累积全集。优先走 runner 注入的 # sync_shot_progress（on_phase_complete 注册时 # sync_canvas 不注入）；partial 标记让 runner 对这类 # 同步的网络性失败不记 canvas_sync_failed。失败不伤 # 渲染主链。 _shot_cb = sync_shot_progress or sync_canvas if _shot_cb is not None: try: _shot_cb(episode_id, PHASE_ID, { "phase": PHASE_ID, "partial": True, "outputs": {"video_clips": [dict(video_clip)]}, }) except Exception: # noqa: BLE001 — 同步失败不伤渲染 logger.warning( "%s: per-shot canvas sync failed for %s (non-fatal)", PHASE_ID, _ckpt_sid, exc_info=True, ) else: logger.warning( "%s: 镜头 %s 已渲染但 mp4 校验失败 (%s, %s),不计入 checkpoint,下次将重渲", PHASE_ID, _ckpt_sid, _ckpt_target, _ckpt_reason, ) _failed_shots.add(_ckpt_sid) # P1-2: fan-out complete — release the H3 engine's httpx connection pool. # (The fan-out block catches every per-shot exception, so reaching here means # the pool has joined and no worker is using the engine anymore.) engine.close() # C10: AFTER the batch loop — now that every completed shot has been # collected + checkpointed, surface the first BgmTriggerError so the # runner's outer retry fires (strip banned words + retry the failing shot). # Non-BgmTriggerError failures are logged above and left out of the output; # they do not abort the whole phase. if _batch_errors: _bgm_errors = [e for _, _, e in _batch_errors if isinstance(e, BgmTriggerError)] logger.warning( "%s: batch finished with %d/%d shot failures (%d BGM)", PHASE_ID, len(_batch_errors), len(shot_list or []), len(_bgm_errors), ) if _bgm_errors: raise _bgm_errors[0] # P1-2: all-engine-degraded fail-loud. When the in-process H3 engine was # actually invoked (reference assets existed on disk) and EVERY attempt # degraded, there are no authoritative clips — fail loudly rather than ship # the delegate's non-rendered (curl-less) paths. No attempts (tests / dev / # missing upstream assets) → no raise, so the existing test contract holds. _eng_attempted = sum(1 for v in _engine_outcomes.values() if v in ("ok", "degraded")) _eng_degraded = sum(1 for v in _engine_outcomes.values() if v == "degraded") if _eng_attempted > 0 and _eng_attempted == _eng_degraded: raise RuntimeError( f"{PHASE_ID}: all {_eng_attempted} in-process H3 renders degraded " f"(H3 service unavailable / rejected every shot). No authoritative " f"clips produced. Investigate the :10588 H3 pipeline." ) aggregated = { "video_clips": video_clips, "lip_sync_reports": lip_sync_reports, } # 3. Write aggregated outputs as SINGLE writes per slot (per the plan's # asset_bus_slots_to_add — "aggregated single write", not per-shot # appends). Downstream p12_composition reads the consolidated arrays. asset_bus_write("video-clips", video_clips) asset_bus_write("lip-sync-reports", lip_sync_reports) # Take log — Seedance retake-protocol inspired iteration tracking. # Written as a JSON slot containing the full take array for this episode. # Enables "same flaw twice = rewrite" rule for downstream analysis. if take_log_entries: # (27-03) 变体尝试留痕搭 take-log（dict slot；video-clips 是纯 list # 不加键——p12 单镜单 clip 契约）。final>1 同样仅在此留档。 _take_log_payload = {"takes": take_log_entries} if _render_variants_meta: _take_log_payload["render_variants"] = _render_variants_meta _take_log_payload["render_variants_config"] = { "requested_pre": p11_pre, "requested_final": p11_final, "effective_final": 1, "note": "final>1 留档不放大 slot（p12 单镜单 clip 契约）", } asset_bus_write("take-log", _take_log_payload) # type: ignore[arg-type] try: from ._manifest import write_manifest, _oss_url _vc = aggregated.get("video_clips", []) clips = _vc if isinstance(_vc, list) else [] manifest_nodes = [] for clip in clips: if not isinstance(clip, dict): continue sid = clip.get("shot_id", clip.get("id", "clip")) label = f"视频片段 {sid}" # Video clips inherit the prompt from the storyboard shot — that # IS the generation prompt that produced this video. Surface it. prompt = _get_vp(clip) or clip.get("image_prompt") or "" motion = clip.get("motion") or clip.get("motion_description") or "" visual = clip.get("visual_description") or "" desc_parts = [] if motion: desc_parts.append(f"运镜: {motion.strip()}") if visual: desc_parts.append(f"画面: {visual.strip()}") description = " · ".join(desc_parts) or f"P11 视频片段 · {sid} · h3 视频生成" params = { "shot_id": sid, "engine": clip.get("engine", "h3"), "duration_sec": clip.get("duration_sec", 0), "resolution": clip.get("resolution", "1280x704"), "description": description, } if prompt: params["prompt"] = prompt.strip() manifest_nodes.append({ "id": f"p11/{sid}", "type": "video", "label": label, "filePath": clip.get("filePath", clip.get("path", "")), "thumbnailUrl": clip.get("thumbnailUrl", clip.get("thumbnail", "")), "params": params, "consumes": [f"p09/{sid}", f"p10/{sid}"], "phase": "p11", }) if manifest_nodes: write_manifest(PHASE_ID, manifest_nodes) except (OSError, RuntimeError): # I/O errors (disk full, permissions) → warn + continue. # ValueError (missing required params) MUST propagate to stop the pipeline. logger.warning("manifest write failed", exc_info=True) # 4. Trigger gate if configured. gate_result = None if GATE_ID and trigger_gate is not None: gate_result = trigger_gate(GATE_ID, episode_id) # P0 #1 (2026-08-14 review): an explicit gate rejection must # block the phase, not just be logged. raise_if_rejected(gate_result, PHASE_ID) logger.info( "p11: gate %s triggered for episode %s -> %s", GATE_ID, episode_id, gate_result, ) result = { "phase": PHASE_ID, "outputs": aggregated, "gate": gate_result, } if sync_canvas is not None: try: sync_canvas(episode_id, PHASE_ID, result) except Exception: logger.warning("%s: canvas sync failed", PHASE_ID, exc_info=True) return result