feat: integrate memory-base + trackFlowBuild with v2 workflows

Two follow-up integrations:

memory-base on_flow_output hook:
  Release-1.10.0 wired the memory-base auto-capture hook into the v1
  build pipeline (api/build.py) so MemoryBases watching a flow get
  notified after every completed run. Wire the same hook into the v2
  endpoint's sync and background paths so MemoryBase auto-capture
  works for v2 workflows too. Background mode fires the hook on
  successful runs only, inside the buffered runner's finally block.
  fire_and_forget_task because we are already running in a background
  context and the hook must not block job-status finalization.

trackFlowBuild analytics:
  Release-1.10.0 wired trackFlowBuild into the v1 build callbacks
  (success -> isError=false, error -> isError=true + error list).
  Those callbacks were deleted with the v1 fallback. Read the bridge's
  final buildInfo after runFlowAGUI resolves and fire trackFlowBuild
  with the same signature as the v1 path:
  - (name, false, {flowId}) on success
  - (name, true, {flowId, error}) on failure

Tests:
  - Background-mode memory-base hook fires on a real chatbot run
    end-to-end (TestMemoryBaseHookBackgroundMode in test_workflow_agui).
  - buildFlow analytics tests stub runFlowAGUI and assert
    trackFlowBuild's args on success + failure.

Author: ogabrielluiz

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

@@ -84,7 +84,7 @@
 from langflow.services.database.models.flow.model import FlowRead
 from langflow.services.database.models.jobs.model import JobType
 from langflow.services.database.models.user.model import UserRead
-from langflow.services.deps import get_job_service, get_queue_service, get_task_service
+from langflow.services.deps import get_job_service, get_memory_base_service, get_queue_service, get_task_service
 
 # Configuration constants
 EXECUTION_TIMEOUT = 300  # 5 minutes default timeout for sync execution
@@ -429,6 +429,18 @@ async def execute_sync_workflow(
             stream=False,
         )
 
+        # Fire memory-base auto-capture hook — non-blocking background effect.
+        try:
+            _run_id_uuid = UUID(graph.run_id) if graph.run_id else None  # type-cast only; same run_id set on graph
+            await get_task_service().fire_and_forget_task(
+                get_memory_base_service().on_flow_output,
+                flow_id=flow.id,
+                session_id=execution_session_id,
+                job_id=_run_id_uuid,
+            )
+        except (RuntimeError, ValueError, OSError):
+            await logger.awarning("Memory base hook scheduling failed for flow %s", flow.id, exc_info=True)
+
         # Build RunResponse
         run_response = RunResponse(outputs=task_result, session_id=execution_session_id)
         # Convert to WorkflowExecutionResponse
@@ -810,6 +822,20 @@ async def _buffer_background_run(
                 JobStatus.FAILED if errored else JobStatus.COMPLETED,
                 finished_timestamp=True,
             )
+        # 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
+        # background coroutine and the hook must not block job finalization.
+        if not errored:
+            try:
+                await get_task_service().fire_and_forget_task(
+                    get_memory_base_service().on_flow_output,
+                    flow_id=flow.id,
+                    session_id=parsed.session_id or str(flow.id),
+                    job_id=job_uuid,
+                )
+            except (RuntimeError, ValueError, OSError):
+                await logger.awarning("Memory base hook scheduling failed for flow %s", flow.id, exc_info=True)
 
 
 async def execute_workflow_background(

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

@@ -811,6 +811,87 @@ async def __call__(self, *args, **kwargs):
         )
 
 
+class TestMemoryBaseHookBackgroundMode:
+    """The memory-base ``on_flow_output`` hook must fire after a background run.
+
+    Sync mode wires the hook directly inside ``execute_workflow_sync`` and the
+    v1 build pipeline wires it after ``end_all_traces`` in ``api/build.py``.
+    The v2 background mode buffers frames in ``_buffer_background_run`` and
+    must dispatch the same hook in its ``finally`` block on successful
+    completion. Without it, MemoryBase auto-capture silently misses every
+    background run.
+    """
+
+    async def test_background_run_fires_memory_base_hook_on_success(
+        self,
+        client: AsyncClient,
+        created_api_key,
+        chatbot_flow,
+        monkeypatch: pytest.MonkeyPatch,
+    ):
+        """Background run schedules ``on_flow_output``.
+
+        ``flow_id``, ``session_id``, and ``job_id`` must reach the hook.
+        """
+        from langflow.api.v2 import workflow as _workflow_module
+
+        captured: list[dict] = []
+
+        class _RecordingMemoryBaseService:
+            async def on_flow_output(self, **kwargs):
+                captured.append(kwargs)
+
+        monkeypatch.setattr(
+            _workflow_module,
+            "get_memory_base_service",
+            lambda: _RecordingMemoryBaseService(),
+        )
+
+        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"]
+
+        events = await client.get(f"api/v2/workflows/{job_id}/events", headers=headers)
+        assert events.status_code == 200
+        assert "RUN_FINISHED" in events.text
+
+        # Wait for the buffer task to finalize the job row.
+        import asyncio as _asyncio
+        from uuid import UUID as _UUID
+
+        from langflow.services.database.models.jobs.model import Job as _Job
+
+        for _ in range(100):
+            async with session_scope() as session:
+                row = await session.get(_Job, _UUID(job_id))
+                if row is not None and row.status.value in ("completed", "failed"):
+                    break
+            await _asyncio.sleep(0.1)
+
+        # The hook is fired via ``fire_and_forget_task`` so allow the loop a
+        # tick to drain the scheduled coroutine.
+        for _ in range(20):
+            if captured:
+                break
+            await _asyncio.sleep(0.05)
+
+        assert captured, (
+            "Memory-base on_flow_output was not called after a successful "
+            "background run. The hook is silently dropped for every "
+            "background-mode workflow run."
+        )
+        assert len(captured) == 1
+        call = captured[0]
+        assert call["flow_id"] == chatbot_flow
+        assert call["session_id"] == "thread-1"  # matches _agui_body session_id
+        assert call["job_id"] == _UUID(job_id)
+
+
 class TestBackgroundModeStreamProtocol:
     """Background mode must honor ``stream_protocol`` end-to-end.
 

src/frontend/src/stores/__tests__/flowStore.test.ts

@@ -50,15 +50,24 @@ jest.mock("../darkStore", () => ({
   },
 }));
 
-jest.mock("../flowsManagerStore", () => ({
-  __esModule: true,
-  default: {
-    getState: () => ({
-      setCurrentFlow: jest.fn(),
-      takeSnapshot: jest.fn(),
-    }),
-  },
-}));
+jest.mock("../flowsManagerStore", () => {
+  const state: { currentFlow: { id: string; name: string } | undefined } = {
+    currentFlow: undefined,
+  };
+  return {
+    __esModule: true,
+    default: {
+      getState: () => ({
+        ...state,
+        setCurrentFlow: jest.fn(),
+        takeSnapshot: jest.fn(),
+      }),
+      __setCurrentFlow: (flow: { id: string; name: string } | undefined) => {
+        state.currentFlow = flow;
+      },
+    },
+  };
+});
 
 jest.mock("../globalVariablesStore/globalVariables", () => ({
   useGlobalVariablesStore: {
@@ -90,6 +99,25 @@ jest.mock("@/utils/utils", () => ({
   brokenEdgeMessage: jest.fn(),
 }));
 
+// runFlowAGUI is exercised end-to-end in its own bridge tests; here we just
+// need a controllable replacement so the buildFlow integration can be tested
+// without touching the network.
+jest.mock("@/controllers/API/agui/run-flow-bridge", () => ({
+  runFlowAGUI: jest.fn(),
+}));
+
+// Keep reactflowUtils' real behaviour for the rest of the suite; only flip
+// validateNodes / validateEdge to no-ops so the buildFlow analytics tests can
+// reach the runFlowAGUI call with an empty (test-fixture) graph.
+jest.mock("../../utils/reactflowUtils", () => {
+  const actual = jest.requireActual("../../utils/reactflowUtils");
+  return {
+    ...actual,
+    validateNodes: jest.fn(() => []),
+    validateEdge: jest.fn(() => []),
+  };
+});
+
 // Note: Some utility modules may not exist in test environment
 // The store should handle missing utilities gracefully
 
@@ -1107,4 +1135,82 @@ describe("useFlowStore", () => {
       expect(latest["output_b"]).toEqual([mockLog2]);
     });
   });
+
+  describe("buildFlow analytics — trackFlowBuild integration", () => {
+    let mockedRunFlow: jest.Mock;
+    let trackFlowBuildMock: jest.Mock;
+
+    beforeAll(() => {
+      // Cast through unknown so the test does not depend on the bridge module's
+      // public type — we only care about controlling the side effect.
+      const bridge = jest.requireMock(
+        "@/controllers/API/agui/run-flow-bridge",
+      ) as { runFlowAGUI: jest.Mock };
+      mockedRunFlow = bridge.runFlowAGUI;
+
+      const analytics = jest.requireMock("@/customization/utils/analytics") as {
+        trackFlowBuild: jest.Mock;
+      };
+      trackFlowBuildMock = analytics.trackFlowBuild;
+
+      // Make `currentFlow` resolvable inside buildFlow.
+      const flowsManager = jest.requireMock("../flowsManagerStore") as {
+        default: {
+          __setCurrentFlow: (
+            flow: { id: string; name: string } | undefined,
+          ) => void;
+        };
+      };
+      flowsManager.default.__setCurrentFlow({
+        id: "flow-abc",
+        name: "Test Flow",
+      });
+    });
+
+    beforeEach(() => {
+      mockedRunFlow.mockReset();
+      trackFlowBuildMock.mockReset();
+      act(() => {
+        useFlowStore.setState({
+          nodes: [],
+          edges: [],
+          buildInfo: null,
+          flowBuildStatus: {},
+          isBuilding: false,
+          componentsToUpdate: [],
+        });
+      });
+    });
+
+    it("fires trackFlowBuild with isError=false after a successful run", async () => {
+      mockedRunFlow.mockImplementation(async () => {
+        // The bridge writes success into the store on `RUN_FINISHED`.
+        useFlowStore.setState({ buildInfo: { success: true } });
+      });
+
+      await useFlowStore.getState().buildFlow({});
+
+      expect(mockedRunFlow).toHaveBeenCalledTimes(1);
+      expect(trackFlowBuildMock).toHaveBeenCalledWith("Test Flow", false, {
+        flowId: "flow-abc",
+      });
+    });
+
+    it("fires trackFlowBuild with isError=true and the error list after a failure", async () => {
+      const errorList = ["boom"];
+      mockedRunFlow.mockImplementation(async () => {
+        useFlowStore.setState({
+          buildInfo: { success: false, error: errorList },
+        });
+      });
+
+      await useFlowStore.getState().buildFlow({});
+
+      expect(mockedRunFlow).toHaveBeenCalledTimes(1);
+      expect(trackFlowBuildMock).toHaveBeenCalledWith("Test Flow", true, {
+        flowId: "flow-abc",
+        error: errorList,
+      });
+    });
+  });
 });