fix: address review findings from CodeRabbit + Copilot

Six real bugs + four hardening + two trivial fixes. Test coverage added
for every behavioral change.

Real bugs:

1. Cancelled background jobs were flipped to COMPLETED/FAILED.
   _buffer_background_run's finally block wrote a terminal status
   unconditionally; if stop_workflow set CANCELLED in parallel, the
   finally overwrote it. Added _finalize_job_status which re-reads the
   row and skips the update when the job is already CANCELLED.

2. Background scheduling failures left orphan buffers. If create_queue
   or start_job raised after _register_background_run, the buffer
   lingered with done=False forever. Wrapped scheduling in try/except,
   call _clear_background_run + finalize FAILED on the failure path.

3. Stop didn't abort the v2 SSE request. flowStore.stopBuilding aborts
   buildController, but buildFlow called runFlowAGUI without a signal.
   Threaded the controller's signal through runFlowAGUI and linked it
   to the agent's AbortController so Stop cancels the in-flight fetch.

4. runFlowAGUI set buildInfo only on RUN_FINISHED/RUN_ERROR. If the
   observable's complete fired without a terminal event (truncated
   stream, proxy timeout), buildInfo stayed null and trackFlowBuild
   read the run as success. Now sets a default failure buildInfo on
   silent complete and on abort.

5. useRunFlow's run() Promise never settled on abort. abort() and
   preempting run() left awaiters deadlocked because RxJS doesn't fire
   error/complete for those paths. Track resolver in a ref, settle on
   abort + preemption.

6. AGUITranslator emitted TextMessage* events with message_id="".
   token and add_message dropped events when the upstream payload
   omitted id, but still emitted START/CONTENT/END for the empty id,
   producing malformed AG-UI streams. Guard the emission paths.

Hardening:

7. state.ts shape validation: STATE_SNAPSHOT with nodes=null or nodes
   as array, STATE_DELTA with non-array delta — all treated as no-ops
   instead of crashing the reducer.

8. _BackgroundRun.frames bounded by _MAX_FRAMES_PER_BACKGROUND_RUN
   (10k). Token-streaming flows used to grow without bound;
   _MAX_BACKGROUND_RUNS only capped the number of buffers. Oldest
   frames are evicted; replay tracks a base_index so re-attach with
   Last-Event-ID past the evicted head starts from the new head.

9. _stream_event_frames inline queue bounded by _EVENT_QUEUE_MAX_SIZE
   (256). Slow consumer applies backpressure to the build loop
   instead of letting the queue grow unbounded.

10. applyStateDelta toggles edge animation in step with node status:
    running -> on, success/error -> off. Mirrors the v1 build path so
    edges animate during the run, not just clear at the end.

Trivial test tightening:

11. Last-Event-ID assertion in test_workflow_agui.py is CRLF-agnostic
    (split per line + check the id list).

12. Three weak "assert str(job_id) in result[...]" checks in
    test_workflow.py changed to exact equality.

Author: ogabrielluiz

src/backend/base/langflow/api/v2/agui_translator.py

@@ -117,7 +117,12 @@ def _translate_token(self, data: dict) -> list[BaseEvent]:
         token boundary) is dropped: re-opening it would emit a second
         ``TEXT_MESSAGE_START`` for an id the protocol considers closed.
         """
-        message_id = str(data.get("id", ""))
+        message_id = str(data.get("id") or "")
+        if not message_id:
+            # Without a stable id the AG-UI lifecycle (START/CONTENT/END) cannot
+            # be correlated. Dropping the event is preferable to emitting a
+            # malformed stream with empty message_ids.
+            return []
         if message_id in self._emitted_text_message_ids and self._open_message_id != message_id:
             return []
         chunk = data.get("chunk", "")
@@ -200,7 +205,10 @@ def _translate_add_message(self, data: dict) -> list[BaseEvent]:
             events.extend(self._close_open_message())
         else:
             text = data.get("text") or ""
-            if text and message_id not in self._emitted_text_message_ids:
+            # Skip text-message lifecycle emission without a stable message_id;
+            # tool-call events above are namespaced by block/content index so
+            # they can still ride a missing id, but TEXT_MESSAGE_* cannot.
+            if text and message_id and message_id not in self._emitted_text_message_ids:
                 self._emitted_text_message_ids.add(message_id)
                 events.append(TextMessageStartEvent(message_id=message_id, role="assistant"))
                 events.append(TextMessageContentEvent(message_id=message_id, delta=text))

src/backend/base/langflow/api/v2/workflow.py

@@ -517,7 +517,10 @@ async def _stream_event_frames(
     playground's chat-view. A follow-up retires this once chat-view
     consumes the AG-UI ``TEXT_MESSAGE_*`` lifecycle directly.
     """
-    queue: asyncio.Queue = asyncio.Queue()
+    # Bounded so a slow consumer can apply backpressure on the build loop
+    # instead of growing without bound. The build loop awaits ``queue.put``
+    # which yields control back to the consumer between frames.
+    queue: asyncio.Queue = asyncio.Queue(maxsize=_EVENT_QUEUE_MAX_SIZE)
     event_manager = create_default_event_manager(queue)
     input_request = _single_input_value_request(parsed)
     flow_data = FlowDataRequest(**parsed.data) if parsed.data else None
@@ -667,17 +670,32 @@ class _BackgroundRun:
     already serialized in the protocol the original POST requested via
     ``stream_protocol``; re-attach replays them as-is. Mixing protocols
     across a single run is not supported.
+
+    Per-run frame count is bounded by ``_MAX_FRAMES_PER_BACKGROUND_RUN`` so a
+    long verbose run (token-by-token streams, repeated tool calls) cannot
+    exhaust process memory while ``_MAX_BACKGROUND_RUNS`` only caps the
+    number of buffers. When the cap is reached the oldest frames are
+    evicted; re-attach with ``Last-Event-ID`` past that point will start
+    from the new buffer head (replay loss is preferred over OOM).
     """
 
     def __init__(self, user_id: str) -> None:
         self.user_id = user_id
         self.frames: list[bytes] = []
+        # Index of the first frame still in ``frames`` (monotonic across the
+        # life of the buffer). Once eviction starts, ``frames[i]`` corresponds
+        # to logical event id ``base_index + i``.
+        self.base_index = 0
         self.done = False
         self._cond = asyncio.Condition()
 
     async def append(self, frame: bytes) -> None:
         async with self._cond:
             self.frames.append(frame)
+            overflow = len(self.frames) - _MAX_FRAMES_PER_BACKGROUND_RUN
+            if overflow > 0:
+                del self.frames[:overflow]
+                self.base_index += overflow
             self._cond.notify_all()
 
     async def finish(self) -> None:
@@ -686,27 +704,71 @@ async def finish(self) -> None:
             self._cond.notify_all()
 
     async def replay(self, start_index: int) -> AsyncIterator[bytes]:
-        """Yield buffered frames from ``start_index`` and tail until done."""
+        """Yield buffered frames from ``start_index`` and tail until done.
+
+        ``start_index`` is in logical event-id space (matches what was emitted
+        on ``id:`` lines). If the caller's ``Last-Event-ID`` points before the
+        buffer's current head (frames evicted under memory pressure), we
+        replay from the head and the caller observes a gap.
+        """
         idx = max(start_index, 0)
         while True:
             async with self._cond:
-                while idx >= len(self.frames) and not self.done:
+                head = self.base_index
+                tail = head + len(self.frames)
+                idx = max(idx, head)
+                while idx >= tail and not self.done:
                     await self._cond.wait()
-                snapshot = self.frames[idx:]
+                    head = self.base_index
+                    tail = head + len(self.frames)
+                    idx = max(idx, head)
+                snapshot = self.frames[idx - head :]
                 finished = self.done
             for frame in snapshot:
                 yield frame
             idx += len(snapshot)
-            if finished and idx >= len(self.frames):
+            if finished and idx >= self.base_index + len(self.frames):
                 return
 
 
 # Process-local registry of background runs keyed by job_id, bounded by
 # ``_MAX_BACKGROUND_RUNS`` (oldest evicted first). Re-attach reads this.
 _MAX_BACKGROUND_RUNS = 100
+# Per-run frame ceiling. Caps memory for a single long/verbose run so a
+# token-streaming flow can't exhaust the process. 10k frames covers minutes
+# of dense token streams with room to spare; beyond that we evict oldest.
+_MAX_FRAMES_PER_BACKGROUND_RUN = 10_000
+# Inline stream queue between the build loop and the SSE consumer. Bounded
+# so a slow consumer applies backpressure to the build loop instead of
+# letting frames accumulate without bound when the network is slow.
+_EVENT_QUEUE_MAX_SIZE = 256
 _BACKGROUND_RUNS: dict[str, _BackgroundRun] = {}
 
 
+async def _finalize_job_status(job_uuid: UUID, terminal_status: JobStatus) -> None:
+    """Update job status to a terminal value, but never overwrite CANCELLED.
+
+    ``stop_workflow`` sets the job to CANCELLED. The buffer task runs in
+    parallel and reaches its ``finally`` block shortly after; if it
+    unconditionally wrote COMPLETED/FAILED it would race with the cancellation
+    and silently overwrite the user's stop intent. Re-read the row first and
+    skip the update if a cancellation already landed.
+    """
+    job_service = get_job_service()
+    try:
+        job = await job_service.get_job_by_job_id(job_id=job_uuid)
+    except Exception:  # noqa: BLE001
+        job = None
+    if job is not None and job.status == JobStatus.CANCELLED:
+        return
+    with contextlib.suppress(Exception):
+        await job_service.update_job_status(
+            job_uuid,
+            terminal_status,
+            finished_timestamp=True,
+        )
+
+
 async def _clear_background_run(job_id: str) -> None:
     """Pop the background run registry entry and wake any re-attach waiters.
 
@@ -780,12 +842,7 @@ async def _buffer_background_run(
         # Fire-and-forget coroutine: do not raise, the route already returned.
         await bg_run.finish()
         job_uuid = UUID(job_id) if isinstance(job_id, str) else job_id
-        with contextlib.suppress(Exception):
-            await get_job_service().update_job_status(
-                job_uuid,
-                JobStatus.FAILED,
-                finished_timestamp=True,
-            )
+        await _finalize_job_status(job_uuid, JobStatus.FAILED)
         return
 
     terminal_error_type = adapter.terminal_error_type
@@ -816,12 +873,7 @@ async def _buffer_background_run(
         # ``update_job_status`` queries the Job table by its UUID primary key;
         # passing the raw string would silently miss every row.
         job_uuid = UUID(job_id) if isinstance(job_id, str) else job_id
-        with contextlib.suppress(Exception):
-            await get_job_service().update_job_status(
-                job_uuid,
-                JobStatus.FAILED if errored else JobStatus.COMPLETED,
-                finished_timestamp=True,
-            )
+        await _finalize_job_status(job_uuid, JobStatus.FAILED if errored else JobStatus.COMPLETED)
         # Fire memory-base auto-capture hook on successful runs only. Matches
         # the sync mode wiring above and the v1 build-pipeline wiring in
         # ``api/build.py``. ``fire_and_forget_task`` because we are already a
@@ -868,19 +920,29 @@ async def execute_workflow_background(
         bg_run = _BackgroundRun(user_id=str(current_user.id))
         _register_background_run(job_id_str, bg_run)
 
-        queue_service = get_queue_service()
-        queue_service.create_queue(job_id_str)
-        queue_service.start_job(
-            job_id_str,
-            _buffer_background_run(
-                bg_run=bg_run,
-                flow=flow,
-                parsed=parsed,
-                job_id=job_id_str,
-                current_user=current_user,
-                stream_protocol=stream_protocol,
-            ),
-        )
+        try:
+            queue_service = get_queue_service()
+            queue_service.create_queue(job_id_str)
+            queue_service.start_job(
+                job_id_str,
+                _buffer_background_run(
+                    bg_run=bg_run,
+                    flow=flow,
+                    parsed=parsed,
+                    job_id=job_id_str,
+                    current_user=current_user,
+                    stream_protocol=stream_protocol,
+                ),
+            )
+        except BaseException:
+            # If queue creation or scheduling fails after the bg_run is
+            # registered, the buffer would stay live with ``done=False`` and
+            # any re-attach client would block on ``_cond.wait()`` forever
+            # (the task that would call ``finish()`` was never scheduled).
+            # Clear the registry, mark the job FAILED, then re-raise.
+            await _clear_background_run(job_id_str)
+            await _finalize_job_status(job_id, JobStatus.FAILED)
+            raise
         return WorkflowJobResponse(job_id=job_id_str, flow_id=parsed.flow_id, status=JobStatus.QUEUED)
 
     except (WorkflowResourceError, WorkflowServiceUnavailableError, WorkflowQueueFullError):

src/backend/tests/unit/api/v2/test_agui_translator.py

@@ -290,6 +290,41 @@ def test_end_vertex_does_not_close_open_text_message():
     assert more[0].message_id == "m1"
 
 
+def test_token_without_message_id_is_dropped():
+    """Token events with missing ``id`` cannot drive the AG-UI lifecycle.
+
+    A TextMessageStart/Content/End triple requires a stable id to correlate
+    chunks. Emitting events with ``message_id=""`` produces malformed
+    streams that some AG-UI clients reject; drop the event instead.
+    """
+    t = AGUITranslator(run_id="r1", thread_id="t1")
+    t.start()
+
+    no_id = t.translate("token", {"chunk": "hello"})
+    none_id = t.translate("token", {"chunk": "hello", "id": None})
+    empty_id = t.translate("token", {"chunk": "hello", "id": ""})
+
+    assert no_id == []
+    assert none_id == []
+    assert empty_id == []
+
+
+def test_add_message_without_message_id_skips_text_lifecycle():
+    """``add_message`` without an id must not emit TextMessage* events.
+
+    Tool-call sub-events can still ride a missing id (they namespace by
+    block/content index), but the text lifecycle needs a stable id.
+    """
+    t = AGUITranslator(run_id="r1", thread_id="t1")
+    t.start()
+
+    out = t.translate("add_message", {"text": "hello"})
+
+    assert all(not isinstance(e, TextMessageStartEvent) for e in out)
+    assert all(not isinstance(e, TextMessageContentEvent) for e in out)
+    assert all(not isinstance(e, TextMessageEndEvent) for e in out)
+
+
 def test_add_message_plain_text_emits_a_text_message():
     t = AGUITranslator(run_id="r1", thread_id="t1")
     t.start()

src/backend/tests/unit/api/v2/test_workflow.py

@@ -391,7 +391,7 @@ async def test_get_workflow_status_forbidden_for_other_user_job(
             assert response.status_code == 404
             result = response.json()
             assert result["detail"]["code"] == "JOB_NOT_FOUND"
-            assert str(job_id) in result["detail"]["job_id"]
+            assert result["detail"]["job_id"] == str(job_id)
         finally:
             async with session_scope() as session:
                 db_job = await session.get(Job, job_id)
@@ -425,7 +425,7 @@ async def test_get_workflow_status_allowed_for_own_job(
 
             assert response.status_code == 200
             result = response.json()
-            assert str(job_id) in result["job_id"]
+            assert result["job_id"] == str(job_id)
         finally:
             async with session_scope() as session:
                 db_job = await session.get(Job, job_id)
@@ -469,7 +469,7 @@ async def test_stop_workflow_forbidden_for_other_user_job(
             assert response.status_code == 404
             result = response.json()
             assert result["detail"]["code"] == "JOB_NOT_FOUND"
-            assert str(job_id) in result["detail"]["job_id"]
+            assert result["detail"]["job_id"] == str(job_id)
         finally:
             async with session_scope() as session:
                 db_job = await session.get(Job, job_id)

src/backend/tests/unit/api/v2/test_workflow_agui.py

@@ -430,7 +430,10 @@ async def test_reattach_replays_after_last_event_id(
         assert response.status_code == 200
         body = response.text
         # Event 0 is RUN_STARTED; with Last-Event-ID=0 it must not be replayed.
-        assert "id: 0\n" not in body
+        # Split per line so CRLF endings don't sneak the event through the
+        # substring check.
+        event_ids = [line.removeprefix("id:").strip() for line in body.splitlines() if line.startswith("id:")]
+        assert "0" not in event_ids
         assert "RUN_FINISHED" in body
 
     async def test_reattach_unknown_job_returns_404(
@@ -892,6 +895,70 @@ async def on_flow_output(self, **kwargs):
         assert call["job_id"] == _UUID(job_id)
 
 
+class TestBackgroundFinalizationGuards:
+    """Cancellation state + cleanup guarantees for the background path.
+
+    ``_buffer_background_run`` and ``execute_workflow_background`` must
+    preserve cancellation state and clean up after scheduling failures.
+    """
+
+    async def test_finalize_does_not_overwrite_cancelled_status(
+        self,
+        client: AsyncClient,
+        created_api_key,
+        chatbot_flow,
+    ):
+        """A run cancelled mid-flight must stay CANCELLED after the buffer ends.
+
+        Race: ``stop_workflow`` sets the job to CANCELLED. The buffer task's
+        ``finally`` block runs shortly after and previously wrote
+        COMPLETED/FAILED unconditionally, silently overwriting the user's
+        stop intent. Guarded by ``_finalize_job_status``.
+        """
+        import asyncio as _asyncio
+        from uuid import UUID as _UUID
+
+        from langflow.services.database.models.jobs.model import Job as _Job
+        from langflow.services.database.models.jobs.model import JobStatus as _JobStatus
+
+        headers = {"x-api-key": created_api_key.api_key}
+        start = await client.post(
+            "api/v2/workflows",
+            json=_agui_body(chatbot_flow, message="hi", mode="background"),
+            headers=headers,
+        )
+        assert start.status_code == 200
+        job_id = start.json()["job_id"]
+        job_uuid = _UUID(job_id)
+
+        # Stop the run before it gets a chance to complete on its own. The
+        # /stop endpoint flips the row to CANCELLED.
+        stop = await client.post(
+            "api/v2/workflows/stop",
+            json={"job_id": job_id},
+            headers=headers,
+        )
+        assert stop.status_code == 200
+
+        # Give the buffer task time to reach its finally block and call
+        # _finalize_job_status. Poll the row for stability.
+        for _ in range(60):
+            async with session_scope() as session:
+                row = await session.get(_Job, job_uuid)
+                if row is not None and row.status in (_JobStatus.COMPLETED, _JobStatus.FAILED):
+                    break
+            await _asyncio.sleep(0.1)
+
+        async with session_scope() as session:
+            row = await session.get(_Job, job_uuid)
+            assert row is not None
+            assert row.status == _JobStatus.CANCELLED, (
+                f"Buffer task overwrote the user's cancellation: got {row.status} "
+                f"(expected CANCELLED). The finally block in _buffer_background_run "
+                f"is racing with stop_workflow."
+            )
+
+
 class TestBackgroundModeStreamProtocol:
     """Background mode must honor ``stream_protocol`` end-to-end.