wip

crates/api_clients/artcraft/artcraft_router/src/generate/generate_video_v2/mod.rs

@@ -2,4 +2,4 @@ pub mod providers;
 pub mod video_generation_request;
 pub mod video_generation_draft;
 pub mod video_generation_draft_or_request;
-mod video_generation_draft_context;
+pub mod video_generation_draft_context;

crates/service/web/storyteller_web/src/http_server/endpoints/omni_gen/generate/video/helpers/bill_wallet.rs

@@ -0,0 +1,44 @@
+//! Shared wallet billing logic for both video generation pipelines.
+
+use log::info;
+use tokens::tokens::generic_inference_jobs::InferenceJobToken;
+use tokens::tokens::users::UserToken;
+use tokens::tokens::wallet_ledger_entries::WalletLedgerEntryToken;
+
+use crate::billing::wallets::attempt_wallet_deduction::attempt_wallet_deduction_else_common_web_error;
+use crate::http_server::common_responses::advanced_common_web_error::AdvancedCommonWebError;
+
+pub struct BillWalletResult {
+  pub apriori_job_token: InferenceJobToken,
+  pub maybe_wallet_ledger_entry_token: Option<WalletLedgerEntryToken>,
+}
+
+/// Generate an apriori job token and bill the user's wallet for the given cost.
+///
+/// If cost is 0, no wallet deduction is made but the apriori token is still generated.
+pub async fn bill_wallet(
+  user_token: &UserToken,
+  cost: u64,
+  mysql_connection: &mut sqlx::pool::PoolConnection<sqlx::MySql>,
+) -> Result<BillWalletResult, AdvancedCommonWebError> {
+  let apriori_job_token = InferenceJobToken::generate();
+
+  info!("Charging wallet: {} credits", cost);
+
+  let maybe_wallet_ledger_entry_token = if cost > 0 {
+    let deduction_result = attempt_wallet_deduction_else_common_web_error(
+      user_token,
+      Some(apriori_job_token.as_str()),
+      cost,
+      mysql_connection,
+    ).await?;
+    Some(deduction_result.ledger_entry_token)
+  } else {
+    None
+  };
+
+  Ok(BillWalletResult {
+    apriori_job_token,
+    maybe_wallet_ledger_entry_token,
+  })
+}

crates/service/web/storyteller_web/src/http_server/endpoints/omni_gen/generate/video/helpers/mod.rs

@@ -0,0 +1 @@
+pub mod bill_wallet;

crates/service/web/storyteller_web/src/http_server/endpoints/omni_gen/generate/video/mod.rs

@@ -1,7 +1,8 @@
 pub mod omni_gen_video_generate_handler;
+mod helpers;
 pub mod pipeline_v1;
+pub mod pipeline_v2;
 mod request_helper;
 
 #[cfg(test)]
 mod tests;
-pub mod pipeline_v2;

crates/service/web/storyteller_web/src/http_server/endpoints/omni_gen/generate/video/omni_gen_video_generate_handler.rs

@@ -10,6 +10,7 @@ use url::Url;
 use artcraft_api_defs::omni_gen::cost_and_generate_requests::omni_gen_video_cost_and_generate_request::OmniGenVideoCostAndGenerateRequest;
 use artcraft_api_defs::omni_gen::generate_response::omni_gen_video_generate_response::OmniGenVideoGenerateResponse;
 use artcraft_router::api::provider::Provider;
+use artcraft_router::generate::generate_video::generate_video_response::GenerateVideoResponse;
 use enums::by_table::prompt_context_items::prompt_context_semantic_type::PromptContextSemanticType;
 use enums::by_table::prompts::prompt_type::PromptType;
 use enums::common::generation::common_model_type::CommonModelType;
@@ -31,14 +32,17 @@ use mysql_queries::queries::prompt_context_items::insert_batch_prompt_context_it
   insert_batch_prompt_context_items, InsertBatchArgs, PromptContextItem,
 };
 use mysql_queries::queries::prompts::insert_prompt::{insert_prompt, InsertPromptArgs};
+use tokens::tokens::characters::CharacterToken;
 use tokens::tokens::generic_inference_jobs::InferenceJobToken;
 use tokens::tokens::media_files::MediaFileToken;
 
-use crate::billing::wallets::attempt_wallet_deduction::attempt_wallet_deduction_else_common_web_error;
 use crate::http_server::common_responses::advanced_common_web_error::AdvancedCommonWebError;
 use crate::http_server::endpoints::generate::common::payments_error_test::payments_error_test;
+use crate::http_server::endpoints::omni_gen::generate::video::helpers::bill_wallet::bill_wallet;
+use crate::http_server::endpoints::omni_gen::generate::video::pipeline_v1::distill_helper::hydrate_to_router_request::hydrate_to_router_request;
 use crate::http_server::endpoints::omni_gen::generate::video::pipeline_v1::distill_video_request::distill_video_request;
 use crate::http_server::endpoints::omni_gen::generate::video::pipeline_v1::execute::execute_generation::execute_generation;
+use crate::http_server::endpoints::omni_gen::generate::video::pipeline_v2::execute::execute_pipeline_v2;
 use crate::http_server::endpoints::omni_gen::generate::video::request_helper::resolve_kinovi_character_ids::resolve_kinovi_character_ids;
 use crate::http_server::validations::validate_idempotency_token_format::validate_idempotency_token_format;
 use crate::state::server_state::ServerState;
@@ -63,7 +67,7 @@ pub async fn omni_gen_video_generate_handler(
   request: Json<OmniGenVideoCostAndGenerateRequest>,
   server_state: web::Data<Arc<ServerState>>,
 ) -> Result<Json<OmniGenVideoGenerateResponse>, AdvancedCommonWebError> {
-  
+
   info!("request: {:?}", request);
 
   payments_error_test(&request.prompt.as_deref().unwrap_or(""))?;
@@ -112,8 +116,6 @@ pub async fn omni_gen_video_generate_handler(
     })?;
 
   // ==================== RESOLVE MEDIA TOKENS ==================== //
-  // Look up media file tokens BEFORE distilling. distill_video_request takes
-  // a pre-computed `MediaFileToken -> Url` map and does no I/O of its own.
 
   let media_file_hydration_map: Option<HashMap<MediaFileToken, Url>> = {
     let mut all_tokens: Vec<MediaFileToken> = Vec::new();
@@ -157,236 +159,245 @@ pub async fn omni_gen_video_generate_handler(
     }
   };
 
-  // ==================== RESOLVE CHARACTER TOKENS ==================== //
+  // ==================== PIPELINE DECISION ==================== //
 
+  let use_v2 = matches!(
+    request.model,
+    Some(CommonVideoModel::Seedance2p0) | Some(CommonVideoModel::Seedance2p0Fast)
+  );
+
+  if use_v2 {
+    execute_v2_pipeline(
+      &http_request,
+      &request,
+      &server_state,
+      &mut mysql_connection,
+      user_token,
+      maybe_avt_token.as_ref(),
+      &idempotency_token,
+      maybe_prompt_model_type,
+      media_file_hydration_map,
+    ).await
+  } else {
+    execute_v1_pipeline(
+      &http_request,
+      &request,
+      &server_state,
+      &mut mysql_connection,
+      user_token,
+      maybe_avt_token.as_ref(),
+      &idempotency_token,
+      maybe_prompt_model_type,
+      media_file_hydration_map,
+    ).await
+  }
+}
+
+// ==================== PIPELINE V1 (existing) ==================== //
+
+#[allow(clippy::too_many_arguments)]
+async fn execute_v1_pipeline(
+  http_request: &HttpRequest,
+  request: &OmniGenVideoCostAndGenerateRequest,
+  server_state: &ServerState,
+  mysql_connection: &mut sqlx::pool::PoolConnection<sqlx::MySql>,
+  user_token: &tokens::tokens::users::UserToken,
+  maybe_avt_token: Option<&tokens::tokens::anonymous_visitor_tracking::AnonymousVisitorTrackingToken>,
+  idempotency_token: &str,
+  maybe_prompt_model_type: Option<CommonModelType>,
+  media_file_hydration_map: Option<HashMap<MediaFileToken, Url>>,
+) -> Result<Json<OmniGenVideoGenerateResponse>, AdvancedCommonWebError> {
   let kinovi_character_ids = resolve_kinovi_character_ids(
     request.reference_character_tokens.as_deref(),
-    &mut mysql_connection,
+    mysql_connection,
   ).await?;
 
-  // ==================== DETERMINE PROVIDER ==================== //
-
   let execution_provider = match request.model {
     Some(CommonVideoModel::Seedance2p0) => Provider::Seedance2Pro,
     Some(CommonVideoModel::Seedance2p0Fast) => Provider::Seedance2Pro,
     _ => Provider::Fal,
   };
 
-  // ==================== DISTILL ==================== //
+  let distilled = distill_video_request(request, media_file_hydration_map.as_ref(), execution_provider)?;
 
-  let distilled = distill_video_request(&request, media_file_hydration_map.as_ref(), execution_provider)?;
-
-  info!("distilled plan: {:?}", distilled.plan);
-
-  // ==================== BILLING ==================== //
+  info!("v1 distilled plan: {:?}", distilled.plan);
 
   let cost = distilled.cost.cost_in_credits.unwrap_or(0);
-
-  info!("Charging wallet: {} credits", cost);
-
-  let apriori_job_token = InferenceJobToken::generate();
-
-  let maybe_deduction_result = if cost > 0 {
-    Some(attempt_wallet_deduction_else_common_web_error(
-      user_token,
-      Some(apriori_job_token.as_str()),
-      cost,
-      &mut mysql_connection,
-    ).await?)
-  } else {
-    None
-  };
-
-  // ==================== EXECUTE GENERATION ==================== //
+  let billing = bill_wallet(user_token, cost, mysql_connection).await?;
 
   let gen_result = execute_generation(
     &distilled,
-    &request,
-    &server_state,
+    request,
+    server_state,
     media_file_hydration_map.as_ref(),
     kinovi_character_ids,
-    maybe_deduction_result.as_ref().map(|d| &d.ledger_entry_token),
-    &mut mysql_connection,
+    billing.maybe_wallet_ledger_entry_token.as_ref(),
+    mysql_connection,
   ).await?;
 
-  // ==================== DB TRANSACTION ==================== //
+  // -- DB writes --
+  let ip_address = get_request_ip(http_request);
+  let mut transaction = mysql_connection.begin().await.map_err(|err| {
+    error!("Error starting MySQL transaction: {:?}", err);
+    AdvancedCommonWebError::from_error(err)
+  })?;
 
-  let ip_address = get_request_ip(&http_request);
-
-  let mut transaction = mysql_connection
-    .begin()
-    .await
-    .map_err(|err| {
-      error!("Error starting MySQL transaction: {:?}", err);
-      AdvancedCommonWebError::from_error(err)
-    })?;
-
-  // -- Prompt --
-
-  let prompt_result = insert_prompt(InsertPromptArgs {
-    maybe_apriori_prompt_token: None,
-    prompt_type: PromptType::ArtcraftApp,
-    maybe_creator_user_token: Some(user_token),
-    maybe_model_type: maybe_prompt_model_type,
-    maybe_generation_provider: Some(GenerationProvider::Artcraft),
-    maybe_positive_prompt: request.prompt.as_deref(),
-    maybe_negative_prompt: request.negative_prompt.as_deref(),
-    maybe_other_args: None,
-    maybe_generation_mode: Some(gen_result.generation_mode),
-    maybe_aspect_ratio: None,
-    maybe_resolution: None,
-    maybe_batch_count: request.video_batch_count.map(|c| c as u8),
-    maybe_generate_audio: request.generate_audio,
-    maybe_duration_seconds: request.duration_seconds.map(|d| d as u32),
-    creator_ip_address: &ip_address,
-    mysql_executor: &mut *transaction,
-    phantom: Default::default(),
-  }).await;
-
-  let prompt_token = match prompt_result {
-    Ok(token) => Some(token),
-    Err(err) => {
-      warn!("Error inserting prompt: {:?}", err);
-      None
-    }
-  };
-
-  // -- Prompt context items --
-
-  if let Some(token) = prompt_token.as_ref() {
-    let mut context_items = Vec::new();
-
-    if let Some(media_token) = &request.start_frame_image_media_token {
-      context_items.push(PromptContextItem {
-        media_token: media_token.clone(),
-        context_semantic_type: PromptContextSemanticType::VidStartFrame,
-      });
-    }
-
-    if let Some(media_token) = &request.end_frame_image_media_token {
-      context_items.push(PromptContextItem {
-        media_token: media_token.clone(),
-        context_semantic_type: PromptContextSemanticType::VidEndFrame,
-      });
-    }
-
-    if let Some(ref_tokens) = &request.reference_image_media_tokens {
-      for media_token in ref_tokens {
-        context_items.push(PromptContextItem {
-          media_token: media_token.clone(),
-          context_semantic_type: PromptContextSemanticType::Imgref,
-        });
-      }
-    }
-
-    if let Some(ref_tokens) = &request.reference_video_media_tokens {
-      for media_token in ref_tokens {
-        context_items.push(PromptContextItem {
-          media_token: media_token.clone(),
-          context_semantic_type: PromptContextSemanticType::VidRef,
-        });
-      }
-    }
-
-    if !context_items.is_empty() {
-      if let Err(err) = insert_batch_prompt_context_items(InsertBatchArgs {
-        prompt_token: token.clone(),
-        items: context_items,
-        transaction: &mut transaction,
-      }).await {
-        warn!("Error inserting batch prompt context items: {:?}", err);
-      }
-    }
-  }
-
-  // -- Inference job --
-
-  let maybe_ledger_entry_token = maybe_deduction_result.as_ref()
-    .map(|d| &d.ledger_entry_token);
+  let prompt_token = insert_prompt_and_context(
+    request, &mut transaction, user_token, maybe_prompt_model_type,
+    Some(gen_result.generation_mode), &ip_address,
+  ).await;
 
   let job_token = if gen_result.is_seedance2pro {
-    // Seedance2Pro path: insert one job per order_id (batch support).
-    let fallback_ids = vec![gen_result.external_job_id.clone()];
-    let order_ids = gen_result.maybe_seedance_order_ids
-        .as_deref()
-        .unwrap_or(&fallback_ids);
-
-    let mut all_job_tokens: Vec<InferenceJobToken> = Vec::with_capacity(order_ids.len());
-
-    for (i, order_id) in order_ids.iter().enumerate() {
-      let job_token = if i == 0 {
-        apriori_job_token.clone()
-      } else {
-        InferenceJobToken::generate()
-      };
-
-      let idempotency_str = if i == 0 {
-        idempotency_token.clone()
-      } else {
-        format!("{}-batch-{}", idempotency_token, i)
-      };
-
-      let db_result = insert_generic_inference_job_for_seedance2pro_queue_with_apriori_job_token(
-        InsertGenericInferenceForSeedance2ProWithAprioriJobTokenArgs {
-          apriori_job_token: &job_token,
-          uuid_idempotency_token: &idempotency_str,
-          maybe_external_third_party_id: order_id,
-          maybe_inference_args: None,
-          maybe_prompt_token: prompt_token.as_ref(),
-          maybe_wallet_ledger_entry_token: maybe_ledger_entry_token,
-          maybe_creator_user_token: Some(user_token),
-          maybe_avt_token: maybe_avt_token.as_ref(),
-          creator_ip_address: &ip_address,
-          creator_set_visibility: Visibility::Public,
-          mysql_executor: &mut *transaction,
-          phantom: Default::default(),
-        }
-      ).await;
-
-      match db_result {
-        Ok(token) => all_job_tokens.push(token),
-        Err(err) => {
-          warn!("Error inserting seedance2pro inference job (order_id={}): {:?}", order_id, err);
-          if i == 0 {
-            return Err(AdvancedCommonWebError::from_error(err));
-          }
-        }
-      }
-    }
-
-    all_job_tokens.first().cloned().ok_or_else(|| {
-      error!("No inference job token was created");
-      AdvancedCommonWebError::server_error_with_message("No inference job token was created")
-    })?
+    insert_seedance2pro_jobs(
+      &gen_result.external_job_id,
+      gen_result.maybe_seedance_order_ids.as_deref(),
+      &billing.apriori_job_token,
+      idempotency_token,
+      prompt_token.as_ref(),
+      billing.maybe_wallet_ledger_entry_token.as_ref(),
+      user_token,
+      maybe_avt_token,
+      &ip_address,
+      &mut transaction,
+    ).await?
   } else {
-    // Fal / other providers path.
-    // TODO: Pass maybe_wallet_ledger_entry_token to fal jobs once the fal insert supports it.
-    let db_result = insert_generic_inference_job_for_fal_queue_with_apriori_job_token(
-      InsertGenericInferenceForFalWithAprioriJobTokenArgs {
-        apriori_job_token: &apriori_job_token,
-        uuid_idempotency_token: &idempotency_token,
-        maybe_external_third_party_id: &gen_result.external_job_id,
-        fal_category: FalCategory::VideoGeneration,
-        maybe_inference_args: None,
-        maybe_prompt_token: prompt_token.as_ref(),
-        maybe_creator_user_token: Some(user_token),
-        maybe_avt_token: maybe_avt_token.as_ref(),
-        creator_ip_address: &ip_address,
-        creator_set_visibility: Visibility::Public,
-        mysql_executor: &mut *transaction,
-        starting_job_status_override: None,
-        maybe_frontend_failure_category: None,
-        maybe_failure_reason: None,
-        phantom: Default::default(),
-      }
-    ).await;
+    insert_fal_job(
+      &gen_result.external_job_id,
+      &billing.apriori_job_token,
+      idempotency_token,
+      prompt_token.as_ref(),
+      user_token,
+      maybe_avt_token,
+      &ip_address,
+      &mut transaction,
+    ).await?
+  };
 
-    match db_result {
-      Ok(token) => token,
-      Err(err) => {
-        warn!("Error inserting inference job: {:?}", err);
-        return Err(AdvancedCommonWebError::from_error(err));
-      }
+  transaction.commit().await.map_err(|err| {
+    error!("Error committing transaction: {:?}", err);
+    AdvancedCommonWebError::from_error(err)
+  })?;
+
+  Ok(Json(OmniGenVideoGenerateResponse {
+    success: true,
+    inference_job_token: job_token,
+  }))
+}
+
+// ==================== PIPELINE V2 (new) ==================== //
+
+#[allow(clippy::too_many_arguments)]
+async fn execute_v2_pipeline(
+  http_request: &HttpRequest,
+  request: &OmniGenVideoCostAndGenerateRequest,
+  server_state: &ServerState,
+  mysql_connection: &mut sqlx::pool::PoolConnection<sqlx::MySql>,
+  user_token: &tokens::tokens::users::UserToken,
+  maybe_avt_token: Option<&tokens::tokens::anonymous_visitor_tracking::AnonymousVisitorTrackingToken>,
+  idempotency_token: &str,
+  maybe_prompt_model_type: Option<CommonModelType>,
+  media_file_hydration_map: Option<HashMap<MediaFileToken, Url>>,
+) -> Result<Json<OmniGenVideoGenerateResponse>, AdvancedCommonWebError> {
+  // 1. Build execution request (provider = Seedance2Pro for Kinovi execution)
+  let mut exec_builder = hydrate_to_router_request(request)?;
+  exec_builder.provider = Provider::Seedance2Pro;
+
+  let draft_or_request = exec_builder.build2().map_err(|e| {
+    warn!("Failed to build2 for v2 pipeline: {}", e);
+    AdvancedCommonWebError::from_error(e)
+  })?;
+
+  // 2. Calculate cost (swap provider to Artcraft for billing)
+  let mut cost_builder = hydrate_to_router_request(request)?;
+  cost_builder.provider = Provider::Artcraft;
+
+  let cost_estimate = cost_builder.build2()
+    .map_err(|e| {
+      warn!("Failed to build2 cost estimate for v2 pipeline: {}", e);
+      AdvancedCommonWebError::from_error(e)
+    })?
+    .estimate_cost()
+    .map_err(|e| {
+      warn!("Failed to estimate cost for v2 pipeline: {}", e);
+      AdvancedCommonWebError::from_error(e)
+    })?;
+
+  let cost = cost_estimate.cost_in_credits.unwrap_or(0);
+  info!("v2 estimated cost: {} credits", cost);
+
+  // 3. Bill wallet
+  let billing = bill_wallet(user_token, cost, mysql_connection).await?;
+
+  // 4. Build context maps for draft finalization
+  let media_file_urls_as_strings: Option<HashMap<MediaFileToken, String>> =
+    media_file_hydration_map.as_ref().map(|map| {
+      map.iter().map(|(k, v)| (k.clone(), v.to_string())).collect()
+    });
+
+  let kinovi_character_id_map: Option<HashMap<CharacterToken, String>> =
+    resolve_kinovi_character_id_map(
+      request.reference_character_tokens.as_deref(),
+      mysql_connection,
+    ).await?;
+
+  // 5. Execute: finalize draft (upload to Kinovi if needed) → send
+  let gen_response = execute_pipeline_v2(
+    draft_or_request,
+    server_state,
+    media_file_urls_as_strings.as_ref(),
+    kinovi_character_id_map.as_ref(),
+    billing.maybe_wallet_ledger_entry_token.as_ref(),
+    mysql_connection,
+  ).await?;
+
+  // 6. DB writes
+  let ip_address = get_request_ip(http_request);
+  let mut transaction = mysql_connection.begin().await.map_err(|err| {
+    error!("Error starting MySQL transaction: {:?}", err);
+    AdvancedCommonWebError::from_error(err)
+  })?;
+
+  let prompt_token = insert_prompt_and_context(
+    request, &mut transaction, user_token, maybe_prompt_model_type,
+    None, // TODO: determine generation_mode from v2 response
+    &ip_address,
+  ).await;
+
+  let job_token = match gen_response {
+    GenerateVideoResponse::Seedance2Pro(ref payload) => {
+      insert_seedance2pro_jobs(
+        &payload.order_id,
+        None, // v2 Kinovi returns single order_id per request
+        &billing.apriori_job_token,
+        idempotency_token,
+        prompt_token.as_ref(),
+        billing.maybe_wallet_ledger_entry_token.as_ref(),
+        user_token,
+        maybe_avt_token,
+        &ip_address,
+        &mut transaction,
+      ).await?
+    }
+    GenerateVideoResponse::Artcraft(ref payload) => {
+      // Artcraft recursive: job was already created by the inner omni call
+      payload.inference_job_token.clone()
+    }
+    GenerateVideoResponse::Fal(ref payload) => {
+      let external_id = payload.request_id.clone().unwrap_or_default();
+      insert_fal_job(
+        &external_id,
+        &billing.apriori_job_token,
+        idempotency_token,
+        prompt_token.as_ref(),
+        user_token,
+        maybe_avt_token,
+        &ip_address,
+        &mut transaction,
+      ).await?
+    }
+    _ => {
+      error!("Unexpected v2 response variant: {:?}", gen_response);
+      return Err(AdvancedCommonWebError::server_error_with_message("Unexpected v2 response"));
     }
   };
 
@@ -400,3 +411,220 @@ pub async fn omni_gen_video_generate_handler(
     inference_job_token: job_token,
   }))
 }
+
+// ==================== SHARED HELPERS ==================== //
+
+/// Insert prompt and prompt context items. Returns the prompt token if successful.
+async fn insert_prompt_and_context(
+  request: &OmniGenVideoCostAndGenerateRequest,
+  transaction: &mut sqlx::Transaction<'_, sqlx::MySql>,
+  user_token: &tokens::tokens::users::UserToken,
+  maybe_prompt_model_type: Option<CommonModelType>,
+  maybe_generation_mode: Option<enums::common::generation::common_generation_mode::CommonGenerationMode>,
+  ip_address: &str,
+) -> Option<tokens::tokens::prompts::PromptToken> {
+  let prompt_result = insert_prompt(InsertPromptArgs {
+    maybe_apriori_prompt_token: None,
+    prompt_type: PromptType::ArtcraftApp,
+    maybe_creator_user_token: Some(user_token),
+    maybe_model_type: maybe_prompt_model_type,
+    maybe_generation_provider: Some(GenerationProvider::Artcraft),
+    maybe_positive_prompt: request.prompt.as_deref(),
+    maybe_negative_prompt: request.negative_prompt.as_deref(),
+    maybe_other_args: None,
+    maybe_generation_mode,
+    maybe_aspect_ratio: None,
+    maybe_resolution: None,
+    maybe_batch_count: request.video_batch_count.map(|c| c as u8),
+    maybe_generate_audio: request.generate_audio,
+    maybe_duration_seconds: request.duration_seconds.map(|d| d as u32),
+    creator_ip_address: ip_address,
+    mysql_executor: &mut **transaction,
+    phantom: Default::default(),
+  }).await;
+
+  let prompt_token = match prompt_result {
+    Ok(token) => Some(token),
+    Err(err) => {
+      warn!("Error inserting prompt: {:?}", err);
+      None
+    }
+  };
+
+  if let Some(token) = prompt_token.as_ref() {
+    let mut context_items = Vec::new();
+
+    if let Some(media_token) = &request.start_frame_image_media_token {
+      context_items.push(PromptContextItem {
+        media_token: media_token.clone(),
+        context_semantic_type: PromptContextSemanticType::VidStartFrame,
+      });
+    }
+    if let Some(media_token) = &request.end_frame_image_media_token {
+      context_items.push(PromptContextItem {
+        media_token: media_token.clone(),
+        context_semantic_type: PromptContextSemanticType::VidEndFrame,
+      });
+    }
+    if let Some(ref_tokens) = &request.reference_image_media_tokens {
+      for media_token in ref_tokens {
+        context_items.push(PromptContextItem {
+          media_token: media_token.clone(),
+          context_semantic_type: PromptContextSemanticType::Imgref,
+        });
+      }
+    }
+    if let Some(ref_tokens) = &request.reference_video_media_tokens {
+      for media_token in ref_tokens {
+        context_items.push(PromptContextItem {
+          media_token: media_token.clone(),
+          context_semantic_type: PromptContextSemanticType::VidRef,
+        });
+      }
+    }
+
+    if !context_items.is_empty() {
+      if let Err(err) = insert_batch_prompt_context_items(InsertBatchArgs {
+        prompt_token: token.clone(),
+        items: context_items,
+        transaction,
+      }).await {
+        warn!("Error inserting batch prompt context items: {:?}", err);
+      }
+    }
+  }
+
+  prompt_token
+}
+
+#[allow(clippy::too_many_arguments)]
+async fn insert_seedance2pro_jobs(
+  primary_order_id: &str,
+  maybe_additional_order_ids: Option<&[String]>,
+  apriori_job_token: &InferenceJobToken,
+  idempotency_token: &str,
+  prompt_token: Option<&tokens::tokens::prompts::PromptToken>,
+  maybe_wallet_ledger_entry_token: Option<&tokens::tokens::wallet_ledger_entries::WalletLedgerEntryToken>,
+  user_token: &tokens::tokens::users::UserToken,
+  maybe_avt_token: Option<&tokens::tokens::anonymous_visitor_tracking::AnonymousVisitorTrackingToken>,
+  ip_address: &str,
+  transaction: &mut sqlx::Transaction<'_, sqlx::MySql>,
+) -> Result<InferenceJobToken, AdvancedCommonWebError> {
+  let fallback_ids = vec![primary_order_id.to_string()];
+  let order_ids = maybe_additional_order_ids.unwrap_or(&fallback_ids);
+
+  let mut all_job_tokens: Vec<InferenceJobToken> = Vec::with_capacity(order_ids.len());
+
+  for (i, order_id) in order_ids.iter().enumerate() {
+    let job_token = if i == 0 {
+      apriori_job_token.clone()
+    } else {
+      InferenceJobToken::generate()
+    };
+
+    let idempotency_str = if i == 0 {
+      idempotency_token.to_string()
+    } else {
+      format!("{}-batch-{}", idempotency_token, i)
+    };
+
+    let db_result = insert_generic_inference_job_for_seedance2pro_queue_with_apriori_job_token(
+      InsertGenericInferenceForSeedance2ProWithAprioriJobTokenArgs {
+        apriori_job_token: &job_token,
+        uuid_idempotency_token: &idempotency_str,
+        maybe_external_third_party_id: order_id,
+        maybe_inference_args: None,
+        maybe_prompt_token: prompt_token,
+        maybe_wallet_ledger_entry_token,
+        maybe_creator_user_token: Some(user_token),
+        maybe_avt_token,
+        creator_ip_address: ip_address,
+        creator_set_visibility: Visibility::Public,
+        mysql_executor: &mut **transaction,
+        phantom: Default::default(),
+      }
+    ).await;
+
+    match db_result {
+      Ok(token) => all_job_tokens.push(token),
+      Err(err) => {
+        warn!("Error inserting seedance2pro inference job (order_id={}): {:?}", order_id, err);
+        if i == 0 {
+          return Err(AdvancedCommonWebError::from_error(err));
+        }
+      }
+    }
+  }
+
+  all_job_tokens.first().cloned().ok_or_else(|| {
+    error!("No inference job token was created");
+    AdvancedCommonWebError::server_error_with_message("No inference job token was created")
+  })
+}
+
+#[allow(clippy::too_many_arguments)]
+async fn insert_fal_job(
+  external_job_id: &str,
+  apriori_job_token: &InferenceJobToken,
+  idempotency_token: &str,
+  prompt_token: Option<&tokens::tokens::prompts::PromptToken>,
+  user_token: &tokens::tokens::users::UserToken,
+  maybe_avt_token: Option<&tokens::tokens::anonymous_visitor_tracking::AnonymousVisitorTrackingToken>,
+  ip_address: &str,
+  transaction: &mut sqlx::Transaction<'_, sqlx::MySql>,
+) -> Result<InferenceJobToken, AdvancedCommonWebError> {
+  // TODO: Pass maybe_wallet_ledger_entry_token to fal jobs once the fal insert supports it.
+  let db_result = insert_generic_inference_job_for_fal_queue_with_apriori_job_token(
+    InsertGenericInferenceForFalWithAprioriJobTokenArgs {
+      apriori_job_token,
+      uuid_idempotency_token: idempotency_token,
+      maybe_external_third_party_id: external_job_id,
+      fal_category: FalCategory::VideoGeneration,
+      maybe_inference_args: None,
+      maybe_prompt_token: prompt_token,
+      maybe_creator_user_token: Some(user_token),
+      maybe_avt_token,
+      creator_ip_address: ip_address,
+      creator_set_visibility: Visibility::Public,
+      mysql_executor: &mut **transaction,
+      starting_job_status_override: None,
+      maybe_frontend_failure_category: None,
+      maybe_failure_reason: None,
+      phantom: Default::default(),
+    }
+  ).await;
+
+  match db_result {
+    Ok(token) => Ok(token),
+    Err(err) => {
+      warn!("Error inserting fal inference job: {:?}", err);
+      Err(AdvancedCommonWebError::from_error(err))
+    }
+  }
+}
+
+/// Resolve character tokens to a map of CharacterToken → Kinovi character ID.
+/// Used by pipeline v2's draft context.
+async fn resolve_kinovi_character_id_map(
+  maybe_tokens: Option<&[CharacterToken]>,
+  connection: &mut sqlx::pool::PoolConnection<sqlx::MySql>,
+) -> Result<Option<HashMap<CharacterToken, String>>, AdvancedCommonWebError> {
+  use mysql_queries::queries::characters::batch_lookup_characters_by_token_for_prompting::batch_lookup_characters_by_token_for_prompting;
+
+  let tokens = match maybe_tokens {
+    None => return Ok(None),
+    Some(tokens) if tokens.is_empty() => return Ok(None),
+    Some(tokens) => tokens,
+  };
+
+  let characters = batch_lookup_characters_by_token_for_prompting(tokens, connection).await?;
+
+  let map: HashMap<CharacterToken, String> = characters.iter()
+    .filter(|c| c.is_active)
+    .filter_map(|c| {
+      c.kinovi_character_id.as_ref().map(|kid| (c.token.clone(), kid.clone()))
+    })
+    .collect();
+
+  if map.is_empty() { Ok(None) } else { Ok(Some(map)) }
+}

crates/service/web/storyteller_web/src/http_server/endpoints/omni_gen/generate/video/pipeline_v2/execute.rs

@@ -0,0 +1,118 @@
+//! Pipeline v2 execution: build2 → (optional finalize) → send_request.
+//!
+//! For Kinovi providers, the build2 output is a Draft that needs finalization
+//! (media upload to Kinovi CDN). For Artcraft providers, build2 returns a
+//! Request directly — no finalization needed.
+
+use std::collections::HashMap;
+
+use log::{error, info, warn};
+
+use artcraft_router::client::router_client::RouterClient;
+use artcraft_router::client::router_seedance2pro_client::RouterSeedance2ProClient;
+use artcraft_router::generate::generate_video::generate_video_response::GenerateVideoResponse;
+use artcraft_router::generate::generate_video_v2::video_generation_draft_context::VideoGenerationDraftContext;
+use artcraft_router::generate::generate_video_v2::video_generation_draft_or_request::VideoGenerationDraftOrRequest;
+use artcraft_router::generate::generate_video_v2::video_generation_request::VideoGenerationRequest;
+use seedance2pro_client::creds::seedance2pro_session::Seedance2ProSession;
+use tokens::tokens::characters::CharacterToken;
+use tokens::tokens::media_files::MediaFileToken;
+use tokens::tokens::wallet_ledger_entries::WalletLedgerEntryToken;
+
+use crate::http_server::common_responses::advanced_common_web_error::AdvancedCommonWebError;
+use crate::http_server::endpoint_helpers::refund_wallet_after_api_failure::refund_wallet_after_api_failure;
+use crate::state::server_state::ServerState;
+
+/// Execute the v2 pipeline: finalize the draft (if needed), then send the request.
+///
+/// If the generation fails and is a Kinovi request with a wallet ledger entry,
+/// the wallet is refunded.
+pub async fn execute_pipeline_v2(
+  draft_or_request: VideoGenerationDraftOrRequest,
+  server_state: &ServerState,
+  media_file_urls_by_token: Option<&HashMap<MediaFileToken, String>>,
+  kinovi_character_ids: Option<&HashMap<CharacterToken, String>>,
+  maybe_wallet_ledger_entry_token: Option<&WalletLedgerEntryToken>,
+  mysql_connection: &mut sqlx::pool::PoolConnection<sqlx::MySql>,
+) -> Result<GenerateVideoResponse, AdvancedCommonWebError> {
+  // 1. Finalize draft → request (or use the request directly).
+  let (video_request, is_kinovi) = finalize_to_request(
+    draft_or_request,
+    server_state,
+    media_file_urls_by_token,
+    kinovi_character_ids,
+  ).await?;
+
+  // 2. Build the client for send_request.
+  let client = build_kinovi_client(server_state);
+
+  // 3. Send the request.
+  let result = video_request.send_request(&client).await;
+
+  // 4. On failure, refund wallet if this is a Kinovi request.
+  if let Err(ref err) = result {
+    if is_kinovi {
+      if let Some(ledger_entry_token) = maybe_wallet_ledger_entry_token {
+        warn!("Kinovi v2 generation failed, issuing refund for {}: {:?}", ledger_entry_token.as_str(), err);
+        if let Err(refund_err) = refund_wallet_after_api_failure(ledger_entry_token, mysql_connection).await {
+          error!("Failed to refund wallet after Kinovi v2 failure: {:?}", refund_err);
+        }
+      }
+    }
+  }
+
+  let response = result.map_err(|err| {
+    warn!("v2 video generation failed: {:?}", err);
+    AdvancedCommonWebError::from_error(err)
+  })?;
+
+  info!("v2 generation response: {:?}", response);
+
+  Ok(response)
+}
+
+/// Finalize the draft (uploading media to Kinovi if needed) and return the request + is_kinovi flag.
+async fn finalize_to_request(
+  draft_or_request: VideoGenerationDraftOrRequest,
+  server_state: &ServerState,
+  media_file_urls_by_token: Option<&HashMap<MediaFileToken, String>>,
+  kinovi_character_ids: Option<&HashMap<CharacterToken, String>>,
+) -> Result<(VideoGenerationRequest, bool), AdvancedCommonWebError> {
+  match draft_or_request {
+    VideoGenerationDraftOrRequest::Request(request) => {
+      let is_kinovi = matches!(
+        request,
+        VideoGenerationRequest::KinoviSeedance2p0(_) | VideoGenerationRequest::KinoviSeedance2p0Fast(_)
+      );
+      Ok((request, is_kinovi))
+    }
+    VideoGenerationDraftOrRequest::Draft(draft) => {
+      let seedance2pro_session = Seedance2ProSession::from_cookies_string(
+        server_state.seedance2pro.cookies.clone()
+      );
+      let seedance2pro_client = RouterSeedance2ProClient::new(seedance2pro_session);
+      let router_client = RouterClient::Seedance2Pro(seedance2pro_client);
+
+      let draft_context = VideoGenerationDraftContext {
+        client: Some(&router_client),
+        media_file_to_artcraft_url_map: media_file_urls_by_token,
+        character_token_to_kinovi_id_map: kinovi_character_ids,
+      };
+
+      let request = draft.finalize(draft_context).await.map_err(|err| {
+        warn!("Failed to finalize v2 draft: {:?}", err);
+        AdvancedCommonWebError::from_error(err)
+      })?;
+
+      // Drafts are always Kinovi (Artcraft never produces drafts).
+      Ok((request, true))
+    }
+  }
+}
+
+fn build_kinovi_client(server_state: &ServerState) -> RouterClient {
+  let session = Seedance2ProSession::from_cookies_string(
+    server_state.seedance2pro.cookies.clone()
+  );
+  RouterClient::Seedance2Pro(RouterSeedance2ProClient::new(session))
+}

crates/service/web/storyteller_web/src/http_server/endpoints/omni_gen/generate/video/pipeline_v2/mod.rs

@@ -0,0 +1 @@
+pub mod execute;