feat(nodes): add conditional if/else branching node (#628)

nodes/src/nodes/conditional/IGlobal.py

@@ -0,0 +1,42 @@
+# =============================================================================
+# MIT License
+# Copyright (c) 2026 Aparavi Software AG
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in
+# all copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+# =============================================================================
+
+from rocketlib import IGlobalBase, warning
+from ai.common.config import Config
+
+
+class IGlobal(IGlobalBase):
+    def beginGlobal(self):
+        config = Config.getNodeConfig(self.glb.logicalType, self.glb.connConfig)
+        # Python expression evaluated per entry
+        self.condition = config.get('condition', '')
+
+    def validateConfig(self):
+        """Validate configuration in the editor before pipeline runs."""
+        config = Config.getNodeConfig(self.glb.logicalType, self.glb.connConfig)
+
+        if not config.get('condition'):
+            warning('No condition defined')
+
+    def endGlobal(self):
+        self.condition = None

nodes/src/nodes/conditional/IInstance.py

@@ -0,0 +1,109 @@
+# =============================================================================
+# MIT License
+# Copyright (c) 2026 Aparavi Software AG
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in
+# all copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+# =============================================================================
+
+"""
+Conditional routing node.
+
+Catalog: ``conditional_text``, ``conditional_questions``, and ``conditional_answers``
+(one input lane each). Evaluates a sandboxed Python expression and routes to
+``then`` (true) or ``else`` (false).
+
+Chunks for a single object are buffered until ``close`` and the condition
+is evaluated once over the full accumulated payload, so an object is never
+split across both branches (e.g. text arriving as two chunks still matches
+a phrase that spans the chunk boundary).
+"""
+
+from __future__ import annotations
+
+from typing import Any, List
+
+from rocketlib import Entry, IInstanceBase, debug
+
+from .IGlobal import IGlobal
+
+
+class IInstance(IInstanceBase):
+    IGlobal: IGlobal
+
+    def open(self, object: Entry):
+        """Reset per-object buffers. Called once at the start of each object."""
+        self._text_chunks: List[str] = []
+        self._questions: List[Any] = []
+        self._answers: List[Any] = []
+
+    def writeText(self, text: str):
+        # Do not forward yet — the branch decision must be made over the full
+        # object payload, not per chunk.
+        self._text_chunks.append(text)
+
+    def writeQuestions(self, questions):
+        self._questions.append(questions)
+
+    def writeAnswers(self, answers):
+        self._answers.append(answers)
+
+    def close(self):
+        """
+        Evaluate the condition once over the full accumulated payload, then
+        flush buffered chunks to the selected branch. Only one of the three
+        lane buffers is populated in practice (each ``conditional_*`` service
+        declares a single input lane), but all three are handled defensively.
+        """
+        try:
+            if self._text_chunks:
+                full_text = ''.join(self._text_chunks)
+                self.instance.selectBranch(self._evaluate({'text': full_text}))
+                for chunk in self._text_chunks:
+                    self.instance.writeText(chunk)
+                self.instance.clearBranchSelection()
+
+            if self._questions:
+                self.instance.selectBranch(self._evaluate({'questions': self._questions}))
+                for q in self._questions:
+                    self.instance.writeQuestions(q)
+                self.instance.clearBranchSelection()
+
+            if self._answers:
+                self.instance.selectBranch(self._evaluate({'answers': self._answers}))
+                for a in self._answers:
+                    self.instance.writeAnswers(a)
+                self.instance.clearBranchSelection()
+        finally:
+            # Always release the branch selection, even if a write raised, so
+            # subsequent dispatches (e.g. close fan-out) are not misrouted.
+            self.instance.clearBranchSelection()
+
+    def _evaluate(self, scope: dict) -> int:
+        """Return branch index: 0 (then) if the condition is true, else 1."""
+        from ai.common.sandbox import execute_sandboxed
+
+        try:
+            code = f'result = bool({self.IGlobal.condition})'
+            output = execute_sandboxed(code, extra_globals=scope)
+            if output.get('exit_code', 1) != 0:
+                raise RuntimeError(f'sandbox error: {output.get("stderr", "unknown")}')
+            return 0 if output.get('result', False) else 1
+        except Exception as exc:
+            debug(f'conditional: evaluation error: {exc}')
+            return 1

nodes/src/nodes/conditional/services.answers.json

@@ -0,0 +1,49 @@
+{
+	"title": "Conditional (answers)",
+	"protocol": "conditional_answers://",
+	"classType": ["conditional"],
+	"capabilities": [],
+	"register": "filter",
+	"node": "python",
+	"path": "nodes.conditional",
+	"prefix": "conditional",
+	"description": ["Evaluates a Python condition on incoming answers and routes each item to one of two branches: ", "``then`` (true) or ``else`` (false). Use with downstream nodes that consume the ``answers`` lane."],
+	"icon": "util-conditional.svg",
+	"documentation": "https://docs.rocketride.org",
+	"lanes": {
+		"answers": ["then", "else"]
+	},
+	"input": [
+		{
+			"lane": "answers",
+			"description": "Answers input",
+			"output": [
+				{ "lane": "then", "description": "Routed when condition is true" },
+				{ "lane": "else", "description": "Routed when condition is false" }
+			]
+		}
+	],
+	"preconfig": {
+		"default": "python",
+		"profiles": {
+			"python": {
+				"title": "Python expression",
+				"condition": ""
+			}
+		}
+	},
+	"fields": {
+		"condition": {
+			"type": "string",
+			"title": "Python condition",
+			"description": "Evaluated in a sandbox with ``answers`` (object) in scope. Truthy → ``then``; falsy → ``else``."
+		}
+	},
+	"shape": [
+		{
+			"section": "Pipe",
+			"title": "Conditional (answers)",
+			"properties": ["condition"]
+		}
+	]
+}

nodes/src/nodes/conditional/services.questions.json

@@ -0,0 +1,49 @@
+{
+	"title": "Conditional (questions)",
+	"protocol": "conditional_questions://",
+	"classType": ["conditional"],
+	"capabilities": [],
+	"register": "filter",
+	"node": "python",
+	"path": "nodes.conditional",
+	"prefix": "conditional",
+	"description": ["Evaluates a Python condition on incoming questions and routes each item to one of two branches: ", "``then`` (true) or ``else`` (false). Use with downstream nodes that consume the ``questions`` lane."],
+	"icon": "util-conditional.svg",
+	"documentation": "https://docs.rocketride.org",
+	"lanes": {
+		"questions": ["then", "else"]
+	},
+	"input": [
+		{
+			"lane": "questions",
+			"description": "Questions input",
+			"output": [
+				{ "lane": "then", "description": "Routed when condition is true" },
+				{ "lane": "else", "description": "Routed when condition is false" }
+			]
+		}
+	],
+	"preconfig": {
+		"default": "python",
+		"profiles": {
+			"python": {
+				"title": "Python expression",
+				"condition": ""
+			}
+		}
+	},
+	"fields": {
+		"condition": {
+			"type": "string",
+			"title": "Python condition",
+			"description": "Evaluated in a sandbox with ``questions`` (object) in scope. Truthy → ``then``; falsy → ``else``."
+		}
+	},
+	"shape": [
+		{
+			"section": "Pipe",
+			"title": "Conditional (questions)",
+			"properties": ["condition"]
+		}
+	]
+}

nodes/src/nodes/conditional/services.text.json

@@ -0,0 +1,49 @@
+{
+	"title": "Conditional (text)",
+	"protocol": "conditional_text://",
+	"classType": ["conditional"],
+	"capabilities": [],
+	"register": "filter",
+	"node": "python",
+	"path": "nodes.conditional",
+	"prefix": "conditional",
+	"description": ["Evaluates a Python condition on incoming text and routes each item to one of two branches: ", "``then`` (true) or ``else`` (false). Use with downstream nodes that consume the ``text`` lane."],
+	"icon": "util-conditional.svg",
+	"documentation": "https://docs.rocketride.org",
+	"lanes": {
+		"text": ["then", "else"]
+	},
+	"input": [
+		{
+			"lane": "text",
+			"description": "Text input",
+			"output": [
+				{ "lane": "then", "description": "Routed when condition is true" },
+				{ "lane": "else", "description": "Routed when condition is false" }
+			]
+		}
+	],
+	"preconfig": {
+		"default": "python",
+		"profiles": {
+			"python": {
+				"title": "Python expression",
+				"condition": ""
+			}
+		}
+	},
+	"fields": {
+		"condition": {
+			"type": "string",
+			"title": "Python condition",
+			"description": "Evaluated in a sandbox with ``text`` (str) in scope. Truthy → ``then``; falsy → ``else``."
+		}
+	},
+	"shape": [
+		{
+			"section": "Pipe",
+			"title": "Conditional (text)",
+			"properties": ["condition"]
+		}
+	]
+}

nodes/test/test_contracts.py

@@ -599,10 +599,11 @@ class TestNodeLanes:
     def test_lane_has_valid_name(self, service: ServiceConfig, lane: LaneDefinition):
         """Verify lane names are valid."""
         valid_lanes = {
-            'text', 'documents', 'questions', 'answers', 
+            'text', 'documents', 'questions', 'answers',
             'table', 'image', 'audio', 'video',
             'classifications', 'classificationContext',
-            'tags', '_source'  # Special lanes
+            'tags', '_source',  # Special lanes
+            'then', 'else',  # Conditional branch outputs
         }
         assert lane.lane in valid_lanes or lane.lane.startswith('_'), \
             f"{service.test_id}: Unknown lane '{lane.lane}'"
@@ -616,9 +617,10 @@ def test_lane_outputs_valid(self, service: ServiceConfig, lane: LaneDefinition):
         """Verify output lane names are valid."""
         valid_lanes = {
             'text', 'documents', 'questions', 'answers',
-            'table', 
+            'table',
             'image', 'audio', 'video',
-            'classifications', 'classificationContext', 'tags'
+            'classifications', 'classificationContext', 'tags',
+            'then', 'else',  # Conditional branch outputs
         }
         for output in lane.outputs:
             assert output in valid_lanes, \

packages/client-typescript/src/client/types/pipeline.ts

@@ -31,6 +31,9 @@ export interface PipelineInputConnection {
 
 	/** Source component ID providing the data */
 	from: string;
+
+	/** From a conditional node: 0 = then, 1 = else. */
+	branch?: number;
 }
 
 /**