Parallel & Conditional Workflows

Fork/Join Model

Sayiir supports fork/join parallelism, allowing you to run multiple branches in parallel and collect their results in a join task. This pattern is ideal for independent operations that can execute concurrently, such as validating payment while checking inventory.

Basic Fork/Join

The fork/join pattern splits execution into multiple parallel branches, then merges the results:

Python

from sayiir import task, Flow, run_workflow

@task
def validate_payment(order: dict) -> dict:
    return {"payment": "valid"}

@task
def check_inventory(order: dict) -> dict:
    return {"stock": "available"}

@task
def finalize(results: dict) -> str:
    return f"Order complete: {results}"

workflow = (
    Flow("checkout")
    .fork()
        .branch(validate_payment)
        .branch(check_inventory)
    .join(finalize)
    .build()
)

result = run_workflow(workflow, {"order_id": 1})

Node.js

import { task, flow, branch, runWorkflow } from "sayiir";

const validatePayment = task("validate_payment", (order: Record<string, unknown>) => {
    return { payment: "valid" };
});

const checkInventory = task("check_inventory", (order: Record<string, unknown>) => {
    return { stock: "available" };
});

const workflow = flow<Record<string, unknown>>("checkout")
    .fork([
        branch("validate_payment", validatePayment),
        branch("check_inventory", checkInventory),
    ])
    .join("finalize", ([payment, inventory]) => {
        return `Order complete: payment=${JSON.stringify(payment)}, stock=${JSON.stringify(inventory)}`;
    })
    .build();

const result = await runWorkflow(workflow, { order_id: 1 });

Rust

let workflow = WorkflowBuilder::new(ctx)
    .with_registry()
    .then_task::<FetchOrderTask>()
    .fork()
        .add_task::<ValidatePaymentTask>()
        .add_task::<CheckInventoryTask>()
        .add_task::<CalculateShippingTask>()
    .join("finalize_order", |results: BranchOutputs<JsonCodec>| async move {
        let payment = results.get::<ValidatePaymentTask>()?;
        let inventory = results.get::<CheckInventoryTask>()?;
        let shipping = results.get::<CalculateShippingTask>()?;
        Ok(Order::finalize(payment, inventory, shipping))
    })
    .build()?;

Workflow Macro Syntax

Rust also supports a concise macro syntax for fork/join:

let wf = workflow! {
    name: "checkout",
    registry: registry,
    steps: [validate_payment || check_inventory, finalize]
};

The || operator denotes parallel execution, and , separates sequential steps including the join handler.

Multi-Step Branches

In Python, you can chain multiple steps within a single branch using .branch(step_a, step_b):

workflow = (
    Flow("complex_checkout")
    .fork()
        .branch(validate_payment, charge_card)
        .branch(check_inventory, reserve_items)
        .branch(calculate_shipping, book_courier)
    .join(finalize_order)
    .build()
)

Each branch executes its steps sequentially, but branches run in parallel with each other.

Join Handler

The join task receives a dictionary (Python), tuple of branch results (Node.js), or map (Rust) containing the results from all branches:

Python

@task
def finalize(results: dict) -> str:
    payment = results["validate_payment"]
    inventory = results["check_inventory"]
    shipping = results["calculate_shipping"]

    return f"Order processed: payment={payment}, stock={inventory}, shipping={shipping}"

Node.js

// Join receives a tuple of branch results in declaration order
.join("finalize", ([payment, inventory, shipping]) => {
    return `Order processed: payment=${JSON.stringify(payment)}, stock=${JSON.stringify(inventory)}, shipping=${JSON.stringify(shipping)}`;
})

Rust

.join("finalize_order", |results: BranchOutputs<JsonCodec>| async move {
    let payment: PaymentResult = results.get_by_id(ValidatePaymentTask::task_id())?;
    let inventory: InventoryResult = results.get_by_id(CheckInventoryTask::task_id())?;
    let shipping: ShippingResult = results.get_by_id(CalculateShippingTask::task_id())?;

    Ok(format!(
        "Order processed: payment={:?}, stock={:?}, shipping={:?}",
        payment, inventory, shipping
    ))
})

The join task only executes after all branches complete successfully. If any branch fails, the entire fork/join fails and triggers workflow retry logic.

Conditional Branching

While fork/join runs all branches in parallel, conditional branching routes execution to exactly one branch based on a runtime value. A key function extracts a routing key from the previous step’s output, and the matching branch runs. All languages now support declaring the set of valid keys upfront, enabling exhaustiveness checks at build time.

Python

from sayiir import task, Flow, run_workflow

@task
def classify(ticket: dict) -> str:
    return "billing" if ticket["type"] == "invoice" else "tech"

@task
def handle_billing(ticket: dict) -> str:
    return f"Billing resolved: {ticket['id']}"

@task
def handle_tech(ticket: dict) -> str:
    return f"Tech resolved: {ticket['id']}"

@task
def fallback(ticket: dict) -> str:
    return f"General: {ticket['id']}"

workflow = (
    Flow("support-router")
    .route(classify, keys=["billing", "tech"])
        .branch("billing", handle_billing)
        .branch("tech", handle_tech)
        .default_branch(fallback)
    .done()
    .build()
)

result = run_workflow(workflow, {"id": 1, "type": "invoice"})
# {"branch": "billing", "result": "Billing resolved: 1"}

Node.js

import { task, flow, runWorkflow } from "sayiir";

const classify = task("classify", (ticket: { id: number; type: string }) => {
  return ticket.type === "invoice" ? "billing" : "tech";
});

const handleBilling = task("handle-billing", (ticket: { id: number }) => {
  return `Billing resolved: ${ticket.id}`;
});

const handleTech = task("handle-tech", (ticket: { id: number }) => {
  return `Tech resolved: ${ticket.id}`;
});

const workflow = flow<{ id: number; type: string }>("support-router")
  .route(
    (ticket) => ticket.type === "invoice" ? "billing" : "tech",
    ["billing", "tech"] as const
  )
    .branch("billing", handleBilling)
    .branch("tech", handleTech)
    .defaultBranch("fallback", (ticket) => `General: ${ticket.id}`)
  .done()
  .build();

const result = await runWorkflow(workflow, { id: 1, type: "invoice" });
// { branch: "billing", result: "Billing resolved: 1" }

Rust

// Builder API — #[derive(BranchKey)] gives static type checking:
// the compiler rejects typos, missing branches, and orphan keys.
#[derive(BranchKey)]
enum Route {
    Billing,
    Tech,
}

let wf = WorkflowBuilder::new(ctx)
    .with_registry()
    .then_task::<ClassifyTask>()
    .route::<_, Route, _, _>( |input: ClassifyOutput| async move {
        Ok(if input.intent == "billing" { Route::Billing } else { Route::Tech })
    })
        .branch(Route::Billing, |sub| sub.then_task::<HandleBillingTask>())
        .branch(Route::Tech, |sub| sub.then_task::<HandleTechTask>())
        .default_branch(|sub| sub.then_task::<FallbackTask>())
    .done()
    .build()?;

// Or using the workflow! macro with typed keys
let wf = workflow! {
    name: "support-router",
    registry: registry,
    steps: [
        classify_ticket,
        route extract_intent -> Route {
            Billing => [handle_billing],
            Tech    => [handle_tech],
            _ => [fallback],
        }
    ]
};

// String-based keys also work (without `-> Type`)
let wf = workflow! {
    name: "support-router",
    registry: registry,
    steps: [
        classify_ticket,
        route extract_intent {
            "billing" => [handle_billing],
            "tech"    => [handle_tech],
            _ => [fallback],
        }
    ]
};

How It Works

1. Declare your keys upfront — keys= in Python, as const array in Node.js, #[derive(BranchKey)] enum in Rust. This is the recommended approach across all three languages: it catches typos and missing branches at build time
2. The key function receives the current step’s output and returns one of the declared keys
3. Sayiir matches the key against named branches — only the matching branch executes
4. If no branch matches and a default is set, the default runs; otherwise the workflow fails with BranchKeyNotFound
5. .done() performs exhaustiveness checks — missing or orphan branches are caught at build time. In Rust, #[derive(BranchKey)] enums give you compile-time safety: the type system rejects typos and missing cases before the code runs. The workflow! macro supports typed keys via route key_fn -> EnumType { Variant => [...] }, which generates a compile-time match exhaustiveness check
6. The output is a BranchEnvelope containing branch (the matched key) and result (the branch output)
7. The branch result is checkpointed — on resume, the key function is not re-evaluated

String keys for dynamic routing

All three languages also support plain string keys without upfront declaration — useful when routing keys are determined at runtime or workflows are built dynamically. In that case, omit keys= (Python), use a plain string[] instead of as const (Node.js), or use string-based route without -> EnumType (Rust). Exhaustiveness is not checked in this mode.

Multi-Step Branches

Each branch can contain a chain of tasks:

workflow = (
    Flow("order-router")
    .route(classify_order, keys=["express", "standard"])
        .branch("express", validate_express, ship_express, notify_customer)
        .branch("standard", validate_standard, ship_standard)
    .done()
    .then(send_confirmation)
    .build()
)

Combining with Fork/Join

Conditional branching composes naturally with other workflow primitives. You can branch after a fork/join, fork within a branch, or chain additional steps after a branch: