From 591b8964f57b7c8fb5b4dcaa6879ed71a159c50d Mon Sep 17 00:00:00 2001 From: malkuthro Date: Sun, 7 Jun 2026 22:30:36 +0200 Subject: [PATCH] feat(setups): add published setup execution API --- CHANGELOG.md | 7 + docs/maintainers/published-setups.md | 94 ++++++ koolook_routes.py | 65 +++- koolook_setup_runner.py | 317 ++++++++++++++++++++ scripts/sync_to_dev.py | 1 + tests/server/test_published_setup_api.py | 173 ++++++++++- tests/server/test_published_setup_runner.py | 172 +++++++++++ 7 files changed, 826 insertions(+), 3 deletions(-) create mode 100644 koolook_setup_runner.py create mode 100644 tests/server/test_published_setup_runner.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 6c7158c..94ee09c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,13 @@ The format is inspired by Keep a Changelog and SemVer. a bundled sample fixture, and read-only external catalog endpoints: `GET /koolook/api/setups` for summaries and `GET /koolook/api/setups/{id}` for detail contracts. +- **Published setup execution API.** Added the first external run path for + callable published setups: `POST /koolook/api/setups/{id}/run` clones the + stored API prompt, accepts only declared setup/app inputs, injects approved + values, submits the prompt to ComfyUI `/prompt`, and returns a stable Koolook + run id plus ComfyUI prompt id. `GET /koolook/api/runs/{runId}` translates + ComfyUI history/queue data into queued/running/succeeded/failed state and + exposes published output-contract items for external frontends. - **Sidebar publish flow for callable setups.** Saved sidebar workflows now expose an explicit **Publish setup...** context-menu action that captures setup metadata, source workflow reference, input/output contract JSON, and diff --git a/docs/maintainers/published-setups.md b/docs/maintainers/published-setups.md index d8863cf..5ec3035 100644 --- a/docs/maintainers/published-setups.md +++ b/docs/maintainers/published-setups.md @@ -124,6 +124,16 @@ External callers should use: - `PublishedSetupRegistry.getSetup(id)` - `PublishedSetupRegistry.publishSetup(visualGraph, metadata, inputContract, outputContract, source)` +Execution lives behind [`../../koolook_setup_runner.py`](../../koolook_setup_runner.py). +External routes and future app adapters should use: + +- `PublishedSetupRunner.runSetup(id, inputs)` +- `PublishedSetupRunner.getRun(runId)` + +The runner hides prompt cloning, input injection, ComfyUI `/prompt` +submission, run-id mapping, history/queue polling, and output flattening from +route callers. + Invalid setup objects are omitted from list/detail results. Diagnostics are kept on `registry.diagnostics` and logged by the HTTP adapter. File-level storage diagnostics, such as unreadable JSON, are reported through the same @@ -192,6 +202,90 @@ Success returns `{ "ok": true, "setup": { ... } }` with the generated `apiPrompt` stored alongside the original `visualGraph`. Validation failures return HTTP `400` with `{ "ok": false, "errors": [...] }`. +## Run API + +`POST /koolook/api/setups/{id}/run` + +Queues a callable published setup on the managed ComfyUI server. The body must +be a JSON object with an `inputs` object: + +```json +{ + "inputs": { + "prompt": "A cinematic close-up" + } +} +``` + +Only fields declared by `inputContract.inputs` or `setupSurface.app.inputs` +are accepted. `setupSurface.app.switch` is also accepted when present. The +runner deep-clones the stored `apiPrompt`, injects approved values into their +declared targets, and submits `{ "prompt": }` to ComfyUI +`/prompt`; the stored setup record is not mutated by a run request. + +Success returns: + +```json +{ + "ok": true, + "run": { + "runId": "run-000001", + "promptId": "f0c2...", + "status": "queued" + } +} +``` + +Failure responses use the standard route error shape: + +```json +{ + "ok": false, + "errors": ["input 'seed' is not declared by this setup"] +} +``` + +Missing setup ids return `404`; invalid inputs and non-callable draft setups +return `400`; ComfyUI queue failures return `502`. + +`GET /koolook/api/runs/{runId}` + +Looks up a stable Koolook run id and translates ComfyUI history/queue data into +external state: + +```json +{ + "ok": true, + "run": { + "runId": "run-000001", + "setupId": "ltx-director-demo", + "promptId": "f0c2...", + "status": "succeeded", + "outputs": [ + { + "key": "video", + "label": "Video", + "type": "video", + "items": [ + { + "nodeId": "20", + "kind": "videos", + "filename": "demo.mp4", + "subfolder": "koolook", + "type": "output" + } + ] + } + ] + } +} +``` + +Status is one of `queued`, `running`, `succeeded`, or `failed`. History entries +produce terminal state and output items; otherwise the runner checks ComfyUI +queue data to distinguish `running` from still `queued`. Unknown run ids return +`404`; ComfyUI status lookup failures return `502`. + ## Comfy-Native Setup Surface Issue #219 amends the #209 direction: curators should define a setup's app diff --git a/koolook_routes.py b/koolook_routes.py index d681ba2..3d205fe 100644 --- a/koolook_routes.py +++ b/koolook_routes.py @@ -36,8 +36,20 @@ from aiohttp import web try: from .koolook_setups import PublishedSetupRegistry, default_registry + from .koolook_setup_runner import ( + AiohttpComfyClient, + InMemorySetupRunStore, + PublishedSetupRunner, + SetupRunError, + ) except ImportError: # pragma: no cover - standalone test/import context from koolook_setups import PublishedSetupRegistry, default_registry + from koolook_setup_runner import ( + AiohttpComfyClient, + InMemorySetupRunStore, + PublishedSetupRunner, + SetupRunError, + ) ENV_VAR = "KFORGELABS_PRESETS" DEFAULT_SUBDIR = "koolook-presets" @@ -384,7 +396,7 @@ def _resolve_target(lib_base: Path, subdir: str, name: str) -> tuple[Path, Path] return target_dir, file_path -def register_routes(routes, setup_registry_factory=None) -> None: +def register_routes(routes, setup_registry_factory=None, setup_runner_factory=None) -> None: """Attach the preset endpoints to the given aiohttp ``RouteTableDef``. Called once from ``__init__.py`` at custom-node load time. Splitting @@ -393,6 +405,16 @@ def register_routes(routes, setup_registry_factory=None) -> None: """ if setup_registry_factory is None: setup_registry_factory = default_registry + run_store = InMemorySetupRunStore() + + def _default_setup_runner(request, registry: PublishedSetupRegistry) -> PublishedSetupRunner: + base_url = f"{request.scheme}://{request.host}" + return PublishedSetupRunner(registry, AiohttpComfyClient(base_url), run_store) + + def _setup_runner(request, registry: PublishedSetupRegistry): + if setup_runner_factory is not None: + return setup_runner_factory() + return _default_setup_runner(request, registry) def _log_setup_diagnostics(registry: PublishedSetupRegistry) -> None: for diagnostic in registry.diagnostics: @@ -445,6 +467,47 @@ def register_routes(routes, setup_registry_factory=None) -> None: ) return web.json_response({"ok": True, "setup": result.setup}) + @routes.post("/koolook/api/setups/{setup_id}/run") + async def run_published_setup(request): + try: + payload = await request.json() + except Exception as exc: + return web.json_response( + {"ok": False, "errors": [f"Body must be JSON: {exc}"]}, + status=400, + ) + if not isinstance(payload, dict): + return web.json_response( + {"ok": False, "errors": ["Body must be a JSON object."]}, + status=400, + ) + registry: PublishedSetupRegistry = setup_registry_factory() + runner = _setup_runner(request, registry) + try: + result = await runner.runSetup( + request.match_info["setup_id"], + payload.get("inputs", {}), + ) + except SetupRunError as exc: + return web.json_response( + {"ok": False, "errors": exc.errors}, + status=exc.status_code, + ) + return web.json_response({"ok": True, "run": result}) + + @routes.get("/koolook/api/runs/{run_id}") + async def get_published_setup_run(request): + registry: PublishedSetupRegistry = setup_registry_factory() + runner = _setup_runner(request, registry) + try: + result = await runner.getRun(request.match_info["run_id"]) + except SetupRunError as exc: + return web.json_response( + {"ok": False, "errors": exc.errors}, + status=exc.status_code, + ) + return web.json_response({"ok": True, "run": result}) + @routes.get("/koolook/presets/info") async def info(_request): base, source = _configured_dir() diff --git a/koolook_setup_runner.py b/koolook_setup_runner.py new file mode 100644 index 0000000..5a99397 --- /dev/null +++ b/koolook_setup_runner.py @@ -0,0 +1,317 @@ +"""Execution boundary for published Koolook workflow setups.""" +from __future__ import annotations + +from copy import deepcopy +from dataclasses import dataclass +from typing import Any + +try: + from .koolook_setups import PublishedSetupRegistry +except ImportError: # pragma: no cover - standalone test/import context + from koolook_setups import PublishedSetupRegistry + + +class SetupRunError(Exception): + """API-facing setup runner error with stable HTTP status semantics.""" + + def __init__(self, status_code: int, errors: list[str]) -> None: + super().__init__("; ".join(errors)) + self.status_code = status_code + self.errors = errors + + +@dataclass(frozen=True) +class RunRecord: + run_id: str + setup_id: str + prompt_id: str + status: str + + +class InMemorySetupRunStore: + """Small run-id mapping store for a running ComfyUI process.""" + + def __init__(self) -> None: + self._next_id = 1 + self._records: dict[str, RunRecord] = {} + + def create(self, *, setup_id: str, prompt_id: str) -> RunRecord: + run_id = f"run-{self._next_id:06d}" + self._next_id += 1 + record = RunRecord( + run_id=run_id, + setup_id=setup_id, + prompt_id=prompt_id, + status="queued", + ) + self._records[run_id] = record + return record + + def get(self, run_id: str) -> RunRecord | None: + return self._records.get(run_id) + + +class AiohttpComfyClient: + """HTTP adapter for the managed ComfyUI server.""" + + def __init__(self, base_url: str) -> None: + self._base_url = base_url.rstrip("/") + + async def queue_prompt(self, prompt: dict) -> dict: + import aiohttp + + async with aiohttp.ClientSession() as session: + async with session.post(f"{self._base_url}/prompt", json={"prompt": prompt}) as response: + try: + payload = await response.json() + except Exception: + payload = {"error": await response.text()} + if response.status >= 400: + detail = payload.get("error") if isinstance(payload, dict) else payload + raise SetupRunError( + 502, + [f"ComfyUI queue request failed with HTTP {response.status}: {detail}"], + ) + return payload if isinstance(payload, dict) else {} + + async def get_history(self, prompt_id: str) -> dict: + import aiohttp + + async with aiohttp.ClientSession() as session: + async with session.get(f"{self._base_url}/history/{prompt_id}") as response: + try: + payload = await response.json() + except Exception: + payload = {"error": await response.text()} + if response.status >= 400: + detail = payload.get("error") if isinstance(payload, dict) else payload + raise SetupRunError( + 502, + [f"ComfyUI history request failed with HTTP {response.status}: {detail}"], + ) + return payload if isinstance(payload, dict) else {} + + async def get_queue(self) -> dict: + import aiohttp + + async with aiohttp.ClientSession() as session: + async with session.get(f"{self._base_url}/queue") as response: + try: + payload = await response.json() + except Exception: + payload = {"error": await response.text()} + if response.status >= 400: + detail = payload.get("error") if isinstance(payload, dict) else payload + raise SetupRunError( + 502, + [f"ComfyUI queue status request failed with HTTP {response.status}: {detail}"], + ) + return payload if isinstance(payload, dict) else {} + + +class PublishedSetupRunner: + """Runs callable published setups without exposing ComfyUI internals.""" + + def __init__( + self, + registry: PublishedSetupRegistry, + comfy_client: Any, + run_store: InMemorySetupRunStore | None = None, + ) -> None: + self._registry = registry + self._comfy_client = comfy_client + self._run_store = run_store or InMemorySetupRunStore() + + async def runSetup(self, setup_id: str, inputs: dict[str, Any]) -> dict[str, Any]: + setup = self._registry.getSetup(setup_id) + if setup is None: + raise SetupRunError(404, [f"Published setup '{setup_id}' not found."]) + if not isinstance(inputs, dict): + raise SetupRunError(400, ["inputs must be a JSON object"]) + + api_prompt = setup.get("apiPrompt") + if not isinstance(api_prompt, dict) or setup.get("validation", {}).get("status") != "valid": + raise SetupRunError(400, [f"Published setup '{setup_id}' is not callable."]) + + prompt = deepcopy(api_prompt) + input_fields = _declared_input_fields(setup) + errors = _validate_run_inputs(input_fields, inputs) + if errors: + raise SetupRunError(400, errors) + + for key, value in inputs.items(): + field = input_fields[key] + target = field["target"] + prompt[str(target["node"])]["inputs"][str(target["input"])] = value + + try: + queued = await self._comfy_client.queue_prompt(prompt) + except SetupRunError: + raise + except Exception as exc: + raise SetupRunError(502, [f"ComfyUI queue request failed: {exc}"]) from exc + + prompt_id = queued.get("prompt_id") if isinstance(queued, dict) else None + if not isinstance(prompt_id, str) or not prompt_id: + raise SetupRunError(502, ["ComfyUI queue response did not include prompt_id"]) + + record = self._run_store.create(setup_id=setup_id, prompt_id=prompt_id) + return {"runId": record.run_id, "promptId": record.prompt_id, "status": record.status} + + async def getRun(self, run_id: str) -> dict[str, Any]: + record = self._run_store.get(run_id) + if record is None: + raise SetupRunError(404, [f"Koolook run '{run_id}' not found."]) + + setup = self._registry.getSetup(record.setup_id) + if setup is None: + raise SetupRunError(404, [f"Published setup '{record.setup_id}' not found."]) + + try: + history = await self._comfy_client.get_history(record.prompt_id) + except SetupRunError: + raise + except Exception as exc: + raise SetupRunError(502, [f"ComfyUI history request failed: {exc}"]) from exc + + history_entry = history.get(record.prompt_id) if isinstance(history, dict) else None + if isinstance(history_entry, dict): + status = _status_from_history(history_entry) + return { + "runId": record.run_id, + "setupId": record.setup_id, + "promptId": record.prompt_id, + "status": status, + "outputs": _summarize_outputs(setup, history_entry.get("outputs", {})), + } + + try: + queue = await self._comfy_client.get_queue() + except SetupRunError: + raise + except Exception as exc: + raise SetupRunError(502, [f"ComfyUI queue status request failed: {exc}"]) from exc + return { + "runId": record.run_id, + "setupId": record.setup_id, + "promptId": record.prompt_id, + "status": _status_from_queue(record.prompt_id, queue), + "outputs": [], + } + + +def _declared_input_fields(setup: dict[str, Any]) -> dict[str, dict[str, Any]]: + fields: dict[str, dict[str, Any]] = {} + for field in setup.get("inputContract", {}).get("inputs", []): + if not isinstance(field, dict): + continue + key = field.get("key") + target = field.get("target") + if isinstance(key, str) and key and isinstance(target, dict): + fields[key] = field + app = setup.get("setupSurface", {}).get("app", {}) + if isinstance(app, dict): + for field in app.get("inputs", []): + if not isinstance(field, dict): + continue + key = field.get("key") + target = field.get("target") + if isinstance(key, str) and key and isinstance(target, dict): + fields.setdefault(key, field) + switch = app.get("switch") + if isinstance(switch, dict): + key = switch.get("key") + target = switch.get("target") + if isinstance(key, str) and key and isinstance(target, dict): + fields.setdefault(key, switch) + return fields + + +def _validate_run_inputs(input_fields: dict[str, dict[str, Any]], inputs: dict[str, Any]) -> list[str]: + errors: list[str] = [] + for key in inputs: + if key not in input_fields: + errors.append(f"input '{key}' is not declared by this setup") + for key, field in input_fields.items(): + if field.get("required") is True and key not in inputs: + errors.append(f"required input '{key}' is missing") + return errors + + +def _status_from_history(history_entry: dict[str, Any]) -> str: + status = history_entry.get("status") + if not isinstance(status, dict): + return "succeeded" + status_text = str(status.get("status_str", "")).lower() + if status.get("completed") is True and status_text not in {"error", "failed", "failure"}: + return "succeeded" + return "failed" + + +def _status_from_queue(prompt_id: str, queue: Any) -> str: + if not isinstance(queue, dict): + return "queued" + if _queue_contains_prompt(queue.get("queue_running", []), prompt_id): + return "running" + if _queue_contains_prompt(queue.get("queue_pending", []), prompt_id): + return "queued" + return "queued" + + +def _queue_contains_prompt(entries: Any, prompt_id: str) -> bool: + if not isinstance(entries, list): + return False + for entry in entries: + if isinstance(entry, dict) and entry.get("prompt_id") == prompt_id: + return True + if isinstance(entry, list) and prompt_id in entry: + return True + return False + + +def _summarize_outputs(setup: dict[str, Any], raw_outputs: Any) -> list[dict[str, Any]]: + output_items = _flatten_history_outputs(raw_outputs) + summaries = [] + for output in setup.get("outputContract", {}).get("outputs", []): + if not isinstance(output, dict): + continue + summaries.append( + { + "key": output.get("key", ""), + "label": output.get("label", output.get("key", "")), + "type": output.get("type", ""), + "items": [ + item + for item in output_items + if _history_output_matches_contract(item, str(output.get("type", ""))) + ], + } + ) + return summaries + + +def _flatten_history_outputs(raw_outputs: Any) -> list[dict[str, Any]]: + if not isinstance(raw_outputs, dict): + return [] + items: list[dict[str, Any]] = [] + for node_id, node_outputs in raw_outputs.items(): + if not isinstance(node_outputs, dict): + continue + for kind, values in node_outputs.items(): + if not isinstance(values, list): + continue + for value in values: + if isinstance(value, dict): + item = {"nodeId": str(node_id), "kind": str(kind)} + item.update(value) + items.append(item) + return items + + +def _history_output_matches_contract(item: dict[str, Any], contract_type: str) -> bool: + kind = str(item.get("kind", "")).lower() + if contract_type == "video": + return kind in {"video", "videos", "animated"} + if contract_type == "image": + return kind in {"image", "images"} + return True diff --git a/scripts/sync_to_dev.py b/scripts/sync_to_dev.py index e36790b..595da3e 100644 --- a/scripts/sync_to_dev.py +++ b/scripts/sync_to_dev.py @@ -165,6 +165,7 @@ RUNTIME_PATHS: tuple[str, ...] = ( "k_video_load.py", "koolook_install_guard.py", "koolook_routes.py", + "koolook_setup_runner.py", "koolook_setups.py", "koolook_versioning.py", "forks", diff --git a/tests/server/test_published_setup_api.py b/tests/server/test_published_setup_api.py index 88f04dc..5decd89 100644 --- a/tests/server/test_published_setup_api.py +++ b/tests/server/test_published_setup_api.py @@ -8,6 +8,7 @@ from aiohttp import web from aiohttp.test_utils import make_mocked_request import koolook_routes +from koolook_setup_runner import PublishedSetupRunner from koolook_setups import PublishedSetupRegistry, StaticSetupStorage from tests.server.test_published_setup_registry import _valid_setup @@ -171,9 +172,177 @@ def test_publish_route_rejects_missing_metadata_with_clear_error() -> None: asyncio.run(exercise()) -def _app_with_registry(registry: PublishedSetupRegistry) -> web.Application: +def test_run_route_queues_callable_setup_and_returns_stable_run_id() -> None: + async def exercise() -> None: + registry = PublishedSetupRegistry(StaticSetupStorage([_valid_setup()])) + comfy = _FakeComfyClient() + app = _app_with_registry( + registry, + setup_runner_factory=lambda: PublishedSetupRunner(registry, comfy), + ) + + response = await _handle_json( + app, + "POST", + "/koolook/api/setups/ltx-director-demo/run", + {"inputs": {"prompt": "external prompt"}}, + ) + + assert response.status == 200 + assert _json_body(response) == { + "ok": True, + "run": { + "runId": "run-000001", + "promptId": "comfy-prompt-1", + "status": "queued", + }, + } + assert comfy.submitted_prompts == [ + {"12": {"class_type": "Text Multiline", "inputs": {"text": "external prompt"}}} + ] + + asyncio.run(exercise()) + + +def test_run_status_route_reports_succeeded_outputs_from_comfy_history() -> None: + async def exercise() -> None: + registry = PublishedSetupRegistry(StaticSetupStorage([_valid_setup()])) + comfy = _FakeComfyClient( + history={ + "comfy-prompt-1": { + "status": {"completed": True, "status_str": "success"}, + "outputs": { + "20": { + "videos": [ + { + "filename": "demo.mp4", + "subfolder": "koolook", + "type": "output", + } + ] + } + }, + } + } + ) + runner = PublishedSetupRunner(registry, comfy) + app = _app_with_registry(registry, setup_runner_factory=lambda: runner) + + run_response = await _handle_json( + app, + "POST", + "/koolook/api/setups/ltx-director-demo/run", + {"inputs": {"prompt": "external prompt"}}, + ) + run_id = _json_body(run_response)["run"]["runId"] + status_response = await _handle(app, "GET", f"/koolook/api/runs/{run_id}") + + assert status_response.status == 200 + assert _json_body(status_response) == { + "ok": True, + "run": { + "runId": "run-000001", + "setupId": "ltx-director-demo", + "promptId": "comfy-prompt-1", + "status": "succeeded", + "outputs": [ + { + "key": "video", + "label": "Video", + "type": "video", + "items": [ + { + "nodeId": "20", + "kind": "videos", + "filename": "demo.mp4", + "subfolder": "koolook", + "type": "output", + } + ], + } + ], + }, + } + + asyncio.run(exercise()) + + +def test_run_route_returns_clear_errors_for_invalid_inputs() -> None: + async def exercise() -> None: + registry = PublishedSetupRegistry(StaticSetupStorage([_valid_setup()])) + app = _app_with_registry( + registry, + setup_runner_factory=lambda: PublishedSetupRunner(registry, _FakeComfyClient()), + ) + + response = await _handle_json( + app, + "POST", + "/koolook/api/setups/ltx-director-demo/run", + {"inputs": {"prompt": "ok", "seed": 123}}, + ) + + assert response.status == 400 + assert _json_body(response) == { + "ok": False, + "errors": ["input 'seed' is not declared by this setup"], + } + + asyncio.run(exercise()) + + +def test_run_route_returns_clear_error_for_missing_setup() -> None: + async def exercise() -> None: + registry = PublishedSetupRegistry(StaticSetupStorage([_valid_setup()])) + app = _app_with_registry( + registry, + setup_runner_factory=lambda: PublishedSetupRunner(registry, _FakeComfyClient()), + ) + + response = await _handle_json( + app, + "POST", + "/koolook/api/setups/missing/run", + {"inputs": {"prompt": "ok"}}, + ) + + assert response.status == 404 + assert _json_body(response) == { + "ok": False, + "errors": ["Published setup 'missing' not found."], + } + + asyncio.run(exercise()) + + +class _FakeComfyClient: + def __init__(self, *, history=None, queue=None) -> None: + self.submitted_prompts: list[dict] = [] + self.history = history or {} + self.queue = queue or {"queue_running": [], "queue_pending": []} + + async def queue_prompt(self, prompt: dict) -> dict: + self.submitted_prompts.append(prompt) + return {"prompt_id": "comfy-prompt-1"} + + async def get_history(self, prompt_id: str) -> dict: + return {prompt_id: self.history[prompt_id]} if prompt_id in self.history else {} + + async def get_queue(self) -> dict: + return self.queue + + +def _app_with_registry( + registry: PublishedSetupRegistry, + *, + setup_runner_factory=None, +) -> web.Application: routes = web.RouteTableDef() - koolook_routes.register_routes(routes, setup_registry_factory=lambda: registry) + koolook_routes.register_routes( + routes, + setup_registry_factory=lambda: registry, + setup_runner_factory=setup_runner_factory, + ) app = web.Application() app.add_routes(routes) return app diff --git a/tests/server/test_published_setup_runner.py b/tests/server/test_published_setup_runner.py new file mode 100644 index 0000000..b4d2d72 --- /dev/null +++ b/tests/server/test_published_setup_runner.py @@ -0,0 +1,172 @@ +from __future__ import annotations + +import asyncio +from copy import deepcopy + +import pytest + +from koolook_setup_runner import PublishedSetupRunner, SetupRunError +from koolook_setups import PublishedSetupRegistry, StaticSetupStorage +from tests.server.test_published_setup_registry import _valid_setup + + +class FakeComfyClient: + def __init__(self, prompt_response=None, *, history=None, queue=None, queue_error=None) -> None: + self.prompt_response = prompt_response or {"prompt_id": "comfy-prompt-1"} + self.submitted_prompts: list[dict] = [] + self.history = history or {} + self.queue = queue or {"queue_running": [], "queue_pending": []} + self.queue_error = queue_error + + async def queue_prompt(self, prompt: dict) -> dict: + if self.queue_error is not None: + raise self.queue_error + self.submitted_prompts.append(prompt) + return self.prompt_response + + async def get_history(self, prompt_id: str) -> dict: + return {prompt_id: self.history[prompt_id]} if prompt_id in self.history else {} + + async def get_queue(self) -> dict: + return self.queue + + +def test_run_setup_injects_declared_inputs_into_cloned_prompt() -> None: + async def exercise() -> None: + setup = _valid_setup() + original_prompt = deepcopy(setup["apiPrompt"]) + registry = PublishedSetupRegistry(StaticSetupStorage([setup])) + comfy = FakeComfyClient() + runner = PublishedSetupRunner(registry, comfy) + + result = await runner.runSetup("ltx-director-demo", {"prompt": "external prompt"}) + + assert result == { + "runId": "run-000001", + "promptId": "comfy-prompt-1", + "status": "queued", + } + assert comfy.submitted_prompts == [ + {"12": {"class_type": "Text Multiline", "inputs": {"text": "external prompt"}}} + ] + assert setup["apiPrompt"] == original_prompt + + asyncio.run(exercise()) + + +def test_run_setup_rejects_inputs_not_declared_by_contract() -> None: + async def exercise() -> None: + registry = PublishedSetupRegistry(StaticSetupStorage([_valid_setup()])) + runner = PublishedSetupRunner(registry, FakeComfyClient()) + + with pytest.raises(SetupRunError) as exc_info: + await runner.runSetup("ltx-director-demo", {"prompt": "ok", "seed": 123}) + + assert exc_info.value.status_code == 400 + assert exc_info.value.errors == ["input 'seed' is not declared by this setup"] + + asyncio.run(exercise()) + + +def test_run_setup_rejects_non_callable_setup() -> None: + async def exercise() -> None: + setup = _valid_setup() + setup["apiPrompt"] = None + setup["validation"] = {"status": "draft", "diagnostics": ["API prompt conversion pending."]} + registry = PublishedSetupRegistry(StaticSetupStorage([setup])) + runner = PublishedSetupRunner(registry, FakeComfyClient()) + + with pytest.raises(SetupRunError) as exc_info: + await runner.runSetup("ltx-director-demo", {"prompt": "external prompt"}) + + assert exc_info.value.status_code == 400 + assert exc_info.value.errors == ["Published setup 'ltx-director-demo' is not callable."] + + asyncio.run(exercise()) + + +def test_run_setup_translates_comfy_queue_failures() -> None: + async def exercise() -> None: + registry = PublishedSetupRegistry(StaticSetupStorage([_valid_setup()])) + runner = PublishedSetupRunner(registry, FakeComfyClient(queue_error=RuntimeError("server down"))) + + with pytest.raises(SetupRunError) as exc_info: + await runner.runSetup("ltx-director-demo", {"prompt": "external prompt"}) + + assert exc_info.value.status_code == 502 + assert exc_info.value.errors == ["ComfyUI queue request failed: server down"] + + asyncio.run(exercise()) + + +def test_get_run_reports_running_state_from_comfy_queue() -> None: + async def exercise() -> None: + registry = PublishedSetupRegistry(StaticSetupStorage([_valid_setup()])) + runner = PublishedSetupRunner( + registry, + FakeComfyClient(queue={"queue_running": [[7, "comfy-prompt-1", {}]], "queue_pending": []}), + ) + + queued = await runner.runSetup("ltx-director-demo", {"prompt": "external prompt"}) + status = await runner.getRun(queued["runId"]) + + assert status == { + "runId": "run-000001", + "setupId": "ltx-director-demo", + "promptId": "comfy-prompt-1", + "status": "running", + "outputs": [], + } + + asyncio.run(exercise()) + + +def test_run_setup_accepts_inputs_declared_by_app_surface_contract() -> None: + async def exercise() -> None: + setup = _valid_setup() + setup["inputContract"] = {"inputs": []} + setup["outputContract"] = {"outputs": []} + setup["visualGraph"]["groups"] = [ + {"title": "Koolook Input", "bounding": [0, 0, 300, 200]}, + {"title": "Koolook Output", "bounding": [400, 0, 300, 200]}, + ] + setup["visualGraph"]["nodes"][0]["pos"] = [20, 20] + setup["setupSurface"] = { + "sourceInputs": [ + { + "group": "Koolook Input", + "nodes": [{"id": "12", "type": "Text Multiline", "title": "Prompt"}], + } + ], + "outputs": [ + { + "group": "Koolook Output", + "nodes": [{"id": "20", "type": "Preview Image", "title": "Preview"}], + } + ], + "controls": [], + "app": { + "inputs": [ + { + "key": "prompt", + "label": "Prompt", + "visible": True, + "target": {"node": "12", "input": "text"}, + "default": "demo prompt", + } + ], + "outputs": [], + "results": [], + }, + } + registry = PublishedSetupRegistry(StaticSetupStorage([setup])) + comfy = FakeComfyClient() + runner = PublishedSetupRunner(registry, comfy) + + await runner.runSetup("ltx-director-demo", {"prompt": "app contract prompt"}) + + assert comfy.submitted_prompts == [ + {"12": {"class_type": "Text Multiline", "inputs": {"text": "app contract prompt"}}} + ] + + asyncio.run(exercise())