Conditional branching (IF/else) for RocketRide pipelines

[dsapandora]

We want true IF/else semantics in pipelines: at runtime, only one downstream branch runs for a given evaluation, instead of unintended fan-out to every child listening on the same lane.

Today, pipeline mode dispatches a named lane (Binder method: "text", "video", "questions", "documents", etc.) to all bound listeners for that method. We use text / writeText as examples only; the same fan-out issue applies to any lane that shares a name across two exclusive children (e.g. two edges both "video", both "questions").

Other media are not a weird fit—they already flow through the same pipeline and Binder::MethodNames. A conditional node can:

• Evaluate on data available early (tags, open, MIME/metadata), then only forward the relevant stream (writeVideo, writeQuestions, …) down the selected branch, or
• Accept a given input lane (e.g. video in) and declare branch outputs as different allowed methods if the product wants typed then/else paths (same trade-offs as the text-only workaround vs full engine exclusivity).

This document outlines the idea, a phased implementation, and where we expect to touch code in Python (node + catalog) and C++ (pipeline wiring / Binder dispatch).

---

Problem

• Pipeline edges use lane strings that must be members of Binder::MethodNames (see engLib/store/headers/binder.hpp).
• Binder::callMethods invokes every instance registered for that method when in pipeline mode (engLib/store/core/binder.cpp).
• So two components both connected with the same lane from the same parent (e.g. { "lane": "text", "from": "conditional_1" } on both, or the same for "video" / "questions") will both receive every matching write unless we change binding or use distinct lane names per branch (limited by downstream inputs).

flowchart LR
  subgraph today [Current semantics same lane]
    C[conditional_1]
    C -->|text| A[branch_A]
    C -->|text| B[branch_B]
  end

Loading

One dispatch on that lane (e.g. one writeText for "text") → A and B both run. The same pattern holds for video, questions, documents, etc., when two children share that lane name from one parent.

---

Goal

• Exclusive branch execution: after evaluating a condition, the engine routes the full relevant API stream (open, the appropriate write* calls for that lane type, close, etc.) to only the chosen child’s subgraph (or a defined default / error path)—not only for text.

Non-goals (initially)

• Arbitrary user Python inside condition strings in SaaS without a sandbox policy (product/security decision).
• Changing unrelated non-pipeline ingest paths.

---

Design idea (two layers)

1. Node / UX layer (Python + catalog)

• A conditional component appears in the pipeline editor.
• It holds configuration: how to evaluate the condition (expression DSL, JSON pointer, tag checks—TBD), error handling, optional else branch id or wiring convention.
• The node implementation evaluates the condition and signals the engine which branch is active or relies entirely on engine dispatch if the engine routes per-edge.

2. Engine layer (C++)

• Edge-scoped or branch-scoped dispatch: when a component has multiple outgoing edges that are marked as mutually exclusive (the conditional construct), the Binder must not register both targets under the same flat methodMap bucket for that lane (e.g. "text", "video", "questions") in the naive way.
• Concrete options (to decide in implementation):
  • Per-outgoing-edge binder keys internally (while preserving user-visible lane compatibility in validation), or
  • A router filter inserted by the pipeline compiler that owns the fan-out boundary, or
  • Slot / branch index in the pipeline graph metadata that callMethods consults.

The exact mechanism is an implementation detail; the invariant is: at most one exclusive child subgraph receives the event stream for that branch evaluation.

flowchart LR
  subgraph target [Target semantics]
    C2[conditional_1]
    C2 -->|if true| A2[branch_A]
    C2 -.->|if false| B2[branch_B]
  end

Loading

Only one of the solid paths is active per object / evaluation cycle.

---

Implementation phases

Phase A — Spike / RFC lock-in

• Confirm invariant with tests: same lane, two children, must not both fire when exclusivity is enabled.
• Choose dispatch strategy (above) and document migration for existing graphs (exclusive vs parallel fan-out must remain distinguishable).

Phase B — Engine (C++)

• Extend pipeline configuration model if needed (e.g. mark outgoing edges from a conditional provider as exclusiveGroup or equivalent—schema TBD).
• Implement dispatch so exclusive edges do not share a single multi-listener methodMap entry incorrectly.
• Primary files (expected touch list):
  • packages/server/engine-lib/engLib/store/pipeline/pipeline_config.cpp — validation, linking LaneInfo, possible new edge metadata.
  • packages/server/engine-lib/engLib/store/headers/binder.hpp — MethodNames only if we add new logical lanes (discourage unless necessary).
  • packages/server/engine-lib/engLib/store/core/binder.cpp — callMethods, binding, pipeline trace metadata.
  • Potentially filter/stack assembly under packages/server/engine-lib/engLib/store/ where pipeline components are instantiated and bound.

Phase C — Node (Python)

• New provider under nodes/src/nodes/ (name TBD, e.g. conditional / pipeline_if):
  • services.json — protocol, classType, lanes mapping (inputs → possible outputs; must align with validated graph).
  • IInstance.py — read incoming data, evaluate condition, cooperate with engine contract (e.g. set branch, or emit on engine-selected path only—depends on Phase B).
  • IEndpoint.py / IGlobal.py — only if following patterns from other filter nodes.
• Register in services catalog / packaging so the VS Code flow lists the node.
• Primary files (expected touch list):
  • nodes/src/nodes/<new_node>/services.json
  • nodes/src/nodes/<new_node>/IInstance.py
  • nodes/src/nodes/<new_node>/IEndpoint.py (if needed)
  • nodes/src/nodes/services-catalog.json (or generated catalog pipeline used by the repo)
  • Node tests under nodes/test/framework/ per README-node-testing.md

Phase D — UI / DX

• Flow editor: two handles or labeled outputs, validation messages when users mix exclusive and parallel semantics incorrectly.
• Docs: link from agents/ROCKETRIDE_PIPELINE_RULES.md and component reference when stable.

---

Interim workaround (without engine change)

If we shipped only a Python node without engine changes:

• Declare branches using different Binder method names (e.g. then → text / else → documents, or then → video / else → text if types and downstream inputs allow), and only call the matching write* for each lane.
• Limitation: two downstream components that only accept the same lane (e.g. both require questions, or both require video) cannot be made mutually exclusive without adapters or engine work.

This is not the long-term “go all in” design above, but documents the MVP escape hatch.

---

Risks

• Compatibility: existing pipelines rely on intentional fan-out; exclusivity must be opt-in per component or edge group.
• Tracing / debugging: pipeline trace events should show which branch was taken.
• Cross-platform: no OS-specific assumptions; behavior must match local engine and SaaS deployments.

---

References

• Binder::MethodNames — engLib/store/headers/binder.hpp
• callMethods / fan-out — engLib/store/core/binder.cpp
• Pipeline lane linking and validation — engLib/store/pipeline/pipeline_config.cpp
• Agent docs — ROCKETRIDE_PIPELINE_RULES.md, ROCKETRIDE_COMPONENT_REFERENCE.md

---

Related prior work

Exploratory implementation by Nihal Nihalani (GitHub: nihalnihalani):

[dsapandora]

Current binding flow (reference)

Understanding the current flow is critical. The pipeline binding happens in three stages:

Stage 1 — Validation (pipeline_config.cpp:71–376)

• Parses input edges from pipeline JSON
• Validates that lane ∈ Binder::MethodNames
• Builds internal LaneInfo / CompInfo graph
• Detects cycles, builds chain

Stage 2 — Connection building (stack.cpp:104–221)

• Reads validated config, builds connections vector:

  std::vector<std::tuple<int, int, std::string>> connections;
  //                     from, to,  lane

• Deduplicates by tuple equality

Stage 3 — Binding (endpoint.pipes.cpp:43–155)

• For each connection tuple, calls:

  (*pFrom)->binder.bind("open", pTo->get());    // auto-lifecycle
  (*pFrom)->binder.bind("closing", pTo->get());  // auto-lifecycle
  (*pFrom)->binder.bind("close", pTo->get());    // auto-lifecycle
  (*pFrom)->binder.bind(lane, pTo->get());        // data lane

• Binder::bind() appends pTo into methodMap[lane] vector

Stage 4 — Runtime dispatch (binder.cpp:90–171)

• callMethods(binder, "text", callback, trace) iterates all instances in methodMap["text"] and calls callback(instance) for each one
• This is where the fan-out happens — no filtering, no exclusivity

---

[dsapandora]

Design: Branch-indexed exclusive dispatch

Each edge from a conditional node gets assigned to an exclusive group identified by the conditional's component ID. Within a group, each edge has a branch index (0 = then, 1 = else, etc.). At runtime, the conditional node calls selectBranch(groupId, branchIndex) on its Binder, and callMethods only dispatches to instances whose branch matches.

Instances not in any exclusive group are dispatched normally (backwards compatible).

---

File-by-file changes

1. Pipeline JSON schema (new edge fields)

Two new optional fields on input edges:

{
  "id": "llm_branch_a",
  "provider": "llm_openai",
  "input": [
    {
      "lane": "text",
      "from": "conditional_1",
      "exclusiveGroup": "conditional_1",
      "branchIndex": 0
    }
  ]
}

{
  "id": "llm_branch_b",
  "provider": "llm_openai",
  "input": [
    {
      "lane": "text",
      "from": "conditional_1",
      "exclusiveGroup": "conditional_1",
      "branchIndex": 1
    }
  ]
}

Rules:

• Both fields must be present together or both absent (if one is present without the other → validation error)
• Within an exclusiveGroup, all edges must share the same from component
• Within an exclusiveGroup, branchIndex values must be unique
• branchIndex is a non-negative integer
• An edge without these fields behaves exactly as today (parallel fan-out)

---

2. pipeline_config.cpp — Validation

File: packages/server/engine-lib/engLib/store/pipeline/pipeline_config.cpp

2a. Extend LaneInfo struct (around line 115)

struct LaneInfo {
    static Text makeKey(TextView lane, TextView from, TextView to) noexcept {
        return _ts(from, "-[", lane, "]->", to);
    }

    Text name;
    CompInfo *from = nullptr, *to = nullptr;
    std::vector<LaneInfo *> outputs;
    bool visited = false;

    // --- NEW: exclusive branch support ---
    Text exclusiveGroup;   // empty = normal fan-out
    int branchIndex = -1;  // -1 = not exclusive
};

2b. Parse exclusive fields during input validation (around lines 185–227)

After validating lane and from, add:

// Parse optional exclusive branch fields
Text exclusiveGroup = input.lookup<Text>("exclusiveGroup");
int branchIndex = -1;

if (input.isMember("exclusiveGroup") || input.isMember("branchIndex")) {
    // Both must be present together
    if (!input.isMember("exclusiveGroup") || !input.isMember("branchIndex"))
        return APERR(Ec::InvalidParam, "Component", id,
                     "'exclusiveGroup' and 'branchIndex' must both be present");

    if (!input["exclusiveGroup"].isString() || !exclusiveGroup)
        return APERR(Ec::InvalidParam, "Component", id,
                     "'exclusiveGroup' must be a non-empty string");

    if (!input["branchIndex"].isInt() || input["branchIndex"].asInt() < 0)
        return APERR(Ec::InvalidParam, "Component", id,
                     "'branchIndex' must be a non-negative integer");

    branchIndex = input["branchIndex"].asInt();
}

2c. Cross-validate exclusive groups (after all edges are parsed)

Add a new validation block after the lane-linking block (after line 303):

// Validate exclusive groups
_block() {
    // group name → { from component, set of used branch indices }
    struct ExGroupInfo {
        Text fromId;
        std::set<int> usedIndices;
    };
    std::unordered_map<Text, ExGroupInfo> exclusiveGroups;

    for (auto &[_, lane] : lanes) {
        if (!lane.exclusiveGroup) continue;

        auto &group = exclusiveGroups[lane.exclusiveGroup];
        if (!group.fromId) {
            group.fromId = lane.from->id;
        } else if (group.fromId != lane.from->id) {
            return APERR(Ec::InvalidParam,
                         "Exclusive group", lane.exclusiveGroup,
                         "has edges from different source components:",
                         group.fromId, "and", lane.from->id);
        }

        if (!group.usedIndices.insert(lane.branchIndex).second) {
            return APERR(Ec::InvalidParam,
                         "Exclusive group", lane.exclusiveGroup,
                         "has duplicate branchIndex:", lane.branchIndex);
        }
    }
}

---

3. stack.cpp — Connection building

File: packages/server/engine-lib/engLib/store/stack.cpp

3a. Extend the connection tuple (around line 102)

Change the connections type from 3-tuple to 5-tuple:

// Before:
std::vector<std::tuple<int, int, std::string>> connections;

// After:
struct Connection {
    int fromIndex;
    int toIndex;
    std::string lane;
    std::string exclusiveGroup;  // empty = normal
    int branchIndex = -1;        // -1 = not exclusive

    bool operator==(const Connection &o) const {
        return fromIndex == o.fromIndex && toIndex == o.toIndex &&
               lane == o.lane;
    }
};
std::vector<Connection> connections;

Note: This requires updating the connections field declaration in the endpoint header where it lives. Look for the existing std::vector<std::tuple<int, int, std::string>> connections; declaration.

3b. Update buildConnections() (lines 170–188)

auto exclusiveGroup = comp.lookup<std::string>("exclusiveGroup");
int branchIndex = comp.isMember("branchIndex")
    ? comp["branchIndex"].asInt() : -1;

Connection conn{fromId, toId, lane, exclusiveGroup, branchIndex};
if (std::find(connections.begin(), connections.end(), conn) != connections.end())
    continue;

connections.emplace_back(conn);

---

4. binder.hpp — Binder extensions

File: packages/server/engine-lib/engLib/store/headers/binder.hpp

4a. Add exclusive dispatch tracking

class Binder {
private:
    // ... existing fields ...

    // NEW: track exclusive group membership per instance
    struct ExclusiveInfo {
        std::string group;
        int branchIndex;
    };
    std::unordered_map<IServiceFilterInstance *, ExclusiveInfo> m_exclusiveMap;

    // NEW: currently selected branch per group
    std::unordered_map<std::string, int> m_activeBranch;

public:
    // ... existing methods ...

    // NEW: bind with exclusive group info
    virtual Error bindExclusive(const std::string &methodName,
                                IServiceFilterInstance *pInstance,
                                const std::string &exclusiveGroup,
                                int branchIndex) noexcept;

    // NEW: select which branch is active for a group
    void selectBranch(const std::string &exclusiveGroup,
                      int branchIndex) noexcept;

    // NEW: clear branch selection (resets to fan-out)
    void clearBranchSelection(const std::string &exclusiveGroup) noexcept;
};

---

5. binder.cpp — Implementation

File: packages/server/engine-lib/engLib/store/core/binder.cpp

5a. New bindExclusive method

Error Binder::bindExclusive(const std::string &methodName,
                            IServiceFilterInstance *pInstance,
                            const std::string &exclusiveGroup,
                            int branchIndex) noexcept {
    // Standard bind first
    if (auto ccode = bind(methodName, pInstance)) return ccode;

    // Record exclusive membership
    m_exclusiveMap[pInstance] = {exclusiveGroup, branchIndex};
    return {};
}

5b. selectBranch / clearBranchSelection

void Binder::selectBranch(const std::string &exclusiveGroup,
                          int branchIndex) noexcept {
    m_activeBranch[exclusiveGroup] = branchIndex;
}

void Binder::clearBranchSelection(
    const std::string &exclusiveGroup) noexcept {
    m_activeBranch.erase(exclusiveGroup);
}

5c. Modify callMethods (lines 114–169)

Add a filter check inside the for loop, before the existing trace/callback logic:

for (auto *pInstance : *(it->second)) {
    // --- NEW: exclusive branch filtering ---
    auto exIt = pThis->m_exclusiveMap.find(pInstance);
    if (exIt != pThis->m_exclusiveMap.end()) {
        auto &info = exIt->second;
        auto branchIt = pThis->m_activeBranch.find(info.group);
        if (branchIt != pThis->m_activeBranch.end() &&
            branchIt->second != info.branchIndex) {
            // This instance belongs to a non-active branch — skip
            continue;
        }
        // If no branch selected yet for this group, skip all exclusive
        // instances in that group (defensive: condition not yet evaluated)
        if (branchIt == pThis->m_activeBranch.end()) {
            continue;
        }
    }
    // --- END NEW ---

    // ... existing trace + callback logic unchanged ...
}

Key behavior:

• If an instance is not in m_exclusiveMap → dispatched normally (backwards compatible)
• If an instance is exclusive and its branch is active → dispatched
• If an instance is exclusive and its branch is NOT active → skipped
• If no branch has been selected yet for a group → all exclusive instances in that group are skipped (safe default: no data leaks to either branch before condition is evaluated)

---

6. endpoint.pipes.cpp — Binding with exclusive info

File: packages/server/engine-lib/engLib/store/core/endpoint/endpoint.pipes.cpp

6a. Update connection iteration (lines 74–118)

for (auto &connection : this->connections) {
    auto fromIndex = connection.fromIndex;
    auto toIndex = connection.toIndex;
    auto &lane = connection.lane;
    auto &exclusiveGroup = connection.exclusiveGroup;
    auto branchIndex = connection.branchIndex;

    // ... existing pFrom/pTo resolution (lines 82-96) ...

    // Create a key using fromName + toName
    std::string connectionKey = std::to_string(toIndex);

    if (boundConnections.find(connectionKey) == boundConnections.end()) {
        // Lifecycle bindings: also exclusive-aware
        if (!exclusiveGroup.empty()) {
            if (auto ccode = (*pFrom)->binder.bindExclusive(
                    "open", pTo->get(), exclusiveGroup, branchIndex))
                return ccode;
            if (auto ccode = (*pFrom)->binder.bindExclusive(
                    "closing", pTo->get(), exclusiveGroup, branchIndex))
                return ccode;
            if (auto ccode = (*pFrom)->binder.bindExclusive(
                    "close", pTo->get(), exclusiveGroup, branchIndex))
                return ccode;
        } else {
            if (auto ccode = (*pFrom)->binder.bind("open", pTo->get()))
                return ccode;
            if (auto ccode = (*pFrom)->binder.bind("closing", pTo->get()))
                return ccode;
            if (auto ccode = (*pFrom)->binder.bind("close", pTo->get()))
                return ccode;
        }
        boundConnections[connectionKey] = true;
    }

    // Data lane binding
    if (!exclusiveGroup.empty()) {
        if (auto ccode = (*pFrom)->binder.bindExclusive(
                lane, pTo->get(), exclusiveGroup, branchIndex))
            return ccode;
    } else {
        if (auto ccode = (*pFrom)->binder.bind(lane, pTo->get()))
            return ccode;
    }
}

[dsapandora]

---

Lifecycle semantics for exclusive branches

conditional_1 evaluates condition during open(entry)
  ├── calls binder.selectBranch("conditional_1", 0)  // chose "then"
  ├── binder.open(entry)     → only branch_a gets open()
  ├── binder.writeText(...)  → only branch_a gets writeText()
  ├── binder.closing()       → only branch_a gets closing()
  └── binder.close()         → only branch_a gets close()
                                branch_b never activates

branch_b never receives open, so it never initializes, never allocates resources, and never needs cleanup. This is the safest default — no wasted work, no side effects on the unselected path.

The conditional node (Python) must call selectBranch before calling any binder.open(). This means condition evaluation happens either:

• During the node's own open() handler (based on tags/metadata available at open time), or
• On the first data event if the condition depends on content (in which case open must be deferred — see "Deferred evaluation" below)

---

[Rod-Christensen]

I don't quite understand why this needs any engine support or changes at all.

Say we setup a conditional if/then/else node, in python, the node itself should be able to make those decisions on it's own without rewiring the pipeline. For example, if the if/then node receives data on the writeText lane, it tests the condition and either returns None to continue routing through to the next stage of the pipeline or Ec::PreventDefault which will stop it from forwarding it on.

The else case seems like it would be easy, and internally actually is, you just route it to a different set of pipes. The biggest challenge is in the UI and the lane connectors. Do we add a "Then text" and "Else text" lanes displayed on the right side of the node, or do we limit the condition to a simple If/then statement. By limiting it to a simple If/then, there would be no UI changes or ambiguity, no engine changes. You could simulate an else by adding another node with the "NOT" of the original test.

From a UX standpoint, this is not ideal, since a traditional if/then/else would actually be two nodes on the canvas, but may be the best way to wire it up with the least number of code changes. It would just work pretty much as is.

Another challenge, do we expect users to know the internals of what the python code provides and use python code as the conditional, or do we use an NLP statement like "If text has the word red in it, skip it"? Can we have this NLP statement converted at init time to a true python statement that we could use our isolated/sandboxed Python node execute?

[dsapandora]

thanks @Rod-Christensen The reason I considered engine changes was the fan-out problem in callMethods, if a conditional node has two output branches on the same lane (e.g., both on text), the Binder dispatches to both with no way to make them mutually exclusive. The exclusiveGroup/branchIndex design solved that at the C++ dispatch level.

But reviewing your suggested approach, you're right — if the conditional node acts as an inline filter using PreventDefault (like anonymize, question, embedding_transformer, etc. already do), and the else is simulated with a second node using the negated condition, we never hit the exclusive fan-out topology. We avoid the problem instead of solving it in the engine.

I'll rework the plan: a pure Python node, no C++ changes, using the PreventDefault pattern that's already battle-tested across the codebase.

[dsapandora]

My original plan proposed exclusiveGroup / branchIndex / selectBranch() in the C++ Binder to support exclusive dispatch when two children share the same lane from a conditional node.

The problem is real — callMethods in binder.cpp:115 dispatches to ALL bound instances on a lane with no exclusion mechanism. But the architect's insight is: we never need that topology.

If the conditional node acts as an inline filter (like anonymize, question, embedding_transformer already do), it uses Ec::PreventDefault to stop data or return None to continue. No two children compete on the same lane from the same parent. We avoid the fan-out problem instead of solving it in the engine.

The chain in the engine:

1. Parent's binder.callMethods("text", ...) dispatches to the conditional node's writeText
2. The Python handler evaluates the condition
3. Return None → __checkCallParent (call.hpp:200) calls the C++ parent class → which calls binder.callMethods on the node's own binder → data flows to children
4. self.preventDefault() → __checkCallParent (call.hpp:203) resets error, returns false → parent class NOT called → data stops here

[dsapandora]

So I suggest:

Phase 1 — IF/THEN only (no UI changes, no engine changes)

A new Python filter node: conditional (or pipeline_if).

Pipeline topology:

Source → Conditional Node → Next Stage

The node receives data on any input lane (e.g., writeText). It evaluates its configured condition:

• Condition passes → return None (data continues to next stage)
• Condition fails → self.preventDefault() (data stops)

Simulating ELSE: add a second conditional node with the negated condition:

Source → IF(has keyword "red")     → Branch A
Source → IF(NOT has keyword "red") → Branch B

From a UX standpoint this is not ideal — a traditional if/then/else is two nodes on the canvas — but it works with zero code changes to engine or UI. It just works as-is.

---

[dsapandora]

@Rod-Christensen About your question, Do we expect users to know Python internals, or do we use NLP?

My Proposed approach: Both, via a conditionType config field

1. "python" — Direct Python expression

For advanced users who know the data model:

{
  "conditionType": "python",
  "condition": "'red' in text.lower()"
}

The expression is evaluated in the sandboxed Python environment (same sandbox as tool_python node — uses execute_sandboxed() from ai.common.sandbox). Variables available in the expression scope:

• text — the current text chunk (in writeText)
• entry — the Entry object from open() (tags, metadata, MIME type, etc.)
• documents — the document list (in writeDocuments)

2. "nlp" — Natural language statement

For non-technical users:

{
  "conditionType": "nlp",
  "condition": "If text has the word red in it, skip it"
}

1. During IGlobal.beginGlobal(), the NLP statement is sent to the pipeline's LLM
2. The LLM converts it to a Python expression (e.g., 'red' in text.lower())
3. The generated Python is validated (syntax check + dry run in sandbox)
4. The compiled expression is cached in IGlobal for the lifetime of the pipeline
5. At runtime (writeText, writeDocuments, etc.), the cached expression is evaluated per event

This means the LLM is called once at pipeline init, not per document. Runtime performance is pure Python evaluation.

Prompt template for NLP → Python conversion:

Convert the following condition to a Python boolean expression.
Available variables: text (str), entry (object with .tags, .mimeType, .metadata), documents (list).
Return ONLY the Python expression, nothing else.

Condition: "{user_condition}"

[dsapandora]

I can also think in "tags" — Tag-based condition

For simple metadata checks without any Python or NLP:

{
  "conditionType": "tags",
  "condition": {
    "hasTag": "confidential"
  }
}

Evaluated during open() against entry.tags. Pure Python dict/list lookup, no sandbox needed.

[dsapandora]

It will look something like this, and we will use flow positive and flow negative. But It will require two nodes from the same text lane, one positive and one negative

[dsapandora]

But I can think of a conditional node with two output lanes (Then and Else) visible in the UI
The pipeline builder UI already renders output lanes declared in services.json automatically. We only needed the C++ engine to recognize the new lane names.

Two lanes can be added to the binder: "then" and "else".

• MethodNames: 15 → 17 entries (2 additional strings)
• writeThen() / writeElse(): binder methods that dispatch to children on their respective lanes but call pInstance->writeText(text) — downstream nodes receive standard text without needing any changes

Files that I can modified:

File	Change
binder.hpp	"then" and "else" in MethodNames, writeThen() and writeElse() declarations
binder.cpp	writeThen() and writeElse() implementations
filter.hpp	cb_writeThen() and cb_writeElse() declarations
filter.callbacks.target.cpp	Callback implementations
bindings.cpp	.PYBIND(writeThen, ...) and .PYBIND(writeElse, ...)

The UI framework reads the input[].output[] section from each node's services.json and renders output connectors automatically. By declaring:

"output": [
    { "lane": "then", "description": "Text when condition is TRUE" },
    { "lane": "else", "description": "Text when condition is FALSE" }
]

The UI displays both connectors on the right side of the node with no additional code.

Result

A single node handles IF/THEN/ELSE. No duplicate nodes with opposite conditions, no frontend changes.

[kwit75]

Endorsing the two-lane single-node design with small C++ changes

@dsapandora — caught up on the full thread. The progression here is great and lands in the right place. Quick recap of where we are, then my decision and a few technical notes.

The arc

1. Original RFC — full exclusiveGroup / branchIndex / selectBranch() machinery in Binder. Solves the fan-out problem at the dispatch layer, but it's a meaningful architectural change and adds two new concepts (groups, branch state) that other nodes will never use.
2. Rod's pushback — "why does this need engine changes at all?" — PreventDefault + two opposite-condition nodes gets you there with zero engine work.
3. Your PreventDefault Phase 1 — works, but the UX is bad: a true if/then/else becomes two nodes the user has to keep in sync, and the negation is implicit.
4. Your final two-lane design (2026-04-09 12:09) — single node, two output lanes (then / else), small additive C++ work: 2 entries in MethodNames, writeThen/writeElse on the binder, matching cb_* in filter.callbacks.target.cpp, pybind bindings. UI gets the connectors for free from services.json.

This is the right answer. It's the smallest engine change that gives users the right mental model, and it composes cleanly with everything already in the codebase. Going with this.

Why I prefer it over the PreventDefault-only approach

• Single node = single source of truth for the condition. With two opposite nodes you have to maintain "X" and "NOT X" in lockstep. First time someone updates one and forgets the other, you get a silent gap or silent overlap and it's miserable to debug because the data just goes the wrong way with no error.
• Discoverability. New users see a node named conditional with then / else connectors and immediately understand it. They don't need to know the PreventDefault idiom or the negation convention.
• Tracing. When you writeThen() vs a generic filter that returned None, the trace tells you why a downstream node received data, not just that it did. Worth its weight in debugging time.

Why I prefer it over the original exclusiveGroup design

• The exclusiveGroup design generalizes a problem we don't actually have. Nothing else in the engine needs cross-edge mutual exclusion, and adding m_exclusiveMap + m_activeBranch to Binder makes every future Binder change harder to reason about. New lane names are an additive change; new dispatch state is not.
• The new lanes carry semantic meaning (then / else), not just a bookkeeping group. That's worth making them first-class.

Technical notes / things to nail down before you open the PR

1. MethodNames is a versioned API surface. Adding entries is fine, but please add them to the end of the array, never reorder. Anything that has serialized a method index (binder traces, on-disk pipeline metadata, recorded test fixtures) will break if text=4 becomes text=5. Add then and else as new tail entries.

2. writeThen / writeElse should dispatch to the downstream node's writeText, not require downstream nodes to implement writeThen/writeElse. You said this already in your comment but I want it explicit in the design: the new lanes are a routing concept on the producer side; consumers still see writeText. Otherwise every existing text consumer becomes incompatible the day this lands.

3. Lane coercion is one-directional. then / else → writeText on the consumer is fine. The reverse (a node bound on text sending to a then listener) should be a validation error in pipeline_config.cpp, not an implicit upcast — otherwise you can build a graph that bypasses the conditional entirely and it'll "work" with surprising semantics.

4. What about non-text lanes? The standup discussion mentioned video, questions, documents. Two options:

   • (a) Ship then / else as text-only and document the limitation. Add thenVideo / elseVideo etc. when there's actual demand. Smallest scope, fastest to ship.
   • (b) Make then / else polymorphic — the conditional node accepts any input lane and emits on the matching then* / else* lane. Bigger MethodNames footprint (8 new entries instead of 2) but symmetric and avoids a phase-2 retrofit.

   I'd ship (a) first and watch for actual user demand. The else* for media lanes is dead code until someone asks for it.

5. conditionType: "python" — sandbox path. Confirm execute_sandboxed() from ai.common.sandbox is the same path tool_python uses, and that the Python expression scope has no implicit access to the filesystem, network, or entry.metadata mutation. If you can't reach os.system from the expression scope, we're fine. If you can, that's a security regression and it needs to be locked down before this ships.

6. conditionType: "nlp" — defer to phase 2. I love the idea but it adds an LLM call to pipeline init, which means startup latency and a new failure mode (LLM unavailable → pipeline doesn't start). Also: "the LLM converts to a Python expression" is a prompt-injection surface — if the user's NLP statement contains "; import os; os.system('rm -rf /'); '", the conversion path needs to defend against that. Worth a separate RFC. Ship python and tags first.

7. conditionType: "tags" — yes, ship in phase 1. This is the highest-value, lowest-risk variant. Most real-world conditional routing is "is this confidential?" / "is this from this customer?" / "does this have tag X?" — pure dict lookup, no sandbox, no LLM, no expression evaluation. Make this the documented happy path in the node README and most users will never need python.

8. Binder trace event format. When a writeThen happens, the trace should record {conditional_id, branch: "then", condition_result: true} so the trace viewer can show the decision. Without this, debugging "why didn't my else branch fire?" is back to printf.

9. Error handling: condition raises an exception. Define this in the node spec, not as an implementation accident. Three reasonable defaults: (a) treat as false and route to else, (b) treat as PreventDefault and stop the pipeline branch, (c) propagate the exception and fail the pipeline. I'd pick (c) for python (so users notice broken expressions immediately) and (a) for tags / nlp (so a missing tag doesn't blow up the run). Make it configurable per node with a sensible default.

Suggested phase 1 scope

• New Python node nodes/src/nodes/conditional/ with services.json, IGlobal.py, IInstance.py, IEndpoint.py as needed.
• conditionType: "tags" and conditionType: "python" only. NLP deferred.
• Two output lanes: then, else. Text-only. Document the media-lane limitation.
• C++ touch list:
  • binder.hpp — append "then", "else" to MethodNames (end of array, not middle).
  • binder.cpp — writeThen() / writeElse() that dispatch to downstream writeText.
  • filter.hpp — cb_writeThen() / cb_writeElse() declarations.
  • filter.callbacks.target.cpp — the implementations.
  • bindings.cpp — .PYBIND entries.
  • pipeline_config.cpp — validation: a then/else lane producer must be a conditional node; consumer side accepts as text.
• Test coverage: at least one happy-path pipeline with both branches, plus a graph that should fail validation (then-lane producer that isn't a conditional).
• Trace format extension for branch decisions.

What I'd like before the PR

• Open this as feat(nodes): conditional if/then/else node against develop.
• Link this discussion in the PR description.
• I'll review as soon as it's up.

Nice work iterating on this so quickly. The final design is the right one.

[asclearuc]

Some comments

• conditional could be on any lane: text , answers, question and any other lane

• question - should conditional be available on tags lane ?

• there might be different types of conditionals even with contains :

  • does input text lane contain word red
  • does input question lane contain word red
  • does input answer contain word red
  • does any input lane contain word red

  You see, depending on what is conditional in the case above - execution may be routed to either to 'true' / 'false' execution path.

  In last scenario appears next question - when condition needs to be evaluated ? Consider for example

  • question arrives without word 'red'
  • answer arrives without word 'red'
  • text arrives and it contains word 'red'

  if condition is 'any input lane contains word 'red'' -> then the condition is true, and all that arrived before (question, answer, text) should be redirected to node

[dsapandora]

@asclearuc Great points, all three are valid:

1. Conditional on any lane — Agreed. The architecture supports this naturally. Phase 1 ships then/else as text-only (2 new MethodNames entries with coercion to writeText). Extending to other media lanes (thenVideo/elseVideo, thenDocuments/elseDocuments, etc.) is the same pattern — it's a Phase 2 scope expansion, not a design change. We'll document this as a known limitation.

2. Tags lane as input — Good distinction to make. conditionType: "tags" already evaluates Entry tags/metadata at open() time — it's lane-agnostic, so the condition itself doesn't need a tags input pipe. The deeper question is: should the conditional node be able to forward the tags lane to downstream nodes? Likely yes long-term, but not blocking for Phase 1 since tag-based routing already works via Entry metadata.

3. Different "contains" variants per lane — This maps well to conditionType: "python". The sandbox exposes different variables depending on the input lane context (text, documents, questions, entry). In Phase 1 we expose text + entry (tags/metadata). As we add multi-lane support in Phase 2, the same expression engine gets access to the additional variables — no redesign needed.

TL;DR — Phase 1 = text + tags evaluation only. The design is lane-agnostic by construction, so extending to other lanes is additive, not a retrofit. I should document the Phase 2 lane matrix so expectations are clear

[dsapandora]

Key differences

Aspect	feat/conditional-node-if-else	nihalnihalani/branch-node
Scope	C++ engine + Python + UI	Python + services.json only
Expressiveness	Any Python expression (unbounded)	Fixed set of rules
Safety model	Depends on Python sandbox	Curated rules, regex with timeout
Output lanes	Generic then / else over any original lane	Only routes between questions ↔ answers
Payload preservation	Yes — buffers and re-flushes the whole object	No — converts Q ↔ A losing enrichment
Tests	Only tweaks in test_contracts.py	Dedicated for profile
UI	Native rendering of then / else edges	No UI changes
Profile	Thought as a basic structure to later support LLM decision and other lane types	Specific profile defined

Comparison between two proposals that solve conditional data routing in a pipeline:

feat/conditional-node-if-else #649

What it implements: a conditional node with initial three variants — conditional_text, conditional_questions, conditional_answers. Each variant has one input lane and two output lanes: then / else.

Mechanics

• C++ engine changes: adds selectBranch(int) / clearBranchSelection() to Binder, new fields m_activeBranch and m_branchMap, a new filter callback cb_selectBranch, and a new Python binding. The "then" / "else" lanes are meta-lanes that the binder resolves dynamically to the actual data lane at runtime.
• Python: IInstance buffers writeText / writeQuestions / writeAnswers until close(), evaluates an arbitrary Python expression (via execute_sandboxed) over the full payload, then calls selectBranch(0|1) before re-dispatching the buffered chunks.
• Frontend: adds branch?: number to PipelineInputConnection, plus changes in QuickAddPopup, FlowGraphContext, graph.ts, and helpers.tsx to render the two output paths in the canvas.

nihalnihalani/feature/conditional-branch-node #528

What it implements: a single branch node. Input / output lanes use the existing native lanes (questions, answers). It routes between them and converts Question ↔ Answer carrying text only (history / documents / context are dropped on conversion).

Mechanics

• No C++ engine changes. Everything lives in Python.
• BranchEngine with a closed set of evaluators: contains, regex (with anti-ReDoS timeout), length, score_threshold, field_equals, sentiment (keyword-based classification), always_true / always_false. Rules are evaluated in order; first-match wins; default_lane is used when nothing matches.
• Five preconfigured profiles in the UI (keyword, regex, length, score, sentiment).

[asclearuc]

The main difference
- #528 is not branching, it is more like a transformation between question and answers depending on condition

[dsapandora]

Draft PR #690

Two-branch conditional router. Evaluates a Python expression per chunk — truthy → THEN, falsy/error/timeout → ELSE (fail-closed).

Contract

Register	filter
Lanes	{"*": ["*"]} (wildcard, same type in/out)
Branches	["then", "else"]
Evaluation	sandboxed Python, per chunk
Fields	condition (str), timeout_s (1–60s, default 5), branches (autopopulated by canvas)

Runtime flow

sequenceDiagram
    autonumber
    participant Up as Upstream node
    participant Eng as Engine (Binder)
    participant Py as flow_if_else (Python)
    participant Down as THEN / ELSE targets

    Up->>Eng: writeText(chunk)
    Eng->>Py: dispatch via callMethods
    Note over Py: AutoGatingMixin intercepts writeText<br/>→ _gate(chunk, 'text', forward)
    Py->>Py: IfElseDriver.evaluate(expression, bindings)
    alt truthy
        Py->>Py: branch = "then"
    else falsy / error / timeout
        Py->>Py: branch = "else"
    end
    Py->>Eng: setTargetFilter(targetId)
    Py->>Eng: forward(chunk) → writeText(chunk)
    Eng->>Down: callMethods skips all but targetId
    Py->>Eng: setTargetFilter("")
    Py->>Eng: preventDefault() — block default broadcast

Loading

Architecture

flowchart LR
    subgraph UI [Canvas - shared-ui]
      A[target-any]
      B[source-any-then]
      C[source-any-else]
    end

    subgraph Py [Python - flow_base + flow_if_else]
      D[AutoGatingMixin<br/>writeXxx overrides] --> E[_gate]
      E --> F[IfElseDriver.evaluate]
      F -->|True| G[setTargetFilter<br/>then nodeId]
      F -->|False| H[setTargetFilter<br/>else nodeId]
    end

    subgraph Cpp [Engine - C++]
      I[Binder.callMethods<br/>filter guard in loop]
      J[pipeline_config<br/>wildcard validator]
      K[services.getServiceSchemas<br/>exposes branches]
    end

    A --> D
    G --> I
    H --> I
    I --> B & C
    J -.serves.-> Py
    K -.feeds catalog.-> UI

Loading

Condition expression

Bindings inside the sandbox:

• payload — bound by lane name (text, image, audio, video, table, documents, questions, answers, classifications).
• state — PerChunkState, dict-like, scope = single chunk invocation.
• cond.* — pre-approved helpers:
  • cond.contains(text, keywords, mode='any'|'all', case_insensitive=True)
  • cond.regex(text, pattern, flags=0)
  • cond.length(text, min=, max=)
  • cond.score_threshold(score, op, threshold) — ops: >= <= == > <
  • cond.field_equals(obj, field, expected)
  • cond.sentiment(text) → 'positive'|'negative'|'neutral'
  • cond.always_true / cond.always_false
• Builtins (whitelist): abs all any bool dict enumerate filter float int len list map max min range round set sorted str sum tuple zip.

Blocked at parse time: import, def, class, lambda, assignments, try/except, raise, with, yield, await, dunder attribute access.

Examples:

cond.contains(text, ['error', 'fatal'], mode='any')
cond.regex(text, r'\d{4}-\d{2}-\d{2}') and cond.length(text, min=50)
cond.score_threshold(state.get('confidence', 0), '>=', 0.8)

C++ changes (+41 / −5 across 7 files, comments included; mostly additive)

Default state (no setTargetFilter + no "*" in lanes + no branches) is byte-for-byte identical to pre-change.

File	+/−	What
headers/binder.hpp	+10	m_targetFilter member + setTargetFilter setter
core/binder.cpp	+4	Guard in callMethods loop: skip instances when pipeType.id != m_targetFilter
headers/filter.hpp	+2	cb_setTargetFilter virtual decl
core/filter/filter.callbacks.target.cpp	+5	cb_setTargetFilter body — delegates to Binder
python/bindings.cpp	+1	PYBIND(setTargetFilter, …)
pipeline/pipeline_config.cpp	+15 / −5	Wildcard "*" support in lane validator
services/services.cpp	+4	Copy branches metadata to UI schema

Dispatch filter (binder.cpp)

if (!m_targetFilter.empty() && pInstance->pipeType.id != m_targetFilter) continue;

Empty default = broadcast. Non-empty = single-target dispatch. Python manages lifecycle (set → emit → clear).

Wildcard validator (pipeline_config.cpp)

Accepts services with lanes: {"*": [...]}. When the allow-list is ["*"] too, enforces input_lane == output_lane (blocks cross-type wiring like text → image). Non-wildcard services follow the original path unchanged.

Thread-safety

m_targetFilter is unsynchronized. The Binder assumes single-threaded dispatch per pipeline. Add a mutex or thread_local if multi-threaded dispatch is introduced.

UI changes (shared-ui)

Wildcard handle rendering (NodeLanes.tsx)

When a service declares lanes: {"*": [...]}:

• 1 input port target-any
• N output ports source-any-{branch} (one per branches entry) — or a single source-any if no branches

Connection validation (FlowGraphContext.tsx)

isValidConnection accepts the sentinel 'any' on either side as compatible with any concrete lane.

Handle ID formats:

• source-<lane> / target-<lane> — concrete
• source-<lane>-<branch> — branched source
• source-any[-<branch>] / target-any — wildcard

resolveEdgeLane() extracts concrete lane + branch from a handle pair for persistence.

Edge reconstruction on load (graph.ts)

getEdgesFromNodes() receives the service catalog and rebuilds handles as target-any / source-any-{branch} for wildcard providers, matching the rendered DOM.

QuickAdd (QuickAddPopup.tsx)

Wildcard-aware filtering: if the clicked handle or any candidate service is wildcard, the popup accepts the pairing.

Adding a new engine lane

Zero code changes in flow_if_else. When the engine adds a content type (e.g. writeJson):

1. IInstanceBase in rocketlib grows writeJson(value) -> None: pass.
2. AutoGatingMixin.__init_subclass__ re-scans on next import and synthesizes a gated writeJson override automatically.
3. Any service declaring the new lane gets a handle in the UI; target-any on flow_if_else accepts it without modification.

No release bump, no config change — the payoff of the {"*":["*"]} contract.

[asclearuc]

Thanks @dsapandora — one specific piece of the spec I want to push on:

"Two-branch conditional router. Evaluates a Python expression per chunk — truthy → THEN, falsy/error/timeout → ELSE (fail-closed)."

I think the "error/timeout → ELSE" half of that rule is operationally weaker than it looks, and it's worth discussing here before it gets locked in as the node's public contract.

As written, the rule collapses three distinct outcomes into the same branch:

1. Falsy condition — user's rule says "don't go down THEN." Intentional.
2. Sandbox error — user's expression is broken, or payload has an unexpected shape.
3. Timeout — engine couldn't evaluate at all, possibly ReDoS or pathological input.

Merging (1) with (2)/(3) has a cost that shows up later, in operations:

• Observability collapses. A pipeline silently hitting a 5% sandbox-error rate or timeout rate looks nominal — "some chunks went to ELSE, that's what ELSE is for." The operator only notices when downstream data quality degrades. Exceptions and metrics make the problem visible at the source; silent ELSE does not.
• "Fail-closed" in most contexts means deny AND log AND alert, not silently deny. The current rule borrows the security-term framing but drops the second and third parts. Fail-obscure might be a more honest label.
• Failures genuinely signal real problems — a condition that works in dev and explodes in prod on one chunk, a regex with catastrophic backtracking, a payload shape the author didn't anticipate. These all deserve surface area.

[github-actions[bot]]

[stepmikhaylov]

Feedback on implementation

All comments have been made for revision 2e7970d21925198857306e6ea16860ae6ef414c1

1. Overly complex Python syntax, aka condition expression

The syntax of Python expressions appears unintuitive and overly complex.
Why not add the current object and the parameters of the relevant lane to the execution context and use them directly? In my view, they are safe in terms of vulnerabilities related to malicious code injection.
For example, instead of:
cond.contains(text, ‘foo’) && cond.field_greater_or_equals(obj, ‘size’, 1048576)
define it as follows:
text.contains(‘foo’) && obj.size >= 1024 * 1024)

2. Implicit lane passing

The AutoGatingMixin class implicitly defines write<Lane> methods and the corresponding instance.write<Lane> calls at runtime to pass the lane to the next node. This seems overly complicated. We don’t have thousands of lanes, and we aren’t adding hundreds of lanes every month. Therefore, I would prefer to see lane methods defined explicitly – writeText -> instance.writeText, writeDocuments -> instance.writeDocuments, writeClassifications -> instance.writeClassification, and so on.

3. Branch routing in C++

A new member of the std::string Binder::m_targetFilter class and the C++ routing logic associated with it. I don’t see the need for these changes on the C++ side. It seems to me that the Python objects are sufficiently self-contained to store references to all branching nodes, check conditions, and pass the lane to the appropriate next node on the Python side.

4. Over-engineering nodes/flow_base

The code nodes/src/nodes/flow_base is not a node as such, but is essentially a library. A new class system is being introduced, essentially to evaluate Python expressions from strings at runtime. It seems to me that this class subsystem is over-engineered. It does not align with the current node architecture, not extensible within that context and can be simplified as described below.

Proposal for refactoring

All code snippets are simplified to clearly illustrate the concept.

1. Base class only with routing

Located in nodes/src/nodes/flow/base/.

class IInstance(IInstanceBase):
  """
  Base base class providing general flow logic (routing, branching, forwarding between flow instances).
  Declares `checkCondition()` as an abstract method, leaving the actual condition evaluation strategy to subclasses.
  The incoming lane methods use `checkCondition` to pass the lanes to appropriate next nodes.
  """

  @abstractmethod
  def checkCondition(condition: str, **kwargs) -> bool:
    """
    kwargs - appropriate lane arguments, e.g. 'text' of 'writeText', `documents` of `writeDocs` etc.
    """

  def writeText(self, text: str):
    if self.checkCondition(self, self.config.condition[1], text):
      self.flowInstances[1].writeText(text)
    elif self.checkCondition(self, self.config.condition[2], text):
      self.flowInstances[2].writeText(text)
    ...
    else:
      self.flowInstanceDefault.writeText(text)

2. Derived class with Python evaluation

Located in nodes/src/nodes/flow/pyeval/.

class IInstance(base.IInstance):
  """
  Extends the base class. Implements checkCondition() using Python's ast module to parse
  and evaluate condition strings at runtime against the current object's properties and any
  additional keyword arguments passed in.
  """
  
  def checkCondition(self, condition: str, **kwargs):
    import ast
    namespace = {
      "obj": self.currentObject,
      **kwargs
    }
    tree = ast.parse(condition)
    compiled = compile(tree)
    res = eval(compiled)
    return res

3. Another derived class with LLM evaluation

Not currently implemented, posted to demonstrate an approach for the potential future implementation.
Located in nodes/src/nodes/flow/llm/.

class IInstance(base.IInstance):
  """
  Extends the base class. Implements checkCondition() by delegating to a LLM provider:
  it packages the condition string and lane arguments (kwargs) into a question and sends it
  using writeQuestion(). Links a LLM instance via control composition.
  """

  def checkCondition(self, condition: str, **kwargs):
    """
    Send a condition and context args to linked LLM condition as a question.
    """
    question = Question()
    # Add current object and lane arguments to question
    question.addQuestion(...)
    self.instance.writeQuestion(question)
    ...

Diagram

[dsapandora]

@stepmikhaylov a quick summary of what landed vs. what didn't, and why.

What was implemented (matches your proposal)

• flow_base/IInstance.py → FlowBaseIInstance(IInstanceBase) with @abstractmethod checkCondition(condition: str, **kwargs) -> bool, exactly as you sketched. **kwargs carries the active lane's payload (text=... for writeText, etc.), so the base is lane-agnostic by construction.
• flow_pyeval/ — derived class with Python expression evaluation (your flow/pyeval/ location proposal, renamed flat for consistency with the rest of nodes/src/nodes/).
• flow_llm/ — second derived class that lets the condition be evaluated by an LLM via invoke, reusing the same base. Validates that the abstract surface really is generic.
• Lane handlers, AVI streaming buffer, and peer-direct branch dispatch (getInstance + acceptXxx) live in the base — subclasses only implement checkCondition. That keeps the contract minimal, as you suggested.

What's different from your proposal

• Two branches (then / else) instead of flowInstances[1..N] + flowInstanceDefault. The branches dict is {'then': [...], 'else': [...]} rather than an N-way elif chain inside the node.

• Reasons:

  • The team consensus (see Rod's earlier comment in Conditional branching (IF/else) for RocketRide pipelines #628) was to keep this a true binary IF/ELSE for v1, with N-way achievable today by chaining nodes. Your N-way design stays valid as a Phase 2 expansion — the base class accommodates it without breaking changes; only the dispatch loop and branches shape would grow.
  • Mutually-exclusive dispatch is now backed by the C++ setTargetFilter change (~41 LOC in binder.cpp, pipeline_config.cpp) that asclearuc already cleared in his review pass.

You can see the latest implementation here #690