streamer45
diff --git a/‎Cargo.lock‎
Lines changed: 4 additions & 4 deletions b/‎Cargo.lock‎
Lines changed: 4 additions & 4 deletions
diff --git a/‎apps/skit/src/websocket_handlers.rs‎
Lines changed: 32 additions & 0 deletions b/‎apps/skit/src/websocket_handlers.rs‎
Lines changed: 32 additions & 0 deletions
diff --git a/‎apps/skit/tests/session_lifecycle_test.rs‎
Lines changed: 12 additions & 13 deletions b/‎apps/skit/tests/session_lifecycle_test.rs‎
Lines changed: 12 additions & 13 deletions
diff --git a/‎crates/core/src/state.rs‎
Lines changed: 11 additions & 0 deletions b/‎crates/core/src/state.rs‎
Lines changed: 11 additions & 0 deletions
@@ -519,6 +519,11 @@ async fn handle_add_node(
 
     {
         let mut pipeline = session.pipeline.lock().await;
+        if pipeline.nodes.contains_key(&node_id) {
+            return Some(ResponsePayload::Error {
+                message: format!("Node '{node_id}' already exists in the pipeline"),
+            });
+        }
         pipeline.nodes.insert(
             node_id.clone(),
             streamkit_api::Node { kind: kind.clone(), params: params.clone(), state: None },
@@ -1224,6 +1229,33 @@ async fn handle_apply_batch(
         });
     }
 
+    // Pre-validate duplicate node_ids against the pipeline model.
+    // Simulate the batch's Add/Remove sequence so that Remove→Add for
+    // the same ID within the batch is allowed, but duplicate Adds
+    // (without intervening Remove) are rejected before any mutation.
+    {
+        let pipeline = session.pipeline.lock().await;
+        let mut live_ids: std::collections::HashSet<&str> =
+            pipeline.nodes.keys().map(String::as_str).collect();
+        for op in &operations {
+            match op {
+                streamkit_api::BatchOperation::AddNode { node_id, .. } => {
+                    if !live_ids.insert(node_id.as_str()) {
+                        return Some(ResponsePayload::Error {
+                            message: format!(
+                                "Batch rejected: node '{node_id}' already exists in the pipeline"
+                            ),
+                        });
+                    }
+                },
+                streamkit_api::BatchOperation::RemoveNode { node_id } => {
+                    live_ids.remove(node_id.as_str());
+                },
+                _ => {},
+            }
+        }
+    } // Pipeline lock released after pre-validation
+
     // Validate permissions for all operations
     for op in &operations {
         if let streamkit_api::BatchOperation::AddNode { kind, params, .. } = op {
 
@@ -458,19 +458,15 @@ async fn test_session_destroy_shuts_down_pipeline() {
 
     println!("✅ Session created: {}", session_id);
 
-    // Add a source node (silence generator)
+    // Add a source node (audio::gain is a registered core node with in/out pins)
     let add_source_request = Request {
         message_type: MessageType::Request,
         correlation_id: Some("add-source".to_string()),
         payload: RequestPayload::AddNode {
             session_id: session_id.clone(),
             node_id: "source".to_string(),
-            kind: "silence".to_string(),
-            params: Some(json!({
-                "duration_ms": 10000, // 10 seconds
-                "sample_rate": 48000,
-                "channels": 2
-            })),
+            kind: "audio::gain".to_string(),
+            params: Some(json!({"gain": 1.0})),
         },
     };
 
@@ -488,15 +484,15 @@ async fn test_session_destroy_shuts_down_pipeline() {
 
     println!("✅ Added source node");
 
-    // Add a gain node
+    // Add a second gain node
     let add_gain_request = Request {
         message_type: MessageType::Request,
         correlation_id: Some("add-gain".to_string()),
         payload: RequestPayload::AddNode {
             session_id: session_id.clone(),
             node_id: "gain".to_string(),
-            kind: "gain".to_string(),
-            params: Some(json!({"gain": 1.0})),
+            kind: "audio::gain".to_string(),
+            params: Some(json!({"gain": 0.5})),
         },
     };
 
@@ -562,18 +558,21 @@ async fn test_session_destroy_shuts_down_pipeline() {
         ResponsePayload::Pipeline { pipeline } => {
             assert_eq!(pipeline.nodes.len(), 2);
 
-            // Check that nodes are in Running state (not Failed or Stopped)
+            // Both nodes use "audio::gain" which is a registered built-in
+            // node type.  With async creation they may still be in
+            // Creating state, or have progressed to Initializing/Ready.
             for (node_id, node) in &pipeline.nodes {
                 if let Some(state) = &node.state {
                     println!("Node '{}' state: {:?}", node_id, state);
                     assert!(
                         matches!(
                             state,
-                            streamkit_core::NodeState::Initializing
+                            streamkit_core::NodeState::Creating
+                                | streamkit_core::NodeState::Initializing
                                 | streamkit_core::NodeState::Ready
                                 | streamkit_core::NodeState::Running
                         ),
-                        "Node '{}' should be initializing/ready/running, got: {:?}",
+                        "Node '{}' should be creating/initializing/ready/running, got: {:?}",
                         node_id,
                         state
                     );
 
@@ -12,6 +12,8 @@
 //! Nodes transition through these states during their lifecycle:
 //!
 //! ```text
+//!      Creating
+//!          ↓
 //!     Initializing
 //!          ↓
 //!        Ready ──────────┐
@@ -89,6 +91,8 @@ impl From<String> for StopReason {
 /// Nodes transition through these states during their lifecycle:
 ///
 /// ```text
+///      Creating
+///          ↓
 ///     Initializing
 ///          ↓
 ///        Ready ──────────┐
@@ -105,6 +109,8 @@ impl From<String> for StopReason {
 /// ```
 ///
 /// ### Valid Transitions:
+/// - `Creating` → `Initializing` (node factory completed successfully)
+/// - `Creating` → `Failed` (node factory returned an error)
 /// - `Initializing` → `Ready` (source nodes) or `Running` (processing nodes)
 /// - `Ready` → `Running` (when pipeline is ready)
 /// - `Running` → `Recovering` (temporary issues, will retry)
@@ -120,6 +126,11 @@ impl From<String> for StopReason {
 #[derive(Debug, Clone, Serialize, Deserialize, TS)]
 #[ts(export)]
 pub enum NodeState {
+    /// Node is being created by the factory (e.g., loading ONNX models via FFI).
+    /// This state is set immediately when `AddNode` is received, before the
+    /// (potentially slow) constructor runs in a background task.
+    Creating,
+
     /// Node is starting up and performing initialization.
     /// Examples: Opening connections, loading resources, validating configuration.
     Initializing,