collective
Browse / github.com/storytold/artcraft / patch
fix(router): preserve ArtCraft batch job tracking
fix(router): preserve ArtCraft batch job tracking The ArtCraft video and audio router adapters currently discard secondary job IDs even when the API returns a batch. The legacy desktop video flow also persists only the primary job. Preserve the complete API token arrays and create a tracking task for every distinct returned video job, carrying the original frontend subscriber metadata. Keep the primary job first, including reordered/empty list fallback consistent with the modern Omni path. The public Tauri response and enqueue-notification shapes remain unchanged. Local HTTP regressions exercise the public video/audio send_request APIs; both
- Status
- open
- Score
- 0.00
- Adopters
- 0 other operators
- Author
- euisuh (imported from pull request #2275, open)
- Base
d4e4ac69bbd93d237eaf5d0260bc6a5ff3a5e373- Size
- +253 −38 in 7 files
- Review
- pending
- Shared
- 2026-10-28
- Id
340a70a7623d67a949fc9a43
How it went
No other operator has reported using it yet.
Diff
raw · crates/api_clients/artcraft/artcraft_router/src/generate/generate_audio/providers/artcraft/request_common.rs crates/api_clients/artcraft/artcraft_router/src/generate/generate_video/providers/artcraft/request_common.rs crates/api_clients/artcraft/artcraft_router/tests/artcraft_job_tokens.rs crates/desktop/artcraft/src/core/commands/generate/generate_video/artcraft/handle_artcraft_video.rs crates/desktop/artcraft/src/core/commands/generate/generate_video/artcraft/handle_artcraft_video_via_router.rs crates/desktop/artcraft/src/core/commands/generate/generate_video/generate_video_command.rs netlify.toml
@@ -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, })) }@@ -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, })) }new file mode 100644@@ -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::<Vec<_>>(), 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::<Vec<_>>(), 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::<usize>().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)+}@@ -14,7 +14,7 @@ pub async fn handle_video_artcraft( app: &AppHandle, app_env_configs: &AppEnvConfigs, storyteller_creds_manager: &StorytellerCredentialManager,-) -> Result<TaskEnqueueSuccess, GenerateError> {+) -> Result<Vec<TaskEnqueueSuccess>, GenerateError> { let creds = get_storyteller_creds_or_error(app, storyteller_creds_manager)?; @@ -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<TaskEnqueueSuccess, GenerateError> {+) -> Result<Vec<TaskEnqueueSuccess>, 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<TaskEnqueueSuccess> {+ 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::<Vec<_>>(), 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)));+ }+ }+ }+}@@ -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<TaskEnqueueSuccess>, 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::<Vec<_>>(), 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::<Vec<_>>(), 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 }+ }+}new file mode 100644@@ -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"