From ec3cdd7049d44a37ceade3b5fbc1064925be18a8 Mon Sep 17 00:00:00 2001 From: Gateway Date: Mon, 11 May 2026 15:21:53 +0700 Subject: [PATCH 01/20] Add experimental Graph Studio vertical slice --- apps/api/app/graph/__init__.py | 2 + apps/api/app/graph/compiler.py | 65 ++++ apps/api/app/graph/events.py | 9 + apps/api/app/graph/executors/__init__.py | 2 + apps/api/app/graph/executors/base.py | 35 ++ apps/api/app/graph/executors/batch_ops.py | 1 + apps/api/app/graph/executors/image_ops.py | 2 + apps/api/app/graph/executors/kie_model.py | 64 +++ apps/api/app/graph/executors/media_load.py | 45 +++ apps/api/app/graph/executors/media_save.py | 21 + apps/api/app/graph/executors/prompt_ops.py | 2 + apps/api/app/graph/executors/video_ops.py | 2 + apps/api/app/graph/media_probe.py | 2 + apps/api/app/graph/registry.py | 183 +++++++++ apps/api/app/graph/routes.py | 205 ++++++++++ apps/api/app/graph/runtime.py | 125 ++++++ apps/api/app/graph/schemas.py | 214 ++++++++++ apps/api/app/graph/validator.py | 107 +++++ apps/api/app/main.py | 2 + apps/api/app/store.py | 204 ++++++++++ apps/api/app/store_support.py | 115 ++++++ apps/api/tests/test_db_admin.py | 17 +- apps/api/tests/test_graph_studio.py | 127 ++++++ apps/web/app/globals.css | 218 +++++++++++ apps/web/app/graph-studio/page.tsx | 6 + .../components/graph-studio/graph-node.tsx | 108 +++++ .../components/graph-studio/graph-studio.tsx | 368 ++++++++++++++++++ apps/web/components/graph-studio/types.ts | 102 +++++ apps/web/components/studio-admin-shell.tsx | 3 +- apps/web/package.json | 1 + package-lock.json | 234 ++++++++++- scripts/smoke_studio_playwright.mjs | 6 + 32 files changed, 2587 insertions(+), 10 deletions(-) create mode 100644 apps/api/app/graph/__init__.py create mode 100644 apps/api/app/graph/compiler.py create mode 100644 apps/api/app/graph/events.py create mode 100644 apps/api/app/graph/executors/__init__.py create mode 100644 apps/api/app/graph/executors/base.py create mode 100644 apps/api/app/graph/executors/batch_ops.py create mode 100644 apps/api/app/graph/executors/image_ops.py create mode 100644 apps/api/app/graph/executors/kie_model.py create mode 100644 apps/api/app/graph/executors/media_load.py create mode 100644 apps/api/app/graph/executors/media_save.py create mode 100644 apps/api/app/graph/executors/prompt_ops.py create mode 100644 apps/api/app/graph/executors/video_ops.py create mode 100644 apps/api/app/graph/media_probe.py create mode 100644 apps/api/app/graph/registry.py create mode 100644 apps/api/app/graph/routes.py create mode 100644 apps/api/app/graph/runtime.py create mode 100644 apps/api/app/graph/schemas.py create mode 100644 apps/api/app/graph/validator.py create mode 100644 apps/api/tests/test_graph_studio.py create mode 100644 apps/web/app/graph-studio/page.tsx create mode 100644 apps/web/components/graph-studio/graph-node.tsx create mode 100644 apps/web/components/graph-studio/graph-studio.tsx create mode 100644 apps/web/components/graph-studio/types.ts diff --git a/apps/api/app/graph/__init__.py b/apps/api/app/graph/__init__.py new file mode 100644 index 0000000..3890630 --- /dev/null +++ b/apps/api/app/graph/__init__.py @@ -0,0 +1,2 @@ +"""Graph Studio backend package.""" + diff --git a/apps/api/app/graph/compiler.py b/apps/api/app/graph/compiler.py new file mode 100644 index 0000000..383fcba --- /dev/null +++ b/apps/api/app/graph/compiler.py @@ -0,0 +1,65 @@ +from __future__ import annotations + +from collections import defaultdict, deque +from typing import Dict, List + +from .registry import registry +from .schemas import GraphCompiledGraph, GraphCompiledNode, GraphWorkflow +from .validator import validate_workflow + + +class GraphCompileError(ValueError): + pass + + +def compile_workflow(workflow: GraphWorkflow) -> GraphCompiledGraph: + validation = validate_workflow(workflow) + if not validation.valid: + raise GraphCompileError("; ".join(error.message for error in validation.errors)) + + definitions = registry.definitions_by_type() + outgoing: Dict[str, List[str]] = defaultdict(list) + indegree: Dict[str, int] = {node.id: 0 for node in workflow.nodes} + input_edges: Dict[str, Dict[str, List[str]]] = {node.id: defaultdict(list) for node in workflow.nodes} + depends_on: Dict[str, set[str]] = {node.id: set() for node in workflow.nodes} + for edge in workflow.edges: + outgoing[edge.source].append(edge.target) + indegree[edge.target] += 1 + input_edges[edge.target][edge.target_port].append(edge.id) + depends_on[edge.target].add(edge.source) + + queue = deque([node_id for node_id, count in indegree.items() if count == 0]) + execution_order: List[str] = [] + while queue: + node_id = queue.popleft() + execution_order.append(node_id) + for target_id in outgoing[node_id]: + indegree[target_id] -= 1 + if indegree[target_id] == 0: + queue.append(target_id) + + node_by_id = {node.id: node for node in workflow.nodes} + compiled_nodes: Dict[str, GraphCompiledNode] = {} + output_node_ids: List[str] = [] + used_types = set() + for node_id in execution_order: + node = node_by_id[node_id] + used_types.add(node.type) + definition = definitions[node.type] + if definition.execution.get("output_node"): + output_node_ids.append(node_id) + compiled_nodes[node_id] = GraphCompiledNode( + node_id=node.id, + node_type=node.type, + depends_on=sorted(depends_on[node.id]), + input_edges={key: list(value) for key, value in input_edges[node.id].items()}, + fields=dict(node.fields), + ) + + return GraphCompiledGraph( + execution_order=execution_order, + nodes=compiled_nodes, + output_node_ids=output_node_ids, + node_definitions={node_type: definitions[node_type] for node_type in sorted(used_types)}, + warnings=validation.warnings, + ) diff --git a/apps/api/app/graph/events.py b/apps/api/app/graph/events.py new file mode 100644 index 0000000..954426c --- /dev/null +++ b/apps/api/app/graph/events.py @@ -0,0 +1,9 @@ +from __future__ import annotations + +from typing import Any, Dict, Optional + +from .. import store + + +def emit(run_id: str, event_type: str, payload: Optional[Dict[str, Any]] = None, node_id: Optional[str] = None) -> Dict[str, Any]: + return store.append_graph_run_event(run_id, event_type, payload or {}, node_id=node_id) diff --git a/apps/api/app/graph/executors/__init__.py b/apps/api/app/graph/executors/__init__.py new file mode 100644 index 0000000..432094c --- /dev/null +++ b/apps/api/app/graph/executors/__init__.py @@ -0,0 +1,2 @@ +"""Graph node executors.""" + diff --git a/apps/api/app/graph/executors/base.py b/apps/api/app/graph/executors/base.py new file mode 100644 index 0000000..6c9eb79 --- /dev/null +++ b/apps/api/app/graph/executors/base.py @@ -0,0 +1,35 @@ +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, Dict, List + +from ..schemas import GraphOutputRef, GraphWorkflow, GraphWorkflowNode + + +@dataclass +class GraphExecutionContext: + run_id: str + workflow: GraphWorkflow + edge_outputs: Dict[str, List[GraphOutputRef]] = field(default_factory=dict) + node_outputs: Dict[str, Dict[str, List[GraphOutputRef]]] = field(default_factory=dict) + + def inputs_for(self, node: GraphWorkflowNode, port_id: str) -> List[GraphOutputRef]: + values: List[GraphOutputRef] = [] + for edge in self.workflow.edges: + if edge.target == node.id and edge.target_port == port_id: + values.extend(self.edge_outputs.get(edge.id, [])) + return values + + def publish_outputs(self, node: GraphWorkflowNode, outputs: Dict[str, List[GraphOutputRef]]) -> None: + self.node_outputs[node.id] = outputs + for edge in self.workflow.edges: + if edge.source == node.id: + self.edge_outputs[edge.id] = outputs.get(edge.source_port, []) + + +class GraphExecutor: + node_type: str + + def execute(self, node: GraphWorkflowNode, context: GraphExecutionContext) -> Dict[str, List[GraphOutputRef]]: + raise NotImplementedError + diff --git a/apps/api/app/graph/executors/batch_ops.py b/apps/api/app/graph/executors/batch_ops.py new file mode 100644 index 0000000..2d2252f --- /dev/null +++ b/apps/api/app/graph/executors/batch_ops.py @@ -0,0 +1 @@ +"""Batch graph executors reserved for the next Graph Studio slice.""" diff --git a/apps/api/app/graph/executors/image_ops.py b/apps/api/app/graph/executors/image_ops.py new file mode 100644 index 0000000..f9be8d8 --- /dev/null +++ b/apps/api/app/graph/executors/image_ops.py @@ -0,0 +1,2 @@ +"""Image utility graph executors reserved for the next Graph Studio slice.""" + diff --git a/apps/api/app/graph/executors/kie_model.py b/apps/api/app/graph/executors/kie_model.py new file mode 100644 index 0000000..d9fe0a8 --- /dev/null +++ b/apps/api/app/graph/executors/kie_model.py @@ -0,0 +1,64 @@ +from __future__ import annotations + +import time +from typing import Dict, List + +from ... import service, store +from ...schemas import MediaRefInput, ValidateRequest +from ..events import emit +from ..registry import registry +from ..schemas import GraphOutputRef, GraphWorkflowNode +from .base import GraphExecutionContext, GraphExecutor + + +def _to_media_ref(value: GraphOutputRef) -> MediaRefInput: + return MediaRefInput(asset_id=value.asset_id, reference_id=value.reference_id) + + +class KieModelExecutor(GraphExecutor): + node_type = "model.kie.nano_banana_pro" + + def execute(self, node: GraphWorkflowNode, context: GraphExecutionContext) -> Dict[str, List[GraphOutputRef]]: + definition = registry.get_definition(node.type) + model_key = str(definition.source.get("model_key") or "nano-banana-pro") + image_refs = context.inputs_for(node, "image_refs") + prompt = str(node.fields.get("prompt") or "").strip() + if not prompt: + raise ValueError("Model node prompt is required.") + option_keys = {field.id for field in definition.fields if field.id != "prompt"} + options = {key: value for key, value in node.fields.items() if key in option_keys and value is not None and value != ""} + request = ValidateRequest( + model_key=model_key, + task_mode="image_edit" if image_refs else "text_to_image", + prompt=prompt, + images=[_to_media_ref(item) for item in image_refs], + options=options, + output_count=1, + ) + emit(context.run_id, "kie.validating", {"model_key": model_key}, node_id=node.id) + service.build_validation_bundle(request) + emit(context.run_id, "kie.submitted", {"model_key": model_key}, node_id=node.id) + batch, jobs = service.submit_jobs(request) + job = jobs[0] + 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 + current = job + while time.time() < deadline: + current = store.get_job(job["job_id"]) or current + if current["status"] in {"completed", "failed", "cancelled"}: + break + runner.tick() + time.sleep(0.25) + current = store.get_job(job["job_id"]) or current + if current["status"] != "completed": + raise ValueError(current.get("error") or f"KIE job did not complete: {current['status']}") + asset = store.get_asset_by_job_id(current["job_id"]) + if not asset: + raise ValueError("KIE job completed without creating an asset.") + return { + "image": [GraphOutputRef(kind="asset", media_type="image", asset_id=asset["asset_id"], job_id=current["job_id"])], + "job": [GraphOutputRef(kind="job", job_id=current["job_id"], metadata={"batch_id": batch["batch_id"]})], + } diff --git a/apps/api/app/graph/executors/media_load.py b/apps/api/app/graph/executors/media_load.py new file mode 100644 index 0000000..58f5581 --- /dev/null +++ b/apps/api/app/graph/executors/media_load.py @@ -0,0 +1,45 @@ +from __future__ import annotations + +from typing import Dict, List + +from ... import store +from ..schemas import GraphOutputRef, GraphWorkflowNode +from .base import GraphExecutionContext, GraphExecutor + + +class LoadImageExecutor(GraphExecutor): + node_type = "media.load_image" + + def execute(self, node: GraphWorkflowNode, context: GraphExecutionContext) -> Dict[str, List[GraphOutputRef]]: + asset_id = node.fields.get("asset_id") + reference_id = node.fields.get("reference_id") + if asset_id: + asset = store.get_asset(str(asset_id)) + if not asset: + raise ValueError("Load Image asset does not exist.") + return { + "image": [ + GraphOutputRef( + kind="asset", + media_type="image", + asset_id=str(asset_id), + metadata={"model_key": asset.get("model_key")}, + ) + ] + } + if reference_id: + reference = store.get_reference_media(str(reference_id)) + if not reference: + raise ValueError("Load Image reference media does not exist.") + return { + "image": [ + GraphOutputRef( + kind="reference_media", + media_type="image", + reference_id=str(reference_id), + metadata={"stored_path": reference.get("stored_path")}, + ) + ] + } + raise ValueError("Load Image requires asset_id or reference_id.") + diff --git a/apps/api/app/graph/executors/media_save.py b/apps/api/app/graph/executors/media_save.py new file mode 100644 index 0000000..2e1c131 --- /dev/null +++ b/apps/api/app/graph/executors/media_save.py @@ -0,0 +1,21 @@ +from __future__ import annotations + +from typing import Dict, List + +from ..events import emit +from ..schemas import GraphOutputRef, GraphWorkflowNode +from .base import GraphExecutionContext, GraphExecutor + + +class SaveImageExecutor(GraphExecutor): + node_type = "media.save_image" + + def execute(self, node: GraphWorkflowNode, context: GraphExecutionContext) -> Dict[str, List[GraphOutputRef]]: + image_refs = context.inputs_for(node, "image") + if not image_refs: + raise ValueError("Save Image requires an image input.") + first = image_refs[0] + if first.asset_id: + emit(context.run_id, "asset.created", {"asset_id": first.asset_id}, node_id=node.id) + return {"asset": [first], "image": [first]} + diff --git a/apps/api/app/graph/executors/prompt_ops.py b/apps/api/app/graph/executors/prompt_ops.py new file mode 100644 index 0000000..d3a2cd1 --- /dev/null +++ b/apps/api/app/graph/executors/prompt_ops.py @@ -0,0 +1,2 @@ +"""Prompt utility graph executors reserved for the next Graph Studio slice.""" + diff --git a/apps/api/app/graph/executors/video_ops.py b/apps/api/app/graph/executors/video_ops.py new file mode 100644 index 0000000..e121915 --- /dev/null +++ b/apps/api/app/graph/executors/video_ops.py @@ -0,0 +1,2 @@ +"""Video utility graph executors reserved for the next Graph Studio slice.""" + diff --git a/apps/api/app/graph/media_probe.py b/apps/api/app/graph/media_probe.py new file mode 100644 index 0000000..b0d2cf8 --- /dev/null +++ b/apps/api/app/graph/media_probe.py @@ -0,0 +1,2 @@ +"""Media probing helpers reserved for future graph utility nodes.""" + diff --git a/apps/api/app/graph/registry.py b/apps/api/app/graph/registry.py new file mode 100644 index 0000000..577d48d --- /dev/null +++ b/apps/api/app/graph/registry.py @@ -0,0 +1,183 @@ +from __future__ import annotations + +from typing import Any, Dict, List, Optional + +from .. import kie_adapter, store +from .schemas import GraphNodeDefinition, GraphNodeField, GraphNodePort + + +def _title_from_key(value: str) -> str: + return " ".join(part.capitalize() for part in value.replace("_", " ").replace("-", " ").split()) + + +def _field_from_option(key: str, spec: Dict[str, Any]) -> Optional[GraphNodeField]: + if spec.get("hidden_from_studio"): + return None + option_type = str(spec.get("type") or "text") + field_type = "text" + if option_type == "enum": + field_type = "select" + elif option_type == "bool": + field_type = "boolean" + elif option_type == "int_range": + field_type = "integer" + elif option_type == "float_range": + field_type = "float" + label = spec.get("label") or _title_from_key(key) + return GraphNodeField( + id=key, + label=str(label), + type=field_type, + required=bool(spec.get("required")), + default=spec.get("default"), + options=list(spec.get("allowed") or []), + min=spec.get("min"), + max=spec.get("max"), + help_text=spec.get("help_text") or spec.get("notes"), + advanced=bool(spec.get("advanced")), + ) + + +class GraphNodeRegistry: + def __init__(self) -> None: + self._definitions: Optional[List[GraphNodeDefinition]] = None + + def list_definitions(self, *, refresh: bool = False) -> List[GraphNodeDefinition]: + if self._definitions is None or refresh: + self._definitions = self._build_definitions() + try: + fingerprint = kie_adapter.model_diagnostics().get("kie_spec_version") or "unknown" + store.cache_graph_node_definitions( + str(fingerprint), + [item.model_dump(mode="json") for item in self._definitions], + ) + except Exception: + pass + return list(self._definitions) + + def get_definition(self, node_type: str) -> GraphNodeDefinition: + for definition in self.list_definitions(): + if definition.type == node_type: + return definition + raise KeyError(node_type) + + def definitions_by_type(self) -> Dict[str, GraphNodeDefinition]: + return {definition.type: definition for definition in self.list_definitions()} + + def _build_definitions(self) -> List[GraphNodeDefinition]: + definitions = [ + GraphNodeDefinition( + type="media.load_image", + title="Load Image", + description="Load an existing Media Studio asset or reference image.", + category="Media", + search_aliases=["asset", "reference", "input", "image"], + tags=["media", "image", "input"], + source={"kind": "system"}, + execution={"executor": "media.load_image", "mode": "sync", "cacheable": True, "output_node": False}, + ui={"default_size": {"width": 280, "height": 260}, "accent": "green", "icon": "image"}, + ports={ + "inputs": [], + "outputs": [GraphNodePort(id="image", label="Image", type="image")], + }, + fields=[ + GraphNodeField(id="asset_id", label="Asset ID", type="asset_picker", required=False), + GraphNodeField(id="reference_id", label="Reference ID", type="reference_media_picker", required=False), + ], + ), + GraphNodeDefinition( + type="media.save_image", + title="Save Image", + description="Expose an image as a normal Media Studio graph output.", + category="Media", + search_aliases=["save", "output", "asset"], + tags=["media", "image", "output"], + source={"kind": "system"}, + execution={"executor": "media.save_image", "mode": "sync", "cacheable": False, "output_node": True}, + ui={"default_size": {"width": 260, "height": 180}, "accent": "yellow", "icon": "save"}, + ports={ + "inputs": [GraphNodePort(id="image", label="Image", type="image", required=True, min=1, max=1, accepts=["image"])], + "outputs": [GraphNodePort(id="asset", label="Asset", type="asset")], + }, + fields=[GraphNodeField(id="label", label="Label", type="text", required=False)], + ), + ] + model = self._nano_banana_pro_model() + if model: + definitions.append(self._kie_model_definition(model)) + return definitions + + def _nano_banana_pro_model(self) -> Optional[Dict[str, Any]]: + models = kie_adapter.list_models() + for key in ("nano-banana-pro", "nanobanana-pro", "nano_banana_pro"): + match = next((item for item in models if item.get("key") == key), None) + if match: + return match + return next( + ( + item + for item in models + if "nano" in str(item.get("key") or "").lower() + and "pro" in str(item.get("key") or "").lower() + and "image" in (item.get("media_types") or ["image"]) + ), + None, + ) + + def _kie_model_definition(self, model: Dict[str, Any]) -> GraphNodeDefinition: + raw = model.get("raw") or {} + image_inputs = (raw.get("inputs") or {}).get("image") or {} + max_images = int(image_inputs.get("required_max") or 0) + fields = [ + GraphNodeField( + id="prompt", + label="Prompt", + type="textarea", + required=True, + default="", + placeholder="Describe the image to generate or edit...", + connectable=True, + port_type="text", + ) + ] + for key, option in (raw.get("options") or {}).items(): + field = _field_from_option(str(key), option if isinstance(option, dict) else {}) + if field: + fields.append(field) + return GraphNodeDefinition( + type="model.kie.nano_banana_pro", + title=str(model.get("label") or "Nano Banana Pro"), + description="KIE image model node for prompt-guided image generation and editing.", + category="Models/Image", + search_aliases=["nano", "banana", "pro", "image", "edit", "generation"], + tags=["model", "image", "kie"], + source={ + "kind": "kie_model", + "model_key": model.get("key"), + "kie_spec_version": model.get("kie_spec_version"), + }, + execution={"executor": "kie.model", "mode": "async", "cacheable": True, "output_node": False, "retryable": True}, + ui={"default_size": {"width": 360, "height": 520}, "accent": "blue", "icon": "sparkles", "show_preview": True}, + ports={ + "inputs": [ + GraphNodePort( + id="image_refs", + label="Reference Images", + type="image", + array=True, + min=int(image_inputs.get("required_min") or 0), + max=max_images or None, + required=bool(image_inputs.get("required_min") or 0), + accepts=["image"], + ) + ], + "outputs": [ + GraphNodePort(id="image", label="Image", type="image"), + GraphNodePort(id="job", label="Job", type="job", advanced=True), + ], + }, + fields=fields, + ) + + +registry = GraphNodeRegistry() diff --git a/apps/api/app/graph/routes.py b/apps/api/app/graph/routes.py new file mode 100644 index 0000000..2257ba8 --- /dev/null +++ b/apps/api/app/graph/routes.py @@ -0,0 +1,205 @@ +from __future__ import annotations + +from typing import List, Optional + +from fastapi import APIRouter, HTTPException, Query + +from .. import store +from .registry import registry +from .runtime import runtime +from .schemas import ( + GraphNodeDefinition, + GraphNodeDefinitionsResponse, + GraphRun, + GraphRunCreateRequest, + GraphRunEventsResponse, + GraphRunListResponse, + GraphTemplate, + GraphTemplateListResponse, + GraphTemplateRecord, + GraphValidationResult, + GraphWorkflow, + GraphWorkflowListResponse, + GraphWorkflowRecord, +) +from .validator import validate_workflow + +router = APIRouter(prefix="/media/graph", tags=["media-graph"]) + + +def _not_found(name: str) -> HTTPException: + return HTTPException(status_code=404, detail=f"{name} not found") + + +def _bad_request(message: str) -> HTTPException: + return HTTPException(status_code=400, detail=message) + + +def _workflow_from_record(record: dict) -> GraphWorkflow: + workflow_json = dict(record.get("workflow_json") or {}) + workflow_json.setdefault("workflow_id", record["workflow_id"]) + workflow_json.setdefault("name", record.get("name") or "Untitled Graph") + workflow_json.setdefault("description", record.get("description")) + return GraphWorkflow(**workflow_json) + + +def _shape_run(record: dict) -> GraphRun: + return GraphRun( + **record, + nodes=[item for item in (runtime._shape_run(record).nodes)], + ) + + +@router.get("/node-definitions", response_model=GraphNodeDefinitionsResponse) +def list_node_definitions() -> GraphNodeDefinitionsResponse: + return GraphNodeDefinitionsResponse(items=registry.list_definitions()) + + +@router.get("/node-definitions/{node_type:path}", response_model=GraphNodeDefinition) +def get_node_definition(node_type: str) -> GraphNodeDefinition: + try: + return registry.get_definition(node_type) + except KeyError: + raise _not_found("node definition") + + +@router.post("/node-definitions/refresh", response_model=GraphNodeDefinitionsResponse) +def refresh_node_definitions() -> GraphNodeDefinitionsResponse: + return GraphNodeDefinitionsResponse(items=registry.list_definitions(refresh=True)) + + +@router.get("/workflows", response_model=GraphWorkflowListResponse) +def list_workflows() -> GraphWorkflowListResponse: + return GraphWorkflowListResponse(items=[GraphWorkflowRecord(**item) for item in store.list_graph_workflows()]) + + +@router.post("/workflows", response_model=GraphWorkflowRecord) +def create_workflow(payload: GraphWorkflow) -> GraphWorkflowRecord: + record = store.create_or_update_graph_workflow( + { + "workflow_id": payload.workflow_id, + "name": payload.name, + "description": payload.description, + "schema_version": payload.schema_version, + "workflow_json": payload.model_dump(mode="json"), + } + ) + return GraphWorkflowRecord(**record) + + +@router.get("/workflows/{workflow_id}", response_model=GraphWorkflowRecord) +def get_workflow(workflow_id: str) -> GraphWorkflowRecord: + record = store.get_graph_workflow(workflow_id) + if not record: + raise _not_found("workflow") + return GraphWorkflowRecord(**record) + + +@router.patch("/workflows/{workflow_id}", response_model=GraphWorkflowRecord) +def update_workflow(workflow_id: str, payload: GraphWorkflow) -> GraphWorkflowRecord: + current = store.get_graph_workflow(workflow_id) + if not current: + raise _not_found("workflow") + record = store.create_or_update_graph_workflow( + { + **current, + "name": payload.name, + "description": payload.description, + "schema_version": payload.schema_version, + "workflow_json": payload.model_dump(mode="json"), + } + ) + return GraphWorkflowRecord(**record) + + +@router.delete("/workflows/{workflow_id}", response_model=GraphWorkflowRecord) +def delete_workflow(workflow_id: str) -> GraphWorkflowRecord: + try: + return GraphWorkflowRecord(**store.archive_graph_workflow(workflow_id)) + except KeyError: + raise _not_found("workflow") + + +@router.post("/workflows/{workflow_id}/validate", response_model=GraphValidationResult) +def validate_saved_workflow(workflow_id: str, payload: Optional[GraphWorkflow] = None) -> GraphValidationResult: + record = store.get_graph_workflow(workflow_id) + if not record: + raise _not_found("workflow") + return validate_workflow(payload or _workflow_from_record(record)) + + +@router.post("/workflows/{workflow_id}/runs", response_model=GraphRun) +def create_run(workflow_id: str, payload: Optional[GraphRunCreateRequest] = None) -> GraphRun: + record = store.get_graph_workflow(workflow_id) + if not record: + raise _not_found("workflow") + workflow = payload.workflow if payload and payload.workflow else _workflow_from_record(record) + try: + return runtime.create_run(workflow_id, workflow, start=True) + except ValueError as exc: + raise _bad_request(str(exc)) + + +@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/{run_id}", response_model=GraphRun) +def get_run(run_id: str) -> GraphRun: + record = store.get_graph_run(run_id) + if not record: + raise _not_found("graph run") + return _shape_run(record) + + +@router.get("/runs/{run_id}/events", response_model=GraphRunEventsResponse) +def list_run_events(run_id: str, after_event_id: Optional[str] = Query(default=None)) -> GraphRunEventsResponse: + if not store.get_graph_run(run_id): + raise _not_found("graph run") + return GraphRunEventsResponse(items=store.list_graph_run_events(run_id, after_event_id=after_event_id)) + + +@router.post("/runs/{run_id}/cancel", response_model=GraphRun) +def cancel_run(run_id: str) -> GraphRun: + if not store.get_graph_run(run_id): + raise _not_found("graph run") + return runtime.cancel_run(run_id) + + +@router.get("/templates", response_model=GraphTemplateListResponse) +def list_templates() -> GraphTemplateListResponse: + return GraphTemplateListResponse(items=[GraphTemplateRecord(**item) for item in store.list_graph_templates()]) + + +@router.post("/templates", response_model=GraphTemplateRecord) +def create_template(payload: GraphTemplate) -> GraphTemplateRecord: + record = store.create_or_update_graph_template( + { + "template_id": payload.template_id, + "name": payload.name, + "description": payload.description, + "tags_json": payload.tags, + "thumbnail_path": payload.thumbnail_path, + "workflow_json": payload.workflow_json, + } + ) + return GraphTemplateRecord(**record) + + +@router.post("/templates/{template_id}/instantiate", response_model=GraphWorkflowRecord) +def instantiate_template(template_id: str) -> GraphWorkflowRecord: + template = store.get_graph_template(template_id) + if not template: + raise _not_found("template") + workflow = dict(template.get("workflow_json") or {}) + workflow.pop("workflow_id", None) + workflow.setdefault("name", template.get("name") or "Template Workflow") + record = store.create_or_update_graph_workflow( + { + "name": workflow.get("name") or template.get("name") or "Template Workflow", + "description": template.get("description"), + "workflow_json": workflow, + } + ) + return GraphWorkflowRecord(**record) diff --git a/apps/api/app/graph/runtime.py b/apps/api/app/graph/runtime.py new file mode 100644 index 0000000..7820d10 --- /dev/null +++ b/apps/api/app/graph/runtime.py @@ -0,0 +1,125 @@ +from __future__ import annotations + +import logging +import threading +from typing import Dict, List + +from .. import store +from .compiler import compile_workflow +from .events import emit +from .executors.base import GraphExecutionContext, GraphExecutor +from .executors.kie_model import KieModelExecutor +from .executors.media_load import LoadImageExecutor +from .executors.media_save import SaveImageExecutor +from .schemas import GraphRun, GraphRunNode, GraphWorkflow +from .validator import validate_workflow + +logger = logging.getLogger(__name__) + + +class GraphRuntime: + def __init__(self) -> None: + executors: List[GraphExecutor] = [ + LoadImageExecutor(), + KieModelExecutor(), + SaveImageExecutor(), + ] + self.executors: Dict[str, GraphExecutor] = {executor.node_type: executor for executor in executors} + + def create_run(self, workflow_id: str, workflow: GraphWorkflow, *, start: bool = True) -> GraphRun: + emit_payload = workflow.model_dump(mode="json") + compiled = compile_workflow(workflow) + run = store.create_graph_run( + { + "workflow_id": workflow_id, + "status": "queued", + "schema_version": workflow.schema_version, + "workflow_json": emit_payload, + "compiled_graph_json": compiled.model_dump(mode="json"), + }, + [ + { + "node_id": node.id, + "node_type": node.type, + "status": "queued", + "input_snapshot_json": node.fields, + } + for node in workflow.nodes + ], + ) + emit(run["run_id"], "run.created", {"workflow_id": workflow_id}) + if start: + thread = threading.Thread(target=self.execute_run, args=(run["run_id"],), name=f"graph-run-{run['run_id']}", daemon=True) + thread.start() + return self._shape_run(run) + + def execute_run(self, run_id: str) -> None: + run = store.get_graph_run(run_id) + if not run: + return + try: + workflow = GraphWorkflow(**run["workflow_json"]) + emit(run_id, "run.validating") + validation = validate_workflow(workflow) + if not validation.valid: + raise ValueError("; ".join(error.message for error in validation.errors)) + compiled = compile_workflow(workflow) + store.update_graph_run(run_id, {"status": "running", "started_at": store.utcnow_iso(), "compiled_graph_json": compiled.model_dump(mode="json")}) + emit(run_id, "run.compiled", {"node_count": len(workflow.nodes), "edge_count": len(workflow.edges)}) + emit(run_id, "run.started") + context = GraphExecutionContext(run_id=run_id, workflow=workflow) + nodes_by_id = {node.id: node for node in workflow.nodes} + for node_id in compiled.execution_order: + node = nodes_by_id[node_id] + executor = self.executors.get(node.type) + if not executor: + raise ValueError(f"No executor for node type: {node.type}") + emit(run_id, "node.queued", node_id=node.id) + store.update_graph_run_node(run_id, node.id, {"status": "running", "started_at": store.utcnow_iso(), "progress": 0.1}) + emit(run_id, "node.started", {"node_type": node.type}, node_id=node.id) + outputs = executor.execute(node, context) + context.publish_outputs(node, outputs) + output_payload = {key: [item.model_dump(mode="json") for item in value] for key, value in outputs.items()} + store.update_graph_run_node( + run_id, + node.id, + { + "status": "completed", + "progress": 1, + "output_snapshot_json": output_payload, + "finished_at": store.utcnow_iso(), + }, + ) + emit(run_id, "node.completed", output_payload, node_id=node.id) + output_snapshot = { + node_id: {port: [item.model_dump(mode="json") for item in refs] for port, refs in outputs.items()} + for node_id, outputs in context.node_outputs.items() + } + store.update_graph_run( + run_id, + { + "status": "completed", + "output_snapshot_json": output_snapshot, + "finished_at": store.utcnow_iso(), + }, + ) + emit(run_id, "run.completed", {"outputs": output_snapshot}) + except Exception as exc: + logger.exception("graph run failed", extra={"run_id": run_id}) + store.update_graph_run(run_id, {"status": "failed", "error": str(exc), "finished_at": store.utcnow_iso()}) + emit(run_id, "run.failed", {"error": str(exc)}) + + def cancel_run(self, run_id: str) -> GraphRun: + run = store.update_graph_run(run_id, {"status": "cancelled", "finished_at": store.utcnow_iso()}) + emit(run_id, "run.cancelled") + return self._shape_run(run) + + def _shape_run(self, record: Dict) -> GraphRun: + return GraphRun( + **record, + nodes=[GraphRunNode(**item) for item in store.list_graph_run_nodes(record["run_id"])], + ) + + +runtime = GraphRuntime() + diff --git a/apps/api/app/graph/schemas.py b/apps/api/app/graph/schemas.py new file mode 100644 index 0000000..7a9663d --- /dev/null +++ b/apps/api/app/graph/schemas.py @@ -0,0 +1,214 @@ +from __future__ import annotations + +from typing import Any, Dict, List, Literal, Optional + +from pydantic import BaseModel, Field + + +class GraphNodePort(BaseModel): + id: str + label: str + type: str + array: bool = False + min: int = 0 + max: Optional[int] = None + required: bool = False + accepts: List[str] = Field(default_factory=list) + description: Optional[str] = None + advanced: bool = False + + +class GraphNodeField(BaseModel): + id: str + label: str + type: str + required: bool = False + default: Any = None + placeholder: Optional[str] = None + options: List[Any] = Field(default_factory=list) + min: Optional[float] = None + max: Optional[float] = None + help_text: Optional[str] = None + advanced: bool = False + hidden: bool = False + connectable: bool = False + port_type: Optional[str] = None + + +class GraphNodeDefinition(BaseModel): + schema_version: int = 1 + type: str + title: str + description: Optional[str] = None + category: str + search_aliases: List[str] = Field(default_factory=list) + tags: List[str] = Field(default_factory=list) + source: Dict[str, Any] = Field(default_factory=dict) + execution: Dict[str, Any] = Field(default_factory=dict) + ui: Dict[str, Any] = Field(default_factory=dict) + ports: Dict[str, List[GraphNodePort]] = Field(default_factory=lambda: {"inputs": [], "outputs": []}) + fields: List[GraphNodeField] = Field(default_factory=list) + + +class GraphWorkflowNode(BaseModel): + id: str + type: str + position: Dict[str, float] = Field(default_factory=lambda: {"x": 0, "y": 0}) + fields: Dict[str, Any] = Field(default_factory=dict) + metadata: Dict[str, Any] = Field(default_factory=dict) + + +class GraphWorkflowEdge(BaseModel): + id: str + source: str + source_port: str + target: str + target_port: str + metadata: Dict[str, Any] = Field(default_factory=dict) + + +class GraphWorkflow(BaseModel): + schema_version: int = 1 + workflow_id: Optional[str] = None + name: str = "Untitled Graph" + description: Optional[str] = None + nodes: List[GraphWorkflowNode] = Field(default_factory=list) + edges: List[GraphWorkflowEdge] = Field(default_factory=list) + viewport: Dict[str, Any] = Field(default_factory=dict) + metadata: Dict[str, Any] = Field(default_factory=dict) + + +class GraphWorkflowRecord(BaseModel): + workflow_id: str + name: str + description: Optional[str] = None + status: str = "active" + schema_version: int = 1 + workflow_json: Dict[str, Any] = Field(default_factory=dict) + created_at: Optional[str] = None + updated_at: Optional[str] = None + + +class GraphTemplate(BaseModel): + template_id: Optional[str] = None + name: str + description: Optional[str] = None + tags: List[str] = Field(default_factory=list) + thumbnail_path: Optional[str] = None + workflow_json: Dict[str, Any] = Field(default_factory=dict) + + +class GraphTemplateRecord(GraphTemplate): + template_id: str + status: str = "active" + created_at: Optional[str] = None + updated_at: Optional[str] = None + + +class GraphOutputRef(BaseModel): + kind: Literal["asset", "reference_media", "job", "value"] + media_type: Optional[str] = None + asset_id: Optional[str] = None + reference_id: Optional[str] = None + job_id: Optional[str] = None + value: Any = None + metadata: Dict[str, Any] = Field(default_factory=dict) + + +class GraphError(BaseModel): + code: str + message: str + node_id: Optional[str] = None + edge_id: Optional[str] = None + field_id: Optional[str] = None + port_id: Optional[str] = None + + +class GraphValidationResult(BaseModel): + valid: bool + errors: List[GraphError] = Field(default_factory=list) + warnings: List[GraphError] = Field(default_factory=list) + + +class GraphCompiledNode(BaseModel): + node_id: str + node_type: str + depends_on: List[str] = Field(default_factory=list) + input_edges: Dict[str, List[str]] = Field(default_factory=dict) + fields: Dict[str, Any] = Field(default_factory=dict) + + +class GraphCompiledGraph(BaseModel): + schema_version: int = 1 + execution_order: List[str] = Field(default_factory=list) + nodes: Dict[str, GraphCompiledNode] = Field(default_factory=dict) + output_node_ids: List[str] = Field(default_factory=list) + node_definitions: Dict[str, GraphNodeDefinition] = Field(default_factory=dict) + warnings: List[GraphError] = Field(default_factory=list) + + +class GraphRun(BaseModel): + run_id: str + workflow_id: str + status: str = "queued" + schema_version: int = 1 + workflow_json: Dict[str, Any] = Field(default_factory=dict) + compiled_graph_json: Dict[str, Any] = Field(default_factory=dict) + output_snapshot_json: Dict[str, Any] = Field(default_factory=dict) + error: Optional[str] = None + nodes: List["GraphRunNode"] = Field(default_factory=list) + 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 + node_id: str + node_type: str + status: str = "queued" + progress: Optional[float] = None + input_snapshot_json: Dict[str, Any] = Field(default_factory=dict) + output_snapshot_json: Dict[str, Any] = Field(default_factory=dict) + error: Optional[str] = None + started_at: Optional[str] = None + finished_at: Optional[str] = None + updated_at: Optional[str] = None + + +class GraphRunEvent(BaseModel): + event_id: str + run_id: str + node_id: Optional[str] = None + event_type: str + payload_json: Dict[str, Any] = Field(default_factory=dict) + created_at: Optional[str] = None + + +class GraphRunCreateRequest(BaseModel): + workflow: Optional[GraphWorkflow] = None + + +class GraphNodeDefinitionsResponse(BaseModel): + items: List[GraphNodeDefinition] = Field(default_factory=list) + + +class GraphWorkflowListResponse(BaseModel): + items: List[GraphWorkflowRecord] = Field(default_factory=list) + + +class GraphRunListResponse(BaseModel): + items: List[GraphRun] = Field(default_factory=list) + + +class GraphRunEventsResponse(BaseModel): + items: List[GraphRunEvent] = Field(default_factory=list) + + +class GraphTemplateListResponse(BaseModel): + items: List[GraphTemplateRecord] = Field(default_factory=list) + + +GraphRun.model_rebuild() diff --git a/apps/api/app/graph/validator.py b/apps/api/app/graph/validator.py new file mode 100644 index 0000000..38cb66f --- /dev/null +++ b/apps/api/app/graph/validator.py @@ -0,0 +1,107 @@ +from __future__ import annotations + +from collections import defaultdict, deque +from typing import Dict, List, Set + +from .. import store +from .registry import registry +from .schemas import GraphError, GraphValidationResult, GraphWorkflow, GraphWorkflowEdge, GraphWorkflowNode + + +def _port_map(definition, direction: str) -> Dict[str, object]: + return {port.id: port for port in definition.ports.get(direction, [])} + + +def validate_workflow(workflow: GraphWorkflow) -> GraphValidationResult: + definitions = registry.definitions_by_type() + errors: List[GraphError] = [] + warnings: List[GraphError] = [] + + node_ids: Set[str] = set() + nodes_by_id: Dict[str, GraphWorkflowNode] = {} + for node in workflow.nodes: + if node.id in node_ids: + errors.append(GraphError(code="duplicate_node_id", message=f"Duplicate node id: {node.id}", node_id=node.id)) + continue + node_ids.add(node.id) + nodes_by_id[node.id] = node + definition = definitions.get(node.type) + if not definition: + errors.append(GraphError(code="missing_node_type", message=f"Unknown node type: {node.type}", node_id=node.id)) + continue + for field in definition.fields: + if field.required and field.id not in node.fields: + errors.append( + GraphError(code="missing_required_field", message=f"Missing required field: {field.label}", node_id=node.id, field_id=field.id) + ) + if node.type == "media.load_image" and not node.fields.get("asset_id") and not node.fields.get("reference_id"): + errors.append(GraphError(code="missing_media_reference", message="Load Image needs an asset or reference image.", node_id=node.id)) + if node.fields.get("asset_id") and not store.get_asset(str(node.fields["asset_id"])): + errors.append(GraphError(code="missing_asset", message="Referenced asset does not exist.", node_id=node.id, field_id="asset_id")) + 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")) + + edge_ids: Set[str] = set() + incoming_by_target_port: Dict[tuple[str, str], int] = defaultdict(int) + outgoing: Dict[str, List[str]] = defaultdict(list) + indegree: Dict[str, int] = {node.id: 0 for node in workflow.nodes} + for edge in workflow.edges: + if edge.id in edge_ids: + errors.append(GraphError(code="duplicate_edge_id", message=f"Duplicate edge id: {edge.id}", edge_id=edge.id)) + continue + edge_ids.add(edge.id) + source = nodes_by_id.get(edge.source) + target = nodes_by_id.get(edge.target) + if not source or not target: + errors.append(GraphError(code="missing_edge_node", message="Edge references a missing node.", edge_id=edge.id)) + continue + source_def = definitions.get(source.type) + target_def = definitions.get(target.type) + if not source_def or not target_def: + continue + source_port = _port_map(source_def, "outputs").get(edge.source_port) + target_port = _port_map(target_def, "inputs").get(edge.target_port) + if not source_port: + errors.append(GraphError(code="missing_source_port", message=f"Unknown source port: {edge.source_port}", edge_id=edge.id, port_id=edge.source_port)) + continue + if not target_port: + errors.append(GraphError(code="missing_target_port", message=f"Unknown target port: {edge.target_port}", edge_id=edge.id, port_id=edge.target_port)) + continue + source_type = getattr(source_port, "type", "") + accepted = getattr(target_port, "accepts", None) or [getattr(target_port, "type", "")] + if source_type not in accepted: + errors.append(GraphError(code="incompatible_edge", message=f"Cannot connect {source_type} to {getattr(target_port, 'type', '')}.", edge_id=edge.id)) + key = (edge.target, edge.target_port) + incoming_by_target_port[key] += 1 + max_count = getattr(target_port, "max", None) + if max_count is not None and incoming_by_target_port[key] > max_count: + errors.append(GraphError(code="input_cardinality_exceeded", message="Too many edges connected to input.", edge_id=edge.id, port_id=edge.target_port)) + outgoing[edge.source].append(edge.target) + indegree[edge.target] = indegree.get(edge.target, 0) + 1 + + for node in workflow.nodes: + definition = definitions.get(node.type) + if not definition: + continue + for port in definition.ports.get("inputs", []): + if port.required and incoming_by_target_port[(node.id, port.id)] < max(1, port.min): + errors.append(GraphError(code="missing_required_input", message=f"Missing required input: {port.label}", node_id=node.id, port_id=port.id)) + + visited_count = 0 + queue = deque([node_id for node_id, count in indegree.items() if count == 0]) + while queue: + node_id = queue.popleft() + visited_count += 1 + for target_id in outgoing[node_id]: + indegree[target_id] -= 1 + if indegree[target_id] == 0: + queue.append(target_id) + if workflow.nodes and visited_count != len(workflow.nodes): + errors.append(GraphError(code="cycle_detected", message="Workflow contains a cycle.")) + + connected_node_ids = {edge.source for edge in workflow.edges} | {edge.target for edge in workflow.edges} + for node in workflow.nodes: + if len(workflow.nodes) > 1 and node.id not in connected_node_ids: + warnings.append(GraphError(code="disconnected_node", message="Node is disconnected.", node_id=node.id)) + + return GraphValidationResult(valid=not errors, errors=errors, warnings=warnings) diff --git a/apps/api/app/main.py b/apps/api/app/main.py index c1f164d..558423e 100644 --- a/apps/api/app/main.py +++ b/apps/api/app/main.py @@ -13,6 +13,7 @@ from . import kie_adapter, service, store from .control_auth import validate_control_request +from .graph.routes import router as graph_router from .runner import runner from .schemas import ( AssetListResponse, @@ -102,6 +103,7 @@ async def lifespan(_: FastAPI): app = FastAPI(title=settings.app_name, lifespan=lifespan) +app.include_router(graph_router) settings.data_root.mkdir(parents=True, exist_ok=True) diff --git a/apps/api/app/store.py b/apps/api/app/store.py index 7b53672..b1cdcfd 100644 --- a/apps/api/app/store.py +++ b/apps/api/app/store.py @@ -434,6 +434,210 @@ def delete_system_prompt(prompt_id: str) -> None: _delete_table("media_system_prompts", "prompt_id", prompt_id) +def list_graph_workflows() -> List[Dict[str, Any]]: + with get_connection() as connection: + rows = connection.execute( + "SELECT * FROM graph_workflows WHERE status != 'archived' ORDER BY updated_at DESC, name ASC" + ).fetchall() + return [_decode_row(row) for row in rows] + + +def get_graph_workflow(workflow_id: str) -> Optional[Dict[str, Any]]: + return _get_table("graph_workflows", "workflow_id", workflow_id) + + +def create_or_update_graph_workflow(payload: Dict[str, Any]) -> Dict[str, Any]: + payload = payload.copy() + if not payload.get("workflow_id"): + payload["workflow_id"] = new_id("graphwf") + payload.setdefault("schema_version", 1) + payload.setdefault("status", "active") + record = _upsert_table("graph_workflows", "workflow_id", payload) + version_count = count_graph_workflow_versions(record["workflow_id"]) + create_graph_workflow_version( + { + "workflow_id": record["workflow_id"], + "version_number": version_count + 1, + "workflow_json": record.get("workflow_json") or {}, + } + ) + return record + + +def archive_graph_workflow(workflow_id: str) -> Dict[str, Any]: + record = get_graph_workflow(workflow_id) + if record is None: + raise KeyError("workflow not found") + record["status"] = "archived" + return create_or_update_graph_workflow(record) + + +def count_graph_workflow_versions(workflow_id: str) -> int: + with get_connection() as connection: + row = connection.execute( + "SELECT COUNT(*) AS count FROM graph_workflow_versions WHERE workflow_id = ?", + (workflow_id,), + ).fetchone() + return int(row["count"] if row else 0) + + +def create_graph_workflow_version(payload: Dict[str, Any]) -> Dict[str, Any]: + payload = payload.copy() + payload.setdefault("version_id", new_id("graphver")) + payload.setdefault("created_at", utcnow_iso()) + return _upsert_table("graph_workflow_versions", "version_id", payload) + + +def list_graph_templates() -> List[Dict[str, Any]]: + with get_connection() as connection: + rows = connection.execute( + "SELECT * FROM graph_templates WHERE status != 'archived' ORDER BY updated_at DESC, name ASC" + ).fetchall() + return [_decode_row(row) for row in rows] + + +def get_graph_template(template_id: str) -> Optional[Dict[str, Any]]: + return _get_table("graph_templates", "template_id", template_id) + + +def create_or_update_graph_template(payload: Dict[str, Any]) -> Dict[str, Any]: + payload = payload.copy() + if not payload.get("template_id"): + payload["template_id"] = new_id("graphtpl") + payload.setdefault("status", "active") + payload.setdefault("tags_json", []) + return _upsert_table("graph_templates", "template_id", payload) + + +def create_graph_run(payload: Dict[str, Any], node_payloads: List[Dict[str, Any]]) -> Dict[str, Any]: + now = utcnow_iso() + run = payload.copy() + run.setdefault("run_id", new_id("grun")) + run.setdefault("status", "queued") + run.setdefault("schema_version", 1) + run.setdefault("created_at", now) + run["updated_at"] = now + run = _upsert_table("graph_runs", "run_id", run) + with get_connection() as connection: + for item in node_payloads: + node = item.copy() + node.setdefault("run_node_id", new_id("grnode")) + node["run_id"] = run["run_id"] + node.setdefault("status", "queued") + node.setdefault("input_snapshot_json", {}) + node.setdefault("output_snapshot_json", {}) + node["updated_at"] = now + _insert_or_update(connection, "graph_run_nodes", "run_node_id", node) + return get_graph_run(run["run_id"]) # type: ignore + + +def list_graph_runs(limit: int = 100) -> List[Dict[str, Any]]: + with get_connection() as connection: + rows = connection.execute( + "SELECT * FROM graph_runs ORDER BY created_at DESC LIMIT ?", + (limit,), + ).fetchall() + return [_decode_row(row) for row in rows] + + +def get_graph_run(run_id: str) -> Optional[Dict[str, Any]]: + return _get_table("graph_runs", "run_id", run_id) + + +def update_graph_run(run_id: str, payload: Dict[str, Any]) -> Dict[str, Any]: + current = get_graph_run(run_id) + if not current: + raise KeyError("graph run not found") + current.update(payload) + current["updated_at"] = utcnow_iso() + return _upsert_table("graph_runs", "run_id", current) + + +def list_graph_run_nodes(run_id: str) -> List[Dict[str, Any]]: + with get_connection() as connection: + rows = connection.execute( + "SELECT * FROM graph_run_nodes WHERE run_id = ? ORDER BY rowid ASC", + (run_id,), + ).fetchall() + return [_decode_row(row) for row in rows] + + +def get_graph_run_node(run_id: str, node_id: str) -> Optional[Dict[str, Any]]: + with get_connection() as connection: + row = connection.execute( + "SELECT * FROM graph_run_nodes WHERE run_id = ? AND node_id = ? LIMIT 1", + (run_id, node_id), + ).fetchone() + return _decode_row(row) if row else None + + +def update_graph_run_node(run_id: str, node_id: str, payload: Dict[str, Any]) -> Dict[str, Any]: + current = get_graph_run_node(run_id, node_id) + if not current: + raise KeyError("graph run node not found") + current.update(payload) + current["updated_at"] = utcnow_iso() + return _upsert_table("graph_run_nodes", "run_node_id", current) + + +def append_graph_run_event(run_id: str, event_type: str, payload: Dict[str, Any], node_id: Optional[str] = None) -> Dict[str, Any]: + event_payload = { + "event_id": new_id("grevent"), + "run_id": run_id, + "node_id": node_id, + "event_type": event_type, + "payload_json": payload, + "created_at": utcnow_iso(), + } + with get_connection() as connection: + connection.execute( + """ + INSERT INTO graph_run_events (event_id, run_id, node_id, event_type, payload_json, created_at) + VALUES (?, ?, ?, ?, ?, ?) + """, + ( + event_payload["event_id"], + event_payload["run_id"], + event_payload["node_id"], + event_payload["event_type"], + _encode(event_payload["payload_json"]), + event_payload["created_at"], + ), + ) + return event_payload + + +def list_graph_run_events(run_id: str, after_event_id: Optional[str] = None) -> List[Dict[str, Any]]: + params: List[Any] = [run_id] + clause = "run_id = ?" + if after_event_id: + marker = None + with get_connection() as connection: + marker = connection.execute( + "SELECT created_at FROM graph_run_events WHERE event_id = ? AND run_id = ?", + (after_event_id, run_id), + ).fetchone() + if marker: + clause += " AND created_at > ?" + params.append(marker["created_at"]) + with get_connection() as connection: + rows = connection.execute( + f"SELECT * FROM graph_run_events WHERE {clause} ORDER BY created_at ASC, event_id ASC", + params, + ).fetchall() + return [_decode_row(row) for row in rows] + + +def cache_graph_node_definitions(source_fingerprint: str, definitions: List[Dict[str, Any]]) -> Dict[str, Any]: + payload = { + "cache_id": "default", + "source_fingerprint": source_fingerprint, + "definitions_json": definitions, + "updated_at": utcnow_iso(), + } + return _upsert_table("graph_node_definitions_cache", "cache_id", payload) + + def list_enhancement_configs() -> List[Dict[str, Any]]: return _list_table("media_enhancement_configs", "model_key ASC") diff --git a/apps/api/app/store_support.py b/apps/api/app/store_support.py index b9285c7..b751375 100644 --- a/apps/api/app/store_support.py +++ b/apps/api/app/store_support.py @@ -38,6 +38,14 @@ "payload_json", "provider_capabilities_json", "metadata_json", + "workflow_json", + "compiled_graph_json", + "definition_json", + "definitions_json", + "node_snapshot_json", + "input_snapshot_json", + "output_snapshot_json", + "error_json", } MIGRATION_TABLES = {"schema_meta", "schema_migrations"} @@ -1180,6 +1188,107 @@ def _apply_baseline_schema(connection: sqlite3.Connection) -> None: _seed_default_model_queue_policies(connection) +def _apply_graph_studio_schema(connection: sqlite3.Connection) -> None: + connection.executescript( + """ + CREATE TABLE IF NOT EXISTS graph_workflows ( + workflow_id TEXT PRIMARY KEY, + name TEXT NOT NULL, + description TEXT, + status TEXT NOT NULL DEFAULT 'active', + schema_version INTEGER NOT NULL DEFAULT 1, + workflow_json TEXT NOT NULL DEFAULT '{}', + created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, + updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP + ); + + CREATE TABLE IF NOT EXISTS graph_workflow_versions ( + version_id TEXT PRIMARY KEY, + workflow_id TEXT NOT NULL, + version_number INTEGER NOT NULL, + workflow_json TEXT NOT NULL DEFAULT '{}', + created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, + FOREIGN KEY(workflow_id) REFERENCES graph_workflows(workflow_id) + ); + + CREATE TABLE IF NOT EXISTS graph_templates ( + template_id TEXT PRIMARY KEY, + name TEXT NOT NULL, + description TEXT, + status TEXT NOT NULL DEFAULT 'active', + tags_json TEXT NOT NULL DEFAULT '[]', + thumbnail_path TEXT, + workflow_json TEXT NOT NULL DEFAULT '{}', + created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, + updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP + ); + + CREATE TABLE IF NOT EXISTS graph_runs ( + run_id TEXT PRIMARY KEY, + workflow_id TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'queued', + schema_version INTEGER NOT NULL DEFAULT 1, + workflow_json TEXT NOT NULL DEFAULT '{}', + compiled_graph_json TEXT NOT NULL DEFAULT '{}', + output_snapshot_json TEXT NOT NULL DEFAULT '{}', + error TEXT, + created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, + started_at TEXT, + finished_at TEXT, + updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, + FOREIGN KEY(workflow_id) REFERENCES graph_workflows(workflow_id) + ); + + CREATE TABLE IF NOT EXISTS graph_run_nodes ( + run_node_id TEXT PRIMARY KEY, + run_id TEXT NOT NULL, + node_id TEXT NOT NULL, + node_type TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'queued', + progress REAL, + input_snapshot_json TEXT NOT NULL DEFAULT '{}', + output_snapshot_json TEXT NOT NULL DEFAULT '{}', + error TEXT, + started_at TEXT, + finished_at TEXT, + updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, + UNIQUE(run_id, node_id), + FOREIGN KEY(run_id) REFERENCES graph_runs(run_id) + ); + + CREATE TABLE IF NOT EXISTS graph_run_events ( + event_id TEXT PRIMARY KEY, + run_id TEXT NOT NULL, + node_id TEXT, + event_type TEXT NOT NULL, + payload_json TEXT NOT NULL DEFAULT '{}', + created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, + FOREIGN KEY(run_id) REFERENCES graph_runs(run_id) + ); + + CREATE TABLE IF NOT EXISTS graph_node_definitions_cache ( + cache_id TEXT PRIMARY KEY, + source_fingerprint TEXT, + definitions_json TEXT NOT NULL DEFAULT '[]', + created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, + updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP + ); + """ + ) + connection.execute( + """ + CREATE INDEX IF NOT EXISTS idx_graph_run_events_run_created + ON graph_run_events(run_id, created_at) + """ + ) + connection.execute( + """ + CREATE INDEX IF NOT EXISTS idx_graph_runs_workflow_created + ON graph_runs(workflow_id, created_at) + """ + ) + + MIGRATIONS = [ SchemaMigration( migration_id="20260419_001_tracked_baseline", @@ -1210,6 +1319,12 @@ def _apply_baseline_schema(connection: sqlite3.Connection) -> None: description="Enable GPT Image 2 on built-in image presets and preserve Seedance 2 default availability.", apply=_apply_default_model_release_updates, ), + SchemaMigration( + migration_id="20260511_005_graph_studio", + version=5, + description="Add Graph Studio workflow, template, run, event, and node definition tables.", + apply=_apply_graph_studio_schema, + ), ] LATEST_SCHEMA_VERSION = MIGRATIONS[-1].version diff --git a/apps/api/tests/test_db_admin.py b/apps/api/tests/test_db_admin.py index 326dc9c..18a829c 100644 --- a/apps/api/tests/test_db_admin.py +++ b/apps/api/tests/test_db_admin.py @@ -62,13 +62,14 @@ def test_create_clean_database_bootstraps_schema_and_defaults(app_modules, tmp_p assert any("gpt-image-2-image-to-image" in json.loads(preset_row[0]) for preset_row in preset_rows) assert any("gpt-image-2-text-to-image" in json.loads(preset_row[0]) for preset_row in preset_rows) - assert status["schema_version"] == 4 - assert status["latest_version"] == 4 - assert len(status["applied_migrations"]) == 4 + assert status["schema_version"] == status["latest_version"] + assert status["latest_version"] == 5 + assert len(status["applied_migrations"]) == 5 assert status["applied_migrations"][0]["migration_id"] == "20260419_001_tracked_baseline" assert status["applied_migrations"][1]["migration_id"] == "20260419_002_project_cover_references" assert status["applied_migrations"][2]["migration_id"] == "20260419_003_project_visibility_flags" assert status["applied_migrations"][3]["migration_id"] == "20260501_004_default_model_release_updates" + assert status["applied_migrations"][4]["migration_id"] == "20260511_005_graph_studio" assert status["pending_migrations"] == [] @@ -130,8 +131,8 @@ def test_bootstrap_schema_updates_v3_default_model_release_settings(app_modules, connection = sqlite3.connect(legacy_db) try: connection.execute( - "DELETE FROM schema_migrations WHERE migration_id = ?", - ("20260501_004_default_model_release_updates",), + "DELETE FROM schema_migrations WHERE migration_id IN (?, ?)", + ("20260501_004_default_model_release_updates", "20260511_005_graph_studio"), ) connection.execute("UPDATE schema_meta SET value = ? WHERE key = ?", ("3", "schema_version")) connection.execute( @@ -187,7 +188,8 @@ def test_bootstrap_schema_updates_v3_default_model_release_settings(app_modules, assert seedance_policy is not None assert int(seedance_policy[0] or 0) == 1 assert int(seedance_policy[1] or 0) == 1 - assert store.get_schema_status(legacy_db)["schema_version"] == 4 + status = store.get_schema_status(legacy_db) + assert status["schema_version"] == status["latest_version"] == 5 def test_backup_database_copies_existing_database(app_modules, tmp_path: Path) -> None: @@ -238,12 +240,13 @@ def test_bootstrap_schema_creates_backup_before_upgrading_existing_database(app_ assert _count_rows(backup_path, "media_jobs") == 1 status = store.get_schema_status(legacy_db) - assert status["schema_version"] == 4 + assert status["schema_version"] == status["latest_version"] == 5 assert status["pending_migrations"] == [] assert status["applied_migrations"][0]["migration_id"] == "20260419_001_tracked_baseline" assert status["applied_migrations"][1]["migration_id"] == "20260419_002_project_cover_references" assert status["applied_migrations"][2]["migration_id"] == "20260419_003_project_visibility_flags" assert status["applied_migrations"][3]["migration_id"] == "20260501_004_default_model_release_updates" + assert status["applied_migrations"][4]["migration_id"] == "20260511_005_graph_studio" def test_deduplicate_assets_by_job_id_keeps_latest_asset(app_modules) -> None: diff --git a/apps/api/tests/test_graph_studio.py b/apps/api/tests/test_graph_studio.py new file mode 100644 index 0000000..2686608 --- /dev/null +++ b/apps/api/tests/test_graph_studio.py @@ -0,0 +1,127 @@ +from __future__ import annotations + +import time +from pathlib import Path + +PNG_1X1_BYTES = ( + b"\x89PNG\r\n\x1a\n" + b"\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01\x08\x02\x00\x00\x00\x90wS\xde" + b"\x00\x00\x00\x0cIDATx\x9cc\xf8\xff\xff?\x00\x05\xfe\x02\xfeA\x0b~\x90" + b"\x00\x00\x00\x00IEND\xaeB`\x82" +) + + +def _create_reference_image(app_modules) -> str: + data_root = app_modules["main"].settings.data_root + target = data_root / "reference-media" / "images" / "graph-source.png" + target.parent.mkdir(parents=True, exist_ok=True) + target.write_bytes(PNG_1X1_BYTES) + record = app_modules["store"].create_or_reuse_reference_media( + { + "kind": "image", + "original_filename": "graph-source.png", + "stored_path": "reference-media/images/graph-source.png", + "mime_type": "image/png", + "file_size_bytes": len(PNG_1X1_BYTES), + "sha256": "graph-source-hash", + "width": 1, + "height": 1, + "metadata_json": {}, + }, + increment_usage=False, + ) + return record["reference_id"] + + +def _workflow(reference_id: str) -> dict: + return { + "schema_version": 1, + "name": "Graph smoke", + "nodes": [ + { + "id": "load", + "type": "media.load_image", + "position": {"x": 0, "y": 0}, + "fields": {"reference_id": reference_id}, + }, + { + "id": "model", + "type": "model.kie.nano_banana_pro", + "position": {"x": 360, "y": 0}, + "fields": {"prompt": "Create a cinematic editorial image.", "resolution": "1K"}, + }, + { + "id": "save", + "type": "media.save_image", + "position": {"x": 760, "y": 0}, + "fields": {"label": "Final"}, + }, + ], + "edges": [ + {"id": "edge-load-model", "source": "load", "source_port": "image", "target": "model", "target_port": "image_refs"}, + {"id": "edge-model-save", "source": "model", "source_port": "image", "target": "save", "target_port": "image"}, + ], + } + + +def test_graph_node_definitions_include_first_slice_nodes(client) -> None: + response = client.get("/media/graph/node-definitions") + assert response.status_code == 200, response.text + node_types = {item["type"] for item in response.json()["items"]} + assert {"media.load_image", "model.kie.nano_banana_pro", "media.save_image"}.issubset(node_types) + + +def test_graph_validation_rejects_invalid_connections(client, app_modules) -> None: + reference_id = _create_reference_image(app_modules) + workflow = _workflow(reference_id) + workflow["edges"][0]["source_port"] = "missing" + + created = client.post("/media/graph/workflows", json=workflow) + assert created.status_code == 200, created.text + workflow_id = created.json()["workflow_id"] + + response = client.post(f"/media/graph/workflows/{workflow_id}/validate", json=workflow) + assert response.status_code == 200, response.text + payload = response.json() + assert payload["valid"] is False + assert any(error["code"] == "missing_source_port" for error in payload["errors"]) + + +def test_graph_validation_detects_cycles(client, app_modules) -> None: + reference_id = _create_reference_image(app_modules) + workflow = _workflow(reference_id) + workflow["edges"].append({"id": "cycle", "source": "save", "source_port": "asset", "target": "model", "target_port": "image_refs"}) + + created = client.post("/media/graph/workflows", json=workflow) + assert created.status_code == 200, created.text + response = client.post(f"/media/graph/workflows/{created.json()['workflow_id']}/validate", json=workflow) + + assert response.status_code == 200, response.text + assert any(error["code"] == "cycle_detected" for error in response.json()["errors"]) + + +def test_graph_load_image_nano_save_runs_offline_and_creates_asset(client, app_modules) -> None: + reference_id = _create_reference_image(app_modules) + create_response = client.post("/media/graph/workflows", json=_workflow(reference_id)) + assert create_response.status_code == 200, create_response.text + workflow_id = create_response.json()["workflow_id"] + + run_response = client.post(f"/media/graph/workflows/{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 + events = client.get(f"/media/graph/runs/{run_id}/events").json()["items"] + assert any(event["event_type"] == "run.completed" for event in events) + assets = app_modules["store"].list_assets(limit=20) + assert any(asset["model_key"] == "nano-banana-pro" for asset in assets) diff --git a/apps/web/app/globals.css b/apps/web/app/globals.css index c917354..74e76ee 100644 --- a/apps/web/app/globals.css +++ b/apps/web/app/globals.css @@ -1,5 +1,223 @@ +@import "@xyflow/react/dist/style.css"; + @import "tailwindcss"; +.graph-studio-shell { + display: grid; + grid-template-columns: 320px minmax(0, 1fr); + height: 100vh; + min-height: 760px; + background: #0e1110; + color: #f7f6f0; +} + +.graph-sidebar { + border-right: 1px solid rgba(247, 246, 240, 0.12); + background: #151817; + padding: 18px; + overflow: auto; +} + +.graph-sidebar-title, +.graph-toolbar, +.graph-search, +.graph-drop-hint { + display: flex; + align-items: center; + gap: 0.6rem; +} + +.graph-sidebar-title { + font-weight: 800; + letter-spacing: 0.02em; + margin-bottom: 18px; +} + +.graph-workflow-name, +.graph-node-field { + display: grid; + gap: 0.45rem; + font-size: 0.76rem; + color: rgba(247, 246, 240, 0.72); + text-transform: uppercase; + letter-spacing: 0.08em; +} + +.graph-workflow-name input, +.graph-search input, +.graph-node-field-control { + width: 100%; + color: #f7f6f0; + background: #303333; + border: 1px solid rgba(247, 246, 240, 0.09); + border-radius: 8px; + outline: none; + padding: 0.7rem 0.75rem; + text-transform: none; + letter-spacing: 0; +} + +.graph-search { + margin: 16px 0; + padding: 0.15rem 0.55rem; + background: #303333; + border: 1px solid rgba(247, 246, 240, 0.09); + border-radius: 8px; +} + +.graph-search input { + border: 0; + padding-left: 0; +} + +.graph-node-list { + display: grid; + gap: 0.5rem; +} + +.graph-node-list button, +.graph-toolbar button { + color: #f7f6f0; + background: #242827; + border: 1px solid rgba(247, 246, 240, 0.12); + border-radius: 8px; + padding: 0.75rem 0.8rem; + cursor: pointer; +} + +.graph-node-list button { + display: grid; + text-align: left; + gap: 0.2rem; +} + +.graph-node-list span, +.graph-drop-hint, +.graph-node-kind, +.graph-node-port-row small { + color: rgba(247, 246, 240, 0.58); + font-size: 0.72rem; +} + +.graph-drop-hint { + margin-top: 18px; + line-height: 1.35; +} + +.graph-main { + min-width: 0; + display: grid; + grid-template-rows: auto minmax(0, 1fr) 180px; +} + +.graph-toolbar { + min-height: 58px; + padding: 10px 14px; + border-bottom: 1px solid rgba(247, 246, 240, 0.12); + background: #111413; +} + +.graph-toolbar button { + display: inline-flex; + align-items: center; + gap: 0.45rem; +} + +.graph-run-status { + margin-left: auto; + color: rgba(247, 246, 240, 0.72); + font-size: 0.84rem; +} + +.graph-canvas { + min-height: 0; + background: #0e1110; +} + +.graph-console { + overflow: auto; + border-top: 1px solid rgba(247, 246, 240, 0.12); + background: #090b0a; + padding: 0.75rem 1rem; + font-family: ui-monospace, SFMono-Regular, Menlo, Monaco, Consolas, monospace; + font-size: 0.78rem; + color: rgba(247, 246, 240, 0.78); +} + +.graph-node { + width: 340px; + overflow: hidden; + border: 1px solid rgba(247, 246, 240, 0.14); + border-radius: 8px; + background: #171b1a; + box-shadow: 0 14px 42px rgba(0, 0, 0, 0.35); +} + +.graph-node-header { + display: flex; + justify-content: space-between; + gap: 0.75rem; + padding: 0.8rem 0.9rem; + background: #202524; + border-bottom: 1px solid rgba(247, 246, 240, 0.1); +} + +.graph-node-title { + font-weight: 800; + font-size: 0.95rem; +} + +.graph-node-status { + align-self: start; + border: 1px solid rgba(209, 255, 71, 0.25); + color: #d1ff47; + border-radius: 999px; + padding: 0.18rem 0.45rem; + font-size: 0.68rem; +} + +.graph-node-body { + display: grid; + gap: 0.65rem; + padding: 0.8rem 0.9rem 0.95rem; +} + +.graph-node-field textarea { + min-height: 110px; + resize: vertical; +} + +.graph-node-port-row { + position: relative; + display: flex; + align-items: center; + gap: 0.45rem; + min-height: 26px; + color: rgba(247, 246, 240, 0.86); +} + +.graph-node-port-output { + justify-content: flex-end; +} + +.graph-handle { + width: 11px; + height: 11px; + border: 2px solid #d1ff47; + background: #111413; +} + +.react-flow__edge-path { + stroke: #d1ff47; + stroke-width: 2; +} + +.react-flow__controls, +.react-flow__minimap { + background: #171b1a; + border: 1px solid rgba(247, 246, 240, 0.12); +} + :root { --ms-surface-primary: rgba(12, 15, 14, 0.94); --ms-surface-secondary: rgba(17, 20, 19, 0.9); diff --git a/apps/web/app/graph-studio/page.tsx b/apps/web/app/graph-studio/page.tsx new file mode 100644 index 0000000..555e290 --- /dev/null +++ b/apps/web/app/graph-studio/page.tsx @@ -0,0 +1,6 @@ +import { GraphStudio } from "@/components/graph-studio/graph-studio"; + +export default function GraphStudioPage() { + return ; +} + diff --git a/apps/web/components/graph-studio/graph-node.tsx b/apps/web/components/graph-studio/graph-node.tsx new file mode 100644 index 0000000..21a0a57 --- /dev/null +++ b/apps/web/components/graph-studio/graph-node.tsx @@ -0,0 +1,108 @@ +"use client"; + +import { Handle, Position, type NodeProps } from "@xyflow/react"; + +import type { GraphNodeData, StudioNode } from "./types"; + +function FieldControl({ + nodeId, + field, + value, + onFieldChange, +}: { + nodeId: string; + field: GraphNodeData["definition"]["fields"][number]; + value: unknown; + onFieldChange: GraphNodeData["onFieldChange"]; +}) { + if (field.hidden) return null; + const commonClass = "graph-node-field-control nodrag"; + if (field.type === "textarea") { + return ( +