# Comfy Cloud 工作流执行适配器 实现计划 > **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. **Goal:** 在 BoosAPI 内实现 ComfyUI 原生协议 ↔ Comfy Cloud API 翻译适配器,让 Pixelle/comfykit 零改动通过 BoosAPI 调用 Comfy Cloud 执行工作流。 **Architecture:** 新增 `/comfy-cloud/{prompt,history,view}` 端点,内部翻译成 `cloud.comfy.org/api/*` + `X-API-Key`。Comfy Cloud 的 base_url + key 从站点管理(`platform='comfy_cloud'` 的 active 站点)读取。纯翻译函数单测,wiremock 集成测。 **Tech Stack:** Rust (axum, reqwest, sqlx, serde_json, wiremock, tokio)。 > **构建/测试:** `cargo build -j 2`、`cargo test -p boosapi --lib comfy_cloud`。分支 `feat/comfy-cloud-adapter`。 --- ## 文件结构 - Create `src/services/comfy_cloud/mod.rs` — 模块导出。 - Create `src/services/comfy_cloud/site.rs` — `resolve_comfy_cloud_site` 读站点配置。 - Create `src/services/comfy_cloud/translate.rs` — `comfy_cloud_job_to_history` 纯翻译函数。 - Create `src/routes/comfy_cloud_adapter.rs` — 三个适配端点 + `routes()`。 - Modify `src/services/mod.rs` — 加 `pub mod comfy_cloud;`。 - Modify `src/routes/mod.rs` — 加 `mod comfy_cloud_adapter;` + `.merge(...)`。 - Modify `src/web/pages/Sites.tsx` — `SITE_PLATFORM_OPTIONS` 加 `comfy_cloud` 选项。 --- ## Task 1: 站点配置读取 `resolve_comfy_cloud_site` **Files:** - Create: `src/services/comfy_cloud/mod.rs` - Create: `src/services/comfy_cloud/site.rs` - Modify: `src/services/mod.rs` - [ ] **Step 1: `src/services/mod.rs` 末尾加模块声明** ```rust pub mod comfy_cloud; ``` - [ ] **Step 2: 写 `src/services/comfy_cloud/mod.rs`** ```rust //! Comfy Cloud 工作流执行适配器 service 层。 pub mod site; pub mod translate; ``` - [ ] **Step 3: 写失败测试 `src/services/comfy_cloud/site.rs`** ```rust use crate::error::{AppError, AppResult}; use sqlx::SqlitePool; /// 解析 Comfy Cloud 站点配置,返回 (base_url, api_key)。 /// 取 platform='comfy_cloud' 且 status='active' 的站点。 pub async fn resolve_comfy_cloud_site(db: &SqlitePool) -> AppResult<(String, String)> { let row: Option<(String, Option)> = sqlx::query_as( "SELECT url, api_key FROM sites WHERE platform = 'comfy_cloud' AND status = 'active' ORDER BY COALESCE(sort_order, 0), id LIMIT 1", ) .fetch_optional(db) .await?; let (url, api_key) = row.ok_or_else(|| { AppError::ServiceUnavailable("未配置 Comfy Cloud 站点(站点管理添加 comfy_cloud 平台)".into()) })?; let api_key = api_key .map(|k| k.trim().to_string()) .filter(|k| !k.is_empty()) .ok_or_else(|| AppError::ServiceUnavailable("Comfy Cloud 站点缺少 API Key".into()))?; let base_url = url.trim().trim_end_matches('/').to_string(); Ok((base_url, api_key)) } #[cfg(test)] mod tests { use super::*; use sqlx::sqlite::SqlitePoolOptions; async fn pool() -> SqlitePool { let pool = SqlitePoolOptions::new().max_connections(1).connect("sqlite::memory:").await.unwrap(); sqlx::query("CREATE TABLE sites (id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT, url TEXT NOT NULL, platform TEXT NOT NULL, status TEXT NOT NULL, api_key TEXT, sort_order INTEGER)") .execute(&pool).await.unwrap(); pool } #[tokio::test] async fn resolves_active_comfy_cloud_site() { let pool = pool().await; sqlx::query("INSERT INTO sites (name,url,platform,status,api_key) VALUES ('cc','https://cloud.comfy.org/','comfy_cloud','active','key-123')") .execute(&pool).await.unwrap(); let (url, key) = resolve_comfy_cloud_site(&pool).await.unwrap(); assert_eq!(url, "https://cloud.comfy.org"); assert_eq!(key, "key-123"); } #[tokio::test] async fn errors_when_not_configured() { let pool = pool().await; let r = resolve_comfy_cloud_site(&pool).await; assert!(matches!(r, Err(AppError::ServiceUnavailable(_)))); } } ``` - [ ] **Step 4: 运行测试,确认失败** Run: `cargo test -p boosapi --lib services::comfy_cloud::site 2>&1 | tail -15` Expected: FAIL —— 模块未挂载/函数未定义(编译错误)。 - [ ] **Step 5: 实现** Step 3 的函数体即实现(测试与实现同文件)。确认 `src/services/comfy_cloud/mod.rs` 已 `pub mod site;`。 - [ ] **Step 6: 运行测试,确认通过** Run: `cargo test -p boosapi --lib services::comfy_cloud::site 2>&1 | tail -10` Expected: PASS(2 个测试)。 - [ ] **Step 7: Commit** ```bash git add src/services/mod.rs src/services/comfy_cloud/mod.rs src/services/comfy_cloud/site.rs git commit -m "feat(comfy-cloud): 站点配置读取 resolve_comfy_cloud_site" ``` ## Task 2: 翻译函数 `comfy_cloud_job_to_history` **Files:** - Create: `src/services/comfy_cloud/translate.rs` - [ ] **Step 1: 写失败测试 `src/services/comfy_cloud/translate.rs`** ```rust use serde_json::{json, Value}; /// 把 Comfy Cloud 的 job status(+ 完成时的 jobs detail outputs)翻译成 /// ComfyUI 原生 /history/{id} 的形状。 /// /// - status pending/in_progress → 返回空对象 {}(comfykit 据此继续轮询) /// - status completed → { "{id}": { "status": {completed:true, status_str:"success"}, "outputs": {...} } } /// - status failed/cancelled → { "{id}": { "status": {completed:false, status_str:"error"} } } pub fn comfy_cloud_job_to_history( prompt_id: &str, status: &str, outputs: Option<&Value>, ) -> Value { match status { "completed" => json!({ prompt_id: { "status": { "completed": true, "status_str": "success" }, "outputs": outputs.cloned().unwrap_or_else(|| json!({})), } }), "failed" | "cancelled" => json!({ prompt_id: { "status": { "completed": false, "status_str": "error" }, "outputs": {}, } }), _ => json!({}), } } #[cfg(test)] mod tests { use super::*; #[test] fn in_progress_returns_empty() { assert_eq!(comfy_cloud_job_to_history("p1", "in_progress", None), json!({})); assert_eq!(comfy_cloud_job_to_history("p1", "pending", None), json!({})); } #[test] fn completed_wraps_outputs() { let outputs = json!({ "9": { "images": [{ "filename": "a.png", "subfolder": "", "type": "output" }] } }); let h = comfy_cloud_job_to_history("p1", "completed", Some(&outputs)); assert_eq!(h["p1"]["status"]["completed"], true); assert_eq!(h["p1"]["outputs"]["9"]["images"][0]["filename"], "a.png"); } #[test] fn failed_marks_error() { let h = comfy_cloud_job_to_history("p1", "failed", None); assert_eq!(h["p1"]["status"]["status_str"], "error"); assert_eq!(h["p1"]["status"]["completed"], false); } } ``` - [ ] **Step 2: 运行测试,确认失败** Run: `cargo test -p boosapi --lib services::comfy_cloud::translate 2>&1 | tail -15` Expected: FAIL —— 模块未挂载/函数未定义。 - [ ] **Step 3: 实现** Step 1 的函数体即实现。确认 `mod.rs` 已 `pub mod translate;`(Task 1 Step 2 已加)。 - [ ] **Step 4: 运行测试,确认通过** Run: `cargo test -p boosapi --lib services::comfy_cloud::translate 2>&1 | tail -10` Expected: PASS(3 个测试)。 - [ ] **Step 5: Commit** ```bash git add src/services/comfy_cloud/translate.rs git commit -m "feat(comfy-cloud): job→ComfyUI history 翻译函数" ``` ## Task 3: 适配端点骨架 + `POST /comfy-cloud/prompt` **Files:** - Create: `src/routes/comfy_cloud_adapter.rs` - [ ] **Step 1: 写骨架 + 测试 `src/routes/comfy_cloud_adapter.rs`** ```rust use crate::error::{AppError, AppResult}; use crate::services::comfy_cloud::site::resolve_comfy_cloud_site; use crate::services::comfy_cloud::translate::comfy_cloud_job_to_history; use crate::AppState; use axum::{ body::Body, extract::{Path, RawQuery, State}, http::{header, StatusCode}, response::Response, routing::{get, post}, Json, Router, }; use serde_json::Value; pub fn routes() -> Router { Router::new() .route("/comfy-cloud/prompt", post(submit_prompt)) .route("/comfy-cloud/history/{prompt_id}", get(get_history)) .route("/comfy-cloud/view", get(get_view)) } fn client() -> reqwest::Client { reqwest::Client::builder() .timeout(std::time::Duration::from_secs(120)) .build() .unwrap_or_else(|_| reqwest::Client::new()) } async fn submit_prompt( State(state): State, Json(body): Json, ) -> AppResult> { let (base_url, api_key) = resolve_comfy_cloud_site(&state.db).await?; let resp = client() .post(format!("{base_url}/api/prompt")) .header("X-API-Key", api_key) .json(&body) .send() .await .map_err(|e| AppError::UpstreamError(e.to_string()))?; let status = resp.status(); let text = resp.text().await.map_err(|e| AppError::UpstreamError(e.to_string()))?; if !status.is_success() { return Err(AppError::UpstreamError(format!("Comfy Cloud prompt [{status}]: {text}"))); } let value: Value = serde_json::from_str(&text).unwrap_or_else(|_| serde_json::json!({ "body": text })); Ok(Json(value)) } #[cfg(test)] mod tests { use super::*; use wiremock::matchers::{header as match_header, method, path}; use wiremock::{Mock, MockServer, ResponseTemplate}; use sqlx::sqlite::SqlitePoolOptions; async fn db_with_site(base_url: &str) -> sqlx::SqlitePool { let pool = SqlitePoolOptions::new().max_connections(1).connect("sqlite::memory:").await.unwrap(); sqlx::query("CREATE TABLE sites (id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT, url TEXT NOT NULL, platform TEXT NOT NULL, status TEXT NOT NULL, api_key TEXT, sort_order INTEGER)") .execute(&pool).await.unwrap(); sqlx::query("INSERT INTO sites (name,url,platform,status,api_key) VALUES ('cc',?,'comfy_cloud','active','key-123')") .bind(base_url).execute(&pool).await.unwrap(); pool } #[tokio::test] async fn submit_forwards_with_api_key() { let server = MockServer::start().await; Mock::given(method("POST")) .and(path("/api/prompt")) .and(match_header("X-API-Key", "key-123")) .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ "prompt_id": "p1" }))) .mount(&server).await; let pool = db_with_site(&server.uri()).await; let (base_url, api_key) = resolve_comfy_cloud_site(&pool).await.unwrap(); // 直接验证转发契约(handler 依赖 AppState,这里测底层契约) let resp = client().post(format!("{base_url}/api/prompt")) .header("X-API-Key", api_key) .json(&serde_json::json!({ "prompt": {} })) .send().await.unwrap(); let v: Value = resp.json().await.unwrap(); assert_eq!(v["prompt_id"], "p1"); } } ``` > 注:`submit_prompt` handler 需 `AppState`,完整 handler 走 Task 6 联调验证;本 task 的 > wiremock 测试验证「X-API-Key 注入 + `/api/prompt` 路径 + prompt_id 透传」这一核心契约。 - [ ] **Step 2: 运行测试,确认失败** Run: `cargo test -p boosapi --lib routes::comfy_cloud_adapter 2>&1 | tail -15` Expected: FAIL —— 模块未挂载(`get_history`/`get_view` 未定义,编译错误)。 > 因 `routes()` 引用了 Task 4/5 的 `get_history`/`get_view`,本 task 先写它们的空桩 > 让编译过(Step 3),Task 4/5 再填实现。 - [ ] **Step 3: 加 `get_history`/`get_view` 空桩(接在 `submit_prompt` 之后,让编译通过)** ```rust async fn get_history( State(state): State, Path(prompt_id): Path, ) -> AppResult> { let _ = (&state, &prompt_id); Ok(Json(serde_json::json!({}))) } async fn get_view( State(state): State, RawQuery(query): RawQuery, ) -> AppResult { let _ = (&state, &query); Ok(Response::builder().status(StatusCode::NOT_IMPLEMENTED).body(Body::empty()) .map_err(|e| AppError::Internal(anyhow::anyhow!(e)))?) } ``` - [ ] **Step 4: 运行测试,确认通过** Run: `cargo test -p boosapi --lib routes::comfy_cloud_adapter 2>&1 | tail -10` Expected: PASS(1 个测试)。 > 此时 `routes::comfy_cloud_adapter` 模块还没挂进 `routes/mod.rs`,测试是模块内单测, > 不需要挂载即可跑。挂载在 Task 6。 - [ ] **Step 5: Commit** ```bash git add src/routes/comfy_cloud_adapter.rs git commit -m "feat(comfy-cloud): 适配端点骨架 + POST /prompt 转发" ``` ## Task 4: `GET /comfy-cloud/history/{id}` 轮询翻译 **Files:** - Modify: `src/routes/comfy_cloud_adapter.rs` - [ ] **Step 1: 写失败测试(加到测试模块内)** ```rust #[tokio::test] async fn history_in_progress_then_completed() { let server = MockServer::start().await; // 状态查询:completed Mock::given(method("GET")) .and(path("/api/job/p1/status")) .and(match_header("X-API-Key", "key-123")) .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ "status": "completed" }))) .mount(&server).await; // 详情:outputs Mock::given(method("GET")) .and(path("/api/jobs/p1")) .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ "outputs": { "9": { "images": [{ "filename": "a.png", "subfolder": "", "type": "output" }] } } }))) .mount(&server).await; let pool = db_with_site(&server.uri()).await; let history = fetch_history(&pool, "p1").await.unwrap(); assert_eq!(history["p1"]["status"]["completed"], true); assert_eq!(history["p1"]["outputs"]["9"]["images"][0]["filename"], "a.png"); } ``` - [ ] **Step 2: 运行测试,确认失败** Run: `cargo test -p boosapi --lib routes::comfy_cloud_adapter::tests::history 2>&1 | tail -15` Expected: FAIL —— `fetch_history` 未定义。 - [ ] **Step 3: 实现 `fetch_history` + 替换 `get_history` 空桩** 把 Task 3 的 `get_history` 空桩替换为: ```rust async fn fetch_history(db: &sqlx::SqlitePool, prompt_id: &str) -> AppResult { let (base_url, api_key) = resolve_comfy_cloud_site(db).await?; let http = client(); let status_resp = http .get(format!("{base_url}/api/job/{prompt_id}/status")) .header("X-API-Key", &api_key) .send().await .map_err(|e| AppError::UpstreamError(e.to_string()))?; if !status_resp.status().is_success() { let st = status_resp.status(); let body = status_resp.text().await.unwrap_or_default(); return Err(AppError::UpstreamError(format!("Comfy Cloud status [{st}]: {body}"))); } let status_json: Value = status_resp.json().await.map_err(|e| AppError::UpstreamError(e.to_string()))?; let status = status_json.get("status").and_then(Value::as_str).unwrap_or("in_progress"); if status != "completed" { return Ok(comfy_cloud_job_to_history(prompt_id, status, None)); } let detail_resp = http .get(format!("{base_url}/api/jobs/{prompt_id}")) .header("X-API-Key", &api_key) .send().await .map_err(|e| AppError::UpstreamError(e.to_string()))?; let detail_json: Value = detail_resp.json().await.map_err(|e| AppError::UpstreamError(e.to_string()))?; let outputs = detail_json.get("outputs"); Ok(comfy_cloud_job_to_history(prompt_id, status, outputs)) } async fn get_history( State(state): State, Path(prompt_id): Path, ) -> AppResult> { Ok(Json(fetch_history(&state.db, &prompt_id).await?)) } ``` - [ ] **Step 4: 运行测试,确认通过** Run: `cargo test -p boosapi --lib routes::comfy_cloud_adapter 2>&1 | tail -10` Expected: PASS(submit + history 共 2 个测试)。 - [ ] **Step 5: Commit** ```bash git add src/routes/comfy_cloud_adapter.rs git commit -m "feat(comfy-cloud): /history 轮询翻译(status+jobs→history)" ``` ## Task 5: `GET /comfy-cloud/view` 产出文件透传 **Files:** - Modify: `src/routes/comfy_cloud_adapter.rs` - [ ] **Step 1: 写失败测试(加到测试模块内)** ```rust #[tokio::test] async fn view_proxies_file_bytes() { let server = MockServer::start().await; Mock::given(method("GET")) .and(path("/api/view")) .and(match_header("X-API-Key", "key-123")) .respond_with(ResponseTemplate::new(200) .insert_header("content-type", "image/png") .set_body_bytes(b"PNGDATA".to_vec())) .mount(&server).await; let pool = db_with_site(&server.uri()).await; let (mime, bytes) = fetch_view(&pool, Some("filename=a.png&type=output")).await.unwrap(); assert_eq!(mime, "image/png"); assert_eq!(bytes, b"PNGDATA"); } ``` - [ ] **Step 2: 运行测试,确认失败** Run: `cargo test -p boosapi --lib routes::comfy_cloud_adapter::tests::view 2>&1 | tail -15` Expected: FAIL —— `fetch_view` 未定义。 - [ ] **Step 3: 实现 `fetch_view` + 替换 `get_view` 空桩** 把 Task 3 的 `get_view` 空桩替换为: ```rust async fn fetch_view(db: &sqlx::SqlitePool, query: Option<&str>) -> AppResult<(String, Vec)> { let (base_url, api_key) = resolve_comfy_cloud_site(db).await?; let q = query.unwrap_or(""); let url = if q.is_empty() { format!("{base_url}/api/view") } else { format!("{base_url}/api/view?{q}") }; let resp = client() .get(url) .header("X-API-Key", api_key) .send().await .map_err(|e| AppError::UpstreamError(e.to_string()))?; if !resp.status().is_success() { let st = resp.status(); return Err(AppError::UpstreamError(format!("Comfy Cloud view [{st}]"))); } let mime = resp .headers() .get(header::CONTENT_TYPE) .and_then(|v| v.to_str().ok()) .unwrap_or("application/octet-stream") .to_string(); let bytes = resp.bytes().await.map_err(|e| AppError::UpstreamError(e.to_string()))?.to_vec(); Ok((mime, bytes)) } async fn get_view( State(state): State, RawQuery(query): RawQuery, ) -> AppResult { let (mime, bytes) = fetch_view(&state.db, query.as_deref()).await?; Response::builder() .status(StatusCode::OK) .header(header::CONTENT_TYPE, mime) .body(Body::from(bytes)) .map_err(|e| AppError::Internal(anyhow::anyhow!(e))) } ``` > reqwest 默认跟随 302,所以 Comfy Cloud `/api/view` 的签名 URL 跳转会被自动跟随, > 最终拿到文件字节。无需手动处理 Location。 - [ ] **Step 4: 运行测试,确认通过** Run: `cargo test -p boosapi --lib routes::comfy_cloud_adapter 2>&1 | tail -10` Expected: PASS(submit + history + view 共 3 个测试)。 - [ ] **Step 5: Commit** ```bash git add src/routes/comfy_cloud_adapter.rs git commit -m "feat(comfy-cloud): /view 产出文件透传" ``` ## Task 6: 挂载路由 + 前端平台选项 + 全量验证 **Files:** - Modify: `src/routes/mod.rs` - Modify: `src/web/pages/Sites.tsx` - [ ] **Step 1: `src/routes/mod.rs` 顶部模块声明区加** ```rust mod comfy_cloud_adapter; ``` - [ ] **Step 2: `build_router` 的 merge 链里加(`.merge(picture_agent::routes())` 附近)** ```rust .merge(comfy_cloud_adapter::routes()) ``` - [ ] **Step 3: 前端 `SITE_PLATFORM_OPTIONS` 加 comfy_cloud 选项** 在 `src/web/pages/Sites.tsx` 的 `SITE_PLATFORM_OPTIONS` 数组(约 line 700-710, `{ value: 'cliproxyapi', ... }` 附近)加一项: ```tsx { value: 'comfy_cloud', label: 'Comfy Cloud', description: '工作流执行 API(数字人/图片视频)' }, ``` - [ ] **Step 4: 后端全量构建 + comfy_cloud 测试** Run: `cargo build -j 2 2>&1 | tail -3 && cargo test -p boosapi --lib comfy_cloud 2>&1 | command grep "test result"` Expected: `Finished`;comfy_cloud service + routes 测试全 PASS。 - [ ] **Step 5: 前端 typecheck** Run: `npm run typecheck:web 2>&1 | tail -5` Expected: 无错误。 - [ ] **Step 6: Commit** ```bash git add src/routes/mod.rs src/web/pages/Sites.tsx git commit -m "feat(comfy-cloud): 挂载适配路由 + 站点管理 comfy_cloud 平台选项" ``` --- ## Task 7: 真实联调(手动验证,非自动测试) > 这一步需要运行中的 BoosAPI + Pixelle,属手动验证。前 6 个 task 完成且 push 后执行。 - [ ] **Step 1: 站点管理添加 Comfy Cloud 站点** 在管理控制台「站点管理」新增站点:平台选 `Comfy Cloud`,URL 填 `https://cloud.comfy.org`, API Key 填有效的 Comfy Cloud key(`X-API-Key`),状态 active。 - [ ] **Step 2: 改 Pixelle config.yaml 的 comfyui_url** 把 `/opt/Pixelle-Video/config.yaml` 的 `comfyui.comfyui_url` 改为 `http://:/comfy-cloud`(指向 BoosAPI 适配端点), 然后 `docker compose restart api`(在 `/opt/Pixelle-Video`)。 - [ ] **Step 3: 跑一次 selfhost 视频生成** 从前端触发数字人/图片 selfhost 工作流生成视频,确认: - 不再出现 `401 AuthenticationRequired`。 - 任务能提交(拿到 prompt_id)、轮询到 completed、取回产出文件。 - [ ] **Step 4: 失败排查指引(若仍报错)** - 仍 401:检查站点 api_key 是否为有效 Comfy Cloud key(`curl -H "X-API-Key: KEY" https://cloud.comfy.org/api/user` 应回 200)。 - 404/路径错:确认 Pixelle comfyui_url 末尾无多余 `/`,BoosAPI 适配端点前缀为 `/comfy-cloud`。 - 取不到产出:看 BoosAPI 日志里 `/api/jobs/{id}` 的响应 outputs 结构,必要时调 translate 映射。 --- ## 完成标准 - 后端:`resolve_comfy_cloud_site` + `comfy_cloud_job_to_history` + 三个适配端点 (`/comfy-cloud/{prompt,history,view}`)实现并测试通过,`cargo build -j 2` 干净。 - 前端:站点管理可选 `comfy_cloud` 平台,`npm run typecheck:web` 干净。 - 联调:Pixelle 指向 BoosAPI 适配端点后,selfhost 视频生成不再 401 且能出片。 - 全部提交在 `feat/comfy-cloud-adapter` 分支。