`collect` joins pre-fan-out outputs instead of only the preceding fan-out group

Author: bossjoker1Created Jun 23, 2026Updated Jun 23, 2026

Labelsbug

Description

StepMode::Collect currently joins outputs that were produced before the immediately preceding fan-out group.

According to docs/workflows.md, collect should gather outputs from the preceding fan-out group only. In the current engine implementation, collect joins the global all_outputs buffer in crates/openfang-kernel/src/workflow.rs, and that buffer already contains outputs from earlier sequential steps.

This changes the payload that downstream steps receive. In a workflow like:

sequential -> fan_out -> fan_out -> collect

the collect step does not just merge the two fan-out branch outputs. It also includes the earlier sequential output. That means later synthesis or decision steps can receive stale pre-fan-out context mixed into the branch merge, and multi-phase workflows can keep carrying forward extra content across later collect steps.

Expected Behavior

collect should only join outputs produced by the immediately preceding fan-out group.

For the reproduction below, the collected payload should be:

A PRE seed

---

B PRE seed

and should not include the earlier PRE seed sequential output.

Steps to Reproduce

Add this test to crates/openfang-kernel/src/workflow.rs (or an integration test under crates/openfang-kernel/tests/) and run:

bash

cargo test -p openfang-kernel collect_should_only_join_fanout_outputs -- --nocapture

rust

use chrono::Utc;
use openfang_kernel::workflow::{
    ErrorMode, StepAgent, StepMode, Workflow, WorkflowEngine, WorkflowId, WorkflowStep,
};
use openfang_types::agent::AgentId;

fn resolver(agent: &StepAgent) -> Option<(AgentId, String)> {
    match agent {
        StepAgent::ByName { name } => Some((AgentId::from_string(name), name.clone())),
        StepAgent::ById { id } => id.parse::<AgentId>().ok().map(|agent_id| (agent_id, id.clone())),
    }
}

#[tokio::test]
async fn collect_should_only_join_fanout_outputs() {
    let engine = WorkflowEngine::new();
    let workflow = Workflow {
        id: WorkflowId::new(),
        name: "collect-leak".to_string(),
        description: "".to_string(),
        steps: vec![
            WorkflowStep {
                name: "preprocess".to_string(),
                agent: StepAgent::ByName { name: "prep".to_string() },
                prompt_template: "PRE {{input}}".to_string(),
                mode: StepMode::Sequential,
                timeout_secs: 5,
                error_mode: ErrorMode::Fail,
                output_var: None,
            },
            WorkflowStep {
                name: "branch-a".to_string(),
                agent: StepAgent::ByName { name: "fan-a".to_string() },
                prompt_template: "A {{input}}".to_string(),
                mode: StepMode::FanOut,
                timeout_secs: 5,
                error_mode: ErrorMode::Fail,
                output_var: None,
            },
            WorkflowStep {
                name: "branch-b".to_string(),
                agent: StepAgent::ByName { name: "fan-b".to_string() },
                prompt_template: "B {{input}}".to_string(),
                mode: StepMode::FanOut,
                timeout_secs: 5,
                error_mode: ErrorMode::Fail,
                output_var: None,
            },
            WorkflowStep {
                name: "collect".to_string(),
                agent: StepAgent::ByName { name: "unused".to_string() },
                prompt_template: "unused".to_string(),
                mode: StepMode::Collect,
                timeout_secs: 5,
                error_mode: ErrorMode::Fail,
                output_var: None,
            },
        ],
        created_at: Utc::now(),
    };

    let workflow_id = engine.register(workflow).await;
    let run_id = engine.create_run(workflow_id, "seed".to_string()).await.unwrap();

    let output = engine
        .execute_run(run_id, resolver, |_agent_id, msg| async move {
            Ok::<_, String>((msg, 1, 1))
        })
        .await
        .unwrap();

    println!("{output}");
}

OpenFang Version

0.6.9

Operating System

Linux (x86_64)

Logs / Screenshots

Observed output from a fresh repro run:

probe=collect_leak
state=Completed
step_results=3
final_output<<EOF
PRE seed

---

A PRE seed

---

B PRE seed
EOF

Source: RightNow-AI/openfang

View original on GitHub View discussion on GitHub