Nvidia/BossAI
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Handle Comfy Cloud video job status

Browse Source
This commit is contained in:
boos
2026-06-05 22:01:03 +08:00
parent 3f22d7b1ab
commit 598b2f48d6
3 changed files with 98 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/proxy/executors/mod.rs
+30 -2
View File
@@ -963,16 +963,44 @@ async fn comfy_json(
}
async fn comfy_history(state: &AppState, prompt_id: &str) -> AppResult<Value> {
let history = comfy_json(state, reqwest::Method::GET, &format!("/history/{prompt_id}"), None).await?;
let path = if state.config.comfyui.uses_cloud_backend() {
format!("/jobs/{prompt_id}")
} else {
format!("/history/{prompt_id}")
};
let history = comfy_json(state, reqwest::Method::GET, &path, None).await?;
let failed = comfy_failure_message(&history);
Ok(json!({
"id": prompt_id,
"object": "video.generation.task",
"backend": "comfy",
"status": if comfy_output_objects(&history) { "succeeded" } else { "running" },
"status": if failed.is_some() {
"failed"
} else if comfy_output_objects(&history) {
"succeeded"
} else {
"running"
},
"error": failed,
"history": history,
}))
}
fn comfy_failure_message(value: &Value) -> Option<String> {
let status = scalar_string(value.get("status").or_else(|| value.get("state")))
.unwrap_or_default()
.to_ascii_lowercase();
if matches!(status.as_str(), "failed" | "error" | "cancelled" | "canceled") {
return Some(
scalar_string(value.pointer("/execution_error/exception_message"))
.or_else(|| scalar_string(value.get("error")))
.or_else(|| scalar_string(value.get("message")))
.unwrap_or_else(|| format!("Comfy job {status}")),
);
}
scalar_string(value.pointer("/execution_error/exception_message"))
}
fn comfy_output_objects(value: &Value) -> bool {
match value {
Value::Object(object) => {
src/routes/digital_human.rs
+24 -1
View File
@@ -720,7 +720,27 @@ async fn submit_prompt(state: &AppState, prompt: Value) -> AppResult<(String, Va
}
async fn prompt_history(state: &AppState, prompt_id: &str) -> AppResult<Value> {
native_json(state, Method::GET, &format!("/history/{prompt_id}"), None).await
let path = if state.config.comfyui.uses_cloud_backend() {
format!("/jobs/{prompt_id}")
} else {
format!("/history/{prompt_id}")
};
native_json(state, Method::GET, &path, None).await
}
fn prompt_failure_message(value: &Value) -> Option<String> {
let status = scalar_string(value.get("status").or_else(|| value.get("state")))
.unwrap_or_default()
.to_ascii_lowercase();
if matches!(status.as_str(), "failed" | "error" | "cancelled" | "canceled") {
return Some(
scalar_string(value.pointer("/execution_error/exception_message"))
.or_else(|| scalar_string(value.get("error")))
.or_else(|| scalar_string(value.get("message")))
.unwrap_or_else(|| format!("Comfy job {status}")),
);
}
scalar_string(value.pointer("/execution_error/exception_message"))
}
fn output_objects(value: &Value) -> Vec<Value> {
@@ -1015,6 +1035,9 @@ async fn create_job(
update_progress(&state_for_task, &task_id_for_spawn, 3, "Comfy 正在生成视频")
.await?;
let history = prompt_history(&state_for_task, &prompt_id).await?;
if let Some(message) = prompt_failure_message(&history) {
return Err(AppError::UpstreamError(message));
}
let outputs = output_objects(&history);
if !outputs.is_empty() {
update_progress(
src/routes/money_printer.rs
+44 -1
View File
@@ -967,7 +967,30 @@ async fn submit_comfy_prompt(state: &AppState, prompt: Value) -> AppResult<(Stri
}
async fn comfy_history(state: &AppState, prompt_id: &str) -> AppResult<Value> {
native_json(state, Method::GET, &format!("/history/{prompt_id}"), None).await
let path = if state.config.comfyui.uses_cloud_backend() {
format!("/jobs/{prompt_id}")
} else {
format!("/history/{prompt_id}")
};
native_json(state, Method::GET, &path, None).await
}
fn comfy_failure_message(value: &Value) -> Option<String> {
let status = scalar_string(value.get("status").or_else(|| value.get("state")))
.unwrap_or_default()
.to_ascii_lowercase();
if matches!(status.as_str(), "failed" | "error" | "cancelled" | "canceled") {
return Some(
scalar_string(value.pointer("/execution_error/exception_message"))
.or_else(|| scalar_string(value.get("error")))
.or_else(|| scalar_string(value.get("message")))
.unwrap_or_else(|| format!("Comfy job {status}")),
);
}
if let Some(message) = scalar_string(value.pointer("/execution_error/exception_message")) {
return Some(message);
}
None
}
fn comfy_output_objects(value: &Value) -> Vec<Value> {
@@ -1316,6 +1339,23 @@ async fn create_job(
.and_then(Value::as_str)
.unwrap_or_default()
.to_string();
let submitted_result = json!({
"userId": auth_for_task.user_id,
"role": auth_for_task.role,
"keyName": auth_for_task.key_name,
"topic": topic,
"request": body,
"script": script,
"videoBackend": video_backend,
"model": "comfy-api",
"prompt": prompt,
"video": submitted,
});
sqlx::query("UPDATE background_tasks SET result_json = ?, updated_at = datetime('now') WHERE id = ? AND status IN ('pending', 'running')")
.bind(submitted_result.to_string())
.bind(&task_id_for_spawn)
.execute(&*state_for_task.db)
.await?;
let mut final_history = Value::Null;
for _ in 0..COMFY_MAX_POLLS {
if is_cancelled(&state_for_task, &task_id_for_spawn).await? {
@@ -1323,6 +1363,9 @@ async fn create_job(
}
update_progress(&state_for_task, &task_id_for_spawn, 4, "Comfy 正在生成视频").await?;
let history = comfy_history(&state_for_task, &prompt_id).await?;
if let Some(message) = comfy_failure_message(&history) {
return Err(AppError::UpstreamError(message));
}
let outputs = comfy_output_objects(&history);
if !outputs.is_empty() {
update_progress(&state_for_task, &task_id_for_spawn, 5, "Comfy 视频完成，正在归档").await?;