Rust Valtron Usage

Rust Valtron Usage

Core Philosophy

Valtron is a progress-driven execution engine. Operations produce TaskIterators or StreamIterators that yield incremental results. The execution engine drives progress — callers decide when and where to synchronize.

The fundamental rule: Do not turn async operations into sync operations at the leaf. Instead, schedule work via execute(), return the stream to the caller, and let them decide when to collect results. This preserves composability, parallelism, and progress observability.

Blocking should happen at boundaries — the outermost point where a concrete value is actually needed — not inside every individual operation.

When Blocking Internally IS Acceptable

Exception to the fundamental rule: For single-value operations where the result is required immediately for subsequent operations, blocking internally is acceptable:

• Authentication checks — auth_check(), whoami() — need to know before proceeding
• Single-item lookups — getting user/org info needed for subsequent requests
• CRUD operations — create_repo(), delete_repo() — caller needs the result immediately

Pattern for single-value blocking:

use foundation_core::valtron::{from_future, execute, collect_one};

pub fn whoami(&self) -> Result<User> {
    let future = from_future(async move {
        // ... async work ...
        Ok::<_, Error>(user)
    });
    
    let stream = execute(future, None)?;
    collect_one(stream).ok_or_else(|| Error::NoResult)
}

Pattern for multi-value streaming:

pub fn list_models(&self) -> Result<impl StreamIterator<D = Result<ModelInfo>, P = ()> + Send> {
    let future = from_future(async move {
        // ... async work ...
        Ok::<_, Error>(vec_of_models)
    });
    
    let stream = execute(future, None)?;
    Ok(stream.flat_map_next(|result| {
        // Expand Vec into stream items
    }))
}

Decision flowchart:

1. Does the operation return multiple values? → Return a stream
2. Is the result needed immediately for subsequent operations? → Blocking internally is OK
3. Could the caller benefit from composing this with other operations? → Return a stream
4. Is this a one-shot initialization or CLI tool? → Blocking is acceptable

Default: Return streams. Block internally only when there's a clear justification.

Code Style: Clear, Simple, Succinct

Async first then sync via valtron calling async

Where ever possible, implementation is always in the async functions and then sync call valtron to run async code and return result, unless due to technical issues or unnecessary complexity should we clone and duplicate code for sync but this ok but rare where the complexity of calling the async via valtron is not worth it and reimplementing the logic for sync when the api already support syncs makes the most sense. This will probably mean two traits one for async and one for sync with the async one having methods ending with *_async to avoid conflict with the sync ones.

NOT A RULE BUT SOMETHING TO CONSIDER: If it makes sense to reduce duplication and one trait can have async methods that others implement with sync methods with default implementation great, or if duplication is causing bloat, we can just define a trait with both async and sync methods but i doubt rust like this. So use your judgement here.

IMPORTANT: In the desire to support sync and async versions, we may increase code size but i think its still worth it where sensible to ensure we can use these in sync and unsync context but if not we default to async versions and can provide wrappers that use valtron to call the async in a way that works for the users, meaning all things by default are async and just let the user know if its not possible and they can decide to call async code however they need in their different contexts.

IMPORTANT: Yes, remember there are caveats, this is why we need to be detailed, maybe sync cant wrap async due to issues like &mut self, we need to clearly articulate the design in the feature to clearly know its possible to do that.

Prefer direct, minimal combinator chains. Each combinator should have a clear purpose:

// GOOD: Only transform what needs transforming
let task = task
    .map_ready(|v| v * 2)      // transforming Ready
    .filter_ready(|v| v > 10); // filtering Ready

// BAD: Unnecessary map_pending that does nothing
let task = task
    .map_pending(|p| p)        // useless - remove
    .map_ready(|v| v * 2);

// GOOD: Use specific combinators for clarity
let stream = stream.filter_done(|v| v.is_ok());

// BAD: Over-engineered with unnecessary maps
let stream = stream
    .map_done(|v| v)           // useless identity map
    .map_pending(|p| p)        // useless identity map
    .filter_done(|v| v.is_ok());

Rules of thumb:

• Don't add map_pending unless you actually need to transform Pending values
• Don't chain multiple maps when one would do
• Use the most specific combinator (filter_done vs map_done + conditional)
• If a combinator doesn't change behavior, remove it

Combinator Usage: Prefer StreamIteratorExt Over Raw Iterator

Never manually match all Stream variants — use StreamIteratorExt combinators that handle pass-through automatically.

Anti-Pattern: Manual Stream Matching

// BAD: Verbose, error-prone, manually handles all Stream variants
let mapped = raw_stream.filter_map(|s| match s {
    Stream::Next(Some(json_str)) => match serde_json::from_str::<V>(&json_str) {
        Ok(v) => Some(Stream::Next(Some(v))),
        Err(e) => {
            tracing::error!("Deserialization error: {e}");
            None
        }
    },
    Stream::Next(None) => Some(Stream::Next(None)),
    Stream::Pending(p) => Some(Stream::Pending(p)),      // boilerplate
    Stream::Init => Some(Stream::Init),                   // boilerplate
    Stream::Ignore => Some(Stream::Ignore),               // boilerplate
    Stream::Delayed(d) => Some(Stream::Delayed(d)),       // boilerplate
});

Problems:

• 15+ lines of boilerplate just to transform Next
• Must manually update if Stream adds new variants
• Hard to read — logic buried in match arms

Correct: Use map_done Combinator

// GOOD: map_done automatically passes through Pending, Init, Ignore, Delayed
let mapped = raw_stream.map_done(|opt_json| {
    opt_json.and_then(|json_str| {
        serde_json::from_str::<V>(&json_str)
            .map_err(|e| tracing::error!("Deserialization error: {e}"))
            .ok()
    })
});

Benefits:

• 5 lines instead of 15+
• Future-proof — new Stream variants handled automatically
• Clear intent — "transform Next values, pass through rest"

When map_pending IS Necessary

Only use map_pending when you need to change the Pending type:

// GOOD: Converting Pending type for type compatibility
let stream = raw_stream
    .map_done(|v| v * 2)
    .map_pending(|p| MyPending::from(p));  // Type conversion needed

// BAD: Unnecessary map_pending — Pending type doesn't matter
let stream = raw_stream
    .map_done(|v| v * 2)
    .map_pending(|_| ());  // Useless unless type requires it

Key insight: At collection boundaries (find_map, collect_result), only Stream::Next values are extracted. Pending is progress information that gets discarded. Don't transform it unless the type signature requires it.

Common Transformations

// Deserialize JSON in Next values
let stream = stream.map_done(|json_opt| {
    json_opt.and_then(|s| serde_json::from_str(&s).ok())
});

// Convert Result to Option, log errors
let stream = stream.map_done(|result| {
    result.map_err(|e| tracing::error!("Error: {e}")).ok()
});

// Extract field from Next value
let stream = stream.map_done(|user| user.name);

The Execution Pipeline

create future/task
    → apply TaskIterator combinators (map_ready, map_pending, filter_ready, etc.)
    → execute() — schedules on thread pool, returns StreamIterator
    → apply StreamIterator combinators (map_done, filter_done, collect, etc.)
    → return stream to caller
    → caller composes multiple streams
    → caller collects at boundary (Iterator::find_map, sync_one, sync_all, etc.)

Every step is optional. The key insight: separating launch from collection enables parallelism even with heterogeneous types.

Pattern 1: Methods Return Streams (The Default)

Methods that perform I/O should schedule the work and return the stream. The caller controls when to collect.

use foundation_core::valtron::{execute, from_future, Stream};

fn get<V: DeserializeOwned + Send + 'static>(
    &self, key: &str,
) -> StorageResult<impl Iterator<Item = Stream<Option<V>, ()>>> {
    let key = key.to_string();
    let conn = Arc::clone(&self.conn);

    let task = from_future(async move {
        let mut stmt = conn.prepare("SELECT value FROM kv_store WHERE key = ?").await?;
        let mut rows = stmt.query([key]).await?;
        match rows.next().await? {
            Some(row) => {
                let value: String = row.get(0)?;
                let deserialized: V = serde_json::from_str(&value)?;
                Ok::<_, BackendError>(Some(deserialized))
            }
            None => Ok(None),
        }
    });

    // Schedule the work — returns immediately, work runs on pool
    let stream = execute(task, None)
        .map_err(|e| StorageError::Backend(format!("Valtron scheduling failed: {e}")))?;
    Ok(stream)
}

Why StorageResult<impl Iterator<...>>: execute() returns a Result indicating whether the task was successfully scheduled. The StorageResult wraps that scheduling result. The stream itself carries the async operation's outcome as Stream::Next(value).

Caller Composes and Collects at Boundary

// Launch two independent lookups — both start executing immediately
let user_stream = db.get::<User>("users:alice")?;
let config_stream = db.get::<Config>("app:config")?;

// Both are running in parallel on the thread pool.
// Collect at the boundary — second may already be done.
let user_results = collect_result(user_stream);   // Vec<Option<User>>
let config_results = collect_result(config_stream); // Vec<Option<Config>>

Pattern 2: TaskIterator Composition Before Execute (Parallel Homogeneous Tasks)

When multiple tasks have the same output type, compose them with execute_collect_all for parallel execution with synchronized collection.

Reference implementation: bin/platform/src/gen_model_descriptors/mod.rs

// Each task is a Box<dyn TaskIterator<Ready=Vec<ModelEntry>, Pending=FetchPending, ...>>
let models_dev_task = create_fetch_task(&mut client, "models.dev", URL_A, parse_a)?;
let openrouter_task = create_fetch_task(&mut client, "openrouter", URL_B, parse_b)?;
let ai_gateway_task = create_fetch_task(&mut client, "ai-gateway", URL_C, parse_c)?;

// All three run in parallel — synchronize when all complete
let result_stream = valtron::execute_collect_all(
    vec![models_dev_task, openrouter_task, ai_gateway_task],
    None,
).expect("scheduling succeeded");

// Collect at boundary
for item in result_stream {
    if let Stream::Next(models) = item {
        all_models.extend(models.into_iter().flatten());
    }
}

Pre-Execute Combinators Shape the Task

Apply TaskIteratorExt combinators before execute() to transform the task's output:

let task = SendRequestTask::new(request, 5, pool, config)
    // Transform Ready: HttpResponse → Vec<ModelEntry>
    .map_ready(move |intro| match intro {
        RequestIntro::Success { stream, .. } => {
            let body = body_reader::collect_string(stream);
            parser(&body, source)
        }
        RequestIntro::Failed(e) => Vec::new(),
    })
    // Transform Pending: HttpPending → FetchPending
    .map_pending(move |p| FetchPending::from_http(p, source));

Pattern 3: Heterogeneous Parallel Execution

When tasks return different types, launch each individually — they still run in parallel:

// Launch all three — work begins immediately for each
let user_stream = execute(user_task, None)?;
let session_stream = execute(session_task, None)?;
let oauth_stream = execute(oauth_task, None)?;

// Collect at boundary — by the time we finish the first,
// the others may already be done
let user = collect_result(user_stream);       // Vec<Option<User>>
let session = collect_result(session_stream); // Vec<Option<Session>>
let oauth = collect_result(oauth_stream);     // Vec<Option<OAuthState>>

Pattern 4: Rich Pending Types (Per-Method Judgment)

The Pending type in Stream<D, P> carries progress information. Whether to use a rich type or () is a per-method judgment call based on whether the caller can do something useful with the progress state.

Simple get — no meaningful progress, use ():

fn get(&self, key: &str) -> StorageResult<impl Iterator<Item = Stream<Option<V>, ()>>>

Migration — caller wants progress:

#[derive(Debug, Clone)]
pub enum MigrationProgress {
    Applying { current: usize, total: usize, name: String },
    Verifying { migration: String },
}

fn migrate(&self) -> StorageResult<impl Iterator<Item = Stream<MigrationResult, MigrationProgress>>>

HTTP fetch — caller wants connection state:

#[derive(Debug, Clone)]
pub enum FetchPending {
    Connecting { source: &'static str },
    AwaitingResponse { source: &'static str },
}

fn fetch(&self, url: &str) -> Result<impl Iterator<Item = Stream<Response, FetchPending>>>

Guideline: Use () unless the operation is long-running or multi-step and the caller would benefit from observability. Don't force richness where it adds no value.

Sync Boundary Helpers (New Valtron Primitives)

These are the tools for collecting at boundaries. They are the only place where blocking should occur.

collect_result — Drain a Stream and Collect All Results

The primary boundary helper. Drains the entire stream, collecting every Stream::Next value. Blocks the calling thread until the stream is exhausted.

/// Blocks until stream is exhausted. Collects ALL `Stream::Next` values.
/// Works for single-value streams (Vec will have one item) and multi-value streams alike.
pub fn collect_result<D, P>(stream: impl Iterator<Item = Stream<D, P>>) -> Vec<D> {
    stream
        .filter_map(|s| match s {
            Stream::Next(v) => Some(v),
            _ => None,
        })
        .collect()
}

For multi-value streams (like list_keys or query), the Vec contains all results.

collect_one — Extract First Result from a Stream

For single-value operations (like get) where you know the stream produces exactly one Next:

/// Blocks until first `Stream::Next(value)`, returns it. Returns None if stream exhausts.
pub fn collect_one<D, P>(stream: impl Iterator<Item = Stream<D, P>>) -> Option<D> {
    stream.find_map(|s| match s {
        Stream::Next(v) => Some(v),
        _ => None,
    })
}

sync_collect_one — Execute Task, Return Single Value

The single-value counterpart to sync_one. Returns Result<T::Ready>, not Result<Vec<T::Ready>>:

/// Execute a single task, block until first result. Returns the value directly.
pub fn sync_collect_one<T>(task: T) -> GenericResult<T::Ready>
where
    T: TaskIterator + Send + 'static,
    T::Ready: Send + 'static,
    T::Pending: Send + 'static,
    T::Spawner: ExecutionAction + Send + 'static,
{
    let stream = execute(task, None)?;
    collect_one(stream).ok_or_else(|| /* error */)
}

sync_one — Execute Task and Block for All Results

For tasks that produce multiple values:

/// Execute a single task and block until complete. Collects ALL results.
pub fn sync_one<T>(task: T) -> GenericResult<Vec<T::Ready>>
where
    T: TaskIterator + Send + 'static,
    T::Ready: Send + 'static,
    T::Pending: Send + 'static,
    T::Spawner: ExecutionAction + Send + 'static,
{
    let stream = execute(task, None)?;
    Ok(collect_result(stream))
}

sync_all — Execute Multiple Tasks, Block Until All Complete

/// Execute multiple homogeneous tasks in parallel, block until all complete.
pub fn sync_all<T>(tasks: Vec<T>) -> GenericResult<Vec<T::Ready>>
where
    T: TaskIterator + Send + 'static,
    T::Ready: Send + 'static,
    T::Pending: Send + 'static,
    T::Spawner: ExecutionAction + Send + 'static,
{
    let stream = execute_collect_all(tasks, None)?;
    // execute_collect_all already buffers and collects all results internally.
    // It yields Stream::Pending(count) while in flight, then a single
    // Stream::Next(Vec<T::Ready>) when all complete.
    stream
        .find_map(|s| match s {
            Stream::Next(v) => Some(v),
            _ => None,
        })
        .ok_or_else(|| /* GenericError: no results produced */)
}

Between vs. At Boundaries

Between operations, use StreamIteratorExt combinators — they preserve Valtron's execution model without blocking:

// BETWEEN operations — StreamIteratorExt (non-blocking, preserves Stream protocol)
let transformed_stream = stream
    .map_done(|user| user.name)
    .filter_done(|name| !name.is_empty());

At boundaries, use standard Iterator methods — these block the thread and extract raw values:

// AT BOUNDARY — standard Iterator (blocking, extracts values)
let names: Vec<String> = transformed_stream
    .filter_map(|s| match s { Stream::Next(v) => Some(v), _ => None })
    .collect();

The !Send Constraint

Many database crates (Turso, libsql) return row iterators that are !Send. These cannot cross the Valtron execution boundary. Consume them fully inside the async block, collecting into Vec<T> before the future returns.

// CORRECT: Collect inside async block — Vec<String> is Send
let task = from_future(async move {
    let mut rows = stmt.query([]).await?;
    let mut keys = Vec::new();
    while let Some(row) = rows.next().await? {
        keys.push(row.get::<String>(0)?);
    }
    Ok::<_, BackendError>(keys)
});
let stream = execute(task, None)?;

// WRONG: turso::Rows is !Send — this won't compile
let task = from_future(async move {
    stmt.query([]).await  // Returns Rows which is !Send
});

Using run_future_iter for Streaming !Send Iterators

When you need to stream rows (not collect to Vec), use run_future_iter to spawn a worker thread that owns the !Send iterator forever:

use foundation_core::valtron::{run_future_iter, Stream, ThreadedValue};

fn list(&self) -> Result<StateStoreStream<String>, StorageError> {
    let conn = Arc::clone(&self.conn);

    // run_future_iter spawns a worker thread that owns the !Send iterator
    let iter = run_future_iter(
        move || async move {
            // !Send rows iterator is created and consumed inside this async block
            let mut stmt = conn
                .prepare("SELECT id FROM deployment_resources ORDER BY id")
                .await
                .map_err(|e| StorageError::Backend(e.to_string()))?;
            let rows = stmt
                .query([libsql::Value::Null; 0])
                .await
                .map_err(|e| StorageError::Backend(e.to_string()))?;
            
            // Wrap the !Send iterator with a transformation function
            // The iterator stays on the worker thread forever
            Ok::<_, StorageError>(LibsqlRowsIterator::new(rows, |row| {
                row.get::<String>(0)
                    .map_err(|e| StorageError::SqlConversion(e.to_string()))
            }))
        },
        None,
        None,
    )
    .map_err(|e| StorageError::Backend(e.to_string()))?;

    // Transform ThreadedValue to Stream protocol
    let stream = iter.map(|threaded_value| match threaded_value {
        ThreadedValue::Value(result) => Stream::Next(result),
    });

    Ok(Box::new(stream))
}

Key pattern:

Why not from_future + collect? For large result sets, collecting to Vec before streaming causes OOM. run_future_iter enables true streaming: rows are fetched, transformed, and yielded one at a time across the thread boundary.

The Send + 'static Requirement

All data captured by async blocks must be Send + 'static for Valtron scheduling:

fn query(&self, sql: &str) -> StorageResult<impl Iterator<Item = Stream<Vec<Row>, ()>>> {
    let sql = sql.to_string();          // &str → owned String
    let conn = Arc::clone(&self.conn);   // Arc clone, not borrow

    let task = from_future(async move {
        // sql and conn are moved in — both Send + 'static
        conn.prepare(&sql).await?.query([]).await
    });
    let stream = execute(task, None)?;
    Ok(stream)
}

Turbo-Fish for Async Block Error Types

When the compiler can't infer the error type in an async block, annotate explicitly:

exec_future(async move {
    conn.execute_batch(&sql).await?;
    Ok::<_, turso::Error>(true)  // Turbo-fish needed
})?;

When Sync Blocking IS Acceptable

Some operations genuinely need to complete before anything else can proceed. Use sync_one or exec_future for:

• One-shot initialization — creating a DB connection, loading a model
• Migrations — must complete before the application starts serving
• CLI tools — where the entire program is sequential by nature

Even here, consider whether multiple initializations could run in parallel (e.g., connect to DB AND load config simultaneously).

Anti-Patterns

Anti-Pattern 1: Blocking at the Leaf (exec_future as Default)

// BAD: Every method blocks immediately — no parallelism possible
fn get(&self, key: &str) -> StorageResult<Option<V>> {
    exec_future(async move { /* ... */ })  // Blocks here!
}

fn set(&self, key: &str, value: &V) -> StorageResult<()> {
    exec_future(async move { /* ... */ })  // Blocks here!
}

// Caller has no choice — two sequential blocking calls
let user = db.get("user:1")?;      // blocks
let config = db.get("config")?;    // blocks (could have been parallel)

Anti-Pattern 2: Blocking in the Middle of a Chain

// BAD: Blocking mid-pipeline kills composability
let partial = collect_result(stream_a);  // blocks!
let transformed = transform(partial);
let final_stream = execute(make_task(transformed), None)?;
// The first block prevented us from overlapping work

Anti-Pattern 3: Using loop {} in StreamIterator::next() or TaskIterator::next_status

// BAD: Blocks the executor thread
impl StreamIterator for MyStream {
    fn next(&mut self) -> Stream<D, P> {
        loop {
            if let Some(val) = self.try_get() {
                return Stream::Next(val);
            }
        }
    }
}

// GOOD: Return control to the executor
impl StreamIterator for MyStream {
    fn next(&mut self) -> Stream<D, P> {
        match self.try_get() {
            Some(val) => Stream::Next(val),
            None if self.done => Stream::Init, // signal completion
            None => Stream::Ignore,            // yield back to executor
        }
    }
}

Spawner Type: Prefer BoxedSendExecutionAction over NoAction

When implementing TaskIterator, the Spawner associated type controls whether the task can spawn sub-tasks during execution. NoAction should be rarely used — it prevents the task from ever spawning work.

Default choice: Use BoxedSendExecutionAction for the Spawner type. This allows tasks to spawn sub-work if needed and is compatible with all executor functions.

use foundation_core::valtron::{BoxedSendExecutionAction, TaskIterator, TaskStatus};

// PREFERRED: Allows spawning sub-tasks
impl TaskIterator for MyTask {
    type Ready = MyResult;
    type Pending = MyProgress;
    type Spawner = BoxedSendExecutionAction;

    fn next_status(&mut self) -> Option<TaskStatus<Self::Ready, Self::Pending, Self::Spawner>> {
        // ...
    }
}

// AVOID: Only use when you are certain the task will never need to spawn
impl TaskIterator for SimplePureComputeTask {
    type Ready = u32;
    type Pending = ();
    type Spawner = NoAction;  // Rarely appropriate
    // ...
}

When NoAction is acceptable: Trivial tasks that are pure computation with no possibility of needing to delegate work (e.g., test fixtures, simple wrappers around from_future).

Valtron Quick Reference

Core Types

Executor Functions

Pre-Execute Combinators (TaskIteratorExt)

Applied before execute() — transforms the task itself:

Quick examples:

// Transform Ready values before execution
let task = task.map_ready(|v| v * 2);

// Filter out unwanted results (filtered items become Ignore)
let task = task.filter_ready(|v| v > 10);

// Stop immediately when seeing an error, returning it
let task = task.map_circuit(|status| match status {
    TaskStatus::Ready(Err(e)) => TaskShortCircuit::ReturnAndStop(TaskStatus::Ready(Err(e))),
    _ => TaskShortCircuit::Continue(status),
});

// Flat map Ready values
let task = task.flat_map_ready(|vec| vec.into_iter().map(TaskStatus::Ready));

// Split stream to observe first match while continuing
let (observer, continuation) = task.split_collect_one(|item| item.is_success());

Why map_circuit for error handling:

When a task should fail immediately on error, you need to communicate that error to the caller while stopping iteration. Without map_circuit, you'd need wrapper enums or lose the error:

// WITHOUT map_circuit — verbose, creates wrapper states
enum ResultWithDone<T> {
    Continue(T),
    Done(T),  // Need a separate variant just to signal "stop with this value"
}

let task = task.map_ready(|result| match result {
    Err(e) => ResultWithDone::Done(Err(e)),  // Custom enum needed
    Ok(v) => ResultWithDone::Continue(Ok(v)),
});

// WITH map_circuit — clear, succinct, no wrapper types
let task = task.map_circuit(|status| match status {
    TaskStatus::Ready(Err(e)) => TaskShortCircuit::ReturnAndStop(TaskStatus::Ready(Err(e))),
    _ => TaskShortCircuit::Continue(status),
});

The key insight: ReturnAndStop(value) returns the value AND stops — the caller receives the error via Stream::Next(error) and knows iteration is done. This preserves the error without inventing wrapper types.

Use map_circuit before execute() when:

• The task should terminate early on error
• You want to propagate the error to the stream consumer
• You need to stop immediately without creating wrapper enums

Post-Execute Combinators (StreamIteratorExt)

Applied after execute() — transforms the stream. Use these between operations, not at boundaries:

Quick examples:

// Transform Next values
let stream = stream.map_done(|v| v.to_string());

// Filter Next values (filtered items become Ignore)
let stream = stream.filter_done(|v| !v.is_empty());

// Stop immediately when seeing an error, preserving it for the caller
let stream = stream.map_circuit(|item| match item {
    Stream::Next(Err(e)) => ShortCircuit::ReturnAndStop(Stream::Next(Err(e))),
    _ => ShortCircuit::Continue(item),
});

// Chain combinators for clean pipelines
let stream = stream
    .filter_done(|v| v.is_ok())
    .map_done(|v| v.unwrap());

// Flat map Next values
let stream = stream.flat_map_next(|vec| vec.into_iter().map(Stream::Next));

// Split stream to observe first match while continuing
let (observer, continuation) = stream.split_collect_one(|item| matches!(item, Stream::Next(v) if v > 10));

Why map_circuit after execute():

When consuming a stream, you often want to stop on error while preserving the error for the caller. Without map_circuit, you'd need to return None and lose the error, or wrap results in custom enums:

// WITHOUT map_circuit — error is lost, caller can't distinguish
let stream = stream.filter_done(|v| v.is_ok());  // Errors become Ignore, swallowed

// OR: verbose wrapper enum
enum StreamValue<T> {
    More(T),
    Last(T),  // Extra variant just to say "stop with this"
}

// WITH map_circuit — error preserved, iteration stops cleanly
let stream = stream.map_circuit(|item| match item {
    Stream::Next(Err(e)) => ShortCircuit::ReturnAndStop(Stream::Next(Err(e))),
    Stream::Next(Ok(v)) => ShortCircuit::Continue(Stream::Next(Ok(v))),
    _ => ShortCircuit::Stop,
});

Use map_circuit after execute() when:

• The stream should terminate early on error
• You want to propagate errors to the final consumer
• You need clean error handling without wrapper enums

Boundary Collection (Standard Iterator — Use at Sync Points Only)

simple_http + Valtron Integration

When using foundation_core::simple_http with Valtron, follow the pattern from generated API clients:

Core Pattern: ClientRequestBuilder → SendRequestTask → StreamIterator

The correct pattern uses ClientRequestBuilder, build_send_request(), and RequestIntro handling:

use foundation_core::valtron::{execute, Stream};
use foundation_core::simple_http::client::SimpleHttpClient;
use foundation_core::simple_http::{RequestIntro, body_reader};

/// Single-value HTTP operation (blocking OK for auth/user info)
pub fn whoami(client: &SimpleHttpClient, token: &str) -> Result<User> {
    let client = client.clone();
    let token = token.to_string();
    let url = "https://huggingface.co/api/whoami-v2".to_string();
    
    // Start with ClientRequestBuilder
    let builder = client.get(&url)?
        .header("Authorization", format!("Bearer {}", token));
    
    // Build SendRequestTask and transform
    let task = builder
        .build_send_request()
        .map_err(|e| Error::RequestBuildFailed(e.to_string()))?
        .map_ready(|intro| match intro {
            RequestIntro::Success { stream, status } => {
                let headers = status.headers().clone();
                if !status.is_success() {
                    return Err(Error::HttpStatus {
                        code: status.as_u16(),
                        headers,
                    });
                }
                // Read body using body_reader helper
                let body = body_reader::collect_string(stream);
                // Parse JSON inside map_ready
                let user: User = serde_json::from_str(&body)
                    .map_err(|e| Error::Json(e.to_string()))?;
                Ok(user)
            }
            RequestIntro::Failed(e) => Err(Error::RequestSendFailed(e.to_string())),
        })
        .map_pending(|_| ());  // Discard progress info
    
    // Execute and collect single result
    let stream = execute(task, None)
        .map_err(|e| Error::Valtron(e.to_string()))?;
    
    // Use standard Iterator::find_map to extract first Next value
    stream.find_map(|s| match s {
        Stream::Next(result) => Some(result),
        _ => None,
    }).ok_or_else(|| Error::NoResult)
}

Multi-Value HTTP Operations (Return Stream)

For listing endpoints, return a StreamIterator that yields individual items:

use foundation_core::valtron::{execute, Stream};
use foundation_core::simple_http::{RequestIntro, body_reader};

pub type ApiStream<T> = Box<dyn Iterator<Item = Stream<Result<T, Error>, ()>> + Send>;

pub fn list_models(client: &SimpleHttpClient) -> Result<ApiStream<ModelInfo>> {
    let client = client.clone();
    let url = "https://huggingface.co/api/models".to_string();
    
    let builder = client.get(&url)?;
    
    let task = builder
        .build_send_request()
        .map_err(|e| Error::RequestBuildFailed(e.to_string()))?
        .map_ready(|intro| match intro {
            RequestIntro::Success { stream, status } => {
                let headers = status.headers().clone();
                if !status.is_success() {
                    return Err(Error::HttpStatus {
                        code: status.as_u16(),
                        headers,
                    });
                }
                let body = body_reader::collect_string(stream);
                // Parse Vec<ModelInfo> from response
                let models: Vec<ModelInfo> = serde_json::from_str(&body)
                    .map_err(|e| Error::Json(e.to_string()))?;
                Ok(models)
            }
            RequestIntro::Failed(e) => Err(Error::RequestSendFailed(e.to_string())),
        })
        .map_pending(|_| ());
    
    let stream = execute(task, None)
        .map_err(|e| Error::Valtron(e.to_string()))?;
    
    // Expand Vec into individual stream items using flat_map_next
    Ok(Box::new(stream.flat_map_next(|result| {
        match result {
            Ok(models) => models.into_iter().map(|m| Stream::Next(Ok(m))).collect::<Vec<_>>().into_iter(),
            Err(e) => vec![Stream::Next(Err(e))].into_iter(),
        }
    })))
}

Reference Implementation Pattern

From backends/foundation_deployment/src/providers/prisma_postgres/clients/mod.rs:

pub fn get_v1_compute_services_execute(
    builder: ClientRequestBuilder<SystemDnsResolver>,
) -> Result<
    impl S

…(truncated)