"""Provider adapter contracts. Every video-generation provider (fal.ai, atlas, piapi, ...) implements the `ProviderAdapter` Protocol. The VideoModelClient never sees provider-specific code — it only calls the Protocol methods on whatever adapter the registry returned. This module owns: - ProviderAdapter Protocol (duck-typed contract) - Shared dataclasses: UnifiedVideoPayload, SubmitRequest, PollRequest, PollResult, ProviderJob - ProviderCapabilityError (raised when a payload requests an unsupported capability and no exception is declared in provider_strategy.json) Nothing here does I/O. Nothing imports fal_client, urllib, or sqlite3. Adapters and the registry own I/O. """ from __future__ import annotations from dataclasses import dataclass, field from typing import TYPE_CHECKING, Any, Optional, Protocol, runtime_checkable if TYPE_CHECKING: from recoil.execution.providers.payload_hints import PayloadHints # ------------------------------------------------------------------ # Unified payload — the single shape adapters translate from # ------------------------------------------------------------------ @dataclass class UnifiedVideoPayload: """Model-agnostic video generation request. Adapters translate this into provider-specific submit bodies. Callers assemble one of these once; routing + translation happens downstream in the adapter. """ prompt: str duration_s: int = 5 resolution: str = "720p" aspect_ratio: Optional[str] = None # Start frame (base64 str OR raw bytes OR URL). Presence => I2V. image: Optional[Any] = None # End frame (same shape as `image`). Presence => first_last_frames. image_tail: Optional[Any] = None # Reference images for R2V (list of URLs, local paths, or bytes). reference_images: list = field(default_factory=list) # Reference videos for R2V (list of URLs or local paths to .mp4/.mov/etc.). # Seedance 2.0 accepts up to 3; tagged @Video1..@Video3 in the prompt. reference_videos: list = field(default_factory=list) # Reference audios for R2V (list of URLs or local paths to .mp3/.wav/.m4a). # Seedance 2.0 accepts up to 3; tagged @Audio1..@Audio3 in the prompt. # Native audio input channel — preferred over bundling audio inside a blank # reference video, since the bundle approach drops audio in multi-shot mode. # See pipeline-learnings §17c. reference_audios: list = field(default_factory=list) generate_audio: bool = False negative_prompt: Optional[str] = None # Provider hints — typed at the adapter boundary. # Phase C of the engine-architectural-audit fix sprint (2026-04-30, T2.21/MF-10): # canonical type is a PayloadHints subclass (WanHints, KlingHints, # GoogleHints, StepRunnerHints). Phase 11 narrowed this annotation from # Optional[Any] → Optional[PayloadHints]. Legacy dict callers still work # at runtime (Python annotations aren't enforced) and adapters call # `coerce_to_dict(payload.hints)` from recoil.execution.providers.payload_hints # to read regardless of shape; the narrower annotation flags dict-passers # for static type-checkers and makes the canonical contract explicit. hints: Optional["PayloadHints"] = None # Caller-owned identifiers for observability. shot_id: Optional[str] = None model_id: str = "seeddance-2.0" # ------------------------------------------------------------------ # Transport-level dataclasses # ------------------------------------------------------------------ @dataclass class SubmitRequest: """What the adapter tells the registry to POST.""" method: str # always "POST" today url: str headers: dict body: dict @dataclass class PollRequest: """What the adapter tells the registry to GET when polling.""" method: str # "GET" for status, "GET" for result fetch url: str headers: dict @dataclass class ProviderJob: """Opaque-to-client handle returned from build_submit/parse_submit. `native_id` is the provider's task / request id. `native_state` lets adapters stash provider-specific URLs (e.g. fal.ai status_url, response_url) without leaking them to the VideoModelClient. """ provider_id: str model_id: str native_id: str tier: str duration_s: int resolution: str native_state: dict = field(default_factory=dict) @dataclass class PollResult: """Normalized poll output. status ∈ {"IN_PROGRESS", "COMPLETED", "FAILED"}. When COMPLETED: video_url populated, audio_url optional, observed_cost populated iff provider supplies billing info. When FAILED: error is set. """ status: str video_url: Optional[str] = None audio_url: Optional[str] = None observed_cost: Optional[float] = None error: Optional[str] = None # Raw provider payload preserved for observability writer. raw: dict = field(default_factory=dict) # ------------------------------------------------------------------ # Capability keys — canonical set. Adapters must answer every key. # ------------------------------------------------------------------ CAPABILITY_KEYS: tuple[str, ...] = ( "t2v", # text-to-video "i2v", # image-to-video (start frame) "r2v", # reference-to-video (ref images) "end_frame", # first+last frame pair "audio", # generate audio alongside video "negative_prompt", # honors negative prompt field "resolution_480p", "resolution_720p", "resolution_1080p", ) # ------------------------------------------------------------------ # The Protocol # ------------------------------------------------------------------ @runtime_checkable class ProviderAdapter(Protocol): """Contract every provider adapter implements. Attributes are class-level constants (or properties) the registry reads without instantiating. Methods are pure translation — they do not perform I/O. The registry + VideoModelClient perform I/O using the SubmitRequest / PollRequest the adapter builds. """ provider_id: str supported_models: list # e.g. ["seeddance-2.0"] auth_env_var: str base_url: str capabilities: dict # keys from CAPABILITY_KEYS -> bool max_prompt_chars: int | None status: str # "primary" | "fallback" | "deprecated" def build_submit(self, payload: UnifiedVideoPayload, tier: str) -> SubmitRequest: ... def parse_submit(self, resp: dict, payload: UnifiedVideoPayload, tier: str) -> ProviderJob: ... def build_poll(self, job: ProviderJob) -> PollRequest: ... def parse_poll(self, resp: dict, job: ProviderJob) -> PollResult: ... def build_result_fetch(self, job: ProviderJob) -> Optional[PollRequest]: ... def parse_result(self, resp: dict, job: ProviderJob) -> PollResult: ... def compute_cost(self, duration_s: float, tier: str, profile: dict) -> float: ... # ------------------------------------------------------------------ # Capability refusal error # ------------------------------------------------------------------ class ProviderCapabilityError(RuntimeError): """Raised when a payload requests a capability the selected adapter does not support AND no exception is declared in provider_strategy.json. Never silently degrade — surface the gap. """ def __init__( self, *, model_id: str, provider_id: str, capability: str, supported_providers: list[str], ): self.model_id = model_id self.provider_id = provider_id self.capability = capability self.supported_providers = list(supported_providers) msg = ( f"ProviderCapabilityError: {model_id} via {provider_id} does not " f"support {capability!r}.\n" f"Supported providers for {capability}: {self.supported_providers}\n" f"Resolutions:\n" f" 1. Add {capability!r} to capability_exceptions in provider_strategy.json\n" f" 2. Switch primary provider for this model\n" f" 3. Remove {capability} from the payload" ) super().__init__(msg) __all__ = [ "UnifiedVideoPayload", "SubmitRequest", "PollRequest", "PollResult", "ProviderJob", "ProviderAdapter", "ProviderCapabilityError", "CAPABILITY_KEYS", ]