Add Comfy digital human jobs

src/routes/digital_human.rs

@@ -1,26 +1,51 @@
 use crate::{
     auth::jwt::validate_jwt,
-    db::models::downstream_api_key::DownstreamApiKey,
+    db::models::{background_task::BackgroundTask, downstream_api_key::DownstreamApiKey},
     error::{AppError, AppResult},
+    services::workspace::{self, WorkspaceAssetInput},
+    workflow::storage,
     AppState,
 };
 use axum::{
-    extract::State,
-    http::{header, HeaderMap},
+    body::Bytes,
+    extract::{Multipart, Path, Query, State},
+    http::{header, HeaderMap, Method},
     routing::{get, post},
     Json, Router,
 };
+use serde::Deserialize;
 use serde_json::{json, Map, Value};
+use std::{collections::HashMap, time::Duration};
+
+const TASK_TYPE_DIGITAL_HUMAN: &str = "digital-human-generation";
+const DIGITAL_HUMAN_TIMEOUT_SECS: u64 = 900;
+const DIGITAL_HUMAN_POLL_INTERVAL_SECS: u64 = 5;
+const DIGITAL_HUMAN_MAX_POLLS: usize = 180;
 
 #[derive(Debug, Clone)]
 struct DigitalHumanAuth {
+    user_id: Option<i64>,
+    role: String,
     outbound_key: String,
 }
 
+#[derive(Debug, Deserialize)]
+struct JobListQuery {
+    limit: Option<i64>,
+}
+
 pub fn routes() -> Router<AppState> {
     Router::new()
         .route("/api/digital-human/status", get(status))
         .route("/api/digital-human/generate", post(generate))
+        .route(
+            "/api/digital-human/jobs",
+            get(list_jobs).post(create_job),
+        )
+        .route("/api/digital-human/jobs/clear", post(clear_jobs))
+        .route("/api/digital-human/jobs/{id}", get(get_job))
+        .route("/api/digital-human/jobs/{id}/cancel", post(cancel_job))
+        .route("/api/digital-human/upload", post(upload_asset))
 }
 
 fn local_v1_base(state: &AppState) -> String {
@@ -32,6 +57,15 @@ fn local_v1_base(state: &AppState) -> String {
     format!("http://{}:{}/v1", host, state.config.server.port)
 }
 
+fn public_base_url(state: &AppState) -> String {
+    let host = if state.config.server.host == "0.0.0.0" || state.config.server.host == "::" {
+        "127.0.0.1".to_string()
+    } else {
+        state.config.server.host.clone()
+    };
+    format!("http://{}:{}", host, state.config.server.port)
+}
+
 fn extract_bearer_token(headers: &HeaderMap) -> Option<&str> {
     headers
         .get(header::AUTHORIZATION)
@@ -72,6 +106,8 @@ async fn resolve_auth(headers: &HeaderMap, state: &AppState) -> Result<DigitalHu
 
     if token == state.config.auth.auth_token || token == state.config.auth.proxy_token {
         return Ok(DigitalHumanAuth {
+            user_id: None,
+            role: "admin".to_string(),
             outbound_key: state.config.auth.proxy_token.clone(),
         });
     }
@@ -88,9 +124,13 @@ async fn resolve_auth(headers: &HeaderMap, state: &AppState) -> Result<DigitalHu
         .map_err(|_| AppError::Unauthorized("Invalid user ID in token".into()))?;
     match user_downstream_key(state, user_id).await {
         Ok(key) => Ok(DigitalHumanAuth {
+            user_id: Some(user_id),
+            role: claims.role,
             outbound_key: key.key,
         }),
         Err(_) if claims.role == "admin" => Ok(DigitalHumanAuth {
+            user_id: Some(user_id),
+            role: claims.role,
             outbound_key: state.config.auth.proxy_token.clone(),
         }),
         Err(message) => Err(AppError::Forbidden(message)),
@@ -101,17 +141,22 @@ async fn status(State(state): State<AppState>, headers: HeaderMap) -> AppResult<
     let _ = resolve_auth(&headers, &state).await?;
     Ok(Json(json!({
         "success": true,
-        "mode": if state.config.comfyui.uses_runninghub_backend() { "runninghub-ai-app" } else { "boosapi-video-route" },
+        "mode": if state.config.comfyui.uses_native_backend() { "comfy-native" } else { "boosapi-video-route" },
         "runninghub": {
             "configured": state.config.comfyui.api_key.as_deref().map(str::trim).is_some_and(|value| !value.is_empty()),
             "endpoint": "https://www.runninghub.cn/task/openapi/ai-app/run"
         },
+        "comfy": {
+            "backend": format!("{:?}", state.config.comfyui.backend).to_ascii_lowercase(),
+            "configured": state.config.comfyui.uses_native_backend(),
+            "baseUrl": state.config.comfyui.target_base_url()
+        },
         "capabilities": [
-            "text_to_avatar_video",
-            "image_to_avatar_video",
-            "audio_driven_avatar",
-            "multi_person_prompt",
-            "segmented_long_video"
+            "product_explainer_video",
+            "image_or_video_avatar",
+            "audio_lip_sync",
+            "voice_reference",
+            "workspace_archive"
         ]
     })))
 }
@@ -295,7 +340,7 @@ async fn generate_runninghub(state: &AppState, body: Value) -> AppResult<Json<Va
     let api_key = runninghub_api_key(state, &body)?;
     let request_body = runninghub_request_body(&body, &api_key)?;
     let response = reqwest::Client::builder()
-        .timeout(std::time::Duration::from_secs(300))
+        .timeout(Duration::from_secs(300))
         .build()
         .unwrap_or_else(|_| reqwest::Client::new())
         .post("https://www.runninghub.cn/task/openapi/ai-app/run")
@@ -363,7 +408,7 @@ async fn generate(
         object.insert("task_type".to_string(), json!("digital_human"));
     }
     let response = reqwest::Client::builder()
-        .timeout(std::time::Duration::from_secs(300))
+        .timeout(Duration::from_secs(300))
         .build()
         .unwrap_or_else(|_| reqwest::Client::new())
         .post(format!("{}/videos", local_v1_base(&state)))
@@ -387,3 +432,821 @@ async fn generate(
         "response": value
     })))
 }
+
+fn scalar_string(value: Option<&Value>) -> Option<String> {
+    match value? {
+        Value::String(value) => Some(value.trim().to_string()).filter(|value| !value.is_empty()),
+        Value::Number(value) => Some(value.to_string()),
+        Value::Bool(value) => Some(value.to_string()),
+        _ => None,
+    }
+}
+
+fn require_string(body: &Value, keys: &[&str], message: &str) -> AppResult<String> {
+    for key in keys {
+        if let Some(value) = scalar_string(body.get(*key)) {
+            return Ok(value);
+        }
+    }
+    Err(AppError::BadRequest(message.into()))
+}
+
+fn optional_string(body: &Value, keys: &[&str]) -> Option<String> {
+    keys.iter().find_map(|key| scalar_string(body.get(*key)))
+}
+
+fn mapping_value<'a>(mapping: &'a Value, keys: &[&str]) -> Option<&'a Value> {
+    keys.iter().find_map(|key| mapping.get(*key))
+}
+
+fn mapping_node_id(mapping: &Value, keys: &[&str]) -> Option<String> {
+    mapping_value(mapping, keys).and_then(|value| {
+        if value.is_object() {
+            scalar_string(value.get("nodeId").or_else(|| value.get("node_id")).or(Some(value)))
+        } else {
+            scalar_string(Some(value))
+        }
+    })
+}
+
+fn mapping_field(mapping: &Value, keys: &[&str], default_field: &str) -> String {
+    mapping_value(mapping, keys)
+        .and_then(|value| {
+            if value.is_object() {
+                scalar_string(value.get("field").or_else(|| value.get("fieldName")))
+            } else {
+                None
+            }
+        })
+        .unwrap_or_else(|| default_field.to_string())
+}
+
+fn default_mapping() -> Value {
+    json!({
+        "prompt": { "nodeId": "", "field": "prompt" },
+        "script": { "nodeId": "", "field": "text" },
+        "avatar": { "nodeId": "", "field": "image" },
+        "referenceAudio": { "nodeId": "", "field": "audio" },
+        "voiceReference": { "nodeId": "", "field": "audio" },
+        "output": { "nodeId": "", "field": "video" }
+    })
+}
+
+fn workflow_mapping_from_body(body: &Value, workflow: &Value) -> Value {
+    body.get("nodeMapping")
+        .or_else(|| body.get("node_mapping"))
+        .cloned()
+        .or_else(|| workflow.pointer("/metadata/digitalHuman/nodeMapping").cloned())
+        .or_else(|| workflow.pointer("/digitalHuman/nodeMapping").cloned())
+        .unwrap_or_else(default_mapping)
+}
+
+fn apply_node_input(prompt: &mut Value, node_id: &str, field: &str, value: Value) -> AppResult<()> {
+    if node_id.trim().is_empty() {
+        return Err(AppError::BadRequest("node mapping contains empty nodeId".into()));
+    }
+    let Some(node) = prompt.get_mut(node_id) else {
+        return Err(AppError::BadRequest(format!(
+            "workflow prompt missing node {node_id}"
+        )));
+    };
+    let Some(inputs) = node.get_mut("inputs").and_then(Value::as_object_mut) else {
+        return Err(AppError::BadRequest(format!(
+            "workflow node {node_id} has no inputs"
+        )));
+    };
+    inputs.insert(field.to_string(), value);
+    Ok(())
+}
+
+fn workflow_prompt_from_graph(graph: &Value) -> Value {
+    graph.get("prompt")
+        .cloned()
+        .or_else(|| graph.get("workflow").cloned())
+        .or_else(|| graph.get("comfyPrompt").cloned())
+        .unwrap_or_else(|| graph.clone())
+}
+
+fn inject_digital_human_inputs(
+    prompt: &mut Value,
+    mapping: &Value,
+    request: &Value,
+) -> AppResult<String> {
+    let avatar_url = require_string(
+        request,
+        &["avatarUrl", "avatar_url", "imageUrl", "image_url", "videoUrl", "video_url"],
+        "人物图片或视频素材是必填项",
+    )?;
+    let product_script = require_string(
+        request,
+        &["script", "productScript", "product_script"],
+        "商品文案是必填项",
+    )?;
+    let reference_audio_url = require_string(
+        request,
+        &["referenceAudioUrl", "reference_audio_url", "audioUrl", "audio_url"],
+        "参考音频是必填项",
+    )?;
+    let audio_role = optional_string(request, &["audioRole", "audio_role"])
+        .unwrap_or_else(|| "both".to_string());
+    if !matches!(audio_role.as_str(), "lip_sync" | "voice_reference" | "both") {
+        return Err(AppError::BadRequest(
+            "audioRole must be lip_sync, voice_reference or both".into(),
+        ));
+    }
+
+    let prompt_text = optional_string(request, &["prompt"]).unwrap_or_else(|| {
+        format!("真人数字人主播讲解商品，口播自然，有带货节奏。\n商品文案：{product_script}")
+    });
+    let prompt_node = mapping_node_id(mapping, &["prompt", "promptNode", "prompt_node"])
+        .ok_or_else(|| AppError::BadRequest("缺少提示词节点映射".into()))?;
+    let script_node = mapping_node_id(mapping, &["script", "scriptNode", "script_node"])
+        .ok_or_else(|| AppError::BadRequest("缺少文案节点映射".into()))?;
+    let avatar_node = mapping_node_id(mapping, &["avatar", "image", "video", "avatarNode"])
+        .ok_or_else(|| AppError::BadRequest("缺少人物素材节点映射".into()))?;
+    let audio_node = mapping_node_id(
+        mapping,
+        &["referenceAudio", "audio", "lipSyncAudio", "audioNode"],
+    )
+    .ok_or_else(|| AppError::BadRequest("缺少参考音频节点映射".into()))?;
+
+    apply_node_input(
+        prompt,
+        &prompt_node,
+        &mapping_field(mapping, &["prompt", "promptNode", "prompt_node"], "prompt"),
+        json!(prompt_text),
+    )?;
+    apply_node_input(
+        prompt,
+        &script_node,
+        &mapping_field(mapping, &["script", "scriptNode", "script_node"], "text"),
+        json!(product_script),
+    )?;
+    apply_node_input(
+        prompt,
+        &avatar_node,
+        &mapping_field(mapping, &["avatar", "image", "video", "avatarNode"], "image"),
+        json!(avatar_url),
+    )?;
+    apply_node_input(
+        prompt,
+        &audio_node,
+        &mapping_field(
+            mapping,
+            &["referenceAudio", "audio", "lipSyncAudio", "audioNode"],
+            "audio",
+        ),
+        json!(reference_audio_url),
+    )?;
+    if matches!(audio_role.as_str(), "voice_reference" | "both") {
+        if let Some(voice_node) =
+            mapping_node_id(mapping, &["voiceReference", "voice", "voiceReferenceNode"])
+        {
+            apply_node_input(
+                prompt,
+                &voice_node,
+                &mapping_field(
+                    mapping,
+                    &["voiceReference", "voice", "voiceReferenceNode"],
+                    "audio",
+                ),
+                json!(reference_audio_url),
+            )?;
+        }
+    }
+    for (keys, request_keys, default_field) in [
+        (
+            &["ratio", "ratioNode"][..],
+            &["ratio", "aspectRatio", "aspect_ratio"][..],
+            "ratio",
+        ),
+        (&["resolution", "resolutionNode"][..], &["resolution"][..], "resolution"),
+        (&["subtitles", "subtitleNode"][..], &["subtitles", "subtitle"][..], "enabled"),
+    ] {
+        if let Some(node_id) = mapping_node_id(mapping, keys) {
+            if let Some(value) = request_keys.iter().find_map(|key| request.get(*key).cloned()) {
+                apply_node_input(prompt, &node_id, &mapping_field(mapping, keys, default_field), value)?;
+            }
+        }
+    }
+    Ok(product_script)
+}
+
+fn native_base(state: &AppState) -> String {
+    state.config.comfyui.target_base_url()
+}
+
+fn native_path(state: &AppState, path: &str) -> String {
+    if state.config.comfyui.uses_cloud_backend() {
+        format!(
+            "/api{}",
+            path.strip_prefix('/')
+                .map(|value| format!("/{value}"))
+                .unwrap_or_else(|| path.to_string())
+        )
+    } else {
+        path.to_string()
+    }
+}
+
+fn native_url(state: &AppState, path: &str) -> String {
+    format!("{}{}", native_base(state), native_path(state, path))
+}
+
+fn native_request(state: &AppState, method: Method, path: &str) -> reqwest::RequestBuilder {
+    let client = reqwest::Client::builder()
+        .timeout(Duration::from_secs(DIGITAL_HUMAN_TIMEOUT_SECS))
+        .build()
+        .unwrap_or_else(|_| reqwest::Client::new());
+    let reqwest_method =
+        reqwest::Method::from_bytes(method.as_str().as_bytes()).unwrap_or(reqwest::Method::GET);
+    let mut request = client.request(reqwest_method, native_url(state, path));
+    if let Some(api_key) = state
+        .config
+        .comfyui
+        .api_key
+        .as_deref()
+        .map(str::trim)
+        .filter(|value| !value.is_empty())
+    {
+        request = request
+            .header("X-API-Key", api_key)
+            .header("Authorization", format!("Bearer {api_key}"));
+    }
+    request
+}
+
+async fn native_json(
+    state: &AppState,
+    method: Method,
+    path: &str,
+    body: Option<Value>,
+) -> AppResult<Value> {
+    let mut request = native_request(state, method, path);
+    if let Some(body) = body {
+        request = request.json(&body);
+    }
+    let response = request
+        .send()
+        .await
+        .map_err(|err| AppError::UpstreamError(format!("Comfy backend unreachable: {err}")))?;
+    let status = response.status();
+    let text = response
+        .text()
+        .await
+        .map_err(|err| AppError::UpstreamError(err.to_string()))?;
+    let value = serde_json::from_str::<Value>(&text).unwrap_or_else(|_| json!({ "body": text }));
+    if !status.is_success() {
+        return Err(AppError::UpstreamError(value.to_string()));
+    }
+    Ok(value)
+}
+
+async fn submit_prompt(state: &AppState, prompt: Value) -> AppResult<(String, Value)> {
+    let body = json!({
+        "client_id": format!("boosapi-digital-human-{}", uuid::Uuid::new_v4().simple()),
+        "prompt": prompt,
+    });
+    let value = native_json(state, Method::POST, "/prompt", Some(body)).await?;
+    let prompt_id = scalar_string(
+        value
+            .get("prompt_id")
+            .or_else(|| value.get("promptId"))
+            .or_else(|| value.pointer("/data/prompt_id"))
+            .or_else(|| value.pointer("/data/promptId")),
+    )
+    .ok_or_else(|| AppError::UpstreamError(format!("Comfy response missing prompt_id: {value}")))?;
+    Ok((prompt_id, value))
+}
+
+async fn prompt_history(state: &AppState, prompt_id: &str) -> AppResult<Value> {
+    native_json(state, Method::GET, &format!("/history/{prompt_id}"), None).await
+}
+
+fn output_objects(value: &Value) -> Vec<Value> {
+    let mut out = Vec::new();
+    fn walk(value: &Value, out: &mut Vec<Value>) {
+        match value {
+            Value::Object(object) => {
+                for key in ["videos", "gifs", "images", "audio"] {
+                    if let Some(items) = object.get(key).and_then(Value::as_array) {
+                        out.extend(items.iter().filter(|item| item.is_object()).cloned());
+                    }
+                }
+                for value in object.values() {
+                    walk(value, out);
+                }
+            }
+            Value::Array(items) => {
+                for item in items {
+                    walk(item, out);
+                }
+            }
+            _ => {}
+        }
+    }
+    walk(value, &mut out);
+    out
+}
+
+fn output_kind(item: &Value) -> &'static str {
+    let filename = scalar_string(item.get("filename")).unwrap_or_default();
+    let lower = filename.to_ascii_lowercase();
+    if lower.ends_with(".mp4") || lower.ends_with(".webm") || lower.ends_with(".mov") {
+        "video"
+    } else if lower.ends_with(".mp3") || lower.ends_with(".wav") || lower.ends_with(".m4a") {
+        "audio"
+    } else {
+        "image"
+    }
+}
+
+fn output_query(item: &Value) -> HashMap<String, String> {
+    let mut params = HashMap::new();
+    if let Some(filename) = scalar_string(item.get("filename")) {
+        params.insert("filename".to_string(), filename);
+    }
+    if let Some(file_type) = scalar_string(item.get("type")) {
+        params.insert("type".to_string(), file_type);
+    } else {
+        params.insert("type".to_string(), "output".to_string());
+    }
+    if let Some(subfolder) = scalar_string(item.get("subfolder")) {
+        params.insert("subfolder".to_string(), subfolder);
+    }
+    params
+}
+
+fn encode(value: &str) -> String {
+    let mut out = String::new();
+    for byte in value.bytes() {
+        if byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.' | b'~') {
+            out.push(byte as char);
+        } else {
+            out.push_str(&format!("%{byte:02X}"));
+        }
+    }
+    out
+}
+
+fn query_string(params: &HashMap<String, String>) -> String {
+    if params.is_empty() {
+        return String::new();
+    }
+    let mut parts = params
+        .iter()
+        .map(|(key, value)| format!("{}={}", encode(key), encode(value)))
+        .collect::<Vec<_>>();
+    parts.sort();
+    format!("?{}", parts.join("&"))
+}
+
+async fn archive_output(
+    state: &AppState,
+    auth: &DigitalHumanAuth,
+    item: &Value,
+    prompt_id: &str,
+    script: &str,
+) -> Option<Value> {
+    let params = output_query(item);
+    let filename = params.get("filename")?.clone();
+    let kind = output_kind(item).to_string();
+    let path = if kind == "video" { "/viewvideo" } else { "/view" };
+    let response = native_request(state, Method::GET, &format!("{path}{}", query_string(&params)))
+        .send()
+        .await
+        .ok()?;
+    if !response.status().is_success() {
+        return None;
+    }
+    let content_type = response
+        .headers()
+        .get(reqwest::header::CONTENT_TYPE)
+        .and_then(|value| value.to_str().ok())
+        .unwrap_or("application/octet-stream")
+        .to_string();
+    let bytes = response.bytes().await.ok()?;
+    let input = WorkspaceAssetInput {
+        user_id: auth.user_id,
+        source: "digital-human".to_string(),
+        kind: kind.clone(),
+        title: Some(filename.chars().take(80).collect()),
+        prompt: Some(script.chars().take(1000).collect()),
+        model: Some("comfy-native".to_string()),
+        remote_url: None,
+        metadata: json!({
+            "digitalHuman": {
+                "promptId": prompt_id,
+                "filename": filename,
+                "type": params.get("type"),
+                "subfolder": params.get("subfolder"),
+                "raw": item,
+            }
+        }),
+    };
+    match workspace::record_bytes_asset(state, input, &bytes, &content_type).await {
+        Ok(asset) => Some(json!({
+            "publicId": asset.public_id,
+            "kind": kind,
+            "filename": filename,
+            "status": asset.status,
+        })),
+        Err(err) => {
+            tracing::warn!(%err, "failed to archive digital human output");
+            None
+        }
+    }
+}
+
+fn job_visible_to_auth(task: &BackgroundTask, auth: &DigitalHumanAuth) -> bool {
+    if auth.role == "admin" && auth.user_id.is_none() {
+        return true;
+    }
+    let Some(raw) = task.result_json.as_deref() else {
+        return false;
+    };
+    let Ok(value) = serde_json::from_str::<Value>(raw) else {
+        return false;
+    };
+    value.get("userId").and_then(Value::as_i64) == auth.user_id
+}
+
+fn task_to_json(task: BackgroundTask) -> Value {
+    let result = task
+        .result_json
+        .as_deref()
+        .and_then(|raw| serde_json::from_str::<Value>(raw).ok())
+        .unwrap_or_else(|| json!({}));
+    json!({
+        "id": task.id,
+        "type": task.task_type,
+        "title": task.title,
+        "status": task.status,
+        "message": task.message,
+        "error": task.error,
+        "progress": task.progress.unwrap_or(0),
+        "progressTotal": task.progress_total.unwrap_or(5),
+        "result": result,
+        "createdAt": task.created_at,
+        "updatedAt": task.updated_at,
+        "startedAt": task.started_at,
+        "finishedAt": task.finished_at,
+    })
+}
+
+fn terminal_status(status: &str) -> bool {
+    matches!(status, "succeeded" | "failed" | "cancelled")
+}
+
+async fn load_job(state: &AppState, id: &str, auth: &DigitalHumanAuth) -> AppResult<Value> {
+    let task: BackgroundTask =
+        sqlx::query_as("SELECT * FROM background_tasks WHERE id = ? AND task_type = ? LIMIT 1")
+            .bind(id)
+            .bind(TASK_TYPE_DIGITAL_HUMAN)
+            .fetch_optional(&*state.db)
+            .await?
+            .ok_or_else(|| AppError::NotFound("Digital human job not found".into()))?;
+    if !job_visible_to_auth(&task, auth) {
+        return Err(AppError::NotFound("Digital human job not found".into()));
+    }
+    Ok(task_to_json(task))
+}
+
+async fn update_progress(state: &AppState, id: &str, progress: i64, message: &str) -> AppResult<()> {
+    sqlx::query(
+        "UPDATE background_tasks
+         SET progress = ?, message = ?, updated_at = datetime('now')
+         WHERE id = ? AND status IN ('pending', 'running')",
+    )
+    .bind(progress)
+    .bind(message)
+    .bind(id)
+    .execute(&*state.db)
+    .await?;
+    Ok(())
+}
+
+async fn is_cancelled(state: &AppState, id: &str) -> AppResult<bool> {
+    let status: Option<String> =
+        sqlx::query_scalar("SELECT status FROM background_tasks WHERE id = ? LIMIT 1")
+            .bind(id)
+            .fetch_optional(&*state.db)
+            .await?;
+    Ok(status.as_deref() == Some("cancelled"))
+}
+
+async fn create_job(
+    State(state): State<AppState>,
+    headers: HeaderMap,
+    Json(body): Json<Value>,
+) -> AppResult<Json<Value>> {
+    let auth = resolve_auth(&headers, &state).await?;
+    if !state.config.comfyui.uses_native_backend() || state.config.comfyui.uses_runninghub_backend()
+    {
+        return Err(AppError::BadRequest(
+            "数字人商品讲解需要 comfyui.backend = cloud 或 external".into(),
+        ));
+    }
+    let template_id = require_string(&body, &["templateId", "template_id"], "请选择工作流模板")?;
+    let workflow = storage::get_workflow(
+        &state.db,
+        &template_id,
+        auth.user_id,
+        auth.role == "admin",
+    )
+    .await?;
+    let mapping = workflow_mapping_from_body(&body, &workflow.graph);
+    let mut prompt = workflow_prompt_from_graph(&workflow.graph);
+    let script = inject_digital_human_inputs(&mut prompt, &mapping, &body)?;
+    let task_id = format!("dh_{}", uuid::Uuid::new_v4().simple());
+    let title = script.chars().take(30).collect::<String>();
+    let initial_result = json!({
+        "userId": auth.user_id,
+        "role": auth.role,
+        "templateId": template_id,
+        "templateName": workflow.name,
+        "audioRole": optional_string(&body, &["audioRole", "audio_role"]).unwrap_or_else(|| "both".to_string()),
+        "request": body,
+        "mapping": mapping,
+    });
+    sqlx::query(
+        "INSERT INTO background_tasks
+         (id, task_type, title, status, message, dedupe_key, result_json, progress, progress_total, started_at)
+         VALUES (?, ?, ?, 'running', '任务已创建，准备提交 Comfy', ?, ?, 1, 5, datetime('now'))",
+    )
+    .bind(&task_id)
+    .bind(TASK_TYPE_DIGITAL_HUMAN)
+    .bind(if title.trim().is_empty() { "数字人讲解任务" } else { &title })
+    .bind(format!("digital-human:{}", auth.user_id.map(|id| id.to_string()).unwrap_or_else(|| "admin".into())))
+    .bind(initial_result.to_string())
+    .execute(&*state.db)
+    .await?;
+
+    let state_for_task = state.clone();
+    let task_id_for_spawn = task_id.clone();
+    let auth_for_task = auth.clone();
+    tokio::spawn(async move {
+        let result = async {
+            update_progress(&state_for_task, &task_id_for_spawn, 2, "正在提交 Comfy 工作流")
+                .await?;
+            let (prompt_id, submit_response) = submit_prompt(&state_for_task, prompt).await?;
+            let mid_result = json!({
+                "userId": auth_for_task.user_id,
+                "role": auth_for_task.role,
+                "templateId": template_id,
+                "templateName": workflow.name,
+                "promptId": prompt_id,
+                "submitResponse": submit_response,
+                "request": body,
+                "mapping": mapping,
+            });
+            sqlx::query(
+                "UPDATE background_tasks SET result_json = ?, updated_at = datetime('now') WHERE id = ?",
+            )
+            .bind(mid_result.to_string())
+            .bind(&task_id_for_spawn)
+            .execute(&*state_for_task.db)
+            .await?;
+
+            for _ in 0..DIGITAL_HUMAN_MAX_POLLS {
+                if is_cancelled(&state_for_task, &task_id_for_spawn).await? {
+                    return Ok(());
+                }
+                update_progress(&state_for_task, &task_id_for_spawn, 3, "Comfy 正在生成视频")
+                    .await?;
+                let history = prompt_history(&state_for_task, &prompt_id).await?;
+                let outputs = output_objects(&history);
+                if !outputs.is_empty() {
+                    update_progress(
+                        &state_for_task,
+                        &task_id_for_spawn,
+                        4,
+                        "生成完成，正在归档产出",
+                    )
+                    .await?;
+                    let mut assets = Vec::new();
+                    for output in &outputs {
+                        if let Some(asset) = archive_output(
+                            &state_for_task,
+                            &auth_for_task,
+                            output,
+                            &prompt_id,
+                            &script,
+                        )
+                        .await
+                        {
+                            assets.push(asset);
+                        }
+                    }
+                    let final_result = json!({
+                        "userId": auth_for_task.user_id,
+                        "role": auth_for_task.role,
+                        "templateId": template_id,
+                        "templateName": workflow.name,
+                        "promptId": prompt_id,
+                        "request": body,
+                        "mapping": mapping,
+                        "history": history,
+                        "outputs": outputs,
+                        "assets": assets,
+                    });
+                    sqlx::query(
+                        "UPDATE background_tasks
+                         SET status = 'succeeded', message = '数字人讲解视频生成完成', result_json = ?, progress = 5,
+                             updated_at = datetime('now'), finished_at = datetime('now')
+                         WHERE id = ? AND status IN ('pending', 'running')",
+                    )
+                    .bind(final_result.to_string())
+                    .bind(&task_id_for_spawn)
+                    .execute(&*state_for_task.db)
+                    .await?;
+                    return Ok(());
+                }
+                tokio::time::sleep(Duration::from_secs(DIGITAL_HUMAN_POLL_INTERVAL_SECS)).await;
+            }
+            Err(AppError::GatewayTimeout("Comfy 数字人任务等待超时".into()))
+        }
+        .await;
+        if let Err(err) = result {
+            let _ = sqlx::query(
+                "UPDATE background_tasks
+                 SET status = 'failed', message = '数字人任务失败', error = ?, updated_at = datetime('now'), finished_at = datetime('now')
+                 WHERE id = ? AND status IN ('pending', 'running')",
+            )
+            .bind(err.to_string())
+            .bind(&task_id_for_spawn)
+            .execute(&*state_for_task.db)
+            .await;
+        }
+    });
+
+    let job = load_job(&state, &task_id, &auth).await?;
+    Ok(Json(json!({ "success": true, "job": job })))
+}
+
+async fn get_job(
+    State(state): State<AppState>,
+    headers: HeaderMap,
+    Path(id): Path<String>,
+) -> AppResult<Json<Value>> {
+    let auth = resolve_auth(&headers, &state).await?;
+    let job = load_job(&state, &id, &auth).await?;
+    Ok(Json(json!({ "success": true, "job": job })))
+}
+
+async fn list_jobs(
+    State(state): State<AppState>,
+    headers: HeaderMap,
+    Query(query): Query<JobListQuery>,
+) -> AppResult<Json<Value>> {
+    let auth = resolve_auth(&headers, &state).await?;
+    let limit = query.limit.unwrap_or(20).clamp(1, 100);
+    let tasks: Vec<BackgroundTask> = sqlx::query_as(
+        "SELECT * FROM background_tasks
+         WHERE task_type = ?
+         ORDER BY created_at DESC
+         LIMIT ?",
+    )
+    .bind(TASK_TYPE_DIGITAL_HUMAN)
+    .bind(limit * 3)
+    .fetch_all(&*state.db)
+    .await?;
+    let jobs = tasks
+        .into_iter()
+        .filter(|task| job_visible_to_auth(task, &auth))
+        .take(limit as usize)
+        .map(task_to_json)
+        .collect::<Vec<_>>();
+    Ok(Json(json!({ "success": true, "jobs": jobs })))
+}
+
+async fn cancel_job(
+    State(state): State<AppState>,
+    headers: HeaderMap,
+    Path(id): Path<String>,
+) -> AppResult<Json<Value>> {
+    let auth = resolve_auth(&headers, &state).await?;
+    let task: BackgroundTask =
+        sqlx::query_as("SELECT * FROM background_tasks WHERE id = ? AND task_type = ? LIMIT 1")
+            .bind(&id)
+            .bind(TASK_TYPE_DIGITAL_HUMAN)
+            .fetch_optional(&*state.db)
+            .await?
+            .ok_or_else(|| AppError::NotFound("Digital human job not found".into()))?;
+    if !job_visible_to_auth(&task, &auth) {
+        return Err(AppError::NotFound("Digital human job not found".into()));
+    }
+    if terminal_status(&task.status) {
+        return Ok(Json(json!({ "success": true, "job": task_to_json(task) })));
+    }
+    sqlx::query(
+        "UPDATE background_tasks
+         SET status = 'cancelled', message = '已取消', updated_at = datetime('now'), finished_at = datetime('now')
+         WHERE id = ? AND task_type = ? AND status IN ('pending', 'running')",
+    )
+    .bind(&id)
+    .bind(TASK_TYPE_DIGITAL_HUMAN)
+    .execute(&*state.db)
+    .await?;
+    let job = load_job(&state, &id, &auth).await?;
+    Ok(Json(json!({ "success": true, "job": job })))
+}
+
+async fn clear_jobs(State(state): State<AppState>, headers: HeaderMap) -> AppResult<Json<Value>> {
+    let auth = resolve_auth(&headers, &state).await?;
+    let tasks: Vec<BackgroundTask> = sqlx::query_as(
+        "SELECT * FROM background_tasks WHERE task_type = ? ORDER BY created_at DESC",
+    )
+    .bind(TASK_TYPE_DIGITAL_HUMAN)
+    .fetch_all(&*state.db)
+    .await?;
+    let mut cleared = 0_i64;
+    for task in tasks {
+        if !job_visible_to_auth(&task, &auth) || !terminal_status(&task.status) {
+            continue;
+        }
+        let affected = sqlx::query("DELETE FROM background_tasks WHERE id = ? AND task_type = ?")
+            .bind(&task.id)
+            .bind(TASK_TYPE_DIGITAL_HUMAN)
+            .execute(&*state.db)
+            .await?
+            .rows_affected();
+        cleared += i64::try_from(affected).unwrap_or(0);
+    }
+    Ok(Json(json!({ "success": true, "cleared": cleared })))
+}
+
+fn upload_kind_from_mime(mime: &str) -> &'static str {
+    if mime.starts_with("video/") {
+        "video"
+    } else if mime.starts_with("audio/") {
+        "audio"
+    } else {
+        "image"
+    }
+}
+
+async fn upload_asset(
+    State(state): State<AppState>,
+    headers: HeaderMap,
+    mut multipart: Multipart,
+) -> AppResult<Json<Value>> {
+    let auth = resolve_auth(&headers, &state).await?;
+    let mut file_name = "digital-human-upload.bin".to_string();
+    let mut mime_type = "application/octet-stream".to_string();
+    let mut bytes: Option<Bytes> = None;
+    let mut purpose = "avatar".to_string();
+    while let Some(field) = multipart
+        .next_field()
+        .await
+        .map_err(|err| AppError::BadRequest(err.to_string()))?
+    {
+        let name = field.name().unwrap_or_default().to_string();
+        if name == "file" {
+            file_name = field.file_name().unwrap_or(&file_name).to_string();
+            mime_type = field
+                .content_type()
+                .map(str::to_string)
+                .unwrap_or_else(|| mime_type.clone());
+            bytes = Some(
+                field
+                    .bytes()
+                    .await
+                    .map_err(|err| AppError::BadRequest(err.to_string()))?,
+            );
+        } else if name == "purpose" {
+            purpose = field.text().await.unwrap_or_else(|_| purpose.clone());
+        }
+    }
+    let Some(bytes) = bytes else {
+        return Err(AppError::BadRequest("file is required".into()));
+    };
+    let kind = upload_kind_from_mime(&mime_type);
+    let input = WorkspaceAssetInput {
+        user_id: auth.user_id,
+        source: "digital-human".to_string(),
+        kind: kind.to_string(),
+        title: Some(file_name.clone()),
+        prompt: None,
+        model: None,
+        remote_url: None,
+        metadata: json!({ "digitalHumanUpload": { "purpose": purpose, "fileName": file_name } }),
+    };
+    let asset = workspace::record_bytes_asset(&state, input, &bytes, &mime_type).await?;
+    let file_url = format!(
+        "{}/api/{}/workspace/assets/{}/file",
+        public_base_url(&state),
+        if auth.user_id.is_some() { "user" } else { "" }.trim_matches('/'),
+        asset.public_id
+    )
+    .replace("/api//", "/api/");
+    Ok(Json(json!({
+        "success": true,
+        "asset": {
+            "publicId": asset.public_id,
+            "kind": kind,
+            "status": asset.status,
+            "fileUrl": file_url,
+        }
+    })))
+}