Struct FlowRunner 

pub struct FlowRunner { /* private fields */ }

High-level helper that orchestrates the common load → execute → save pattern.

FlowRunner provides a convenient wrapper around the lower-level graph execution API. It automatically handles session loading, execution, and persistence.

When to Use FlowRunner

• Web services: Execute one step per HTTP request
• Interactive applications: Step-by-step workflow progression
• Simple demos: Minimal boilerplate for common use cases

Performance

FlowRunner is lightweight and efficient:

• Creation cost: ~2 pointer copies (negligible)
• Memory overhead: 16 bytes (2 × Arc<T>)
• Runtime cost: Identical to manual approach

Examples

Basic Usage

use graph_flow::{FlowRunner, Graph, InMemorySessionStorage, Session, SessionStorage};
use std::sync::Arc;

let graph = Arc::new(Graph::new("my_workflow"));
let storage = Arc::new(InMemorySessionStorage::new());
let runner = FlowRunner::new(graph, storage.clone());

// Create a session first
let session = Session::new_from_task("session_id".to_string(), "start_task");
storage.save(session).await?;

// Execute workflow step
let result = runner.run("session_id").await?;
println!("Response: {:?}", result.response);

Shared Runner Pattern (Recommended for Web Services)

use graph_flow::FlowRunner;
use std::sync::Arc;

// Application state
struct AppState {
    flow_runner: Arc<FlowRunner>,
}

impl AppState {
    fn new(runner: FlowRunner) -> Self {
        Self {
            flow_runner: Arc::new(runner),
        }
    }
}

// Request handler
async fn handle_request(
    state: Arc<AppState>,
    session_id: String,
) -> Result<String, Box<dyn std::error::Error>> {
    let result = state.flow_runner.run(&session_id).await?;
    Ok(result.response.unwrap_or_default())
}

Implementations

impl FlowRunner

pub fn new(graph: Arc<Graph>, storage: Arc<dyn SessionStorage>) -> Self

Create a new FlowRunner from an Arc<Graph> and any SessionStorage implementation.

Parameters

• graph - The workflow graph to execute
• storage - Storage backend for session persistence

Examples

use graph_flow::{FlowRunner, Graph, InMemorySessionStorage};
use std::sync::Arc;

let graph = Arc::new(Graph::new("my_workflow"));
let storage = Arc::new(InMemorySessionStorage::new());
let runner = FlowRunner::new(graph, storage);

With PostgreSQL Storage

use graph_flow::{FlowRunner, Graph, PostgresSessionStorage};
use std::sync::Arc;

let graph = Arc::new(Graph::new("my_workflow"));
let storage = Arc::new(
    PostgresSessionStorage::connect("postgresql://localhost/mydb").await?
);
let runner = FlowRunner::new(graph, storage);

pub async fn run(&self, session_id: &str) -> Result<ExecutionResult>

Execute exactly one task for the given session_id and persist the updated session.

This method:

1. Loads the session from storage
2. Executes the current task
3. Saves the updated session back to storage
4. Returns the execution result

Parameters

• session_id - Unique identifier for the session to execute

Returns

Returns the same ExecutionResult that Graph::execute_session does, so callers can inspect the assistant’s response and the status (WaitingForInput, Completed, etc.).

Errors

Returns an error if:

• The session doesn’t exist
• Task execution fails
• Storage operations fail

Concurrency

Sessions are protected by optimistic locking: if another run call (or any other writer) saved the same session between this call’s load and save, the save fails with crate::GraphError::SessionConflict instead of silently losing the other writer’s update. On conflict, retry by calling run again (it reloads the latest session state). Note that the task’s side effects from the conflicting attempt are not rolled back, so tasks should be idempotent if concurrent runs are possible.

Examples

Basic Execution

let result = runner.run("test_session").await?;

match result.status {
    graph_flow::ExecutionStatus::Completed => {
        println!("Workflow completed: {:?}", result.response);
    }
    graph_flow::ExecutionStatus::WaitingForInput => {
        println!("Waiting for user input: {:?}", result.response);
    }
    graph_flow::ExecutionStatus::Paused { next_task_id, reason } => {
        println!("Paused, next task: {}, reason: {}", next_task_id, reason);
    }
}

Interactive Loop

loop {
    let result = runner.run("session_id").await?;
     
    match result.status {
        ExecutionStatus::Completed => break,
        ExecutionStatus::WaitingForInput => {
            // Get user input and update context
            // Then continue loop
            break; // For demo
        }
        ExecutionStatus::Paused { .. } => {
            // Continue to next step
            continue;
        }
    }
}

Error Handling

match runner.run("nonexistent_session").await {
    Ok(result) => {
        println!("Success: {:?}", result.response);
    }
    Err(GraphError::SessionNotFound(session_id)) => {
        eprintln!("Session not found: {}", session_id);
    }
    Err(GraphError::TaskExecutionFailed(msg)) => {
        eprintln!("Task failed: {}", msg);
    }
    Err(e) => {
        eprintln!("Other error: {}", e);
    }
}

Trait Implementations

impl Clone for FlowRunner

fn clone(&self) -> FlowRunner

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.