feat: workflow graph system — blocks 1-20 complete checkpoint

Full graph-based workflow execution engine with:
- WorkflowGraphRuntime with two-phase dispatch (collect → fire Celery tasks)
- Node registry with contract validation, socket types, execution kinds
- WorkflowRuntimeServices: preflight, context resolution, shadow-mode A/B
- WorkflowRouter: CRUD + dispatch/preflight/run-history API endpoints
- OutputTypeContracts: workflow binding, rollout-mode resolution
- Admin router: output-type workflow binding endpoints
- Frontend: complete workflow editor with drag-drop canvas, node inspector,
  module bundles, reference bundles, preflight panel, validation banner,
  authoring guidance, blueprint templates, shadow/rollout gate UI
- Tests: comprehensive coverage for all workflow modules
- Docs: NODE_CONTRACT_AUDIT and VALIDATION_ERROR_INVENTORY

All 20 blocks from NEXT_20_BLOCK_BATCH_PLAN completed.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

backend/tests/domains/test_output_types_api.py

@@ -244,6 +244,79 @@ async def test_create_output_type_rejects_workflow_artifact_mismatch(
     assert "blend_asset" in response.json()["detail"]
 
 
+@pytest.mark.asyncio
+async def test_create_output_type_accepts_blend_asset_binding_for_still_graph_blend_blueprint(
+    client,
+    db,
+    auth_headers,
+):
+    workflow = WorkflowDefinition(
+        name=f"Still Blend Graph {uuid.uuid4().hex[:8]}",
+        config=build_workflow_blueprint_config("still_graph_blend_reference"),
+        is_active=True,
+    )
+    db.add(workflow)
+    await db.commit()
+    await db.refresh(workflow)
+
+    response = await client.post(
+        "/api/output-types",
+        json={
+            "name": f"Blend Graph {uuid.uuid4().hex[:8]}",
+            "renderer": "blender",
+            "output_format": "blend",
+            "render_backend": "celery",
+            "workflow_family": "order_line",
+            "artifact_kind": "blend_asset",
+            "workflow_definition_id": str(workflow.id),
+            "workflow_rollout_mode": "graph",
+        },
+        headers=auth_headers,
+    )
+
+    assert response.status_code == 201, response.text
+    payload = response.json()
+    assert payload["artifact_kind"] == "blend_asset"
+    assert payload["workflow_definition_id"] == str(workflow.id)
+
+
+@pytest.mark.asyncio
+async def test_create_output_type_accepts_still_binding_for_alpha_graph_blueprint(
+    client,
+    db,
+    auth_headers,
+):
+    workflow = WorkflowDefinition(
+        name=f"Still Alpha Graph {uuid.uuid4().hex[:8]}",
+        config=build_workflow_blueprint_config("still_graph_alpha_reference"),
+        is_active=True,
+    )
+    db.add(workflow)
+    await db.commit()
+    await db.refresh(workflow)
+
+    response = await client.post(
+        "/api/output-types",
+        json={
+            "name": f"Still Alpha {uuid.uuid4().hex[:8]}",
+            "renderer": "blender",
+            "output_format": "png",
+            "render_backend": "celery",
+            "workflow_family": "order_line",
+            "artifact_kind": "still_image",
+            "workflow_definition_id": str(workflow.id),
+            "workflow_rollout_mode": "graph",
+            "transparent_bg": True,
+        },
+        headers=auth_headers,
+    )
+
+    assert response.status_code == 201, response.text
+    payload = response.json()
+    assert payload["artifact_kind"] == "still_image"
+    assert payload["transparent_bg"] is True
+
+
 @pytest.mark.asyncio
 async def test_create_output_type_rejects_artifact_kind_incompatible_with_family(
     client,

backend/tests/domains/test_workflow_config_utils.py

@@ -7,6 +7,7 @@ from app.domains.rendering.workflow_config_utils import (
     get_workflow_execution_mode,
     workflow_config_requires_canonicalization,
 )
+from app.domains.rendering.output_type_contracts import derive_supported_artifact_kinds_from_workflow_config
 
 
 def test_build_preset_workflow_config_creates_canonical_dag():
@@ -368,6 +369,11 @@ def test_build_workflow_blueprint_config_creates_order_rendering_family_graph():
     assert any(node["step"] == "blender_turntable" for node in config["nodes"])
     assert any(node["step"] == "export_blend" for node in config["nodes"])
     assert sum(1 for node in config["nodes"] if node["step"] == "notify") == 3
+    assert derive_supported_artifact_kinds_from_workflow_config(config) == (
+        "blend_asset",
+        "still_image",
+        "turntable_video",
+    )
 
 
 def test_build_workflow_blueprint_config_creates_still_graph_reference():
@@ -390,6 +396,33 @@ def test_build_workflow_blueprint_config_creates_still_graph_reference():
     ]
     render_node = next(node for node in config["nodes"] if node["step"] == "blender_still")
     assert render_node["params"]["use_custom_render_settings"] is False
+    assert derive_supported_artifact_kinds_from_workflow_config(config) == ("still_image",)
+
+
+def test_build_workflow_blueprint_config_creates_still_graph_alpha_reference():
+    config = build_workflow_blueprint_config("still_graph_alpha_reference")
+
+    assert config["version"] == 1
+    assert config["ui"]["blueprint"] == "still_graph_alpha_reference"
+    assert config["ui"]["execution_mode"] == "graph"
+    render_node = next(node for node in config["nodes"] if node["step"] == "blender_still")
+    assert render_node["params"]["transparent_bg"] is True
+    assert derive_supported_artifact_kinds_from_workflow_config(config) == ("still_image",)
+
+
+def test_build_workflow_blueprint_config_creates_still_graph_blend_reference():
+    config = build_workflow_blueprint_config("still_graph_blend_reference")
+
+    assert config["version"] == 1
+    assert config["ui"]["blueprint"] == "still_graph_blend_reference"
+    assert config["ui"]["execution_mode"] == "graph"
+    assert any(node["step"] == "export_blend" for node in config["nodes"])
+    save_blend = next(node for node in config["nodes"] if node["id"] == "save_blend")
+    assert save_blend["params"]["expected_artifact_role"] == "blend_export"
+    assert derive_supported_artifact_kinds_from_workflow_config(config) == (
+        "blend_asset",
+        "still_image",
+    )
 
 
 def test_build_starter_workflow_config_creates_minimal_valid_custom_graph():

backend/tests/domains/test_workflow_dispatch_service.py

@@ -143,6 +143,14 @@ def _derive_rollout_mode_from_config(workflow_config: dict | None) -> str:
     return "legacy_only"
 
 
+def _assert_generated_task_ids(task_ids: list[str], expected_count: int) -> list[str]:
+    assert len(task_ids) == expected_count
+    assert len(set(task_ids)) == expected_count
+    for task_id in task_ids:
+        uuid.UUID(task_id)
+    return task_ids
+
+
 async def _seed_order_line(
     db,
     admin_user,
@@ -420,7 +428,7 @@ async def test_dispatch_render_with_workflow_graph_mode_dispatches_supported_cus
 
     monkeypatch.setattr(
         "app.tasks.celery_app.celery_app.send_task",
-        lambda task_name, args, kwargs: type("Result", (), {"id": "graph-task-1"})(),
+        lambda task_name, args, kwargs, **task_options: type("Result", (), {"id": "graph-task-1"})(),
     )
 
     result = dispatch_render_with_workflow(str(order_line.id))
@@ -437,7 +445,7 @@ async def test_dispatch_render_with_workflow_graph_mode_dispatches_supported_cus
 
     assert result["backend"] == "workflow_graph"
     assert result["execution_mode"] == "graph"
-    assert result["task_ids"] == ["graph-task-1"]
+    generated_task_ids = _assert_generated_task_ids(result["task_ids"], 1)
     assert result["rollout_gate_status"] == "graph_authoritative"
     assert result["rollout_gate_verdict"] == "pass"
     assert result["workflow_rollout_ready"] is True
@@ -447,6 +455,7 @@ async def test_dispatch_render_with_workflow_graph_mode_dispatches_supported_cus
     assert node_results["setup"].status == "completed"
     assert node_results["template"].status == "completed"
     assert node_results["render"].status == "queued"
+    assert node_results["render"].output["task_id"] == generated_task_ids[0]
     assert node_results["render"].output["publish_asset_enabled"] is True
     assert node_results["render"].output["graph_authoritative_output_enabled"] is False
 
@@ -476,7 +485,7 @@ async def test_dispatch_render_with_workflow_graph_mode_uses_output_save_as_auth
 
     calls: list[tuple[str, list[str], dict]] = []
 
-    def _fake_send_task(task_name: str, args: list[str], kwargs: dict):
+    def _fake_send_task(task_name: str, args: list[str], kwargs: dict, **task_options):
         calls.append((task_name, args, kwargs))
         return type("Result", (), {"id": "graph-output-save-task-1"})()
 
@@ -495,13 +504,14 @@ async def test_dispatch_render_with_workflow_graph_mode_uses_output_save_as_auth
     node_results = {node_result.node_name: node_result for node_result in run.node_results}
 
     assert result["backend"] == "workflow_graph"
-    assert result["task_ids"] == ["graph-output-save-task-1"]
+    generated_task_ids = _assert_generated_task_ids(result["task_ids"], 1)
     assert len(calls) == 1
     assert calls[0][0] == "app.domains.rendering.tasks.render_order_line_still_task"
     assert calls[0][1] == [str(order_line.id)]
     assert calls[0][2]["publish_asset_enabled"] is False
     assert calls[0][2]["graph_authoritative_output_enabled"] is True
     assert calls[0][2]["graph_output_node_ids"] == ["output"]
+    assert node_results["render"].output["task_id"] == generated_task_ids[0]
     assert node_results["output"].status == "pending"
     assert node_results["output"].output["publication_mode"] == "awaiting_graph_authoritative_save"
     assert node_results["output"].output["handoff_state"] == "armed"
@@ -537,7 +547,7 @@ async def test_dispatch_render_with_workflow_graph_mode_canonicalizes_legacy_pre
 
     monkeypatch.setattr(
         "app.tasks.celery_app.celery_app.send_task",
-        lambda task_name, args, kwargs: type("Result", (), {"id": "legacy-graph-task-1"})(),
+        lambda task_name, args, kwargs, **task_options: type("Result", (), {"id": "legacy-graph-task-1"})(),
     )
 
     result = dispatch_render_with_workflow(str(order_line.id))
@@ -554,11 +564,12 @@ async def test_dispatch_render_with_workflow_graph_mode_canonicalizes_legacy_pre
 
     assert result["backend"] == "workflow_graph"
     assert result["execution_mode"] == "graph"
-    assert result["task_ids"] == ["legacy-graph-task-1"]
+    generated_task_ids = _assert_generated_task_ids(result["task_ids"], 1)
     assert run.execution_mode == "graph"
     assert node_results["setup"].status == "completed"
     assert node_results["template"].status == "completed"
     assert node_results["render"].status == "queued"
+    assert node_results["render"].output["task_id"] == generated_task_ids[0]
     assert node_results["output"].status == "pending"
 
 
@@ -674,7 +685,7 @@ async def test_dispatch_render_with_workflow_shadow_mode_keeps_legacy_authoritat
 
     calls: list[tuple[str, list[str], dict]] = []
 
-    def _fake_send_task(task_name: str, args: list[str], kwargs: dict):
+    def _fake_send_task(task_name: str, args: list[str], kwargs: dict, **task_options):
         calls.append((task_name, args, kwargs))
         return type("Result", (), {"id": "shadow-task-1"})()
 
@@ -694,18 +705,20 @@ async def test_dispatch_render_with_workflow_shadow_mode_keeps_legacy_authoritat
         .options(selectinload(WorkflowRun.node_results))
     )
     run = run_result.scalar_one()
+    node_results = {node_result.node_name: node_result for node_result in run.node_results}
     render_call = calls[0]
 
     assert result["backend"] == "legacy"
     assert result["execution_mode"] == "shadow"
     assert result["shadow_status"] == "dispatched"
-    assert result["shadow_task_ids"] == ["shadow-task-1"]
+    shadow_task_ids = _assert_generated_task_ids(result["shadow_task_ids"], 1)
     assert result["rollout_gate_status"] == "pending_shadow_verdict"
     assert result["rollout_gate_verdict"] is None
     assert result["workflow_rollout_ready"] is False
     assert result["output_type_rollout_ready"] is False
     assert run.execution_mode == "shadow"
     assert run.status == "pending"
+    assert node_results["render"].output["task_id"] == shadow_task_ids[0]
     assert render_call[0] == "app.domains.rendering.tasks.render_order_line_still_task"
     assert render_call[1] == [str(order_line.id)]
     assert render_call[2]["publish_asset_enabled"] is False
@@ -745,7 +758,7 @@ async def test_dispatch_render_with_workflow_shadow_mode_canonicalizes_legacy_pr
 
     calls: list[tuple[str, list[str], dict]] = []
 
-    def _fake_send_task(task_name: str, args: list[str], kwargs: dict):
+    def _fake_send_task(task_name: str, args: list[str], kwargs: dict, **task_options):
         calls.append((task_name, args, kwargs))
         return type("Result", (), {"id": "legacy-shadow-task-1"})()
 
@@ -1021,7 +1034,7 @@ def test_dispatch_render_with_workflow_unit_marks_shadow_dispatch_as_pending_rol
     )
     monkeypatch.setattr(
         "app.domains.rendering.workflow_graph_runtime.execute_graph_workflow",
-        lambda *args, **kwargs: SimpleNamespace(task_ids=["shadow-task-1"]),
+        lambda *args, **kwargs: SimpleNamespace(task_ids=["shadow-task-1"], task_specs=[]),
     )
 
     result = dispatch_render_with_workflow(order_line_id)
@@ -1065,7 +1078,7 @@ async def test_workflow_dispatch_endpoint_returns_workflow_run_with_node_results
 
     calls: list[tuple[str, list[str], dict]] = []
 
-    def _fake_send_task(task_name: str, args: list[str], kwargs: dict):
+    def _fake_send_task(task_name: str, args: list[str], kwargs: dict, **task_options):
         calls.append((task_name, args, kwargs))
         return type("Result", (), {"id": f"task-{len(calls)}"})()
 
@@ -1083,7 +1096,7 @@ async def test_workflow_dispatch_endpoint_returns_workflow_run_with_node_results
     assert body["context_id"] == context_id
     assert body["execution_mode"] == "graph"
     assert body["dispatched"] == 2
-    assert body["task_ids"] == ["task-1", "task-2"]
+    generated_task_ids = _assert_generated_task_ids(body["task_ids"], 2)
     assert [call[0] for call in calls] == [
         "app.domains.rendering.tasks.render_order_line_still_task",
         "app.domains.rendering.tasks.export_blend_for_order_line_task",
@@ -1103,12 +1116,12 @@ async def test_workflow_dispatch_endpoint_returns_workflow_run_with_node_results
     node_results = {node["node_name"]: node for node in body["workflow_run"]["node_results"]}
     assert body["workflow_run"]["status"] == "pending"
     assert body["workflow_run"]["execution_mode"] == "graph"
-    assert body["workflow_run"]["celery_task_id"] == "task-1"
+    assert body["workflow_run"]["celery_task_id"] == generated_task_ids[0]
     assert body["workflow_run"]["order_line_id"] == str(order_line.id)
     assert node_results["render"]["status"] == "queued"
-    assert node_results["render"]["output"]["task_id"] == "task-1"
+    assert node_results["render"]["output"]["task_id"] == generated_task_ids[0]
     assert node_results["blend"]["status"] == "queued"
-    assert node_results["blend"]["output"]["task_id"] == "task-2"
+    assert node_results["blend"]["output"]["task_id"] == generated_task_ids[1]
     assert node_results["setup"]["status"] == "completed"
     assert node_results["setup"]["output"]["order_line_id"] == str(order_line.id)
     assert node_results["template"]["status"] == "completed"
@@ -1120,7 +1133,7 @@ async def test_workflow_dispatch_endpoint_returns_workflow_run_with_node_results
 
 
 @pytest.mark.asyncio
-async def test_workflow_dispatch_endpoint_rejects_output_save_for_export_blend_only_graph(
+async def test_workflow_dispatch_endpoint_arms_output_save_for_export_blend_only_graph(
     client,
     db,
     admin_user,
@@ -1141,7 +1154,7 @@ async def test_workflow_dispatch_endpoint_rejects_output_save_for_export_blend_o
 
     calls: list[tuple[str, list[str], dict]] = []
 
-    def _fake_send_task(task_name: str, args: list[str], kwargs: dict):
+    def _fake_send_task(task_name: str, args: list[str], kwargs: dict, **task_options):
         calls.append((task_name, args, kwargs))
         return type("Result", (), {"id": f"task-{len(calls)}"})()
 
@@ -1153,9 +1166,35 @@ async def test_workflow_dispatch_endpoint_rejects_output_save_for_export_blend_o
         headers=auth_headers,
     )
 
-    assert response.status_code == 422
-    assert "output_save" in response.json()["detail"]
-    assert calls == []
+    assert response.status_code == 200
+    body = response.json()
+
+    assert body["context_id"] == context_id
+    assert body["execution_mode"] == "graph"
+    assert body["dispatched"] == 1
+    generated_task_ids = _assert_generated_task_ids(body["task_ids"], 1)
+    assert len(calls) == 1
+    assert calls[0][0] == "app.domains.rendering.tasks.export_blend_for_order_line_task"
+    assert calls[0][1] == [context_id]
+    assert calls[0][2]["workflow_node_id"] == "blend"
+    assert calls[0][2]["publish_asset_enabled"] is False
+    assert calls[0][2]["graph_authoritative_output_enabled"] is True
+    assert calls[0][2]["graph_output_node_ids"] == ["output"]
+    assert "workflow_run_id" in calls[0][2]
+
+    node_results = {node["node_name"]: node for node in body["workflow_run"]["node_results"]}
+    assert body["workflow_run"]["status"] == "pending"
+    assert body["workflow_run"]["execution_mode"] == "graph"
+    assert body["workflow_run"]["celery_task_id"] == generated_task_ids[0]
+    assert body["workflow_run"]["order_line_id"] == str(order_line.id)
+    assert node_results["setup"]["status"] == "completed"
+    assert node_results["template"]["status"] == "completed"
+    assert node_results["blend"]["status"] == "queued"
+    assert node_results["blend"]["output"]["task_id"] == generated_task_ids[0]
+    assert node_results["output"]["status"] == "pending"
+    assert node_results["output"]["output"]["publication_mode"] == "awaiting_graph_authoritative_save"
+    assert node_results["output"]["output"]["handoff_state"] == "armed"
+    assert node_results["output"]["output"]["handoff_node_ids"] == ["blend"]
 
 
 @pytest.mark.asyncio
@@ -1180,7 +1219,7 @@ async def test_workflow_dispatch_endpoint_arms_output_save_for_turntable(
 
     calls: list[tuple[str, list[str], dict]] = []
 
-    def _fake_send_task(task_name: str, args: list[str], kwargs: dict):
+    def _fake_send_task(task_name: str, args: list[str], kwargs: dict, **task_options):
         calls.append((task_name, args, kwargs))
         return type("Result", (), {"id": f"task-{len(calls)}"})()
 
@@ -1198,7 +1237,7 @@ async def test_workflow_dispatch_endpoint_arms_output_save_for_turntable(
     assert body["context_id"] == context_id
     assert body["execution_mode"] == "graph"
     assert body["dispatched"] == 1
-    assert body["task_ids"] == ["task-1"]
+    generated_task_ids = _assert_generated_task_ids(body["task_ids"], 1)
     assert calls[0][0] == "app.domains.rendering.tasks.render_turntable_task"
     assert calls[0][1] == [context_id]
     assert calls[0][2]["workflow_run_id"] == body["workflow_run"]["id"]
@@ -1210,6 +1249,7 @@ async def test_workflow_dispatch_endpoint_arms_output_save_for_turntable(
 
     node_results = {node["node_name"]: node for node in body["workflow_run"]["node_results"]}
     assert node_results["turntable"]["status"] == "queued"
+    assert node_results["turntable"]["output"]["task_id"] == generated_task_ids[0]
     assert node_results["turntable"]["output"]["predicted_asset_type"] == "turntable"
     assert node_results["turntable"]["output"]["publish_asset_enabled"] is False
     assert node_results["turntable"]["output"]["graph_authoritative_output_enabled"] is True
@@ -1242,7 +1282,7 @@ async def test_workflow_dispatch_endpoint_arms_notify_handoff_for_render_node(
 
     calls: list[tuple[str, list[str], dict]] = []
 
-    def _fake_send_task(task_name: str, args: list[str], kwargs: dict):
+    def _fake_send_task(task_name: str, args: list[str], kwargs: dict, **task_options):
         calls.append((task_name, args, kwargs))
         return type("Result", (), {"id": "task-1"})()
 
@@ -1260,7 +1300,7 @@ async def test_workflow_dispatch_endpoint_arms_notify_handoff_for_render_node(
     assert body["context_id"] == context_id
     assert body["execution_mode"] == "graph"
     assert body["dispatched"] == 1
-    assert body["task_ids"] == ["task-1"]
+    generated_task_ids = _assert_generated_task_ids(body["task_ids"], 1)
     assert len(calls) == 1
     assert calls[0][0] == "app.domains.rendering.tasks.render_order_line_still_task"
     assert calls[0][1] == [context_id]
@@ -1271,6 +1311,7 @@ async def test_workflow_dispatch_endpoint_arms_notify_handoff_for_render_node(
 
     node_results = {node["node_name"]: node for node in body["workflow_run"]["node_results"]}
     assert node_results["render"]["status"] == "queued"
+    assert node_results["render"]["output"]["task_id"] == generated_task_ids[0]
     assert node_results["render"]["output"]["graph_notify_node_ids"] == ["notify"]
     assert node_results["notify"]["status"] == "pending"
     assert node_results["notify"]["output"]["notification_mode"] == "deferred_to_render_task"
@@ -1321,6 +1362,54 @@ async def test_workflow_preflight_endpoint_reports_render_graph_readiness(
     assert node_checks["blend"]["status"] == "ready"
 
 
+@pytest.mark.asyncio
+async def test_workflow_order_line_contexts_keep_completed_orders_renderable_for_editor_reruns(
+    client,
+    db,
+    admin_user,
+    auth_headers,
+    tmp_path,
+):
+    ready_line = await _seed_renderable_order_line(db, admin_user, tmp_path)
+    completed_line = await _seed_renderable_order_line(db, admin_user, tmp_path)
+    blocked_line = await _seed_renderable_order_line(db, admin_user, tmp_path)
+
+    completed_order = await db.get(Order, completed_line.order_id)
+    assert completed_order is not None
+    completed_order.status = OrderStatus.completed
+
+    blocked_order = await db.get(Order, blocked_line.order_id)
+    assert blocked_order is not None
+    blocked_order.status = OrderStatus.rejected
+    await db.commit()
+
+    response = await client.get(
+        "/api/workflows/contexts/order-lines",
+        headers=auth_headers,
+    )
+
+    assert response.status_code == 200
+    body = response.json()
+    flattened = [
+        option
+        for group in body
+        for option in group["options"]
+    ]
+    by_id = {option["value"]: option for option in flattened}
+
+    assert by_id[str(ready_line.id)]["is_renderable"] is True
+    assert by_id[str(ready_line.id)]["renderability_reason"] is None
+    assert by_id[str(completed_line.id)]["is_renderable"] is True
+    assert by_id[str(completed_line.id)]["renderability_reason"] is None
+    assert by_id[str(blocked_line.id)]["is_renderable"] is False
+    assert by_id[str(blocked_line.id)]["renderability_reason"] == "order_closed"
+
+    renderable_ids = {option["value"] for option in flattened if option["is_renderable"]}
+    assert str(ready_line.id) in renderable_ids
+    assert str(completed_line.id) in renderable_ids
+    assert str(blocked_line.id) not in renderable_ids
+
+
 @pytest.mark.asyncio
 async def test_workflow_draft_dispatch_endpoint_dispatches_unsaved_render_graph(
     client,
@@ -1342,7 +1431,7 @@ async def test_workflow_draft_dispatch_endpoint_dispatches_unsaved_render_graph(
 
     calls: list[tuple[str, list[str], dict]] = []
 
-    def _fake_send_task(task_name: str, args: list[str], kwargs: dict):
+    def _fake_send_task(task_name: str, args: list[str], kwargs: dict, **task_options):
         calls.append((task_name, args, kwargs))
         return type("Result", (), {"id": f"draft-task-{len(calls)}"})()
 
@@ -1364,7 +1453,7 @@ async def test_workflow_draft_dispatch_endpoint_dispatches_unsaved_render_graph(
     assert body["context_id"] == str(order_line.id)
     assert body["execution_mode"] == "graph"
     assert body["dispatched"] == 1
-    assert body["task_ids"] == ["draft-task-1"]
+    generated_task_ids = _assert_generated_task_ids(body["task_ids"], 1)
     assert body["workflow_run"]["workflow_def_id"] == str(workflow_definition.id)
     assert body["workflow_run"]["execution_mode"] == "graph"
     assert body["workflow_run"]["order_line_id"] == str(order_line.id)
@@ -1375,6 +1464,7 @@ async def test_workflow_draft_dispatch_endpoint_dispatches_unsaved_render_graph(
     assert node_results["setup"]["status"] == "completed"
     assert node_results["template"]["status"] == "completed"
     assert node_results["render"]["status"] == "queued"
+    assert node_results["render"]["output"]["task_id"] == generated_task_ids[0]
 
 
 @pytest.mark.asyncio
@@ -1394,7 +1484,7 @@ async def test_workflow_draft_dispatch_endpoint_marks_submitted_order_processing
 
     monkeypatch.setattr(
         "app.tasks.celery_app.celery_app.send_task",
-        lambda task_name, args, kwargs: type("Result", (), {"id": "draft-task-1"})(),
+        lambda task_name, args, kwargs, **task_options: type("Result", (), {"id": "draft-task-1"})(),
     )
     response = await client.post(
         "/api/workflows/dispatch",
@@ -1412,6 +1502,44 @@ async def test_workflow_draft_dispatch_endpoint_marks_submitted_order_processing
     assert order_line.order.completed_at is None
 
 
+@pytest.mark.asyncio
+async def test_workflow_preflight_endpoint_allows_completed_order_context_for_editor_reruns(
+    client,
+    db,
+    admin_user,
+    auth_headers,
+    tmp_path,
+):
+    order_line = await _seed_renderable_order_line(db, admin_user, tmp_path)
+    order = await db.get(Order, order_line.order_id)
+    assert order is not None
+    order.status = OrderStatus.completed
+    await db.commit()
+
+    workflow_definition = WorkflowDefinition(
+        name=f"Completed Order Preflight {uuid.uuid4().hex[:8]}",
+        config=_build_valid_custom_still_graph(),
+        is_active=True,
+    )
+    db.add(workflow_definition)
+    await db.commit()
+    await db.refresh(workflow_definition)
+
+    response = await client.get(
+        f"/api/workflows/{workflow_definition.id}/preflight",
+        headers=auth_headers,
+        params={"context_id": str(order_line.id)},
+    )
+
+    assert response.status_code == 200
+    body = response.json()
+    assert body["graph_dispatch_allowed"] is True
+    assert body["summary"] == "Preflight passed with warnings."
+    node_checks = {node["node_id"]: node for node in body["nodes"]}
+    assert node_checks["setup"]["status"] == "ready"
+    assert not any(issue["code"] == "order_line_skipped" for issue in body["issues"])
+
+
 @pytest.mark.asyncio
 async def test_workflow_draft_dispatch_endpoint_rejects_invalid_graph_config(
     client,

backend/tests/domains/test_workflow_executor.py

@@ -102,7 +102,7 @@ def test_dispatch_workflow_preserves_mapped_task_order_around_phase_3_bridge_nod
                 {"id": "setup", "step": StepName.ORDER_LINE_SETUP.value, "params": {}},
                 {"id": "render", "step": StepName.BLENDER_STILL.value, "params": {"width": 1024}},
                 {"id": "save", "step": StepName.OUTPUT_SAVE.value, "params": {}},
-                {"id": "export", "step": StepName.EXPORT_BLEND.value, "params": {"include_textures": True}},
+                {"id": "export", "step": StepName.EXPORT_BLEND.value, "params": {"output_name_suffix": "client"}},
                 {"id": "notify", "step": StepName.NOTIFY.value, "params": {}},
             ],
             "edges": [
@@ -125,6 +125,6 @@ def test_dispatch_workflow_preserves_mapped_task_order_around_phase_3_bridge_nod
         (
             "app.domains.rendering.tasks.export_blend_for_order_line_task",
             ["line-456"],
-            {"include_textures": True},
+            {"output_name_suffix": "client"},
         ),
     ]

backend/tests/domains/test_workflow_node_registry.py

@@ -3,7 +3,16 @@ import pytest
 from app.core.process_steps import StepName
 from app.domains.rendering.models import OutputType, WorkflowDefinition, WorkflowRun
 from app.domains.rendering.workflow_config_utils import build_preset_workflow_config
-from app.domains.rendering.workflow_graph_runtime import _STILL_TASK_KEYS, _TURNTABLE_TASK_KEYS
+from app.domains.rendering.workflow_graph_runtime import (
+    _AUTO_POPULATE_RUNTIME_KEYS,
+    _GLB_BBOX_RUNTIME_KEYS,
+    _MATERIAL_MAP_RUNTIME_KEYS,
+    _NOTIFY_RUNTIME_KEYS,
+    _OUTPUT_SAVE_RUNTIME_KEYS,
+    _RESOLVE_TEMPLATE_RUNTIME_KEYS,
+    _STILL_TASK_KEYS,
+    _TURNTABLE_TASK_KEYS,
+)
 from app.domains.rendering.workflow_node_registry import (
     get_node_definition,
     list_node_definitions,
@@ -121,6 +130,29 @@ def test_graph_render_node_fields_are_supported_by_runtime_dispatch():
     assert turntable_runtime_fields <= _TURNTABLE_TASK_KEYS
 
 
+def test_bridge_node_fields_match_runtime_dispatch_contracts():
+    resolve_template = get_node_definition(StepName.RESOLVE_TEMPLATE)
+    material_map = get_node_definition(StepName.MATERIAL_MAP_RESOLVE)
+    auto_populate = get_node_definition(StepName.AUTO_POPULATE_MATERIALS)
+    glb_bbox = get_node_definition(StepName.GLB_BBOX)
+    output_save = get_node_definition(StepName.OUTPUT_SAVE)
+    notify = get_node_definition(StepName.NOTIFY)
+
+    assert resolve_template is not None
+    assert material_map is not None
+    assert auto_populate is not None
+    assert glb_bbox is not None
+    assert output_save is not None
+    assert notify is not None
+
+    assert {field.key for field in resolve_template.fields} == _RESOLVE_TEMPLATE_RUNTIME_KEYS
+    assert {field.key for field in material_map.fields} == _MATERIAL_MAP_RUNTIME_KEYS
+    assert {field.key for field in auto_populate.fields} == _AUTO_POPULATE_RUNTIME_KEYS
+    assert {field.key for field in glb_bbox.fields} == _GLB_BBOX_RUNTIME_KEYS
+    assert {field.key for field in output_save.fields} == _OUTPUT_SAVE_RUNTIME_KEYS
+    assert {field.key for field in notify.fields} == _NOTIFY_RUNTIME_KEYS
+
+
 def test_order_line_setup_and_template_contracts_expose_runtime_outputs():
     setup = get_node_definition(StepName.ORDER_LINE_SETUP)
     template = get_node_definition(StepName.RESOLVE_TEMPLATE)
@@ -178,7 +210,13 @@ def test_order_line_setup_and_template_contracts_expose_runtime_outputs():
         "include_populated_products",
     }
     assert output.input_contract["requires"] == ["order_line_context"]
-    assert output.input_contract["requires_any"] == ["rendered_image", "rendered_frames", "rendered_video"]
+    assert output.input_contract["requires_any"] == [
+        "rendered_image",
+        "rendered_frames",
+        "rendered_video",
+        "blend_asset",
+    ]
+    assert "blend_asset" in output.artifact_roles_consumed
     assert set(output.output_contract["provides"]) >= {"media_asset", "workflow_result"}
     assert {field.key for field in output.fields} == {
         "expected_artifact_role",
@@ -193,6 +231,7 @@ def test_order_line_setup_and_template_contracts_expose_runtime_outputs():
         "rendered_frames",
         "rendered_video",
         "workflow_result",
+        "blend_asset",
     ]
     assert {field.key for field in notify.fields} == {"channel", "require_armed_render"}
 

backend/tests/domains/test_workflow_runtime_services.py

@@ -581,6 +581,32 @@ def test_prepare_order_line_render_context_skips_closed_orders(sync_session, tmp
     assert line.render_status == "cancelled"
 
 
+def test_prepare_order_line_render_context_allows_completed_rerender_for_workflow_editor(
+    sync_session,
+    tmp_path,
+    monkeypatch,
+):
+    from app.config import settings
+
+    monkeypatch.setattr(settings, "upload_dir", str(tmp_path / "uploads"))
+    line = _seed_order_line_graph(sync_session, tmp_path)
+    line.order.status = OrderStatus.completed
+    sync_session.commit()
+
+    result = prepare_order_line_render_context(
+        sync_session,
+        str(line.id),
+        persist_state=False,
+        allow_completed_order_rerender=True,
+    )
+    sync_session.refresh(line)
+
+    assert result.status == "ready"
+    assert result.reason is None
+    assert result.is_ready is True
+    assert line.render_status == "pending"
+
+
 def test_build_order_line_render_invocation_applies_output_and_line_overrides(tmp_path):
     step_path = tmp_path / "parts" / "bearing.step"
     step_path.parent.mkdir(parents=True, exist_ok=True)

backend/tests/domains/test_workflow_schema.py

@@ -399,6 +399,54 @@ def test_workflow_schema_accepts_transitive_contract_wiring():
     assert config.ui.execution_mode == "graph"
 
 
+def test_workflow_schema_accepts_export_blend_notify_contract_wiring():
+    config = WorkflowConfig.model_validate(
+        {
+            "version": 1,
+            "nodes": [
+                {"id": "setup", "step": "order_line_setup", "params": {}},
+                {"id": "template", "step": "resolve_template", "params": {}},
+                {"id": "export", "step": "export_blend", "params": {"output_name_suffix": "review"}},
+                {"id": "notify", "step": "notify", "params": {}},
+            ],
+            "edges": [
+                {"from": "setup", "to": "template"},
+                {"from": "setup", "to": "export"},
+                {"from": "template", "to": "export"},
+                {"from": "export", "to": "notify"},
+            ],
+            "ui": {"family": "order_line", "execution_mode": "graph"},
+        }
+    )
+
+    assert config.ui is not None
+    assert config.ui.execution_mode == "graph"
+
+
+def test_workflow_schema_accepts_export_blend_output_save_contract_wiring():
+    config = WorkflowConfig.model_validate(
+        {
+            "version": 1,
+            "nodes": [
+                {"id": "setup", "step": "order_line_setup", "params": {}},
+                {"id": "template", "step": "resolve_template", "params": {}},
+                {"id": "export", "step": "export_blend", "params": {"output_name_suffix": "review"}},
+                {"id": "output", "step": "output_save", "params": {"expected_artifact_role": "blend_export"}},
+            ],
+            "edges": [
+                {"from": "setup", "to": "template"},
+                {"from": "setup", "to": "export"},
+                {"from": "template", "to": "export"},
+                {"from": "export", "to": "output"},
+            ],
+            "ui": {"family": "order_line", "execution_mode": "graph"},
+        }
+    )
+
+    assert config.ui is not None
+    assert config.ui.execution_mode == "graph"
+
+
 def test_workflow_schema_accepts_cad_intake_contract_wiring_with_shared_bbox_node():
     config = WorkflowConfig.model_validate(
         {