feat(enhance): config, templates, ComfyUI client, jobs, endpoints (#192)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Cameron CordesandClaude Opus 5.5 committed 2026-10-30 15:27:28 -04:00
1 parent ebed15bd59
commit 17d74ebafe
19 files changed
+3912

No files matched your search

+423
View File
@@ -0,0 +1,423 @@
//! Thin client for the parts of ComfyUI's HTTP API Enhance uses (#192).
//!
//! | Step | Call |
//! |-----------|----------------------------------------|
//! | reachable | `GET /system_stats` |
//! | submit | `POST /prompt {prompt, client_id}` |
//! | status | `GET /queue`, `GET /history/{id}` |
//! | preview | `GET /view?filename=&subfolder=&type=` |
//! | approve | `POST /imageapi/approve {id}` |
//! | free VRAM | `POST /free` |
use std::time::Duration;
use bytes::Bytes;
use serde::Deserialize;
use serde_json::{Value, json};
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ComfyError {
/// Connect/timeout failure: ComfyUI isn't running (a normal state).
Unreachable(String),
/// ComfyUI answered with a status we don't handle.
Http { status: u16, body: String },
/// ComfyUI answered with something we can't read.
BadResponse(String),
}
impl std::fmt::Display for ComfyError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
ComfyError::Unreachable(e) => write!(f, "ComfyUI is unreachable: {e}"),
ComfyError::Http { status, body } => write!(f, "ComfyUI returned {status}: {body}"),
ComfyError::BadResponse(e) => write!(f, "unexpected ComfyUI response: {e}"),
}
}
}
impl std::error::Error for ComfyError {}
impl From<reqwest::Error> for ComfyError {
fn from(e: reqwest::Error) -> Self {
if e.is_connect() || e.is_timeout() || e.is_request() {
ComfyError::Unreachable(e.without_url().to_string())
} else {
ComfyError::BadResponse(e.without_url().to_string())
}
}
}
/// `POST /prompt` outcomes other than success.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum SubmitError {
Comfy(ComfyError),
/// ComfyUI refused the prompt (HTTP 400): validation failed.
Rejected {
message: String,
/// `node_errors` was non-empty (a node is missing an input, a model
/// or a link target).
node_errors: bool,
},
}
impl std::fmt::Display for SubmitError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
SubmitError::Comfy(e) => e.fmt(f),
SubmitError::Rejected { message, .. } => write!(f, "ComfyUI rejected the workflow: {message}"),
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ApproveError {
Comfy(ComfyError),
/// The staged result is gone (ComfyUI restarted: its temp folder is cleared).
Gone,
/// ComfyUI refused the approval (bad id, write failure).
Refused(String),
}
/// Where `GET /view` finds a staged result.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, Deserialize)]
pub struct ViewRef {
pub filename: String,
#[serde(default)]
pub subfolder: String,
#[serde(default = "temp", rename = "type")]
pub kind: String,
}
fn temp() -> String {
"temp".into()
}
/// A result the save node staged: ComfyUI's pending id, where to preview it,
/// and the library path when it was already saved (never, with approval on).
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, Deserialize)]
pub struct Staged {
pub id: String,
pub view: ViewRef,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct HistoryEntry {
pub completed: bool,
/// `success` / `error`.
pub status_str: String,
/// ComfyUI's exception message when the run failed.
pub error: Option<String>,
pub staged: Vec<Staged>,
/// Last event time, unix millis.
pub timestamp_ms: Option<i64>,
}
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct QueueState {
pub running: Vec<String>,
pub pending: Vec<String>,
}
impl QueueState {
pub fn is_empty(&self) -> bool {
self.running.is_empty() && self.pending.is_empty()
}
/// Jobs ahead of `prompt_id`; `None` when it isn't queued.
pub fn position_of(&self, prompt_id: &str) -> Option<(bool, usize)> {
if self.running.iter().any(|p| p == prompt_id) {
return Some((true, 0));
}
self.pending
.iter()
.position(|p| p == prompt_id)
.map(|i| (false, self.running.len() + i))
}
}
#[derive(Clone)]
pub struct ComfyClient {
http: reqwest::Client,
base: String,
/// Client id sent with submissions.
client_id: String,
}
impl ComfyClient {
pub fn new(base_url: &str, timeout: Duration) -> Self {
let http = reqwest::Client::builder()
.timeout(timeout)
.connect_timeout(Duration::from_secs(3).min(timeout))
.build()
.expect("reqwest client");
ComfyClient {
http,
base: base_url.trim_end_matches('/').to_string(),
client_id: "imageapi".to_string(),
}
}
fn url(&self, path: &str) -> String {
format!("{}{}", self.base, path)
}
/// `GET /system_stats` with its own (short) timeout.
pub async fn probe(&self, timeout: Duration) -> bool {
self.http
.get(self.url("/system_stats"))
.timeout(timeout)
.send()
.await
.is_ok_and(|r| r.status().is_success())
}
pub async fn submit(&self, prompt: &Value) -> Result<String, SubmitError> {
let body = json!({ "prompt": prompt, "client_id": self.client_id });
let resp = self
.http
.post(self.url("/prompt"))
.json(&body)
.send()
.await
.map_err(|e| SubmitError::Comfy(e.into()))?;
let status = resp.status();
let text = resp
.text()
.await
.map_err(|e| SubmitError::Comfy(e.into()))?;
let json: Value = serde_json::from_str(&text).unwrap_or(Value::Null);
if status.is_success() {
// A 200 can still carry node_errors for outputs that were pruned;
// the prompt itself queued.
return json["prompt_id"]
.as_str()
.map(str::to_string)
.ok_or_else(|| {
SubmitError::Comfy(ComfyError::BadResponse(format!("no prompt_id in {text}")))
});
}
if status.as_u16() == 400 {
return Err(SubmitError::Rejected {
message: rejection_message(&json, &text),
node_errors: json["node_errors"]
.as_object()
.is_some_and(|m| !m.is_empty()),
});
}
Err(SubmitError::Comfy(ComfyError::Http {
status: status.as_u16(),
body: truncate(&text),
}))
}
pub async fn queue(&self) -> Result<QueueState, ComfyError> {
let json: Value = self.get_json("/queue").await?;
let ids = |key: &str| -> Vec<String> {
json[key]
.as_array()
.map(|a| {
a.iter()
.filter_map(|item| item.get(1).and_then(Value::as_str))
.map(str::to_string)
.collect()
})
.unwrap_or_default()
};
Ok(QueueState {
running: ids("queue_running"),
pending: ids("queue_pending"),
})
}
/// `GET /history/{id}`; `None` when ComfyUI doesn't know the id.
pub async fn history(&self, prompt_id: &str) -> Result<Option<HistoryEntry>, ComfyError> {
let json: Value = self
.get_json(&format!("/history/{}", urlencoding::encode(prompt_id)))
.await?;
Ok(json.get(prompt_id).map(parse_history_entry))
}
/// The newest finished run, for the idle check.
pub async fn newest_history(&self) -> Result<Option<HistoryEntry>, ComfyError> {
let json: Value = self.get_json("/history?max_items=1").await?;
Ok(json
.as_object()
.and_then(|m| m.values().next())
.map(parse_history_entry))
}
/// `GET /view`: the staged PNG. `Ok(None)` for 404.
pub async fn view(&self, r: &ViewRef) -> Result<Option<(String, Bytes)>, ComfyError> {
let resp = self
.http
.get(self.url("/view"))
.query(&[
("filename", r.filename.as_str()),
("subfolder", r.subfolder.as_str()),
("type", r.kind.as_str()),
])
.send()
.await?;
if resp.status().as_u16() == 404 {
return Ok(None);
}
if !resp.status().is_success() {
return Err(ComfyError::Http {
status: resp.status().as_u16(),
body: truncate(&resp.text().await.unwrap_or_default()),
});
}
let content_type = resp
.headers()
.get(reqwest::header::CONTENT_TYPE)
.and_then(|v| v.to_str().ok())
.unwrap_or("image/png")
.to_string();
Ok(Some((content_type, resp.bytes().await?)))
}
/// `POST /imageapi/approve`: ComfyUI writes the file; returns its path
/// as ComfyUI sees it.
pub async fn approve(&self, pending_id: &str) -> Result<String, ApproveError> {
let resp = self
.http
.post(self.url("/imageapi/approve"))
.json(&json!({ "id": pending_id }))
.send()
.await
.map_err(|e| ApproveError::Comfy(e.into()))?;
let status = resp.status();
let text = resp
.text()
.await
.map_err(|e| ApproveError::Comfy(e.into()))?;
let json: Value = serde_json::from_str(&text).unwrap_or(Value::Null);
if status.is_success() {
return json["path"].as_str().map(str::to_string).ok_or_else(|| {
ApproveError::Comfy(ComfyError::BadResponse(format!("no path in {text}")))
});
}
let message = json["error"].as_str().unwrap_or(&text).to_string();
match status.as_u16() {
404 => Err(ApproveError::Gone),
400 | 500 => Err(ApproveError::Refused(truncate(&message))),
s => Err(ApproveError::Comfy(ComfyError::Http {
status: s,
body: truncate(&text),
})),
}
}
/// `POST /free`: unload models and release cached memory.
pub async fn free(&self) -> Result<(), ComfyError> {
let resp = self
.http
.post(self.url("/free"))
.json(&json!({ "unload_models": true, "free_memory": true }))
.send()
.await?;
if resp.status().is_success() {
Ok(())
} else {
Err(ComfyError::Http {
status: resp.status().as_u16(),
body: truncate(&resp.text().await.unwrap_or_default()),
})
}
}
async fn get_json(&self, path: &str) -> Result<Value, ComfyError> {
let resp = self.http.get(self.url(path)).send().await?;
if !resp.status().is_success() {
return Err(ComfyError::Http {
status: resp.status().as_u16(),
body: truncate(&resp.text().await.unwrap_or_default()),
});
}
Ok(resp.json().await?)
}
}
fn truncate(s: &str) -> String {
s.chars().take(300).collect()
}
fn rejection_message(json: &Value, raw: &str) -> String {
let err = &json["error"];
let head = err["message"].as_str().or_else(|| err.as_str());
let detail = err["details"].as_str().filter(|d| !d.is_empty());
// Per-node errors say which input is wrong.
let nodes: Vec<String> = json["node_errors"]
.as_object()
.map(|m| {
m.values()
.filter_map(|n| {
let class = n["class_type"].as_str().unwrap_or("node");
let first = n["errors"].as_array()?.first()?;
Some(format!(
"{class}: {}",
first["message"].as_str().unwrap_or("invalid")
))
})
.collect()
})
.unwrap_or_default();
let mut parts: Vec<String> = Vec::new();
parts.extend(head.map(str::to_string));
parts.extend(detail.map(str::to_string));
parts.extend(nodes);
if parts.is_empty() {
truncate(raw)
} else {
truncate(&parts.join("; "))
}
}
pub(crate) fn parse_history_entry(entry: &Value) -> HistoryEntry {
let status = &entry["status"];
let status_str = status["status_str"].as_str().unwrap_or_default().to_string();
let completed = status["completed"].as_bool().unwrap_or(false);
let messages = status["messages"].as_array().cloned().unwrap_or_default();
let error = messages.iter().find_map(|m| {
let (kind, data) = (m.get(0)?.as_str()?, m.get(1)?);
(kind == "execution_error").then(|| {
data["exception_message"]
.as_str()
.unwrap_or("execution error")
.trim()
.to_string()
})
});
let timestamp_ms = messages
.iter()
.filter_map(|m| m.get(1)?.get("timestamp")?.as_i64())
.max();
let mut staged = Vec::new();
if let Some(outputs) = entry["outputs"].as_object() {
for out in outputs.values() {
let Some(results) = out["imageapi_versions"].as_array() else {
continue;
};
let images = out["images"].as_array().cloned().unwrap_or_default();
for (i, r) in results.iter().enumerate() {
let (Some(id), Some(image)) = (r["id"].as_str(), images.get(i)) else {
continue;
};
if let Ok(view) = serde_json::from_value::<ViewRef>(image.clone()) {
staged.push(Staged {
id: id.to_string(),
view,
});
}
}
}
}
HistoryEntry {
completed,
status_str,
error,
staged,
timestamp_ms,
}
}
+263
View File
@@ -0,0 +1,263 @@
//! Enhance configuration (#192): one `enhance.toml`, next to `llm.toml`.
//!
//! Path from `ENHANCE_CONFIG_PATH`, else `./enhance.toml`. A missing file at
//! the default path means Enhance is simply off (no `url`); a missing
//! explicit path is an error. Nothing here is a secret: ComfyUI has no auth
//! and stays LAN-only. `enhance.example.toml` is the annotated schema (a
//! test parses it).
use std::path::{Path, PathBuf};
use anyhow::{Context as _, Result, bail};
use serde::Deserialize;
use super::path_map::PathMap;
const DEFAULT_PATH: &str = "enhance.toml";
/// How the app's text combines with the template's own prompt.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize, serde::Serialize, Default)]
#[serde(rename_all = "lowercase")]
pub enum PromptMode {
/// The template's prompt is a full base instruction; the app's text is
/// appended after `prompt_join`. Empty text leaves the prompt as is.
#[default]
Append,
/// The app's text *is* the prompt (and is required).
Replace,
}
#[derive(Debug, Clone, PartialEq, Deserialize)]
#[serde(default, deny_unknown_fields)]
pub struct EnhanceConfig {
/// ComfyUI base URL, e.g. `http://192.168.10.218:8188`. Unset = Enhance
/// is disabled (`/enhance/templates` reports `available: false`).
pub url: Option<String>,
/// Per-request timeout for ComfyUI calls.
pub timeout_secs: u64,
/// How long the `available` probe result is cached.
pub probe_ttl_secs: u64,
/// Job poll interval.
pub poll_secs: u64,
/// Idle time (empty queue, newest history entry this old) before
/// ImageApi calls `POST /free` once. 0 disables it.
pub idle_free_secs: u64,
/// An active job whose ComfyUI stays unreachable this long becomes
/// `unavailable`.
pub unreachable_grace_secs: u64,
/// Template preselected by clients; default = the first one.
pub default_template: Option<String>,
/// Clients warn above this many megapixels in the original.
pub warn_above_megapixels: f64,
/// Library path prefix <-> ComfyUI path prefix. Longest prefix wins.
pub path_map: Vec<PathMapping>,
#[serde(rename = "template")]
pub templates: Vec<TemplateConfig>,
}
impl Default for EnhanceConfig {
fn default() -> Self {
Self {
url: None,
timeout_secs: 15,
probe_ttl_secs: 10,
poll_secs: 2,
idle_free_secs: 300,
unreachable_grace_secs: 120,
default_template: None,
warn_above_megapixels: 1.5,
path_map: Vec::new(),
templates: Vec::new(),
}
}
}
#[derive(Debug, Clone, PartialEq, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct PathMapping {
/// Prefix as ImageApi sees it, e.g. `/media/nas/`.
pub from: String,
/// Prefix as ComfyUI sees it, e.g. `Z:\` or `\\nas\`.
pub to: String,
}
#[derive(Debug, Clone, PartialEq, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct TemplateConfig {
/// Stable id clients send (`POST /enhance { template }`).
pub name: String,
pub label: String,
#[serde(default)]
pub description: String,
/// API-format workflow JSON (ComfyUI -> Workflow -> Export (API)).
/// Relative paths resolve against the config file's directory.
pub file: PathBuf,
#[serde(default)]
pub prompt_mode: PromptMode,
/// Text between the template's prompt and the app's text (append mode).
#[serde(default = "default_join")]
pub prompt_join: String,
/// Approximate output size, for the client's large-original warning.
#[serde(default = "default_output_mp")]
pub output_megapixels: f64,
}
fn default_join() -> String {
" ".into()
}
fn default_output_mp() -> f64 {
1.0
}
impl EnhanceConfig {
pub fn validate(&self) -> Result<()> {
if let Some(url) = &self.url
&& !(url.starts_with("http://") || url.starts_with("https://"))
{
bail!("url must start with http:// or https://, got {url:?}");
}
for (key, v) in [
("timeout_secs", self.timeout_secs),
("probe_ttl_secs", self.probe_ttl_secs),
("poll_secs", self.poll_secs),
("unreachable_grace_secs", self.unreachable_grace_secs),
] {
if v == 0 {
bail!("{key} must be greater than 0");
}
}
if !(self.warn_above_megapixels > 0.0) {
bail!("warn_above_megapixels must be greater than 0");
}
let mut seen = std::collections::HashSet::new();
for t in &self.templates {
let ok = !t.name.is_empty()
&& t.name
.chars()
.all(|c| c.is_ascii_alphanumeric() || c == '_' || c == '-');
if !ok {
bail!(
"template name {:?} must be non-empty and use only letters, digits, '_' and '-'",
t.name
);
}
if !seen.insert(t.name.as_str()) {
bail!("duplicate template name {:?}", t.name);
}
if t.label.trim().is_empty() {
bail!("template {:?} needs a label", t.name);
}
if !(t.output_megapixels > 0.0) {
bail!("template {:?}: output_megapixels must be > 0", t.name);
}
}
if let Some(d) = &self.default_template
&& !self.templates.iter().any(|t| &t.name == d)
{
bail!("default_template {d:?} is not a configured template");
}
// Parses every mapping (rejects empty prefixes).
PathMap::new(&self.path_map)?;
Ok(())
}
/// The preselected template's name.
pub fn default_name(&self) -> Option<&str> {
self.default_template
.as_deref()
.or_else(|| self.templates.first().map(|t| t.name.as_str()))
}
}
/// Where the config was read from, and what it said.
#[derive(Debug)]
pub struct Loaded {
pub config: EnhanceConfig,
/// Directory relative template `file`s resolve against.
pub base_dir: PathBuf,
/// The file, when one was read.
pub path: Option<PathBuf>,
}
/// Read `enhance.toml` (see module docs). A bad file is an error so a typo
/// fails startup instead of silently disabling the feature.
pub fn load() -> Result<Loaded> {
let (path, explicit) = match std::env::var("ENHANCE_CONFIG_PATH") {
Ok(p) if !p.trim().is_empty() => (PathBuf::from(p.trim()), true),
_ => (PathBuf::from(DEFAULT_PATH), false),
};
match std::fs::read_to_string(&path) {
Ok(text) => {
let config = parse(&text).with_context(|| format!("in {}", path.display()))?;
let base_dir = path
.parent()
.filter(|p| !p.as_os_str().is_empty())
.map(Path::to_path_buf)
.unwrap_or_else(|| PathBuf::from("."));
Ok(Loaded {
config,
base_dir,
path: Some(path),
})
}
Err(e) if e.kind() == std::io::ErrorKind::NotFound && !explicit => Ok(Loaded {
config: EnhanceConfig::default(),
base_dir: PathBuf::from("."),
path: None,
}),
Err(e) => Err(e).with_context(|| format!("reading enhance config {}", path.display())),
}
}
pub fn parse(text: &str) -> Result<EnhanceConfig> {
let config: EnhanceConfig = toml::from_str(text)?;
config.validate()?;
Ok(config)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn example_file_parses() {
let cfg = parse(include_str!("../../enhance.example.toml")).unwrap();
assert!(cfg.url.is_some());
assert_eq!(cfg.templates[0].name, "restore");
assert_eq!(cfg.default_name(), Some("restore"));
assert!(
cfg.templates
.iter()
.any(|t| t.prompt_mode == PromptMode::Replace)
);
}
#[test]
fn empty_file_is_disabled_defaults() {
let cfg = parse("").unwrap();
assert_eq!(cfg, EnhanceConfig::default());
assert!(cfg.url.is_none());
}
#[test]
fn rejects_unknown_keys_and_bad_values() {
assert!(parse("nope = 1").is_err());
assert!(parse("url = \"ftp://x\"").is_err());
assert!(parse("poll_secs = 0").is_err());
assert!(parse("default_template = \"x\"").is_err());
let dup = "[[template]]\nname='a'\nlabel='A'\nfile='a.json'\n[[template]]\nname='a'\nlabel='B'\nfile='b.json'";
assert!(parse(dup).is_err());
assert!(parse("[[template]]\nname='a b'\nlabel='A'\nfile='a.json'").is_err());
assert!(parse("[[path_map]]\nfrom=''\nto='Z:\\\\'").is_err());
}
#[test]
fn template_defaults() {
let cfg = parse("[[template]]\nname='a'\nlabel='A'\nfile='a.json'").unwrap();
let t = &cfg.templates[0];
assert_eq!(t.prompt_mode, PromptMode::Append);
assert_eq!(t.prompt_join, " ");
assert_eq!(t.output_megapixels, 1.0);
}
}
+515
View File
@@ -0,0 +1,515 @@
//! HTTP surface of Enhance (#192). ImageApi relays everything, previews
//! included: the apps never talk to ComfyUI.
//!
//! - `GET /enhance/templates`
//! - `POST /enhance`
//! - `GET /enhance/jobs?path=`
//! - `GET /enhance/jobs/{id}`
//! - `GET /enhance/jobs/{id}/preview/{n}`
//! - `POST /enhance/jobs/{id}/approve`
//! - `POST /enhance/jobs/{id}/discard`
use std::path::{Component, Path, PathBuf};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use actix_web::{HttpResponse, Responder, get, post, web};
use log::warn;
use serde::{Deserialize, Serialize};
use serde_json::json;
use super::config::PromptMode;
use super::service::{EnhanceError, EnhanceService, SourceFile};
use crate::data::Claims;
use crate::database::ExifDao;
use crate::database::enhance_dao::{EnhanceJobRow, STATUS_APPROVED, STATUS_READY};
use crate::libraries::{self, Library};
use crate::state::AppState;
use crate::utils::normalize_path;
use crate::versions::handlers::build_response;
type ExifData = web::Data<Mutex<Box<dyn ExifDao>>>;
fn error_response(e: &EnhanceError) -> HttpResponse {
let (mut builder, code) = match e {
EnhanceError::Disabled => (HttpResponse::ServiceUnavailable(), "disabled"),
EnhanceError::Unavailable => (HttpResponse::ServiceUnavailable(), "unavailable"),
EnhanceError::UnknownTemplate(_) => (HttpResponse::NotFound(), "unknown_template"),
EnhanceError::BadRequest(_) => (HttpResponse::BadRequest(), "bad_request"),
EnhanceError::Rejected(_) => (HttpResponse::UnprocessableEntity(), "rejected"),
EnhanceError::Expired => (HttpResponse::Gone(), "expired"),
EnhanceError::NotFound => (HttpResponse::NotFound(), "not_found"),
EnhanceError::Unplaceable(_) => (HttpResponse::BadGateway(), "unplaceable"),
EnhanceError::Upstream(_) => (HttpResponse::BadGateway(), "upstream"),
EnhanceError::Internal(_) => (HttpResponse::InternalServerError(), "internal"),
};
builder.json(json!({ "error": e.message(), "code": code }))
}
fn bad(msg: &str) -> HttpResponse {
error_response(&EnhanceError::BadRequest(msg.to_string()))
}
// ---- views --------------------------------------------------------------
#[derive(Debug, Serialize)]
struct PendingView {
/// Index for `preview/{n}` and `approve { n }`.
n: usize,
/// Path of the preview on this API.
preview: String,
approved: bool,
}
#[derive(Debug, Serialize)]
struct SavedView {
n: usize,
path: Option<String>,
library_id: Option<i32>,
}
#[derive(Debug, Serialize)]
pub struct JobView {
id: String,
status: String,
#[serde(skip_serializing_if = "Option::is_none")]
queue_position: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
error: Option<String>,
path: String,
library_id: i32,
template: String,
#[serde(skip_serializing_if = "Option::is_none")]
prompt: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
reference_path: Option<String>,
created_at: i64,
updated_at: i64,
pending: Vec<PendingView>,
saved: Vec<SavedView>,
}
pub fn job_view(job: &EnhanceJobRow) -> JobView {
let saved = EnhanceService::saved_of(job);
// An expired job's staged files are gone; don't offer them.
let pending = if job.status == STATUS_READY || job.status == STATUS_APPROVED {
EnhanceService::pending_of(job)
.iter()
.enumerate()
.map(|(n, _)| PendingView {
n,
preview: format!("/enhance/jobs/{}/preview/{n}", job.id),
approved: saved.iter().any(|s| s.n == n),
})
.collect()
} else {
Vec::new()
};
let queued = matches!(job.status.as_str(), "queued" | "running");
JobView {
id: job.id.clone(),
status: job.status.clone(),
queue_position: job.queue_position.filter(|_| queued),
error: job.error.clone(),
path: job.rel_path.clone(),
library_id: job.library_id,
template: job.template.clone(),
prompt: job.prompt.clone(),
reference_path: job.reference_path.clone(),
created_at: job.created_at,
updated_at: job.updated_at,
pending,
saved: saved
.into_iter()
.map(|s| SavedView {
n: s.n,
path: s.rel_path,
library_id: s.library_id,
})
.collect(),
}
}
// ---- file resolution ----------------------------------------------------
fn safe_rel(rel: &str) -> bool {
!rel.is_empty()
&& Path::new(rel)
.components()
.all(|c| matches!(c, Component::Normal(_)))
}
/// An indexed, hashed file in a library.
fn resolve_file(
app_state: &AppState,
exif_dao: &Mutex<Box<dyn ExifDao>>,
path: &str,
library: Option<&str>,
) -> Result<SourceFile, HttpResponse> {
let library_id = match libraries::resolve_library_param_state(app_state, library) {
Ok(l) => l.map(|l| l.id),
Err(msg) => return Err(bad(&msg)),
};
let rel = normalize_path(path);
if !safe_rel(&rel) {
return Err(bad("invalid path"));
}
let mut rows = exif_dao
.lock()
.expect("Unable to lock ExifDao")
.get_exif_batch(library_id, std::slice::from_ref(&rel))
.map_err(|_| error_response(&EnhanceError::Internal("lookup failed".into())))?;
rows.sort_by_key(|r| r.id);
let Some(row) = rows.iter().find(|r| r.content_hash.is_some()) else {
return Err(error_response(&EnhanceError::NotFound));
};
let Some(lib) = app_state.library_by_id(row.library_id) else {
return Err(error_response(&EnhanceError::NotFound));
};
let abs = Path::new(&lib.root_path).join(&rel);
if !abs.is_file() {
return Err(error_response(&EnhanceError::NotFound));
}
Ok(SourceFile {
library_id: row.library_id,
rel_path: rel,
abs,
hash: row.content_hash.clone().unwrap_or_default(),
})
}
/// Library and relative path of an absolute file ComfyUI wrote. The job's own
/// library wins when roots overlap.
fn place_in_libraries(
libs: &[Library],
prefer: i32,
abs: &Path,
) -> Option<(i32, String)> {
let mut ordered: Vec<&Library> = libs.iter().collect();
ordered.sort_by_key(|l| l.id != prefer);
for lib in ordered {
if let Ok(rest) = abs.strip_prefix(&lib.root_path) {
let rel = rest.to_str()?.replace('\\', "/");
if safe_rel(&rel) {
return Some((lib.id, rel));
}
}
}
None
}
// ---- templates ----------------------------------------------------------
#[derive(Serialize)]
struct TemplateView {
name: String,
label: String,
description: String,
accepts_prompt: bool,
prompt_mode: PromptMode,
/// Replace-mode templates need non-empty text (empty is a 400).
prompt_required: bool,
accepts_reference: bool,
default: bool,
/// Approximate output size; clients warn when the original is larger.
output_megapixels: f64,
warn_above_megapixels: f64,
}
#[get("/enhance/templates")]
pub async fn get_templates(_: Claims, svc: web::Data<EnhanceService>) -> impl Responder {
let available = svc.available().await;
let reason = if !svc.is_configured() {
Some(super::service::MSG_DISABLED)
} else if !available {
Some(super::service::MSG_UNAVAILABLE)
} else {
None
};
let default = svc.default_template().map(str::to_string);
let templates: Vec<TemplateView> = svc
.templates()
.iter()
.map(|t| TemplateView {
name: t.cfg.name.clone(),
label: t.cfg.label.clone(),
description: t.cfg.description.clone(),
accepts_prompt: t.accepts_prompt(),
prompt_mode: t.cfg.prompt_mode,
prompt_required: t.prompt_required(),
accepts_reference: t.accepts_reference(),
default: default.as_deref() == Some(t.cfg.name.as_str()),
output_megapixels: t.cfg.output_megapixels,
warn_above_megapixels: svc.config.warn_above_megapixels,
})
.collect();
let mut body = json!({
"available": available,
"default": default,
"warn_above_megapixels": svc.config.warn_above_megapixels,
"templates": templates,
});
if let Some(r) = reason {
body["reason"] = json!(r);
}
HttpResponse::Ok().json(body)
}
// ---- jobs ---------------------------------------------------------------
#[derive(Debug, Deserialize)]
pub struct LibraryQuery {
pub library: Option<String>,
}
#[derive(Debug, Deserialize)]
pub struct SubmitBody {
pub path: String,
pub template: String,
#[serde(default)]
pub prompt: Option<String>,
#[serde(default)]
pub reference_path: Option<String>,
#[serde(default)]
pub library: Option<String>,
}
#[post("/enhance")]
#[tracing::instrument(name = "enhance_submit", skip_all)]
pub async fn submit(
_: Claims,
query: web::Query<LibraryQuery>,
body: web::Json<SubmitBody>,
svc: web::Data<EnhanceService>,
app_state: web::Data<AppState>,
exif_dao: ExifData,
) -> impl Responder {
if !svc.is_configured() {
return error_response(&EnhanceError::Disabled);
}
let library = body.library.as_deref().or(query.library.as_deref());
let photo = match resolve_file(&app_state, &exif_dao, &body.path, library) {
Ok(f) => f,
Err(r) => return r,
};
let reference = match body.reference_path.as_deref().filter(|p| !p.trim().is_empty()) {
Some(p) => {
// The reference lives in the photo's own library.
let lib = app_state.library_by_id(photo.library_id).map(|l| l.name.clone());
match resolve_file(&app_state, &exif_dao, p, lib.as_deref()) {
Ok(f) => Some(f),
Err(r) => return r,
}
}
None => None,
};
match svc
.submit(&photo, &body.template, body.prompt.as_deref(), reference.as_ref())
.await
{
Ok(job) => HttpResponse::Ok().json(job_view(&job)),
Err(e) => error_response(&e),
}
}
#[get("/enhance/jobs/{id}")]
pub async fn get_job(
_: Claims,
id: web::Path<String>,
svc: web::Data<EnhanceService>,
) -> impl Responder {
match svc.get_job(&id) {
Ok(job) => HttpResponse::Ok().json(job_view(&job)),
Err(e) => error_response(&e),
}
}
#[derive(Debug, Deserialize)]
pub struct JobsQuery {
pub path: String,
pub library: Option<String>,
}
/// A photo's recent jobs, newest first, so the viewer can pick up an enhance
/// in progress after reopening. Discarded jobs are not listed.
#[get("/enhance/jobs")]
pub async fn list_jobs(
_: Claims,
q: web::Query<JobsQuery>,
svc: web::Data<EnhanceService>,
app_state: web::Data<AppState>,
exif_dao: ExifData,
) -> impl Responder {
let file = match resolve_file(&app_state, &exif_dao, &q.path, q.library.as_deref()) {
Ok(f) => f,
Err(r) => return r,
};
match svc.jobs_for_hash(&file.hash) {
Ok(jobs) => {
let views: Vec<JobView> = jobs.iter().map(job_view).collect();
HttpResponse::Ok().json(json!({ "jobs": views }))
}
Err(e) => error_response(&e),
}
}
#[get("/enhance/jobs/{id}/preview/{n}")]
pub async fn preview(
_: Claims,
path: web::Path<(String, usize)>,
svc: web::Data<EnhanceService>,
) -> impl Responder {
let (id, n) = path.into_inner();
let job = match svc.get_job(&id) {
Ok(j) => j,
Err(e) => return error_response(&e),
};
match svc.preview(&job, n).await {
Ok((content_type, bytes)) => HttpResponse::Ok()
.content_type(content_type)
.insert_header(("Cache-Control", "private, max-age=3600"))
.body(bytes),
Err(e) => error_response(&e),
}
}
#[post("/enhance/jobs/{id}/discard")]
pub async fn discard(
_: Claims,
id: web::Path<String>,
svc: web::Data<EnhanceService>,
) -> impl Responder {
match svc.discard(&id) {
Ok(job) => HttpResponse::Ok().json(job_view(&job)),
Err(e) => error_response(&e),
}
}
#[derive(Debug, Deserialize)]
pub struct ApproveBody {
#[serde(default)]
pub n: usize,
}
/// Ask ComfyUI to write result `n` into the library, ingest that file now,
/// and answer with the version (an `/image/versions` `versions[]` entry).
#[post("/enhance/jobs/{id}/approve")]
#[tracing::instrument(name = "enhance_approve", skip_all)]
pub async fn approve(
_: Claims,
id: web::Path<String>,
body: Option<web::Json<ApproveBody>>,
svc: web::Data<EnhanceService>,
app_state: web::Data<AppState>,
exif_dao: ExifData,
) -> impl Responder {
let n = body.map(|b| b.n).unwrap_or(0);
let job = match svc.get_job(&id) {
Ok(j) => j,
Err(e) => return error_response(&e),
};
let libs = app_state.libraries.clone();
let prefer = job.library_id;
let place = move |abs: &Path| place_in_libraries(&libs, prefer, abs);
let (job, saved) = match svc.approve(&id, n, &place).await {
Ok(v) => v,
Err(e) => return error_response(&e),
};
let (Some(library_id), Some(rel)) = (saved.library_id, saved.rel_path.clone()) else {
return error_response(&EnhanceError::Unplaceable(
"the saved file isn't in a library".into(),
));
};
let Some(library) = app_state.library_by_id(library_id).cloned() else {
return error_response(&EnhanceError::Unplaceable("unknown library".into()));
};
let abs = PathBuf::from(&library.root_path).join(&rel);
// The file was written by another machine over a share; give a lagging
// mount a moment to show it.
for _ in 0..25 {
if abs.is_file() {
break;
}
tokio::time::sleep(Duration::from_millis(200)).await;
}
if !abs.is_file() {
return error_response(&EnhanceError::Unplaceable(format!(
"ComfyUI saved {rel} but it isn't visible to ImageApi yet; it will appear after the next scan"
)));
}
let version_dao = app_state.version_dao.clone();
let exif_arc: Arc<Mutex<Box<dyn ExifDao>>> = exif_dao.clone().into_inner();
let (lib2, rel2, abs2) = (library.clone(), rel.clone(), abs.clone());
let ingested = web::block(move || {
ingest_saved(&lib2, &rel2, &abs2, &exif_arc, &version_dao)
})
.await;
let hash = match ingested {
Ok(Some(h)) => h,
_ => {
warn!("enhance: ingest of {rel} failed");
return HttpResponse::Accepted().json(json!({
"linked": false,
"path": rel,
"library_id": library_id,
"job": job_view(&job),
}));
}
};
// Which version row did that become?
let entry = {
let mut dao = app_state.version_dao.lock().expect("Unable to lock VersionDao");
let original = match dao.get_link(&hash) {
Ok(Some(link)) => link.original_hash,
_ => job.original_hash.clone(),
};
build_response(dao.as_mut(), &original, job.library_id)
.ok()
.and_then(|r| r.versions.into_iter().find(|v| v.hash == hash))
};
match entry {
Some(v) => HttpResponse::Ok().json(v),
None => HttpResponse::Accepted().json(json!({
"linked": false,
"path": rel,
"library_id": library_id,
"job": job_view(&job),
})),
}
}
/// Register the saved file (if the watcher hasn't yet) and link it as a
/// version. Idempotent. Returns the file's content hash.
pub fn ingest_saved(
library: &Library,
rel: &str,
abs: &Path,
exif_dao: &Arc<Mutex<Box<dyn ExifDao>>>,
version_dao: &Arc<Mutex<Box<dyn crate::database::VersionDao>>>,
) -> Option<String> {
let rel_s = rel.to_string();
let existing = exif_dao
.lock()
.expect("Unable to lock ExifDao")
.get_exif_batch(Some(library.id), std::slice::from_ref(&rel_s))
.ok()?;
let file = vec![(abs.to_path_buf(), rel_s.clone())];
if existing.is_empty() {
crate::watcher::register_new_files(library, file.clone(), exif_dao);
}
let linked = crate::versions::link_new_files(library, &file, exif_dao, version_dao)
+ crate::versions::sweep_name_links(library, exif_dao, version_dao);
if linked > 0 {
crate::duplicates::invalidate_caches();
}
exif_dao
.lock()
.expect("Unable to lock ExifDao")
.get_exif_batch(Some(library.id), std::slice::from_ref(&rel_s))
.ok()?
.into_iter()
.find_map(|r| r.content_hash)
}
View File
Whitespace-only changes.
+25
View File
@@ -0,0 +1,25 @@
//! Enhance on demand (#192): run a ComfyUI restore/edit workflow on a photo
//! from the apps, approve the result in-app, and have it land in the library
//! as a version (#191).
//!
//! - [`config`]: `enhance.toml` (templates, path map, timings).
//! - [`template`]: workflow loading/validation and per-photo filling.
//! - [`comfy`]: the ComfyUI HTTP calls used.
//! - [`service`]: submit, poll, approve, idle `/free`.
//! - [`handlers`]: the HTTP endpoints.
//!
//! ImageApi never writes under a library root: ComfyUI writes the approved
//! file, ImageApi only ingests it.
pub mod comfy;
pub mod config;
pub mod handlers;
pub mod path_map;
pub mod service;
pub mod template;
#[cfg(test)]
mod integration_tests;
pub use handlers::{approve, discard, get_job, get_templates, list_jobs, preview, submit};
pub use service::EnhanceService;
+174
View File
@@ -0,0 +1,174 @@
//! Library path <-> ComfyUI path mapping (#192).
//!
//! ImageApi sees `/media/nas/Media/Pictures/a.jpg`; ComfyUI (on Windows)
//! sees `Z:\Media\Pictures\a.jpg` or `\\nas\Media\Pictures\a.jpg`. Each
//! mapping is a prefix pair; the longest matching prefix wins and the
//! separators are converted. Used in both directions: library -> ComfyUI on
//! submit, and ComfyUI -> library to recognise the path `approve` returns.
//! With no mappings configured, paths pass through unchanged.
use std::path::{Path, PathBuf};
use anyhow::{Result, bail};
use super::config::PathMapping;
#[derive(Debug, Clone)]
struct Entry {
/// Library side, `/`-separated, always ending in `/`.
from: String,
/// ComfyUI side without a trailing separator.
to: String,
/// Separator ComfyUI's side uses.
to_sep: char,
}
#[derive(Debug, Clone, Default)]
pub struct PathMap {
entries: Vec<Entry>,
}
impl PathMap {
pub fn new(mappings: &[PathMapping]) -> Result<Self> {
let mut entries = Vec::new();
for m in mappings {
let from = m.from.replace('\\', "/");
let to = m.to.trim();
if from.trim_matches('/').is_empty() || to.is_empty() {
bail!("path_map entries need non-empty from and to ({m:?})");
}
let to_sep = if to.contains('\\') { '\\' } else { '/' };
entries.push(Entry {
from: format!("{}/", from.trim_end_matches('/')),
to: to.trim_end_matches(['/', '\\']).to_string(),
to_sep,
});
}
Ok(PathMap { entries })
}
pub fn is_identity(&self) -> bool {
self.entries.is_empty()
}
/// The path ComfyUI should open for a library file. `None` when mappings
/// exist but none covers the file.
pub fn to_comfy(&self, abs: &Path) -> Option<String> {
let text = abs.to_str()?.replace('\\', "/");
if self.entries.is_empty() {
return Some(abs.to_string_lossy().into_owned());
}
let best = self
.entries
.iter()
.filter(|e| text.starts_with(e.from.as_str()))
.max_by_key(|e| e.from.len())?;
let rest = &text[best.from.len()..];
let sep = best.to_sep.to_string();
Some(format!("{}{}{}", best.to, sep, rest.replace('/', &sep)))
}
/// The library-side absolute path of a path ComfyUI reported. Matching
/// is case-insensitive (Windows). `None` when no mapping covers it or it
/// tries to climb out with `..`.
pub fn from_comfy(&self, comfy: &str) -> Option<PathBuf> {
let text = comfy.trim().replace('\\', "/");
if self.entries.is_empty() {
return Some(PathBuf::from(text));
}
let best = self
.entries
.iter()
.filter_map(|e| {
let prefix = format!("{}/", e.to.replace('\\', "/"));
let head = text.get(..prefix.len())?;
head.eq_ignore_ascii_case(&prefix).then_some((e, prefix.len()))
})
.max_by_key(|(_, len)| *len)?;
let rest = &text[best.1..];
if rest.split('/').any(|c| c == "..") {
return None;
}
Some(PathBuf::from(format!("{}{}", best.0.from, rest)))
}
}
#[cfg(test)]
mod tests {
use super::*;
fn map(pairs: &[(&str, &str)]) -> PathMap {
let m: Vec<PathMapping> = pairs
.iter()
.map(|(f, t)| PathMapping {
from: f.to_string(),
to: t.to_string(),
})
.collect();
PathMap::new(&m).unwrap()
}
#[test]
fn drive_letter_round_trip() {
let m = map(&[("/media/nas/", "Z:\\")]);
let c = m
.to_comfy(Path::new("/media/nas/Media/Pictures/a b.jpg"))
.unwrap();
assert_eq!(c, "Z:\\Media\\Pictures\\a b.jpg");
assert_eq!(
m.from_comfy(&c).unwrap(),
PathBuf::from("/media/nas/Media/Pictures/a b.jpg")
);
// Drive letters are case-insensitive.
assert_eq!(
m.from_comfy("z:\\x\\y.png").unwrap(),
PathBuf::from("/media/nas/x/y.png")
);
}
#[test]
fn unc_target_and_missing_trailing_slash() {
let m = map(&[("/media/nas", "\\\\nas\\")]);
assert_eq!(
m.to_comfy(Path::new("/media/nas/Media/a.jpg")).unwrap(),
"\\\\nas\\Media\\a.jpg"
);
assert_eq!(
m.from_comfy("\\\\nas\\Media\\a.jpg").unwrap(),
PathBuf::from("/media/nas/Media/a.jpg")
);
}
#[test]
fn longest_prefix_wins_both_ways() {
let m = map(&[("/media/nas/", "Z:\\"), ("/media/nas/Media/", "Y:\\")]);
assert_eq!(
m.to_comfy(Path::new("/media/nas/Media/a.jpg")).unwrap(),
"Y:\\a.jpg"
);
assert_eq!(
m.to_comfy(Path::new("/media/nas/Other/a.jpg")).unwrap(),
"Z:\\Other\\a.jpg"
);
assert_eq!(
m.from_comfy("Y:\\a.jpg").unwrap(),
PathBuf::from("/media/nas/Media/a.jpg")
);
}
#[test]
fn uncovered_and_escaping_paths_are_none() {
let m = map(&[("/media/nas/", "Z:\\")]);
assert!(m.to_comfy(Path::new("/elsewhere/a.jpg")).is_none());
assert!(m.to_comfy(Path::new("/media/nasty/a.jpg")).is_none());
assert!(m.from_comfy("C:\\a.png").is_none());
assert!(m.from_comfy("Z:\\..\\etc\\passwd").is_none());
}
#[test]
fn no_mappings_is_identity() {
let m = PathMap::default();
assert_eq!(m.to_comfy(Path::new("/a/b.jpg")).unwrap(), "/a/b.jpg");
assert_eq!(m.from_comfy("/a/b.png").unwrap(), PathBuf::from("/a/b.png"));
}
}
+733
View File
@@ -0,0 +1,733 @@
//! Enhance orchestration (#192): submit, poll, approve, idle `/free`.
//!
//! The job row (`enhance_jobs`) is the only state that matters. ComfyUI is
//! allowed to be down (a normal state) and to forget everything when it
//! restarts (unapproved results are cleared), so every transition tolerates
//! both.
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use log::{info, warn};
use serde::{Deserialize, Serialize};
use super::comfy::{
ApproveError, ComfyClient, ComfyError, HistoryEntry, Staged, SubmitError,
};
use super::config::{EnhanceConfig, Loaded, PromptMode};
use super::path_map::PathMap;
use super::template::Template;
use crate::database::enhance_dao::{
EnhanceJobDao, EnhanceJobRow, JobUpdate, STATUS_APPROVED, STATUS_EXPIRED, STATUS_FAILED,
STATUS_QUEUED, STATUS_READY, STATUS_RUNNING, STATUS_UNAVAILABLE,
};
/// Short timeout for the `available` probe, so the templates call stays snappy
/// when ComfyUI's host is off.
const PROBE_TIMEOUT: Duration = Duration::from_secs(2);
/// How often the idle check looks at ComfyUI.
const IDLE_CHECK_EVERY: Duration = Duration::from_secs(30);
/// Finished jobs are kept this long.
const JOB_RETENTION_SECS: i64 = 14 * 24 * 3600;
/// Jobs shown per photo.
pub const JOBS_PER_PHOTO: i64 = 20;
pub const MSG_UNAVAILABLE: &str = "ComfyUI isn't running";
pub const MSG_EXPIRED: &str =
"This result expired (ComfyUI was restarted). Run it again.";
pub const MSG_DISABLED: &str = "Enhance isn't configured";
/// A library file a request names.
#[derive(Debug, Clone)]
pub struct SourceFile {
pub library_id: i32,
pub rel_path: String,
pub abs: PathBuf,
pub hash: String,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum EnhanceError {
/// No `url` configured.
Disabled,
/// ComfyUI isn't reachable.
Unavailable,
UnknownTemplate(String),
BadRequest(String),
/// ComfyUI refused the workflow.
Rejected(String),
/// The result is gone (ComfyUI restarted).
Expired,
NotFound,
/// Approved, but ImageApi can't place the file in a library.
Unplaceable(String),
Upstream(String),
Internal(String),
}
impl EnhanceError {
pub fn message(&self) -> String {
match self {
EnhanceError::Disabled => MSG_DISABLED.into(),
EnhanceError::Unavailable => MSG_UNAVAILABLE.into(),
EnhanceError::UnknownTemplate(t) => format!("unknown template {t:?}"),
EnhanceError::BadRequest(m)
| EnhanceError::Rejected(m)
| EnhanceError::Unplaceable(m)
| EnhanceError::Upstream(m)
| EnhanceError::Internal(m) => m.clone(),
EnhanceError::Expired => MSG_EXPIRED.into(),
EnhanceError::NotFound => "not found".into(),
}
}
}
fn comfy_err(e: ComfyError) -> EnhanceError {
match e {
ComfyError::Unreachable(_) => EnhanceError::Unavailable,
other => EnhanceError::Upstream(other.to_string()),
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SavedEntry {
pub n: usize,
pub comfy_path: String,
pub library_id: Option<i32>,
pub rel_path: Option<String>,
}
/// What the idle check decided.
#[derive(Debug, Default)]
pub struct IdleState {
initialised: bool,
marker: Option<i64>,
seen_at_ms: i64,
freed: bool,
}
impl IdleState {
/// One observation of a reachable ComfyUI. `marker` identifies the
/// newest history entry (its last event time): when it changes, or the
/// queue isn't empty, or one of our jobs is active, ComfyUI is in use.
/// Returns true once per quiet spell when it has been idle for
/// `idle_ms` and `/free` should be called.
///
/// The marker's *value* is ComfyUI's clock; only changes in it matter,
/// and "how long ago" is measured on ImageApi's clock, so clock skew
/// between the hosts can't cause an early unload. After ImageApi
/// (re)starts the first observation counts as activity.
pub fn observe(
&mut self,
now_ms: i64,
queue_empty: bool,
active_jobs: usize,
marker: Option<i64>,
idle_ms: i64,
) -> bool {
if !self.initialised || marker != self.marker || !queue_empty || active_jobs > 0 {
self.initialised = true;
self.marker = marker;
self.seen_at_ms = now_ms;
self.freed = false;
return false;
}
if !self.freed && now_ms - self.seen_at_ms >= idle_ms {
self.freed = true;
return true;
}
false
}
/// ComfyUI went away: the next observation starts a fresh baseline.
pub fn reset(&mut self) {
*self = IdleState::default();
}
/// `/free` failed; try again on the next check.
pub fn free_failed(&mut self) {
self.freed = false;
}
}
pub struct EnhanceService {
pub config: EnhanceConfig,
templates: Vec<Template>,
client: Option<ComfyClient>,
pub path_map: PathMap,
dao: Arc<Mutex<Box<dyn EnhanceJobDao>>>,
probe: Mutex<Option<(Instant, bool)>>,
unreachable_since: Mutex<HashMap<String, Instant>>,
idle: Mutex<IdleState>,
/// Serialises approvals (rare; keeps "approve twice" simple).
approve_lock: tokio::sync::Mutex<()>,
}
impl EnhanceService {
/// Load the templates named in `loaded`, logging (not failing on) bad
/// ones: one broken template shouldn't take the rest down.
pub fn from_loaded(
loaded: Loaded,
dao: Arc<Mutex<Box<dyn EnhanceJobDao>>>,
) -> anyhow::Result<Self> {
let mut templates = Vec::new();
for cfg in &loaded.config.templates {
match Template::load(cfg, &loaded.base_dir) {
Ok(t) => {
for w in &t.warnings {
warn!("enhance: template '{}': {w}", cfg.name);
}
info!(
"enhance: template '{}' loaded (prompt: {}{}, reference: {})",
cfg.name,
if t.accepts_prompt() { "yes" } else { "no" },
if t.accepts_prompt() {
match cfg.prompt_mode {
PromptMode::Append => " append",
PromptMode::Replace => " replace",
}
} else {
""
},
if t.accepts_reference() { "yes" } else { "no" },
);
templates.push(t);
}
Err(e) => warn!("enhance: template '{}' rejected: {e:#}", cfg.name),
}
}
match &loaded.path {
Some(p) if loaded.config.url.is_some() => info!(
"enhance: config {} ({} of {} templates usable)",
p.display(),
templates.len(),
loaded.config.templates.len()
),
Some(p) => info!("enhance: config {} has no url; Enhance is off", p.display()),
None => info!("enhance: no enhance.toml; Enhance is off"),
}
Self::new(loaded.config, templates, dao)
}
pub fn new(
config: EnhanceConfig,
templates: Vec<Template>,
dao: Arc<Mutex<Box<dyn EnhanceJobDao>>>,
) -> anyhow::Result<Self> {
let client = config
.url
.as_deref()
.map(|u| ComfyClient::new(u, Duration::from_secs(config.timeout_secs)));
let path_map = PathMap::new(&config.path_map)?;
Ok(EnhanceService {
config,
templates,
client,
path_map,
dao,
probe: Mutex::new(None),
unreachable_since: Mutex::new(HashMap::new()),
idle: Mutex::new(IdleState::default()),
approve_lock: tokio::sync::Mutex::new(()),
})
}
pub fn is_configured(&self) -> bool {
self.client.is_some()
}
pub fn templates(&self) -> &[Template] {
&self.templates
}
pub fn template(&self, name: &str) -> Option<&Template> {
self.templates.iter().find(|t| t.cfg.name == name)
}
pub fn default_template(&self) -> Option<&str> {
let wanted = self.config.default_name()?;
self.template(wanted)
.map(|t| t.cfg.name.as_str())
.or_else(|| self.templates.first().map(|t| t.cfg.name.as_str()))
}
/// Cached `GET /system_stats` probe.
pub async fn available(&self) -> bool {
let Some(client) = &self.client else {
return false;
};
let ttl = Duration::from_secs(self.config.probe_ttl_secs);
if let Some((at, up)) = *self.probe.lock().unwrap()
&& at.elapsed() < ttl
{
return up;
}
let up = client.probe(PROBE_TIMEOUT).await;
*self.probe.lock().unwrap() = Some((Instant::now(), up));
up
}
/// Fresh probe (ignores and refreshes the cache): a submit shouldn't be
/// refused because ComfyUI came up since the last probe.
async fn probe_now(&self) -> bool {
let Some(client) = &self.client else {
return false;
};
let up = client.probe(PROBE_TIMEOUT).await;
*self.probe.lock().unwrap() = Some((Instant::now(), up));
up
}
// ---- jobs -----------------------------------------------------------
pub async fn submit(
&self,
photo: &SourceFile,
template_name: &str,
prompt: Option<&str>,
reference: Option<&SourceFile>,
) -> Result<EnhanceJobRow, EnhanceError> {
let Some(client) = &self.client else {
return Err(EnhanceError::Disabled);
};
let template = self
.template(template_name)
.ok_or_else(|| EnhanceError::UnknownTemplate(template_name.to_string()))?;
let unmapped = |p: &Path| {
EnhanceError::Unplaceable(format!(
"{} isn't under any path_map entry, so ComfyUI can't open it",
p.display()
))
};
let source = self
.path_map
.to_comfy(&photo.abs)
.ok_or_else(|| unmapped(&photo.abs))?;
let reference_path = match reference {
Some(r) => Some(
self.path_map
.to_comfy(&r.abs)
.ok_or_else(|| unmapped(&r.abs))?,
),
None => None,
};
let filled = template
.fill(
&source,
prompt,
reference_path.as_deref(),
&mut || rand::random::<u64>() >> 12,
)
.map_err(|e| EnhanceError::BadRequest(e.0))?;
if !self.probe_now().await {
return Err(EnhanceError::Unavailable);
}
let prompt_id = match client.submit(&filled.prompt).await {
Ok(id) => id,
Err(SubmitError::Comfy(e)) => return Err(comfy_err(e)),
Err(SubmitError::Rejected { message, node_errors }) => {
// A reference-less run of a template that wants one fails
// validation on the dropped node's consumers.
return Err(if filled.dropped_reference && node_errors {
EnhanceError::Rejected(format!(
"this template needs a reference photo ({message})"
))
} else {
EnhanceError::Rejected(message)
});
}
};
let now = chrono::Utc::now().timestamp();
let user_prompt = prompt.map(str::trim).filter(|p| !p.is_empty());
let row = EnhanceJobRow {
id: uuid::Uuid::new_v4().to_string(),
original_hash: photo.hash.clone(),
library_id: photo.library_id,
rel_path: photo.rel_path.clone(),
template: template.cfg.name.clone(),
prompt: user_prompt.map(str::to_string),
reference_hash: reference.map(|r| r.hash.clone()),
reference_path: reference.map(|r| r.rel_path.clone()),
comfy_prompt_id: prompt_id,
status: STATUS_QUEUED.to_string(),
error: None,
queue_position: None,
pending: "[]".to_string(),
saved: "[]".to_string(),
discarded: 0,
created_at: now,
updated_at: now,
};
self.dao
.lock()
.unwrap()
.insert(&row)
.map_err(|e| EnhanceError::Internal(format!("saving the job failed: {e}")))?;
info!(
"enhance: job {} submitted ({} for {}, prompt {})",
row.id, row.template, row.rel_path, row.comfy_prompt_id
);
Ok(row)
}
pub fn get_job(&self, id: &str) -> Result<EnhanceJobRow, EnhanceError> {
self.dao
.lock()
.unwrap()
.get(id)
.map_err(|e| EnhanceError::Internal(e.to_string()))?
.ok_or(EnhanceError::NotFound)
}
pub fn jobs_for_hash(&self, hash: &str) -> Result<Vec<EnhanceJobRow>, EnhanceError> {
self.dao
.lock()
.unwrap()
.list_for_hash(hash, JOBS_PER_PHOTO)
.map_err(|e| EnhanceError::Internal(e.to_string()))
}
/// Hide a job from the per-photo list. Nothing is deleted on ComfyUI:
/// staged results vanish on its restart anyway.
pub fn discard(&self, id: &str) -> Result<EnhanceJobRow, EnhanceError> {
self.get_job(id)?;
self.dao
.lock()
.unwrap()
.discard(id)
.map_err(|e| EnhanceError::Internal(e.to_string()))?;
self.get_job(id)
}
fn update_job(&self, id: &str, u: &JobUpdate) {
if let Err(e) = self.dao.lock().unwrap().update(id, u) {
warn!("enhance: updating job {id} failed: {e}");
}
}
/// The staged result `n` of a job, for the preview relay.
pub fn pending_of(job: &EnhanceJobRow) -> Vec<Staged> {
serde_json::from_str(&job.pending).unwrap_or_default()
}
pub fn saved_of(job: &EnhanceJobRow) -> Vec<SavedEntry> {
serde_json::from_str(&job.saved).unwrap_or_default()
}
/// Relay the staged PNG `n` of a job. A 404 from ComfyUI means it was
/// restarted: the job becomes `expired`.
pub async fn preview(
&self,
job: &EnhanceJobRow,
n: usize,
) -> Result<(String, bytes::Bytes), EnhanceError> {
let client = self.client.as_ref().ok_or(EnhanceError::Disabled)?;
let staged = Self::pending_of(job);
let item = staged.get(n).ok_or(EnhanceError::NotFound)?;
match client.view(&item.view).await {
Ok(Some(found)) => Ok(found),
Ok(None) => {
self.expire(job);
Err(EnhanceError::Expired)
}
Err(e) => Err(comfy_err(e)),
}
}
/// Mark a job whose results are gone `expired` (unless it was approved:
/// the saved file is in the library regardless).
fn expire(&self, job: &EnhanceJobRow) {
if job.status == STATUS_READY {
let mut u = JobUpdate::status(STATUS_EXPIRED);
u.error = Some(MSG_EXPIRED.to_string());
self.update_job(&job.id, &u);
}
}
// ---- poller ---------------------------------------------------------
/// Advance every active job once. Public so tests (and the loop) drive it.
pub async fn poll_once(&self) {
let Some(client) = &self.client else { return };
let active = match self.dao.lock().unwrap().list_active() {
Ok(a) => a,
Err(e) => {
warn!("enhance: listing active jobs failed: {e}");
return;
}
};
if active.is_empty() {
self.unreachable_since.lock().unwrap().clear();
return;
}
// Queue first, history second: a job that finishes between the two
// calls is then found in history, never misread as forgotten.
let queue = match client.queue().await {
Ok(q) => q,
Err(ComfyError::Unreachable(_)) => {
self.note_unreachable(&active);
return;
}
Err(e) => {
warn!("enhance: /queue failed: {e}");
return;
}
};
for job in active {
self.unreachable_since.lock().unwrap().remove(&job.id);
if let Some((running, ahead)) = queue.position_of(&job.comfy_prompt_id) {
let status = if running { STATUS_RUNNING } else { STATUS_QUEUED };
let mut u = JobUpdate::status(status);
u.queue_position = Some(ahead as i32);
if job.status != status || job.queue_position != u.queue_position {
self.update_job(&job.id, &u);
}
continue;
}
match client.history(&job.comfy_prompt_id).await {
Ok(Some(entry)) if entry.completed || entry.status_str == "error" => {
self.finish(&job, entry)
}
// Known but not finished (a race with the queue): keep waiting.
Ok(Some(_)) => {}
Ok(None) => {
info!("enhance: job {} expired (ComfyUI forgot it)", job.id);
let mut u = JobUpdate::status(STATUS_EXPIRED);
u.error = Some(MSG_EXPIRED.to_string());
self.update_job(&job.id, &u);
}
Err(ComfyError::Unreachable(_)) => self.note_unreachable(std::slice::from_ref(&job)),
Err(e) => warn!("enhance: /history for job {} failed: {e}", job.id),
}
}
}
fn finish(&self, job: &EnhanceJobRow, entry: HistoryEntry) {
if entry.status_str == "error" || (!entry.completed && entry.error.is_some()) {
let mut u = JobUpdate::status(STATUS_FAILED);
u.error = Some(
entry
.error
.unwrap_or_else(|| "ComfyUI reported an error".to_string()),
);
self.update_job(&job.id, &u);
return;
}
if entry.staged.is_empty() {
let mut u = JobUpdate::status(STATUS_FAILED);
u.error = Some("ComfyUI finished without producing a result".to_string());
self.update_job(&job.id, &u);
return;
}
let mut u = JobUpdate::status(STATUS_READY);
u.pending = serde_json::to_string(&entry.staged).ok();
self.update_job(&job.id, &u);
info!("enhance: job {} ready ({} result(s))", job.id, entry.staged.len());
}
fn note_unreachable(&self, jobs: &[EnhanceJobRow]) {
let grace = Duration::from_secs(self.config.unreachable_grace_secs);
let mut seen = self.unreachable_since.lock().unwrap();
for job in jobs {
let since = *seen.entry(job.id.clone()).or_insert_with(Instant::now);
if since.elapsed() >= grace {
seen.remove(&job.id);
let mut u = JobUpdate::status(STATUS_UNAVAILABLE);
u.error = Some(MSG_UNAVAILABLE.to_string());
self.update_job(&job.id, &u);
}
}
}
/// Release ComfyUI's VRAM once it has been idle. The idle test uses
/// ComfyUI's own history, so a session you're working in by hand keeps
/// it from unloading.
pub async fn idle_check(&self) {
let (Some(client), true) = (&self.client, self.config.idle_free_secs > 0) else {
return;
};
let reachable = async {
let queue = client.queue().await?;
let newest = client.newest_history().await?;
Ok::<_, ComfyError>((queue, newest))
}
.await;
let (queue, newest) = match reachable {
Ok(v) => v,
Err(_) => {
self.idle.lock().unwrap().reset();
return;
}
};
let active = self
.dao
.lock()
.unwrap()
.list_active()
.map(|a| a.len())
.unwrap_or(1);
let marker = newest.and_then(|h| h.timestamp_ms);
let should_free = self.idle.lock().unwrap().observe(
chrono::Utc::now().timestamp_millis(),
queue.is_empty(),
active,
marker,
self.config.idle_free_secs as i64 * 1000,
);
if should_free {
match client.free().await {
Ok(()) => info!(
"enhance: ComfyUI idle for {}s, released its models",
self.config.idle_free_secs
),
Err(e) => {
warn!("enhance: POST /free failed: {e}");
self.idle.lock().unwrap().free_failed();
}
}
}
}
/// The background loop: poll active jobs every `poll_secs`, run the idle
/// check every 30 s, prune old jobs once at start.
pub fn spawn(self: &Arc<Self>) {
if !self.is_configured() {
return;
}
let svc = Arc::clone(self);
tokio::spawn(async move {
let cutoff = chrono::Utc::now().timestamp() - JOB_RETENTION_SECS;
if let Err(e) = svc.dao.lock().unwrap().prune(cutoff) {
warn!("enhance: pruning old jobs failed: {e}");
}
let mut last_idle = Instant::now() - IDLE_CHECK_EVERY;
loop {
svc.poll_once().await;
if last_idle.elapsed() >= IDLE_CHECK_EVERY {
last_idle = Instant::now();
svc.idle_check().await;
}
tokio::time::sleep(Duration::from_secs(svc.config.poll_secs)).await;
}
});
}
// ---- approve --------------------------------------------------------
/// Ask ComfyUI to write result `n` into the library. Idempotent: a result
/// that is already saved returns the stored entry without calling ComfyUI.
/// The caller ingests the returned file.
pub async fn approve(
&self,
job_id: &str,
n: usize,
place: &dyn Fn(&Path) -> Option<(i32, String)>,
) -> Result<(EnhanceJobRow, SavedEntry), EnhanceError> {
let client = self.client.as_ref().ok_or(EnhanceError::Disabled)?;
let _guard = self.approve_lock.lock().await;
let job = self.get_job(job_id)?;
let mut saved = Self::saved_of(&job);
if let Some(done) = saved.iter().find(|s| s.n == n) {
return Ok((job.clone(), done.clone()));
}
match job.status.as_str() {
STATUS_READY | STATUS_APPROVED => {}
STATUS_EXPIRED => return Err(EnhanceError::Expired),
other => {
return Err(EnhanceError::BadRequest(format!(
"the job is {other}, there is nothing to approve"
)));
}
}
let staged = Self::pending_of(&job);
let item = staged
.get(n)
.ok_or_else(|| EnhanceError::BadRequest(format!("this job has no result {n}")))?;
let comfy_path = match client.approve(&item.id).await {
Ok(p) => p,
Err(ApproveError::Gone) => {
self.expire(&job);
return Err(EnhanceError::Expired);
}
Err(ApproveError::Refused(m)) => return Err(EnhanceError::Upstream(m)),
Err(ApproveError::Comfy(e)) => return Err(comfy_err(e)),
};
let located = self
.path_map
.from_comfy(&comfy_path)
.and_then(|abs| place(&abs));
let entry = SavedEntry {
n,
comfy_path: comfy_path.clone(),
library_id: located.as_ref().map(|(l, _)| *l),
rel_path: located.map(|(_, r)| r),
};
// Record it even when it can't be placed: the file exists now.
saved.push(entry.clone());
let mut u = JobUpdate::status(STATUS_APPROVED);
u.error = None;
u.saved = serde_json::to_string(&saved).ok();
self.update_job(&job.id, &u);
let job = self.get_job(job_id)?;
if entry.rel_path.is_none() {
return Err(EnhanceError::Unplaceable(format!(
"ComfyUI saved {comfy_path}, but it isn't under any library or path_map entry"
)));
}
Ok((job, entry))
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn idle_frees_once_after_the_quiet_period() {
let mut s = IdleState::default();
let idle = 300_000;
// First sight is activity.
assert!(!s.observe(0, true, 0, Some(5), idle));
assert!(!s.observe(299_000, true, 0, Some(5), idle));
assert!(s.observe(300_000, true, 0, Some(5), idle));
// Only once.
assert!(!s.observe(400_000, true, 0, Some(5), idle));
assert!(!s.observe(9_000_000, true, 0, Some(5), idle));
// A new history entry (e.g. an interactive run) restarts the clock
// and re-arms the free.
assert!(!s.observe(9_100_000, true, 0, Some(6), idle));
assert!(!s.observe(9_399_000, true, 0, Some(6), idle));
assert!(s.observe(9_400_000, true, 0, Some(6), idle));
}
#[test]
fn idle_never_frees_while_busy() {
let mut s = IdleState::default();
assert!(!s.observe(0, true, 0, Some(1), 1000));
// Queue not empty: activity, clock restarts each time.
assert!(!s.observe(5000, false, 0, Some(1), 1000));
assert!(!s.observe(5500, true, 0, Some(1), 1000));
// One of our jobs is active.
assert!(!s.observe(9000, true, 1, Some(1), 1000));
assert!(!s.observe(9500, true, 0, Some(1), 1000));
assert!(s.observe(10_000, true, 0, Some(1), 1000));
}
#[test]
fn idle_restarts_after_comfy_goes_away_and_retries_failures() {
let mut s = IdleState::default();
assert!(!s.observe(0, true, 0, None, 1000));
assert!(s.observe(1000, true, 0, None, 1000));
s.free_failed();
assert!(s.observe(1500, true, 0, None, 1000), "retried");
s.reset();
assert!(!s.observe(10_000, true, 0, None, 1000), "fresh baseline");
assert!(s.observe(11_000, true, 0, None, 1000));
}
}
+490
View File
@@ -0,0 +1,490 @@
//! Workflow templates (#192): load, validate, and fill in an API-format
//! ComfyUI workflow for one photo.
//!
//! Nodes are found by `_meta.title`, never by id, so re-exporting after
//! editing the graph keeps working.
use std::path::Path;
use anyhow::{Context as _, Result, bail};
use serde_json::{Map, Value, json};
use super::config::{PromptMode, TemplateConfig};
pub const TITLE_SOURCE: &str = "imageapi:source";
pub const TITLE_SAVE: &str = "imageapi:save";
pub const TITLE_PROMPT: &str = "imageapi:prompt";
pub const TITLE_REFERENCE: &str = "imageapi:reference";
/// Prompt-node inputs, in the order they are tried.
const PROMPT_INPUTS: &[&str] = &["text", "prompt", "value"];
/// Output nodes that write into ComfyUI's output folder on every API run.
const STRAY_OUTPUT_CLASSES: &[&str] = &[
"SaveImage",
"SaveAnimatedWEBP",
"SaveAnimatedPNG",
"SaveWEBM",
"SaveVideo",
"SaveGLB",
];
#[derive(Debug, Clone)]
pub struct Template {
pub cfg: TemplateConfig,
workflow: Map<String, Value>,
source_id: String,
save_id: String,
/// `(node id, input name)` of the prompt text.
prompt: Option<(String, &'static str)>,
reference_id: Option<String>,
/// Startup warnings (stray output nodes).
pub warnings: Vec<String>,
}
/// Why a request can't be turned into a ComfyUI prompt. The message is for
/// the caller, so it names the problem in the user's terms.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct FillError(pub String);
impl std::fmt::Display for FillError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str(&self.0)
}
}
/// A filled-in prompt plus whether the reference node had to be removed
/// (so a ComfyUI rejection can be reported as "needs a reference photo").
#[derive(Debug)]
pub struct Filled {
pub prompt: Value,
pub dropped_reference: bool,
}
impl Template {
pub fn load(cfg: &TemplateConfig, base_dir: &Path) -> Result<Template> {
let path = if cfg.file.is_absolute() {
cfg.file.clone()
} else {
base_dir.join(&cfg.file)
};
let text = std::fs::read_to_string(&path)
.with_context(|| format!("reading workflow {}", path.display()))?;
let value: Value = serde_json::from_str(&text)
.with_context(|| format!("parsing workflow {}", path.display()))?;
Self::from_value(cfg.clone(), value)
}
pub fn from_value(cfg: TemplateConfig, value: Value) -> Result<Template> {
let Value::Object(workflow) = value else {
bail!("workflow is not a JSON object (is it the API-format export?)");
};
for (id, node) in &workflow {
if node.get("class_type").and_then(Value::as_str).is_none()
|| !node.get("inputs").is_some_and(Value::is_object)
{
bail!(
"node {id} has no class_type/inputs: this looks like a UI-format workflow, \
export with Workflow -> Export (API)"
);
}
}
let find = |title: &str| -> Result<Option<String>> {
let ids: Vec<&String> = workflow
.iter()
.filter(|(_, n)| title_of(n) == Some(title))
.map(|(id, _)| id)
.collect();
match ids.len() {
0 => Ok(None),
1 => Ok(Some(ids[0].clone())),
_ => bail!("more than one node is titled {title:?}"),
}
};
let Some(source_id) = find(TITLE_SOURCE)? else {
bail!("no node titled {TITLE_SOURCE:?} (the Load Image From Path that gets the photo)");
};
let Some(save_id) = find(TITLE_SAVE)? else {
bail!("no node titled {TITLE_SAVE:?} (the Save Image Version node)");
};
require_input(&workflow, &source_id, "path", TITLE_SOURCE)?;
let prompt = match find(TITLE_PROMPT)? {
None => None,
Some(id) => {
let inputs = inputs_of(&workflow, &id);
let Some(name) = PROMPT_INPUTS
.iter()
.copied()
.find(|n| inputs.get(*n).is_some_and(Value::is_string))
else {
bail!(
"node {TITLE_PROMPT:?} has none of the inputs {PROMPT_INPUTS:?} \
as a plain string (a linked input can't take the user's text)"
);
};
Some((id, name))
}
};
if cfg.prompt_mode == PromptMode::Replace && prompt.is_none() {
bail!("prompt_mode = \"replace\" needs a node titled {TITLE_PROMPT:?}");
}
let reference_id = find(TITLE_REFERENCE)?;
if let Some(id) = &reference_id {
require_input(&workflow, id, "path", TITLE_REFERENCE)?;
}
let mut warnings = Vec::new();
for (id, node) in &workflow {
let class = node["class_type"].as_str().unwrap_or_default();
if STRAY_OUTPUT_CLASSES.contains(&class) {
warnings.push(format!(
"node {id} ({class}) also writes into ComfyUI's output folder on every run"
));
}
}
Ok(Template {
cfg,
workflow,
source_id,
save_id,
prompt,
reference_id,
warnings,
})
}
pub fn accepts_prompt(&self) -> bool {
self.prompt.is_some()
}
pub fn accepts_reference(&self) -> bool {
self.reference_id.is_some()
}
pub fn prompt_required(&self) -> bool {
self.cfg.prompt_mode == PromptMode::Replace
}
/// Fill the workflow for one photo.
///
/// - `source` / `reference` are paths as ComfyUI sees them.
/// - Without a reference the reference node and every link to it are
/// removed: its leftover export path must never be used.
/// - Every `seed` / `noise_seed` / `*.seed` input gets a fresh value
/// (ComfyUI's "randomize after generate" only exists in its page), or a
/// retry would return the cached result.
pub fn fill(
&self,
source: &str,
prompt: Option<&str>,
reference: Option<&str>,
next_seed: &mut dyn FnMut() -> u64,
) -> Result<Filled, FillError> {
let user_text = prompt.map(str::trim).filter(|t| !t.is_empty());
match (&self.prompt, user_text) {
(None, Some(_)) => {
return Err(FillError(
"this template doesn't take a prompt".to_string(),
));
}
(Some(_), None) if self.prompt_required() => {
return Err(FillError(
"this template needs an instruction (the prompt can't be empty)".to_string(),
));
}
_ => {}
}
if reference.is_some() && self.reference_id.is_none() {
return Err(FillError(
"this template doesn't take a reference photo".to_string(),
));
}
let mut wf = self.workflow.clone();
set_input(&mut wf, &self.source_id, "path", json!(source));
set_input(&mut wf, &self.save_id, "require_approval", json!(true));
if let (Some((id, input)), Some(text)) = (&self.prompt, user_text) {
let base = inputs_of(&wf, id)
.get(*input)
.and_then(Value::as_str)
.unwrap_or_default()
.to_string();
let combined = match self.cfg.prompt_mode {
PromptMode::Replace => text.to_string(),
PromptMode::Append if base.trim().is_empty() => text.to_string(),
PromptMode::Append => format!("{base}{}{text}", self.cfg.prompt_join),
};
set_input(&mut wf, id, input, json!(combined));
}
let mut dropped_reference = false;
if let Some(ref_id) = &self.reference_id {
match reference {
Some(path) => set_input(&mut wf, ref_id, "path", json!(path)),
None => {
wf.remove(ref_id);
for node in wf.values_mut() {
if let Some(inputs) = node.get_mut("inputs").and_then(Value::as_object_mut)
{
inputs.retain(|_, v| !links_to(v, ref_id));
}
}
dropped_reference = true;
}
}
}
for node in wf.values_mut() {
let Some(inputs) = node.get_mut("inputs").and_then(Value::as_object_mut) else {
continue;
};
for (key, value) in inputs.iter_mut() {
if is_seed_input(key) && value.is_number() {
*value = json!(next_seed());
}
}
}
Ok(Filled {
prompt: Value::Object(wf),
dropped_reference,
})
}
}
fn title_of(node: &Value) -> Option<&str> {
node.get("_meta")?.get("title")?.as_str()
}
fn inputs_of<'a>(wf: &'a Map<String, Value>, id: &str) -> &'a Map<String, Value> {
static EMPTY: std::sync::LazyLock<Map<String, Value>> = std::sync::LazyLock::new(Map::new);
wf.get(id)
.and_then(|n| n.get("inputs"))
.and_then(Value::as_object)
.unwrap_or(&EMPTY)
}
fn require_input(wf: &Map<String, Value>, id: &str, input: &str, title: &str) -> Result<()> {
if !inputs_of(wf, id).get(input).is_some_and(Value::is_string) {
bail!("node {title:?} has no string input {input:?}");
}
Ok(())
}
fn set_input(wf: &mut Map<String, Value>, id: &str, input: &str, value: Value) {
if let Some(inputs) = wf
.get_mut(id)
.and_then(|n| n.get_mut("inputs"))
.and_then(Value::as_object_mut)
{
inputs.insert(input.to_string(), value);
}
}
/// `["<id>", n]`: a link to output `n` of node `id`.
fn links_to(v: &Value, id: &str) -> bool {
matches!(v.as_array().map(Vec::as_slice), Some([Value::String(s), Value::Number(_)]) if s == id)
}
fn is_seed_input(key: &str) -> bool {
key == "seed" || key == "noise_seed" || key.ends_with(".seed")
}
#[cfg(test)]
mod tests {
use super::*;
pub const RESTORE: &str = include_str!("../../tests/fixtures/enhance/restore_turbo.json");
pub const EDIT: &str = include_str!("../../tests/fixtures/enhance/edit_turbo.json");
pub const DEV: &str = include_str!("../../tests/fixtures/enhance/restore_dev.json");
fn cfg(name: &str, mode: PromptMode) -> TemplateConfig {
TemplateConfig {
name: name.into(),
label: name.into(),
description: String::new(),
file: "x.json".into(),
prompt_mode: mode,
prompt_join: " ".into(),
output_megapixels: 1.0,
}
}
fn load(json: &str, mode: PromptMode) -> Result<Template> {
Template::from_value(cfg("t", mode), serde_json::from_str(json).unwrap())
}
fn seeds() -> impl FnMut() -> u64 {
let mut n = 1000;
move || {
n += 1;
n
}
}
fn input<'a>(p: &'a Value, id: &str, name: &str) -> &'a Value {
&p[id]["inputs"][name]
}
#[test]
fn real_templates_load() {
let restore = load(RESTORE, PromptMode::Append).unwrap();
assert!(restore.accepts_prompt());
assert!(!restore.accepts_reference());
assert!(!restore.prompt_required());
let edit = load(EDIT, PromptMode::Replace).unwrap();
assert!(edit.accepts_prompt() && edit.accepts_reference() && edit.prompt_required());
assert!(load(DEV, PromptMode::Append).is_ok());
// The prompt node is a String Multiline primitive: input `value`.
assert_eq!(restore.prompt.as_ref().unwrap().1, "value");
assert!(restore.warnings.is_empty());
}
#[test]
fn missing_required_titles_are_rejected_with_a_clear_message() {
let no_source = RESTORE.replace("imageapi:source", "something else");
let err = load(&no_source, PromptMode::Append).unwrap_err().to_string();
assert!(err.contains("imageapi:source"), "{err}");
let no_save = RESTORE.replace("imageapi:save", "something else");
let err = load(&no_save, PromptMode::Append).unwrap_err().to_string();
assert!(err.contains("imageapi:save"), "{err}");
}
#[test]
fn rejects_ui_format_duplicates_and_bad_prompt_nodes() {
let ui = r#"{"nodes": [], "links": []}"#;
assert!(load(ui, PromptMode::Append).is_err());
let dup = RESTORE.replace("\"Load CLIP\"", "\"imageapi:source\"");
assert!(load(&dup, PromptMode::Append).is_err());
// Prompt node with none of text / prompt / value.
let mut v: Value = serde_json::from_str(RESTORE).unwrap();
v["13"]["inputs"] = json!({"other": "x"});
assert!(Template::from_value(cfg("t", PromptMode::Append), v).is_err());
// replace without a prompt node
let no_prompt = RESTORE.replace("imageapi:prompt", "nope");
assert!(load(&no_prompt, PromptMode::Replace).is_err());
assert!(load(&no_prompt, PromptMode::Append).is_ok());
}
#[test]
fn prompt_node_accepts_text_or_prompt_inputs() {
let mut v: Value = serde_json::from_str(RESTORE).unwrap();
v["13"]["inputs"] = json!({"prompt": "base"});
let t = Template::from_value(cfg("t", PromptMode::Append), v).unwrap();
let f = t.fill("Z:\\a.jpg", Some("more"), None, &mut seeds()).unwrap();
assert_eq!(input(&f.prompt, "13", "prompt"), "base more");
}
#[test]
fn stray_save_image_is_a_warning_not_an_error() {
let mut v: Value = serde_json::from_str(RESTORE).unwrap();
v["99"] = json!({"class_type": "SaveImage", "inputs": {"images": ["6", 0]}});
let t = Template::from_value(cfg("t", PromptMode::Append), v).unwrap();
assert_eq!(t.warnings.len(), 1);
assert!(t.warnings[0].contains("SaveImage"));
}
#[test]
fn fill_sets_path_forces_approval_and_randomises_every_seed() {
let t = load(RESTORE, PromptMode::Append).unwrap();
let f = t.fill("Z:\\p\\a.jpg", None, None, &mut seeds()).unwrap();
assert_eq!(input(&f.prompt, "37", "path"), "Z:\\p\\a.jpg");
assert_eq!(input(&f.prompt, "36", "require_approval"), true);
// KSampler.seed and TextGenerate's sampling_mode.seed, both fresh.
let a = input(&f.prompt, "5", "seed").as_u64().unwrap();
let b = input(&f.prompt, "14", "sampling_mode.seed").as_u64().unwrap();
assert!(a > 1000 && b > 1000 && a != b);
assert_ne!(a, 774744764437710);
// Two fills with a real RNG differ.
let mut rng = || rand::random::<u64>() >> 12;
let x = t.fill("p", None, None, &mut rng).unwrap();
let y = t.fill("p", None, None, &mut rng).unwrap();
assert_ne!(input(&x.prompt, "5", "seed"), input(&y.prompt, "5", "seed"));
}
#[test]
fn force_approval_even_if_the_export_had_it_off() {
let off = RESTORE.replace("\"require_approval\": true", "\"require_approval\": false");
let t = load(&off, PromptMode::Append).unwrap();
let f = t.fill("p", None, None, &mut seeds()).unwrap();
assert_eq!(input(&f.prompt, "36", "require_approval"), true);
}
#[test]
fn append_uses_the_join_text_and_empty_leaves_the_base() {
let base = "Increase the clarity";
let mut c = cfg("t", PromptMode::Append);
c.prompt_join = "\n\nAdditional instructions: ".into();
let mut v: Value = serde_json::from_str(RESTORE).unwrap();
v["13"]["inputs"]["value"] = json!(base);
let t = Template::from_value(c, v).unwrap();
let f = t.fill("p", Some(" warmer "), None, &mut seeds()).unwrap();
assert_eq!(
input(&f.prompt, "13", "value"),
"Increase the clarity\n\nAdditional instructions: warmer"
);
let f = t.fill("p", Some(" "), None, &mut seeds()).unwrap();
assert_eq!(input(&f.prompt, "13", "value"), base);
let f = t.fill("p", None, None, &mut seeds()).unwrap();
assert_eq!(input(&f.prompt, "13", "value"), base);
}
#[test]
fn default_join_is_a_single_space_sentence() {
let t = load(RESTORE, PromptMode::Append).unwrap();
let f = t.fill("p", Some("Make it warmer."), None, &mut seeds()).unwrap();
let text = input(&f.prompt, "13", "value").as_str().unwrap();
assert!(text.ends_with("Remove any phone interface overlays. Make it warmer."));
}
#[test]
fn replace_mode_requires_and_uses_the_instruction() {
let t = load(EDIT, PromptMode::Replace).unwrap();
let err = t.fill("p", None, None, &mut seeds()).unwrap_err();
assert!(err.0.contains("instruction"));
assert!(t.fill("p", Some(" "), None, &mut seeds()).is_err());
let f = t.fill("p", Some("make it black and white"), None, &mut seeds()).unwrap();
assert_eq!(input(&f.prompt, "13", "value"), "make it black and white");
}
#[test]
fn prompt_or_reference_the_template_cant_take_are_refused() {
let mut v: Value = serde_json::from_str(RESTORE).unwrap();
v.as_object_mut().unwrap().remove("13");
// 14 links to 13; drop that link so the graph stays plausible.
v["14"]["inputs"].as_object_mut().unwrap().remove("prompt");
let t = Template::from_value(cfg("t", PromptMode::Append), v).unwrap();
assert!(t.fill("p", Some("x"), None, &mut seeds()).is_err());
assert!(t.fill("p", None, None, &mut seeds()).is_ok());
let restore = load(RESTORE, PromptMode::Append).unwrap();
assert!(restore.fill("p", None, Some("r"), &mut seeds()).is_err());
}
#[test]
fn reference_given_sets_its_path_and_keeps_links() {
let t = load(EDIT, PromptMode::Replace).unwrap();
let f = t
.fill("src", Some("x"), Some("Z:\\ref.jpg"), &mut seeds())
.unwrap();
assert!(!f.dropped_reference);
assert_eq!(input(&f.prompt, "38", "path"), "Z:\\ref.jpg");
assert_eq!(input(&f.prompt, "36", "reference_1"), &json!(["38", 1]));
}
#[test]
fn no_reference_removes_the_node_and_every_link_to_it() {
let t = load(EDIT, PromptMode::Replace).unwrap();
let f = t.fill("src", Some("x"), None, &mut seeds()).unwrap();
assert!(f.dropped_reference);
let wf = f.prompt.as_object().unwrap();
assert!(!wf.contains_key("38"));
// No input anywhere still links to node 38, and the leftover export
// path can't leak.
assert!(!serde_json::to_string(&f.prompt).unwrap().contains("[\"38\""));
assert!(wf["36"]["inputs"].get("reference_1").is_none());
// Links to other nodes survive.
assert_eq!(input(&f.prompt, "36", "source"), &json!(["37", 1]));
}
}