From 5fbfd09caf5fb09ca72fab20ec24560d687d2f96 Mon Sep 17 00:00:00 2001 From: Gateway <519563+gateway@users.noreply.github.com> Date: Sat, 23 May 2026 20:28:52 +0700 Subject: [PATCH 1/2] Harden dynamic media preset graph node --- apps/api/app/graph/executors/kie_model.py | 108 +++---- apps/api/app/graph/executors/preset_ops.py | 140 ++++----- apps/api/app/graph/normalization.py | 32 +- apps/api/app/graph/preset_catalog.py | 275 ++++++++++++++++++ apps/api/app/graph/pricing.py | 2 + apps/api/app/graph/registry.py | 91 +----- apps/api/app/graph/routes.py | 31 ++ apps/api/app/graph/runtime.py | 2 - apps/api/app/graph/schemas.py | 19 ++ apps/api/app/graph/system_nodes_preset.py | 75 +++-- apps/api/app/graph/validator.py | 77 +++-- apps/api/app/store.py | 2 + apps/api/app/store_graph.py | 37 +++ apps/api/tests/test_graph_studio.py | 159 ++++++++-- .../graph-studio/graph-library-dialogs.tsx | 6 +- .../graph-studio/graph-node-field.tsx | 19 +- .../graph-node-media-preview.test.tsx | 21 ++ .../graph-studio/graph-node-media-preview.tsx | 4 +- .../components/graph-studio/graph-node.tsx | 20 +- .../graph-studio/graph-run-history-panel.tsx | 17 +- .../graph-studio/graph-studio-dialogs.tsx | 6 +- .../components/graph-studio/graph-studio.tsx | 36 ++- .../hooks/use-graph-run-history.ts | 6 +- apps/web/components/graph-studio/types.ts | 8 + .../utils/graph-media-preset.test.ts | 108 +++++++ .../graph-studio/utils/graph-media-preset.ts | 127 ++++++++ .../graph-studio/utils/graph-media-preview.ts | 4 +- .../graph-studio/utils/graph-tabs.ts | 29 +- apps/web/lib/graph-media-preview.test.ts | 14 + apps/web/lib/graph-tabs.test.ts | 19 ++ 30 files changed, 1152 insertions(+), 342 deletions(-) create mode 100644 apps/api/app/graph/preset_catalog.py create mode 100644 apps/web/components/graph-studio/utils/graph-media-preset.test.ts create mode 100644 apps/web/components/graph-studio/utils/graph-media-preset.ts diff --git a/apps/api/app/graph/executors/kie_model.py b/apps/api/app/graph/executors/kie_model.py index e957dfd..cea68b0 100644 --- a/apps/api/app/graph/executors/kie_model.py +++ b/apps/api/app/graph/executors/kie_model.py @@ -202,6 +202,64 @@ def _to_media_ref(value: GraphOutputRef, *, role: Optional[str] = None) -> Media return MediaRefInput(asset_id=value.asset_id, reference_id=value.reference_id, role=role) +def submit_and_wait_for_kie_request( + *, + node: GraphWorkflowNode, + context: GraphExecutionContext, + request: ValidateRequest, + model_key: str, +) -> Dict[str, List[GraphOutputRef]]: + emit(context.run_id, "kie.validating", {"model_key": model_key}, node_id=node.id) + validation_started = time.perf_counter() + service.build_validation_bundle(request) + context.record_node_metric(node, "kie_validation_duration_seconds", round(time.perf_counter() - validation_started, 4)) + submit_started = time.perf_counter() + batch, jobs = service.submit_jobs(request) + context.record_node_metric(node, "kie_submit_duration_seconds", round(time.perf_counter() - submit_started, 4)) + job = jobs[0] + context.record_node_metric(node, "batch_id", batch["batch_id"]) + context.record_node_metric(node, "job_id", job["job_id"]) + emit(context.run_id, "kie.submitted", {"model_key": model_key, "job_id": job["job_id"], "batch_id": batch["batch_id"]}, node_id=node.id) + emit(context.run_id, "kie.polling", {"job_id": job["job_id"], "batch_id": batch["batch_id"]}, node_id=node.id) + + from ...runner import runner + + deadline = time.time() + 3600 + polling_started = time.perf_counter() + current = job + poll_count = 0 + sleep_seconds = 0.5 + while time.time() < deadline: + if context.is_cancel_requested(): + cancel_batch_jobs(batch["batch_id"]) + raise GraphRunCancelled(GRAPH_RUN_CANCELLED_MESSAGE) + current = store.get_job(job["job_id"]) or current + if current["status"] in {"completed", "failed", "cancelled"}: + break + runner.tick() + poll_count += 1 + elapsed = time.perf_counter() - polling_started + sleep_seconds = _adaptive_graph_kie_poll_interval(elapsed) + sleep_deadline = time.perf_counter() + sleep_seconds + while time.perf_counter() < sleep_deadline: + if context.is_cancel_requested(): + cancel_batch_jobs(batch["batch_id"]) + raise GraphRunCancelled(GRAPH_RUN_CANCELLED_MESSAGE) + time.sleep(min(0.25, max(0.0, sleep_deadline - time.perf_counter()))) + context.record_node_metric(node, "kie_polling_duration_seconds", round(time.perf_counter() - polling_started, 4)) + context.record_node_metric(node, "kie_poll_count", poll_count) + context.record_node_metric(node, "kie_poll_interval_seconds", sleep_seconds) + current = store.get_job(job["job_id"]) or current + if current["status"] == "cancelled" and context.is_cancel_requested(): + raise GraphRunCancelled(GRAPH_RUN_CANCELLED_MESSAGE) + if current["status"] != "completed": + raise ValueError(current.get("error") or f"KIE job did not complete: {current['status']}") + assets = store.get_assets_by_job_id(current["job_id"]) + if not assets: + raise ValueError("KIE job completed without creating an asset.") + return completed_kie_job_outputs(node=node, job=current, assets=assets, batch_id=batch["batch_id"]) + + class KieModelExecutor(GraphExecutor): node_type = "model.kie" @@ -272,55 +330,7 @@ def execute(self, node: GraphWorkflowNode, context: GraphExecutionContext) -> Di options=options, output_count=1, ) - emit(context.run_id, "kie.validating", {"model_key": model_key}, node_id=node.id) - validation_started = time.perf_counter() - service.build_validation_bundle(request) - context.record_node_metric(node, "kie_validation_duration_seconds", round(time.perf_counter() - validation_started, 4)) - submit_started = time.perf_counter() - batch, jobs = service.submit_jobs(request) - context.record_node_metric(node, "kie_submit_duration_seconds", round(time.perf_counter() - submit_started, 4)) - job = jobs[0] - context.record_node_metric(node, "batch_id", batch["batch_id"]) - context.record_node_metric(node, "job_id", job["job_id"]) - emit(context.run_id, "kie.submitted", {"model_key": model_key, "job_id": job["job_id"], "batch_id": batch["batch_id"]}, node_id=node.id) - emit(context.run_id, "kie.polling", {"job_id": job["job_id"], "batch_id": batch["batch_id"]}, node_id=node.id) - - from ...runner import runner - - deadline = time.time() + 3600 - polling_started = time.perf_counter() - current = job - poll_count = 0 - sleep_seconds = 0.5 - while time.time() < deadline: - if context.is_cancel_requested(): - cancel_batch_jobs(batch["batch_id"]) - raise GraphRunCancelled(GRAPH_RUN_CANCELLED_MESSAGE) - current = store.get_job(job["job_id"]) or current - if current["status"] in {"completed", "failed", "cancelled"}: - break - runner.tick() - poll_count += 1 - elapsed = time.perf_counter() - polling_started - sleep_seconds = _adaptive_graph_kie_poll_interval(elapsed) - sleep_deadline = time.perf_counter() + sleep_seconds - while time.perf_counter() < sleep_deadline: - if context.is_cancel_requested(): - cancel_batch_jobs(batch["batch_id"]) - raise GraphRunCancelled(GRAPH_RUN_CANCELLED_MESSAGE) - time.sleep(min(0.25, max(0.0, sleep_deadline - time.perf_counter()))) - context.record_node_metric(node, "kie_polling_duration_seconds", round(time.perf_counter() - polling_started, 4)) - context.record_node_metric(node, "kie_poll_count", poll_count) - context.record_node_metric(node, "kie_poll_interval_seconds", sleep_seconds) - current = store.get_job(job["job_id"]) or current - if current["status"] == "cancelled" and context.is_cancel_requested(): - raise GraphRunCancelled(GRAPH_RUN_CANCELLED_MESSAGE) - if current["status"] != "completed": - raise ValueError(current.get("error") or f"KIE job did not complete: {current['status']}") - assets = store.get_assets_by_job_id(current["job_id"]) - if not assets: - raise ValueError("KIE job completed without creating an asset.") - return completed_kie_job_outputs(node=node, job=current, assets=assets, batch_id=batch["batch_id"]) + return submit_and_wait_for_kie_request(node=node, context=context, request=request, model_key=model_key) def _select_task_mode( diff --git a/apps/api/app/graph/executors/preset_ops.py b/apps/api/app/graph/executors/preset_ops.py index a39a9fe..ae3aae3 100644 --- a/apps/api/app/graph/executors/preset_ops.py +++ b/apps/api/app/graph/executors/preset_ops.py @@ -3,10 +3,11 @@ import json from typing import Any, Dict, List -from ... import service, store -from ...schemas import MediaRefInput +from ... import kie_adapter, store +from ...schemas import MediaRefInput, ValidateRequest from ..schemas import GraphOutputRef, GraphWorkflowNode from .base import GraphExecutionContext, GraphExecutor +from .kie_model import _select_task_mode, submit_and_wait_for_kie_request def _dict_field(value: Any) -> Dict[str, Any]: @@ -19,8 +20,12 @@ def _dict_field(value: Any) -> Dict[str, Any]: return {} +def _graph_ref_to_media_input(ref: GraphOutputRef) -> MediaRefInput: + return MediaRefInput(asset_id=ref.asset_id, reference_id=ref.reference_id) + + def _graph_ref_to_media_ref(ref: GraphOutputRef) -> Dict[str, Any]: - return MediaRefInput(asset_id=ref.asset_id, reference_id=ref.reference_id).model_dump(exclude_none=True) + return _graph_ref_to_media_input(ref).model_dump(exclude_none=True) class PresetRenderExecutor(GraphExecutor): @@ -28,107 +33,78 @@ class PresetRenderExecutor(GraphExecutor): def execute(self, node: GraphWorkflowNode, context: GraphExecutionContext) -> Dict[str, List[GraphOutputRef]]: preset_id = str(node.fields.get("preset_id") or "").strip() - if not preset_id and node.type.startswith("preset.render."): - from ..registry import registry - - preset_id = str(registry.get_definition(node.type).source.get("preset_id") or "").strip() if not preset_id: - raise ValueError("Preset Render requires a preset.") + raise ValueError("Media Preset requires a preset.") preset = store.get_preset(preset_id) if not preset: - raise ValueError("Preset Render preset does not exist.") + raise ValueError("Media Preset does not exist.") - text_values = _dict_field(node.fields.get("text_values") or node.fields.get("text_values_json")) - image_slots = _dict_field(node.fields.get("image_slots") or node.fields.get("image_slots_json")) + text_values: Dict[str, str] = { + key: str(value) + for key, value in _dict_field(node.fields.get("text_values") or node.fields.get("text_values_json")).items() + if value is not None and value != "" + } + image_slots: Dict[str, List[MediaRefInput]] = {} for field in preset.get("input_schema_json") or []: key = str(field.get("key") or "").strip() if not key: continue dynamic_value = node.fields.get(f"text__{_slug(key)}") if dynamic_value is not None and dynamic_value != "": - text_values[key] = dynamic_value + text_values[key] = str(dynamic_value) for group in preset.get("choice_groups_json") or []: key = str(group.get("key") or group.get("id") or "").strip() if not key: continue dynamic_value = node.fields.get(f"choice__{_slug(key)}") if dynamic_value is not None and dynamic_value != "": - text_values[key] = dynamic_value - connected_images = [_graph_ref_to_media_ref(ref) for ref in context.inputs_for(node, "image_refs")] - - cursor = 0 + text_values[key] = str(dynamic_value) for slot in preset.get("input_slots_json") or []: key = str(slot.get("key") or "").strip() - if not key or image_slots.get(key): - continue - dynamic_slot_refs = [_graph_ref_to_media_ref(ref) for ref in context.inputs_for(node, f"slot__{_slug(key)}")] - if dynamic_slot_refs: - image_slots[key] = dynamic_slot_refs + if not key: continue - max_files = int(slot.get("max_files") or 1) - selected = connected_images[cursor : cursor + max_files] - cursor += len(selected) + selected = [_graph_ref_to_media_input(ref) for ref in context.inputs_for(node, f"slot__{_slug(key)}")] if selected: image_slots[key] = selected - - missing_text = [] - for field in preset.get("input_schema_json") or []: - key = str(field.get("key") or "").strip() - if field.get("required") and not str(text_values.get(key) or field.get("default_value") or "").strip(): - missing_text.append(key) - if key and key not in text_values and field.get("default_value"): - text_values[key] = str(field.get("default_value")) - if missing_text: - raise ValueError("Preset Render missing required text field: %s" % ", ".join(missing_text)) - - missing_slots = [] - for slot in preset.get("input_slots_json") or []: - key = str(slot.get("key") or "").strip() - if slot.get("required") and not image_slots.get(key): - missing_slots.append(key) - if missing_slots: - raise ValueError("Preset Render missing required image slot: %s" % ", ".join(missing_slots)) - - rendered_prompt = service._render_preset_prompt(str(preset.get("prompt_template") or ""), text_values, image_slots) - image_refs = [] - for refs in image_slots.values(): - if isinstance(refs, list): - for item in refs: - if not isinstance(item, dict): - continue - image_refs.append( - GraphOutputRef( - kind="reference_media" if item.get("reference_id") else "asset", - media_type="image", - asset_id=item.get("asset_id"), - reference_id=item.get("reference_id"), - ) - ) + image_inputs = [item for refs in image_slots.values() for item in refs] + model_key = self._selected_model_key(node, preset) + if not model_key: + raise ValueError("Media Preset does not define a compatible model.") + model = next((item for item in kie_adapter.list_models() if str(item.get("key") or "") == model_key), {}) + task_modes = [str(item) for item in ((model or {}).get("task_modes") or ((model or {}).get("raw") or {}).get("task_modes") or [])] + task_mode = _select_task_mode( + task_modes, + output_media_type="image", + has_images=bool(image_inputs), + has_videos=False, + has_audios=False, + model_key=model_key, + ) + options = preset.get("default_options_json") if isinstance(preset.get("default_options_json"), dict) else {} context.record_node_metric(node, "preset_text_field_count", len(text_values)) - context.record_node_metric(node, "preset_image_ref_count", len(image_refs)) - return { - "prompt": [GraphOutputRef(kind="value", value=rendered_prompt, metadata={"type": "text", "preset_id": preset_id})], - "image_refs": image_refs, - "preset": [ - GraphOutputRef( - kind="value", - value={ - "preset_id": preset_id, - "key": preset.get("key"), - "label": preset.get("label"), - "recommended_models": preset.get("applies_to_models_json") or [], - }, - metadata={"type": "json"}, - ) - ], - "recommended_models": [ - GraphOutputRef( - kind="value", - value=preset.get("applies_to_models_json") or [], - metadata={"type": "json", "preset_id": preset_id}, - ) - ], - } + context.record_node_metric(node, "preset_image_ref_count", len(image_inputs)) + request = ValidateRequest( + model_key=model_key, + task_mode=task_mode, + prompt="", + images=image_inputs, + options=options, + preset_id=preset_id, + preset_text_values=text_values, + preset_image_slots=image_slots, + output_count=1, + ) + return submit_and_wait_for_kie_request(node=node, context=context, request=request, model_key=model_key) + + def _selected_model_key(self, node: GraphWorkflowNode, preset: Dict[str, Any]) -> str: + compatible = [str(item).strip() for item in (preset.get("applies_to_models_json") or []) if str(item).strip()] + default_model = str(preset.get("model_key") or "").strip() + if default_model and default_model not in compatible: + compatible.insert(0, default_model) + selected = str(node.fields.get("preset_model_key") or "").strip() + if selected and (not compatible or selected in compatible): + return selected + return compatible[0] if compatible else default_model def _slug(value: str) -> str: diff --git a/apps/api/app/graph/normalization.py b/apps/api/app/graph/normalization.py index a0e17c9..0340528 100644 --- a/apps/api/app/graph/normalization.py +++ b/apps/api/app/graph/normalization.py @@ -3,6 +3,7 @@ from copy import deepcopy from typing import Dict, Iterable +from .preset_catalog import media_preset_catalog from .prompt_recipe_catalog import prompt_recipe_for_node_type, prompt_recipe_catalog from .registry import registry from .schemas import GraphNodeDefinition, GraphWorkflow, GraphWorkflowEdge, GraphWorkflowNode @@ -12,6 +13,10 @@ def _recipe_by_id(catalog: Iterable[dict]) -> Dict[str, dict]: return {str(item.get("recipe_id") or ""): item for item in catalog if str(item.get("recipe_id") or "").strip()} +def _preset_by_id(catalog: Iterable[dict]) -> Dict[str, dict]: + return {str(item.get("preset_id") or ""): item for item in catalog if str(item.get("preset_id") or "").strip()} + + SEEDANCE_LEGACY_TARGET_PORTS = { "image_refs": "reference_images", "video_refs": "reference_videos", @@ -49,6 +54,28 @@ def normalize_prompt_recipe_node( return node.model_copy(update={"fields": fields, "type": node.type}) +def normalize_media_preset_node( + node: GraphWorkflowNode, + *, + preset_catalog_items: list[dict] | None = None, + preset_lookup: Dict[str, dict] | None = None, +) -> GraphWorkflowNode: + fields = dict(node.fields) + changed = False + catalog = preset_catalog_items if preset_catalog_items is not None else media_preset_catalog(status="all") + by_id = preset_lookup if preset_lookup is not None else _preset_by_id(catalog) + if node.type == "preset.render": + preset = by_id.get(str(fields.get("preset_id") or "").strip()) + if preset and not str(fields.get("preset_model_key") or "").strip(): + default_model_key = str(preset.get("default_model_key") or "") + if default_model_key: + fields["preset_model_key"] = default_model_key + changed = True + if not changed: + return node + return node.model_copy(update={"fields": fields, "type": node.type}) + + def materialize_node_field_defaults( node: GraphWorkflowNode, definition: GraphNodeDefinition | None, @@ -75,9 +102,12 @@ def materialize_workflow_defaults( definitions = definitions_by_type or registry.definitions_by_type() all_recipe_catalog = prompt_recipe_catalog(status="all") recipe_lookup = _recipe_by_id(all_recipe_catalog) + all_preset_catalog = media_preset_catalog(status="all") + preset_lookup = _preset_by_id(all_preset_catalog) nodes = [] for node in workflow.nodes: - normalized = normalize_prompt_recipe_node(node, recipe_catalog_items=all_recipe_catalog, recipe_lookup=recipe_lookup) + normalized = normalize_media_preset_node(node, preset_catalog_items=all_preset_catalog, preset_lookup=preset_lookup) + normalized = normalize_prompt_recipe_node(normalized, recipe_catalog_items=all_recipe_catalog, recipe_lookup=recipe_lookup) nodes.append(materialize_node_field_defaults(normalized, definitions.get(normalized.type))) seedance_node_ids = {node.id for node in nodes if node.type == "model.kie.seedance_2_0"} edges: list[GraphWorkflowEdge] = [] diff --git a/apps/api/app/graph/preset_catalog.py b/apps/api/app/graph/preset_catalog.py new file mode 100644 index 0000000..acc9b83 --- /dev/null +++ b/apps/api/app/graph/preset_catalog.py @@ -0,0 +1,275 @@ +from __future__ import annotations + +import sqlite3 +from typing import Any, Dict, Iterable, List + +from .. import kie_adapter, store +from .prompt_recipe_catalog import slug, title_from_key +from .schemas import GraphNodeField, GraphNodePort + + +def _model_labels() -> Dict[str, str]: + labels: Dict[str, str] = {} + for model in kie_adapter.list_models(): + key = str(model.get("key") or "").strip() + if not key: + continue + labels[key] = str(model.get("label") or model.get("name") or title_from_key(key)) + return labels + + +def _compatible_models(preset: Dict[str, Any]) -> List[str]: + models = [str(item).strip() for item in (preset.get("applies_to_models_json") or preset.get("applies_to_models") or []) if str(item).strip()] + model_key = str(preset.get("model_key") or "").strip() + if model_key and model_key not in models: + models.insert(0, model_key) + return models + + +def _field_help_text(*, required: bool, detail: str) -> str: + prefix = "Required." if required else "Optional." + return f"{prefix} {detail}".strip() if detail else prefix + + +def _selection_summary(preset: Dict[str, Any], compatible_models: List[Dict[str, str]]) -> Dict[str, Any]: + label = str(preset.get("label") or preset.get("key") or preset.get("preset_id") or "Media Preset") + image_slots = preset.get("input_slots_json") or [] + required_slots = [str(slot.get("label") or title_from_key(str(slot.get("key") or ""))) for slot in image_slots if slot.get("required")] + slot_count = len(image_slots) + model_label = compatible_models[0]["label"] if compatible_models else "No compatible model" + details = [f"Model: {model_label}", f"Image slots: {slot_count}"] + if required_slots: + details.append("Required images: " + ", ".join(required_slots)) + else: + details.append("Required images: none") + return { + "title": label, + "subtitle": "Media Preset", + "description": str(preset.get("description") or "Run this saved Media Preset."), + "details": details, + } + + +def media_preset_catalog(*, status: str = "all") -> List[Dict[str, Any]]: + try: + presets = store.list_presets() + except sqlite3.OperationalError as exc: + if "no such table: media_presets" not in str(exc): + raise + presets = [] + model_labels = _model_labels() + catalog: List[Dict[str, Any]] = [] + for preset in presets: + status_value = str(preset.get("status") or "active") + if status == "active" and status_value != "active": + continue + preset_id = str(preset.get("preset_id") or "").strip() + if not preset_id: + continue + compatible_model_keys = _compatible_models(preset) + compatible_models = [ + {"value": model_key, "label": model_labels.get(model_key) or title_from_key(model_key)} + for model_key in compatible_model_keys + ] + text_fields = [] + for field in preset.get("input_schema_json") or []: + key = str(field.get("key") or "").strip() + if not key: + continue + detail = str(field.get("help_text") or field.get("description") or "").strip() + text_fields.append( + { + "key": key, + "label": str(field.get("label") or title_from_key(key)), + "type": "textarea" if field.get("multiline") else "text", + "required": bool(field.get("required")), + "default_value": field.get("default_value"), + "placeholder": field.get("placeholder"), + "help_text": detail, + "display_help_text": _field_help_text(required=bool(field.get("required")), detail=detail), + } + ) + image_slots = [] + for slot in preset.get("input_slots_json") or []: + key = str(slot.get("key") or "").strip() + if not key: + continue + max_files = int(slot.get("max_files") or 1) + detail = str(slot.get("help_text") or slot.get("description") or "").strip() + image_slots.append( + { + "key": key, + "label": str(slot.get("label") or title_from_key(key)), + "required": bool(slot.get("required")), + "max_files": max(1, max_files), + "help_text": detail, + } + ) + choice_groups = [] + for group in preset.get("choice_groups_json") or []: + key = str(group.get("key") or group.get("id") or "").strip() + choices = group.get("choices") or group.get("options") or [] + if not key or not choices: + continue + choice_groups.append( + { + "key": key, + "label": str(group.get("label") or title_from_key(key)), + "required": bool(group.get("required")), + "default_value": group.get("default"), + "options": choices, + "help_text": str(group.get("help_text") or group.get("description") or "").strip(), + } + ) + catalog.append( + { + "preset_id": preset_id, + "key": str(preset.get("key") or preset_id), + "label": str(preset.get("label") or preset.get("key") or preset_id), + "description": str(preset.get("description") or ""), + "status": status_value, + "compatible_models": compatible_models, + "default_model_key": compatible_model_keys[0] if compatible_model_keys else "", + "text_fields": text_fields, + "image_slots": image_slots, + "choice_groups": choice_groups, + "selection_summary": _selection_summary(preset, compatible_models), + } + ) + catalog.sort(key=lambda item: (str(item.get("label") or "").lower(), str(item.get("preset_id") or "").lower())) + return catalog + + +def media_preset_picker_options(catalog: Iterable[Dict[str, Any]]) -> List[Dict[str, Any]]: + return [ + { + "value": str(item["preset_id"]), + "label": str(item["label"]), + "description": str(item.get("description") or ""), + "selection_summary": dict(item.get("selection_summary") or {}), + } + for item in catalog + ] + + +def media_preset_search_aliases(catalog: Iterable[Dict[str, Any]]) -> List[str]: + aliases = ["media preset", "preset", "image preset"] + for item in catalog: + aliases.extend([str(item.get("label") or ""), str(item.get("key") or "")]) + deduped: List[str] = [] + seen: set[str] = set() + for alias in aliases: + normalized = alias.strip().lower() + if not normalized or normalized in seen: + continue + seen.add(normalized) + deduped.append(alias) + return deduped + + +def media_preset_model_options(catalog: Iterable[Dict[str, Any]]) -> List[Dict[str, Any]]: + options: List[Dict[str, Any]] = [] + seen: set[str] = set() + for preset in catalog: + for model in preset.get("compatible_models") or []: + value = str(model.get("value") or "").strip() + if not value or value in seen: + continue + seen.add(value) + options.append({"value": value, "label": str(model.get("label") or title_from_key(value))}) + return options + + +def media_preset_input_ports(catalog: Iterable[Dict[str, Any]]) -> List[GraphNodePort]: + merged: Dict[str, Dict[str, Any]] = {} + for preset in catalog: + preset_id = str(preset.get("preset_id") or "") + for slot in preset.get("image_slots") or []: + key = str(slot.get("key") or "").strip() + if not key: + continue + port_id = f"slot__{slug(key)}" + entry = merged.setdefault( + port_id, + { + "label": str(slot.get("label") or title_from_key(key)), + "max_files": 1, + "preset_ids": [], + }, + ) + entry["preset_ids"].append(preset_id) + entry["max_files"] = max(int(entry["max_files"]), int(slot.get("max_files") or 1)) + ports: List[GraphNodePort] = [] + for port_id, entry in sorted(merged.items(), key=lambda item: str(item[1]["label"]).lower()): + ports.append( + GraphNodePort( + id=port_id, + label=str(entry["label"]), + type="image", + array=True, + min=0, + max=int(entry["max_files"]) or None, + required=False, + accepts=["image"], + description="Image input for the selected Media Preset.", + visible_if={"field": "preset_id", "in": sorted({str(item) for item in entry["preset_ids"]})}, + ) + ) + return ports + + +def media_preset_dynamic_fields(catalog: Iterable[Dict[str, Any]]) -> List[GraphNodeField]: + merged: Dict[str, Dict[str, Any]] = {} + for preset in catalog: + preset_id = str(preset.get("preset_id") or "") + for field in preset.get("text_fields") or []: + key = str(field.get("key") or "").strip() + if not key: + continue + field_id = f"text__{slug(key)}" + entry = merged.setdefault( + field_id, + { + "label": str(field.get("label") or title_from_key(key)), + "type": str(field.get("type") or "text"), + "placeholder": str(field.get("placeholder") or ""), + "help_text": str(field.get("display_help_text") or field.get("help_text") or ""), + "preset_ids": [], + }, + ) + entry["preset_ids"].append(preset_id) + for group in preset.get("choice_groups") or []: + key = str(group.get("key") or "").strip() + if not key: + continue + field_id = f"choice__{slug(key)}" + entry = merged.setdefault( + field_id, + { + "label": str(group.get("label") or title_from_key(key)), + "type": "select", + "placeholder": "", + "help_text": str(group.get("help_text") or ""), + "options": list(group.get("options") or []), + "preset_ids": [], + }, + ) + entry["preset_ids"].append(preset_id) + if not entry.get("options") and group.get("options"): + entry["options"] = list(group.get("options") or []) + fields: List[GraphNodeField] = [] + for field_id, entry in sorted(merged.items(), key=lambda item: str(item[1]["label"]).lower()): + fields.append( + GraphNodeField( + id=field_id, + label=str(entry["label"]), + type=str(entry["type"]), + required=False, + default=None, + placeholder=str(entry.get("placeholder") or "") or None, + options=list(entry.get("options") or []), + help_text=str(entry.get("help_text") or "") or None, + visible_if={"field": "preset_id", "in": sorted({str(item) for item in entry["preset_ids"]})}, + ) + ) + return fields diff --git a/apps/api/app/graph/pricing.py b/apps/api/app/graph/pricing.py index d145243..3f545e7 100644 --- a/apps/api/app/graph/pricing.py +++ b/apps/api/app/graph/pricing.py @@ -602,6 +602,8 @@ def _output_port(definition: Optional[GraphNodeDefinition], port_id: str): def _model_options(node: GraphWorkflowNode, definition: GraphNodeDefinition) -> Dict[str, Any]: + if definition.source.get("preset_id"): + return {} keys = {field.id for field in definition.fields if field.id not in {"prompt", "output_count"}} return {key: value for key, value in node.fields.items() if key in keys and value is not None and value != ""} diff --git a/apps/api/app/graph/registry.py b/apps/api/app/graph/registry.py index 5fa2022..35c59cf 100644 --- a/apps/api/app/graph/registry.py +++ b/apps/api/app/graph/registry.py @@ -1,6 +1,5 @@ from __future__ import annotations -import sqlite3 from typing import Any, Dict, List, Optional from .. import kie_adapter, store @@ -53,16 +52,6 @@ def _is_supported_graph_model_option(model_key: str, option_key: str) -> bool: return option_key not in blocked -def _list_active_presets_for_graph() -> List[Dict[str, Any]]: - try: - presets = store.list_presets() - except sqlite3.OperationalError as exc: - if "no such table: media_presets" not in str(exc): - raise - presets = [] - return [preset for preset in presets if str(preset.get("status") or "active") == "active"] - - def _visible_condition_from_option(spec: Dict[str, Any]) -> Optional[Dict[str, Any]]: raw_condition = spec.get("ui_visible_when") if not isinstance(raw_condition, dict) or not raw_condition: @@ -240,7 +229,7 @@ def _layout_ui(definition: GraphNodeDefinition) -> GraphNodeDefinition: if definition.category.startswith("Models/"): computed_min_width = max(computed_min_width, 340) computed_min_height = max(computed_min_height, 440) - if definition.type == "preset.render" or definition.type.startswith("preset.render."): + if definition.type == "preset.render": computed_min_width = max(computed_min_width, 340) computed_min_height = max(computed_min_height, 380) if definition.type == "prompt.recipe" or definition.type.startswith("prompt.recipe."): @@ -385,8 +374,6 @@ def _build_definitions(self) -> List[GraphNodeDefinition]: continue definitions.append(definition) seen_model_nodes.add(definition.type) - for preset in _list_active_presets_for_graph(): - definitions.append(self._preset_render_definition(preset)) definitions = [_layout_ui(definition) for definition in definitions] validate_node_definitions(definitions) return definitions @@ -617,80 +604,4 @@ def _kie_model_definition(self, model: Dict[str, Any]) -> GraphNodeDefinition: fields=fields, ) - def _preset_render_definition(self, preset: Dict[str, Any]) -> GraphNodeDefinition: - preset_id = str(preset.get("preset_id") or "") - preset_key = str(preset.get("key") or preset_id) - input_ports = [] - for slot in preset.get("input_slots_json") or []: - key = str(slot.get("key") or "").strip() - if not key: - continue - input_ports.append( - GraphNodePort( - id=f"slot__{_slug(key)}", - label=str(slot.get("label") or _title_from_key(key)), - type="image", - array=True, - min=1 if slot.get("required") else 0, - max=int(slot.get("max_files") or 1), - required=bool(slot.get("required")), - accepts=["image"], - ) - ) - fields = [ - GraphNodeField(id="preset_id", label="Preset", type="text", required=False, default=preset_id, hidden=True), - ] - for field in preset.get("input_schema_json") or []: - key = str(field.get("key") or "").strip() - if not key: - continue - fields.append( - GraphNodeField( - id=f"text__{_slug(key)}", - label=str(field.get("label") or _title_from_key(key)), - type="textarea" if field.get("multiline") else "text", - required=bool(field.get("required")), - default=field.get("default_value") or "", - placeholder=field.get("placeholder"), - help_text=field.get("help_text") or field.get("description"), - ) - ) - for group in preset.get("choice_groups_json") or []: - key = str(group.get("key") or group.get("id") or "").strip() - choices = group.get("choices") or group.get("options") or [] - if not key or not choices: - continue - fields.append( - GraphNodeField( - id=f"choice__{_slug(key)}", - label=str(group.get("label") or _title_from_key(key)), - type="select", - required=bool(group.get("required")), - default=group.get("default"), - options=choices, - ) - ) - return GraphNodeDefinition( - type=f"preset.render.{_slug(preset_key or preset_id)}", - title=str(preset.get("label") or preset_key or "Render Preset"), - description=str(preset.get("description") or "Render this structured Media Studio preset."), - category="Preset", - search_aliases=["preset", "render", preset_key, str(preset.get("label") or "")], - tags=["preset", "prompt", "image"], - source={"kind": "preset", "preset_id": preset_id, "preset_key": preset_key}, - execution={"executor": "preset.render", "mode": "sync", "cacheable": True, "output_node": False}, - limits={"max_input_images": 8}, - ui={"default_size": {"width": 360, "height": 460}, "accent": "purple", "icon": "preset"}, - ports={ - "inputs": input_ports, - "outputs": [ - GraphNodePort(id="prompt", label="Prompt", type="text"), - GraphNodePort(id="image_refs", label="Image Refs", type="image", array=True), - GraphNodePort(id="preset", label="Preset", type="json"), - GraphNodePort(id="recommended_models", label="Recommended Models", type="json", advanced=True), - ], - }, - fields=fields, - ) - registry = GraphNodeRegistry() diff --git a/apps/api/app/graph/routes.py b/apps/api/app/graph/routes.py index 04420ac..d9d936a 100644 --- a/apps/api/app/graph/routes.py +++ b/apps/api/app/graph/routes.py @@ -24,6 +24,8 @@ GraphRunListResponse, GraphRunStatusNode, GraphRunStatusResponse, + GraphRunSummary, + GraphRunSummaryListResponse, GraphTemplate, GraphTemplateListResponse, GraphTemplateRecord, @@ -76,6 +78,23 @@ def _shape_run(record: dict) -> GraphRun: return shaped +def _shape_run_summary(record: dict) -> GraphRunSummary: + return GraphRunSummary( + run_id=str(record["run_id"]), + workflow_id=str(record["workflow_id"]), + status=str(record.get("status") or "queued"), + schema_version=int(record.get("schema_version") or 1), + metrics_json=record.get("metrics_json") if isinstance(record.get("metrics_json"), dict) else {}, + error=record.get("error"), + node_count=int(record.get("node_count") or 0), + artifact_count=int(record.get("artifact_count") or 0), + created_at=record.get("created_at"), + started_at=record.get("started_at"), + finished_at=record.get("finished_at"), + updated_at=record.get("updated_at"), + ) + + def _shape_run_status(record: dict) -> GraphRunStatusResponse: nodes = [ GraphRunStatusNode( @@ -220,11 +239,23 @@ def list_workflow_runs(workflow_id: str, limit: int = Query(default=50, ge=1, le return GraphRunListResponse(items=[_shape_run(item) for item in store.list_graph_runs_for_workflow(workflow_id, limit=limit)]) +@router.get("/workflows/{workflow_id}/runs/summary", response_model=GraphRunSummaryListResponse) +def list_workflow_run_summaries(workflow_id: str, limit: int = Query(default=15, ge=1, le=50)) -> GraphRunSummaryListResponse: + if not store.get_graph_workflow(workflow_id): + raise _not_found("workflow") + return GraphRunSummaryListResponse(items=[_shape_run_summary(item) for item in store.list_graph_run_summaries_for_workflow(workflow_id, limit=limit)]) + + @router.get("/runs", response_model=GraphRunListResponse) def list_runs(limit: int = Query(default=100, ge=1, le=500)) -> GraphRunListResponse: return GraphRunListResponse(items=[_shape_run(item) for item in store.list_graph_runs(limit=limit)]) +@router.get("/runs/summary", response_model=GraphRunSummaryListResponse) +def list_run_summaries(limit: int = Query(default=15, ge=1, le=50)) -> GraphRunSummaryListResponse: + return GraphRunSummaryListResponse(items=[_shape_run_summary(item) for item in store.list_graph_run_summaries(limit=limit)]) + + @router.get("/runs/{run_id}", response_model=GraphRun) def get_run(run_id: str) -> GraphRun: record = store.get_graph_run(run_id) diff --git a/apps/api/app/graph/runtime.py b/apps/api/app/graph/runtime.py index 800124f..6cf6c6e 100644 --- a/apps/api/app/graph/runtime.py +++ b/apps/api/app/graph/runtime.py @@ -550,8 +550,6 @@ def execute_run(self, run_id: str, *, resume: bool = False) -> None: executor = self.executors.get(node.type) if not executor and node.type.startswith("model.kie."): executor = self.executors.get("model.kie") - if not executor and node.type.startswith("preset.render."): - executor = self.executors.get("preset.render") if not executor and node.type.startswith("prompt.recipe."): executor = self.executors.get("prompt.recipe") if not executor: diff --git a/apps/api/app/graph/schemas.py b/apps/api/app/graph/schemas.py index 39190df..e3ef9a7 100644 --- a/apps/api/app/graph/schemas.py +++ b/apps/api/app/graph/schemas.py @@ -208,6 +208,21 @@ class GraphRun(BaseModel): updated_at: Optional[str] = None +class GraphRunSummary(BaseModel): + run_id: str + workflow_id: str + status: str = "queued" + schema_version: int = 1 + metrics_json: Dict[str, Any] = Field(default_factory=dict) + error: Optional[str] = None + node_count: int = 0 + artifact_count: int = 0 + created_at: Optional[str] = None + started_at: Optional[str] = None + finished_at: Optional[str] = None + updated_at: Optional[str] = None + + class GraphRunNode(BaseModel): run_node_id: str run_id: str @@ -264,6 +279,10 @@ class GraphRunListResponse(BaseModel): items: List[GraphRun] = Field(default_factory=list) +class GraphRunSummaryListResponse(BaseModel): + items: List[GraphRunSummary] = Field(default_factory=list) + + class GraphRunEventsResponse(BaseModel): items: List[GraphRunEvent] = Field(default_factory=list) diff --git a/apps/api/app/graph/system_nodes_preset.py b/apps/api/app/graph/system_nodes_preset.py index bcab06f..12923d3 100644 --- a/apps/api/app/graph/system_nodes_preset.py +++ b/apps/api/app/graph/system_nodes_preset.py @@ -1,52 +1,63 @@ from __future__ import annotations -import sqlite3 from typing import List +from .preset_catalog import ( + media_preset_catalog, + media_preset_dynamic_fields, + media_preset_input_ports, + media_preset_model_options, + media_preset_picker_options, + media_preset_search_aliases, +) from .schemas import GraphNodeDefinition, GraphNodeField, GraphNodePort -from .. import store - - -def _preset_options() -> List[dict[str, str]]: - try: - presets = store.list_presets() - except sqlite3.OperationalError as exc: - if "no such table: media_presets" not in str(exc): - raise - presets = [] - return [ - {"value": str(item["preset_id"]), "label": str(item.get("label") or item.get("key") or item["preset_id"])} - for item in presets - if str(item.get("status") or "active") == "active" - ] - def preset_node_definitions() -> List[GraphNodeDefinition]: + all_catalog = media_preset_catalog(status="all") + active_catalog = [item for item in all_catalog if str(item.get("status") or "active") == "active"] + input_ports = media_preset_input_ports(all_catalog) return [ GraphNodeDefinition( type="preset.render", - title="Render Preset", - description="Render an existing Media Studio structured preset into prompt text and image refs.", + title="Media Preset", + description="Run any saved Media Preset from one schema-driven graph node.", + help_text="Choose a saved Media Preset, then fill only the fields and image inputs that appear for that preset.", category="Preset", - search_aliases=["preset", "render", "template", "prompt"], - tags=["preset", "prompt", "image"], - source={"kind": "system"}, - execution={"executor": "preset.render", "mode": "sync", "cacheable": True, "output_node": False}, - limits={"max_input_images": 8}, - ui={"default_size": {"width": 360, "height": 420}, "accent": "purple", "icon": "preset"}, + search_aliases=media_preset_search_aliases(active_catalog), + tags=["preset", "image", "media"], + source={"kind": "media_preset", "preset_catalog": all_catalog}, + execution={"executor": "preset.render", "mode": "async", "cacheable": True, "output_node": False, "retryable": True}, + limits={ + "max_input_images": max((port.max or 0) for port in input_ports) if input_ports else 0, + "output_count": {"default": 1, "max": 1}, + }, + ui={"default_size": {"width": 420, "height": 620}, "accent": "blue", "icon": "preset", "field_layout": "stack"}, ports={ - "inputs": [GraphNodePort(id="image_refs", label="Image Refs", type="image", array=True, required=False, max=8, accepts=["image"])], + "inputs": input_ports, "outputs": [ - GraphNodePort(id="prompt", label="Prompt", type="text"), - GraphNodePort(id="image_refs", label="Image Refs", type="image", array=True), - GraphNodePort(id="preset", label="Preset", type="json"), + GraphNodePort(id="image", label="Image", type="image"), + GraphNodePort(id="job", label="Job", type="job", advanced=True), ], }, fields=[ - GraphNodeField(id="preset_id", label="Preset", type="preset_picker", required=True, options=_preset_options()), - GraphNodeField(id="text_values_json", label="Text Values JSON", type="textarea", required=False, default="{}", placeholder='{"subject":"..."}'), - GraphNodeField(id="image_slots_json", label="Image Slots JSON", type="textarea", required=False, default="{}", placeholder='{"subject":[{"reference_id":"..."}]}'), + GraphNodeField( + id="preset_id", + label="Media Preset", + type="preset_picker", + required=True, + options=media_preset_picker_options(active_catalog), + help_text="Choose the saved preset to run. The fields and image inputs below update to match it.", + ), + GraphNodeField( + id="preset_model_key", + label="Model", + type="select", + required=False, + options=media_preset_model_options(all_catalog), + help_text="Model used for this preset. Options are limited to models the selected preset supports.", + ), + *media_preset_dynamic_fields(all_catalog), ], ), ] diff --git a/apps/api/app/graph/validator.py b/apps/api/app/graph/validator.py index 9b9b560..f5f4e05 100644 --- a/apps/api/app/graph/validator.py +++ b/apps/api/app/graph/validator.py @@ -53,11 +53,18 @@ def _preset_id_for_node(node: GraphWorkflowNode, definition) -> str: preset_id = str(node.fields.get("preset_id") or "").strip() if preset_id: return preset_id - if node.type.startswith("preset.render."): - return str(definition.source.get("preset_id") or "").strip() return "" +def _preset_model_key_for_node(node: GraphWorkflowNode, preset: Dict[str, Any]) -> tuple[str, List[str]]: + compatible = [str(item).strip() for item in (preset.get("applies_to_models_json") or []) if str(item).strip()] + default_model = str(preset.get("model_key") or "").strip() + if default_model and default_model not in compatible: + compatible.insert(0, default_model) + selected = str(node.fields.get("preset_model_key") or "").strip() + return selected or (compatible[0] if compatible else default_model), compatible + + def _node_execution_mode(node: GraphWorkflowNode) -> str: execution = node.metadata.get("execution") if isinstance(node.metadata.get("execution"), dict) else {} mode = str(execution.get("mode") or "enabled") @@ -194,7 +201,7 @@ def validate_workflow(workflow: GraphWorkflow) -> GraphValidationResult: if node.fields.get("reference_id") and not store.get_reference_media(str(node.fields["reference_id"])): errors.append(GraphError(code="missing_reference_media", message="Referenced reference media does not exist.", node_id=node.id, field_id="reference_id")) preset_id = _preset_id_for_node(node, definition) - if (node.type == "preset.render" or node.type.startswith("preset.render.")) and preset_id: + if node.type == "preset.render" and preset_id: preset = store.get_preset(preset_id) if not preset: errors.append(GraphError(code="missing_preset", message="Referenced preset does not exist.", node_id=node.id, field_id="preset_id")) @@ -205,13 +212,40 @@ def validate_workflow(workflow: GraphWorkflow) -> GraphValidationResult: dynamic_value = node.fields.get(f"text__{_slug(key)}") if key and dynamic_value is not None and dynamic_value != "": text_values[key] = dynamic_value - missing_text = [ - str(field.get("key")) - for field in (preset.get("input_schema_json") or []) - if field.get("required") and not str(text_values.get(str(field.get("key"))) or field.get("default_value") or "").strip() - ] - for key in missing_text: - errors.append(GraphError(code="missing_preset_text", message=f"Missing required preset text field: {key}", node_id=node.id, field_id="text_values_json")) + for field in preset.get("input_schema_json") or []: + key = str(field.get("key") or "").strip() + if field.get("required") and not str(text_values.get(key) or field.get("default_value") or "").strip(): + errors.append( + GraphError( + code="missing_preset_text", + message=f"Missing required preset text field: {key}", + node_id=node.id, + field_id=f"text__{_slug(key)}", + ) + ) + for group in preset.get("choice_groups_json") or []: + key = str(group.get("key") or group.get("id") or "").strip() + if key and group.get("required") and not str(node.fields.get(f"choice__{_slug(key)}") or group.get("default") or "").strip(): + errors.append( + GraphError( + code="missing_preset_choice", + message=f"Missing required preset choice: {key}", + node_id=node.id, + field_id=f"choice__{_slug(key)}", + ) + ) + model_key, compatible_models = _preset_model_key_for_node(node, preset) + if not model_key: + errors.append(GraphError(code="missing_preset_model", message="Media Preset has no compatible model.", node_id=node.id, field_id="preset_model_key")) + elif compatible_models and model_key not in compatible_models: + errors.append( + GraphError( + code="preset_model_not_compatible", + message="Selected model is not compatible with this Media Preset.", + node_id=node.id, + field_id="preset_model_key", + ) + ) if node.type == "prompt.recipe" or node.type.startswith("prompt.recipe."): prompt_recipe_context = validate_prompt_recipe_node_setup(node, definition, errors=errors) if prompt_recipe_context: @@ -390,22 +424,31 @@ def validate_workflow(workflow: GraphWorkflow) -> GraphValidationResult: ) ) preset_id = _preset_id_for_node(node, definition) - if (node.type == "preset.render" or node.type.startswith("preset.render.")) and preset_id: + if node.type == "preset.render" and preset_id: preset = store.get_preset(preset_id) if preset: - slot_values = _dict_field(node.fields.get("image_slots") or node.fields.get("image_slots_json")) - connected_count = incoming_by_target_port[(node.id, "image_refs")] - has_connected_refs = connected_count > 0 for slot in preset.get("input_slots_json") or []: key = str(slot.get("key") or "").strip() - dynamic_connected_count = incoming_by_target_port[(node.id, f"slot__{_slug(key)}")] - if slot.get("required") and not slot_values.get(key) and not has_connected_refs and dynamic_connected_count <= 0: + port_id = f"slot__{_slug(key)}" + connected_count = incoming_by_target_port[(node.id, port_id)] + available_count = available_incoming_by_target_port[(node.id, port_id)] + if slot.get("required") and available_count <= 0: errors.append( GraphError( code="missing_preset_image_slot", message=f"Missing required preset image slot: {key}", node_id=node.id, - port_id="image_refs", + port_id=port_id, + ) + ) + max_files = int(slot.get("max_files") or 1) + if max_files > 0 and connected_count > max_files: + errors.append( + GraphError( + code="preset_image_slot_max_exceeded", + message=f"Too many images connected to preset image slot: {key}", + node_id=node.id, + port_id=port_id, ) ) if node.type == "prompt.recipe" or node.type.startswith("prompt.recipe."): diff --git a/apps/api/app/store.py b/apps/api/app/store.py index 6b6ba26..b5177b0 100644 --- a/apps/api/app/store.py +++ b/apps/api/app/store.py @@ -49,6 +49,8 @@ list_graph_artifacts_for_run, list_graph_run_events, list_graph_run_nodes, + list_graph_run_summaries, + list_graph_run_summaries_for_workflow, list_graph_runs, list_graph_runs_for_workflow, list_graph_templates, diff --git a/apps/api/app/store_graph.py b/apps/api/app/store_graph.py index 5e0f5c5..17e371a 100644 --- a/apps/api/app/store_graph.py +++ b/apps/api/app/store_graph.py @@ -138,6 +138,39 @@ def list_graph_runs(limit: int = 100) -> List[Dict[str, Any]]: return [_decode_row(row) for row in rows] +GRAPH_RUN_SUMMARY_SELECT = """ +SELECT + graph_runs.run_id, + graph_runs.workflow_id, + graph_runs.status, + graph_runs.schema_version, + graph_runs.metrics_json, + graph_runs.error, + graph_runs.created_at, + graph_runs.started_at, + graph_runs.finished_at, + graph_runs.updated_at, + (SELECT COUNT(*) FROM graph_run_nodes WHERE graph_run_nodes.run_id = graph_runs.run_id) AS node_count, + (SELECT COUNT(*) FROM graph_artifacts WHERE graph_artifacts.run_id = graph_runs.run_id) AS artifact_count +FROM graph_runs +""" + + +def _list_graph_run_summaries(*, limit: int, workflow_id: str | None = None) -> List[Dict[str, Any]]: + where_clause = "WHERE workflow_id = ?" if workflow_id else "" + params: tuple[Any, ...] = (workflow_id, limit) if workflow_id else (limit,) + with get_connection() as connection: + rows = connection.execute( + f"{GRAPH_RUN_SUMMARY_SELECT} {where_clause} ORDER BY created_at DESC LIMIT ?", + params, + ).fetchall() + return [_decode_row(row) for row in rows] + + +def list_graph_run_summaries(limit: int = 100) -> List[Dict[str, Any]]: + return _list_graph_run_summaries(limit=limit) + + def list_graph_runs_for_workflow(workflow_id: str, limit: int = 100) -> List[Dict[str, Any]]: with get_connection() as connection: rows = connection.execute( @@ -152,6 +185,10 @@ def list_graph_runs_for_workflow(workflow_id: str, limit: int = 100) -> List[Dic return [_decode_row(row) for row in rows] +def list_graph_run_summaries_for_workflow(workflow_id: str, limit: int = 100) -> List[Dict[str, Any]]: + return _list_graph_run_summaries(limit=limit, workflow_id=workflow_id) + + def get_graph_run(run_id: str) -> Optional[Dict[str, Any]]: return _get_table("graph_runs", "run_id", run_id) diff --git a/apps/api/tests/test_graph_studio.py b/apps/api/tests/test_graph_studio.py index 98cb9db..f3231bd 100644 --- a/apps/api/tests/test_graph_studio.py +++ b/apps/api/tests/test_graph_studio.py @@ -561,6 +561,34 @@ def test_graph_note_node_runs_without_ports(client) -> None: assert note_node["metrics_json"]["note_character_count"] == len("# Plan\n\n- Connect source image\n- Run final model") +def test_graph_run_summary_lists_do_not_embed_full_run_payloads(client) -> None: + workflow = { + "schema_version": 1, + "name": "Summary payload guard", + "nodes": [ + { + "id": "note", + "type": "utility.note", + "position": {"x": 0, "y": 0}, + "fields": {"body": "Keep history lightweight."}, + } + ], + "edges": [], + } + final_payload = _run_graph_workflow(client, workflow) + + summary = client.get(f"/media/graph/workflows/{final_payload['workflow_id']}/runs/summary?limit=10") + assert summary.status_code == 200, summary.text + item = next(run for run in summary.json()["items"] if run["run_id"] == final_payload["run_id"]) + + assert item["node_count"] == 1 + assert item["artifact_count"] == 0 + assert "workflow_json" not in item + assert "compiled_graph_json" not in item + assert "output_snapshot_json" not in item + assert "nodes" not in item + + def test_graph_node_definitions_include_valid_layout_metadata(client) -> None: response = client.get("/media/graph/node-definitions") assert response.status_code == 200, response.text @@ -3994,7 +4022,7 @@ def test_graph_preset_render_validates_required_slots_and_runs(client, app_modul "id": "preset", "type": "preset.render", "position": {"x": 0, "y": 0}, - "fields": {"preset_id": preset["preset_id"], "text_values_json": '{"style":"cinematic"}'}, + "fields": {"preset_id": preset["preset_id"], "text__style": "cinematic", "preset_model_key": "nano-banana-pro"}, } ], "edges": [], @@ -4006,23 +4034,53 @@ def test_graph_preset_render_validates_required_slots_and_runs(client, app_modul assert invalid.json()["valid"] is False assert any(error["code"] == "missing_preset_image_slot" for error in invalid.json()["errors"]) - workflow = _workflow(reference_id) - workflow["nodes"].insert( - 1, - { - "id": "preset", - "type": "preset.render", - "position": {"x": 220, "y": -180}, - "fields": {"preset_id": preset["preset_id"], "text_values_json": '{"style":"cinematic"}'}, - }, - ) - model_node = next(node for node in workflow["nodes"] if node["id"] == "model") - model_node["fields"].pop("prompt") + muted_slot_workflow = { + "schema_version": 1, + "name": "Preset muted slot", + "nodes": [ + { + "id": "load", + "type": "media.load_image", + "position": {"x": 0, "y": 0}, + "fields": {"reference_id": reference_id}, + "metadata": {"execution": {"mode": "muted"}}, + }, + { + "id": "preset", + "type": "preset.render", + "position": {"x": 320, "y": 0}, + "fields": {"preset_id": preset["preset_id"], "text__style": "cinematic", "preset_model_key": "nano-banana-pro"}, + }, + ], + "edges": [ + {"id": "edge-load-preset", "source": "load", "source_port": "image", "target": "preset", "target_port": "slot__subject"}, + ], + } + created = client.post("/media/graph/workflows", json=muted_slot_workflow) + assert created.status_code == 200, created.text + invalid = client.post(f"/media/graph/workflows/{created.json()['workflow_id']}/validate", json=muted_slot_workflow) + assert invalid.status_code == 200, invalid.text + assert invalid.json()["valid"] is False + assert any(error["code"] == "missing_preset_image_slot" for error in invalid.json()["errors"]) + + workflow = { + "schema_version": 1, + "name": "Preset render", + "nodes": [ + {"id": "load", "type": "media.load_image", "position": {"x": 0, "y": 0}, "fields": {"reference_id": reference_id}}, + { + "id": "preset", + "type": "preset.render", + "position": {"x": 320, "y": 0}, + "fields": {"preset_id": preset["preset_id"], "text__style": "cinematic", "preset_model_key": "nano-banana-pro"}, + }, + {"id": "save", "type": "media.save_image", "position": {"x": 720, "y": 0}, "fields": {"label": "Preset final"}}, + ], + "edges": [], + } workflow["edges"] = [ - {"id": "edge-load-preset", "source": "load", "source_port": "image", "target": "preset", "target_port": "image_refs"}, - {"id": "edge-preset-model-prompt", "source": "preset", "source_port": "prompt", "target": "model", "target_port": "prompt"}, - {"id": "edge-preset-model-image", "source": "preset", "source_port": "image_refs", "target": "model", "target_port": "image_refs"}, - {"id": "edge-model-save", "source": "model", "source_port": "image", "target": "save", "target_port": "image"}, + {"id": "edge-load-preset", "source": "load", "source_port": "image", "target": "preset", "target_port": "slot__subject"}, + {"id": "edge-preset-save", "source": "preset", "source_port": "image", "target": "save", "target_port": "image"}, ] created = client.post("/media/graph/workflows", json=workflow) assert created.status_code == 200, created.text @@ -4046,12 +4104,13 @@ def test_graph_preset_render_validates_required_slots_and_runs(client, app_modul assert final_payload["status"] == "completed", final_payload preset_node = next(node for node in final_payload["nodes"] if node["node_id"] == "preset") assert preset_node["metrics_json"]["preset_image_ref_count"] == 1 - assert "cinematic editorial image" in preset_node["output_snapshot_json"]["prompt"][0]["value"] + assert "image" in preset_node["output_snapshot_json"] def test_graph_dynamic_preset_node_renders_fields_and_slots(client, app_modules) -> None: store = app_modules["store"] reference_id = _create_reference_image(app_modules) + second_reference_id = _create_named_reference_image(app_modules, name="graph-source-second.png", sha="graph-source-second-hash") preset = store.create_or_update_preset( { "preset_id": "graph-dynamic-preset-test", @@ -4064,26 +4123,47 @@ def test_graph_dynamic_preset_node_renders_fields_and_slots(client, app_modules) "applies_to_models_json": ["nano-banana-pro"], "prompt_template": "Create a {{style}} portrait from [[subject]].", "input_schema_json": [{"key": "style", "label": "Style", "required": True}], - "input_slots_json": [{"key": "subject", "label": "Subject", "required": True, "max_files": 1}], + "input_slots_json": [{"key": "subject", "label": "Subject", "required": True, "max_files": 3}], "choice_groups_json": [], "default_options_json": {}, "rules_json": {}, } ) definitions = client.post("/media/graph/node-definitions/refresh").json()["items"] - node_type = "preset.render.graph_dynamic_preset_test" - dynamic_definition = next(item for item in definitions if item["type"] == node_type) + dynamic_definition = next(item for item in definitions if item["type"] == "preset.render") + assert not any(item["type"].startswith("preset.render.") for item in definitions) + preset_picker = next(field for field in dynamic_definition["fields"] if field["id"] == "preset_id") + assert any(option["value"] == preset["preset_id"] for option in preset_picker["options"]) + model_picker = next(field for field in dynamic_definition["fields"] if field["id"] == "preset_model_key") + assert any(option["value"] == "nano-banana-pro" for option in model_picker["options"]) assert any(field["id"] == "text__style" for field in dynamic_definition["fields"]) - assert any(port["id"] == "slot__subject" for port in dynamic_definition["ports"]["inputs"]) + subject_port = next(port for port in dynamic_definition["ports"]["inputs"] if port["id"] == "slot__subject") + assert subject_port["visible_if"]["field"] == "preset_id" + assert preset["preset_id"] in subject_port["visible_if"]["in"] + assert subject_port["max"] >= 3 + assert dynamic_definition["source"]["kind"] == "media_preset" + assert any(port["id"] == "image" and port["type"] == "image" for port in dynamic_definition["ports"]["outputs"]) + assert not any(port["id"] in {"prompt", "image_refs", "preset"} for port in dynamic_definition["ports"]["outputs"]) workflow = { "schema_version": 1, "name": "Dynamic preset", "nodes": [ {"id": "load", "type": "media.load_image", "position": {"x": 0, "y": 0}, "fields": {"reference_id": reference_id}}, - {"id": "preset", "type": node_type, "position": {"x": 320, "y": 0}, "fields": {"text__style": "cinematic"}}, + {"id": "load-2", "type": "media.load_image", "position": {"x": 0, "y": 180}, "fields": {"reference_id": second_reference_id}}, + { + "id": "preset", + "type": "preset.render", + "position": {"x": 320, "y": 0}, + "fields": {"preset_id": preset["preset_id"], "preset_model_key": "nano-banana-pro", "text__style": "cinematic"}, + }, + {"id": "save", "type": "media.save_image", "position": {"x": 700, "y": 0}, "fields": {"label": "Dynamic preset final"}}, + ], + "edges": [ + {"id": "edge-load-preset", "source": "load", "source_port": "image", "target": "preset", "target_port": "slot__subject"}, + {"id": "edge-load-2-preset", "source": "load-2", "source_port": "image", "target": "preset", "target_port": "slot__subject"}, + {"id": "edge-preset-save", "source": "preset", "source_port": "image", "target": "save", "target_port": "image"}, ], - "edges": [{"id": "edge-load-preset", "source": "load", "source_port": "image", "target": "preset", "target_port": "slot__subject"}], } created = client.post("/media/graph/workflows", json=workflow) assert created.status_code == 200, created.text @@ -4091,6 +4171,26 @@ def test_graph_dynamic_preset_node_renders_fields_and_slots(client, app_modules) assert validation.status_code == 200, validation.text assert validation.json()["valid"] is True + run_response = client.post(f"/media/graph/workflows/{created.json()['workflow_id']}/runs", json={}) + assert run_response.status_code == 200, run_response.text + run_id = run_response.json()["run_id"] + final_payload = None + for _ in range(60): + current = client.get(f"/media/graph/runs/{run_id}") + assert current.status_code == 200 + final_payload = current.json() + if final_payload["status"] in {"completed", "failed"}: + break + time.sleep(0.1) + + assert final_payload is not None + assert final_payload["status"] == "completed", final_payload + preset_node = next(node for node in final_payload["nodes"] if node["node_id"] == "preset") + assert preset_node["metrics_json"]["preset_image_ref_count"] == 2 + assert "image" in preset_node["output_snapshot_json"] + save_node = next(node for node in final_payload["nodes"] if node["node_id"] == "save") + assert save_node["output_snapshot_json"]["image"][0]["asset_id"] + def test_graph_node_definitions_auto_invalidate_after_prompt_recipe_save(client) -> None: initial = client.get("/media/graph/node-definitions") @@ -4163,5 +4263,12 @@ def test_graph_node_definitions_auto_invalidate_after_preset_save(client) -> Non refreshed = client.get("/media/graph/node-definitions") assert refreshed.status_code == 200, refreshed.text - node_types = {item["type"] for item in refreshed.json()["items"]} - assert "preset.render.auto_refresh_preset" in node_types + definitions = refreshed.json()["items"] + dynamic_definition = next(item for item in definitions if item["type"] == "preset.render") + assert dynamic_definition["source"]["kind"] == "media_preset" + preset_picker = next(field for field in dynamic_definition["fields"] if field["id"] == "preset_id") + assert any(option["label"] == "Auto Refresh Preset" for option in preset_picker["options"]) + subject_port = next(port for port in dynamic_definition["ports"]["inputs"] if port["id"] == "slot__subject") + assert created.json()["preset_id"] in subject_port["visible_if"]["in"] + assert any(port["id"] == "image" and port["type"] == "image" for port in dynamic_definition["ports"]["outputs"]) + assert not any(port["id"] in {"prompt", "image_refs", "preset"} for port in dynamic_definition["ports"]["outputs"]) diff --git a/apps/web/components/graph-studio/graph-library-dialogs.tsx b/apps/web/components/graph-studio/graph-library-dialogs.tsx index 2b2aa4d..340aa43 100644 --- a/apps/web/components/graph-studio/graph-library-dialogs.tsx +++ b/apps/web/components/graph-studio/graph-library-dialogs.tsx @@ -7,7 +7,7 @@ import type { MediaAsset, MediaReference } from "@/lib/types"; import { GraphNodeTypeBadge } from "./components/graph-node-type-badge"; import { GraphRunHistoryPanel } from "./graph-run-history-panel"; import { GraphTemplateBrowser } from "./graph-template-browser"; -import type { GraphArtifact, GraphNodeDefinition, GraphRun, GraphTemplateRecord, GraphWorkflowRecord } from "./types"; +import type { GraphArtifact, GraphNodeDefinition, GraphRunHistoryItem, GraphTemplateRecord, GraphWorkflowRecord } from "./types"; import { graphDefinitionHiddenInSearch, rankGraphNodeDefinitions } from "./hooks/use-graph-node-search"; import { graphMediaDragPayload } from "./utils/graph-media-preview"; import { formatGraphTimestamp } from "./utils/graph-time"; @@ -48,7 +48,7 @@ export function GraphLibraryDialog({ references: MediaReference[]; assets: MediaAsset[]; workflowId: string | null; - runHistory: GraphRun[]; + runHistory: GraphRunHistoryItem[]; selectedHistoryRunId: string | null; selectedRunArtifacts: GraphArtifact[]; onClose: () => void; @@ -62,7 +62,7 @@ export function GraphLibraryDialog({ onAddLoadImageNode: (fields: Record) => void; onRefreshRunHistory: () => void; onInspectRun: (runId: string) => void; - onRestoreRun: (run: GraphRun) => void; + onRestoreRun: (run: GraphRunHistoryItem) => void | Promise; onPinArtifact: (artifact: GraphArtifact) => void; }) { const [nodeLibraryQuery, setNodeLibraryQuery] = useState(""); diff --git a/apps/web/components/graph-studio/graph-node-field.tsx b/apps/web/components/graph-studio/graph-node-field.tsx index 7d75ad5..975b1ac 100644 --- a/apps/web/components/graph-studio/graph-node-field.tsx +++ b/apps/web/components/graph-studio/graph-node-field.tsx @@ -6,6 +6,7 @@ import { GraphMarkdownNoteField } from "./graph-markdown-note"; import { GraphNodeProviderModelField } from "./graph-node-provider-model-field"; import { useGraphProviderModelCatalogContext, type GraphProviderKind } from "./hooks/use-graph-provider-model-catalog"; import type { GraphNodeData } from "./types"; +import { graphMediaPresetFieldOverride, graphMediaPresetSelectionDefaults } from "./utils/graph-media-preset"; import { graphPromptRuntimeFieldOverride } from "./utils/graph-prompt-provider"; import { graphPromptRecipeFieldOverride, graphPromptRecipeFilteredOptions, graphPromptRecipeOptionLabel, graphPromptRecipeSelectionDefaults } from "./utils/graph-prompt-recipe"; @@ -125,7 +126,7 @@ function GraphNodeTextareaField({ const textareaRef = useRef(null); const selectionRef = useRef<{ start: number; end: number } | null>(null); const textValue = String(value ?? ""); - const override = graphPromptRecipeFieldOverride(definition, nodeFields, field); + const override = graphMediaPresetFieldOverride(definition, nodeFields, field) ?? graphPromptRecipeFieldOverride(definition, nodeFields, field); const runtimeOverride = graphPromptRuntimeFieldOverride(definition.type, nodeFields, field); const placeholder = override?.placeholder ?? runtimeOverride?.placeholder ?? field.placeholder ?? ""; const markdownPreviewField = typeof definition.ui?.markdown_preview_field === "string" ? definition.ui.markdown_preview_field : null; @@ -190,7 +191,7 @@ export function GraphNodeFieldControl({ }) { if (field.hidden) return null; const commonClass = "graph-node-field-control nodrag"; - const fieldOverride = graphPromptRecipeFieldOverride(definition, nodeFields, field); + const fieldOverride = graphMediaPresetFieldOverride(definition, nodeFields, field) ?? graphPromptRecipeFieldOverride(definition, nodeFields, field); const runtimeOverride = graphPromptRuntimeFieldOverride(definition.type, nodeFields, field); if (field.type === "textarea") { return ; @@ -233,8 +234,8 @@ export function GraphNodeFieldControl({ ); } const options = field.type === "prompt_recipe_picker" ? graphPromptRecipeFilteredOptions(field, nodeFields) : fieldOverride?.options ?? field.options ?? []; - const emptyLabel = field.type === "prompt_recipe_picker" ? "Select recipe" : field.id === "project_id" ? "No group" : "Auto"; - const showEmptyOption = field.type === "prompt_recipe_picker" || (!field.required && (field.default === undefined || field.default === null || field.default === "")); + const emptyLabel = field.type === "prompt_recipe_picker" ? "Select recipe" : field.type === "preset_picker" ? "Select preset" : field.id === "project_id" ? "No group" : "Auto"; + const showEmptyOption = field.type === "prompt_recipe_picker" || field.type === "preset_picker" || (!field.required && (field.default === undefined || field.default === null || field.default === "")); return (