minimax worker endpoints (#1855)

2026-10-20 17:02:53 -04:00 · 2026-08-07 01:24:20 -04:00
parent bb10c993a4
commit a7a32a3862
44 changed files with 1824 additions and 35 deletions
@@ -0,0 +1,12 @@
+{
+  "db_name": "MySQL",
+  "query": "\nUPDATE generic_inference_jobs\nSET\n  status = ?,\n  on_success_result_entity_type = ?,\n  on_success_result_entity_token = ?,\n  failure_reason = NULL,\n  internal_debugging_failure_reason = NULL,\n  success_execution_millis = ?,\n  success_inference_execution_millis = ?,\n  retry_at = NULL,\n  successfully_completed_at = NOW()\nWHERE token = ?\nAND status NOT IN ('complete_success', 'complete_failure')\n    ",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Right": 6
+    },
+    "nullable": []
+  },
+  "hash": "03919ba8286d18980ecaa0038df76f43f5388db0d3838181fa8f15f4cbb932af"
+}
@@ -0,0 +1,12 @@
+{
+  "db_name": "MySQL",
+  "query": "\nUPDATE generic_inference_jobs\nSET\n  status = ?,\n  assigned_worker = ?,\n  last_assigned_worker = ?,\n  assigned_cluster = ?,\n  first_started_at = NOW()\nWHERE token = ?\n    ",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Right": 5
+    },
+    "nullable": []
+  },
+  "hash": "258ac40d3fab39f110e1e82d50e73542423bfe9a21f8127e6cd83214c1726e25"
+}
@@ -0,0 +1,36 @@
+{
+  "db_name": "MySQL",
+  "query": "\nSELECT\n  token as `job_token: tokens::tokens::generic_inference_jobs::InferenceJobToken`,\n  maybe_prompt_token as `maybe_prompt_token: tokens::tokens::prompts::PromptToken`\nFROM generic_inference_jobs\nWHERE job_type = ?\nAND status = 'pending'\nORDER BY id ASC\nLIMIT 1\nFOR UPDATE SKIP LOCKED\n    ",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "job_token: tokens::tokens::generic_inference_jobs::InferenceJobToken",
+        "type_info": {
+          "type": "VarString",
+          "flags": "NOT_NULL | UNIQUE_KEY | BINARY | NO_DEFAULT_VALUE",
+          "char_set": 224,
+          "max_size": 128
+        }
+      },
+      {
+        "ordinal": 1,
+        "name": "maybe_prompt_token: tokens::tokens::prompts::PromptToken",
+        "type_info": {
+          "type": "VarString",
+          "flags": "BINARY",
+          "char_set": 224,
+          "max_size": 128
+        }
+      }
+    ],
+    "parameters": {
+      "Right": 1
+    },
+    "nullable": [
+      false,
+      true
+    ]
+  },
+  "hash": "6ca0fcdf495624cf347f612586868f6dea93f52d9f942a92dcd2b3e6aba7c63c"
+}
@@ -0,0 +1,12 @@
+{
+  "db_name": "MySQL",
+  "query": "\nUPDATE generic_inference_jobs\nSET\n  status = ?,\n  frontend_failure_category = ?,\n  failure_reason = ?,\n  internal_debugging_failure_reason = ?,\n  success_execution_millis = ?,\n  success_inference_execution_millis = ?\nWHERE token = ?\nAND status NOT IN ('complete_success', 'complete_failure')\n    ",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Right": 7
+    },
+    "nullable": []
+  },
+  "hash": "96599b4854248d84188b092db84edb50dc5402bb7771fd8ea96a96ad43e732c0"
+}
@@ -0,0 +1,124 @@
+{
+  "db_name": "MySQL",
+  "query": "\nSELECT\n  token as `job_token: tokens::tokens::generic_inference_jobs::InferenceJobToken`,\n  status as `status: enums::common::job_status_plus::JobStatusPlus`,\n  job_type as `job_type: enums::by_table::generic_inference_jobs::inference_job_type::InferenceJobType`,\n  maybe_creator_user_token as `maybe_creator_user_token: tokens::tokens::users::UserToken`,\n  maybe_creator_anonymous_visitor_token as `maybe_creator_anonymous_visitor_token: tokens::tokens::anonymous_visitor_tracking::AnonymousVisitorTrackingToken`,\n  creator_ip_address,\n  maybe_prompt_token as `maybe_prompt_token: tokens::tokens::prompts::PromptToken`,\n  maybe_debug_log_event_token as `maybe_debug_log_event_token: tokens::tokens::non_unique::debug_logs_event_token::DebugLogEventToken`,\n  platform_type as `maybe_platform_type: enums::common::platform_type::PlatformType`,\n  on_success_result_entity_token as `maybe_result_media_file_token: tokens::tokens::media_files::MediaFileToken`\nFROM generic_inference_jobs\nWHERE token = ?\nAND job_type IN ('artcraft_minimax_h3_turbo', 'artcraft_minimax_h3_ultra')\n    ",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "job_token: tokens::tokens::generic_inference_jobs::InferenceJobToken",
+        "type_info": {
+          "type": "VarString",
+          "flags": "NOT_NULL | UNIQUE_KEY | BINARY | NO_DEFAULT_VALUE",
+          "char_set": 224,
+          "max_size": 128
+        }
+      },
+      {
+        "ordinal": 1,
+        "name": "status: enums::common::job_status_plus::JobStatusPlus",
+        "type_info": {
+          "type": "String",
+          "flags": "NOT_NULL | MULTIPLE_KEY | BINARY | ENUM",
+          "char_set": 224,
+          "max_size": 76
+        }
+      },
+      {
+        "ordinal": 2,
+        "name": "job_type: enums::by_table::generic_inference_jobs::inference_job_type::InferenceJobType",
+        "type_info": {
+          "type": "VarString",
+          "flags": "MULTIPLE_KEY | BINARY",
+          "char_set": 224,
+          "max_size": 128
+        }
+      },
+      {
+        "ordinal": 3,
+        "name": "maybe_creator_user_token: tokens::tokens::users::UserToken",
+        "type_info": {
+          "type": "VarString",
+          "flags": "MULTIPLE_KEY | BINARY",
+          "char_set": 224,
+          "max_size": 128
+        }
+      },
+      {
+        "ordinal": 4,
+        "name": "maybe_creator_anonymous_visitor_token: tokens::tokens::anonymous_visitor_tracking::AnonymousVisitorTrackingToken",
+        "type_info": {
+          "type": "VarString",
+          "flags": "MULTIPLE_KEY | BINARY",
+          "char_set": 224,
+          "max_size": 128
+        }
+      },
+      {
+        "ordinal": 5,
+        "name": "creator_ip_address",
+        "type_info": {
+          "type": "VarString",
+          "flags": "NOT_NULL | MULTIPLE_KEY | BINARY | NO_DEFAULT_VALUE",
+          "char_set": 224,
+          "max_size": 160
+        }
+      },
+      {
+        "ordinal": 6,
+        "name": "maybe_prompt_token: tokens::tokens::prompts::PromptToken",
+        "type_info": {
+          "type": "VarString",
+          "flags": "BINARY",
+          "char_set": 224,
+          "max_size": 128
+        }
+      },
+      {
+        "ordinal": 7,
+        "name": "maybe_debug_log_event_token: tokens::tokens::non_unique::debug_logs_event_token::DebugLogEventToken",
+        "type_info": {
+          "type": "VarString",
+          "flags": "BINARY",
+          "char_set": 224,
+          "max_size": 128
+        }
+      },
+      {
+        "ordinal": 8,
+        "name": "maybe_platform_type: enums::common::platform_type::PlatformType",
+        "type_info": {
+          "type": "VarString",
+          "flags": "BINARY",
+          "char_set": 224,
+          "max_size": 64
+        }
+      },
+      {
+        "ordinal": 9,
+        "name": "maybe_result_media_file_token: tokens::tokens::media_files::MediaFileToken",
+        "type_info": {
+          "type": "VarString",
+          "flags": "MULTIPLE_KEY | BINARY",
+          "char_set": 224,
+          "max_size": 128
+        }
+      }
+    ],
+    "parameters": {
+      "Right": 1
+    },
+    "nullable": [
+      false,
+      false,
+      true,
+      true,
+      true,
+      false,
+      true,
+      true,
+      true,
+      true
+    ]
+  },
+  "hash": "f1d41ad17cb01f5aef0ce2eea042ccb3e5e649365a0befaf0f46492b75523442"
+}
@@ -0,0 +1,29 @@
+use serde_derive::{Deserialize, Serialize};
+use utoipa::ToSchema;
+
+use enums::by_table::generic_inference_jobs::frontend_failure_category::FrontendFailureCategory;
+
+/// Request body for `POST /v1/internal/minimax_jobs/job/{job_token}/failure`.
+#[derive(Clone, Debug, Serialize, Deserialize, ToSchema)]
+pub struct MarkMinimaxJobFailureRequest {
+  /// Optional failure class the frontend can localize and present.
+  pub maybe_frontend_failure_category: Option<FrontendFailureCategory>,
+
+  /// Optional failure reason we can show the user (512 characters max).
+  pub maybe_user_failure_reason: Option<String>,
+
+  /// Optional internal-only stack trace or error (512 characters max).
+  pub maybe_internal_debugging_failure_reason: Option<String>,
+
+  /// Optional total wall-clock runtime of the failed attempt, in milliseconds.
+  pub execution_duration_millis: Option<u64>,
+
+  /// Optional inference-only runtime of the failed attempt, in milliseconds.
+  pub inference_duration_millis: Option<u64>,
+}
+
+/// Response body for `POST /v1/internal/minimax_jobs/job/{job_token}/failure`.
+#[derive(Clone, Debug, Serialize, Deserialize, ToSchema)]
+pub struct MarkMinimaxJobFailureResponse {
+  pub success: bool,
+}
@@ -0,0 +1,29 @@
+use serde_derive::{Deserialize, Serialize};
+use utoipa::ToSchema;
+
+use tokens::tokens::media_files::MediaFileToken;
+
+// NB: The success request is form-multipart (video bytes + optional
+// `duration_millis` / `width` / `height` metadata fields), so its form struct
+// lives with the actix handler; only the response shape is declared here.
+
+/// Response body for `POST /v1/internal/minimax_jobs/job/{job_token}/success`.
+#[derive(Clone, Debug, Serialize, Deserialize, ToSchema)]
+pub struct MarkMinimaxJobSuccessResponse {
+  pub success: bool,
+
+  /// The media file created from the uploaded video.
+  pub media_file_token: MediaFileToken,
+
+  /// Video duration stored on the media file record, in milliseconds
+  /// (worker-supplied, or probed from the file when not supplied).
+  pub maybe_duration_millis: Option<u64>,
+
+  /// Frame width stored on the media file record, in pixels
+  /// (worker-supplied, or probed from the file when not supplied).
+  pub maybe_width: Option<u32>,
+
+  /// Frame height stored on the media file record, in pixels
+  /// (worker-supplied, or probed from the file when not supplied).
+  pub maybe_height: Option<u32>,
+}
@@ -0,0 +1,33 @@
+use serde_derive::{Deserialize, Serialize};
+use utoipa::ToSchema;
+
+/// The first-party Minimax models a worker can request jobs for.
+///
+/// This is deliberately its own enum (not `CommonVideoModel`) so the internal
+/// worker API surface stays small and stable.
+#[derive(Copy, Clone, Debug, Eq, PartialEq, Serialize, Deserialize, ToSchema)]
+pub enum MinimaxWorkerModel {
+  #[serde(rename = "minimax_h3_turbo")]
+  MinimaxH3Turbo,
+
+  #[serde(rename = "minimax_h3_ultra")]
+  MinimaxH3Ultra,
+}
+
+#[cfg(test)]
+mod tests {
+  use super::*;
+
+  #[test]
+  fn serde_round_trip() {
+    let cases = [
+      (MinimaxWorkerModel::MinimaxH3Turbo, "\"minimax_h3_turbo\""),
+      (MinimaxWorkerModel::MinimaxH3Ultra, "\"minimax_h3_ultra\""),
+    ];
+    for (variant, json) in cases {
+      assert_eq!(serde_json::to_string(&variant).unwrap(), json);
+      let parsed: MinimaxWorkerModel = serde_json::from_str(json).unwrap();
+      assert_eq!(parsed, variant);
+    }
+  }
+}
@@ -0,0 +1,4 @@
+pub mod mark_minimax_job_failure;
+pub mod mark_minimax_job_success;
+pub mod minimax_worker_model;
+pub mod obtain_minimax_job;
@@ -0,0 +1,85 @@
+use serde_derive::{Deserialize, Serialize};
+use utoipa::ToSchema;
+
+use enums::by_table::prompt_context_items::prompt_context_semantic_type::PromptContextSemanticType;
+use enums::common::generation::common_aspect_ratio::CommonAspectRatio;
+use enums::common::generation::common_bitrate::CommonBitrate;
+use enums::common::generation::common_generation_mode::CommonGenerationMode;
+use enums::common::generation::common_resolution::CommonResolution;
+use tokens::tokens::generic_inference_jobs::InferenceJobToken;
+use tokens::tokens::media_files::MediaFileToken;
+use tokens::tokens::prompts::PromptToken;
+
+use crate::common::responses::media_links::MediaLinks;
+use crate::internal::minimax_jobs::minimax_worker_model::MinimaxWorkerModel;
+
+/// Request body for `POST /v1/internal/minimax_jobs/obtain_job`.
+#[derive(Clone, Debug, Serialize, Deserialize, ToSchema)]
+pub struct ObtainMinimaxJobRequest {
+  /// Which model the worker wants a job for.
+  pub model: MinimaxWorkerModel,
+
+  /// The worker's hostname (linux hostname, k8s pod name), if reported.
+  pub worker_hostname: Option<String>,
+
+  /// The cluster the worker runs in (e.g. "runpod", "lambdalabs"), if reported.
+  pub cluster_name: Option<String>,
+}
+
+/// Response body for `POST /v1/internal/minimax_jobs/obtain_job`.
+#[derive(Clone, Debug, Serialize, Deserialize, ToSchema)]
+pub struct ObtainMinimaxJobResponse {
+  pub success: bool,
+
+  /// The obtained job, or `None` when no pending job is available.
+  pub maybe_job: Option<ObtainedMinimaxJob>,
+}
+
+/// A job the worker now holds a lock on (status was moved to `started`).
+#[derive(Clone, Debug, Serialize, Deserialize, ToSchema)]
+pub struct ObtainedMinimaxJob {
+  pub job_token: InferenceJobToken,
+
+  /// Echo of the requested model.
+  pub model: MinimaxWorkerModel,
+
+  /// The prompt and inputs the worker needs to run inference. Only `None`
+  /// when the job was somehow enqueued without a prompt record (or the
+  /// prompt lookup failed) — workers should fail such jobs via the failure
+  /// endpoint.
+  pub maybe_prompt: Option<MinimaxJobPromptDetails>,
+}
+
+/// The prompt (text + generation settings) and its attached media references.
+#[derive(Clone, Debug, Serialize, Deserialize, ToSchema)]
+pub struct MinimaxJobPromptDetails {
+  pub prompt_token: PromptToken,
+
+  pub maybe_positive_prompt: Option<String>,
+  pub maybe_negative_prompt: Option<String>,
+
+  pub maybe_generation_mode: Option<CommonGenerationMode>,
+  pub maybe_aspect_ratio: Option<CommonAspectRatio>,
+  pub maybe_resolution: Option<CommonResolution>,
+  pub maybe_bitrate: Option<CommonBitrate>,
+  pub maybe_generate_audio: Option<bool>,
+  pub maybe_duration_seconds: Option<u32>,
+
+  /// Image / video / audio references attached to the prompt, with full CDN
+  /// links, in their original order.
+  pub media_references: Vec<MinimaxJobMediaReference>,
+}
+
+/// One media input attached to the prompt (start frame, end frame, or an
+/// image / video / audio reference), with full CDN links.
+#[derive(Clone, Debug, Serialize, Deserialize, ToSchema)]
+pub struct MinimaxJobMediaReference {
+  pub media_file_token: MediaFileToken,
+
+  /// The role the media plays (start frame, end frame, image / video / audio
+  /// reference, etc.).
+  pub semantic_type: PromptContextSemanticType,
+
+  /// Full CDN links to the media file.
+  pub media_links: MediaLinks,
+}
@@ -0,0 +1,4 @@
+//! Internal-facing API surfaces (worker fleets, etc.). These are
+//! authenticated with internal API keys, never consumer credentials.
+
+pub mod minimax_jobs;
@@ -5,6 +5,7 @@ pub mod common;
 pub mod credits;
 pub mod folders;
 pub mod generate;
+pub mod internal;
 pub mod jobs;
 pub mod media_file;
 pub mod moderation;
@@ -0,0 +1,25 @@
+use enums::by_table::generic_inference_jobs::inference_job_type::InferenceJobType;
+use enums::by_table::generic_inference_jobs::inference_model_type::InferenceModelType;
+
+/// The first-party Minimax H3 model tier a job runs on.
+#[derive(Debug, Clone, Copy, Eq, PartialEq)]
+pub enum FirstPartyMinimaxH3Model {
+  Turbo,
+  Ultra,
+}
+
+impl FirstPartyMinimaxH3Model {
+  pub fn inference_job_type(&self) -> InferenceJobType {
+    match self {
+      Self::Turbo => InferenceJobType::ArtcraftMinimaxH3Turbo,
+      Self::Ultra => InferenceJobType::ArtcraftMinimaxH3Ultra,
+    }
+  }
+
+  pub fn inference_model_type(&self) -> InferenceModelType {
+    match self {
+      Self::Turbo => InferenceModelType::MinimaxH3Turbo,
+      Self::Ultra => InferenceModelType::MinimaxH3Ultra,
+    }
+  }
+}
@@ -0,0 +1,99 @@
+//! Look up a first-party Minimax H3 job by token.
+//!
+//! Only returns rows whose `job_type` is one of the first-party Minimax H3
+//! job types — the internal worker API must never read (or later mutate)
+//! unrelated jobs, so an unknown or non-Minimax token yields `Ok(None)`.
+
+use sqlx::{Executor, MySql};
+use std::marker::PhantomData;
+
+use enums::by_table::generic_inference_jobs::inference_job_type::InferenceJobType;
+use enums::common::job_status_plus::JobStatusPlus;
+use enums::common::platform_type::PlatformType;
+use tokens::tokens::anonymous_visitor_tracking::AnonymousVisitorTrackingToken;
+use tokens::tokens::generic_inference_jobs::InferenceJobToken;
+use tokens::tokens::media_files::MediaFileToken;
+use tokens::tokens::non_unique::debug_logs_event_token::DebugLogEventToken;
+use tokens::tokens::prompts::PromptToken;
+use tokens::tokens::users::UserToken;
+
+pub struct GetFirstPartyMinimaxH3JobByTokenArgs<'e, 'c, E>
+  where E: 'e + Executor<'c, Database = MySql>
+{
+  pub job_token: &'e InferenceJobToken,
+
+  pub mysql_executor: E,
+
+  pub phantom: PhantomData<&'c E>,
+}
+
+#[derive(Debug)]
+pub struct FirstPartyMinimaxH3JobDetails {
+  pub job_token: InferenceJobToken,
+
+  pub status: JobStatusPlus,
+  pub job_type: InferenceJobType,
+
+  pub maybe_creator_user_token: Option<UserToken>,
+  pub maybe_creator_anonymous_visitor_token: Option<AnonymousVisitorTrackingToken>,
+  pub creator_ip_address: String,
+
+  pub maybe_prompt_token: Option<PromptToken>,
+  pub maybe_debug_log_event_token: Option<DebugLogEventToken>,
+
+  pub maybe_platform_type: Option<PlatformType>,
+
+  /// Set once the job has succeeded: the resulting media file.
+  pub maybe_result_media_file_token: Option<MediaFileToken>,
+}
+
+pub async fn get_first_party_minimax_h3_job_by_token<'e, 'c : 'e, E>(
+  args: GetFirstPartyMinimaxH3JobByTokenArgs<'e, 'c, E>
+) -> Result<Option<FirstPartyMinimaxH3JobDetails>, sqlx::Error>
+  where E: 'e + Executor<'c, Database = MySql>
+{
+  let maybe_row = sqlx::query!(
+    r#"
+SELECT
+  token as `job_token: tokens::tokens::generic_inference_jobs::InferenceJobToken`,
+  status as `status: enums::common::job_status_plus::JobStatusPlus`,
+  job_type as `job_type: enums::by_table::generic_inference_jobs::inference_job_type::InferenceJobType`,
+  maybe_creator_user_token as `maybe_creator_user_token: tokens::tokens::users::UserToken`,
+  maybe_creator_anonymous_visitor_token as `maybe_creator_anonymous_visitor_token: tokens::tokens::anonymous_visitor_tracking::AnonymousVisitorTrackingToken`,
+  creator_ip_address,
+  maybe_prompt_token as `maybe_prompt_token: tokens::tokens::prompts::PromptToken`,
+  maybe_debug_log_event_token as `maybe_debug_log_event_token: tokens::tokens::non_unique::debug_logs_event_token::DebugLogEventToken`,
+  platform_type as `maybe_platform_type: enums::common::platform_type::PlatformType`,
+  on_success_result_entity_token as `maybe_result_media_file_token: tokens::tokens::media_files::MediaFileToken`
+FROM generic_inference_jobs
+WHERE token = ?
+AND job_type IN ('artcraft_minimax_h3_turbo', 'artcraft_minimax_h3_ultra')
+    "#,
+    args.job_token.as_str(),
+  )
+    .fetch_optional(args.mysql_executor)
+    .await?;
+
+  let Some(row) = maybe_row else {
+    return Ok(None);
+  };
+
+  let job_type = match row.job_type {
+    Some(job_type) => job_type,
+    // Unreachable given the WHERE clause, but don't panic on it.
+    None => return Ok(None),
+  };
+
+  Ok(Some(FirstPartyMinimaxH3JobDetails {
+    job_token: row.job_token,
+    status: row.status,
+    job_type,
+    maybe_creator_user_token: row.maybe_creator_user_token,
+    maybe_creator_anonymous_visitor_token: row.maybe_creator_anonymous_visitor_token,
+    creator_ip_address: row.creator_ip_address,
+    maybe_prompt_token: row.maybe_prompt_token,
+    maybe_debug_log_event_token: row.maybe_debug_log_event_token,
+    maybe_platform_type: row.maybe_platform_type,
+    maybe_result_media_file_token: row.maybe_result_media_file_token,
+  }))
+}
@@ -11,8 +11,6 @@ use std::marker::PhantomData;

 use enums::by_table::generic_inference_jobs::inference_category::InferenceCategory;
 use enums::by_table::generic_inference_jobs::inference_job_product_category::InferenceJobProductCategory;
-use enums::by_table::generic_inference_jobs::inference_job_type::InferenceJobType;
-use enums::by_table::generic_inference_jobs::inference_model_type::InferenceModelType;
 use enums::common::job_status_plus::JobStatusPlus;
 use enums::common::platform_type::PlatformType;
 use enums::common::visibility::Visibility;
@@ -26,6 +24,7 @@ use tokens::tokens::wallet_ledger_entries::WalletLedgerEntryToken;
 use crate::errors::database_query_error::DatabaseQueryError;
 use crate::payloads::generic_inference_args::generic_inference_args::GenericInferenceArgs;
 use crate::queries::generic_inference::common::job_cost_estimates::JobCostEstimates;
+use crate::queries::generic_inference::first_party::minimax_h3::first_party_minimax_h3_model::FirstPartyMinimaxH3Model;
 use crate::queries::generic_inference::common::insert_full_generic_inference_job_record::{
  insert_full_generic_inference_job_record,
  InsertFullGenericInferenceJobRecordArgs,
@@ -69,28 +68,13 @@ pub struct InsertGenericInferenceForFirstPartyMinimaxH3WithAprioriJobTokenArgs<'
  pub phantom: PhantomData<&'c E>,
 }

-/// The first-party Minimax H3 model tier being enqueued.
-#[derive(Debug, Clone, Copy, Eq, PartialEq)]
-pub enum FirstPartyMinimaxH3Model {
-  Turbo,
-  Ultra,
-}
-
 pub async fn insert_generic_inference_job_for_first_party_minimax_h3_with_apriori_job_token<'e, 'c : 'e, E>(
  args: InsertGenericInferenceForFirstPartyMinimaxH3WithAprioriJobTokenArgs<'e, 'c, E>
 ) -> Result<InferenceJobToken, DatabaseQueryError>
  where E: 'e + Executor<'c, Database = MySql>
 {
-  let (job_type, model_type) = match args.minimax_model {
-    FirstPartyMinimaxH3Model::Turbo => (
-      InferenceJobType::ArtcraftMinimaxH3Turbo,
-      InferenceModelType::MinimaxH3Turbo,
-    ),
-    FirstPartyMinimaxH3Model::Ultra => (
-      InferenceJobType::ArtcraftMinimaxH3Ultra,
-      InferenceModelType::MinimaxH3Ultra,
-    ),
-  };
+  let job_type = args.minimax_model.inference_job_type();
+  let model_type = args.minimax_model.inference_model_type();

  let cost_estimates = JobCostEstimates {
    // First-party jobs have no external provider, so no provider-side costs.
@@ -0,0 +1,88 @@
+//! Permanently mark a first-party Minimax H3 job as failed.
+//!
+//! Never flips an already-terminal job: a late or duplicate failure report
+//! against a `complete_success` / `complete_failure` row is silently ignored.
+
+use sqlx::{Executor, MySql};
+use std::convert::TryFrom;
+use std::marker::PhantomData;
+
+use enums::by_table::generic_inference_jobs::frontend_failure_category::FrontendFailureCategory;
+use enums::common::job_status_plus::JobStatusPlus;
+use tokens::tokens::generic_inference_jobs::InferenceJobToken;
+
+/// `failure_reason` / `internal_debugging_failure_reason` are VARCHAR(512).
+const MAX_FAILURE_REASON_CHARS: usize = 512;
+
+pub struct MarkFirstPartyMinimaxH3JobFailedArgs<'e, 'c, E>
+  where E: 'e + Executor<'c, Database = MySql>
+{
+  pub job_token: &'e InferenceJobToken,
+
+  pub maybe_frontend_failure_category: Option<FrontendFailureCategory>,
+
+  /// User-facing failure reason. Clamped to 512 characters.
+  pub maybe_failure_reason: Option<&'e str>,
+
+  /// Internal-only stack trace or error. Clamped to 512 characters.
+  pub maybe_internal_debugging_failure_reason: Option<&'e str>,
+
+  /// Total wall-clock runtime of the failed attempt, in milliseconds.
+  /// NB: Written to `success_execution_millis` — the column name is
+  /// unfortunate, but we want the runtime data for failures too.
+  pub maybe_execution_duration_millis: Option<u64>,
+
+  /// Inference-only runtime of the failed attempt, in milliseconds.
+  /// NB: Written to `success_inference_execution_millis` — the column name is
+  /// unfortunate, but we want the runtime data for failures too.
+  pub maybe_inference_duration_millis: Option<u64>,
+
+  pub mysql_executor: E,
+
+  pub phantom: PhantomData<&'c E>,
+}
+
+pub async fn mark_first_party_minimax_h3_job_failed<'e, 'c : 'e, E>(
+  args: MarkFirstPartyMinimaxH3JobFailedArgs<'e, 'c, E>
+) -> Result<(), sqlx::Error>
+  where E: 'e + Executor<'c, Database = MySql>
+{
+  let maybe_failure_reason = args.maybe_failure_reason
+    .map(|reason| clamp_chars(reason, MAX_FAILURE_REASON_CHARS));
+
+  let maybe_internal_debugging_failure_reason = args.maybe_internal_debugging_failure_reason
+    .map(|reason| clamp_chars(reason, MAX_FAILURE_REASON_CHARS));
+
+  sqlx::query!(
+    r#"
+UPDATE generic_inference_jobs
+SET
+  status = ?,
+  frontend_failure_category = ?,
+  failure_reason = ?,
+  internal_debugging_failure_reason = ?,
+  success_execution_millis = ?,
+  success_inference_execution_millis = ?
+WHERE token = ?
+AND status NOT IN ('complete_success', 'complete_failure')
+    "#,
+    JobStatusPlus::CompleteFailure.to_str(),
+    args.maybe_frontend_failure_category.map(|category| category.to_str()),
+    maybe_failure_reason,
+    maybe_internal_debugging_failure_reason,
+    args.maybe_execution_duration_millis.and_then(|millis| u32::try_from(millis).ok()),
+    args.maybe_inference_duration_millis.and_then(|millis| u32::try_from(millis).ok()),
+    args.job_token.as_str(),
+  )
+    .execute(args.mysql_executor)
+    .await?;
+
+  Ok(())
+}
+
+fn clamp_chars(value: &str, max_chars: usize) -> &str {
+  match value.char_indices().nth(max_chars) {
+    Some((byte_index, _)) => &value[..byte_index],
+    None => value,
+  }
+}
@@ -0,0 +1,68 @@
+//! Mark a first-party Minimax H3 job as started and record which worker
+//! obtained it.
+//!
+//! Takes a transaction (not a generic executor): this must run in the same
+//! transaction as [`select_pending_first_party_minimax_h3_job_for_update`]
+//! so the select-then-mark pair is atomic.
+//!
+//! [`select_pending_first_party_minimax_h3_job_for_update`]:
+//! crate::queries::generic_inference::first_party::minimax_h3::select_pending_first_party_minimax_h3_job_for_update
+
+use sqlx::{MySql, Transaction};
+
+use enums::common::job_status_plus::JobStatusPlus;
+use tokens::tokens::generic_inference_jobs::InferenceJobToken;
+
+/// `assigned_worker` / `assigned_cluster` are VARCHAR(128).
+const MAX_WORKER_FIELD_CHARS: usize = 128;
+
+pub struct MarkFirstPartyMinimaxH3JobStartedArgs<'a, 'tx> {
+  pub job_token: &'a InferenceJobToken,
+
+  /// The worker's hostname (linux hostname, k8s pod name), when reported.
+  pub maybe_worker_hostname: Option<&'a str>,
+
+  /// The cluster the worker runs in (e.g. "runpod", "lambda"), when reported.
+  pub maybe_cluster_name: Option<&'a str>,
+
+  pub transaction: &'a mut Transaction<'tx, MySql>,
+}
+
+pub async fn mark_first_party_minimax_h3_job_started(
+  args: MarkFirstPartyMinimaxH3JobStartedArgs<'_, '_>
+) -> Result<(), sqlx::Error> {
+  let maybe_worker_hostname = args.maybe_worker_hostname
+    .map(|hostname| clamp_chars(hostname, MAX_WORKER_FIELD_CHARS));
+
+  let maybe_cluster_name = args.maybe_cluster_name
+    .map(|cluster| clamp_chars(cluster, MAX_WORKER_FIELD_CHARS));
+
+  sqlx::query!(
+    r#"
+UPDATE generic_inference_jobs
+SET
+  status = ?,
+  assigned_worker = ?,
+  last_assigned_worker = ?,
+  assigned_cluster = ?,
+  first_started_at = NOW()
+WHERE token = ?
+    "#,
+    JobStatusPlus::Started.to_str(),
+    maybe_worker_hostname,
+    maybe_worker_hostname,
+    maybe_cluster_name,
+    args.job_token.as_str(),
+  )
+    .execute(&mut **args.transaction)
+    .await?;
+
+  Ok(())
+}
+
+fn clamp_chars(value: &str, max_chars: usize) -> &str {
+  match value.char_indices().nth(max_chars) {
+    Some((byte_index, _)) => &value[..byte_index],
+    None => value,
+  }
+}
@@ -0,0 +1,67 @@
+//! Mark a first-party Minimax H3 job as successfully completed, pointing at
+//! the resulting media file. Mirrors the Fal webhook success write.
+//!
+//! Never flips an already-terminal job: a late or duplicate success report
+//! against a `complete_success` / `complete_failure` row is silently ignored.
+
+use sqlx::{Executor, MySql};
+use std::convert::TryFrom;
+use std::marker::PhantomData;
+
+use enums::by_table::generic_inference_jobs::inference_result_type::InferenceResultType;
+use enums::common::job_status_plus::JobStatusPlus;
+use tokens::tokens::generic_inference_jobs::InferenceJobToken;
+use tokens::tokens::media_files::MediaFileToken;
+
+pub struct MarkFirstPartyMinimaxH3JobSucceededArgs<'e, 'c, E>
+  where E: 'e + Executor<'c, Database = MySql>
+{
+  pub job_token: &'e InferenceJobToken,
+
+  /// The media file created from the worker's uploaded video.
+  pub media_file_token: &'e MediaFileToken,
+
+  /// Total wall-clock runtime of the job, in milliseconds.
+  pub maybe_execution_duration_millis: Option<u64>,
+
+  /// Inference-only runtime of the job, in milliseconds.
+  pub maybe_inference_duration_millis: Option<u64>,
+
+  pub mysql_executor: E,
+
+  pub phantom: PhantomData<&'c E>,
+}
+
+pub async fn mark_first_party_minimax_h3_job_succeeded<'e, 'c : 'e, E>(
+  args: MarkFirstPartyMinimaxH3JobSucceededArgs<'e, 'c, E>
+) -> Result<(), sqlx::Error>
+  where E: 'e + Executor<'c, Database = MySql>
+{
+  sqlx::query!(
+    r#"
+UPDATE generic_inference_jobs
+SET
+  status = ?,
+  on_success_result_entity_type = ?,
+  on_success_result_entity_token = ?,
+  failure_reason = NULL,
+  internal_debugging_failure_reason = NULL,
+  success_execution_millis = ?,
+  success_inference_execution_millis = ?,
+  retry_at = NULL,
+  successfully_completed_at = NOW()
+WHERE token = ?
+AND status NOT IN ('complete_success', 'complete_failure')
+    "#,
+    JobStatusPlus::CompleteSuccess.to_str(),
+    InferenceResultType::MediaFile.to_str(),
+    args.media_file_token.as_str(),
+    args.maybe_execution_duration_millis.and_then(|millis| u32::try_from(millis).ok()),
+    args.maybe_inference_duration_millis.and_then(|millis| u32::try_from(millis).ok()),
+    args.job_token.as_str(),
+  )
+    .execute(args.mysql_executor)
+    .await?;
+
+  Ok(())
+}
@@ -1 +1,7 @@
+pub mod first_party_minimax_h3_model;
+pub mod get_first_party_minimax_h3_job_by_token;
 pub mod insert_generic_inference_job_for_first_party_minimax_h3_with_apriori_job_token;
+pub mod mark_first_party_minimax_h3_job_failed;
+pub mod mark_first_party_minimax_h3_job_started;
+pub mod mark_first_party_minimax_h3_job_succeeded;
+pub mod select_pending_first_party_minimax_h3_job_for_update;
@@ -0,0 +1,67 @@
+//! `SELECT ... FOR UPDATE SKIP LOCKED` the oldest pending first-party
+//! Minimax H3 job of a given model tier.
+//!
+//! Takes a transaction (not a generic executor) — the row lock only means
+//! anything inside a transaction, and the caller must run
+//! [`mark_first_party_minimax_h3_job_started`] in the same transaction so the
+//! select-then-mark pair is atomic. `SKIP LOCKED` keeps concurrent workers
+//! from blocking on (or double-obtaining) the same row: each worker locks a
+//! distinct pending row or gets `None`.
+//!
+//! Performance: the scan is driven by `index_job_type` (job_type, then PK
+//! order, so "oldest first" is index order) and only pending rows survive
+//! the filter; pending first-party rows are expected to be few.
+//!
+//! [`mark_first_party_minimax_h3_job_started`]:
+//! crate::queries::generic_inference::first_party::minimax_h3::mark_first_party_minimax_h3_job_started
+
+use sqlx::{MySql, Transaction};
+
+use tokens::tokens::generic_inference_jobs::InferenceJobToken;
+use tokens::tokens::prompts::PromptToken;
+
+use crate::queries::generic_inference::first_party::minimax_h3::first_party_minimax_h3_model::FirstPartyMinimaxH3Model;
+
+pub struct SelectPendingFirstPartyMinimaxH3JobForUpdateArgs<'a, 'tx> {
+  /// Which model tier to obtain a job for.
+  pub minimax_model: FirstPartyMinimaxH3Model,
+
+  pub transaction: &'a mut Transaction<'tx, MySql>,
+}
+
+/// The locked pending job.
+#[derive(Debug)]
+pub struct PendingFirstPartyMinimaxH3Job {
+  pub job_token: InferenceJobToken,
+  pub maybe_prompt_token: Option<PromptToken>,
+}
+
+/// Returns the locked job, or `None` when no unlocked pending job of the
+/// given tier exists.
+pub async fn select_pending_first_party_minimax_h3_job_for_update(
+  args: SelectPendingFirstPartyMinimaxH3JobForUpdateArgs<'_, '_>
+) -> Result<Option<PendingFirstPartyMinimaxH3Job>, sqlx::Error> {
+  let job_type = args.minimax_model.inference_job_type();
+
+  let maybe_row = sqlx::query!(
+    r#"
+SELECT
+  token as `job_token: tokens::tokens::generic_inference_jobs::InferenceJobToken`,
+  maybe_prompt_token as `maybe_prompt_token: tokens::tokens::prompts::PromptToken`
+FROM generic_inference_jobs
+WHERE job_type = ?
+AND status = 'pending'
+ORDER BY id ASC
+LIMIT 1
+FOR UPDATE SKIP LOCKED
+    "#,
+    job_type.to_str(),
+  )
+    .fetch_optional(&mut **args.transaction)
+    .await?;
+
+  Ok(maybe_row.map(|row| PendingFirstPartyMinimaxH3Job {
+    job_token: row.job_token,
+    maybe_prompt_token: row.maybe_prompt_token,
+  }))
+}
@@ -1,5 +1,5 @@
 use anyhow::anyhow;
-use sqlx::MySqlPool;
+use sqlx::{Acquire, MySql};

 use enums::by_table::generic_synthetic_ids::id_category::IdCategory;
 use enums::by_table::media_files::media_file_class::MediaFileClass;
@@ -24,8 +24,13 @@ use crate::payloads::media_file_extra_info::media_file_extra_info::MediaFileExtr
 use crate::queries::generic_synthetic_ids::transactional_increment_generic_synthetic_id::transactional_increment_generic_synthetic_id;
 use crate::queries::media_files::create::generic_insert::insert_media_file_generic_executor::{insert_media_file_generic_executor, InsertMediaFileGenericExecutorArgs};

-pub struct InsertArgs<'a> {
-    pub pool: &'a MySqlPool,
+pub struct InsertArgs<'a, A>
+where
+    A: Acquire<'a, Database = MySql>,
+{
+    /// Where to get the connection: a `&MySqlPool` (acquires one) or an
+    /// already-acquired `&mut PoolConnection` (reuses it).
+    pub pool: A,

    // Creator info
    pub maybe_creator_user_token: Option<&'a UserToken>,
@@ -111,9 +116,12 @@ pub struct InsertArgs<'a> {
 /// Insert a media file record, allocating the creator's synthetic IDs in the
 /// same transaction. Wraps [`insert_media_file_generic_executor`], which runs
 /// the actual INSERT.
-pub async fn insert_media_file_generic(
-    args: InsertArgs<'_>
+pub async fn insert_media_file_generic<'a, A>(
+    args: InsertArgs<'a, A>
 ) -> AnyhowResult<(MediaFileToken, u64)>
+where
+    A: Acquire<'a, Database = MySql> + Send,
+    <A as Acquire<'a>>::Connection: Send,
 {
    let extra_file_modification_info = args
        .maybe_extra_media_info.map(|extra| extra.to_json_string())
@@ -128,7 +128,7 @@ pub async fn insert_media_file_generic_from_job(
        media_type: args.media_type,
        origin_category: args.origin_category,
        origin_product_category: args.origin_product_category,
-        pool: &args.pool,
+        pool: args.pool,
        public_bucket_directory_hash: args.public_bucket_directory_hash,
    }).await
 }
@@ -8,7 +8,8 @@ use enums::by_table::media_files::media_file_origin_category::MediaFileOriginCat
 use enums::by_table::media_files::media_file_origin_product_category::MediaFileOriginProductCategory;
 use enums::by_table::media_files::media_file_type::MediaFileType;
 use enums::common::visibility::Visibility;
-use sqlx::MySqlPool;
+use sqlx::pool::PoolConnection;
+use sqlx::{Acquire, MySql, MySqlPool};
 use enums::common::generation_provider::GenerationProvider;
 use enums::common::platform_type::PlatformType;
 use tokens::tokens::anonymous_visitor_tracking::AnonymousVisitorTrackingToken;
@@ -293,20 +294,33 @@ impl MediaFileInsertBuilder {
    self
  }

-  // TODO(bt,2025-04-26): Other connector options.
  pub async fn insert_pool(self, mysql_pool: &MySqlPool) -> Result<MediaFileToken, MediaFileInsertBuilderError> {
+    self.insert_acquire(mysql_pool).await
+  }
+
+  /// Insert using an already-acquired connection, so callers holding one
+  /// don't pay for a second pool acquisition.
+  pub async fn insert_connection(self, mysql_connection: &mut PoolConnection<MySql>) -> Result<MediaFileToken, MediaFileInsertBuilderError> {
+    self.insert_acquire(mysql_connection).await
+  }
+
+  async fn insert_acquire<'a, A>(&'a self, connector: A) -> Result<MediaFileToken, MediaFileInsertBuilderError>
+  where
+    A: Acquire<'a, Database = MySql> + Send,
+    <A as Acquire<'a>>::Connection: Send,
+  {
    let media_file_type = self.media_file_type
        .ok_or_else(|| MediaFileInsertBuilderError::MissingRequiredField(
          "Media file type is required".to_string()))?;

-    let checksum_sha2 = self.checksum_sha2
+    let checksum_sha2 = self.checksum_sha2.as_ref()
        .ok_or_else(|| MediaFileInsertBuilderError::MissingRequiredField(
          "Checksum SHA2 is required".to_string()))?;

-    let bucket_path = self.public_bucket_directory_hash
+    let bucket_path = self.public_bucket_directory_hash.as_ref()
        .ok_or_else(|| MediaFileInsertBuilderError::MissingRequiredField(
          "Public bucket directory hash is required".to_string()))?;
-    
+
    let mut origin_category = self.origin_category
        .ok_or_else(|| MediaFileInsertBuilderError::MissingRequiredField(
          "Origin category is required".to_string()))?;
@@ -330,7 +344,7 @@ impl MediaFileInsertBuilder {
    }

    let result = insert_media_file_generic(InsertArgs {
-      pool: mysql_pool,
+      pool: connector,
      maybe_creator_user_token: self.maybe_creator_user_token.as_ref(),
      maybe_creator_anonymous_visitor_token: self.maybe_creator_anonymous_visitor_token.as_ref(),
      creator_ip_address: self.creator_ip_address.as_deref().unwrap_or("127.0.0.1"),
@@ -343,7 +357,7 @@ impl MediaFileInsertBuilder {
      origin_product_category: self.origin_product_category,
      maybe_origin_model_type: None, // TODO
      maybe_origin_model_token: None, // TODO
-      maybe_origin_filename: self.maybe_origin_filename,
+      maybe_origin_filename: self.maybe_origin_filename.clone(),
      maybe_mime_type: self.maybe_mime_type.as_deref(),
      file_size_bytes: self.file_size_bytes,
      maybe_duration_millis: self.maybe_duration_millis,
@@ -352,7 +366,7 @@ impl MediaFileInsertBuilder {
      maybe_frame_width: self.maybe_frame_width,
      maybe_frame_height: self.maybe_frame_height,
      maybe_engine_category: self.maybe_engine_category,
-      checksum_sha2: &checksum_sha2,
+      checksum_sha2: checksum_sha2.as_str(),
      maybe_title: self.maybe_title.as_deref(),
      maybe_text_transcript: None, // TODO
      maybe_scene_source_media_file_token: None, // TODO
@@ -50,7 +50,7 @@ pub async fn insert_media_file_from_studio_scene_render(
 ) -> AnyhowResult<MediaFileToken>
 {
  let (new_media_token, _id) = insert_media_file_generic(InsertArgs {
-    pool: &args.pool,
+    pool: args.pool,

    // Dynamic bits (user)
    maybe_creator_user_token: args.maybe_creator_user_token,
@@ -111,6 +111,10 @@ use crate::http_server::endpoints::api_keys::list_api_keys_handler::*;
 use crate::http_server::endpoints::api_keys::get_api_key_handler::*;
 use crate::http_server::endpoints::api_keys::delete_api_key_handler::*;
 use crate::http_server::endpoints::api_keys::update_api_key_handler::*;
+use artcraft_api_defs::internal::minimax_jobs::mark_minimax_job_failure::*;
+use artcraft_api_defs::internal::minimax_jobs::mark_minimax_job_success::*;
+use artcraft_api_defs::internal::minimax_jobs::minimax_worker_model::*;
+use artcraft_api_defs::internal::minimax_jobs::obtain_minimax_job::*;
 use artcraft_api_defs::api_keys::common::*;
 use artcraft_api_defs::api_keys::create_api_key::*;
 use artcraft_api_defs::api_keys::list_api_keys::*;
@@ -641,6 +645,10 @@ use crate::http_server::endpoints::media_files::list::list_batch_generated_redux
    crate::http_server::endpoints::omni_api::upload::omni_upload_video_media_file_handler::omni_upload_video_media_file_handler,
    crate::http_server::endpoints::omni_api::job_status::get_job_status_handler::omni_api_get_job_status_handler,
    crate::http_server::endpoints::omni_api::job_status::batch_get_job_status_handler::omni_api_batch_get_job_status_handler,
+    // Internal (worker fleets, internal API key authenticated)
+    crate::http_server::endpoints::internal::minimax_jobs::obtain_minimax_job_handler::obtain_minimax_job_handler,
+    crate::http_server::endpoints::internal::minimax_jobs::mark_minimax_job_failure_handler::mark_minimax_job_failure_handler,
+    crate::http_server::endpoints::internal::minimax_jobs::mark_minimax_job_success_handler::mark_minimax_job_success_handler,
    // Moderation
    crate::http_server::endpoints::moderation::alerts::moderation_send_alert_handler::moderation_send_alert_handler,
    crate::http_server::endpoints::moderation::dashboards::moderator_list_databox_dashboards_handler::moderator_list_databox_dashboards_handler,
@@ -757,6 +765,17 @@ use crate::http_server::endpoints::media_files::list::list_batch_generated_redux
    UserToken,
    ZsVoiceDatasetToken,

+    // Internal (worker fleets)
+    MarkMinimaxJobFailureRequest,
+    MarkMinimaxJobFailureResponse,
+    MarkMinimaxJobSuccessResponse,
+    MinimaxJobMediaReference,
+    MinimaxJobPromptDetails,
+    MinimaxWorkerModel,
+    ObtainMinimaxJobRequest,
+    ObtainMinimaxJobResponse,
+    ObtainedMinimaxJob,
+
    // Enums
    BetaKeyProduct,
    CommentEntityType,
@@ -0,0 +1,121 @@
+use std::sync::Arc;
+
+use actix_web::web::Json;
+use actix_web::{web, HttpRequest};
+use log::{error, info, warn};
+use serde_derive::Deserialize;
+
+use artcraft_api_defs::internal::minimax_jobs::mark_minimax_job_failure::{
+  MarkMinimaxJobFailureRequest, MarkMinimaxJobFailureResponse,
+};
+use mysql_queries::queries::generic_inference::first_party::minimax_h3::get_first_party_minimax_h3_job_by_token::{
+  get_first_party_minimax_h3_job_by_token, GetFirstPartyMinimaxH3JobByTokenArgs,
+};
+use mysql_queries::queries::generic_inference::first_party::minimax_h3::mark_first_party_minimax_h3_job_failed::{
+  mark_first_party_minimax_h3_job_failed, MarkFirstPartyMinimaxH3JobFailedArgs,
+};
+use tokens::tokens::generic_inference_jobs::InferenceJobToken;
+
+use crate::http_server::common_responses::common_web_error::CommonWebError;
+use crate::http_server::web_utils::require_internal_api_key::require_internal_api_key;
+use crate::state::server_state::ServerState;
+
+#[derive(Deserialize)]
+pub struct MarkMinimaxJobFailurePathInfo {
+  job_token: InferenceJobToken,
+}
+
+/// Internal (worker-facing): mark a first-party Minimax job as failed.
+#[utoipa::path(
+  post,
+  tag = "Internal (Minimax Jobs)",
+  path = "/v1/internal/minimax_jobs/job/{job_token}/failure",
+  request_body = MarkMinimaxJobFailureRequest,
+  params(
+    ("job_token" = String, Path, description = "The inference job token"),
+  ),
+  responses(
+    (status = 200, description = "Success", body = MarkMinimaxJobFailureResponse),
+    (status = 401, description = "Missing or invalid internal API key"),
+    (status = 404, description = "No such minimax job"),
+    (status = 500, description = "Server error"),
+  ),
+)]
+pub async fn mark_minimax_job_failure_handler(
+  http_request: HttpRequest,
+  path: web::Path<MarkMinimaxJobFailurePathInfo>,
+  request: Json<MarkMinimaxJobFailureRequest>,
+  server_state: web::Data<Arc<ServerState>>,
+) -> Result<Json<MarkMinimaxJobFailureResponse>, CommonWebError> {
+
+  require_internal_api_key(&http_request, &server_state)?;
+
+  let job_token = &path.job_token;
+
+  let mut mysql_connection = server_state.mysql_pool.acquire().await?;
+
+  let maybe_job = get_first_party_minimax_h3_job_by_token(GetFirstPartyMinimaxH3JobByTokenArgs {
+    job_token,
+    mysql_executor: &mut *mysql_connection,
+    phantom: Default::default(),
+  }).await.map_err(|err| {
+    error!("Error looking up minimax job {}: {:?}", job_token, err);
+    CommonWebError::from_error(err)
+  })?;
+
+  let Some(job) = maybe_job else {
+    warn!("No such minimax job: {}", job_token);
+    return Err(CommonWebError::NotFound);
+  };
+
+  // Worker-supplied text is shown to users; strip control characters.
+  // (The query clamps both reasons to the columns' 512 characters.)
+  let maybe_user_failure_reason = request.maybe_user_failure_reason
+    .as_deref()
+    .map(sanitize_user_failure_reason)
+    .filter(|reason| !reason.is_empty());
+
+  mark_first_party_minimax_h3_job_failed(MarkFirstPartyMinimaxH3JobFailedArgs {
+    job_token: &job.job_token,
+    maybe_frontend_failure_category: request.maybe_frontend_failure_category,
+    maybe_failure_reason: maybe_user_failure_reason.as_deref(),
+    maybe_internal_debugging_failure_reason: request.maybe_internal_debugging_failure_reason.as_deref(),
+    maybe_execution_duration_millis: request.execution_duration_millis,
+    maybe_inference_duration_millis: request.inference_duration_millis,
+    mysql_executor: &mut *mysql_connection,
+    phantom: Default::default(),
+  }).await.map_err(|err| {
+    error!("Error marking minimax job {} failed: {:?}", job_token, err);
+    CommonWebError::from_error(err)
+  })?;
+
+  info!("Minimax job {} marked as failed (category: {:?})",
+    job.job_token, request.maybe_frontend_failure_category);
+
+  Ok(Json(MarkMinimaxJobFailureResponse {
+    success: true,
+  }))
+}
+
+/// Replace control characters (including newlines) with spaces and trim.
+fn sanitize_user_failure_reason(reason: &str) -> String {
+  reason
+    .chars()
+    .map(|c| if c.is_control() { ' ' } else { c })
+    .collect::<String>()
+    .trim()
+    .to_string()
+}
+
+#[cfg(test)]
+mod tests {
+  use super::*;
+
+  #[test]
+  fn sanitize_strips_control_characters() {
+    assert_eq!(sanitize_user_failure_reason("line one\nline two\r\n"), "line one line two");
+    assert_eq!(sanitize_user_failure_reason("\t  spaced  \t"), "spaced");
+    assert_eq!(sanitize_user_failure_reason("\u{0000}\u{0007}"), "");
+    assert_eq!(sanitize_user_failure_reason("plain reason"), "plain reason");
+  }
+}
@@ -0,0 +1,361 @@
+use std::io::Read;
+use std::sync::Arc;
+
+use actix_multipart::form::tempfile::TempFile;
+use actix_multipart::form::text::Text;
+use actix_multipart::form::MultipartForm;
+use actix_web::web::Json;
+use actix_web::{web, HttpRequest};
+use log::{error, info, warn};
+use serde_derive::Deserialize;
+use sqlx::pool::PoolConnection;
+use sqlx::MySql;
+use utoipa::ToSchema;
+
+use artcraft_api_defs::internal::minimax_jobs::mark_minimax_job_success::MarkMinimaxJobSuccessResponse;
+use bucket_paths::legacy::typified_paths::public::media_files::bucket_file_path::MediaFileBucketPath;
+use enums::by_table::media_files::media_file_class::MediaFileClass;
+use enums::by_table::media_files::media_file_origin_category::MediaFileOriginCategory;
+use enums::by_table::media_files::media_file_type::MediaFileType;
+use enums::common::generation_provider::GenerationProvider;
+use enums::common::job_status_plus::JobStatusPlus;
+use ffmpeg_utils::ffprobe::ffprobe_get_info::ffprobe_get_info;
+use filesys::path_to_string::path_to_string;
+use hashing::sha256::sha256_hash_bytes::sha256_hash_bytes;
+use mimetypes::mimetype_for_bytes::get_mimetype_for_bytes;
+use mimetypes::mimetype_to_extension::mimetype_to_extension;
+use mysql_queries::queries::generic_inference::first_party::minimax_h3::get_first_party_minimax_h3_job_by_token::{
+  get_first_party_minimax_h3_job_by_token, FirstPartyMinimaxH3JobDetails, GetFirstPartyMinimaxH3JobByTokenArgs,
+};
+use mysql_queries::queries::generic_inference::first_party::minimax_h3::mark_first_party_minimax_h3_job_succeeded::{
+  mark_first_party_minimax_h3_job_succeeded, MarkFirstPartyMinimaxH3JobSucceededArgs,
+};
+use mysql_queries::queries::media_files::create::insert_builder::media_file_insert_builder::MediaFileInsertBuilder;
+use thumbnail_generator::task_client::thumbnail_task::{ThumbnailTaskBuilder, ThumbnailTaskInputMimeType};
+use tokens::tokens::generic_inference_jobs::InferenceJobToken;
+use tokens::tokens::media_files::MediaFileToken;
+
+use crate::http_server::common_responses::common_web_error::CommonWebError;
+use crate::http_server::web_utils::require_internal_api_key::require_internal_api_key;
+use crate::state::server_state::ServerState;
+
+const BUCKET_PATH_PREFIX: Option<&str> = Some("artcraft_");
+
+#[derive(Deserialize)]
+pub struct MarkMinimaxJobSuccessPathInfo {
+  job_token: InferenceJobToken,
+}
+
+/// Form-multipart request fields for the minimax job success endpoint.
+#[derive(MultipartForm, ToSchema)]
+#[multipart(duplicate_field = "deny")]
+pub struct MarkMinimaxJobSuccessForm {
+  /// The generated video file bytes.
+  #[multipart(limit = "1 GiB")]
+  #[schema(value_type = Vec<u8>, format = Binary)]
+  file: TempFile,
+
+  /// Optional: video duration in milliseconds. Probed from the file if absent.
+  #[multipart(limit = "2 KiB")]
+  #[schema(value_type = Option<u64>, format = Binary)]
+  duration_millis: Option<Text<u64>>,
+
+  /// Optional: video frame width in pixels. Probed from the file if absent.
+  #[multipart(limit = "2 KiB")]
+  #[schema(value_type = Option<u32>, format = Binary)]
+  width: Option<Text<u32>>,
+
+  /// Optional: video frame height in pixels. Probed from the file if absent.
+  #[multipart(limit = "2 KiB")]
+  #[schema(value_type = Option<u32>, format = Binary)]
+  height: Option<Text<u32>>,
+
+  /// Optional: total wall-clock runtime of the job, in milliseconds.
+  #[multipart(limit = "2 KiB")]
+  #[schema(value_type = Option<u64>, format = Binary)]
+  execution_duration_millis: Option<Text<u64>>,
+
+  /// Optional: inference-only runtime of the job, in milliseconds.
+  #[multipart(limit = "2 KiB")]
+  #[schema(value_type = Option<u64>, format = Binary)]
+  inference_duration_millis: Option<Text<u64>>,
+}
+
+/// Internal (worker-facing): mark a first-party Minimax job as successful and
+/// upload the generated video. Mirrors what the Fal webhook does when videos
+/// complete: upload to the public bucket, create the media file record, fire
+/// thumbnail tasks, then mark the job `complete_success`.
+#[utoipa::path(
+  post,
+  tag = "Internal (Minimax Jobs)",
+  path = "/v1/internal/minimax_jobs/job/{job_token}/success",
+  params(
+    ("job_token" = String, Path, description = "The inference job token"),
+    (
+      "request" = MarkMinimaxJobSuccessForm,
+      description = "Multipart form: `file` (video bytes) plus optional `duration_millis` / `width` / `height` metadata."
+    ),
+  ),
+  responses(
+    (status = 200, description = "Success", body = MarkMinimaxJobSuccessResponse),
+    (status = 400, description = "Bad input (unreadable or non-video file)"),
+    (status = 401, description = "Missing or invalid internal API key"),
+    (status = 404, description = "No such minimax job"),
+    (status = 500, description = "Server error"),
+  ),
+)]
+pub async fn mark_minimax_job_success_handler(
+  http_request: HttpRequest,
+  path: web::Path<MarkMinimaxJobSuccessPathInfo>,
+  MultipartForm(mut form): MultipartForm<MarkMinimaxJobSuccessForm>,
+  server_state: web::Data<Arc<ServerState>>,
+) -> Result<Json<MarkMinimaxJobSuccessResponse>, CommonWebError> {
+
+  require_internal_api_key(&http_request, &server_state)?;
+
+  let job_token = &path.job_token;
+
+  // ==================== LOOK UP JOB ==================== //
+
+  // NB: The connection is scoped so we don't hold a pool slot across the
+  // (slow) bucket upload below.
+  let maybe_job = {
+    let mut mysql_connection = server_state.mysql_pool.acquire().await?;
+
+    get_first_party_minimax_h3_job_by_token(GetFirstPartyMinimaxH3JobByTokenArgs {
+      job_token,
+      mysql_executor: &mut *mysql_connection,
+      phantom: Default::default(),
+    }).await.map_err(|err| {
+      error!("Error looking up minimax job {}: {:?}", job_token, err);
+      CommonWebError::from_error(err)
+    })?
+  };
+
+  let Some(job) = maybe_job else {
+    warn!("No such minimax job: {}", job_token);
+    return Err(CommonWebError::NotFound);
+  };
+
+  // ==================== IDEMPOTENCY: TERMINAL STATUS CHECK ==================== //
+
+  match job.status {
+    JobStatusPlus::CompleteSuccess => {
+      // A duplicate / retried success report. Don't re-upload or create a
+      // second media file — return the already-recorded result.
+      let Some(existing_media_token) = job.maybe_result_media_file_token else {
+        error!("Minimax job {} is complete_success but has no result media file token", job.job_token);
+        return Err(CommonWebError::server_error_with_message(
+          "job already complete but result media file is missing"));
+      };
+
+      info!("Minimax job {} is already complete_success; ignoring duplicate success report", job.job_token);
+
+      return Ok(Json(MarkMinimaxJobSuccessResponse {
+        success: true,
+        media_file_token: existing_media_token,
+        maybe_duration_millis: None,
+        maybe_width: None,
+        maybe_height: None,
+      }));
+    }
+    JobStatusPlus::CompleteFailure => {
+      warn!("Minimax job {} is already complete_failure; rejecting success report", job.job_token);
+      return Err(CommonWebError::BadInputWithSimpleMessage(
+        "job is already in a terminal state (complete_failure)".to_string()));
+    }
+    _ => {} // Not terminal; proceed.
+  }
+
+  // ==================== READ + VALIDATE FILE ==================== //
+
+  // NB: No database connection is held anywhere in this section — the byte
+  // read, hashing, ffprobe, and bucket upload below are all slow, and
+  // holding a pool slot across them is what causes PoolTimedOut incidents.
+
+  let mut file_bytes = Vec::new();
+  form.file.file.read_to_end(&mut file_bytes)
+    .map_err(|err| {
+      error!("Problem reading uploaded video file: {:?}", err);
+      CommonWebError::server_error_with_message("problem reading uploaded video file")
+    })?;
+
+  if file_bytes.is_empty() {
+    return Err(CommonWebError::BadInputWithSimpleMessage("empty video file".to_string()));
+  }
+
+  let mime_type = get_mimetype_for_bytes(&file_bytes)
+    .map(|mime_type| mime_type.to_string())
+    .ok_or_else(|| {
+      warn!("Could not determine mimetype for uploaded video ({} bytes)", file_bytes.len());
+      CommonWebError::BadInputWithSimpleMessage("could not determine video mimetype".to_string())
+    })?;
+
+  let media_file_type = MediaFileType::try_from_mime_type(&mime_type)
+    .ok_or_else(|| {
+      warn!("Unsupported media file type: {}", mime_type);
+      CommonWebError::BadInputWithSimpleMessage("unsupported media file type".to_string())
+    })?;
+
+  let extension_with_period = mimetype_to_extension(&mime_type)
+    .map(|extension| format!(".{extension}"))
+    .ok_or_else(|| {
+      warn!("Could not determine file extension for mimetype: {}", mime_type);
+      CommonWebError::server_error_with_message("could not determine file extension")
+    })?;
+
+  let file_size_bytes = file_bytes.len();
+  let file_hash = sha256_hash_bytes(&file_bytes)
+    .map_err(|err| {
+      error!("Failed to hash video bytes: {:?}", err);
+      CommonWebError::from_anyhow_error(err)
+    })?;
+
+  // ==================== VIDEO METADATA ==================== //
+
+  // Prefer worker-supplied metadata; probe the file for anything missing.
+  let mut maybe_duration_millis = form.duration_millis.map(|text| text.0);
+  let mut maybe_frame_width = form.width.map(|text| text.0);
+  let mut maybe_frame_height = form.height.map(|text| text.0);
+
+  let needs_probe = maybe_duration_millis.is_none()
+    || maybe_frame_width.is_none()
+    || maybe_frame_height.is_none();
+
+  if needs_probe {
+    if let Ok(video_info) = ffprobe_get_info(&form.file.file.path()) {
+      if maybe_duration_millis.is_none() {
+        maybe_duration_millis = video_info.duration
+          .as_ref()
+          .map(|duration| duration.millis as u64);
+      }
+      if maybe_frame_width.is_none() {
+        maybe_frame_width = video_info.dimensions
+          .as_ref()
+          .map(|dims| dims.width as u32);
+      }
+      if maybe_frame_height.is_none() {
+        maybe_frame_height = video_info.dimensions
+          .as_ref()
+          .map(|dims| dims.height as u32);
+      }
+    }
+  }
+
+  // ==================== UPLOAD + MEDIA FILE RECORD ==================== //
+
+  let public_upload_path = MediaFileBucketPath::generate_new(BUCKET_PATH_PREFIX, Some(&extension_with_period));
+
+  info!("Uploading media to bucket path: {}", public_upload_path.get_full_object_path_str());
+
+  server_state.public_bucket_client.upload_file_with_content_type_process(
+    public_upload_path.get_full_object_path_str(),
+    file_bytes.as_ref(),
+    &mime_type)
+      .await
+      .map_err(|err| {
+        warn!("Failed to upload video to bucket: {:?}", err);
+        CommonWebError::from_anyhow_error(err)
+      })?;
+
+  // ==================== MEDIA FILE RECORD + MARK JOB DONE ==================== //
+
+  // Tight critical section: one connection for both post-upload writes, no
+  // network calls in between.
+  let media_token = {
+    let mut mysql_connection = server_state.mysql_pool.acquire().await?;
+
+    let media_token = insert_media_file_record(
+      &mut mysql_connection,
+      &job,
+      &file_hash,
+      file_size_bytes,
+      maybe_duration_millis,
+      maybe_frame_width,
+      maybe_frame_height,
+      media_file_type,
+      &mime_type,
+      &public_upload_path,
+    ).await?;
+
+    info!("Minimax video media file uploaded with token: {}", media_token);
+
+    mark_first_party_minimax_h3_job_succeeded(MarkFirstPartyMinimaxH3JobSucceededArgs {
+      job_token: &job.job_token,
+      media_file_token: &media_token,
+      maybe_execution_duration_millis: form.execution_duration_millis.map(|text| text.0),
+      maybe_inference_duration_millis: form.inference_duration_millis.map(|text| text.0),
+      mysql_executor: &mut *mysql_connection,
+      phantom: Default::default(),
+    }).await.map_err(|err| {
+      error!("Error marking minimax job {} succeeded: {:?}", job_token, err);
+      CommonWebError::from_error(err)
+    })?;
+
+    media_token
+  };
+
+  // ==================== THUMBNAILS ==================== //
+
+  let thumbnail_task_result =
+      ThumbnailTaskBuilder::new_for_source_mimetype(ThumbnailTaskInputMimeType::MP4)
+          .with_bucket(server_state.public_bucket_client.bucket_name().as_str())
+          .with_path(&*path_to_string(public_upload_path.to_full_object_pathbuf()))
+          .with_output_suffix("thumb")
+          .with_event_id(&media_token.to_string())
+          .send_all()
+          .await;
+
+  if let Err(err) = thumbnail_task_result {
+    // Fail open
+    error!("Failed to create some/all thumbnail tasks: {:?}", err);
+  }
+
+  info!("Minimax job {} marked as complete_success with media file {}", job.job_token, media_token);
+
+  Ok(Json(MarkMinimaxJobSuccessResponse {
+    success: true,
+    media_file_token: media_token,
+    maybe_duration_millis,
+    maybe_width: maybe_frame_width,
+    maybe_height: maybe_frame_height,
+  }))
+}
+
+async fn insert_media_file_record(
+  mysql_connection: &mut PoolConnection<MySql>,
+  job: &FirstPartyMinimaxH3JobDetails,
+  file_hash: &str,
+  file_size_bytes: usize,
+  maybe_duration_millis: Option<u64>,
+  maybe_frame_width: Option<u32>,
+  maybe_frame_height: Option<u32>,
+  media_file_type: MediaFileType,
+  mime_type: &str,
+  public_upload_path: &MediaFileBucketPath,
+) -> Result<MediaFileToken, CommonWebError> {
+  MediaFileInsertBuilder::new()
+      .checksum_sha2(file_hash)
+      .creator_ip_address(&job.creator_ip_address)
+      .file_size_bytes(file_size_bytes as u64)
+      .maybe_creator_anonymous_visitor(job.maybe_creator_anonymous_visitor_token.as_ref())
+      .maybe_creator_user(job.maybe_creator_user_token.as_ref())
+      .maybe_duration_millis(maybe_duration_millis)
+      .maybe_frame_height(maybe_frame_height)
+      .maybe_frame_width(maybe_frame_width)
+      .maybe_generation_provider(Some(GenerationProvider::Artcraft))
+      .maybe_prompt_token(job.maybe_prompt_token.as_ref())
+      .maybe_source_job_token(Some(&job.job_token))
+      .maybe_platform_type(job.maybe_platform_type)
+      .media_file_class(MediaFileClass::Video)
+      .media_file_origin_category(MediaFileOriginCategory::Inference)
+      .media_file_type(media_file_type)
+      .mime_type(mime_type)
+      .public_bucket_directory_hash(public_upload_path)
+      .insert_connection(mysql_connection)
+      .await
+      .map_err(|err| {
+        warn!("Failed to insert video media file record: {:?}", err);
+        CommonWebError::from_error(err)
+      })
+}
@@ -0,0 +1,3 @@
+pub mod mark_minimax_job_failure_handler;
+pub mod mark_minimax_job_success_handler;
+pub mod obtain_minimax_job_handler;
@@ -0,0 +1,195 @@
+use std::sync::Arc;
+
+use actix_web::web::Json;
+use actix_web::{web, HttpRequest};
+use log::{error, info, warn};
+use sqlx::pool::PoolConnection;
+use sqlx::{Acquire, MySql};
+
+use artcraft_api_defs::internal::minimax_jobs::minimax_worker_model::MinimaxWorkerModel;
+use artcraft_api_defs::internal::minimax_jobs::obtain_minimax_job::{
+  MinimaxJobMediaReference, MinimaxJobPromptDetails, ObtainMinimaxJobRequest, ObtainMinimaxJobResponse,
+  ObtainedMinimaxJob,
+};
+use bucket_paths::legacy::typified_paths::public::media_files::bucket_file_path::MediaFileBucketPath;
+use mysql_queries::queries::generic_inference::first_party::minimax_h3::first_party_minimax_h3_model::FirstPartyMinimaxH3Model;
+use mysql_queries::queries::generic_inference::first_party::minimax_h3::mark_first_party_minimax_h3_job_started::{
+  mark_first_party_minimax_h3_job_started, MarkFirstPartyMinimaxH3JobStartedArgs,
+};
+use mysql_queries::queries::generic_inference::first_party::minimax_h3::select_pending_first_party_minimax_h3_job_for_update::{
+  select_pending_first_party_minimax_h3_job_for_update, PendingFirstPartyMinimaxH3Job,
+  SelectPendingFirstPartyMinimaxH3JobForUpdateArgs,
+};
+use mysql_queries::queries::prompt_context_items::list_prompt_context_items::list_prompt_context_items;
+use mysql_queries::queries::prompts::get_prompt::get_prompt_from_connection;
+
+use crate::http_server::common_responses::common_web_error::CommonWebError;
+use crate::http_server::common_responses::media::media_links_builder::MediaLinksBuilder;
+use crate::http_server::endpoints::media_files::helpers::get_media_domain::get_media_domain;
+use crate::http_server::web_utils::require_internal_api_key::require_internal_api_key;
+use crate::state::server_state::ServerState;
+
+/// Internal (worker-facing): obtain a lock on a pending first-party Minimax
+/// job by atomically moving it from `pending` to `started`. The response
+/// carries everything the worker needs to run inference: the text prompt,
+/// generation settings, and full CDN links to every media reference.
+#[utoipa::path(
+  post,
+  tag = "Internal (Minimax Jobs)",
+  path = "/v1/internal/minimax_jobs/obtain_job",
+  request_body = ObtainMinimaxJobRequest,
+  responses(
+    (status = 200, description = "Success (maybe_job is null when no pending job exists)", body = ObtainMinimaxJobResponse),
+    (status = 401, description = "Missing or invalid internal API key"),
+    (status = 500, description = "Server error"),
+  ),
+)]
+pub async fn obtain_minimax_job_handler(
+  http_request: HttpRequest,
+  request: Json<ObtainMinimaxJobRequest>,
+  server_state: web::Data<Arc<ServerState>>,
+) -> Result<Json<ObtainMinimaxJobResponse>, CommonWebError> {
+
+  require_internal_api_key(&http_request, &server_state)?;
+
+  let minimax_model = match request.model {
+    MinimaxWorkerModel::MinimaxH3Turbo => FirstPartyMinimaxH3Model::Turbo,
+    MinimaxWorkerModel::MinimaxH3Ultra => FirstPartyMinimaxH3Model::Ultra,
+  };
+
+  let mut mysql_connection = server_state.mysql_pool.acquire().await?;
+
+  // The select-then-mark pair runs in one transaction: `FOR UPDATE SKIP
+  // LOCKED` guarantees two workers can never obtain the same job.
+  let mut transaction = mysql_connection.begin().await.map_err(|err| {
+    error!("Error starting MySQL transaction: {:?}", err);
+    CommonWebError::from_error(err)
+  })?;
+
+  let maybe_pending_job = select_pending_first_party_minimax_h3_job_for_update(
+    SelectPendingFirstPartyMinimaxH3JobForUpdateArgs {
+      minimax_model,
+      transaction: &mut transaction,
+    }
+  ).await.map_err(|err| {
+    error!("Error selecting pending minimax job: {:?}", err);
+    CommonWebError::from_error(err)
+  })?;
+
+  let Some(pending_job) = maybe_pending_job else {
+    return Ok(Json(ObtainMinimaxJobResponse {
+      success: true,
+      maybe_job: None,
+    }));
+  };
+
+  mark_first_party_minimax_h3_job_started(MarkFirstPartyMinimaxH3JobStartedArgs {
+    job_token: &pending_job.job_token,
+    maybe_worker_hostname: request.worker_hostname.as_deref(),
+    maybe_cluster_name: request.cluster_name.as_deref(),
+    transaction: &mut transaction,
+  }).await.map_err(|err| {
+    error!("Error marking minimax job {} started: {:?}", pending_job.job_token, err);
+    CommonWebError::from_error(err)
+  })?;
+
+  transaction.commit().await.map_err(|err| {
+    error!("Error committing transaction: {:?}", err);
+    CommonWebError::from_error(err)
+  })?;
+
+  info!("Worker {:?} (cluster {:?}) obtained minimax job {}",
+    request.worker_hostname, request.cluster_name, pending_job.job_token);
+
+  // Hydrate the prompt + media references the worker needs. Best-effort: the
+  // job is already locked, so a hydration failure yields `maybe_prompt: None`
+  // and the worker should report the job failed rather than us wedging it.
+  let maybe_prompt = build_prompt_details(
+    &http_request,
+    &server_state,
+    &pending_job,
+    &mut mysql_connection,
+  ).await;
+
+  Ok(Json(ObtainMinimaxJobResponse {
+    success: true,
+    maybe_job: Some(ObtainedMinimaxJob {
+      job_token: pending_job.job_token,
+      model: request.model,
+      maybe_prompt,
+    }),
+  }))
+}
+
+/// Fetch the prompt record and its media references, resolving each media
+/// file to full CDN links.
+async fn build_prompt_details(
+  http_request: &HttpRequest,
+  server_state: &ServerState,
+  pending_job: &PendingFirstPartyMinimaxH3Job,
+  mysql_connection: &mut PoolConnection<MySql>,
+) -> Option<MinimaxJobPromptDetails> {
+  let Some(prompt_token) = pending_job.maybe_prompt_token.as_ref() else {
+    warn!("Minimax job {} has no prompt token", pending_job.job_token);
+    return None;
+  };
+
+  let prompt = match get_prompt_from_connection(prompt_token, mysql_connection).await {
+    Ok(Some(prompt)) => prompt,
+    Ok(None) => {
+      warn!("Prompt {} for minimax job {} not found", prompt_token, pending_job.job_token);
+      return None;
+    }
+    Err(err) => {
+      error!("Error fetching prompt {} for minimax job {}: {:?}",
+        prompt_token, pending_job.job_token, err);
+      return None;
+    }
+  };
+
+  let context_items = match list_prompt_context_items(prompt_token, mysql_connection).await {
+    Ok(items) => items,
+    Err(err) => {
+      // Don't hand the worker a prompt with silently-missing references.
+      error!("Error listing prompt context items for {} (minimax job {}): {:?}",
+        prompt_token, pending_job.job_token, err);
+      return None;
+    }
+  };
+
+  let media_domain = get_media_domain(http_request);
+
+  let media_references = context_items
+    .into_iter()
+    .map(|item| {
+      let bucket_path = MediaFileBucketPath::from_object_hash(
+        &item.public_bucket_directory_hash,
+        item.maybe_public_bucket_prefix.as_deref(),
+        item.maybe_public_bucket_extension.as_deref());
+
+      let media_links = MediaLinksBuilder::from_media_path_and_env(
+        media_domain,
+        server_state.server_environment,
+        &bucket_path);
+
+      MinimaxJobMediaReference {
+        media_file_token: item.media_token,
+        semantic_type: item.context_semantic_type,
+        media_links,
+      }
+    })
+    .collect();
+
+  Some(MinimaxJobPromptDetails {
+    prompt_token: prompt.token,
+    maybe_positive_prompt: prompt.maybe_positive_prompt,
+    maybe_negative_prompt: prompt.maybe_negative_prompt,
+    maybe_generation_mode: prompt.maybe_generation_mode,
+    maybe_aspect_ratio: prompt.maybe_aspect_ratio,
+    maybe_resolution: prompt.maybe_resolution,
+    maybe_bitrate: prompt.maybe_bitrate,
+    maybe_generate_audio: prompt.maybe_generate_audio,
+    maybe_duration_seconds: prompt.maybe_duration_seconds,
+    media_references,
+  })
+}
@@ -0,0 +1,5 @@
+//! Internal-facing endpoints for our own worker fleets (GPU inference).
+//! Authenticated with internal API keys (`INTERNAL_API_KEYS`),
+//! never consumer credentials.
+
+pub mod minimax_jobs;
@@ -13,6 +13,7 @@ pub mod folders;
 pub mod generate;
 pub mod image_studio;
 pub mod inference_job;
+pub mod internal;
 pub mod media_files;
 pub mod misc;
 pub mod model_download;
@@ -17,9 +17,9 @@ use enums::common::generation::common_model_type::CommonModelType;
 use enums::common::generation::common_video_model::CommonVideoModel;
 use enums::common::platform_type::PlatformType;
 use enums::common::visibility::Visibility;
+use mysql_queries::queries::generic_inference::first_party::minimax_h3::first_party_minimax_h3_model::FirstPartyMinimaxH3Model;
 use mysql_queries::queries::generic_inference::first_party::minimax_h3::insert_generic_inference_job_for_first_party_minimax_h3_with_apriori_job_token::{
  insert_generic_inference_job_for_first_party_minimax_h3_with_apriori_job_token,
-  FirstPartyMinimaxH3Model,
  InsertGenericInferenceForFirstPartyMinimaxH3WithAprioriJobTokenArgs,
 };
 use tokens::tokens::anonymous_visitor_tracking::AnonymousVisitorTrackingToken;
@@ -96,6 +96,14 @@ pub async fn write_prompt_records(args: WritePromptRecordsArgs<'_, '_>) -> Optio
        });
      }
    }
+    if let Some(ref_tokens) = &request.reference_audio_media_tokens {
+      for media_token in ref_tokens {
+        context_items.push(PromptContextItem {
+          media_token: media_token.clone(),
+          context_semantic_type: PromptContextSemanticType::Audioref,
+        });
+      }
+    }

    if !context_items.is_empty() {
      if let Err(err) = insert_batch_prompt_context_items(InsertBatchArgs {
@@ -8,6 +8,7 @@ use crate::http_server::routes::application_routes::dev_routes::add_dev_routes;
 use crate::http_server::routes::application_routes::featured_item_routes::add_featured_item_routes;
 use crate::http_server::routes::application_routes::folders_routes::add_folders_routes;
 use crate::http_server::routes::application_routes::generate_routes::add_generate_routes;
+use crate::http_server::routes::application_routes::internal_routes::add_internal_routes;
 use crate::http_server::routes::application_routes::job_routes::add_job_routes;
 use crate::http_server::routes::application_routes::media_files_routes::add_media_file_routes;
 use crate::http_server::routes::application_routes::moderation_routes::add_moderator_routes;
@@ -53,6 +54,7 @@ where
  app = add_omni_api_routes(app); // /v1/omni_api/... (API-key authenticated generate)
  app = add_character_routes(app); // /v1/character/... and /v1/characters/...
  app = add_webhook_routes(app); // /v1/webhooks/... (fal)
+  app = add_internal_routes(app); // /v1/internal/... (internal API key authenticated worker fleets)
  
  // Legacy FakeYou surface area that might be useful again one day
  app = add_weights_routes(app); // v1/weights
@@ -0,0 +1,35 @@
+use actix_http::body::MessageBody;
+use actix_service::ServiceFactory;
+use actix_web::dev::{ServiceRequest, ServiceResponse};
+use actix_web::{web, App, Error};
+
+use crate::http_server::endpoints::internal::minimax_jobs::mark_minimax_job_failure_handler::mark_minimax_job_failure_handler;
+use crate::http_server::endpoints::internal::minimax_jobs::mark_minimax_job_success_handler::mark_minimax_job_success_handler;
+use crate::http_server::endpoints::internal::minimax_jobs::obtain_minimax_job_handler::obtain_minimax_job_handler;
+
+/// Internal-facing routes for our own worker fleets (GPU inference).
+/// Every endpoint requires an internal API key (`INTERNAL_API_KEYS`).
+pub fn add_internal_routes<T, B>(app: App<T>) -> App<T>
+where
+    B: MessageBody,
+    T: ServiceFactory<
+      ServiceRequest,
+      Config = (),
+      Response = ServiceResponse<B>,
+      Error = Error,
+      InitError = (),
+    >,
+{
+  app.service(web::scope("/v1/internal")
+      .service(web::scope("/minimax_jobs")
+          .service(web::resource("/obtain_job")
+              .route(web::post().to(obtain_minimax_job_handler))
+          )
+          .service(web::resource("/job/{job_token}/failure")
+              .route(web::post().to(mark_minimax_job_failure_handler))
+          )
+          .service(web::resource("/job/{job_token}/success")
+              .route(web::post().to(mark_minimax_job_success_handler))
+          )
+      ))
+}
@@ -8,6 +8,7 @@ mod dev_routes;
 mod featured_item_routes;
 mod folders_routes;
 mod generate_routes;
+mod internal_routes;
 mod job_routes;
 mod media_files_routes;
 mod moderation_routes;
@@ -8,6 +8,7 @@ pub mod job_keepalives;
 pub mod open_zip_archive;
 pub mod read_multipart_field_bytes;
 pub mod redis_rate_limiter;
+pub mod require_internal_api_key;
 pub mod response_error_helpers;
 pub mod response_success_helpers;
 pub mod scoped_temp_dir_creator;
@@ -0,0 +1,122 @@
+//! Auth guard for internal (worker-facing) endpoints.
+//!
+//! Workers send an internal API key in the `Authorization` header. The key
+//! MUST exactly match one of the keys loaded from
+//! `INTERNAL_API_KEYS` at startup or the request is rejected with
+//! 401. This is entirely separate from consumer API key auth.
+
+use actix_web::HttpRequest;
+use log::warn;
+
+use crate::http_server::common_responses::common_web_error::CommonWebError;
+use crate::state::server_state::ServerState;
+use crate::util::internal_api_key::InternalApiKey;
+
+/// Require a valid internal API key in the `Authorization` header.
+///
+/// Accepts `Bearer <key>`, `Key <key>`, and bare `<key>` header forms
+/// (matching the site-wide Authorization header conventions).
+pub fn require_internal_api_key(
+  http_request: &HttpRequest,
+  server_state: &ServerState,
+) -> Result<(), CommonWebError> {
+  let maybe_header_value = http_request.headers()
+    .get("authorization")
+    .and_then(|value| value.to_str().ok());
+
+  let Some(header_value) = maybe_header_value else {
+    warn!("Internal endpoint request without an Authorization header.");
+    return Err(CommonWebError::NotAuthorized);
+  };
+
+  let Some(candidate_key) = parse_authorization_header_value(header_value) else {
+    warn!("Internal endpoint request with an unusable Authorization header.");
+    return Err(CommonWebError::NotAuthorized);
+  };
+
+  if server_state.internal_api_keys.contains(&InternalApiKey::new(candidate_key)) {
+    Ok(())
+  } else {
+    warn!("Internal endpoint request with an unrecognized internal API key.");
+    Err(CommonWebError::NotAuthorized)
+  }
+}
+
+/// Extract the key from `Bearer <key>`, `Key <key>`, or bare `<key>` forms.
+/// Returns `None` for blank values, unsupported schemes, or scheme-only values.
+fn parse_authorization_header_value(header_value: &str) -> Option<&str> {
+  let trimmed = header_value.trim();
+  if trimmed.is_empty() {
+    return None;
+  }
+
+  let mut parts = trimmed.split_whitespace();
+  let first = parts.next()?;
+
+  match parts.next() {
+    // A single token is a bare key — unless it's a scheme name with no
+    // credential ("Bearer" / "Key" alone).
+    None => {
+      if is_scheme_name(first) {
+        None
+      } else {
+        Some(first)
+      }
+    }
+    Some(second) => {
+      // Two tokens: the first must be a supported scheme.
+      if parts.next().is_some() {
+        return None;
+      }
+      if is_scheme_name(first) {
+        Some(second)
+      } else {
+        None
+      }
+    }
+  }
+}
+
+fn is_scheme_name(token: &str) -> bool {
+  token.eq_ignore_ascii_case("bearer") || token.eq_ignore_ascii_case("key")
+}
+
+#[cfg(test)]
+mod tests {
+  use super::*;
+
+  mod header_value_parsing {
+    use super::*;
+
+    #[test]
+    fn bearer_scheme() {
+      assert_eq!(parse_authorization_header_value("Bearer foo_abc"), Some("foo_abc"));
+      assert_eq!(parse_authorization_header_value("bearer foo_abc"), Some("foo_abc"));
+    }
+
+    #[test]
+    fn key_scheme() {
+      assert_eq!(parse_authorization_header_value("Key foo_abc"), Some("foo_abc"));
+    }
+
+    #[test]
+    fn bare_key() {
+      assert_eq!(parse_authorization_header_value("foo_abc"), Some("foo_abc"));
+      assert_eq!(parse_authorization_header_value("  foo_abc  "), Some("foo_abc"));
+    }
+
+    #[test]
+    fn blank_or_scheme_only_is_none() {
+      assert_eq!(parse_authorization_header_value(""), None);
+      assert_eq!(parse_authorization_header_value("   "), None);
+      assert_eq!(parse_authorization_header_value("Bearer"), None);
+      assert_eq!(parse_authorization_header_value("Bearer  "), None);
+    }
+
+    #[test]
+    fn unsupported_scheme_is_none() {
+      assert_eq!(parse_authorization_header_value("Basic dXNlcjpwYXNz"), None);
+      assert_eq!(parse_authorization_header_value("abc def"), None);
+    }
+  }
+}
@@ -34,6 +34,7 @@ use crate::startup::setup_bans::{
  load_static_container_ip_bans, load_troll_user_token_bans,
 };
 use crate::startup::setup_inference_providers::setup_inference_providers;
+use crate::startup::setup_internal_api_keys::setup_internal_api_keys;
 use crate::startup::setup_dashboards::setup_dashboards;
 use crate::startup::setup_seedance_video_bucket::setup_seedance_video_bucket;
 use crate::startup::setup_static_feature_flags::setup_static_feature_flags;
@@ -235,6 +236,7 @@ pub async fn setup_dependencies(server_hostname: &str) -> AnyhowResult<SetupResu
    sort_key_crypto,
    opaque_cursors: opaque_cursor_encoder,
    static_api_token_set,
+    internal_api_keys: setup_internal_api_keys(),
    inference_providers,
    resend: ResendData {
      api_key: resend_api_key,
@@ -6,6 +6,7 @@ pub mod setup_pager;
 pub mod setup_seedance_video_bucket;
 pub mod setup_disabled_endpoints;
 pub mod setup_inference_providers;
+pub mod setup_internal_api_keys;
 pub mod setup_static_feature_flags;
 pub mod setup_stripe_artcraft;
 pub mod setup_stripe_fakeyou;
@@ -0,0 +1,32 @@
+//! Load the accepted internal API keys (worker fleet auth) from the
+//! environment at startup.
+//!
+//! Missing or empty configuration is not fatal — the internal endpoints will
+//! simply reject every request until keys are configured.
+
+use std::collections::HashSet;
+use std::env;
+
+use log::{info, warn};
+
+use crate::util::internal_api_key::InternalApiKey;
+
+const INTERNAL_API_KEYS_ENV_VAR: &str = "INTERNAL_API_KEYS";
+
+pub fn setup_internal_api_keys() -> HashSet<InternalApiKey> {
+  let keys = match env::var(INTERNAL_API_KEYS_ENV_VAR) {
+    Ok(value) => InternalApiKey::parse_comma_separated_list(&value),
+    Err(_) => HashSet::new(),
+  };
+
+  if keys.is_empty() {
+    warn!(
+      "{} is unset or empty; internal worker endpoints will reject all requests.",
+      INTERNAL_API_KEYS_ENV_VAR,
+    );
+  } else {
+    info!("Loaded {} accepted internal API key(s).", keys.len());
+  }
+
+  keys
+}
@@ -1,3 +1,5 @@
+use std::collections::HashSet;
+
 use crate::configs::app_startup::username_set::UsernameSet;
 use crate::configs::static_api_tokens::StaticApiTokenSet;
 use crate::http_server::deprecated_endpoints::categories::tts::list_fully_computed_assigned_tts_categories::list_fully_computed_assigned_tts_categories::ModelTokensByCategoryToken;
@@ -12,6 +14,7 @@ use crate::state::flags::paging_flags::PagingFlags;
 use crate::state::memory_cache::model_token_to_info_cache::ModelTokenToInfoCache;
 use crate::threads::db_health_checker_thread::db_health_check_status::HealthCheckStatus;
 use crate::http_server::web_utils::web_sort_key_crypto::WebSortKeyCrypto;
+use crate::util::internal_api_key::InternalApiKey;
 use crate::util::troll_user_bans::troll_user_ban_list::TrollUserBanList;

 use actix_artcraft::sessions::anonymous_visitor_tracking::avt_cookie_manager::AvtCookieManager;
@@ -109,6 +112,10 @@ pub struct ServerState {

  pub static_api_token_set: StaticApiTokenSet,

+  /// Accepted internal API keys for our own worker fleets (GPU inference).
+  /// Loaded from `INTERNAL_API_KEYS` at startup; empty when unset.
+  pub internal_api_keys: HashSet<InternalApiKey>,
+
  pub caches: InMemoryCaches,

  pub google_sign_in_cert: GoogleSignInCert,
@@ -0,0 +1,65 @@
+//! Internal-facing API keys for our own worker fleets (GPU inference, etc.).
+//!
+//! These are configured as an env secret (`INTERNAL_API_KEYS`) at app
+//! startup and are completely distinct from consumer-facing
+//! `artcraft_api_keys` infrastructure — do not mix the two.
+
+use std::collections::HashSet;
+use std::fmt;
+
+/// An accepted internal API key.
+#[derive(Clone, Eq, PartialEq, Hash)]
+pub struct InternalApiKey(String);
+
+impl InternalApiKey {
+  pub fn new(key: impl Into<String>) -> Self {
+    Self(key.into())
+  }
+
+  pub fn as_str(&self) -> &str {
+    &self.0
+  }
+
+  /// Parse a comma-separated key list (the `INTERNAL_API_KEYS` env
+  /// var format). Entries are trimmed; blank entries are dropped.
+  pub fn parse_comma_separated_list(value: &str) -> HashSet<InternalApiKey> {
+    value
+      .split(',')
+      .map(str::trim)
+      .filter(|entry| !entry.is_empty())
+      .map(InternalApiKey::new)
+      .collect()
+  }
+}
+
+/// Redacted: keys are secrets and must never leak into logs.
+impl fmt::Debug for InternalApiKey {
+  fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+    write!(f, "InternalApiKey(***)")
+  }
+}
+
+#[cfg(test)]
+mod tests {
+  use super::*;
+
+  #[test]
+  fn parses_comma_separated_list_with_whitespace() {
+    let keys = InternalApiKey::parse_comma_separated_list(" foo_abc , bar_def,, ");
+    assert_eq!(keys.len(), 2);
+    assert!(keys.contains(&InternalApiKey::new("foo_abc")));
+    assert!(keys.contains(&InternalApiKey::new("bar_def")));
+  }
+
+  #[test]
+  fn empty_list_parses_to_empty_set() {
+    assert!(InternalApiKey::parse_comma_separated_list("").is_empty());
+    assert!(InternalApiKey::parse_comma_separated_list(" , ,").is_empty());
+  }
+
+  #[test]
+  fn debug_is_redacted() {
+    let key = InternalApiKey::new("foo_supersecret");
+    assert_eq!(format!("{:?}", key), "InternalApiKey(***)");
+  }
+}
@@ -10,6 +10,7 @@ pub mod enroll_in_studio;
 pub mod generate_api_key;
 pub mod http_download_url_to_bytes;
 pub mod http_download_url_to_tempfile;
+pub mod internal_api_key;
 pub mod lookup;
 pub mod placeholder_images;
 pub mod read_toml_file_to_struct;