From 32be0989438fb8100d48bc423821d8885fd42587 Mon Sep 17 00:00:00 2001 From: Brandon Thomas Date: Mon, 11 May 2026 23:23:26 -0400 Subject: [PATCH] upload finished results from fal (wip) --- crates/desktop/artcraft/Cargo.toml | 1 + .../lifecycle/startup/handle_tauri_startup.rs | 7 + .../handlers/fal/handle_fal_complete.rs | 273 ++++++++++++++++++ .../handlers/fal/handle_fal_failure.rs | 51 ++++ .../handlers/fal/mod.rs | 4 +- .../handlers/fal/poll_fal_tasks.rs | 183 ++++++++++++ .../third_party_task_polling_thread.rs | 53 ++-- crates/desktop/artcraft/src/lib.rs | 2 + 8 files changed, 551 insertions(+), 23 deletions(-) diff --git a/crates/desktop/artcraft/Cargo.toml b/crates/desktop/artcraft/Cargo.toml index cec696fca5..d969c47f6b 100644 --- a/crates/desktop/artcraft/Cargo.toml +++ b/crates/desktop/artcraft/Cargo.toml @@ -33,6 +33,7 @@ build_metadata.workspace = true cookie_store.workspace = true enums.workspace = true errors.workspace = true +fal_client.workspace = true filesys.workspace = true grok_client.workspace = true uuid_utils.workspace = true diff --git a/crates/desktop/artcraft/src/core/lifecycle/startup/handle_tauri_startup.rs b/crates/desktop/artcraft/src/core/lifecycle/startup/handle_tauri_startup.rs index a3d08de217..033ea2236e 100644 --- a/crates/desktop/artcraft/src/core/lifecycle/startup/handle_tauri_startup.rs +++ b/crates/desktop/artcraft/src/core/lifecycle/startup/handle_tauri_startup.rs @@ -21,6 +21,7 @@ use crate::services::sora::state::sora_task_queue::SoraTaskQueue; use crate::services::storyteller::state::storyteller_credential_manager::StorytellerCredentialManager; use crate::services::worldlabs::state::worldlabs_bearer_bridge::WorldlabsBearerBridge; use crate::services::worldlabs::state::worldlabs_credential_manager::WorldlabsCredentialManager; +use crate::core::providers::credentials::provider_credential_loading_cache::ProviderCredentialLoadingCache; use crate::core::threads::third_party_task_polling_thread::third_party_task_polling_thread::third_party_task_polling_thread; use crate::services::worldlabs::threads::worldlabs_marble_task_polling::worldlabs_marble_task_polling; use errors::AnyhowResult; @@ -40,6 +41,7 @@ pub async fn handle_tauri_startup( grok_image_prompt_queue: GrokImagePromptQueue, worldlabs_bearer_bridge: WorldlabsBearerBridge, worldlabs_creds_manager: WorldlabsCredentialManager, + credential_cache: ProviderCredentialLoadingCache, ) -> AnyhowResult<()> { set_app_log_level( @@ -118,7 +120,12 @@ pub async fn handle_tauri_startup( )); tauri::async_runtime::spawn(third_party_task_polling_thread( + app.clone(), + app_env_configs.clone(), + root.clone(), task_database.clone(), + storyteller_creds_manager.clone(), + credential_cache, )); spawn_discord_presence_thread()?; diff --git a/crates/desktop/artcraft/src/core/threads/third_party_task_polling_thread/handlers/fal/handle_fal_complete.rs b/crates/desktop/artcraft/src/core/threads/third_party_task_polling_thread/handlers/fal/handle_fal_complete.rs index e69de29bb2..a4a1388a35 100644 --- a/crates/desktop/artcraft/src/core/threads/third_party_task_polling_thread/handlers/fal/handle_fal_complete.rs +++ b/crates/desktop/artcraft/src/core/threads/third_party_task_polling_thread/handlers/fal/handle_fal_complete.rs @@ -0,0 +1,273 @@ +use crate::core::events::basic_sendable_event_trait::BasicSendableEvent; +use crate::core::events::generation_events::common::{GenerationAction, GenerationServiceProvider}; +use crate::core::events::generation_events::generation_complete_event::GenerationCompleteEvent; +use crate::core::state::app_env_configs::app_env_configs::AppEnvConfigs; +use crate::core::state::data_dir::app_data_root::AppDataRoot; +use crate::core::state::data_dir::trait_data_subdir::DataSubdir; +use crate::core::state::task_database::TaskDatabase; +use crate::services::storyteller::state::storyteller_credential_manager::StorytellerCredentialManager; +use artcraft_api_defs::prompts::create_prompt::CreatePromptRequest; +use artcraft_api_defs::utils::media_links_to_thumbnail_template::media_links_to_thumbnail_template; +use artcraft_client::credentials::storyteller_credential_set::StorytellerCredentialSet; +use artcraft_client::endpoints::media_files::get_media_file::get_media_file; +use artcraft_client::endpoints::media_files::upload_image_media_file_from_file::{ + upload_image_media_file_from_file, UploadImageFromFileArgs, +}; +use artcraft_client::endpoints::media_files::upload_video_media_file_from_file::{ + upload_video_media_file_from_file, UploadVideoFromFileArgs, +}; +use artcraft_client::endpoints::prompts::create_prompt::create_prompt; +use enums::common::generation_provider::GenerationProvider; +use enums::tauri::tasks::task_media_file_class::TaskMediaFileClass; +use enums::tauri::tasks::task_type::TaskType; +use fal_client::polling::poll_job_response::poll_job_response::PollJobResponse; +use fal_client::polling::poll_job_response::success_case_extractors::PollResponseExtractedContents; +use log::{error, info, warn}; +use sqlite_tasks::queries::task::Task; +use sqlite_tasks::queries::update_successful_task_status_with_metadata::{ + update_successful_task_status_with_metadata, UpdateSuccessfulTaskArgs, +}; +use std::fs::File; +use std::io::Write; +use std::path::PathBuf; +use tauri::AppHandle; +use tokens::tokens::media_files::MediaFileToken; +use uuid_utils::uuid::generate_random_uuid; + +pub async fn handle_fal_complete( + app_handle: &AppHandle, + app_env_configs: &AppEnvConfigs, + app_data_root: &AppDataRoot, + task_database: &TaskDatabase, + storyteller_creds_manager: &StorytellerCredentialManager, + task: &Task, + job_response: PollJobResponse, +) { + info!("[FalComplete] Handling completed task {}", task.id.as_str()); + + let result = handle_fal_complete_inner( + app_handle, + app_env_configs, + app_data_root, + task_database, + storyteller_creds_manager, + task, + job_response, + ).await; + + if let Err(err) = result { + error!("[FalComplete] Failed to handle task {}: {:?}", task.id.as_str(), err); + } +} + +async fn handle_fal_complete_inner( + app_handle: &AppHandle, + app_env_configs: &AppEnvConfigs, + app_data_root: &AppDataRoot, + task_database: &TaskDatabase, + storyteller_creds_manager: &StorytellerCredentialManager, + task: &Task, + job_response: PollJobResponse, +) -> Result<(), Box> { + let creds = storyteller_creds_manager.get_credentials()? + .ok_or("No Storyteller credentials available")?; + + let extracted = job_response.extracted_contents; + + // Determine what kind of media we got and collect download URLs. + let (urls, media_class, generation_action) = collect_media_urls(task, &extracted)?; + + if urls.is_empty() { + warn!("[FalComplete] Task {} completed but no downloadable media found in response", task.id.as_str()); + //warn!("[FalComplete] Raw body: {}", &job_response.raw_body[..job_response.raw_body.len().min(500)]); + return Ok(()); + } + + // Create a prompt record + let prompt_response = create_prompt( + &app_env_configs.storyteller_host, + Some(&creds), + CreatePromptRequest { + uuid_idempotency_token: generate_random_uuid(), + positive_prompt: None, + negative_prompt: None, + model_type: None, + generation_provider: Some(GenerationProvider::Fal), + maybe_generation_mode: None, + maybe_aspect_ratio: None, + maybe_resolution: None, + maybe_batch_count: None, + maybe_generate_audio: None, + maybe_duration_seconds: None, + }, + ).await?; + + info!("[FalComplete] Created prompt: {:?}", prompt_response.prompt_token); + + let mut maybe_primary_media_file_token: Option = None; + + for (i, url) in urls.iter().enumerate() { + info!("[FalComplete] Downloading result {} from: {}", i, url); + + let download_path = download_file(url, app_data_root, i).await?; + + info!("[FalComplete] Uploading result {} to backend...", i); + + let media_token = upload_to_backend( + &creds, + app_env_configs, + &download_path, + &prompt_response.prompt_token, + media_class, + ).await?; + + info!("[FalComplete] Uploaded as media file: {:?}", media_token); + + if maybe_primary_media_file_token.is_none() { + maybe_primary_media_file_token = Some(media_token); + } + } + + // Look up CDN/thumbnail URLs for the primary media file + let mut maybe_cdn_url = None; + let mut maybe_thumbnail_url_template = None; + + if let Some(media_file_token) = maybe_primary_media_file_token.as_ref() { + match get_media_file(&app_env_configs.storyteller_host, media_file_token).await { + Ok(response) => { + maybe_cdn_url = Some(response.media_file.media_links.cdn_url.to_string()); + maybe_thumbnail_url_template = media_links_to_thumbnail_template(&response.media_file.media_links) + .map(|s| s.to_string()); + } + Err(err) => { + error!("[FalComplete] Failed to look up media file after upload: {:?} (failing open)", err); + } + } + } + + // Mark the task as completed + let updated = update_successful_task_status_with_metadata(UpdateSuccessfulTaskArgs { + db: task_database.get_connection(), + task_id: &task.id, + maybe_batch_token: None, + maybe_primary_media_file_token: maybe_primary_media_file_token.as_ref(), + maybe_primary_media_file_class: Some(media_class), + maybe_primary_media_file_cdn_url: maybe_cdn_url.as_deref(), + maybe_primary_media_file_thumbnail_url_template: maybe_thumbnail_url_template.as_deref(), + }).await?; + + if updated { + let event = GenerationCompleteEvent { + action: Some(generation_action), + service: GenerationServiceProvider::Fal, + model: None, + }; + event.send_infallible(app_handle); + } + + info!("[FalComplete] Task {} fully handled", task.id.as_str()); + + Ok(()) +} + +// ── Helpers ── + +/// Collect downloadable media URLs from the extracted response contents. +fn collect_media_urls( + task: &Task, + extracted: &Option, +) -> Result<(Vec, TaskMediaFileClass, GenerationAction), Box> { + let extracted = match extracted { + Some(e) => e, + None => return Ok((vec![], TaskMediaFileClass::Image, GenerationAction::GenerateImage)), + }; + + // Images (batch) + if let Some(images) = &extracted.images { + let urls: Vec = images.iter() + .filter_map(|img| img.url.clone()) + .collect(); + if !urls.is_empty() { + return Ok((urls, TaskMediaFileClass::Image, GenerationAction::GenerateImage)); + } + } + + // Single image (e.g. background removal) + if let Some(image) = &extracted.image { + if let Some(url) = &image.url { + return Ok((vec![url.clone()], TaskMediaFileClass::Image, GenerationAction::GenerateImage)); + } + } + + // Video + if let Some(video) = &extracted.video { + if let Some(url) = &video.url { + return Ok((vec![url.clone()], TaskMediaFileClass::Video, GenerationAction::GenerateVideo)); + } + } + + // 3D model (GLB) + if let Some(glb) = &extracted.model_glb { + if let Some(url) = &glb.url { + // TODO: We may want a TaskMediaFileClass::Mesh variant + return Ok((vec![url.clone()], TaskMediaFileClass::Image, GenerationAction::ImageTo3d)); + } + } + + Ok((vec![], TaskMediaFileClass::Image, GenerationAction::GenerateImage)) +} + +async fn download_file( + url: &str, + app_data_root: &AppDataRoot, + index: usize, +) -> Result> { + let response = reqwest::get(url).await?; + let bytes = response.bytes().await?; + + let extension = url_utils::download_extension::extract_download_extension_from_url::extract_download_extension_from_url_str(url) + .map(|ext| ext.as_extension_without_period()) + .unwrap_or("bin"); + + let tempdir = app_data_root.temp_dir().path(); + let filename = format!("fal_{}_{}.{}", generate_random_uuid(), index, extension); + let download_path = tempdir.join(filename); + + let mut file = File::create(&download_path)?; + file.write_all(&bytes)?; + + Ok(download_path) +} + +async fn upload_to_backend( + creds: &StorytellerCredentialSet, + app_env_configs: &AppEnvConfigs, + download_path: &PathBuf, + prompt_token: &tokens::tokens::prompts::PromptToken, + media_class: TaskMediaFileClass, +) -> Result> { + let media_token = match media_class { + TaskMediaFileClass::Video => { + let result = upload_video_media_file_from_file(UploadVideoFromFileArgs { + api_host: &app_env_configs.storyteller_host, + maybe_creds: Some(creds), + path: download_path, + maybe_prompt_token: Some(prompt_token), + }).await?; + result.media_file_token + } + _ => { + let result = upload_image_media_file_from_file(UploadImageFromFileArgs { + api_host: &app_env_configs.storyteller_host, + maybe_creds: Some(creds), + path: download_path, + is_intermediate_system_file: false, + maybe_prompt_token: Some(prompt_token), + maybe_batch_token: None, + }).await?; + result.media_file_token + } + }; + + Ok(media_token) +} diff --git a/crates/desktop/artcraft/src/core/threads/third_party_task_polling_thread/handlers/fal/handle_fal_failure.rs b/crates/desktop/artcraft/src/core/threads/third_party_task_polling_thread/handlers/fal/handle_fal_failure.rs index e69de29bb2..2f754eb9eb 100644 --- a/crates/desktop/artcraft/src/core/threads/third_party_task_polling_thread/handlers/fal/handle_fal_failure.rs +++ b/crates/desktop/artcraft/src/core/threads/third_party_task_polling_thread/handlers/fal/handle_fal_failure.rs @@ -0,0 +1,51 @@ +use crate::core::events::basic_sendable_event_trait::BasicSendableEvent; +use crate::core::events::generation_events::common::{GenerationAction, GenerationServiceProvider}; +use crate::core::events::generation_events::generation_failed_event::GenerationFailedEvent; +use crate::core::state::task_database::TaskDatabase; +use enums::tauri::tasks::task_status::TaskStatus; +use enums::tauri::tasks::task_type::TaskType; +use log::{error, info}; +use sqlite_tasks::queries::task::Task; +use sqlite_tasks::queries::update_task_status::{update_task_status, UpdateTaskArgs}; +use tauri::AppHandle; + +pub async fn handle_fal_failure( + app_handle: &AppHandle, + task_database: &TaskDatabase, + task: &Task, + reason: &str, +) { + info!("[FalPolling] Marking task {} as failed: {}", task.id.as_str(), reason); + + let update_result = update_task_status(UpdateTaskArgs { + db: task_database.get_connection(), + task_id: &task.id, + status: TaskStatus::CompleteFailure, + }).await; + + if let Err(err) = update_result { + error!("[FalPolling] Failed to update task status for {}: {:?}", task.id.as_str(), err); + } + + let action = task_type_to_generation_action(task.task_type); + + let event = GenerationFailedEvent { + action, + service: GenerationServiceProvider::Fal, + model: None, + reason: Some(reason.to_string()), + }; + + event.send_infallible(app_handle); +} + +fn task_type_to_generation_action(task_type: TaskType) -> GenerationAction { + match task_type { + TaskType::ImageGeneration => GenerationAction::GenerateImage, + TaskType::VideoGeneration => GenerationAction::GenerateVideo, + TaskType::BackgroundRemoval => GenerationAction::RemoveBackground, + TaskType::ObjectGeneration => GenerationAction::ImageTo3d, + TaskType::GaussianGeneration => GenerationAction::GenerateGaussian, + TaskType::ImageInpaintEdit => GenerationAction::ImageInpaintEdit, + } +} diff --git a/crates/desktop/artcraft/src/core/threads/third_party_task_polling_thread/handlers/fal/mod.rs b/crates/desktop/artcraft/src/core/threads/third_party_task_polling_thread/handlers/fal/mod.rs index 98d4beb39d..95e4589169 100644 --- a/crates/desktop/artcraft/src/core/threads/third_party_task_polling_thread/handlers/fal/mod.rs +++ b/crates/desktop/artcraft/src/core/threads/third_party_task_polling_thread/handlers/fal/mod.rs @@ -1,3 +1,3 @@ -mod handle_fal_complete; -mod handle_fal_failure; +pub mod handle_fal_complete; +pub mod handle_fal_failure; pub mod poll_fal_tasks; diff --git a/crates/desktop/artcraft/src/core/threads/third_party_task_polling_thread/handlers/fal/poll_fal_tasks.rs b/crates/desktop/artcraft/src/core/threads/third_party_task_polling_thread/handlers/fal/poll_fal_tasks.rs index e69de29bb2..81f9f585ac 100644 --- a/crates/desktop/artcraft/src/core/threads/third_party_task_polling_thread/handlers/fal/poll_fal_tasks.rs +++ b/crates/desktop/artcraft/src/core/threads/third_party_task_polling_thread/handlers/fal/poll_fal_tasks.rs @@ -0,0 +1,183 @@ +use crate::core::providers::credentials::provider_credential_loading_cache::ProviderCredentialLoadingCache; +use crate::core::providers::credentials::payload::provider_credential_payload::ProviderCredentialPayload; +use crate::core::providers::credentials::provider_credential_key::ProviderCredentialKey; +use crate::core::state::app_env_configs::app_env_configs::AppEnvConfigs; +use crate::core::state::data_dir::app_data_root::AppDataRoot; +use crate::core::state::task_database::TaskDatabase; +use crate::core::threads::third_party_task_polling_thread::handlers::fal::handle_fal_complete::handle_fal_complete; +use crate::core::threads::third_party_task_polling_thread::handlers::fal::handle_fal_failure::handle_fal_failure; +use crate::services::storyteller::state::storyteller_credential_manager::StorytellerCredentialManager; +use enums::tauri::tasks::task_status::TaskStatus; +use fal_client::creds::fal_api_key::FalApiKey; +use fal_client::polling::poll_job_response::poll_job_response::{poll_job_response, PollJobResponseArgs}; +use fal_client::polling::poll_job_status::poll_job_status::{poll_job_status, FalJobStatus, PollJobStatusArgs}; +use log::{error, info, warn}; +use sqlite_tasks::queries::task::Task; +use sqlite_tasks::queries::update_task_status::{update_task_status, UpdateTaskArgs}; +use tauri::AppHandle; + +pub async fn poll_fal_tasks( + app_handle: &AppHandle, + app_env_configs: &AppEnvConfigs, + app_data_root: &AppDataRoot, + task_database: &TaskDatabase, + storyteller_creds_manager: &StorytellerCredentialManager, + credential_cache: &ProviderCredentialLoadingCache, + fal_tasks: &[&Task], +) { + let api_key = match load_fal_api_key(credential_cache) { + Some(key) => key, + None => { + warn!("[FalPolling] No FAL API key configured, skipping poll"); + return; + } + }; + + for task in fal_tasks { + let result = poll_single_fal_task( + app_handle, + app_env_configs, + app_data_root, + task_database, + storyteller_creds_manager, + &api_key, + task, + ).await; + + if let Err(err) = result { + error!( + "[FalPolling] Error processing task {}: {:?}", + task.id.as_str(), + err, + ); + } + } +} + +async fn poll_single_fal_task( + app_handle: &AppHandle, + app_env_configs: &AppEnvConfigs, + app_data_root: &AppDataRoot, + task_database: &TaskDatabase, + storyteller_creds_manager: &StorytellerCredentialManager, + api_key: &FalApiKey, + task: &Task, +) -> Result<(), Box> { + let status_url = match &task.queue_status_url { + Some(url) => url.as_str(), + None => { + warn!("[FalPolling] Task {} has no queue_status_url, marking as failed", task.id.as_str()); + handle_fal_failure( + app_handle, + task_database, + task, + "No queue status URL to check", + ).await; + return Ok(()); + } + }; + + // Step 1: Check job status + info!("[FalPolling] Checking status for task {}: {}", task.id.as_str(), status_url); + + let status_response = poll_job_status(PollJobStatusArgs { + status_url, + api_key, + }).await?; + + match status_response.status { + FalJobStatus::InQueue | FalJobStatus::InProgress => { + info!( + "[FalPolling] Task {} still {:?} (queue_position={:?})", + task.id.as_str(), + status_response.status, + status_response.maybe_queue_position, + ); + return Ok(()); + } + FalJobStatus::Unknown(ref status) => { + warn!("[FalPolling] Task {} has unknown status: {}", task.id.as_str(), status); + return Ok(()); + } + FalJobStatus::Completed => { + info!("[FalPolling] Task {} completed, fetching results...", task.id.as_str()); + } + } + + // Step 2: Fetch the response + let response_url = pick_response_url(task, &status_response.response_url); + + let response_url = match response_url { + Some(url) => url, + None => { + warn!("[FalPolling] Task {} completed but has no response URL, marking as failed", task.id.as_str()); + handle_fal_failure( + app_handle, + task_database, + task, + "Job completed but no response URL available", + ).await; + return Ok(()); + } + }; + + let job_response = match poll_job_response(PollJobResponseArgs { + response_url: &response_url, + api_key, + }).await { + Ok(response) => response, + Err(err) => { + error!( + "[FalPolling] Failed to fetch response for task {}: {}. Skipping for now.", + task.id.as_str(), + err, + ); + return Ok(()); + } + }; + + // Step 3: Handle completed job + handle_fal_complete( + app_handle, + app_env_configs, + app_data_root, + task_database, + storyteller_creds_manager, + task, + job_response, + ).await; + + Ok(()) +} + +// ── Helpers ── + +fn load_fal_api_key(credential_cache: &ProviderCredentialLoadingCache) -> Option { + match credential_cache.get_credentials(ProviderCredentialKey::FalApiKey) { + Ok(Some(ProviderCredentialPayload::ApiKey(data))) => { + Some(FalApiKey::from_str(data.as_str())) + } + _ => None, + } +} + +/// Use the task's stored response URL first; fall back to the one from the status poll. +/// Log if they differ. +fn pick_response_url(task: &Task, polled_response_url: &Option) -> Option { + match (&task.queue_response_url, polled_response_url) { + (Some(db_url), Some(polled_url)) => { + if db_url != polled_url { + warn!( + "[FalPolling] Task {} response URLs differ: db={}, polled={}. Using db URL.", + task.id.as_str(), + db_url, + polled_url, + ); + } + Some(db_url.clone()) + } + (Some(db_url), None) => Some(db_url.clone()), + (None, Some(polled_url)) => Some(polled_url.clone()), + (None, None) => None, + } +} diff --git a/crates/desktop/artcraft/src/core/threads/third_party_task_polling_thread/third_party_task_polling_thread.rs b/crates/desktop/artcraft/src/core/threads/third_party_task_polling_thread/third_party_task_polling_thread.rs index 63792fb291..552633c4cd 100644 --- a/crates/desktop/artcraft/src/core/threads/third_party_task_polling_thread/third_party_task_polling_thread.rs +++ b/crates/desktop/artcraft/src/core/threads/third_party_task_polling_thread/third_party_task_polling_thread.rs @@ -1,5 +1,10 @@ +use crate::core::providers::credentials::provider_credential_loading_cache::ProviderCredentialLoadingCache; +use crate::core::state::app_env_configs::app_env_configs::AppEnvConfigs; +use crate::core::state::data_dir::app_data_root::AppDataRoot; use crate::core::state::task_database::TaskDatabase; +use crate::core::threads::third_party_task_polling_thread::handlers::fal::poll_fal_tasks::poll_fal_tasks; use crate::core::utils::task_database_pending_statuses::TASK_DATABASE_PENDING_STATUSES; +use crate::services::storyteller::state::storyteller_credential_manager::StorytellerCredentialManager; use enums::common::generation_provider::GenerationProvider; use log::{error, info, warn}; use sqlite_tasks::queries::list_non_artcraft_pending_tasks::{ @@ -7,24 +12,31 @@ use sqlite_tasks::queries::list_non_artcraft_pending_tasks::{ }; use sqlite_tasks::queries::task::Task; use std::time::Duration; +use tauri::AppHandle; const SLEEP_NO_THIRD_PARTY_JOBS_SEEN: Duration = Duration::from_secs(10); const SLEEP_THIRD_PARTY_JOBS_SEEN: Duration = Duration::from_secs(2); const SLEEP_BETWEEN_FAL_POLLS: Duration = Duration::from_secs(1); const SLEEP_ON_ERROR: Duration = Duration::from_secs(30); -/// Main loop that polls for third-party (non-Artcraft) tasks and processes them. -/// -/// This thread runs for the lifetime of the application. On error, it pauses -/// and gracefully continues. pub async fn third_party_task_polling_thread( + app_handle: AppHandle, + app_env_configs: AppEnvConfigs, + app_data_root: AppDataRoot, task_database: TaskDatabase, + storyteller_creds_manager: StorytellerCredentialManager, + credential_cache: ProviderCredentialLoadingCache, ) -> ! { let mut has_ever_seen_third_party_jobs = false; loop { let result = poll_iteration( + &app_handle, + &app_env_configs, + &app_data_root, &task_database, + &storyteller_creds_manager, + &credential_cache, &mut has_ever_seen_third_party_jobs, ).await; @@ -36,7 +48,12 @@ pub async fn third_party_task_polling_thread( } async fn poll_iteration( + app_handle: &AppHandle, + app_env_configs: &AppEnvConfigs, + app_data_root: &AppDataRoot, task_database: &TaskDatabase, + storyteller_creds_manager: &StorytellerCredentialManager, + credential_cache: &ProviderCredentialLoadingCache, has_ever_seen_third_party_jobs: &mut bool, ) -> Result<(), PollError> { let task_list = list_non_artcraft_pending_tasks(ListNonArtcraftPendingTasksArgs { @@ -58,7 +75,6 @@ async fn poll_iteration( *has_ever_seen_third_party_jobs = true; - // Partition tasks by provider. let fal_tasks: Vec<&Task> = tasks.iter() .filter(|t| t.provider == GenerationProvider::Fal) .collect(); @@ -67,7 +83,6 @@ async fn poll_iteration( .filter(|t| t.provider != GenerationProvider::Fal) .collect(); - // Shed non-FAL jobs for now — we'll add handlers for other providers later. if !non_fal_tasks.is_empty() { for task in &non_fal_tasks { warn!( @@ -80,25 +95,21 @@ async fn poll_iteration( } if fal_tasks.is_empty() { - let sleep_duration = SLEEP_THIRD_PARTY_JOBS_SEEN; - tokio::time::sleep(sleep_duration).await; + tokio::time::sleep(SLEEP_THIRD_PARTY_JOBS_SEEN).await; return Ok(()); } - info!( - "[ThirdPartyPolling] {} FAL job(s) ready to check", - fal_tasks.len(), - ); + info!("[ThirdPartyPolling] {} FAL job(s) ready to check", fal_tasks.len()); - for task in &fal_tasks { - info!( - "[ThirdPartyPolling] task_id={}, provider_job_id={:?}, type={:?}, status_url={:?}", - task.id.as_str(), - task.provider_job_id, - task.task_type, - task.queue_status_url, - ); - } + poll_fal_tasks( + app_handle, + app_env_configs, + app_data_root, + task_database, + storyteller_creds_manager, + credential_cache, + &fal_tasks, + ).await; tokio::time::sleep(SLEEP_BETWEEN_FAL_POLLS).await; diff --git a/crates/desktop/artcraft/src/lib.rs b/crates/desktop/artcraft/src/lib.rs index 47e5c88445..c602406bf1 100644 --- a/crates/desktop/artcraft/src/lib.rs +++ b/crates/desktop/artcraft/src/lib.rs @@ -97,6 +97,7 @@ pub fn run() { let app_preferences = load_app_preferences_or_default(&app_data_root); let provider_credential_cache = crate::core::providers::credentials::provider_credential_loading_cache::ProviderCredentialLoadingCache::new(app_data_root.clone()); + let provider_credential_cache_2 = provider_credential_cache.clone(); // NB: tauri-plugin-http stores the credentials on disk, so we can defer to that for now. // println!("Attempting to read existing artcraft credentials..."); @@ -179,6 +180,7 @@ pub fn run() { grok_prompt_queue_2, worldlabs_bearer_bridge_2, worldlabs_creds_manager_2, + provider_credential_cache_2, ).await; if let Err(err) = result {