From 4d7c3678a970f9f52ca065fbdd5aadf0ae08d58f Mon Sep 17 00:00:00 2001 From: mason5052 Date: Sun, 19 Apr 2026 14:27:06 -0400 Subject: [PATCH] feat: add flow-scoped file uploads --- backend/pkg/server/response/errors.go | 6 + backend/pkg/server/router.go | 11 + backend/pkg/server/services/flow_files.go | 333 ++++++++++++++++++ .../pkg/server/services/flow_files_test.go | 95 +++++ .../src/features/flows/files/flow-files.tsx | 289 +++++++++++++++ frontend/src/features/flows/flow-tabs.tsx | 9 + 6 files changed, 743 insertions(+) create mode 100644 backend/pkg/server/services/flow_files.go create mode 100644 backend/pkg/server/services/flow_files_test.go create mode 100644 frontend/src/features/flows/files/flow-files.tsx diff --git a/backend/pkg/server/response/errors.go b/backend/pkg/server/response/errors.go index b3161a1f..f294ddc3 100644 --- a/backend/pkg/server/response/errors.go +++ b/backend/pkg/server/response/errors.go @@ -109,6 +109,12 @@ var ErrFlowsInvalidRequest = NewHttpError(400, "Flows.InvalidRequest", "invalid var ErrFlowsNotFound = NewHttpError(404, "Flows.NotFound", "flow not found") var ErrFlowsInvalidData = NewHttpError(500, "Flows.InvalidData", "invalid flow data") +// flow files + +var ErrFlowFilesInvalidRequest = NewHttpError(400, "FlowFiles.InvalidRequest", "invalid flow file request data") +var ErrFlowFilesNotFound = NewHttpError(404, "FlowFiles.NotFound", "flow file not found") +var ErrFlowFilesInvalidData = NewHttpError(500, "FlowFiles.InvalidData", "invalid flow file data") + // tasks var ErrTasksInvalidRequest = NewHttpError(400, "Tasks.InvalidRequest", "invalid task request data") diff --git a/backend/pkg/server/router.go b/backend/pkg/server/router.go index 4e5404c3..45bb5a53 100644 --- a/backend/pkg/server/router.go +++ b/backend/pkg/server/router.go @@ -130,6 +130,7 @@ func NewRouter( roleService := services.NewRoleService(orm) providerService := services.NewProviderService(providers) flowService := services.NewFlowService(orm, providers, controller, subscriptions) + flowFileService := services.NewFlowFileService(orm, cfg.DataDir) taskService := services.NewTaskService(orm) subtaskService := services.NewSubtaskService(orm) containerService := services.NewContainerService(orm) @@ -223,6 +224,7 @@ func NewRouter( setProvidersGroup(privateGroup, providerService) setFlowsGroup(privateGroup, flowService) + setFlowFilesGroup(privateGroup, flowFileService) setTasksGroup(privateGroup, taskService) setSubtasksGroup(privateGroup, subtaskService) setContainersGroup(privateGroup, containerService) @@ -367,6 +369,15 @@ func setFlowsGroup(parent *gin.RouterGroup, svc *services.FlowService) { } } +func setFlowFilesGroup(parent *gin.RouterGroup, svc *services.FlowFileService) { + flowFilesGroup := parent.Group("/flows/:flowID/files") + { + flowFilesGroup.GET("/", svc.GetFlowFiles) + flowFilesGroup.POST("/", svc.UploadFlowFiles) + flowFilesGroup.GET("/:fileName", svc.DownloadFlowFile) + } +} + func setContainersGroup(parent *gin.RouterGroup, svc *services.ContainerService) { containersViewGroup := parent.Group("/containers") { diff --git a/backend/pkg/server/services/flow_files.go b/backend/pkg/server/services/flow_files.go new file mode 100644 index 00000000..b7461779 --- /dev/null +++ b/backend/pkg/server/services/flow_files.go @@ -0,0 +1,333 @@ +package services + +import ( + "errors" + "fmt" + "net/http" + "os" + "path" + "path/filepath" + "slices" + "sort" + "strconv" + "strings" + "time" + + "pentagi/pkg/server/logger" + "pentagi/pkg/server/models" + "pentagi/pkg/server/response" + + "github.com/gin-gonic/gin" + "github.com/jinzhu/gorm" +) + +const flowUploadsDirName = "uploads" + +type flowFile struct { + Name string `json:"name"` + Path string `json:"path"` + Size int64 `json:"size"` + ModifiedAt time.Time `json:"modifiedAt"` +} + +type flowFiles struct { + Files []flowFile `json:"files"` + Total uint64 `json:"total"` +} + +type FlowFileService struct { + dataDir string + db *gorm.DB +} + +func NewFlowFileService(db *gorm.DB, dataDir string) *FlowFileService { + return &FlowFileService{ + dataDir: dataDir, + db: db, + } +} + +func (s *FlowFileService) GetFlowFiles(c *gin.Context) { + flowID, err := parseFlowIDParam(c) + if err != nil { + logger.FromContext(c).WithError(err).Error("error parsing flow id") + response.Error(c, response.ErrFlowFilesInvalidRequest, err) + return + } + + if _, err := s.getFlow(c, flowID, false); err != nil { + s.handleFlowLookupError(c, flowID, err) + return + } + + files, err := s.listFlowFiles(flowID) + if err != nil { + logger.FromContext(c).WithError(err).WithField("flow_id", flowID).Error("error listing flow files") + response.Error(c, response.ErrInternal, err) + return + } + + response.Success(c, http.StatusOK, flowFiles{ + Files: files, + Total: uint64(len(files)), + }) +} + +func (s *FlowFileService) UploadFlowFiles(c *gin.Context) { + flowID, err := parseFlowIDParam(c) + if err != nil { + logger.FromContext(c).WithError(err).Error("error parsing flow id") + response.Error(c, response.ErrFlowFilesInvalidRequest, err) + return + } + + if _, err := s.getFlow(c, flowID, true); err != nil { + s.handleFlowLookupError(c, flowID, err) + return + } + + multipartForm, err := c.MultipartForm() + if err != nil { + logger.FromContext(c).WithError(err).WithField("flow_id", flowID).Error("error reading multipart form") + response.Error(c, response.ErrFlowFilesInvalidRequest, err) + return + } + + fileHeaders := multipartForm.File["files"] + if len(fileHeaders) == 0 { + fileHeader, formErr := c.FormFile("file") + if formErr == nil && fileHeader != nil { + fileHeaders = append(fileHeaders, fileHeader) + } + } + if len(fileHeaders) == 0 { + err = errors.New("at least one uploaded file is required") + logger.FromContext(c).WithError(err).WithField("flow_id", flowID).Error("missing uploaded files") + response.Error(c, response.ErrFlowFilesInvalidRequest, err) + return + } + + uploadDir := s.flowUploadsDir(flowID) + if err := os.MkdirAll(uploadDir, 0755); err != nil { + logger.FromContext(c).WithError(err).WithField("flow_id", flowID).Error("error creating upload directory") + response.Error(c, response.ErrInternal, err) + return + } + + savedFiles := make([]flowFile, 0, len(fileHeaders)) + for _, fileHeader := range fileHeaders { + fileName, err := sanitizeFlowFileName(fileHeader.Filename) + if err != nil { + logger.FromContext(c).WithError(err).WithField("flow_id", flowID).Error("invalid uploaded file name") + response.Error(c, response.ErrFlowFilesInvalidData, err) + return + } + + dstPath := filepath.Join(uploadDir, fileName) + if err := c.SaveUploadedFile(fileHeader, dstPath); err != nil { + logger.FromContext(c).WithError(err).WithFields(map[string]any{ + "flow_id": flowID, + "file_name": fileName, + }).Error("error saving uploaded file") + response.Error(c, response.ErrInternal, err) + return + } + + info, err := os.Stat(dstPath) + if err != nil { + logger.FromContext(c).WithError(err).WithFields(map[string]any{ + "flow_id": flowID, + "file_name": fileName, + }).Error("error stating uploaded file") + response.Error(c, response.ErrInternal, err) + return + } + + savedFiles = append(savedFiles, newFlowFile(info)) + } + + sortFlowFiles(savedFiles) + response.Success(c, http.StatusOK, flowFiles{ + Files: savedFiles, + Total: uint64(len(savedFiles)), + }) +} + +func (s *FlowFileService) DownloadFlowFile(c *gin.Context) { + flowID, err := parseFlowIDParam(c) + if err != nil { + logger.FromContext(c).WithError(err).Error("error parsing flow id") + response.Error(c, response.ErrFlowFilesInvalidRequest, err) + return + } + + if _, err := s.getFlow(c, flowID, false); err != nil { + s.handleFlowLookupError(c, flowID, err) + return + } + + fileName, err := sanitizeFlowFileName(c.Param("fileName")) + if err != nil { + logger.FromContext(c).WithError(err).WithField("flow_id", flowID).Error("invalid download file name") + response.Error(c, response.ErrFlowFilesInvalidRequest, err) + return + } + + filePath := filepath.Join(s.flowUploadsDir(flowID), fileName) + info, err := os.Stat(filePath) + if err != nil { + logger.FromContext(c).WithError(err).WithFields(map[string]any{ + "flow_id": flowID, + "file_name": fileName, + }).Error("error reading flow file") + if errors.Is(err, os.ErrNotExist) { + response.Error(c, response.ErrFlowFilesNotFound, err) + } else { + response.Error(c, response.ErrInternal, err) + } + return + } + if info.IsDir() { + err = fmt.Errorf("file '%s' is a directory", fileName) + logger.FromContext(c).WithError(err).WithFields(map[string]any{ + "flow_id": flowID, + "file_name": fileName, + }).Error("invalid flow file type") + response.Error(c, response.ErrFlowFilesNotFound, err) + return + } + + c.FileAttachment(filePath, fileName) +} + +func (s *FlowFileService) getFlow(c *gin.Context, flowID uint64, writeAccess bool) (models.Flow, error) { + var flow models.Flow + + uid := c.GetUint64("uid") + privs := c.GetStringSlice("prm") + scope := flowScopeForFiles(privs, uid, flowID, writeAccess) + if scope == nil { + return flow, response.ErrNotPermitted + } + + if err := s.db.Model(&flow).Scopes(scope).Take(&flow).Error; err != nil { + if gorm.IsRecordNotFoundError(err) { + return flow, response.ErrFlowsNotFound + } + return flow, err + } + + return flow, nil +} + +func (s *FlowFileService) listFlowFiles(flowID uint64) ([]flowFile, error) { + entries, err := os.ReadDir(s.flowUploadsDir(flowID)) + if err != nil { + if errors.Is(err, os.ErrNotExist) { + return []flowFile{}, nil + } + return nil, err + } + + files := make([]flowFile, 0, len(entries)) + for _, entry := range entries { + if entry.IsDir() { + continue + } + + info, err := entry.Info() + if err != nil { + return nil, err + } + + files = append(files, newFlowFile(info)) + } + + sortFlowFiles(files) + return files, nil +} + +func (s *FlowFileService) handleFlowLookupError(c *gin.Context, flowID uint64, err error) { + fields := map[string]any{"flow_id": flowID} + + switch err { + case response.ErrNotPermitted: + logger.FromContext(c).WithFields(fields).Error("error filtering user role permissions: permission not found") + response.Error(c, response.ErrNotPermitted, nil) + case response.ErrFlowsNotFound: + logger.FromContext(c).WithFields(fields).Error("error finding flow for flow files") + response.Error(c, response.ErrFlowsNotFound, err) + default: + logger.FromContext(c).WithError(err).WithFields(fields).Error("error loading flow for flow files") + response.Error(c, response.ErrInternal, err) + } +} + +func (s *FlowFileService) flowUploadsDir(flowID uint64) string { + return filepath.Join(s.dataDir, fmt.Sprintf("flow-%d", flowID), flowUploadsDirName) +} + +func flowScopeForFiles( + privs []string, + uid uint64, + flowID uint64, + writeAccess bool, +) func(db *gorm.DB) *gorm.DB { + if slices.Contains(privs, "flows.admin") { + return func(db *gorm.DB) *gorm.DB { + return db.Where("id = ?", flowID) + } + } + + if writeAccess && slices.Contains(privs, "flows.edit") { + return func(db *gorm.DB) *gorm.DB { + return db.Where("id = ? AND user_id = ?", flowID, uid) + } + } + + if !writeAccess && slices.Contains(privs, "flows.view") { + return func(db *gorm.DB) *gorm.DB { + return db.Where("id = ? AND user_id = ?", flowID, uid) + } + } + + return nil +} + +func parseFlowIDParam(c *gin.Context) (uint64, error) { + return strconv.ParseUint(c.Param("flowID"), 10, 64) +} + +func sanitizeFlowFileName(fileName string) (string, error) { + trimmedName := strings.TrimSpace(fileName) + if trimmedName == "" { + return "", fmt.Errorf("file name is required") + } + + normalizedName := strings.ReplaceAll(trimmedName, "\\", "/") + cleanName := path.Base(path.Clean("/" + normalizedName)) + if cleanName == "." || cleanName == "/" || cleanName == "" { + return "", fmt.Errorf("invalid file name") + } + + return cleanName, nil +} + +func newFlowFile(info os.FileInfo) flowFile { + return flowFile{ + Name: info.Name(), + Path: path.Join("/work", flowUploadsDirName, info.Name()), + Size: info.Size(), + ModifiedAt: info.ModTime(), + } +} + +func sortFlowFiles(files []flowFile) { + sort.Slice(files, func(i, j int) bool { + if files[i].ModifiedAt.Equal(files[j].ModifiedAt) { + return files[i].Name < files[j].Name + } + + return files[i].ModifiedAt.After(files[j].ModifiedAt) + }) +} diff --git a/backend/pkg/server/services/flow_files_test.go b/backend/pkg/server/services/flow_files_test.go new file mode 100644 index 00000000..b40cb440 --- /dev/null +++ b/backend/pkg/server/services/flow_files_test.go @@ -0,0 +1,95 @@ +package services + +import ( + "os" + "path/filepath" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestSanitizeFlowFileName(t *testing.T) { + testCases := []struct { + expected string + fileName string + name string + wantErr bool + }{ + { + name: "keeps plain file name", + fileName: "report.txt", + expected: "report.txt", + }, + { + name: "collapses parent traversal", + fileName: "../report.txt", + expected: "report.txt", + }, + { + name: "normalizes windows separators", + fileName: `nested\brief.md`, + expected: "brief.md", + }, + { + name: "rejects empty names", + fileName: " ", + wantErr: true, + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + actual, err := sanitizeFlowFileName(tc.fileName) + + if tc.wantErr { + require.Error(t, err) + return + } + + require.NoError(t, err) + assert.Equal(t, tc.expected, actual) + }) + } +} + +func TestFlowFileService_ListFlowFiles(t *testing.T) { + dataDir := t.TempDir() + service := NewFlowFileService(nil, dataDir) + uploadDir := filepath.Join(dataDir, "flow-7", "uploads") + + require.NoError(t, os.MkdirAll(uploadDir, 0755)) + require.NoError(t, os.Mkdir(filepath.Join(uploadDir, "nested"), 0755)) + + oldPath := filepath.Join(uploadDir, "old.txt") + newPath := filepath.Join(uploadDir, "new.txt") + + require.NoError(t, os.WriteFile(oldPath, []byte("old"), 0644)) + require.NoError(t, os.WriteFile(newPath, []byte("newer content"), 0644)) + + oldTime := time.Now().Add(-2 * time.Hour) + newTime := time.Now().Add(-1 * time.Hour) + require.NoError(t, os.Chtimes(oldPath, oldTime, oldTime)) + require.NoError(t, os.Chtimes(newPath, newTime, newTime)) + + files, err := service.listFlowFiles(7) + require.NoError(t, err) + require.Len(t, files, 2) + + assert.Equal(t, "new.txt", files[0].Name) + assert.Equal(t, "/work/uploads/new.txt", files[0].Path) + assert.Equal(t, int64(len("newer content")), files[0].Size) + + assert.Equal(t, "old.txt", files[1].Name) + assert.Equal(t, "/work/uploads/old.txt", files[1].Path) + assert.Equal(t, int64(len("old")), files[1].Size) +} + +func TestFlowFileService_ListFlowFiles_MissingDirectory(t *testing.T) { + service := NewFlowFileService(nil, t.TempDir()) + + files, err := service.listFlowFiles(999) + require.NoError(t, err) + assert.Empty(t, files) +} diff --git a/frontend/src/features/flows/files/flow-files.tsx b/frontend/src/features/flows/files/flow-files.tsx new file mode 100644 index 00000000..9c2b05d3 --- /dev/null +++ b/frontend/src/features/flows/files/flow-files.tsx @@ -0,0 +1,289 @@ +import { AlertCircle, Copy, Download, FileUp, FolderUp, Loader2, RefreshCw } from 'lucide-react'; +import { useCallback, useEffect, useMemo, useRef, useState } from 'react'; +import { toast } from 'sonner'; + +import { Button, buttonVariants } from '@/components/ui/button'; +import { Empty, EmptyDescription, EmptyHeader, EmptyMedia, EmptyTitle } from '@/components/ui/empty'; +import { InputGroup, InputGroupAddon, InputGroupButton, InputGroupInput } from '@/components/ui/input-group'; +import { Tooltip, TooltipContent, TooltipTrigger } from '@/components/ui/tooltip'; +import { axios } from '@/lib/axios'; +import { copyToClipboard } from '@/lib/report'; +import { cn } from '@/lib/utils'; +import { formatDate } from '@/lib/utils/format'; +import { baseUrl } from '@/models/api'; +import { useFlow } from '@/providers/flow-provider'; + +interface FlowFile { + modifiedAt: string; + name: string; + path: string; + size: number; +} + +interface FlowFilesResponse { + files: Array; + total: number; +} + +const formatFileSize = (size: number) => { + if (size < 1024) { + return `${size} B`; + } + + const units = ['KB', 'MB', 'GB', 'TB']; + let unitIndex = -1; + let value = size; + + while (value >= 1024 && unitIndex < units.length - 1) { + value /= 1024; + unitIndex += 1; + } + + return `${value.toFixed(value >= 10 ? 0 : 1)} ${units[unitIndex]}`; +}; + +const FlowFiles = () => { + const { flowId } = useFlow(); + const inputRef = useRef(null); + const [files, setFiles] = useState>([]); + const [isLoading, setIsLoading] = useState(false); + const [isUploading, setIsUploading] = useState(false); + const [searchValue, setSearchValue] = useState(''); + + const loadFiles = useCallback(async () => { + if (!flowId) { + setFiles([]); + + return; + } + + setIsLoading(true); + + try { + const data = await axios.get(`/flows/${flowId}/files`); + setFiles(data.files ?? []); + } catch (error) { + const description = error instanceof Error ? error.message : 'An error occurred while loading flow files'; + toast.error('Failed to load files', { + description, + }); + } finally { + setIsLoading(false); + } + }, [flowId]); + + useEffect(() => { + void loadFiles(); + }, [loadFiles]); + + const handleUploadButtonClick = useCallback(() => { + inputRef.current?.click(); + }, []); + + const handleCopyPath = useCallback(async (filePath: string) => { + const success = await copyToClipboard(filePath); + + if (success) { + toast.success('Path copied to clipboard'); + + return; + } + + toast.error('Failed to copy path'); + }, []); + + const handleFileSelection = useCallback( + async (event: React.ChangeEvent) => { + if (!flowId) { + return; + } + + const selectedFiles = Array.from(event.target.files ?? []); + + if (selectedFiles.length === 0) { + return; + } + + const formData = new FormData(); + + for (const file of selectedFiles) { + formData.append('files', file); + } + + setIsUploading(true); + + try { + const data = await axios.post(`/flows/${flowId}/files`, formData); + const uploadedCount = data.files?.length ?? selectedFiles.length; + + toast.success(uploadedCount === 1 ? 'File uploaded successfully' : 'Files uploaded successfully', { + description: + uploadedCount === 1 + ? `${data.files?.[0]?.path ?? '/work/uploads'}` + : `${uploadedCount} files are now available under /work/uploads`, + }); + + await loadFiles(); + } catch (error) { + const description = error instanceof Error ? error.message : 'An error occurred while uploading files'; + toast.error('Failed to upload files', { + description, + }); + } finally { + setIsUploading(false); + event.target.value = ''; + } + }, + [flowId, loadFiles], + ); + + const filteredFiles = useMemo(() => { + const normalizedSearch = searchValue.toLowerCase().trim(); + + if (!normalizedSearch) { + return files; + } + + return files.filter((file) => { + return ( + file.name.toLowerCase().includes(normalizedSearch) || file.path.toLowerCase().includes(normalizedSearch) + ); + }); + }, [files, searchValue]); + + return ( +
+ + +
+
+
+ +
+

Uploaded files are shared with the whole flow.

+

+ PentAGI stores user uploads under /work/uploads, so automation, assistants, + and container commands can access the same files immediately. +

+
+
+
+ +
+ + + + + setSearchValue(event.target.value)} + placeholder="Filter uploaded files..." + type="text" + value={searchValue} + /> + {searchValue && ( + + setSearchValue('')} + type="button" + > + Clear + + + )} + + +
+ + +
+
+
+ + {filteredFiles.length > 0 ? ( +
+ {filteredFiles.map((file) => ( +
+
+
+
+ +

{file.name}

+
+
+

{formatFileSize(file.size)}

+
+ {file.path} + + + + + Copy container path + +
+

{formatDate(new Date(file.modifiedAt))}

+
+
+ + + + Download + +
+
+ ))} +
+ ) : ( + + + + + + No uploaded files + + Upload files into /work/uploads when you want this flow, its assistants, or + container commands to use them. + + + + )} +
+ ); +}; + +export default FlowFiles; diff --git a/frontend/src/features/flows/flow-tabs.tsx b/frontend/src/features/flows/flow-tabs.tsx index a86b74ba..6d911b0a 100644 --- a/frontend/src/features/flows/flow-tabs.tsx +++ b/frontend/src/features/flows/flow-tabs.tsx @@ -4,6 +4,7 @@ import { ScrollArea, ScrollBar } from '@/components/ui/scroll-area'; import { Tabs, TabsContent, TabsList, TabsTrigger } from '@/components/ui/tabs'; import FlowAgents from '@/features/flows/agents/flow-agents'; import FlowDashboard from '@/features/flows/dashboard/flow-dashboard'; +import FlowFiles from '@/features/flows/files/flow-files'; import FlowAssistantMessages from '@/features/flows/messages/flow-assistant-messages'; import FlowAutomationMessages from '@/features/flows/messages/flow-automation-messages'; import FlowScreenshots from '@/features/flows/screenshots/flow-screenshots'; @@ -49,6 +50,7 @@ const FlowTabs = ({ activeTab, onTabChange }: FlowTabsProps) => { Agents Searches Vector Store + Files Screenshots @@ -117,6 +119,13 @@ const FlowTabs = ({ activeTab, onTabChange }: FlowTabsProps) => { + + + +