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.

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),

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.