diff --git a/crates/api_clients/artcraft/artcraft_router/src/generate/generate_audio/providers/artcraft/request_common.rs b/crates/api_clients/artcraft/artcraft_router/src/generate/generate_audio/providers/artcraft/request_common.rs --- a/crates/api_clients/artcraft/artcraft_router/src/generate/generate_audio/providers/artcraft/request_common.rs +++ b/crates/api_clients/artcraft/artcraft_router/src/generate/generate_audio/providers/artcraft/request_common.rs @@ -23,7 +23,7 @@ pub async fn send_artcraft_omni_audio_request( .map_err(|err| ArtcraftRouterError::Provider(ProviderError::Storyteller(err)))?; Ok(GenerateAudioResponse::Artcraft(ArtcraftAudioResponsePayload { - inference_job_token: response.inference_job_token.clone(), - all_inference_job_tokens: vec![response.inference_job_token], + inference_job_token: response.inference_job_token, + all_inference_job_tokens: response.all_job_tokens, })) } diff --git a/crates/api_clients/artcraft/artcraft_router/src/generate/generate_video/providers/artcraft/request_common.rs b/crates/api_clients/artcraft/artcraft_router/src/generate/generate_video/providers/artcraft/request_common.rs --- a/crates/api_clients/artcraft/artcraft_router/src/generate/generate_video/providers/artcraft/request_common.rs +++ b/crates/api_clients/artcraft/artcraft_router/src/generate/generate_video/providers/artcraft/request_common.rs @@ -23,7 +23,7 @@ pub async fn send_artcraft_omni_video_request( .map_err(|err| ArtcraftRouterError::Provider(ProviderError::Storyteller(err)))?; Ok(GenerateVideoResponse::Artcraft(ArtcraftVideoResponsePayload { - inference_job_token: response.inference_job_token.clone(), - all_inference_job_tokens: vec![response.inference_job_token], + inference_job_token: response.inference_job_token, + all_inference_job_tokens: response.all_job_tokens, })) } diff --git a/crates/api_clients/artcraft/artcraft_router/tests/artcraft_job_tokens.rs b/crates/api_clients/artcraft/artcraft_router/tests/artcraft_job_tokens.rs new file mode 100644 --- /dev/null +++ b/crates/api_clients/artcraft/artcraft_router/tests/artcraft_job_tokens.rs @@ -0,0 +1,89 @@ +use artcraft_client::credentials::storyteller_credential_set::StorytellerCredentialSet; +use artcraft_client::utils::api_host::ApiHost; +use artcraft_router::api::router_audio_model::RouterAudioModel; +use artcraft_router::api::router_provider::RouterProvider; +use artcraft_router::api::router_video_model::RouterVideoModel; +use artcraft_router::client::router_artcraft_client::RouterArtcraftClient; +use artcraft_router::client::router_client::RouterClient; +use artcraft_router::generate::generate_audio::audio_generation_draft_or_request::AudioGenerationDraftOrRequest; +use artcraft_router::generate::generate_audio::generate_audio_request_builder::GenerateAudioRequestBuilder; +use artcraft_router::generate::generate_video::generate_video_request_builder::GenerateVideoRequestBuilder; +use artcraft_router::generate::generate_video::video_generation_draft_or_request::VideoGenerationDraftOrRequest; +use std::time::Duration; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::TcpListener; +use tokio::task::JoinHandle; + +const JOBS: [&str; 3] = ["job_primary", "job_second", "job_third"]; +const TIMEOUT: Duration = Duration::from_secs(5); + +#[tokio::test] +async fn video_preserves_batch_and_single_job_tokens() { + for jobs in [&JOBS[..], &JOBS[..1], &[JOBS[1], JOBS[0], JOBS[2]], &[]] { + let (client, server) = serve_jobs("video", jobs).await; + let builder = GenerateVideoRequestBuilder { model: RouterVideoModel::Seedance1p5Pro, provider: RouterProvider::Artcraft, prompt: Some("test video".to_string()), video_batch_count: Some(jobs.len().max(1) as u16), ..Default::default() }; + let VideoGenerationDraftOrRequest::Request(request) = builder.build2().unwrap() else { + panic!("Artcraft should create a ready request"); + }; + let response = tokio::time::timeout(TIMEOUT, request.send_request(&client)).await.unwrap().unwrap(); + let payload = response.get_artcraft_payload().unwrap(); + assert_eq!(payload.inference_job_token.as_str(), JOBS[0]); + assert_eq!(payload.all_inference_job_tokens.iter().map(|token| token.as_str()).collect::>(), jobs); + server.await.unwrap(); + } +} + +#[tokio::test] +async fn audio_preserves_batch_and_single_job_tokens() { + for jobs in [&JOBS[..], &JOBS[..1], &[JOBS[1], JOBS[0], JOBS[2]], &[]] { + let (client, server) = serve_jobs("audio", jobs).await; + let builder = GenerateAudioRequestBuilder { model: RouterAudioModel::SunoMusic, provider: RouterProvider::Artcraft, prompt: Some("test music".to_string()), ..Default::default() }; + let AudioGenerationDraftOrRequest::Request(request) = builder.build2().unwrap() else { + panic!("Artcraft should create a ready request"); + }; + let response = tokio::time::timeout(TIMEOUT, request.send_request(&client)).await.unwrap().unwrap(); + let payload = response.get_artcraft_payload().unwrap(); + assert_eq!(payload.inference_job_token.as_str(), JOBS[0]); + assert_eq!(payload.all_inference_job_tokens.iter().map(|token| token.as_str()).collect::>(), jobs); + server.await.unwrap(); + } +} + +async fn serve_jobs(modality: &'static str, jobs: &[&str]) -> (RouterClient, JoinHandle<()>) { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + let body = serde_json::json!({ + "success": true, + "inference_job_token": JOBS[0], + "all_job_tokens": jobs, + }) + .to_string(); + let server = tokio::spawn(async move { + let (mut stream, _) = tokio::time::timeout(TIMEOUT, listener.accept()).await.unwrap().unwrap(); + let mut request = Vec::new(); + let mut buffer = [0; 4096]; + loop { + let count = tokio::time::timeout(TIMEOUT, stream.read(&mut buffer)).await.unwrap().unwrap(); + assert_ne!(count, 0, "request ended before its body"); + request.extend_from_slice(&buffer[..count]); + if let Some(end) = request.windows(4).position(|bytes| bytes == b"\r\n\r\n") { + let headers = std::str::from_utf8(&request[..end]).unwrap(); + assert!(headers.starts_with(&format!("POST /v1/omni_gen/generate/{modality} HTTP/1.1"))); + let length = headers + .lines() + .find_map(|line| { + let (name, value) = line.split_once(':')?; + name.eq_ignore_ascii_case("content-length").then(|| value.trim().parse::().unwrap()) + }) + .unwrap(); + if request.len() >= end + 4 + length { + break; + } + } + } + let response = format!("HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", body.len()); + stream.write_all(response.as_bytes()).await.unwrap(); + }); + let client = RouterClient::Artcraft(RouterArtcraftClient::new(ApiHost::Localhost { port: u32::from(port) }, StorytellerCredentialSet::empty())); + (client, server) +} diff --git a/crates/desktop/artcraft/src/core/commands/generate/generate_video/artcraft/handle_artcraft_video.rs b/crates/desktop/artcraft/src/core/commands/generate/generate_video/artcraft/handle_artcraft_video.rs --- a/crates/desktop/artcraft/src/core/commands/generate/generate_video/artcraft/handle_artcraft_video.rs +++ b/crates/desktop/artcraft/src/core/commands/generate/generate_video/artcraft/handle_artcraft_video.rs @@ -14,7 +14,7 @@ pub async fn handle_video_artcraft( app: &AppHandle, app_env_configs: &AppEnvConfigs, storyteller_creds_manager: &StorytellerCredentialManager, -) -> Result { +) -> Result, GenerateError> { let creds = get_storyteller_creds_or_error(app, storyteller_creds_manager)?; diff --git a/crates/desktop/artcraft/src/core/commands/generate/generate_video/artcraft/handle_artcraft_video_via_router.rs b/crates/desktop/artcraft/src/core/commands/generate/generate_video/artcraft/handle_artcraft_video_via_router.rs --- a/crates/desktop/artcraft/src/core/commands/generate/generate_video/artcraft/handle_artcraft_video_via_router.rs +++ b/crates/desktop/artcraft/src/core/commands/generate/generate_video/artcraft/handle_artcraft_video_via_router.rs @@ -15,7 +15,7 @@ use artcraft_router::client::request_mismatch_mitigation_strategy::RequestMismat use artcraft_router::client::router_artcraft_client::RouterArtcraftClient; use artcraft_router::client::router_client::RouterClient; use artcraft_router::generate::generate_video::generate_video_request_builder::GenerateVideoRequestBuilder; -use artcraft_router::generate::generate_video::generate_video_response::GenerateVideoResponse; +use artcraft_router::generate::generate_video::generate_video_response::{ArtcraftVideoResponsePayload, GenerateVideoResponse}; use artcraft_router::generate::generate_video::video_generation_draft_or_request::VideoGenerationDraftOrRequest; use enums::common::generation_provider::GenerationProvider; use enums::tauri::tasks::task_type::TaskType; @@ -27,7 +27,7 @@ pub(super) async fn handle_artcraft_video_via_router( creds: &StorytellerCredentialSet, model: RouterVideoModel, generation_model: GenerationModel, -) -> Result { +) -> Result, GenerateError> { let client = RouterClient::Artcraft(RouterArtcraftClient::new( app_env_configs.storyteller_host.clone(), creds.clone(), @@ -67,19 +67,21 @@ pub(super) async fn handle_artcraft_video_via_router( info!("Building request for artcraft_router (v2 pipeline)..."); let response = generate_via_v2(router_request, &client).await?; - let job_id = response.get_artcraft_payload() - .map(|p| p.inference_job_token.to_string()) + let payload = response.get_artcraft_payload() .ok_or(GenerateError::ResponseHadNoJobTokens)?; - Ok(TaskEnqueueSuccess { - task_type: TaskType::VideoGeneration, - model: Some(generation_model), - provider: GenerationProvider::Artcraft, - provider_job_id: Some(job_id), - maybe_queue_status_url: None, - maybe_prompt_token: None, - maybe_queue_response_url: None, - }) + Ok(artcraft_video_tasks(payload, generation_model)) +} + +fn artcraft_video_tasks(payload: ArtcraftVideoResponsePayload, generation_model: GenerationModel) -> Vec { + let mut job_tokens = vec![payload.inference_job_token]; + for job_token in payload.all_inference_job_tokens { + if !job_tokens.contains(&job_token) { + job_tokens.push(job_token); + } + } + + job_tokens.into_iter().map(|job_token| TaskEnqueueSuccess { task_type: TaskType::VideoGeneration, model: Some(generation_model.clone()), provider: GenerationProvider::Artcraft, provider_job_id: Some(job_token.to_string()), maybe_queue_status_url: None, maybe_prompt_token: None, maybe_queue_response_url: None }).collect() } /// V2 pipeline: build2 → send_request (Artcraft skips draft phase). @@ -105,3 +107,26 @@ async fn generate_via_v2( info!("V2 successfully enqueued."); Ok(response) } + +#[cfg(test)] +mod tests { + use super::*; + use tokens::tokens::generic_inference_jobs::InferenceJobToken; + + const JOBS: [&str; 3] = ["job_primary", "job_second", "job_third"]; + + #[test] + fn video_tasks_keep_primary_first_and_track_each_job_once() { + for all_jobs in [&JOBS[..], &JOBS[..1], &[JOBS[1], JOBS[0], JOBS[2], JOBS[1]], &[]] { + let payload = ArtcraftVideoResponsePayload { inference_job_token: InferenceJobToken::new_from_str(JOBS[0]), all_inference_job_tokens: all_jobs.iter().map(|job| InferenceJobToken::new_from_str(job)).collect() }; + let tasks = artcraft_video_tasks(payload, GenerationModel::Seedance1p5Pro); + let expected = if all_jobs.len() > 1 { &JOBS[..] } else { &JOBS[..1] }; + assert_eq!(tasks.iter().map(|task| task.provider_job_id.as_deref().unwrap()).collect::>(), expected); + for task in tasks { + assert_eq!(task.provider, GenerationProvider::Artcraft); + assert_eq!(task.task_type, TaskType::VideoGeneration); + assert!(matches!(task.model, Some(GenerationModel::Seedance1p5Pro))); + } + } + } +} diff --git a/crates/desktop/artcraft/src/core/commands/generate/generate_video/generate_video_command.rs b/crates/desktop/artcraft/src/core/commands/generate/generate_video/generate_video_command.rs --- a/crates/desktop/artcraft/src/core/commands/generate/generate_video/generate_video_command.rs +++ b/crates/desktop/artcraft/src/core/commands/generate/generate_video/generate_video_command.rs @@ -180,7 +180,7 @@ async fn handle_request( app_data_root, app_env_configs, grok_creds_manager, - ).await + ).await.map(|event| vec![event]) } GenerationProvider::Sora => { handle_sora_sora2( @@ -189,7 +189,7 @@ async fn handle_request( app_data_root, app_env_configs, sora_creds_manager, - ).await + ).await.map(|event| vec![event]) } _ => { handle_video_artcraft( @@ -201,27 +201,123 @@ async fn handle_request( } }; - let success_event = match result { - Err(err) => return Err(err), - Ok(event) => event, - }; - - let result = success_event - .insert_into_task_database_with_frontend_payload( - task_database, - request.frontend_caller, - request.frontend_subscriber_id.as_deref(), - request.frontend_subscriber_payload.as_deref() - ) - .await; - - if let Err(err) = result { - error!("Failed to create task in database: {:?}", err); - } + let (success_event, count) = persist_video_tasks(result?, task_database, &request).await?; - if let Err(err) = artcraft_usage_tracker.record_video_generation(1, ArtcraftUsageType::ImageToResult, ArtcraftUsagePage::VideoPage) { + if let Err(err) = artcraft_usage_tracker.record_video_generation(count, ArtcraftUsageType::ImageToResult, ArtcraftUsagePage::VideoPage) { warn!("Failed to report usage: {:?}", err); } Ok(success_event) } + +async fn persist_video_tasks(events: Vec, task_database: &TaskDatabase, request: &TauriGenerateVideoRequest) -> Result<(TaskEnqueueSuccess, u16), GenerateError> { + let count = u16::try_from(events.len()).unwrap_or(u16::MAX); + for (index, event) in events.iter().enumerate() { + let result = event.insert_into_task_database_with_frontend_payload(task_database, request.frontend_caller, request.frontend_subscriber_id.as_deref(), request.frontend_subscriber_payload.as_deref()).await; + + if let Err(err) = result { + // Generation is already accepted; returning an error could prompt duplicate submissions. + let role = if index == 0 { "primary" } else { "secondary" }; + error!("Failed to persist accepted {} video job {:?}: {:?}", role, event.provider_job_id, err); + } + } + + events.into_iter().next().map(|event| (event, count)).ok_or(GenerateError::ResponseHadNoJobTokens) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::core::events::generation_events::common::GenerationModel; + use enums::tauri::tasks::task_status::TaskStatus; + use enums::tauri::tasks::task_type::TaskType; + use enums::tauri::ux::tauri_command_caller::TauriCommandCaller; + use serde_json::json; + use sqlite_tasks::queries::list_tasks_by_provider_and_tokens::{list_tasks_by_provider_and_tokens, ListTasksArgs}; + + const JOBS: [&str; 3] = ["job_primary", "job_second", "job_third"]; + + mod persistence { + use super::*; + + #[tokio::test] + async fn persists_batch_and_single_video_jobs_with_subscriber_metadata() { + for jobs in [&JOBS[..], &JOBS[..1]] { + let directory = tempfile::tempdir().unwrap(); + let root = AppDataRoot::create_existing(directory.path()).unwrap(); + let database = TaskDatabase::connect(&root).await.unwrap(); + let request: TauriGenerateVideoRequest = serde_json::from_value(json!({ + "frontend_caller": "image_to_video", + "frontend_subscriber_id": "video-editor", + "frontend_subscriber_payload": "scene-123", + })) + .unwrap(); + let events = jobs.iter().copied().map(video_event).collect(); + + let (event, count) = persist_video_tasks(events, &database, &request).await.unwrap(); + assert_eq!(event.provider_job_id.as_deref(), Some(JOBS[0])); + assert_eq!(usize::from(count), jobs.len()); + database.get_connection().get_pool().close().await; + + let reopened = TaskDatabase::connect(&root).await.unwrap(); + let mut tasks = list_tasks_by_provider_and_tokens(ListTasksArgs { db: reopened.get_connection(), provider: GenerationProvider::Artcraft, provider_job_ids: None }).await.unwrap().tasks; + tasks.sort_by(|left, right| left.provider_job_id.cmp(&right.provider_job_id)); + assert_eq!(tasks.iter().map(|task| task.provider_job_id.as_deref().unwrap()).collect::>(), jobs); + for task in tasks { + assert_eq!(task.status, TaskStatus::Pending); + assert_eq!(task.task_type, TaskType::VideoGeneration); + assert_eq!(task.frontend_caller, Some(TauriCommandCaller::ImageToVideo)); + assert_eq!(task.frontend_subscriber_id.as_deref(), Some("video-editor")); + assert_eq!(task.frontend_subscriber_payload.as_deref(), Some("scene-123")); + } + reopened.get_connection().get_pool().close().await; + } + } + + #[tokio::test] + async fn preserves_accepted_batch_when_primary_or_secondary_insert_fails() { + for failed_job in &JOBS[..2] { + let directory = tempfile::tempdir().unwrap(); + let root = AppDataRoot::create_existing(directory.path()).unwrap(); + let database = TaskDatabase::connect(&root).await.unwrap(); + let existing_id = video_event(failed_job).insert_into_task_database_with_frontend_payload(&database, None, Some("existing-tracker"), None).await.unwrap(); + let request = serde_json::from_value(json!({ "frontend_subscriber_id": "new-tracker" })).unwrap(); + let events = JOBS.iter().copied().map(video_event).collect(); + + let (event, count) = persist_video_tasks(events, &database, &request).await.unwrap(); + assert_eq!(event.provider_job_id.as_deref(), Some(JOBS[0])); + assert_eq!(usize::from(count), JOBS.len()); + database.get_connection().get_pool().close().await; + + let reopened = TaskDatabase::connect(&root).await.unwrap(); + let mut tasks = list_tasks_by_provider_and_tokens(ListTasksArgs { db: reopened.get_connection(), provider: GenerationProvider::Artcraft, provider_job_ids: None }).await.unwrap().tasks; + tasks.sort_by(|left, right| left.provider_job_id.cmp(&right.provider_job_id)); + assert_eq!(tasks.iter().map(|task| task.provider_job_id.as_deref().unwrap()).collect::>(), JOBS); + for task in tasks { + if task.provider_job_id.as_deref() == Some(*failed_job) { + assert_eq!(task.id, existing_id); + assert_eq!(task.frontend_subscriber_id.as_deref(), Some("existing-tracker")); + } else { + assert_eq!(task.frontend_subscriber_id.as_deref(), Some("new-tracker")); + } + } + reopened.get_connection().get_pool().close().await; + } + } + + #[tokio::test] + async fn rejects_an_empty_video_batch() { + let directory = tempfile::tempdir().unwrap(); + let root = AppDataRoot::create_existing(directory.path()).unwrap(); + let database = TaskDatabase::connect(&root).await.unwrap(); + let request = serde_json::from_value(json!({})).unwrap(); + let result = persist_video_tasks(Vec::new(), &database, &request).await; + assert!(matches!(result, Err(GenerateError::ResponseHadNoJobTokens))); + database.get_connection().get_pool().close().await; + } + } + + fn video_event(job: &str) -> TaskEnqueueSuccess { + TaskEnqueueSuccess { task_type: TaskType::VideoGeneration, model: Some(GenerationModel::Seedance1p5Pro), provider: GenerationProvider::Artcraft, provider_job_id: Some(job.to_string()), maybe_queue_status_url: None, maybe_queue_response_url: None, maybe_prompt_token: None } + } +} diff --git a/netlify.toml b/netlify.toml new file mode 100644 --- /dev/null +++ b/netlify.toml @@ -0,0 +1,5 @@ +# The legacy storyteller-docs build script and docs now live in +# storytold/artcraft-services. Skip only while that script is absent. +# Remove this guard after the Netlify site is repointed to that repository. +[build] + ignore = "test ! -f script/netlify_build_docs.sh"