[codex] Harden dynamic media preset graph node

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(

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,116 +20,91 @@ 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):
     node_type = "preset.render"
 
     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:

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] = []