Streaming support

.gitignore

@@ -47,4 +47,6 @@ Thumbs.db
 
 # Project specific
 target
-TODO.md
+TODO.md
+
+*.wav

crates/stepflow-analysis/src/tracker.rs

@@ -87,6 +87,16 @@ pub struct DependencyTracker {
     completed: BitSet,
 }
 
+impl Clone for DependencyTracker {
+    fn clone(&self) -> Self {
+        Self {
+            dependencies: self.dependencies.clone(),
+            blocking: self.blocking.clone(),
+            completed: self.completed.clone(),
+        }
+    }
+}
+
 impl DependencyTracker {
     pub fn new(dependencies: Arc<Dependencies>) -> Self {
         let blocking = dependencies

crates/stepflow-builtins/src/eval.rs

@@ -84,6 +84,16 @@ impl BuiltinComponent for EvalComponent {
         let result_value = match nested_result {
             FlowResult::Success { result } => result.as_ref().clone(),
             FlowResult::Skipped => serde_json::Value::Null,
+            FlowResult::Streaming { stream_id, metadata, chunk, chunk_index, is_final } => {
+                // For streaming results, return the metadata and chunk info
+                serde_json::json!({
+                    "stream_id": stream_id,
+                    "metadata": metadata.as_ref(),
+                    "chunk": chunk,
+                    "chunk_index": chunk_index,
+                    "is_final": is_final
+                })
+            }
             FlowResult::Failed { error } => {
                 // Propagate the failure from the nested workflow
                 return Ok(FlowResult::Failed { error });

crates/stepflow-core/src/flow_result.rs

@@ -49,6 +49,19 @@ impl FlowError {
 pub enum FlowResult {
     /// The step execution was successful.
     Success { result: ValueRef },
+    /// The step is streaming data.
+    Streaming { 
+        /// Stream identifier
+        stream_id: String,
+        /// Metadata about the stream
+        metadata: ValueRef,
+        /// Base64 encoded chunk data
+        chunk: String,
+        /// Chunk index
+        chunk_index: usize,
+        /// Whether this is the final chunk
+        is_final: bool,
+    },
     /// The step was skipped.
     Skipped,
     /// The step failed with the given error.
@@ -57,6 +70,12 @@ pub enum FlowResult {
 
 impl From<serde_json::Value> for FlowResult {
     fn from(value: serde_json::Value) -> Self {
+        // First try to deserialize as a proper FlowResult
+        if let Ok(flow_result) = serde_json::from_value::<FlowResult>(value.clone()) {
+            return flow_result;
+        }
+
+        // If that fails, wrap in Success as fallback
         let result = ValueRef::new(value);
         Self::Success { result }
     }
@@ -70,6 +89,15 @@ impl FlowResult {
         }
     }
 
+    pub fn streaming(&self) -> Option<(String, ValueRef, String, usize, bool)> {
+        match self {
+            Self::Streaming { stream_id, metadata, chunk, chunk_index, is_final } => {
+                Some((stream_id.clone(), metadata.clone(), chunk.clone(), *chunk_index, *is_final))
+            }
+            _ => None,
+        }
+    }
+
     pub fn skipped(&self) -> bool {
         matches!(self, Self::Skipped)
     }

crates/stepflow-core/src/workflow/step.rs

@@ -3,7 +3,7 @@ use crate::schema::SchemaRef;
 use schemars::JsonSchema;
 
 /// A step in a workflow that executes a component with specific arguments.
-#[derive(serde::Serialize, serde::Deserialize, Debug, PartialEq, JsonSchema, utoipa::ToSchema)]
+#[derive(Clone, serde::Serialize, serde::Deserialize, Debug, PartialEq, JsonSchema, utoipa::ToSchema)]
 #[serde(rename_all = "camelCase")]
 pub struct Step {
     /// Optional identifier for the step
@@ -28,6 +28,10 @@ pub struct Step {
     /// Arguments to pass to the component for this step
     #[serde(default, skip_serializing_if = "ValueRef::is_null")]
     pub input: ValueRef,
+
+    /// Whether this step is a streaming step (doesn't persist results)
+    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
+    pub streaming: bool,
 }
 
 #[derive(
@@ -134,6 +138,7 @@ mod tests {
                 default_value: Some(ValueRef::from("fallback")),
             },
             input: serde_json::Value::Null.into(),
+            streaming: false,
         };
 
         let yaml = serde_yaml_ng::to_string(&step).unwrap();
@@ -152,6 +157,7 @@ mod tests {
             skip_if: None,
             on_error: ErrorAction::Fail,
             input: serde_json::Value::Null.into(),
+            streaming: false,
         };
 
         let yaml = serde_yaml_ng::to_string(&step).unwrap();

crates/stepflow-execution/src/error.rs

@@ -42,6 +42,8 @@ pub enum ExecutionError {
     ExecutionNotFound(Uuid),
     #[error("workflow '{0}' not found")]
     WorkflowNotFound(FlowHash),
+    #[error("streaming operation failed")]
+    StreamingError,
 }
 
 pub type Result<T, E = error_stack::Report<ExecutionError>> = std::result::Result<T, E>;