From 91f650c7639e9e18952d68f883abd3be2bcbbb7d Mon Sep 17 00:00:00 2001 From: ogabrielluiz Date: Mon, 15 Jun 2026 14:11:27 -0300 Subject: [PATCH] fix(api/v2): dedupe repeated inactive node deltas in AG-UI stream build.py keeps reporting a conditionally-excluded vertex in inactivated_vertices on every subsequent end_vertex (the excluded set persists until the ConditionalRouter clears it), so the translator was putting the same inactive STATE_DELTA on the wire once per remaining vertex. Track emitted inactive nodes and skip re-emitting; drop a node from the set when it actually runs again (build_start/end_vertex) so a loop re-activation can still re-emit inactive later. --- .../base/langflow/api/v2/agui_translator.py | 25 ++++++++-- .../tests/unit/api/v2/test_agui_translator.py | 46 +++++++++++++++++++ 2 files changed, 66 insertions(+), 5 deletions(-) diff --git a/src/backend/base/langflow/api/v2/agui_translator.py b/src/backend/base/langflow/api/v2/agui_translator.py index d6672da2bb..60c447e07d 100644 --- a/src/backend/base/langflow/api/v2/agui_translator.py +++ b/src/backend/base/langflow/api/v2/agui_translator.py @@ -88,6 +88,14 @@ class AGUITranslator: # Custom content blocks already emitted as CUSTOM events, mapped to a # fingerprint of their last-emitted payload so in-place updates re-emit. self._emitted_content_state: dict[str, str] = {} + # Nodes already emitted as ``inactive``. ``build.py`` keeps reporting a + # conditionally-excluded vertex in ``inactivated_vertices`` on every + # subsequent ``end_vertex`` (the excluded set persists until the router + # clears it), so without this the same inactive delta goes out once per + # remaining vertex. A node is dropped from this set when it actually runs + # again (``build_start``/``end_vertex`` for it), so a re-activation in a + # loop can re-emit ``inactive`` later. + self._inactivated_nodes: set[str] = set() def start(self) -> list[BaseEvent]: """Open the run. @@ -197,6 +205,8 @@ class AGUITranslator: node_id = data.get("id") if not node_id: return [] + # The node is running again, so a later exclusion may re-emit ``inactive``. + self._inactivated_nodes.discard(node_id) return [ StepStartedEvent(step_name=node_id), StateDeltaEvent(delta=[self._set_node(node_id, "running", None)]), @@ -209,17 +219,22 @@ class AGUITranslator: if not node_id: return [] status = "success" if build_data.get("valid") else "error" + # This node just ran, so a later exclusion may re-emit ``inactive``. + self._inactivated_nodes.discard(node_id) delta = [self._set_node(node_id, status, build_data.get("data"))] # A branch component (If-Else, Conditional Router) reports the vertices on # the not-taken branch in ``inactivated_vertices`` (already unioned with the # transitively excluded set in ``build.py``). The canvas seeds those nodes as # ``pending`` from ``vertices_sorted`` and never builds them, so without this # they stay stuck on ``pending`` instead of rendering as inactive. Mark each - # one ``inactive`` so the frontend maps it to ``BuildStatus.INACTIVE``. - delta.extend( - self._set_node(inactive_id, "inactive", None) - for inactive_id in build_data.get("inactivated_vertices") or [] - ) + # one ``inactive`` so the frontend maps it to ``BuildStatus.INACTIVE``, but + # only once: ``build.py`` re-reports the persisted excluded set on every + # later ``end_vertex``, so dedupe to avoid sending the same delta repeatedly. + for inactive_id in build_data.get("inactivated_vertices") or []: + if inactive_id in self._inactivated_nodes: + continue + self._inactivated_nodes.add(inactive_id) + delta.append(self._set_node(inactive_id, "inactive", None)) return [ StepFinishedEvent(step_name=node_id), StateDeltaEvent(delta=delta), diff --git a/src/backend/tests/unit/api/v2/test_agui_translator.py b/src/backend/tests/unit/api/v2/test_agui_translator.py index 0a29d94a20..6d237e8cfc 100644 --- a/src/backend/tests/unit/api/v2/test_agui_translator.py +++ b/src/backend/tests/unit/api/v2/test_agui_translator.py @@ -520,6 +520,52 @@ def test_end_vertex_marks_inactivated_vertices_inactive(): } +def test_inactivated_vertex_is_emitted_once_across_end_vertices(): + """``build.py`` keeps reporting the persisted excluded set on later end_vertex. + + The translator must emit each node's ``inactive`` delta only once so the first + run of the feature doesn't put the same redundant op on the wire repeatedly. + """ + t = AGUITranslator(run_id="r1", thread_id="t1") + t.start() + + first = t.translate( + "end_vertex", + {"build_data": {"id": "router", "valid": True, "data": None, "inactivated_vertices": ["node-b"]}}, + ) + # The next vertex still carries node-b in the persisted excluded set. + second = t.translate( + "end_vertex", + {"build_data": {"id": "node-a", "valid": True, "data": None, "inactivated_vertices": ["node-b"]}}, + ) + + def inactive_paths(out): + return [op["path"] for op in out[1].delta if op["value"]["status"] == "inactive"] + + assert inactive_paths(first) == ["/nodes/node-b"] + assert inactive_paths(second) == [] # already emitted, not repeated + + +def test_reactivated_vertex_can_emit_inactive_again(): + """A vertex that runs after being excluded (loop re-activation) may go inactive again.""" + t = AGUITranslator(run_id="r1", thread_id="t1") + t.start() + + excluded = t.translate( + "end_vertex", + {"build_data": {"id": "router", "valid": True, "data": None, "inactivated_vertices": ["node-b"]}}, + ) + assert [op["path"] for op in excluded[1].delta if op["value"]["status"] == "inactive"] == ["/nodes/node-b"] + + # node-b actually runs on a later pass, then gets excluded again. + t.translate("build_start", {"id": "node-b"}) + again = t.translate( + "end_vertex", + {"build_data": {"id": "router", "valid": True, "data": None, "inactivated_vertices": ["node-b"]}}, + ) + assert [op["path"] for op in again[1].delta if op["value"]["status"] == "inactive"] == ["/nodes/node-b"] + + def test_node_state_deltas_use_add_so_they_apply_without_vertices_sorted(): """build_start/end_vertex must not depend on vertices_sorted seeding the node.