"""Retry-strategy library + self-healing registry for coverage-pass failures. Drives `_execute_pass_batch`'s retry path. Given a failed `PassResult` and its `CoveragePass`, classifies the failure mode, selects a strategy from a per- mode escalation chain (reordered by learned success rate once data exists), and returns a mutated pass ready for re-execution. Architecture summary (see consultations/recoil/retry-strategy-architecture/): - 25 strategies across 6 categories (A-G). - Accumulate-don't-stack: each retry adds 1 new strategy on top of previous free/cheap mutations. Expensive strategies reset to the original pass. - Autonomous from day 1: static ESCALATION_CHAINS ordering during cold start (n < 8), learned reordering at n >= 8. - Cost cap: max_retry_spend_usd = 6.00 per pass default; escalate to human when next strategy's cost would exceed cap. - Tier 0 + Tier 1 detection in v1 (API errors + gate results + cut count). Tier 2 Opus vision classification for residual UNKNOWN cases. File layout: - FailureMode (imported from recoil.core.critic) - RetryStrategyName (new enum in this module) - StrategyDiff (dataclass — what a strategy changed) - ESCALATION_CHAINS (dict[FailureMode, list[RetryStrategyName]]) - STRATEGY_REGISTRY (dict[RetryStrategyName, StrategyEntry]) - detect_failure_mode() — Tier 0 + Tier 1 classifier - StrategyEngine class — orchestrates selection + application Strategy implementations live in this file (not split across modules) — keeping all ~25 functions in one file is simpler to review and greppable. The file is long (~1200 lines after full implementation) but it's a coherent unit. """ from __future__ import annotations import dataclasses import logging import os import random from dataclasses import dataclass from enum import Enum from pathlib import Path from typing import Callable, Optional, Literal, TYPE_CHECKING import sys _RECOIL_ROOT = str(Path(__file__).resolve().parent.parent.parent) _PIPELINE_ROOT = str(Path(__file__).resolve().parent.parent) if _RECOIL_ROOT not in sys.path: sys.path.insert(0, _RECOIL_ROOT) if _PIPELINE_ROOT not in sys.path: sys.path.insert(0, _PIPELINE_ROOT) from recoil.core.critic import FailureMode # noqa: E402 from orchestrator.coverage_planner import CoveragePass # noqa: E402 from recoil.execution.step_types import PassResult # noqa: E402 if TYPE_CHECKING: from orchestrator.learning_engine import LearningEngine logger = logging.getLogger(__name__) _warned_tier2_opus_unavailable = False def _warn_tier2_opus_unavailable_once() -> None: global _warned_tier2_opus_unavailable if _warned_tier2_opus_unavailable: return _warned_tier2_opus_unavailable = True logger.warning( "Tier-2 Opus classifier unavailable (no Anthropic SDK client) — visual failures degrade to UNKNOWN/0.30" ) # ============================================================================ # Enums + dataclasses # ============================================================================ class RetryStrategyName(str, Enum): """Canonical strategy identifiers. String values appear in PassStore records and LearningEngine stats — never rename an existing member.""" # Category A — Reference image strategies ADD_TURNAROUND_ANGLES = "add_turnaround_angles" REMOVE_COMPETING_REFS = "remove_competing_refs" SWAP_POLE_POSITION_REF = "swap_pole_position_ref" REORDER_REFS_SCENE_FIRST = "reorder_refs_scene_first" # Category B — Prompt-level strategies STRENGTHEN_IDENTITY_LOCK = "strengthen_identity_lock" REWRITE_SHOT_TYPE_PREFIX = "rewrite_shot_type_prefix" SOFTEN_PROMPT = "soften_prompt" BUMP_MOTION_INTENSITY = "bump_motion_intensity" ADD_FILM_STOCK_ANCHOR = "add_film_stock_anchor" FORMAT_MODE_TIMELINE = "format_mode_timeline" FORMAT_MODE_HYBRID = "format_mode_hybrid" REDUCE_FACE_PROMINENCE = "reduce_face_prominence" # Category C — Start frame / blueprint strategies REGEN_PREVIZ_SWAP_FRAME = "regen_previz_swap_frame" SWITCH_I2V_TO_R2V = "switch_i2v_to_r2v" # SWITCH_R2V_TO_I2V deferred — retry loop always calls execute_pass (R2V); # would require conditional dispatch to execute_video. Cut from v1. # Category D — Pass structure strategies (sprint-feasible subset only) REORDER_SEGMENTS = "reorder_segments" DROP_WEAK_SEGMENT = "drop_weak_segment" # RESEGMENT_PASS deferred — needs pass split/merge logic # Category E — Generation config strategies FLIP_ANCHOR_DURATION = "flip_anchor_duration" CHANGE_RESOLUTION = "change_resolution" REDUCE_DURATION = "reduce_duration" REDUCE_TAKES_COUNT = "reduce_takes_count" CHANGE_SEED = "change_seed" # CHANGE_MODEL deferred — Kling integration doesn't exist # Category F — Cost / routing strategies UPGRADE_FAST_TO_PRO = "upgrade_fast_to_pro" DOWNGRADE_PRO_TO_FAST = "downgrade_pro_to_fast" # Category G — Human routing ESCALATE_TO_HUMAN = "escalate_to_human" CostTier = Literal["free", "cheap", "expensive"] @dataclass(frozen=True) class StrategyDiff: """Immutable record of what a strategy changed. Returned by every strategy function. `modified_pass` is the new pass ready for re-execution; `changes` is a human-readable diff for provenance logs and the morning review queue. """ strategy_name: RetryStrategyName modified_pass: CoveragePass changes: dict # e.g. {"element_config.identity_ref_mode": "full_turnaround"} cost_tier: CostTier estimated_cost_usd: float # rough estimate for the retry generation applicable: bool = ( True # some strategies (e.g. switch_r2v_to_i2v w/o keyframe) may be no-op ) @dataclass(frozen=True) class StrategyEntry: """Registry entry for a single strategy.""" name: RetryStrategyName fn: Callable[[CoveragePass, PassResult], StrategyDiff] cost_tier: CostTier applicable_modes: tuple[FailureMode, ...] description: str # NOTE: RetryCostPolicy lives in production_types.py (added in Step 1b). # It is imported here so strategy functions can reference it without a # circular import (production_types.py does NOT import strategy_registry.py). from orchestrator.production_types import RetryCostPolicy # noqa: E402 # ============================================================================ # Escalation chains — static cold-start ordering # ============================================================================ # For each failure mode, an ordered list of strategies to try. Cheap/free # strategies first. Expensive strategies last. The LearningEngine can reorder # these once it has n >= 8 samples per (failure_mode, model) pair. # # Design principle: the most recent empirical knowledge drives initial order. # ADD_TURNAROUND_ANGLES is the seeded winner for IDENTITY_DRIFT (pipeline- # learnings §10g). FORMAT_MODE strategies are the known fix for CUTS_TOO_SOFT. ESCALATION_CHAINS: dict[FailureMode, list[RetryStrategyName]] = { FailureMode.IDENTITY_DRIFT: [ RetryStrategyName.ADD_TURNAROUND_ANGLES, # free — proven winner RetryStrategyName.STRENGTHEN_IDENTITY_LOCK, # free RetryStrategyName.SWAP_POLE_POSITION_REF, # free RetryStrategyName.REORDER_SEGMENTS, # free RetryStrategyName.FLIP_ANCHOR_DURATION, # free RetryStrategyName.REDUCE_DURATION, # free RetryStrategyName.UPGRADE_FAST_TO_PRO, # cheap — quality escalation before expensive regen RetryStrategyName.REGEN_PREVIZ_SWAP_FRAME, # expensive (+previz cost) ], FailureMode.COMPOSITION_WRONG: [ RetryStrategyName.REWRITE_SHOT_TYPE_PREFIX, # free RetryStrategyName.REORDER_REFS_SCENE_FIRST, # free RetryStrategyName.UPGRADE_FAST_TO_PRO, # cheap — quality escalation before expensive regen RetryStrategyName.REGEN_PREVIZ_SWAP_FRAME, # expensive ], FailureMode.MOTION_FAILURE: [ RetryStrategyName.BUMP_MOTION_INTENSITY, # free RetryStrategyName.FORMAT_MODE_TIMELINE, # free RetryStrategyName.FORMAT_MODE_HYBRID, # free RetryStrategyName.FLIP_ANCHOR_DURATION, # free RetryStrategyName.UPGRADE_FAST_TO_PRO, # cheap — pro tier has better motion ], FailureMode.STYLE_DRIFT: [ RetryStrategyName.ADD_FILM_STOCK_ANCHOR, # free RetryStrategyName.REORDER_REFS_SCENE_FIRST, # free RetryStrategyName.UPGRADE_FAST_TO_PRO, # cheap — pro tier has better style consistency ], FailureMode.CUTS_TOO_SOFT: [ RetryStrategyName.FORMAT_MODE_TIMELINE, # free RetryStrategyName.FORMAT_MODE_HYBRID, # free RetryStrategyName.FLIP_ANCHOR_DURATION, # free RetryStrategyName.REORDER_SEGMENTS, # free ], FailureMode.CONTENT_FILTER_HARD_BLOCK: [ RetryStrategyName.SOFTEN_PROMPT, # free (existing behavior) RetryStrategyName.SWITCH_I2V_TO_R2V, # cheap (most effective per §9n) RetryStrategyName.REDUCE_FACE_PROMINENCE, # free ], FailureMode.REF_BLEED: [ RetryStrategyName.REMOVE_COMPETING_REFS, # free RetryStrategyName.SWAP_POLE_POSITION_REF, # free ], FailureMode.PROMPT_DURATION_MISMATCH: [ RetryStrategyName.FLIP_ANCHOR_DURATION, # free RetryStrategyName.REDUCE_DURATION, # free ], FailureMode.COST_OVERRUN: [ RetryStrategyName.DOWNGRADE_PRO_TO_FAST, # cheap — biggest cost lever RetryStrategyName.REDUCE_TAKES_COUNT, # free RetryStrategyName.REDUCE_DURATION, # free RetryStrategyName.CHANGE_RESOLUTION, # free (480p escape) ], FailureMode.GATE_MECHANICAL: [ RetryStrategyName.CHANGE_SEED, # free — mechanical re-roll RetryStrategyName.DROP_WEAK_SEGMENT, # free (if multi-segment) ], FailureMode.TRANSIENT: [ RetryStrategyName.CHANGE_SEED, # free — retry with new state ], FailureMode.UNKNOWN: [ RetryStrategyName.CHANGE_SEED, # safest catchall RetryStrategyName.ESCALATE_TO_HUMAN, ], } # ============================================================================ # Registry — populated by strategy function definitions below # ============================================================================ # Populated at module import by `_register()` decorator calls on each # strategy function. At import time, STRATEGY_REGISTRY maps # RetryStrategyName → StrategyEntry. STRATEGY_REGISTRY: dict[RetryStrategyName, StrategyEntry] = {} def _register( name: RetryStrategyName, cost_tier: CostTier, applicable_modes: tuple[FailureMode, ...], description: str, ) -> Callable: """Decorator — registers a strategy function in STRATEGY_REGISTRY.""" def decorator(fn: Callable[[CoveragePass, PassResult], StrategyDiff]) -> Callable: STRATEGY_REGISTRY[name] = StrategyEntry( name=name, fn=fn, cost_tier=cost_tier, applicable_modes=applicable_modes, description=description, ) return fn return decorator # ============================================================================ # Strategy implementations — Category A-G # ============================================================================ # ── Category A — Reference image strategies ── @_register( name=RetryStrategyName.ADD_TURNAROUND_ANGLES, cost_tier="free", applicable_modes=(FailureMode.IDENTITY_DRIFT,), description=( "Signal _gather_pass_refs to include the full canonical turnaround " "set (hero + front + three_quarter + profile + back) instead of " "defaulting to hero-only. Proven winner for identity drift per " "pipeline-learnings §10g." ), ) def strategy_add_turnaround_angles( coverage_pass: CoveragePass, pass_result: PassResult, ) -> StrategyDiff: new_element_config = dict(coverage_pass.element_config or {}) prev_mode = new_element_config.get("identity_ref_mode", "hero_only") new_element_config["identity_ref_mode"] = "full_turnaround" new_element_config.pop("identity_ref_override", None) modified = dataclasses.replace(coverage_pass, element_config=new_element_config) return StrategyDiff( strategy_name=RetryStrategyName.ADD_TURNAROUND_ANGLES, modified_pass=modified, changes={"element_config.identity_ref_mode": f"{prev_mode} → full_turnaround"}, cost_tier="free", estimated_cost_usd=_estimated_cost_of_next_generation(modified), ) @_register( name=RetryStrategyName.REMOVE_COMPETING_REFS, cost_tier="free", applicable_modes=(FailureMode.REF_BLEED,), description=( "Strip all non-focus characters from element_config.characters. " "Seedance's limited ref budget (9 slots) shouldn't split across " "competing faces when one character is the focus." ), ) def strategy_remove_competing_refs( coverage_pass: CoveragePass, pass_result: PassResult, ) -> StrategyDiff: new_element_config = dict(coverage_pass.element_config or {}) focus = coverage_pass.focus_character if not focus: return StrategyDiff( strategy_name=RetryStrategyName.REMOVE_COMPETING_REFS, modified_pass=coverage_pass, changes={}, cost_tier="free", estimated_cost_usd=_estimated_cost_of_next_generation(coverage_pass), applicable=False, ) chars_before = new_element_config.get("characters", []) if focus: chars_after = [ c for c in chars_before if isinstance(c, dict) and c.get("char_id") == focus ] new_element_config["characters"] = chars_after modified = dataclasses.replace(coverage_pass, element_config=new_element_config) return StrategyDiff( strategy_name=RetryStrategyName.REMOVE_COMPETING_REFS, modified_pass=modified, changes={ "element_config.characters": f"{len(chars_before)} → {len(new_element_config.get('characters', []))} (focus only)", }, cost_tier="free", estimated_cost_usd=_estimated_cost_of_next_generation(modified), ) @_register( name=RetryStrategyName.SWAP_POLE_POSITION_REF, cost_tier="free", applicable_modes=(FailureMode.IDENTITY_DRIFT, FailureMode.REF_BLEED), description=( "Change which reference image occupies @Image1. The pole position " "gets 40-50% of model attention (ADR-M04). If hero isn't locking " "identity, try a turnaround angle relevant to the failing segment's " "camera angle." ), ) def strategy_swap_pole_position_ref( coverage_pass: CoveragePass, pass_result: PassResult, ) -> StrategyDiff: worst_segment = _worst_segment(pass_result) shot_type = None if worst_segment is not None: for seg in coverage_pass.segments: if seg.source_shot_id == worst_segment.source_shot_id: shot_type = seg.shot_type break preferred_angle = { "CU": "front", "ECU": "front", "MCU": "front", "MS": "three_quarter", "MLS": "three_quarter", "WS": "three_quarter", "OTS": "profile", "PROFILE": "profile", }.get(shot_type, "front") new_element_config = dict(coverage_pass.element_config or {}) prev = new_element_config.get("pole_position_angle", "hero") new_element_config["pole_position_angle"] = preferred_angle modified = dataclasses.replace(coverage_pass, element_config=new_element_config) return StrategyDiff( strategy_name=RetryStrategyName.SWAP_POLE_POSITION_REF, modified_pass=modified, changes={"element_config.pole_position_angle": f"{prev} → {preferred_angle}"}, cost_tier="free", estimated_cost_usd=_estimated_cost_of_next_generation(modified), ) @_register( name=RetryStrategyName.REORDER_REFS_SCENE_FIRST, cost_tier="free", applicable_modes=(FailureMode.STYLE_DRIFT, FailureMode.COMPOSITION_WRONG), description=( "Swap ref stacking order so scene/blueprint refs come before " "identity refs. Seedance is recency-biased — when the problem is " "style or framing rather than identity, giving scene @Image1 " "priority anchors the look while identity refs in later slots " "still maintain character." ), ) def strategy_reorder_refs_scene_first( coverage_pass: CoveragePass, pass_result: PassResult, ) -> StrategyDiff: new_element_config = dict(coverage_pass.element_config or {}) prev = new_element_config.get("ref_ordering", "identity_first") new_element_config["ref_ordering"] = "scene_first" modified = dataclasses.replace(coverage_pass, element_config=new_element_config) return StrategyDiff( strategy_name=RetryStrategyName.REORDER_REFS_SCENE_FIRST, modified_pass=modified, changes={"element_config.ref_ordering": f"{prev} → scene_first"}, cost_tier="free", estimated_cost_usd=_estimated_cost_of_next_generation(modified), ) # ── Helper — used by multiple strategies ── def _worst_segment(pass_result: PassResult): """Return the failing segment with the lowest identity_score (or None).""" if not pass_result.segment_results: return None bad = [s for s in pass_result.segment_results if not getattr(s, "usable", True)] pool = bad if bad else list(pass_result.segment_results) def _score(s): return getattr(s, "identity_score", None) or 1.0 pool.sort(key=_score) return pool[0] if pool else None # ── Category B — Prompt-level strategies ── @_register( name=RetryStrategyName.STRENGTHEN_IDENTITY_LOCK, cost_tier="free", applicable_modes=(FailureMode.IDENTITY_DRIFT,), description="Prepend strict identity-lock language to arc_preamble.", ) def strategy_strengthen_identity_lock( coverage_pass: CoveragePass, pass_result: PassResult, ) -> StrategyDiff: lock_prefix = ( "CRITICAL: Maintain EXACT facial identity from @Image1 throughout " "all segments. DO NOT alter bone structure, skin tone, or hair " "between shots. " ) new_preamble = lock_prefix + (coverage_pass.arc_preamble or "") modified = dataclasses.replace(coverage_pass, arc_preamble=new_preamble) return StrategyDiff( strategy_name=RetryStrategyName.STRENGTHEN_IDENTITY_LOCK, modified_pass=modified, changes={"arc_preamble": "prepended identity-lock preamble"}, cost_tier="free", estimated_cost_usd=_estimated_cost_of_next_generation(modified), ) @_register( name=RetryStrategyName.REWRITE_SHOT_TYPE_PREFIX, cost_tier="free", applicable_modes=(FailureMode.COMPOSITION_WRONG,), description=( "Prepend aggressive bracketed shot-type declaration to each segment " "prompt to override model composition defaults." ), ) def strategy_rewrite_shot_type_prefix( coverage_pass: CoveragePass, pass_result: PassResult, ) -> StrategyDiff: _SHOT_PREFIX_MAP = { "CU": "[EXTREME CLOSE UP — FACE ONLY] ", "ECU": "[EXTREME CLOSE UP — FACE ONLY] ", "MCU": "[EXTREME CLOSE UP — FACE ONLY] ", "MS": "[MEDIUM SHOT — WAIST UP] ", "MLS": "[MEDIUM SHOT — WAIST UP] ", "WS": "[WIDE ESTABLISHING SHOT — FULL ENVIRONMENT] ", "LS": "[WIDE ESTABLISHING SHOT — FULL ENVIRONMENT] ", "EWS": "[WIDE ESTABLISHING SHOT — FULL ENVIRONMENT] ", "OTS": "[OVER THE SHOULDER — FOCUS ON SUBJECT] ", } new_segments = [] for seg in coverage_pass.segments: prefix = _SHOT_PREFIX_MAP.get( seg.shot_type, f"[{seg.shot_type.upper()}] " if seg.shot_type else "" ) new_prompt = prefix + seg.prompt new_segments.append(dataclasses.replace(seg, prompt=new_prompt)) modified = dataclasses.replace(coverage_pass, segments=new_segments) return StrategyDiff( strategy_name=RetryStrategyName.REWRITE_SHOT_TYPE_PREFIX, modified_pass=modified, changes={"segments": "prepended shot-type prefix to all segment prompts"}, cost_tier="free", estimated_cost_usd=_estimated_cost_of_next_generation(modified), ) @_register( name=RetryStrategyName.SOFTEN_PROMPT, cost_tier="free", applicable_modes=(FailureMode.CONTENT_FILTER_HARD_BLOCK,), description=( "Run lib.prompt_soften.soften_prompt() on arc_preamble and all " "segment prompts to sanitize for content filter." ), ) def strategy_soften_prompt( coverage_pass: CoveragePass, pass_result: PassResult, ) -> StrategyDiff: from recoil.pipeline._lib.prompt_soften import soften_prompt new_preamble = soften_prompt(coverage_pass.arc_preamble or "") new_segments = [] for seg in coverage_pass.segments: new_segments.append(dataclasses.replace(seg, prompt=soften_prompt(seg.prompt))) modified = dataclasses.replace( coverage_pass, arc_preamble=new_preamble, segments=new_segments, ) return StrategyDiff( strategy_name=RetryStrategyName.SOFTEN_PROMPT, modified_pass=modified, changes={"arc_preamble": "softened", "segments": "softened all prompts"}, cost_tier="free", estimated_cost_usd=_estimated_cost_of_next_generation(modified), ) @_register( name=RetryStrategyName.BUMP_MOTION_INTENSITY, cost_tier="free", applicable_modes=(FailureMode.MOTION_FAILURE,), description=( "Bump motion_preset intensity one level up per segment. If no " "motion_preset exists, append kinetic motion language to prompt." ), ) def strategy_bump_motion_intensity( coverage_pass: CoveragePass, pass_result: PassResult, ) -> StrategyDiff: _INTENSITY_LADDER = ["low", "medium", "high", "extreme"] new_segments = [] for seg in coverage_pass.segments: preset = seg.motion_preset if preset and preset.get("intensity"): current = preset["intensity"] try: idx = _INTENSITY_LADDER.index(current) except ValueError: idx = len(_INTENSITY_LADDER) - 1 next_idx = min(idx + 1, len(_INTENSITY_LADDER) - 1) new_preset = dict(preset) new_preset["intensity"] = _INTENSITY_LADDER[next_idx] new_segments.append(dataclasses.replace(seg, motion_preset=new_preset)) else: motion_suffix = ( " Fast, dynamic kinetic camera movement. " "Strong physical action with visible momentum." ) new_segments.append( dataclasses.replace(seg, prompt=seg.prompt + motion_suffix) ) modified = dataclasses.replace(coverage_pass, segments=new_segments) return StrategyDiff( strategy_name=RetryStrategyName.BUMP_MOTION_INTENSITY, modified_pass=modified, changes={ "segments": "bumped motion intensity one level or appended kinetic language" }, cost_tier="free", estimated_cost_usd=_estimated_cost_of_next_generation(modified), ) @_register( name=RetryStrategyName.ADD_FILM_STOCK_ANCHOR, cost_tier="free", applicable_modes=(FailureMode.STYLE_DRIFT,), description=( "Prepend Kodak Vision3 500T film stock anchor to arc_preamble " "for consistent visual language across segments." ), ) def strategy_add_film_stock_anchor( coverage_pass: CoveragePass, pass_result: PassResult, ) -> StrategyDiff: film_anchor = ( "Shot on Kodak Vision3 500T, 35mm. Consistent warm grain, " "amber shadows, blue-hour highlights. " ) new_preamble = film_anchor + (coverage_pass.arc_preamble or "") modified = dataclasses.replace(coverage_pass, arc_preamble=new_preamble) return StrategyDiff( strategy_name=RetryStrategyName.ADD_FILM_STOCK_ANCHOR, modified_pass=modified, changes={"arc_preamble": "prepended Kodak Vision3 500T film stock anchor"}, cost_tier="free", estimated_cost_usd=_estimated_cost_of_next_generation(modified), ) @_register( name=RetryStrategyName.FORMAT_MODE_TIMELINE, cost_tier="free", applicable_modes=(FailureMode.CUTS_TOO_SOFT, FailureMode.MOTION_FAILURE), description=( "Set generation_config format_mode to 'timeline'. The prompt " "builder already supports this mode — no prompt_engine changes." ), ) def strategy_format_mode_timeline( coverage_pass: CoveragePass, pass_result: PassResult, ) -> StrategyDiff: new_gen_config = dict(coverage_pass.generation_config or {}) prev = new_gen_config.get("format_mode", "default") new_gen_config["format_mode"] = "timeline" modified = dataclasses.replace(coverage_pass, generation_config=new_gen_config) return StrategyDiff( strategy_name=RetryStrategyName.FORMAT_MODE_TIMELINE, modified_pass=modified, changes={"generation_config.format_mode": f"{prev} → timeline"}, cost_tier="free", estimated_cost_usd=_estimated_cost_of_next_generation(modified), ) @_register( name=RetryStrategyName.FORMAT_MODE_HYBRID, cost_tier="free", applicable_modes=(FailureMode.CUTS_TOO_SOFT, FailureMode.MOTION_FAILURE), description=( "Set generation_config format_mode to 'hybrid'. The prompt " "builder already supports this mode — no prompt_engine changes." ), ) def strategy_format_mode_hybrid( coverage_pass: CoveragePass, pass_result: PassResult, ) -> StrategyDiff: new_gen_config = dict(coverage_pass.generation_config or {}) prev = new_gen_config.get("format_mode", "default") new_gen_config["format_mode"] = "hybrid" modified = dataclasses.replace(coverage_pass, generation_config=new_gen_config) return StrategyDiff( strategy_name=RetryStrategyName.FORMAT_MODE_HYBRID, modified_pass=modified, changes={"generation_config.format_mode": f"{prev} → hybrid"}, cost_tier="free", estimated_cost_usd=_estimated_cost_of_next_generation(modified), ) @_register( name=RetryStrategyName.REDUCE_FACE_PROMINENCE, cost_tier="free", applicable_modes=(FailureMode.CONTENT_FILTER_HARD_BLOCK,), description=( "For close-up segments (CU/ECU/MCU), widen the shot type and add " "face-obscuring direction to reduce content filter triggers." ), ) def strategy_reduce_face_prominence( coverage_pass: CoveragePass, pass_result: PassResult, ) -> StrategyDiff: _FACE_SUFFIX = ( " Focus on body language and environment, " "face partially obscured by shadow or angle." ) new_segments = [] changed_count = 0 for seg in coverage_pass.segments: if seg.shot_type in ("CU", "ECU", "MCU"): # CU → MCU, ECU → MCU, MCU → MS new_shot_type = "MCU" if seg.shot_type in ("CU", "ECU") else "MS" new_segments.append( dataclasses.replace( seg, shot_type=new_shot_type, prompt=seg.prompt + _FACE_SUFFIX, ) ) changed_count += 1 else: new_segments.append(seg) modified = dataclasses.replace(coverage_pass, segments=new_segments) return StrategyDiff( strategy_name=RetryStrategyName.REDUCE_FACE_PROMINENCE, modified_pass=modified, changes={ "segments": f"widened shot type + added face-obscuring direction on {changed_count} close-up segments", }, cost_tier="free", estimated_cost_usd=_estimated_cost_of_next_generation(modified), ) # (Categories C-G filled in by Phases 4-6) # ── Category C — Start frame / blueprint strategies ── @_register( name=RetryStrategyName.REGEN_PREVIZ_SWAP_FRAME, cost_tier="expensive", applicable_modes=(FailureMode.IDENTITY_DRIFT, FailureMode.COMPOSITION_WRONG), description=( "Mark the pass for previz frame regeneration on the worst segment. " "The StrategyEngine handles the actual regen call after strategy " "application. Expensive because it triggers a new keyframe generation." ), ) def strategy_regen_previz_swap_frame( coverage_pass: CoveragePass, pass_result: PassResult, ) -> StrategyDiff: worst = _worst_segment(pass_result) segment_index = 0 if worst is not None: for i, seg in enumerate(coverage_pass.segments): if seg.source_shot_id == getattr(worst, "source_shot_id", None): segment_index = i break new_gen_config = dict(coverage_pass.generation_config or {}) new_gen_config["regen_previz_for_segment"] = segment_index modified = dataclasses.replace(coverage_pass, generation_config=new_gen_config) return StrategyDiff( strategy_name=RetryStrategyName.REGEN_PREVIZ_SWAP_FRAME, modified_pass=modified, changes={"generation_config.regen_previz_for_segment": str(segment_index)}, cost_tier="expensive", estimated_cost_usd=_estimated_cost_of_next_generation(modified) + 0.10, ) @_register( name=RetryStrategyName.SWITCH_I2V_TO_R2V, cost_tier="cheap", applicable_modes=(FailureMode.CONTENT_FILTER_HARD_BLOCK,), description=( "Clear the start frame / blueprint, forcing the retry through the " "R2V (ref-to-video) path instead of I2V. Most effective content-" "filter escape per pipeline-learnings §9n — the start frame often " "contains the flagged content." ), ) def strategy_switch_i2v_to_r2v( coverage_pass: CoveragePass, pass_result: PassResult, ) -> StrategyDiff: prev_path = coverage_pass.blueprint_image_path or "(none)" modified = dataclasses.replace( coverage_pass, blueprint_image_path=None, blueprint_source="", ) return StrategyDiff( strategy_name=RetryStrategyName.SWITCH_I2V_TO_R2V, modified_pass=modified, changes={ "blueprint_image_path": f"{prev_path} → None (I2V→R2V for content-filter escape per §9n)" }, cost_tier="cheap", estimated_cost_usd=_estimated_cost_of_next_generation(modified), ) # ── Category D — Pass structure strategies ── @_register( name=RetryStrategyName.REORDER_SEGMENTS, cost_tier="free", applicable_modes=(FailureMode.IDENTITY_DRIFT, FailureMode.CUTS_TOO_SOFT), description=( "Move the worst-scoring segment to position 0 so the model sees it " "first. Seedance's recency bias means the opening segment gets the " "strongest identity lock. Recomputes shot_range for the new order." ), ) def strategy_reorder_segments( coverage_pass: CoveragePass, pass_result: PassResult, ) -> StrategyDiff: worst = _worst_segment(pass_result) if worst is None: return StrategyDiff( strategy_name=RetryStrategyName.REORDER_SEGMENTS, modified_pass=coverage_pass, changes={}, cost_tier="free", estimated_cost_usd=_estimated_cost_of_next_generation(coverage_pass), applicable=False, ) worst_shot_id = worst.source_shot_id worst_seg = None remaining = [] for seg in coverage_pass.segments: if worst_seg is None and seg.source_shot_id == worst_shot_id: worst_seg = seg else: remaining.append(seg) if worst_seg is None: return StrategyDiff( strategy_name=RetryStrategyName.REORDER_SEGMENTS, modified_pass=coverage_pass, changes={}, cost_tier="free", estimated_cost_usd=_estimated_cost_of_next_generation(coverage_pass), applicable=False, ) new_segments = [worst_seg] + remaining new_shot_range = (new_segments[0].source_shot_id, new_segments[-1].source_shot_id) modified = dataclasses.replace( coverage_pass, segments=new_segments, shot_range=new_shot_range, ) return StrategyDiff( strategy_name=RetryStrategyName.REORDER_SEGMENTS, modified_pass=modified, changes={ "segments": f"moved {worst_shot_id} to position 0", "shot_range": str(new_shot_range), }, cost_tier="free", estimated_cost_usd=_estimated_cost_of_next_generation(modified), ) @_register( name=RetryStrategyName.DROP_WEAK_SEGMENT, cost_tier="free", applicable_modes=(FailureMode.GATE_MECHANICAL,), description=( "Remove the worst-scoring segment entirely. The remaining segments " "form a shorter but mechanically sound pass. Not applicable if the " "pass has only 1 segment." ), ) def strategy_drop_weak_segment( coverage_pass: CoveragePass, pass_result: PassResult, ) -> StrategyDiff: if len(coverage_pass.segments) <= 1: return StrategyDiff( strategy_name=RetryStrategyName.DROP_WEAK_SEGMENT, modified_pass=coverage_pass, changes={}, cost_tier="free", estimated_cost_usd=_estimated_cost_of_next_generation(coverage_pass), applicable=False, ) worst = _worst_segment(pass_result) if worst is None: return StrategyDiff( strategy_name=RetryStrategyName.DROP_WEAK_SEGMENT, modified_pass=coverage_pass, changes={}, cost_tier="free", estimated_cost_usd=_estimated_cost_of_next_generation(coverage_pass), applicable=False, ) worst_shot_id = worst.source_shot_id new_segments = [ seg for seg in coverage_pass.segments if seg.source_shot_id != worst_shot_id ] if not new_segments: return StrategyDiff( strategy_name=RetryStrategyName.DROP_WEAK_SEGMENT, modified_pass=coverage_pass, changes={}, cost_tier="free", estimated_cost_usd=_estimated_cost_of_next_generation(coverage_pass), applicable=False, ) new_shot_range = (new_segments[0].source_shot_id, new_segments[-1].source_shot_id) modified = dataclasses.replace( coverage_pass, segments=new_segments, shot_range=new_shot_range, ) return StrategyDiff( strategy_name=RetryStrategyName.DROP_WEAK_SEGMENT, modified_pass=modified, changes={ "segments": f"dropped {worst_shot_id} ({len(coverage_pass.segments)} → {len(new_segments)} segments)", "shot_range": str(new_shot_range), }, cost_tier="free", estimated_cost_usd=_estimated_cost_of_next_generation(modified), ) # ── Category E — Generation config strategies ── @_register( name=RetryStrategyName.FLIP_ANCHOR_DURATION, cost_tier="free", applicable_modes=( FailureMode.IDENTITY_DRIFT, FailureMode.MOTION_FAILURE, FailureMode.CUTS_TOO_SOFT, FailureMode.PROMPT_DURATION_MISMATCH, ), description=( "Toggle anchor_duration_s between None and 1. When set to 1, the " "model anchors segment boundaries more tightly. When None, it flows " "freely. Toggling can break out of repetitive failure loops." ), ) def strategy_flip_anchor_duration( coverage_pass: CoveragePass, pass_result: PassResult, ) -> StrategyDiff: new_gen_config = dict(coverage_pass.generation_config or {}) current = new_gen_config.get("anchor_duration_s") if current is None: new_gen_config["anchor_duration_s"] = 1 change_desc = "None → 1" else: new_gen_config["anchor_duration_s"] = None change_desc = f"{current} → None" modified = dataclasses.replace(coverage_pass, generation_config=new_gen_config) return StrategyDiff( strategy_name=RetryStrategyName.FLIP_ANCHOR_DURATION, modified_pass=modified, changes={"generation_config.anchor_duration_s": change_desc}, cost_tier="free", estimated_cost_usd=_estimated_cost_of_next_generation(modified), ) @_register( name=RetryStrategyName.CHANGE_RESOLUTION, cost_tier="free", applicable_modes=(FailureMode.COST_OVERRUN,), description=( "Force resolution to 480p. Only appears in the COST_OVERRUN " "escalation chain as a last-resort cost reduction." ), ) def strategy_change_resolution( coverage_pass: CoveragePass, pass_result: PassResult, ) -> StrategyDiff: new_gen_config = dict(coverage_pass.generation_config or {}) prev = new_gen_config.get("resolution", "720p") new_gen_config["resolution"] = "480p" modified = dataclasses.replace(coverage_pass, generation_config=new_gen_config) return StrategyDiff( strategy_name=RetryStrategyName.CHANGE_RESOLUTION, modified_pass=modified, changes={"generation_config.resolution": f"{prev} → 480p"}, cost_tier="free", estimated_cost_usd=_estimated_cost_of_next_generation(modified), ) @_register( name=RetryStrategyName.REDUCE_DURATION, cost_tier="free", applicable_modes=( FailureMode.IDENTITY_DRIFT, FailureMode.PROMPT_DURATION_MISMATCH, FailureMode.COST_OVERRUN, ), description=( "Scale total pass duration down to 80% (never below 3s per segment). " "Shorter passes reduce cost and give the model less time to drift." ), ) def strategy_reduce_duration( coverage_pass: CoveragePass, pass_result: PassResult, ) -> StrategyDiff: segments = coverage_pass.segments if not segments: return StrategyDiff( strategy_name=RetryStrategyName.REDUCE_DURATION, modified_pass=coverage_pass, changes={}, cost_tier="free", estimated_cost_usd=_estimated_cost_of_next_generation(coverage_pass), applicable=False, ) old_total = coverage_pass.duration_s new_total = max(3 * len(segments), int(old_total * 0.8)) # Scale each segment proportionally, floor at 3s new_segments = [] for seg in segments: if old_total > 0: scaled = max(3, int(seg.duration_s * new_total / old_total)) else: scaled = 3 new_segments.append(dataclasses.replace(seg, duration_s=scaled)) modified = dataclasses.replace(coverage_pass, segments=new_segments) return StrategyDiff( strategy_name=RetryStrategyName.REDUCE_DURATION, modified_pass=modified, changes={ "duration_s": f"{old_total}s → {modified.duration_s}s ({len(segments)} segments)" }, cost_tier="free", estimated_cost_usd=_estimated_cost_of_next_generation(modified), ) @_register( name=RetryStrategyName.REDUCE_TAKES_COUNT, cost_tier="free", applicable_modes=(FailureMode.COST_OVERRUN,), description="Force takes_count to 1 to reduce generation cost.", ) def strategy_reduce_takes_count( coverage_pass: CoveragePass, pass_result: PassResult, ) -> StrategyDiff: prev = coverage_pass.takes_count modified = dataclasses.replace(coverage_pass, takes_count=1) return StrategyDiff( strategy_name=RetryStrategyName.REDUCE_TAKES_COUNT, modified_pass=modified, changes={"takes_count": f"{prev} → 1"}, cost_tier="free", estimated_cost_usd=_estimated_cost_of_next_generation(modified), ) @_register( name=RetryStrategyName.CHANGE_SEED, cost_tier="free", applicable_modes=( FailureMode.GATE_MECHANICAL, FailureMode.TRANSIENT, FailureMode.UNKNOWN, ), description=( "Inject a random seed into generation_config. If the Seedance API " "doesn't read 'seed', this is a no-op but safe — the request " "payload change may still affect server-side routing." ), ) def strategy_change_seed( coverage_pass: CoveragePass, pass_result: PassResult, ) -> StrategyDiff: new_seed = random.randint(0, 2**32 - 1) new_gen_config = {**(coverage_pass.generation_config or {}), "seed": new_seed} modified = dataclasses.replace(coverage_pass, generation_config=new_gen_config) return StrategyDiff( strategy_name=RetryStrategyName.CHANGE_SEED, modified_pass=modified, changes={"generation_config.seed": str(new_seed)}, cost_tier="free", estimated_cost_usd=_estimated_cost_of_next_generation(modified), ) # ── Category F — Cost / routing strategies ── def _tier_switch_strategy( coverage_pass: CoveragePass, *, target_tier: str, guard_tier: str, strategy_name: RetryStrategyName, cost_tier: str, ) -> StrategyDiff: current_tier = ( (coverage_pass.generation_config or {}).get("tier") or os.environ.get("SEEDANCE_TIER", "pro").lower() ) if current_tier == guard_tier: return StrategyDiff( strategy_name=strategy_name, modified_pass=coverage_pass, changes={}, cost_tier=cost_tier, estimated_cost_usd=_estimated_cost_of_next_generation(coverage_pass), applicable=False, ) new_gen_config = {**(coverage_pass.generation_config or {}), "tier": target_tier} modified = dataclasses.replace(coverage_pass, generation_config=new_gen_config) return StrategyDiff( strategy_name=strategy_name, modified_pass=modified, changes={"generation_config.tier": f"{current_tier} → {target_tier}"}, cost_tier=cost_tier, estimated_cost_usd=_estimated_cost_of_next_generation(modified), ) @_register( name=RetryStrategyName.UPGRADE_FAST_TO_PRO, cost_tier="cheap", applicable_modes=( FailureMode.IDENTITY_DRIFT, FailureMode.COMPOSITION_WRONG, FailureMode.MOTION_FAILURE, FailureMode.STYLE_DRIFT, ), description=( "Switch generation_config['tier'] to 'pro'. Pro tier uses the full " "Seedance 2.0 model with higher quality output. Only applicable when " "current tier is 'fast' — no-op if already on pro." ), ) def strategy_upgrade_fast_to_pro( coverage_pass: CoveragePass, pass_result: PassResult, ) -> StrategyDiff: return _tier_switch_strategy( coverage_pass, target_tier="pro", guard_tier="pro", strategy_name=RetryStrategyName.UPGRADE_FAST_TO_PRO, cost_tier="cheap", ) @_register( name=RetryStrategyName.DOWNGRADE_PRO_TO_FAST, cost_tier="free", applicable_modes=(FailureMode.COST_OVERRUN,), description=( "Switch generation_config['tier'] to 'fast'. Fast tier is ~2x cheaper. " "Only applicable when current tier is 'pro' — no-op if already on fast." ), ) def strategy_downgrade_pro_to_fast( coverage_pass: CoveragePass, pass_result: PassResult, ) -> StrategyDiff: return _tier_switch_strategy( coverage_pass, target_tier="fast", guard_tier="fast", strategy_name=RetryStrategyName.DOWNGRADE_PRO_TO_FAST, cost_tier="free", ) # ── Category G — Human routing ── @_register( name=RetryStrategyName.ESCALATE_TO_HUMAN, cost_tier="free", applicable_modes=( FailureMode.IDENTITY_DRIFT, FailureMode.COMPOSITION_WRONG, FailureMode.MOTION_FAILURE, FailureMode.STYLE_DRIFT, FailureMode.CUTS_TOO_SOFT, FailureMode.CONTENT_FILTER_HARD_BLOCK, FailureMode.REF_BLEED, FailureMode.PROMPT_DURATION_MISMATCH, FailureMode.COST_OVERRUN, FailureMode.GATE_MECHANICAL, FailureMode.TRANSIENT, FailureMode.UNKNOWN, ), description="Terminal strategy — route to human review queue.", ) def strategy_escalate_to_human( coverage_pass: CoveragePass, pass_result: PassResult, ) -> StrategyDiff: return StrategyDiff( strategy_name=RetryStrategyName.ESCALATE_TO_HUMAN, modified_pass=coverage_pass, changes={"status": "strategy_exhausted"}, cost_tier="free", estimated_cost_usd=0.0, ) # ============================================================================ # Failure-mode detection — Tier 0 + Tier 1 (v1), Tier 2 (Opus 4.7 vision) # ============================================================================ _OPUS_CLASSIFIER_PROMPT = """\ You are a video quality classifier for a cinematic AI production pipeline. I am showing you a frame extracted from a failed generated video clip. Classify the PRIMARY failure reason from exactly one of these modes: - IDENTITY_DRIFT: Character looks different from reference (wrong face, body type, skin tone, hair) - COMPOSITION_WRONG: Wrong shot framing (expected CU but got WS, wrong subject in frame) - STYLE_DRIFT: Visual style wrong (wrong color palette, lighting style, film grain) - MOTION_FAILURE: Motion is wrong (static when should move, jittery, impossible physics) - UNKNOWN: Cannot determine from this frame Respond ONLY with a JSON object: {"failure_mode": "", "confidence": <0.0-1.0>, "reason": ""} """ _OPUS_MODE_MAP = { "IDENTITY_DRIFT": FailureMode.IDENTITY_DRIFT, "COMPOSITION_WRONG": FailureMode.COMPOSITION_WRONG, "STYLE_DRIFT": FailureMode.STYLE_DRIFT, "MOTION_FAILURE": FailureMode.MOTION_FAILURE, } def _classify_with_opus(pass_result: PassResult) -> tuple[FailureMode, float]: """Extract first frame from failed video, send to Opus 4.7 for Tier 2 classification.""" import base64 import json import subprocess import tempfile video_path = pass_result.video_path if not video_path: return (FailureMode.UNKNOWN, 0.30) try: with tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) as tf: frame_path = tf.name try: result = subprocess.run( ["ffmpeg", "-y", "-i", str(video_path), "-vframes", "1", "-q:v", "3", frame_path], capture_output=True, timeout=15, ) if result.returncode != 0: return (FailureMode.UNKNOWN, 0.30) with open(frame_path, "rb") as f: frame_b64 = base64.standard_b64encode(f.read()).decode() finally: Path(frame_path).unlink(missing_ok=True) from recoil.tools.engine_constants import get_anthropic_client, ANTHROPIC_OPUS client = get_anthropic_client() if client is None: _warn_tier2_opus_unavailable_once() return (FailureMode.UNKNOWN, 0.30) response = client.messages.create( # REC-38: route through the SSOT constant (was hardcoded retired # 'claude-opus-4-7'). This is the de-facto identity classifier today. model=ANTHROPIC_OPUS, max_tokens=256, messages=[ { "role": "user", "content": [ { "type": "image", "source": {"type": "base64", "media_type": "image/jpeg", "data": frame_b64}, }, {"type": "text", "text": _OPUS_CLASSIFIER_PROMPT}, ], } ], ) raw = response.content[0].text.strip() if "```" in raw: raw = raw.split("```")[1].removeprefix("json").strip() parsed = json.loads(raw) mode_str = parsed.get("failure_mode", "UNKNOWN") conf = float(parsed.get("confidence", 0.30)) mode = _OPUS_MODE_MAP.get(mode_str, FailureMode.UNKNOWN) logger.info("Opus Tier 2 classifier: %s (conf=%.2f) — %s", mode_str, conf, parsed.get("reason", "")) return (mode, conf) except Exception as exc: logger.warning("Opus Tier 2 classifier failed: %s", exc) return (FailureMode.UNKNOWN, 0.30) def detect_failure_mode( pass_result: PassResult, coverage_pass: CoveragePass, ) -> tuple[FailureMode, float]: """Tiered failure-mode classifier. Returns (mode, confidence 0.0-1.0). V1 implementation covers Tier 0 (API errors), Tier 1 (gate results + cut count + duration mismatch), and Tier 2 Opus vision classification for cases Tier 0+1 can't classify. The Tier 2 classifier currently depends on an available Anthropic SDK client. """ # ── Tier 0: delegate to canonical (string + http_status) ───────────── from recoil.pipeline.core.failure_mode import ( classify_failure as _canonical_classify, UnknownFailureEscalation, ) error = (pass_result.error or "").lower() if error: try: mode, conf = _canonical_classify( error_text=error, escalate_unknown=False, # pass-level — fall through to Tier 1 caller="strategy_registry.detect_failure_mode", ) except UnknownFailureEscalation: mode, conf = FailureMode.UNKNOWN, 0.30 if mode is not FailureMode.UNKNOWN and mode is not FailureMode.NONE: return (mode, conf) # Pass didn't even produce a video — treat as TRANSIENT if no error, else UNKNOWN if not pass_result.video_path and not pass_result.success: return (FailureMode.TRANSIENT if not error else FailureMode.UNKNOWN, 0.5) # ── Tier 1: Gate results ───────────────────────────────────────────── # Identity: any segment with score < 0.7 = identity drift for seg in pass_result.segment_results or []: score = getattr(seg, "identity_score", None) if score is not None and score < 0.7: return (FailureMode.IDENTITY_DRIFT, 0.85) # Cuts too soft: detected < expected (with tolerance) expected = getattr(pass_result, "expected_cuts", 0) or 0 detected = getattr(pass_result, "detected_cuts", None) if expected > 0 and detected is not None and detected < max(1, expected - 1): gap = expected - detected confidence = 0.45 if gap == 1 else 0.70 if gap >= 3 else 0.55 return (FailureMode.CUTS_TOO_SOFT, confidence) # Duration mismatch: sum of segment durations ≠ actual video duration prompt_total_s = sum(s.duration_s for s in coverage_pass.segments) if ( coverage_pass.duration_s and abs(prompt_total_s - coverage_pass.duration_s) > 1.0 ): return (FailureMode.PROMPT_DURATION_MISMATCH, 0.80) # ── Tier 1.5: any unusable segment → mechanical failure ────────────── unusable = [ s for s in (pass_result.segment_results or []) if not getattr(s, "usable", True) ] if unusable: return (FailureMode.GATE_MECHANICAL, 0.60) # ── Tier 2: Opus 4.7 vision classification ─────────────────────────── if pass_result.video_path: mode, conf = _classify_with_opus(pass_result) if mode != FailureMode.UNKNOWN: return (mode, conf) return (FailureMode.UNKNOWN, 0.30) # ============================================================================ # StrategyEngine — orchestrator # ============================================================================ class StrategyEngine: """Orchestrates strategy selection + application for failed passes.""" def __init__(self, learning: "LearningEngine", model: str = "seeddance-2.0"): self._learning = learning self._model = model def select_and_apply( self, failure_mode: FailureMode, coverage_pass: CoveragePass, pass_result: PassResult, already_tried: list[str], original_pass: CoveragePass, cumulative_retry_cost: float, cost_policy: RetryCostPolicy, ) -> Optional[StrategyDiff]: """Select and apply the next strategy. Returns StrategyDiff or None if exhausted. Accumulation rules: - Free/cheap strategies: applied on top of `coverage_pass` (previous mutations preserved). - Expensive strategies: reset to `original_pass` before applying. - Hard cap: 3 accumulated free strategies before forced escalation. - Cost cap: skip a strategy if estimated_cost would exceed cost_policy.max_retry_spend_usd. """ # 1. Ask LearningEngine for recommendation rec_name = self._learning.recommend_strategy( failure_mode=failure_mode.value, model=self._model, already_tried=already_tried, ) selection_basis = "learned_reorder" if rec_name else "static_chain" # 2. If no recommendation, fall through to static chain if rec_name is None: static_chain = ESCALATION_CHAINS.get(failure_mode, []) for name in static_chain: if name.value in already_tried: continue entry = STRATEGY_REGISTRY[name] projected_cost = cumulative_retry_cost + _estimate_strategy_cost( entry, coverage_pass ) if projected_cost > cost_policy.max_retry_spend_usd: continue rec_name = name break else: # Convert string recommendation back to enum try: rec_name = RetryStrategyName(rec_name) except ValueError: rec_name = None # Verify cost cap even for learned recommendations if rec_name is not None: entry = STRATEGY_REGISTRY.get(rec_name) if entry: projected_cost = cumulative_retry_cost + _estimate_strategy_cost( entry, coverage_pass ) if projected_cost > cost_policy.max_retry_spend_usd: # Learned rec too expensive — fall through to static chain rec_name = None static_chain = ESCALATION_CHAINS.get(failure_mode, []) for name in static_chain: if name.value in already_tried: continue st_entry = STRATEGY_REGISTRY[name] st_cost = cumulative_retry_cost + _estimate_strategy_cost( st_entry, coverage_pass ) if st_cost > cost_policy.max_retry_spend_usd: continue rec_name = name break if rec_name is None: return _escalate_to_human_diff(coverage_pass) # 3. Apply the selected strategy entry = STRATEGY_REGISTRY[rec_name] if entry.cost_tier == "expensive": input_pass = original_pass else: # Check free-strategy accumulation cap free_accumulated = [] for s in already_tried: try: sn = RetryStrategyName(s) if sn in STRATEGY_REGISTRY and STRATEGY_REGISTRY[sn].cost_tier in ( "free", "cheap", ): free_accumulated.append(s) except ValueError: pass if len(free_accumulated) >= 3: expensive_avail = [ n for n in ESCALATION_CHAINS.get(failure_mode, []) if STRATEGY_REGISTRY[n].cost_tier == "expensive" and n.value not in already_tried ] if expensive_avail: exp_entry = STRATEGY_REGISTRY[expensive_avail[0]] projected_cost = cumulative_retry_cost + _estimate_strategy_cost( exp_entry, original_pass ) if projected_cost > cost_policy.max_retry_spend_usd: return _escalate_to_human_diff(coverage_pass) entry = exp_entry input_pass = original_pass else: return _escalate_to_human_diff(coverage_pass) else: input_pass = coverage_pass diff = entry.fn(input_pass, pass_result) return dataclasses.replace( diff, changes={ **diff.changes, "_selection_basis": selection_basis, }, ) def _estimate_strategy_cost(entry: StrategyEntry, coverage_pass: CoveragePass) -> float: """Estimate the incremental cost of applying this strategy + re-running.""" base = _estimated_cost_of_next_generation(coverage_pass) if entry.cost_tier == "expensive": return base + 0.10 return base def _escalate_to_human_diff(coverage_pass: CoveragePass) -> StrategyDiff: """Build the canonical escape-hatch diff.""" entry = STRATEGY_REGISTRY[RetryStrategyName.ESCALATE_TO_HUMAN] return entry.fn(coverage_pass, None) def _estimated_cost_of_next_generation(coverage_pass: CoveragePass) -> float: """Rough cost estimate for the next generation of this pass. Used by StrategyEngine's cost cap check. Reads the pass's current generation_config for resolution/duration/tier and multiplies by the model_profiles rate. Does NOT account for upload/polling latency. """ from recoil.core import model_profiles gen = coverage_pass.generation_config or {} model = gen.get("model") or "seeddance-2.0" resolution = gen.get("resolution") or "720p" duration_s = coverage_pass.duration_s or 10 profile = model_profiles.get_profile(model) # Default to fal Fast 720p observed rate if tiering unspecified tier = gen.get("tier") or "fast" res_suffix = "" if resolution == "720p" else f"_{resolution}" key = f"cost_per_second_{tier}{res_suffix}" rate = profile.get(key) or profile.get("cost_per_second_fast", 0.2419) return rate * duration_s # ============================================================================ # CP-9 Phase 7 — score-card → FailureMode bridge (substrate only) # # Per loraverse SYNTHESIS Q4 (LOCKED) + Phase 1 audit § 12f items 4-7: # adds a sibling factory function that maps a PanelOfJudges scorecard # (produced by `pipeline.core.eval.PanelOfJudges.score`) onto the SAME # return shape as `detect_failure_mode(...)` — `tuple[FailureMode, float]` # — so callers can swap inputs without changing downstream consumption. # # Substrate only — production switchover at # `pipeline/orchestrator/production_loop.py:1087-1111` is a follow-up CP # gated on JT sign-off after living with PanelOfJudges output for N # production runs. CP-9 does NOT modify `production_loop.py` or # `detect_failure_mode` body/signature (Phase 7 hard gate, Phase 9 # byte-stability check enforces). # # FailureMode is a `(str, Enum)` with 23 stable members — only existing # members are returned; do NOT invent new modes. Score-bucket mapping is # a deliberate placeholder that JT may refine in retry-strategy iteration: # panel_score is None → (UNKNOWN, 0.0) — no score, no signal # panel_score >= 0.7 → (NONE, panel_score) — confident "good" # panel_score >= 0.4 → (UNKNOWN, panel_score) — ambiguous mid-band # panel_score < 0.4 → (IDENTITY_DRIFT, — confident "bad" # 1.0 - panel_score) (placeholder mode; # JT picks the right # bucket→mode map) # ============================================================================ def from_score_card(score_card: dict) -> tuple[FailureMode, float]: """Map a PanelOfJudges scorecard to a FailureMode + confidence. Mirrors :func:`detect_failure_mode` return shape so callers can swap inputs without changing downstream consumption. Substrate only — production switchover is gated on JT sign-off (CP-N+). Args: score_card: a dict with keys ``{panel_id, panel_score, panel_warnings, judges, aggregation, panel_cost_usd}`` as produced by :meth:`pipeline.core.eval.PanelOfJudges.score`. Only ``panel_score`` is read by this function; other keys are tolerated (and may be consulted by future revisions). Returns: ``(FailureMode, confidence)`` where ``confidence`` is in ``[0.0, 1.0]``. Mapping bands (placeholder — see module docstring): - ``panel_score is None`` → ``(FailureMode.UNKNOWN, 0.0)`` (panel produced no score; downstream may treat as TRANSIENT retry or escalation; this function is purely informational) - ``panel_score >= 0.7`` → ``(FailureMode.NONE, panel_score)`` (confident "good"; no failure mode flagged) - ``panel_score >= 0.4`` → ``(FailureMode.UNKNOWN, panel_score)`` (ambiguous mid-band; downstream picks retry vs. escalation) - ``panel_score < 0.4`` → ``(FailureMode.IDENTITY_DRIFT, 1.0 - panel_score)`` (confident "bad"; placeholder mode — JT will refine the bucket→mode map post-CP-9) """ import math panel_score = score_card.get("panel_score") if panel_score is None: return (FailureMode.UNKNOWN, 0.0) try: ps_float = float(panel_score) except (TypeError, ValueError): return (FailureMode.UNKNOWN, 0.0) # NaN / Inf must NOT pass through the clamp — min/max with NaN is # implementation-defined in Python and can produce surprising bucket # routing. Treat as UNKNOWN, parallel to compute_aggregate_score's # Phase 5 guard. if math.isnan(ps_float) or math.isinf(ps_float): return (FailureMode.UNKNOWN, 0.0) # Clamp to [0.0, 1.0] — docstring promises confidence in this range, but # a malformed scorecard could pass through negative or >1.0. Clamp keeps # the contract honest without silently producing confidence > 1.0. panel_score = max(0.0, min(1.0, ps_float)) if panel_score >= 0.7: return (FailureMode.NONE, panel_score) if panel_score >= 0.4: return (FailureMode.UNKNOWN, panel_score) # ambiguous mid-band return (FailureMode.IDENTITY_DRIFT, 1.0 - panel_score) # confident "bad"