feat: add duplicate-safe workflow shadow dispatch

backend/app/domains/rendering/workflow_graph_runtime.py

@@ -208,12 +208,26 @@ def execute_graph_workflow(
 
             from app.tasks.celery_app import celery_app
 
-            result = celery_app.send_task(task_name, args=[workflow_context.context_id], kwargs=node.params)
+            task_kwargs = dict(node.params)
+            task_kwargs["workflow_run_id"] = str(workflow_context.workflow_run_id)
+            task_kwargs["workflow_node_id"] = node.id
+            if workflow_context.execution_mode == "shadow":
+                task_kwargs["publish_asset_enabled"] = False
+                task_kwargs["emit_events"] = False
+                task_kwargs["job_document_enabled"] = False
+                task_kwargs["output_name_suffix"] = f"shadow-{str(workflow_context.workflow_run_id)[:8]}"
+
+            result = celery_app.send_task(
+                task_name,
+                args=[workflow_context.context_id],
+                kwargs=task_kwargs,
+            )
             metadata["task_id"] = result.id
             if definition is not None:
                 metadata["execution_kind"] = definition.execution_kind
             metadata["attempt_count"] = 1
             metadata["max_attempts"] = retry_policy["max_attempts"]
+            metadata["execution_mode"] = workflow_context.execution_mode
             node_result.status = "queued"
             node_result.output = metadata
             node_result.log = None
@@ -371,9 +385,18 @@ def _execute_order_line_setup(
     node_params: dict[str, Any],
 ) -> tuple[dict[str, Any], str, str | None]:
     del node_params
-    setup = prepare_order_line_render_context(session, workflow_context.context_id)
+    shadow_mode = workflow_context.execution_mode == "shadow"
+    if shadow_mode:
+        setup = prepare_order_line_render_context(
+            session,
+            workflow_context.context_id,
+            persist_state=False,
+        )
+    else:
+        setup = prepare_order_line_render_context(session, workflow_context.context_id)
     state.setup = setup
     payload = _serialize_setup_result(setup)
+    payload["shadow_mode"] = shadow_mode
     if setup.status == "ready":
         return payload, "completed", None
     if setup.status == "skip":
@@ -436,17 +459,27 @@ def _execute_auto_populate_materials(
     state: WorkflowGraphState,
     node_params: dict[str, Any],
 ) -> tuple[dict[str, Any], str, str | None]:
-    del workflow_context, node_params
+    del node_params
     if state.setup is None or state.setup.cad_file is None:
         if state.setup is not None and state.setup.status == "skip":
             return _serialize_setup_result(state.setup), "skipped", state.setup.reason
         raise WorkflowGraphRuntimeError("auto_populate_materials requires a resolved cad_file")
-    result = auto_populate_materials_for_cad(session, str(state.setup.cad_file.id))
+    shadow_mode = workflow_context.execution_mode == "shadow"
+    if shadow_mode:
+        result = auto_populate_materials_for_cad(
+            session,
+            str(state.setup.cad_file.id),
+            persist_updates=False,
+        )
+    else:
+        result = auto_populate_materials_for_cad(session, str(state.setup.cad_file.id))
     state.auto_populate = result
-    if state.setup.order_line is not None and state.setup.order_line.product is not None:
+    if not shadow_mode and state.setup.order_line is not None and state.setup.order_line.product is not None:
         session.refresh(state.setup.order_line.product)
         state.setup.materials_source = state.setup.order_line.product.cad_part_materials or []
-    return _serialize_auto_populate_result(result), "completed", None
+    payload = _serialize_auto_populate_result(result)
+    payload["shadow_mode"] = shadow_mode
+    return payload, "completed", None
 
 
 def _execute_glb_bbox(