diff --git a/BACKLOG.md b/BACKLOG.md index d895016..b6e6b20 100644 --- a/BACKLOG.md +++ b/BACKLOG.md @@ -1296,7 +1296,13 @@ Wan 2.2 small (video out, a long job), ACE-Step (audio out), SF3D (a package not on PyPI, gated weights), Hunyuan3D (a licence that excludes the EU, which H6 must be able to say no to). -### H1. A workflow says what it takes and makes (M) +### H1. A workflow says what it takes and makes (M) — built 2026-10-11, not yet tried on the PoC +Built as written, with three choices: a run answers with one file, the +first of its output kind (several outputs as one zip is left); the +workflow's placeholders are checked against the inputs when a run is +served, since the catalog never reads the workflow file; and video, +audio and files count as nothing in usage until a per-run price exists +(per_image and per_mesh still count pictures and meshes). - Catalog: `comfyui.inputs` — a list of `{name, type, required, default, min, max, choices, max_mb, doc}`, type one of text, int, float, bool, choice, image, mask, audio, video, file. Each is a placeholder of the diff --git a/CHANGELOG.md b/CHANGELOG.md index 068f83e..9a9ef78 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -36,6 +36,20 @@ APIs change between minors without a deprecation period. `{"state":"waiting_gpu","held_by":…,"eta_seconds":…}` while another on-demand model holds it; a direct request's 503 carries `X-Llmgw-Waiting-For` and `X-Llmgw-Eta-Seconds`. +- **Any ComfyUI workflow, as a job**: `POST /v1/workflows/runs` with + `{model, inputs}`. A model's catalog entry declares its inputs + (`comfyui.inputs`: text, int, float, bool, choice, image, mask, audio, + video, file, with bounds and defaults) and its output (`comfyui.output`: + pictures, meshes, video, audio or files); a request is checked against + them before a job is written, file inputs are base64, a data URL or a + `/v1/files` id, and `/v1/models` lists each model's `workflow` so a + client can build its form. A PNG result carries the licence in an + `iTXt` chunk, a GLB in `asset.extras` as before. +- **Environments**: `comfyui.environment` names ComfyUI models that agree + on their custom nodes and may share one replica; the catalog refuses + one that cannot. +- **A licence a machine can match**: `license.id`, `commercial`, + `excluded_regions`, `revenue_cap_usd`, `attribution`. - **On-demand models in `/v1/models`**, with `loaded` saying whether one is up now; a picture is cut out and put on grey before TripoSR sees it (`LLMGWPrepareForMesh`, rembg when it has no alpha). diff --git a/docs/admin-api.openapi.json b/docs/admin-api.openapi.json index 75ef4e9..c604298 100644 --- a/docs/admin-api.openapi.json +++ b/docs/admin-api.openapi.json @@ -8774,7 +8774,14 @@ }, "output": { "type": "string", - "description": "Output is what the workflow makes: pictures (the default), asked for at /v1/images/generations, or meshes, asked for as a job at /v1/3d/generations from a picture." + "description": "Output is what the workflow makes: pictures (the default), asked for at /v1/images/generations; meshes, asked for as a job at /v1/3d/generations from a picture; or video, audio or files. Every kind is asked for as a job at /v1/workflows/runs too." + }, + "inputs": { + "type": "array", + "items": { + "$ref": "#/components/schemas/catalog.Input" + }, + "description": "Inputs are what the workflow takes from a request at /v1/workflows/runs, each the placeholder of its name." }, "nodes": { "type": "array", @@ -8879,6 +8886,47 @@ } } }, + "catalog.Input": { + "type": "object", + "description": "Input is one thing a workflow takes from a request: the placeholder {{name}} in its graph, filled with the request's inputs.name.", + "properties": { + "name": { + "type": "string", + "description": "Name is the placeholder's, and the request's key." + }, + "type": { + "type": "string", + "description": "Type is one of text, int, float, bool, choice, image, mask, audio, video, file. The last five are files, sent as base64, a data URL, or the id of a file uploaded to /v1/files, and put in ComfyUI's input directory; the placeholder gets the name they were given there." + }, + "required": { + "type": "boolean", + "description": "Required inputs must be in every request; the others take Default, or the workflow's own {{name|default}}." + }, + "default": {}, + "min": { + "type": "number", + "description": "Min and Max bound a number; for text, Max bounds its length." + }, + "max": { + "type": "number" + }, + "choices": { + "type": "array", + "items": { + "type": "string" + }, + "description": "Choices are what a choice may be." + }, + "max_mb": { + "type": "integer", + "description": "MaxMB bounds a file input; zero is the type's own bound." + }, + "doc": { + "type": "string", + "description": "Doc says what the input is for, to whoever fills it in." + } + } + }, "catalog.License": { "type": "object", "description": "License is the terms what a model makes comes under: said beside every result, and written into a mesh, which travels without the API.", diff --git a/docs/client-api.openapi.json b/docs/client-api.openapi.json index 441e81a..9548e3c 100644 --- a/docs/client-api.openapi.json +++ b/docs/client-api.openapi.json @@ -73,6 +73,9 @@ }, { "name": "threads" + }, + { + "name": "workflows" } ], "paths": { @@ -1477,6 +1480,163 @@ } } } + }, + "/v1/workflows/runs": { + "post": { + "operationId": "postWorkflowsRuns", + "summary": "Run a ComfyUI model's workflow, as a job", + "description": "Inputs are those the model's workflow lists in /v1/models. Answered 202 with the job; download output_file_id from /v1/files once it is completed.", + "tags": [ + "workflows" + ], + "requestBody": { + "required": true, + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/RunRequest" + } + } + } + }, + "responses": { + "202": { + "description": "Accepted: started, and running on.", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/Job" + } + } + } + }, + "401": { + "$ref": "#/components/responses/Unauthorized" + }, + "default": { + "$ref": "#/components/responses/Error" + } + } + }, + "get": { + "operationId": "getWorkflowsRuns", + "summary": "The caller's jobs, newest first", + "tags": [ + "workflows" + ], + "parameters": [ + { + "name": "limit", + "in": "query", + "description": "The most jobs to answer with, at most 100; 20 when absent.", + "schema": { + "type": "integer" + } + }, + { + "name": "after", + "in": "query", + "description": "A job's ID: those asked for before it.", + "schema": { + "type": "string" + } + } + ], + "responses": { + "200": { + "description": "OK", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/JobList" + } + } + } + }, + "401": { + "$ref": "#/components/responses/Unauthorized" + }, + "default": { + "$ref": "#/components/responses/Error" + } + } + } + }, + "/v1/workflows/runs/{id}": { + "get": { + "operationId": "getWorkflowsRunsId", + "summary": "A run: its status, progress and result", + "tags": [ + "workflows" + ], + "parameters": [ + { + "name": "id", + "in": "path", + "required": true, + "description": "The job's ID, job_….", + "schema": { + "type": "string" + } + } + ], + "responses": { + "200": { + "description": "OK", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/Job" + } + } + } + }, + "401": { + "$ref": "#/components/responses/Unauthorized" + }, + "default": { + "$ref": "#/components/responses/Error" + } + } + } + }, + "/v1/workflows/runs/{id}/cancel": { + "post": { + "operationId": "postWorkflowsRunsIdCancel", + "summary": "Stop a run, in the engine's queue too", + "tags": [ + "workflows" + ], + "parameters": [ + { + "name": "id", + "in": "path", + "required": true, + "description": "The job's ID, job_….", + "schema": { + "type": "string" + } + } + ], + "responses": { + "200": { + "description": "OK", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/Job" + } + } + } + }, + "401": { + "$ref": "#/components/responses/Unauthorized" + }, + "default": { + "$ref": "#/components/responses/Error" + } + } + } } }, "components": { @@ -2192,6 +2352,10 @@ "type": "object", "description": "License is the terms what a model makes comes under.", "properties": { + "id": { + "type": "string", + "description": "ID is SPDX where there is one, else the licence's own short name." + }, "name": { "type": "string" }, @@ -2200,6 +2364,24 @@ }, "notice": { "type": "string" + }, + "commercial": { + "type": "boolean", + "description": "Commercial is whether the output may be used commercially; absent is not said." + }, + "excluded_regions": { + "type": "array", + "items": { + "type": "string" + }, + "description": "ExcludedRegions are ISO 3166 codes, EU for the union, where it grants nothing." + }, + "revenue_cap_usd": { + "type": "integer", + "format": "int64" + }, + "attribution": { + "type": "string" } } }, @@ -2389,10 +2571,14 @@ "additionalProperties": { "type": "boolean" }, - "description": "Capabilities: tools, vision, embeddings, images, meshes." + "description": "Capabilities: tools, vision, embeddings, images, meshes, workflows." }, "license": { "$ref": "#/components/schemas/License" + }, + "workflow": { + "$ref": "#/components/schemas/WorkflowInfo", + "description": "Workflow is what a ComfyUI model takes and makes." } } }, @@ -2525,6 +2711,25 @@ } } }, + "RunRequest": { + "type": "object", + "description": "RunRequest asks a ComfyUI model to run its workflow.", + "properties": { + "model": { + "type": "string" + }, + "inputs": { + "type": "object", + "additionalProperties": {}, + "description": "Inputs are by name, as the model's workflow lists them in /v1/models." + }, + "seed": { + "type": "integer", + "format": "int64", + "description": "Seed is the workflow's {{seed}}, where it has one; random otherwise." + } + } + }, "ScoreRequest": { "type": "object", "description": "ScoreRequest scores pairs of texts, for a scorer vLLM serves.", @@ -2608,6 +2813,57 @@ } } }, + "WorkflowInfo": { + "type": "object", + "description": "WorkflowInfo is what a ComfyUI model takes and makes, for a run at /v1/workflows/runs.", + "properties": { + "output": { + "type": "string", + "description": "Output is pictures, meshes, video, audio or files." + }, + "inputs": { + "type": "array", + "items": { + "$ref": "#/components/schemas/WorkflowInput" + } + } + } + }, + "WorkflowInput": { + "type": "object", + "description": "WorkflowInput is one input a workflow takes.", + "properties": { + "name": { + "type": "string" + }, + "type": { + "type": "string", + "description": "Type is text, int, float, bool, choice, image, mask, audio, video or file; the last five are sent as base64, a data URL, or a file's id." + }, + "required": { + "type": "boolean" + }, + "default": {}, + "min": { + "type": "number" + }, + "max": { + "type": "number" + }, + "choices": { + "type": "array", + "items": { + "type": "string" + } + }, + "max_mb": { + "type": "integer" + }, + "doc": { + "type": "string" + } + } + }, "router.KBCollectionView": { "type": "object", "description": "KBCollectionView is a collection as a caller sees it.", diff --git a/docs/reference.md b/docs/reference.md index a3003a7..14b16dd 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -65,6 +65,7 @@ rather than repeated. | `workflow` | `string` | [ComfyUI](#comfyui) | | `checkpoint` | `string` | [ComfyUI](#comfyui) | | `output` | `string` | [ComfyUI](#comfyui) | +| `inputs` | `[]Input` | [ComfyUI](#comfyui) | | `nodes` | `[]CustomNode` | [ComfyUI](#comfyui) | | `environment` | `string` | [ComfyUI](#comfyui) | | `dir` | `string` | [ComfyUIConfig](#comfyuiconfig) | @@ -120,6 +121,15 @@ rather than repeated. | `enabled` | `bool` | [FirewallSettings](#firewallsettings) | | `format` | `string` | [GuardModel](#guardmodel) | | `categories` | `[]string` | [GuardModel](#guardmodel) | +| `name` | `string` | [Input](#input) | +| `type` | `string` | [Input](#input) | +| `required` | `bool` | [Input](#input) | +| `default` | `any` | [Input](#input) | +| `min` | `*float64` | [Input](#input) | +| `max` | `*float64` | [Input](#input) | +| `choices` | `[]string` | [Input](#input) | +| `max_mb` | `int` | [Input](#input) | +| `doc` | `string` | [Input](#input) | | `model` | `string` | [JudgePolicy](#judgepolicy) | | `models` | `[]string` | [JudgePolicy](#judgepolicy) | | `input` | `string` | [JudgePolicy](#judgepolicy) | @@ -789,8 +799,16 @@ must. `string` Output is what the workflow makes: pictures (the default), asked for -at /v1/images/generations, or meshes, asked for as a job at -/v1/3d/generations from a picture. +at /v1/images/generations; meshes, asked for as a job at +/v1/3d/generations from a picture; or video, audio or files. Every +kind is asked for as a job at /v1/workflows/runs too. + +### `inputs` + +`[]Input` + +Inputs are what the workflow takes from a request at +/v1/workflows/runs, each the placeholder of its name. ### `nodes` @@ -1201,6 +1219,62 @@ ShieldGemma: Yes or No), or json ({"safe": false, "categories": Categories the model is asked about, for shieldgemma — each a policy sentence — and the names reported for llamaguard3's S1, S2, …. +## Input + +### `name` + +`string` + +Name is the placeholder's, and the request's key. + +### `type` + +`string` + +Type is one of text, int, float, bool, choice, image, mask, audio, +video, file. The last five are files, sent as base64, a data URL, or +the id of a file uploaded to /v1/files, and put in ComfyUI's input +directory; the placeholder gets the name they were given there. + +### `required` + +`bool` + +Required inputs must be in every request; the others take Default, +or the workflow's own {{name|default}}. + +### `default` + +`any` + +### `min` + +`*float64` + +Min and Max bound a number; for text, Max bounds its length. + +### `max` + +`*float64` + +### `choices` + +`[]string` + +Choices are what a choice may be. + +### `max_mb` + +`int` + +MaxMB bounds a file input; zero is the type's own bound. + +### `doc` + +`string` + +Doc says what the input is for, to whoever fills it in. + ## JudgePolicy ### `model` diff --git a/internal/clientapi/clientapi.go b/internal/clientapi/clientapi.go index 4002552..974d611 100644 --- a/internal/clientapi/clientapi.go +++ b/internal/clientapi/clientapi.go @@ -258,9 +258,51 @@ type TokenCount struct { // License is the terms what a model makes comes under. type License struct { + // ID is SPDX where there is one, else the licence's own short name. + ID string `json:"id,omitempty"` Name string `json:"name"` URL string `json:"url,omitempty"` Notice string `json:"notice,omitempty"` + // Commercial is whether the output may be used commercially; absent is + // not said. + Commercial *bool `json:"commercial,omitempty"` + // ExcludedRegions are ISO 3166 codes, EU for the union, where it grants + // nothing. + ExcludedRegions []string `json:"excluded_regions,omitempty"` + RevenueCapUSD int64 `json:"revenue_cap_usd,omitempty"` + Attribution string `json:"attribution,omitempty"` +} + +// WorkflowInfo is what a ComfyUI model takes and makes, for a run at +// /v1/workflows/runs. +type WorkflowInfo struct { + // Output is pictures, meshes, video, audio or files. + Output string `json:"output"` + Inputs []WorkflowInput `json:"inputs"` +} + +// WorkflowInput is one input a workflow takes. +type WorkflowInput struct { + Name string `json:"name"` + // Type is text, int, float, bool, choice, image, mask, audio, video or + // file; the last five are sent as base64, a data URL, or a file's id. + Type string `json:"type"` + Required bool `json:"required,omitempty"` + Default any `json:"default,omitempty"` + Min *float64 `json:"min,omitempty"` + Max *float64 `json:"max,omitempty"` + Choices []string `json:"choices,omitempty"` + MaxMB int `json:"max_mb,omitempty"` + Doc string `json:"doc,omitempty"` +} + +// RunRequest asks a ComfyUI model to run its workflow. +type RunRequest struct { + Model string `json:"model"` + // Inputs are by name, as the model's workflow lists them in /v1/models. + Inputs map[string]any `json:"inputs"` + // Seed is the workflow's {{seed}}, where it has one; random otherwise. + Seed *int64 `json:"seed,omitempty"` } // ModelInfo is a model a key may use. @@ -276,9 +318,11 @@ type ModelInfo struct { ContextLength int `json:"context_length,omitempty"` MaxOutputTokens int `json:"max_output_tokens,omitempty"` Description string `json:"description,omitempty"` - // Capabilities: tools, vision, embeddings, images, meshes. + // Capabilities: tools, vision, embeddings, images, meshes, workflows. Capabilities map[string]bool `json:"capabilities"` License *License `json:"license,omitempty"` + // Workflow is what a ComfyUI model takes and makes. + Workflow *WorkflowInfo `json:"workflow,omitempty"` } // ModelList is the models a key may use. diff --git a/internal/clientapi/routes.go b/internal/clientapi/routes.go index 6431778..b1f24ef 100644 --- a/internal/clientapi/routes.go +++ b/internal/clientapi/routes.go @@ -18,6 +18,7 @@ var PathParams = map[string]string{ "files/{id}": "The file's ID, file-….", "batches/{id}": "The batch's ID, batch_….", "generations/{id}": "The job's ID, job_….", + "runs/{id}": "The job's ID, job_….", "threads/{id}": "The conversation's ID, as sent in x-llmgw-thread.", "memory/{key}": "The note's key, chosen by the caller.", "collections/{id}": "The collection's ID, its name when that is unique, or ORG/NAME.", @@ -59,6 +60,18 @@ var Routes = []Route{ Response: JobList{}}, {Method: "GET", Path: "/v1/3d/generations/{id}", Summary: "A 3D job: its status, progress and result", Response: Job{}}, {Method: "POST", Path: "/v1/3d/generations/{id}/cancel", Summary: "Stop a 3D job, in the engine's queue too", Response: Job{}}, + {Method: "POST", Path: "/v1/workflows/runs", Summary: "Run a ComfyUI model's workflow, as a job", + Description: "Inputs are those the model's workflow lists in /v1/models. Answered 202 with the job; " + + "download output_file_id from /v1/files once it is completed.", + Request: RunRequest{}, Response: Job{}, Status: 202}, + {Method: "GET", Path: "/v1/workflows/runs", Summary: "The caller's jobs, newest first", + Query: []Param{ + {Name: "limit", Type: "integer", Description: "The most jobs to answer with, at most 100; 20 when absent."}, + {Name: "after", Description: "A job's ID: those asked for before it."}, + }, + Response: JobList{}}, + {Method: "GET", Path: "/v1/workflows/runs/{id}", Summary: "A run: its status, progress and result", Response: Job{}}, + {Method: "POST", Path: "/v1/workflows/runs/{id}/cancel", Summary: "Stop a run, in the engine's queue too", Response: Job{}}, {Method: "POST", Path: "/v1/moderations", Summary: "Texts judged by the guardrails' judge model", Request: ModerationRequest{}, Response: ModerationResponse{}}, {Method: "POST", Path: "/v1/responses", Summary: "OpenAI's Responses API, for Codex and the like", diff --git a/internal/clientapi/routes_test.go b/internal/clientapi/routes_test.go index fd1ab5d..36144c2 100644 --- a/internal/clientapi/routes_test.go +++ b/internal/clientapi/routes_test.go @@ -17,7 +17,8 @@ func routerRoutes(t *testing.T) []string { t.Helper() _, file, _, _ := runtime.Caller(0) dir := filepath.Join(filepath.Dir(file), "..", "router") - consts := map[string]string{"imagesPath": "/v1/images/generations", "meshesPath": "/v1/3d/generations"} + consts := map[string]string{"imagesPath": "/v1/images/generations", "meshesPath": "/v1/3d/generations", + "workflowsPath": "/v1/workflows/runs"} literal := regexp.MustCompile(`mux\.HandleFunc\("([A-Z]+) (/[^"]*)"`) built := regexp.MustCompile(`mux\.HandleFunc\("([A-Z]+) "\+([A-Za-z]+Path)(\+"([^"]*)")?`) files, _ := filepath.Glob(filepath.Join(dir, "*.go")) diff --git a/internal/comfy/comfy.go b/internal/comfy/comfy.go index 98e4757..88dad71 100644 --- a/internal/comfy/comfy.go +++ b/internal/comfy/comfy.go @@ -108,6 +108,11 @@ type Params struct { } func (p Params) value(name string) (any, bool) { + // A workflow's own input of a built-in's name is the request's: an input + // named prompt is the prompt a run was given. + if v, ok := p.Extra[name]; ok { + return v, true + } switch name { case "checkpoint": return p.Checkpoint, p.Checkpoint != "" @@ -130,8 +135,44 @@ func (p Params) value(name string) (any, bool) { case "image": return p.Image, p.Image != "" } - v, ok := p.Extra[name] - return v, ok + return nil, false +} + +// Builtin are the placeholders Params fills by itself, which a workflow may +// use without declaring them as inputs. +var Builtin = []string{"checkpoint", "prompt", "negative_prompt", "width", "height", "n", "seed", "steps", "guidance", "image"} + +// Placeholders are the names a template's placeholders use, each once, +// sorted, and whether each has a default of its own. +func (t *Template) Placeholders() map[string]bool { + out := map[string]bool{} + var walk func(v any) + walk = func(v any) { + switch v := v.(type) { + case map[string]any: + for _, x := range v { + walk(x) + } + case []any: + for _, x := range v { + walk(x) + } + case string: + for s := v; ; { + i := strings.Index(s, "{{") + j := strings.Index(s, "}}") + if i < 0 || j < i { + return + } + name, _, hasDef := strings.Cut(s[i+2:j], "|") + name = strings.TrimSpace(name) + out[name] = out[name] || hasDef + s = s[j+2:] + } + } + } + walk(t.graph) + return out } // Fill is the workflow for one request. @@ -398,9 +439,17 @@ func (c *Client) fetch(ctx context.Context, base, apiKey string, h historyEntry) return nil, err } img.Name = f.Filename - if t := typeByName(f.Filename); t != "" && (img.ContentType == "" || - strings.HasPrefix(img.ContentType, "application/octet-stream") || strings.HasPrefix(img.ContentType, "text/plain")) { - img.ContentType = t + // A type the server did not say is the file's by its name, + // else by its first bytes. + if generic(img.ContentType) { + if t := typeByName(f.Filename); t != "" { + img.ContentType = t + } else { + img.ContentType = http.DetectContentType(img.Data) + } + } + if img.ContentType == "image/png" { + img.Data = withoutPNGText(img.Data) } out = append(out, img) } @@ -497,14 +546,12 @@ func (c *Client) get(ctx context.Context, u, apiKey string) (Image, error) { if resp.StatusCode != http.StatusOK { return Image{}, &Error{Msg: fmt.Sprintf("ComfyUI answered %d for a file it made", resp.StatusCode)} } - ct := resp.Header.Get("Content-Type") - if ct == "" || ct == "application/octet-stream" { - ct = http.DetectContentType(b) - } - if ct == "image/png" { - b = withoutPNGText(b) - } - return Image{Data: b, ContentType: ct}, nil + return Image{Data: b, ContentType: resp.Header.Get("Content-Type")}, nil +} + +// generic is a type that says nothing of what a file is. +func generic(ct string) bool { + return ct == "" || strings.HasPrefix(ct, "application/octet-stream") || strings.HasPrefix(ct, "text/plain") } // withoutPNGText is a PNG without its text chunks. ComfyUI writes the whole @@ -604,12 +651,33 @@ func typeByName(name string) string { return "application/ply" case ".png": return "image/png" + case ".jpg", ".jpeg": + return "image/jpeg" + case ".webp": + return "image/webp" + case ".gif": + return "image/gif" + case ".mp4": + return "video/mp4" + case ".webm": + return "video/webm" + case ".mov": + return "video/quicktime" + case ".wav": + return "audio/wav" + case ".mp3": + return "audio/mpeg" + case ".flac": + return "audio/flac" + case ".ogg", ".opus": + return "audio/ogg" } return "" } -// Upload puts a picture in ComfyUI's input directory, where a LoadImage node -// reads it, and returns the name it was given there. +// Upload puts a file in ComfyUI's input directory, where a loader node reads +// it — LoadImage, LoadAudio, LoadVideo, whatever the workflow uses: ComfyUI's +// /upload/image takes any file — and returns the name it was given there. func (c *Client) Upload(ctx context.Context, base, apiKey, name string, data []byte) (string, error) { base = strings.TrimRight(base, "/") var body bytes.Buffer diff --git a/internal/control/catalog/catalog.go b/internal/control/catalog/catalog.go index c32e9d9..a428769 100644 --- a/internal/control/catalog/catalog.go +++ b/internal/control/catalog/catalog.go @@ -590,9 +590,13 @@ type ComfyUI struct { // must. Checkpoint string `json:"checkpoint,omitempty" koanf:"checkpoint"` // Output is what the workflow makes: pictures (the default), asked for - // at /v1/images/generations, or meshes, asked for as a job at - // /v1/3d/generations from a picture. + // at /v1/images/generations; meshes, asked for as a job at + // /v1/3d/generations from a picture; or video, audio or files. Every + // kind is asked for as a job at /v1/workflows/runs too. Output string `json:"output,omitempty" koanf:"output"` + // Inputs are what the workflow takes from a request at + // /v1/workflows/runs, each the placeholder of its name. + Inputs []Input `json:"inputs,omitempty" koanf:"inputs"` // Nodes are custom nodes the workflow needs, put in each replica's own // custom_nodes at the commit named. Nodes []CustomNode `json:"nodes,omitempty" koanf:"nodes"` @@ -705,11 +709,14 @@ func (m Model) validateComfyUI() error { } } } - if out := m.comfyOutput(); out != OutputPictures && out != OutputMeshes { - return fmt.Errorf("model %q: comfyui.output %q is not pictures or meshes", m.Name, out) + if err := validOutput(m.Name, m.comfyOutput()); err != nil { + return err } - if m.comfyOutput() == OutputMeshes && w == "" { - return fmt.Errorf("model %q: a model that makes meshes names its workflow: no built-in makes them", m.Name) + if out := m.comfyOutput(); out != OutputPictures && w == "" { + return fmt.Errorf("model %q: a model that makes %s names its workflow: no built-in makes them", m.Name, out) + } + if err := validateInputs(m.Name, m.ComfyInputs()); err != nil { + return err } if _, ckpt := m.ComfyWorkflow(); ckpt == "" { return fmt.Errorf("model %q: its source is a set of files, so comfyui.checkpoint must name the one the workflow loads", @@ -1463,11 +1470,13 @@ func (e *External) validate(model string, profiles int) error { if err := checkWorkflow(model, e.Workflow); err != nil { return err } - if e.Output != "" && e.Output != OutputPictures && e.Output != OutputMeshes { - return fmt.Errorf("model %q: external output %q is not pictures or meshes", model, e.Output) + if e.Output != "" { + if err := validOutput(model, e.Output); err != nil { + return err + } } - if e.Output == OutputMeshes && e.Workflow == "" { - return fmt.Errorf("model %q: a model that makes meshes names its workflow: no built-in makes them", model) + if e.Output != "" && e.Output != OutputPictures && e.Workflow == "" { + return fmt.Errorf("model %q: a model that makes %s names its workflow: no built-in makes them", model, e.Output) } default: return fmt.Errorf("model %q: external api %q is not openai or comfyui", model, e.API) diff --git a/internal/control/catalog/workflow.go b/internal/control/catalog/workflow.go new file mode 100644 index 0000000..d39ae1e --- /dev/null +++ b/internal/control/catalog/workflow.go @@ -0,0 +1,156 @@ +package catalog + +import ( + "fmt" + "regexp" + "slices" + "strings" +) + +// What a ComfyUI workflow takes and what it makes, said in the catalog so a +// model is data rather than code: a request is checked against its inputs +// before a job is written, a client can build a form from /v1/models, and +// the answer is picked from the files the workflow saved by its output kind. + +// The kinds of thing a ComfyUI model makes, beside pictures and meshes. +const ( + OutputVideo = "video" + OutputAudio = "audio" + // OutputFiles is whatever the workflow saves, of any type. + OutputFiles = "files" +) + +// Outputs are every output kind a ComfyUI model may name. +var Outputs = []string{OutputPictures, OutputMeshes, OutputVideo, OutputAudio, OutputFiles} + +// The types of a workflow's input. +const ( + InputText = "text" + InputInt = "int" + InputFloat = "float" + InputBool = "bool" + InputChoice = "choice" + InputImage = "image" + InputMask = "mask" + InputAudio = "audio" + InputVideo = "video" + InputFile = "file" +) + +// InputTypes are every type an input may have. +var InputTypes = []string{InputText, InputInt, InputFloat, InputBool, InputChoice, + InputImage, InputMask, InputAudio, InputVideo, InputFile} + +// Input is one thing a workflow takes from a request: the placeholder +// {{name}} in its graph, filled with the request's inputs.name. +type Input struct { + // Name is the placeholder's, and the request's key. + Name string `json:"name" koanf:"name"` + // Type is one of text, int, float, bool, choice, image, mask, audio, + // video, file. The last five are files, sent as base64, a data URL, or + // the id of a file uploaded to /v1/files, and put in ComfyUI's input + // directory; the placeholder gets the name they were given there. + Type string `json:"type" koanf:"type"` + // Required inputs must be in every request; the others take Default, + // or the workflow's own {{name|default}}. + Required bool `json:"required,omitempty" koanf:"required"` + Default any `json:"default,omitempty" koanf:"default"` + // Min and Max bound a number; for text, Max bounds its length. + Min *float64 `json:"min,omitempty" koanf:"min"` + Max *float64 `json:"max,omitempty" koanf:"max"` + // Choices are what a choice may be. + Choices []string `json:"choices,omitempty" koanf:"choices"` + // MaxMB bounds a file input; zero is the type's own bound. + MaxMB int `json:"max_mb,omitempty" koanf:"max_mb"` + // Doc says what the input is for, to whoever fills it in. + Doc string `json:"doc,omitempty" koanf:"doc"` +} + +// IsFile reports whether the input is a file put in ComfyUI's input directory. +func (in Input) IsFile() bool { + switch in.Type { + case InputImage, InputMask, InputAudio, InputVideo, InputFile: + return true + } + return false +} + +// MaxBytes is the most a file input may be. +func (in Input) MaxBytes() int64 { + if in.MaxMB > 0 { + return int64(in.MaxMB) << 20 + } + switch in.Type { + case InputImage, InputMask: + return 16 << 20 + case InputVideo: + return 256 << 20 + } + return 64 << 20 +} + +var inputName = regexp.MustCompile(`^[a-z][a-z0-9_]{0,63}$`) + +func (in Input) validate(model string) error { + if !inputName.MatchString(in.Name) { + return fmt.Errorf("model %q: input %q is not a name of lowercase letters, digits and '_'", model, in.Name) + } + if !slices.Contains(InputTypes, in.Type) { + return fmt.Errorf("model %q: input %q has type %q, not one of %s", model, in.Name, in.Type, strings.Join(InputTypes, ", ")) + } + if in.Type == InputChoice && len(in.Choices) == 0 { + return fmt.Errorf("model %q: input %q is a choice with no choices", model, in.Name) + } + if in.Type != InputChoice && len(in.Choices) > 0 { + return fmt.Errorf("model %q: input %q has choices but is not a choice", model, in.Name) + } + if in.Min != nil && in.Max != nil && *in.Min > *in.Max { + return fmt.Errorf("model %q: input %q has min above max", model, in.Name) + } + if in.Required && in.Default != nil { + return fmt.Errorf("model %q: input %q is required and has a default: it is one or the other", model, in.Name) + } + if in.IsFile() && in.Default != nil { + return fmt.Errorf("model %q: input %q is a file and cannot have a default", model, in.Name) + } + if in.MaxMB < 0 { + return fmt.Errorf("model %q: input %q has a negative max_mb", model, in.Name) + } + return nil +} + +// ComfyOutput is what a ComfyUI model makes: pictures unless it says. +func (m Model) ComfyOutput() string { return m.comfyOutput() } + +// ComfyInputs are what a ComfyUI model's workflow takes from a request. +func (m Model) ComfyInputs() []Input { + if m.ComfyUI == nil { + return nil + } + return m.ComfyUI.Inputs +} + +// RunsWorkflows reports whether a model answers /v1/workflows/runs: any the +// cluster's ComfyUI or a ComfyUI outside it serves. +func (m Model) RunsWorkflows() bool { return m.ServesComfyUI() } + +func validOutput(model, out string) error { + if !slices.Contains(Outputs, out) { + return fmt.Errorf("model %q: comfyui.output %q is not one of %s", model, out, strings.Join(Outputs, ", ")) + } + return nil +} + +func validateInputs(model string, inputs []Input) error { + seen := map[string]bool{} + for _, in := range inputs { + if err := in.validate(model); err != nil { + return err + } + if seen[in.Name] { + return fmt.Errorf("model %q: input %q is declared twice", model, in.Name) + } + seen[in.Name] = true + } + return nil +} diff --git a/internal/control/catalog/workflow_test.go b/internal/control/catalog/workflow_test.go new file mode 100644 index 0000000..3fc9680 --- /dev/null +++ b/internal/control/catalog/workflow_test.go @@ -0,0 +1,47 @@ +package catalog + +import ( + "strings" + "testing" +) + +func TestAWorkflowsInputsAreCheckedWhenTheCatalogIsBuilt(t *testing.T) { + one := 1.0 + zero := 0.0 + run := func(out string, inputs ...Input) Model { + return comfyModel("/w/ace.safetensors", &ComfyUI{Workflow: "w.json", Output: out, Inputs: inputs}) + } + cat, err := New([]Model{run(OutputAudio, + Input{Name: "tags", Type: InputText, Required: true}, + Input{Name: "strength", Type: InputFloat, Min: &zero, Max: &one, Default: 0.5}, + Input{Name: "style", Type: InputChoice, Choices: []string{"pop"}}, + Input{Name: "clip", Type: InputVideo, MaxMB: 512})}) + if err != nil { + t.Fatal(err) + } + m, _ := cat.Resolve("flux") + if !m.RunsWorkflows() || m.MakesImages() || m.MakesMeshes() || m.ComfyOutput() != OutputAudio || len(m.ComfyInputs()) != 4 { + t.Errorf("an audio model: runs %v pictures %v meshes %v output %q", m.RunsWorkflows(), m.MakesImages(), m.MakesMeshes(), m.ComfyOutput()) + } + if m.ComfyInputs()[3].MaxBytes() != 512<<20 || (Input{Type: InputImage}).MaxBytes() != 16<<20 { + t.Error("a file input's bound") + } + for name, c := range map[string]struct { + m Model + want string + }{ + "an unknown kind": {run("hologram"), "not one of pictures"}, + "video, no workflow": {comfyModel("/w/x", &ComfyUI{Output: OutputVideo}), "makes video names its workflow"}, + "a capital name": {run(OutputAudio, Input{Name: "Tags", Type: InputText}), "lowercase"}, + "an unknown type": {run(OutputAudio, Input{Name: "x", Type: "colour"}), "not one of text"}, + "a choice of nothing": {run(OutputAudio, Input{Name: "x", Type: InputChoice}), "no choices"}, + "min above max": {run(OutputAudio, Input{Name: "x", Type: InputInt, Min: &one, Max: &zero}), "min above max"}, + "required and default": {run(OutputAudio, Input{Name: "x", Type: InputInt, Required: true, Default: 3}), "one or the other"}, + "a file's default": {run(OutputAudio, Input{Name: "x", Type: InputImage, Default: "a.png"}), "cannot have a default"}, + "twice": {run(OutputAudio, Input{Name: "x", Type: InputText}, Input{Name: "x", Type: InputInt}), "declared twice"}, + } { + if _, err := New([]Model{c.m}); err == nil || !strings.Contains(err.Error(), c.want) { + t.Errorf("%s: %v", name, err) + } + } +} diff --git a/internal/docgen/reference_gen.go b/internal/docgen/reference_gen.go index 72abb08..bcd061d 100644 --- a/internal/docgen/reference_gen.go +++ b/internal/docgen/reference_gen.go @@ -70,7 +70,8 @@ var Reference = []Block{ {Package: "catalog", Name: "ComfyUI", Title: "ComfyUI", Doc: "", Fields: []Field{ {Key: "workflow", Type: "string", Doc: "Workflow is a built-in's name (flux-schnell, the default, or\nflux-dev) or a JSON file on the control plane, in ComfyUI's API\nformat.", Nested: ""}, {Key: "checkpoint", Type: "string", Doc: "Checkpoint is the file the workflow's {{checkpoint}} names: the\nsource's own name unless set, which a source that is a set of files\nmust.", Nested: ""}, - {Key: "output", Type: "string", Doc: "Output is what the workflow makes: pictures (the default), asked for\nat /v1/images/generations, or meshes, asked for as a job at\n/v1/3d/generations from a picture.", Nested: ""}, + {Key: "output", Type: "string", Doc: "Output is what the workflow makes: pictures (the default), asked for\nat /v1/images/generations; meshes, asked for as a job at\n/v1/3d/generations from a picture; or video, audio or files. Every\nkind is asked for as a job at /v1/workflows/runs too.", Nested: ""}, + {Key: "inputs", Type: "[]Input", Doc: "Inputs are what the workflow takes from a request at\n/v1/workflows/runs, each the placeholder of its name.", Nested: ""}, {Key: "nodes", Type: "[]CustomNode", Doc: "Nodes are custom nodes the workflow needs, put in each replica's own\ncustom_nodes at the commit named.", Nested: ""}, {Key: "environment", Type: "string", Doc: "Environment names the models that may share one ComfyUI replica: every\nmodel of an environment agrees on the custom nodes they both name, so\none replica can carry all of them and ComfyUI moves between their\nweights itself. Empty is an environment of the model alone.", Nested: ""}, }}, @@ -151,6 +152,17 @@ var Reference = []Block{ {Key: "format", Type: "string", Doc: "Format is qwen3guard (Qwen3Guard-Gen: \"Safety: Safe\", \"Unsafe\" or\n\"Controversial\", and the categories), llamaguard3 (Meta's Llama Guard\n3: \"safe\", or \"unsafe\" and the categories), shieldgemma (Google's\nShieldGemma: Yes or No), or json ({\"safe\": false, \"categories\":\n[\"...\"]}).", Nested: ""}, {Key: "categories", Type: "[]string", Doc: "Categories the model is asked about, for shieldgemma — each a policy\nsentence — and the names reported for llamaguard3's S1, S2, ….", Nested: ""}, }}, + {Package: "catalog", Name: "Input", Title: "Input", Doc: "", Fields: []Field{ + {Key: "name", Type: "string", Doc: "Name is the placeholder's, and the request's key.", Nested: ""}, + {Key: "type", Type: "string", Doc: "Type is one of text, int, float, bool, choice, image, mask, audio,\nvideo, file. The last five are files, sent as base64, a data URL, or\nthe id of a file uploaded to /v1/files, and put in ComfyUI's input\ndirectory; the placeholder gets the name they were given there.", Nested: ""}, + {Key: "required", Type: "bool", Doc: "Required inputs must be in every request; the others take Default,\nor the workflow's own {{name|default}}.", Nested: ""}, + {Key: "default", Type: "any", Doc: "", Nested: ""}, + {Key: "min", Type: "*float64", Doc: "Min and Max bound a number; for text, Max bounds its length.", Nested: ""}, + {Key: "max", Type: "*float64", Doc: "", Nested: ""}, + {Key: "choices", Type: "[]string", Doc: "Choices are what a choice may be.", Nested: ""}, + {Key: "max_mb", Type: "int", Doc: "MaxMB bounds a file input; zero is the type's own bound.", Nested: ""}, + {Key: "doc", Type: "string", Doc: "Doc says what the input is for, to whoever fills it in.", Nested: ""}, + }}, {Package: "guard", Name: "JudgePolicy", Title: "JudgePolicy", Doc: "", Fields: []Field{ {Key: "model", Type: "string", Doc: "Model is a catalog model with role: guard.", Nested: ""}, {Key: "models", Type: "[]string", Doc: "Models are judges asked in turn, instead of Model: the first that\nanswers gives the verdict. A judge on a GPU first and one on CPUs\nafter it is the GPU's speed while it is there, and the CPUs' when it\nis not — busy, failed, or the node down.", Nested: ""}, diff --git a/internal/docgen/render.go b/internal/docgen/render.go index 9577755..6101853 100644 --- a/internal/docgen/render.go +++ b/internal/docgen/render.go @@ -26,6 +26,7 @@ var Sources = []string{ "internal/agent/supervisor/supervisor.go", "internal/tlsconf/tlsconf.go", "internal/control/catalog/catalog.go", + "internal/control/catalog/workflow.go", "internal/router/admission/controller.go", "internal/accounting/cost.go", "internal/guard/policy.go", diff --git a/internal/router/handlers.go b/internal/router/handlers.go index 7e57abf..4f23ae7 100644 --- a/internal/router/handlers.go +++ b/internal/router/handlers.go @@ -143,6 +143,12 @@ func (r *Router) Handler() http.Handler { mux.HandleFunc("GET "+meshesPath, r.handleListJobs) mux.HandleFunc("GET "+meshesPath+"/{id}", r.handleGetJob) mux.HandleFunc("POST "+meshesPath+"/{id}/cancel", r.handleCancelJob) + // Workflow runs: whatever a ComfyUI model makes, from the inputs its + // catalog entry declares, as a job like a 3D model. + mux.HandleFunc("POST "+workflowsPath, r.createJobAt(workflowsPath)) + mux.HandleFunc("GET "+workflowsPath, r.handleListJobs) + mux.HandleFunc("GET "+workflowsPath+"/{id}", r.handleGetJob) + mux.HandleFunc("POST "+workflowsPath+"/{id}/cancel", r.handleCancelJob) // Knowledge bases: documents a person's llmgw-kb read and embedded. r.registerKB(mux) // Liveness: this process is up. It says nothing about whether it can @@ -1198,6 +1204,16 @@ func (r *Router) handleModels(w http.ResponseWriter, req *http.Request) { entry["capabilities"] = map[string]bool{ "tools": m.Capabilities.Tools, "vision": m.Capabilities.Vision, "embeddings": m.Capabilities.Embeddings, "images": m.MakesImages(), "meshes": m.MakesMeshes(), + "workflows": m.RunsWorkflows(), + } + // What a run of it takes and makes, so a client can build the form + // without the catalog in hand. + if m.RunsWorkflows() { + var inputs any = m.ComfyInputs() + if len(m.ComfyInputs()) == 0 { + inputs = []any{} + } + entry["workflow"] = map[string]any{"output": m.ComfyOutput(), "inputs": inputs} } data = append(data, entry) } diff --git a/internal/router/images.go b/internal/router/images.go index c370152..35bce2b 100644 --- a/internal/router/images.go +++ b/internal/router/images.go @@ -112,6 +112,18 @@ func parseImageRequest(body []byte) (imageRequest, error) { // endpoint may serve as it likes. func (r *Router) imageRouteRefusal(path, model string, body []byte) string { m, known := r.modelMap()[model] + if path == workflowsPath { + if !known { + return "" + } + if !m.RunsWorkflows() { + return fmt.Sprintf("%s runs no ComfyUI workflow", m.Name) + } + if _, err := parseRunRequest(m, body); err != nil { + return err.Error() + } + return "" + } if path == meshesPath { if known && !m.MakesMeshes() { if m.MakesImages() { @@ -137,6 +149,9 @@ func (r *Router) imageRouteRefusal(path, model string, body []byte) string { return "" } if known && m.ServesComfyUI() { + if !m.MakesMeshes() && !m.MakesImages() { + return fmt.Sprintf("%s makes %s: ask for it at %s, as a job", m.Name, m.ComfyOutput(), workflowsPath) + } if m.MakesMeshes() { return fmt.Sprintf("%s makes 3D models: ask for one at %s, as a job", m.Name, meshesPath) } @@ -149,11 +164,15 @@ func (r *Router) imageRouteRefusal(path, model string, body []byte) string { // proxy would: a failure of this replica that another may not have is // retryable, and nothing is written until the pictures are all in hand. func (r *Router) generateImages(ctx context.Context, w http.ResponseWriter, req *parsedRequest, ep endpoint.Info, body []byte) (proxy.Outcome, error) { - if req.Path == meshesPath { + switch req.Path { + case meshesPath: return r.generateMeshes(ctx, w, req, ep, body) - } - if req.Path != imagesPath { - return proxy.Outcome{}, fmt.Errorf("%s is a ComfyUI server: it answers %s and %s only", ep.ID, imagesPath, meshesPath) + case workflowsPath: + return r.generateRun(ctx, w, req, ep, body) + case imagesPath: + default: + return proxy.Outcome{}, fmt.Errorf("%s is a ComfyUI server: it answers %s, %s and %s only", ep.ID, + imagesPath, meshesPath, workflowsPath) } in, err := parseImageRequest(body) if err != nil { diff --git a/internal/router/jobs.go b/internal/router/jobs.go index 80cbbb9..f688614 100644 --- a/internal/router/jobs.go +++ b/internal/router/jobs.go @@ -66,11 +66,21 @@ func (r *Router) jobJSON(ctx context.Context, j store.Job) map[string]any { return out } -// handleCreateJob queues a request for a 3D model. Refused now: no key, a -// model that is not there or not allowed, a request the model cannot serve. -// Refused when it is served, and said as the job's error: limits, the -// guardrails, a model that will not come up. +// handleCreateJob queues a request for a 3D model. func (r *Router) handleCreateJob(w http.ResponseWriter, req *http.Request) { + r.createJobAt(meshesPath)(w, req) +} + +// createJobAt queues a request at path: a 3D model, a workflow run. Refused +// now: no key, a model that is not there or not allowed, a request the model +// cannot serve, a file it names that is not the caller's. Refused when it is +// served, and said as the job's error: limits, the guardrails, a model that +// will not come up. +func (r *Router) createJobAt(path string) http.HandlerFunc { + return func(w http.ResponseWriter, req *http.Request) { r.createJob(w, req, path) } +} + +func (r *Router) createJob(w http.ResponseWriter, req *http.Request, path string) { p, ok := r.batchPrincipal(w, req) if !ok { return @@ -100,12 +110,21 @@ func (r *Router) handleCreateJob(w http.ResponseWriter, req *http.Request) { auth.WriteForbiddenModel(w, in.Model) return } - if why := r.imageRouteRefusal(meshesPath, in.Model, body); why != "" { + if why := r.imageRouteRefusal(path, in.Model, body); why != "" { writeError(w, http.StatusBadRequest, "invalid_request_error", why) return } + if path == workflowsPath { + if m, known := r.modelMap()[in.Model]; known { + run, _ := parseRunRequest(m, body) + if err := r.checkRunFiles(req.Context(), ownerOf(p), run); err != nil { + writeError(w, http.StatusBadRequest, "invalid_request_error", err.Error()) + return + } + } + } now := time.Now() - j := store.Job{ID: newID("job_"), Owner: ownerOf(p), KeyID: p.KeyID, Endpoint: meshesPath, Model: in.Model, + j := store.Job{ID: newID("job_"), Owner: ownerOf(p), KeyID: p.KeyID, Endpoint: path, Model: in.Model, Input: body, Status: store.JobQueued, CreatedAt: now, ExpiresAt: now.Add(jobWindow)} if err := r.store.PutJob(req.Context(), j); err != nil { writeError(w, http.StatusInternalServerError, "server_error", err.Error()) diff --git a/internal/router/jobworker.go b/internal/router/jobworker.go index af7abf4..872ec6f 100644 --- a/internal/router/jobworker.go +++ b/internal/router/jobworker.go @@ -25,7 +25,7 @@ const ( jobPoll = 5 * time.Second // jobCancelCheck is how often a running job looks for a cancel. jobCancelCheck = 2 * time.Second - // jobMaxResult bounds what one job may answer: a mesh, not a film. + // jobMaxResult bounds what one job may answer: a mesh, a short video. jobMaxResult = 256 << 20 ) @@ -264,6 +264,22 @@ func extensionFor(contentType string) string { return ".jpg" case "image/webp": return ".webp" + case "image/gif": + return ".gif" + case "video/mp4": + return ".mp4" + case "video/webm": + return ".webm" + case "video/quicktime": + return ".mov" + case "audio/wav", "audio/wave", "audio/x-wav": + return ".wav" + case "audio/mpeg": + return ".mp3" + case "audio/flac": + return ".flac" + case "audio/ogg": + return ".ogg" } return ".bin" } diff --git a/internal/router/workflows.go b/internal/router/workflows.go new file mode 100644 index 0000000..433e39e --- /dev/null +++ b/internal/router/workflows.go @@ -0,0 +1,501 @@ +package router + +import ( + "bytes" + "context" + "crypto/rand" + "encoding/base64" + "encoding/binary" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "hash/crc32" + "math" + "net/http" + "path" + "slices" + "sort" + "strconv" + "strings" + "time" + "unicode/utf8" + + "gitea.mynodes.xyz/ilker/llmgw/internal/auth" + "gitea.mynodes.xyz/ilker/llmgw/internal/comfy" + "gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog" + "gitea.mynodes.xyz/ilker/llmgw/internal/router/endpoint" + "gitea.mynodes.xyz/ilker/llmgw/internal/router/proxy" +) + +// Workflow runs: whatever a ComfyUI model makes, from the inputs its catalog +// entry declares. Pictures and meshes have fixed shapes of their own +// (images.go, meshes.go); this is the one shape for everything, so a model +// is added to the catalog as data — its workflow, its inputs, its output +// kind — rather than as code. A run is a job like a 3D model (jobs.go): it +// is queued, asked after, and its result is a file. + +// workflowsPath is where workflow runs are asked for. The synchronous path +// behind it is served to the job worker only; clients get a job. +const workflowsPath = "/v1/workflows/runs" + +// runRequest is a request to run a model's workflow. +type runRequest struct { + Model string `json:"model"` + Inputs map[string]json.RawMessage `json:"inputs"` + Seed *int64 `json:"seed"` + + // values are the inputs that are not files, as the workflow gets them. + values map[string]any + // files are the file inputs, by input name. + files map[string]runFile +} + +// runFile is a file input: its bytes, or the id of a file in /v1/files to be +// read when the run is served. +type runFile struct { + data []byte + fileID string + ext string +} + +// maxTextInput bounds a text input that names no max of its own. +const maxTextInput = 32 << 10 + +// parseRunRequest checks a request against what the model's workflow takes. +// It refuses an input the workflow does not take, a required one left out, a +// value of the wrong type or out of its bounds, and a file that is not what +// the input says. +func parseRunRequest(m catalog.Model, body []byte) (runRequest, error) { + var in runRequest + if err := json.Unmarshal(body, &in); err != nil { + return in, err + } + in.values, in.files = map[string]any{}, map[string]runFile{} + declared := m.ComfyInputs() + for name := range in.Inputs { + if !slices.ContainsFunc(declared, func(d catalog.Input) bool { return d.Name == name }) { + names := make([]string, 0, len(declared)) + for _, d := range declared { + names = append(names, d.Name) + } + if len(names) == 0 { + return in, fmt.Errorf("%s takes no inputs, and was given %q", m.Name, name) + } + return in, fmt.Errorf("%s takes no input %q (it takes %s)", m.Name, name, strings.Join(names, ", ")) + } + } + for _, d := range declared { + raw, given := in.Inputs[d.Name] + if !given || string(raw) == "null" { + if d.Required { + return in, fmt.Errorf("input %q is required: %s", d.Name, describeInput(d)) + } + if d.Default != nil { + in.values[d.Name] = d.Default + } + continue + } + if d.IsFile() { + f, err := parseFileInput(d, raw) + if err != nil { + return in, err + } + in.files[d.Name] = f + continue + } + v, err := parseValueInput(d, raw) + if err != nil { + return in, err + } + in.values[d.Name] = v + } + return in, nil +} + +func describeInput(d catalog.Input) string { + s := d.Type + if d.Type == catalog.InputChoice { + s += " of " + strings.Join(d.Choices, ", ") + } + if d.Doc != "" { + s += " — " + d.Doc + } + return s +} + +func parseValueInput(d catalog.Input, raw json.RawMessage) (any, error) { + bounds := func(f float64) error { + if d.Min != nil && f < *d.Min || d.Max != nil && f > *d.Max { + return fmt.Errorf("input %q is %v, outside %s", d.Name, f, boundsText(d)) + } + return nil + } + switch d.Type { + case catalog.InputText: + var s string + if json.Unmarshal(raw, &s) != nil { + return nil, fmt.Errorf("input %q is text", d.Name) + } + limit := maxTextInput + if d.Max != nil { + limit = int(*d.Max) + } + if n := utf8.RuneCountInString(s); n > limit { + return nil, fmt.Errorf("input %q is %d characters, past %d", d.Name, n, limit) + } + return s, nil + case catalog.InputInt: + var f float64 + if json.Unmarshal(raw, &f) != nil || f != math.Trunc(f) || math.Abs(f) > 1<<53 { + return nil, fmt.Errorf("input %q is a whole number", d.Name) + } + return int64(f), bounds(f) + case catalog.InputFloat: + var f float64 + if json.Unmarshal(raw, &f) != nil || math.IsNaN(f) || math.IsInf(f, 0) { + return nil, fmt.Errorf("input %q is a number", d.Name) + } + return f, bounds(f) + case catalog.InputBool: + var b bool + if json.Unmarshal(raw, &b) != nil { + return nil, fmt.Errorf("input %q is true or false", d.Name) + } + return b, nil + case catalog.InputChoice: + var s string + if json.Unmarshal(raw, &s) != nil || !slices.Contains(d.Choices, s) { + return nil, fmt.Errorf("input %q is one of %s", d.Name, strings.Join(d.Choices, ", ")) + } + return s, nil + } + return nil, fmt.Errorf("input %q has type %q, which takes no value", d.Name, d.Type) +} + +func boundsText(d catalog.Input) string { + switch { + case d.Min != nil && d.Max != nil: + return fmt.Sprintf("%v to %v", *d.Min, *d.Max) + case d.Min != nil: + return fmt.Sprintf("at least %v", *d.Min) + case d.Max != nil: + return fmt.Sprintf("at most %v", *d.Max) + } + return "its bounds" +} + +// parseFileInput reads a file input: base64, a data URL, or a file id. +func parseFileInput(d catalog.Input, raw json.RawMessage) (runFile, error) { + var s string + if json.Unmarshal(raw, &s) != nil || strings.TrimSpace(s) == "" { + return runFile{}, fmt.Errorf("input %q is a file: base64, a data URL, or the id of a file uploaded to /v1/files", d.Name) + } + s = strings.TrimSpace(s) + if strings.HasPrefix(s, "file-") && len(s) < 80 && !strings.ContainsAny(s, " /") { + return runFile{fileID: s}, nil + } + if strings.HasPrefix(s, "data:") { + _, after, ok := strings.Cut(s, ";base64,") + if !ok { + return runFile{}, fmt.Errorf("input %q is a data URL without base64 in it", d.Name) + } + s = after + } + if int64(base64.StdEncoding.DecodedLen(len(s))) > d.MaxBytes()+3 { + return runFile{}, fmt.Errorf("input %q is past %d MB: upload it to /v1/files and give its id", d.Name, d.MaxBytes()>>20) + } + b, err := base64.StdEncoding.DecodeString(s) + if err != nil { + return runFile{}, fmt.Errorf("input %q is not base64: %v", d.Name, err) + } + return checkFileInput(d, b) +} + +// checkFileInput is a file input's bytes, if they are what the input says. +func checkFileInput(d catalog.Input, b []byte) (runFile, error) { + if int64(len(b)) > d.MaxBytes() { + return runFile{}, fmt.Errorf("input %q is past %d MB", d.Name, d.MaxBytes()>>20) + } + ext, kind := sniffFile(b) + want := map[string]string{catalog.InputImage: "image", catalog.InputMask: "image", + catalog.InputAudio: "audio", catalog.InputVideo: "video"}[d.Type] + if want != "" && kind != want { + return runFile{}, fmt.Errorf("input %q is %s, and this is not one it can read", d.Name, map[string]string{ + "image": "a PNG, JPEG, WebP or GIF picture", "audio": "WAV, MP3, FLAC or Ogg audio", + "video": "an MP4, WebM or QuickTime video"}[want]) + } + if ext == "" { + ext = ".bin" + } + return runFile{data: b, ext: ext}, nil +} + +// sniffFile is a file's extension and kind (image, audio, video) by its +// first bytes, or empty when it is none of the kinds a workflow loads. +func sniffFile(b []byte) (ext, kind string) { + has := func(off int, sig string) bool { return len(b) >= off+len(sig) && string(b[off:off+len(sig)]) == sig } + switch { + case has(0, "\x89PNG\r\n\x1a\n"): + return ".png", "image" + case has(0, "\xff\xd8\xff"): + return ".jpg", "image" + case has(0, "RIFF") && has(8, "WEBP"): + return ".webp", "image" + case has(0, "GIF87a"), has(0, "GIF89a"): + return ".gif", "image" + case has(0, "RIFF") && has(8, "WAVE"): + return ".wav", "audio" + case has(0, "fLaC"): + return ".flac", "audio" + case has(0, "OggS"): + return ".ogg", "audio" + case has(0, "ID3"), len(b) > 2 && b[0] == 0xff && b[1]&0xe0 == 0xe0: + return ".mp3", "audio" + case has(4, "ftypqt"): + return ".mov", "video" + case has(4, "ftyp"): + return ".mp4", "video" + case has(0, "\x1a\x45\xdf\xa3"): + return ".webm", "video" + } + return "", "" +} + +// runFileIDs are the file ids a run's inputs name. +func (in runRequest) runFileIDs() []string { + var ids []string + for _, f := range in.files { + if f.fileID != "" { + ids = append(ids, f.fileID) + } + } + sort.Strings(ids) + return ids +} + +// checkRunFiles refuses a run naming a file its owner does not have. +func (r *Router) checkRunFiles(ctx context.Context, owner string, in runRequest) error { + for _, id := range in.runFileIDs() { + f, err := r.store.FileByID(ctx, id) + if err != nil || f.Owner != owner { + return fmt.Errorf("no file %q", id) + } + } + return nil +} + +// generateRun answers a workflow run at a ComfyUI replica: the file inputs +// put in its input directory, the workflow filled in and run, and the file +// it made written back with its type — what the job keeps. +func (r *Router) generateRun(ctx context.Context, w http.ResponseWriter, req *parsedRequest, ep endpoint.Info, body []byte) (proxy.Outcome, error) { + m, known := r.modelMap()[req.Model] + if !known { + return proxy.Outcome{}, fmt.Errorf("%s: no catalog entry for %q says what its workflow takes", ep.ID, req.Model) + } + in, err := parseRunRequest(m, body) + if err != nil { + return proxy.Outcome{}, err + } + tpl, err := comfy.Load(ep.Workflow) + if err != nil { + return proxy.Outcome{}, fmt.Errorf("%s: %w", ep.ID, err) + } + if err := checkPlaceholders(m, tpl); err != nil { + return proxy.Outcome{}, fmt.Errorf("%s: %w", ep.ID, err) + } + if t := r.cfg.Timeouts.Total; t > 0 { + var cancel context.CancelFunc + ctx, cancel = context.WithTimeout(ctx, t) + defer cancel() + } + start := time.Now() + fail := func(err error) (proxy.Outcome, error) { + var ce *comfy.Error + switch { + case errors.As(err, &ce): + return proxy.Outcome{Retryable: ce.Retryable}, err + case errors.Is(err, context.Canceled) && ctxCauseIsClient(ctx): + return proxy.Outcome{}, proxy.ErrClientGone + case errors.Is(err, context.DeadlineExceeded): + return proxy.Outcome{}, fmt.Errorf("%s finished nothing within %s", ep.ID, r.cfg.Timeouts.Total) + } + return proxy.Outcome{}, err + } + var upstream struct { + Model string `json:"model"` + } + _ = json.Unmarshal(body, &upstream) + p := comfy.Params{Checkpoint: upstream.Model, Seed: comfy.RandomSeed(), Extra: in.values} + if in.Seed != nil { + p.Seed = *in.Seed + } + if s, ok := in.values["seed"].(int64); ok { + p.Seed = s + } + // Each file under a name of its own: two runs' files must not overwrite + // each other in the replica's input directory. + names := make([]string, 0, len(in.files)) + for name := range in.files { + names = append(names, name) + } + sort.Strings(names) + for _, name := range names { + f := in.files[name] + if f.fileID != "" { + pr, _ := auth.FromContext(ctx) + d := m.ComfyInputs()[slices.IndexFunc(m.ComfyInputs(), func(x catalog.Input) bool { return x.Name == name })] + meta, err := r.store.FileByID(ctx, f.fileID) + if err != nil || meta.Owner != ownerOf(pr) { + return proxy.Outcome{}, fmt.Errorf("input %q: no file %q", name, f.fileID) + } + b, err := r.store.FileContent(ctx, f.fileID) + if err != nil { + return proxy.Outcome{}, fmt.Errorf("input %q: reading file %q: %w", name, f.fileID, err) + } + if f, err = checkFileInput(d, b); err != nil { + return proxy.Outcome{}, err + } + in.files[name] = f + } + rnd := make([]byte, 8) + _, _ = rand.Read(rnd) + f = in.files[name] + up, err := r.comfy.Upload(ctx, ep.URL, ep.APIKey, "llmgw-"+hex.EncodeToString(rnd)+f.ext, f.data) + if err != nil { + return fail(err) + } + p.Extra[name] = up + } + workflow, err := tpl.Fill(p) + if err != nil { + return proxy.Outcome{}, fmt.Errorf("%s: %w", ep.ID, err) + } + files, err := r.comfy.GenerateWatch(ctx, ep.URL, ep.APIKey, workflow, r.jobProgress(ctx, req)) + if err != nil { + return fail(err) + } + out, err := pickOutput(m.ComfyOutput(), files) + if err != nil { + return proxy.Outcome{}, fmt.Errorf("%s: %w", ep.ID, err) + } + data, ct := out.Data, out.ContentType + if ct == "model/obj" { + if data, err = objToGLB(data); err != nil { + return proxy.Outcome{}, fmt.Errorf("%s: the mesh it saved: %w", ep.ID, err) + } + ct = "model/gltf-binary" + } + if m.License != nil { + if data, err = withLicense(data, ct, *m.License); err != nil { + // Refused rather than sent without it: the file travels without + // the API, and the notice is what the terms ask to travel too. + return proxy.Outcome{}, fmt.Errorf("%s: the licence could not be written into the file: %w", ep.ID, err) + } + w.Header().Set("x-llmgw-license", m.License.Name) + } + if _, ok := tpl.Placeholders()["seed"]; ok { + w.Header().Set("x-llmgw-seed", strconv.FormatInt(p.Seed, 10)) + } + w.Header().Set("Content-Type", ct) + w.WriteHeader(http.StatusOK) + _, _ = w.Write(data) + took := time.Since(start) + o := proxy.Outcome{Wrote: true, Status: http.StatusOK, Bytes: int64(len(data)), TTFT: took, Decode: took} + switch m.ComfyOutput() { + case catalog.OutputPictures: + o.Images = 1 + case catalog.OutputMeshes: + o.Meshes = 1 + } + return o, nil +} + +// checkPlaceholders refuses a workflow that wants what no request can give: +// a placeholder that is neither an input, nor one llmgw fills itself, nor +// has a default of its own. The catalog's to fix, said as such. +func checkPlaceholders(m catalog.Model, tpl *comfy.Template) error { + var missing []string + for name, hasDefault := range tpl.Placeholders() { + if hasDefault || slices.Contains(comfy.Builtin, name) || + slices.ContainsFunc(m.ComfyInputs(), func(d catalog.Input) bool { return d.Name == name }) { + continue + } + missing = append(missing, name) + } + if len(missing) > 0 { + sort.Strings(missing) + return fmt.Errorf("the workflow has placeholders %s that the catalog declares no input for and that have no default", + strings.Join(missing, ", ")) + } + return nil +} + +// pickOutput is the file a run answers with: the first the workflow saved of +// the model's output kind, or the first of any kind for files. +func pickOutput(kind string, files []comfy.Image) (comfy.Image, error) { + is := func(ct string) bool { + switch kind { + case catalog.OutputPictures: + return strings.HasPrefix(ct, "image/") && ct != "image/gif" + case catalog.OutputMeshes: + return isMeshType(ct) + case catalog.OutputVideo: + return strings.HasPrefix(ct, "video/") || ct == "image/gif" || ct == "image/webp" + case catalog.OutputAudio: + return strings.HasPrefix(ct, "audio/") + } + return true + } + for _, f := range files { + if is(f.ContentType) { + return f, nil + } + } + saved := make([]string, 0, len(files)) + for _, f := range files { + saved = append(saved, path.Base(f.Name)+" ("+f.ContentType+")") + } + return comfy.Image{}, fmt.Errorf("the workflow saved no %s: it saved %s", kind, strings.Join(saved, ", ")) +} + +// withLicense is a file with its licence written in, where its format has a +// place for one: a GLB's asset.extras, a PNG's text. Any other is returned as +// it is, and the licence goes beside it in the job's answer only. +func withLicense(data []byte, ct string, lic catalog.License) ([]byte, error) { + switch ct { + case "model/gltf-binary": + return glbWithLicense(data, lic) + case "image/png": + return pngWithLicense(data, lic) + } + return data, nil +} + +// pngWithLicense is a PNG with an iTXt chunk "License" after its header, +// holding the licence as JSON. The workflow's own text chunks were stripped +// when the file was read; this one is the operator's to say. +func pngWithLicense(png []byte, lic catalog.License) ([]byte, error) { + const sig = "\x89PNG\r\n\x1a\n" + if len(png) < len(sig)+25 || string(png[:len(sig)]) != sig || string(png[12:16]) != "IHDR" { + return nil, errors.New("not a PNG") + } + text, err := json.Marshal(lic) + if err != nil { + return nil, err + } + // keyword, NUL, compression flag 0, method 0, empty language, NUL, empty + // translated keyword, NUL, then the UTF-8 text. + var chunk bytes.Buffer + chunk.WriteString("iTXt") + chunk.WriteString("License\x00\x00\x00\x00\x00") + chunk.Write(text) + body := chunk.Bytes() + out := make([]byte, 0, len(png)+len(body)+8) + ihdrEnd := len(sig) + 8 + 13 + 4 + out = append(out, png[:ihdrEnd]...) + out = binary.BigEndian.AppendUint32(out, uint32(len(body)-4)) //nolint:gosec // a licence's JSON + out = append(out, body...) + out = binary.BigEndian.AppendUint32(out, crc32.ChecksumIEEE(body)) + return append(out, png[ihdrEnd:]...), nil +} diff --git a/internal/router/workflows_test.go b/internal/router/workflows_test.go new file mode 100644 index 0000000..95d1e4d --- /dev/null +++ b/internal/router/workflows_test.go @@ -0,0 +1,296 @@ +package router + +import ( + "bytes" + "crypto/sha256" + "encoding/base64" + "encoding/binary" + "encoding/json" + "fmt" + "image" + "image/png" + "io" + "net/http" + "net/http/httptest" + "path/filepath" + "strings" + "sync" + "testing" + "time" + + "gitea.mynodes.xyz/ilker/llmgw/internal/comfy" + "gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog" + "gitea.mynodes.xyz/ilker/llmgw/internal/router/endpoint" + "gitea.mynodes.xyz/ilker/llmgw/internal/store" +) + +// testWAV is the smallest WAV header and a few samples. +func testWAV() []byte { + var b bytes.Buffer + data := []byte{0, 0, 1, 0, 2, 0, 3, 0} + b.WriteString("RIFF") + _ = binary.Write(&b, binary.LittleEndian, uint32(36+len(data))) + b.WriteString("WAVEfmt ") + _ = binary.Write(&b, binary.LittleEndian, []any{uint32(16), uint16(1), uint16(1), uint32(8000), uint32(16000), uint16(2), uint16(16)}) + b.WriteString("data") + _ = binary.Write(&b, binary.LittleEndian, uint32(len(data))) + b.Write(data) + return b.Bytes() +} + +// runComfy is a ComfyUI whose workflow saves what it is told to: a song, a +// picture, under whatever key its save node uses. +type runComfy struct { + srv *httptest.Server + mu sync.Mutex + uploaded []string + workflow map[string]any + out []byte + outName string +} + +func newRunComfy(t *testing.T, out []byte, outName string) *runComfy { + c := &runComfy{out: out, outName: outName} + c.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/upload/image": + f, h, err := r.FormFile("image") + if err != nil { + http.Error(w, err.Error(), 400) + return + } + b, _ := io.ReadAll(f) + ext, _ := sniffFile(b) + c.mu.Lock() + c.uploaded = append(c.uploaded, h.Filename+":"+ext) + c.mu.Unlock() + fmt.Fprintf(w, `{"name":%q,"subfolder":"","type":"input"}`, h.Filename) + case "/prompt": + var b struct { + Prompt map[string]any `json:"prompt"` + } + _ = json.NewDecoder(r.Body).Decode(&b) + c.mu.Lock() + c.workflow = b.Prompt + c.mu.Unlock() + fmt.Fprint(w, `{"prompt_id":"r1"}`) + case "/history/r1": + fmt.Fprintf(w, `{"r1":{"status":{"status_str":"success","completed":true},"outputs":{ + "9":{"audio":[{"filename":%q,"subfolder":"","type":"output"}]}}}}`, c.outName) + case "/view": + w.Header().Set("Content-Type", "application/octet-stream") + _, _ = w.Write(c.out) + default: + http.NotFound(w, r) + } + })) + t.Cleanup(c.srv.Close) + return c +} + +const songWorkflow = `{ + "1": {"class_type": "LoadImage", "inputs": {"image": "{{cover}}"}}, + "2": {"class_type": "TextEncodeAceStepAudio", "inputs": {"tags": "{{tags}}", "lyrics": "{{lyrics|}}", "strength": "{{strength}}"}}, + "3": {"class_type": "EmptyAceStepLatentAudio", "inputs": {"seconds": "{{seconds}}", "style": "{{style}}"}}, + "4": {"class_type": "KSampler", "inputs": {"seed": "{{seed}}", "model": "{{checkpoint}}"}}, + "9": {"class_type": "SaveAudio", "inputs": {"filename_prefix": "audio/song"}} +}` + +func f64(v float64) *float64 { return &v } + +func runRouter(t *testing.T, c *runComfy, workflow string, output string) (*Router, string, string) { + t.Helper() + cfg := baseConfig() + cfg.Models = []catalog.Model{ + {Name: "ace", Source: "external", License: &catalog.License{ID: "Apache-2.0", Name: "Apache 2.0"}, + External: &catalog.External{URL: c.srv.URL, Model: "ace_step.safetensors", API: catalog.APIComfyUI, + Output: output, Workflow: "w.json"}, + ComfyUI: &catalog.ComfyUI{Inputs: []catalog.Input{ + {Name: "tags", Type: catalog.InputText, Required: true, Max: f64(200), Doc: "genres and moods"}, + {Name: "lyrics", Type: catalog.InputText}, + {Name: "strength", Type: catalog.InputFloat, Min: f64(0), Max: f64(1), Default: 0.7}, + {Name: "seconds", Type: catalog.InputInt, Min: f64(5), Max: f64(240), Default: 30}, + {Name: "style", Type: catalog.InputChoice, Choices: []string{"pop", "metal"}, Default: "pop"}, + {Name: "cover", Type: catalog.InputImage, Required: true}, + }}}, + {Name: "flux", Source: "external", External: &catalog.External{URL: c.srv.URL, + Model: "flux1-schnell-fp8.safetensors", API: catalog.APIComfyUI}}, + } + cfg.Auth.Enabled = true + cfg.Auth.Store = filepath.Join(t.TempDir(), "store.db") + cfg.Auth.FlushInterval = 20 * time.Millisecond + cfg.Auth.BatchSize = 1 + rt := newTestRouter(t, cfg) + rt.comfy = &comfy.Client{Poll: time.Millisecond} + rt.SetEndpoints(t.Context(), []endpoint.Info{ + {ID: "external:ace", URL: c.srv.URL, Models: []string{"ace"}, Engine: endpoint.EngineComfyUI, + Pool: "external", Weight: 1, UpstreamModel: "ace_step.safetensors", Workflow: workflow}, + {ID: "external:flux", URL: c.srv.URL, Models: []string{"flux"}, Engine: endpoint.EngineComfyUI, + Pool: "external", Weight: 1, UpstreamModel: "flux1-schnell-fp8.safetensors"}, + }) + org, _ := rt.store.CreateOrg(t.Context(), "studio") + team, _ := rt.store.CreateTeam(t.Context(), org, "sound") + project, err := rt.store.CreateProject(t.Context(), team, "score") + if err != nil { + t.Fatal(err) + } + secret := store.KeyPrefix + strings.Repeat("w", 43) + sum := sha256.Sum256([]byte(secret)) + if err := rt.store.CreateKey(t.Context(), store.Key{ID: "key-w", Hash: sum[:], Prefix: "wwww", Name: "sound", + ProjectID: project, CreatedAt: time.Now()}); err != nil { + t.Fatal(err) + } + return rt, secret, project +} + +func TestAWorkflowRunsFromTheInputsItsCatalogEntryDeclares(t *testing.T) { + wav := testWAV() + c := newRunComfy(t, wav, "song_00001_.wav") + rt, secret, _ := runRouter(t, c, songWorkflow, catalog.OutputAudio) + h := rt.Handler() + + code, j, raw := call(t, h, secret, http.MethodPost, workflowsPath, map[string]any{"model": "ace", "seed": 5, + "inputs": map[string]any{"tags": "synthwave, driving", "seconds": 60, "cover": "data:image/png;base64," + testPicture(t)}}) + if code != http.StatusAccepted { + t.Fatalf("%d %s", code, raw) + } + got := waitJob(t, h, secret, j["id"].(string), jobDone) + if got["status"] != "completed" || got["content_type"] != "audio/wav" { + t.Fatalf("job %v", got) + } + code, _, file := call(t, h, secret, http.MethodGet, "/v1/files/"+got["output_file_id"].(string)+"/content", nil) + if code != http.StatusOK || file != string(wav) { + t.Fatalf("the result is not the song: %d", code) + } + c.mu.Lock() + wf, _ := json.Marshal(c.workflow) + uploaded := strings.Join(c.uploaded, ",") + c.mu.Unlock() + if !strings.HasPrefix(uploaded, "llmgw-") || !strings.HasSuffix(uploaded, ".png:.png") { + t.Errorf("uploaded %q", uploaded) + } + name := strings.TrimSuffix(uploaded, ":.png") + for _, want := range []string{`"image":"` + name + `"`, `"tags":"synthwave, driving"`, `"seconds":60`, + `"strength":0.7`, `"style":"pop"`, `"lyrics":""`, `"seed":5`, `"model":"ace_step.safetensors"`} { + if !strings.Contains(string(wf), want) { + t.Errorf("%s not in the workflow sent: %s", want, wf) + } + } +} + +func TestARunIsRefusedWhatItsWorkflowDoesNotTake(t *testing.T) { + c := newRunComfy(t, testWAV(), "song.wav") + rt, secret, _ := runRouter(t, c, songWorkflow, catalog.OutputAudio) + h := rt.Handler() + cover := testPicture(t) + in := func(extra map[string]any) map[string]any { + m := map[string]any{"tags": "lofi", "cover": cover} + for k, v := range extra { + m[k] = v + } + return map[string]any{"model": "ace", "inputs": m} + } + for name, c := range map[string]struct { + body map[string]any + want string + }{ + "no tags": {map[string]any{"model": "ace", "inputs": map[string]any{"cover": cover}}, `input "tags" is required: text — genres and moods`}, + "an unknown input": {in(map[string]any{"tempo": 120}), `takes no input "tempo"`}, + "a number as text": {in(map[string]any{"tags": 5}), `input "tags" is text`}, + "too long": {in(map[string]any{"tags": strings.Repeat("x", 201)}), "past 200"}, + "out of bounds": {in(map[string]any{"strength": 2}), "outside 0 to 1"}, + "not whole": {in(map[string]any{"seconds": 10.5}), "whole number"}, + "not a choice": {in(map[string]any{"style": "jazz"}), "one of pop, metal"}, + "a song as a picture": {in(map[string]any{"cover": base64.StdEncoding.EncodeToString(testWAV())}), "PNG, JPEG"}, + "someone else's file": {in(map[string]any{"cover": "file-nothere"}), `no file "file-nothere"`}, + "a run of flux's": {map[string]any{"model": "flux", "inputs": map[string]any{"tags": "x"}}, "takes no inputs"}, + "a chat with ace": {map[string]any{"model": "ace", "messages": []any{}}, "makes audio: ask for it at " + workflowsPath}, + } { + path := workflowsPath + if name == "a chat with ace" { + path = "/v1/chat/completions" + } + code, _, raw := call(t, h, secret, http.MethodPost, path, c.body) + if code != http.StatusBadRequest || !strings.Contains(errorMessage([]byte(raw)), c.want) { + t.Errorf("%s: %d %s", name, code, raw) + } + } +} + +func TestARunTakesAFileUploadedBeforeAndNamedByItsID(t *testing.T) { + c := newRunComfy(t, testWAV(), "song.wav") + rt, secret, project := runRouter(t, c, songWorkflow, catalog.OutputAudio) + h := rt.Handler() + pic, _ := base64.StdEncoding.DecodeString(testPicture(t)) + now := time.Now() + if err := rt.store.PutFile(t.Context(), store.File{ID: "file-cover", Owner: "project:" + project, Purpose: "user_data", + Filename: "cover.png", Bytes: int64(len(pic)), CreatedAt: now, ExpiresAt: now.Add(time.Hour)}, pic); err != nil { + t.Fatal(err) + } + if err := rt.store.PutFile(t.Context(), store.File{ID: "file-theirs", Owner: "project:other", Purpose: "user_data", + Filename: "x.png", Bytes: int64(len(pic)), CreatedAt: now, ExpiresAt: now.Add(time.Hour)}, pic); err != nil { + t.Fatal(err) + } + code, _, raw := call(t, h, secret, http.MethodPost, workflowsPath, + map[string]any{"model": "ace", "inputs": map[string]any{"tags": "x", "cover": "file-theirs"}}) + if code != http.StatusBadRequest || !strings.Contains(errorMessage([]byte(raw)), `no file "file-theirs"`) { + t.Errorf("another project's file: %d %s", code, raw) + } + code, j, raw := call(t, h, secret, http.MethodPost, workflowsPath, + map[string]any{"model": "ace", "inputs": map[string]any{"tags": "x", "cover": "file-cover"}}) + if code != http.StatusAccepted { + t.Fatalf("%d %s", code, raw) + } + if got := waitJob(t, h, secret, j["id"].(string), jobDone); got["status"] != "completed" { + t.Fatalf("job %v", got) + } + c.mu.Lock() + defer c.mu.Unlock() + if len(c.uploaded) != 1 || !strings.HasSuffix(c.uploaded[0], ".png:.png") { + t.Errorf("uploaded %v", c.uploaded) + } +} + +func TestAWorkflowWantingWhatNoInputGivesFailsSayingSo(t *testing.T) { + c := newRunComfy(t, testWAV(), "song.wav") + wf := strings.Replace(songWorkflow, `"{{style}}"`, `"{{mood}}"`, 1) + rt, secret, _ := runRouter(t, c, wf, catalog.OutputAudio) + h := rt.Handler() + code, j, raw := call(t, h, secret, http.MethodPost, workflowsPath, + map[string]any{"model": "ace", "inputs": map[string]any{"tags": "x", "cover": testPicture(t)}}) + if code != http.StatusAccepted { + t.Fatalf("%d %s", code, raw) + } + got := waitJob(t, h, secret, j["id"].(string), jobDone) + if e, _ := got["error"].(map[string]any); got["status"] != "failed" || !strings.Contains(fmt.Sprint(e["message"]), "placeholders mood") { + t.Errorf("job %v", got) + } +} + +func TestARunThatSavedTheWrongKindFailsAndAPictureCarriesItsLicence(t *testing.T) { + c := newRunComfy(t, testWAV(), "song.wav") + rt, secret, _ := runRouter(t, c, songWorkflow, catalog.OutputVideo) + h := rt.Handler() + _, j, _ := call(t, h, secret, http.MethodPost, workflowsPath, + map[string]any{"model": "ace", "inputs": map[string]any{"tags": "x", "cover": testPicture(t)}}) + got := waitJob(t, h, secret, j["id"].(string), jobDone) + if e, _ := got["error"].(map[string]any); !strings.Contains(fmt.Sprint(e["message"]), "saved no video: it saved song.wav (audio/wav)") { + t.Errorf("job %v", got) + } + + var b bytes.Buffer + _ = png.Encode(&b, image.NewRGBA(image.Rect(0, 0, 4, 4))) + with, err := pngWithLicense(b.Bytes(), catalog.License{ID: "Apache-2.0", Name: "Apache 2.0", Attribution: "Made with ACE"}) + if err != nil { + t.Fatal(err) + } + if _, err := png.Decode(bytes.NewReader(with)); err != nil { + t.Fatalf("not a PNG any more: %v", err) + } + if !bytes.Contains(with, []byte("iTXtLicense\x00\x00\x00\x00\x00{\"id\":\"Apache-2.0\"")) { + t.Error("no licence chunk in the picture") + } + if _, err := pngWithLicense([]byte("GIF89a"), catalog.License{Name: "x"}); err == nil { + t.Error("a GIF taken for a PNG") + } +}