refactor(dify-agent): download Binding files through Agent Stub (#40143)

Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>

dify-agent-runtime/cmd/dify-agent-cli/main.go

@@ -103,14 +103,21 @@ func newFileCommand() *cobra.Command {
 		Short: "Upload or download workflow files through the Agent Stub.",
 	}
 
+	var noDownloadLink bool
 	upload := &cobra.Command{
 		Use:   "upload PATH",
 		Short: "Upload one sandbox-local file as a ToolFile output reference.",
 		Args:  cobra.ExactArgs(1),
 		RunE: withEnv(func(env *agentcli.Environment, args []string, _ *cobra.Command) error {
-			return agentcli.RunFileUpload(env, args[0])
+			return agentcli.RunFileUpload(env, args[0], noDownloadLink)
 		}),
 	}
+	upload.Flags().BoolVar(
+		&noDownloadLink,
+		"no-download-link",
+		false,
+		"Skip creating a public download link after upload.",
+	)
 
 	var downloadTo string
 	download := &cobra.Command{

dify-agent-runtime/cmd/dify-agent-cli/main_test.go

@@ -53,7 +53,12 @@ func TestCommandHelp(t *testing.T) {
 		{
 			name: "file upload",
 			args: []string{"file", "upload", "--help"},
-			want: []string{"dify-agent file upload", "Upload one sandbox-local file"},
+			want: []string{
+				"dify-agent file upload",
+				"Upload one sandbox-local file",
+				"--no-download-link",
+				"Skip creating a public download link after upload.",
+			},
 		},
 		{
 			name: "file download",

dify-agent-runtime/internal/agentcli/file.go

@@ -2,6 +2,7 @@ package agentcli
 
 import (
 	"context"
+	"encoding/base64"
 	"encoding/json"
 	"fmt"
 	"io"
@@ -15,7 +16,7 @@ import (
 type FileUploadResponse struct {
 	TransferMethod    string `json:"transfer_method"`
 	Reference         string `json:"reference"`
-	PublicDownloadURL string `json:"public_download_url"`
+	PublicDownloadURL string `json:"public_download_url,omitempty"`
 }
 
 // FileDownloadResponse is the response from a file download request.
@@ -27,14 +28,14 @@ type FileDownloadResponse struct {
 }
 
 // RunFileUpload executes the `file upload` command.
-func RunFileUpload(env *Environment, path string) error {
+func RunFileUpload(env *Environment, path string, noDownloadLink bool) error {
 	client, err := NewStubClient(env)
 	if err != nil {
 		return err
 	}
 	defer func() { _ = client.Close() }()
 
-	return runFileUpload(client, path, os.Stdout)
+	return runFileUpload(client, path, noDownloadLink, os.Stdout)
 }
 
 type fileUploadClient interface {
@@ -48,7 +49,7 @@ type fileUploadClient interface {
 	) (*FileDownloadResponse, error)
 }
 
-func runFileUpload(client fileUploadClient, path string, output io.Writer) error {
+func runFileUpload(client fileUploadClient, path string, noDownloadLink bool, output io.Writer) error {
 	absPath, err := filepath.Abs(path)
 	if err != nil {
 		return fmt.Errorf("resolve path: %w", err)
@@ -83,24 +84,47 @@ func runFileUpload(client fileUploadClient, path string, output io.Writer) error
 	if reference == "" {
 		return fmt.Errorf("signed file upload response is missing reference")
 	}
-
-	// Step 3: Request download URL for the uploaded file
-	ref := reference
-	dlResp, err := client.CreateFileDownloadURL(ctx, "tool_file", &ref, nil, true)
-	if err != nil {
-		return err
+	if noDownloadLink && !isCanonicalDifyFileReference(reference) {
+		return fmt.Errorf("signed file upload response has invalid reference")
 	}
 
 	result := FileUploadResponse{
-		TransferMethod:    "tool_file",
-		Reference:         reference,
-		PublicDownloadURL: dlResp.DownloadURL,
+		TransferMethod: "tool_file",
+		Reference:      reference,
+	}
+	if !noDownloadLink {
+		// Step 3: Request a browser-visible URL unless the caller only needs the
+		// canonical ToolFile reference.
+		ref := reference
+		dlResp, err := client.CreateFileDownloadURL(ctx, "tool_file", &ref, nil, true)
+		if err != nil {
+			return err
+		}
+		result.PublicDownloadURL = dlResp.DownloadURL
 	}
 	out, _ := json.Marshal(result)
 	_, _ = fmt.Fprintln(output, string(out))
 	return nil
 }
 
+func isCanonicalDifyFileReference(reference string) bool {
+	encodedPayload, found := strings.CutPrefix(reference, "dify-file-ref:")
+	if !found || encodedPayload == "" {
+		return false
+	}
+	payloadJSON, err := base64.URLEncoding.DecodeString(encodedPayload)
+	if err != nil {
+		return false
+	}
+	var payload struct {
+		RecordID string `json:"record_id"`
+	}
+	if err := json.Unmarshal(payloadJSON, &payload); err != nil {
+		return false
+	}
+	return payload.RecordID != ""
+}
+
 // RunFileDownload executes the `file download` command.
 func RunFileDownload(env *Environment, transferMethod string, referenceOrURL string, localDir string) error {
 	var reference *string

dify-agent-runtime/internal/agentcli/file_test.go

@@ -13,24 +13,48 @@ import (
 )
 
 type fakeFileUploadClient struct {
-	forFrontend bool
+	forFrontend         bool
+	downloadRequestCall int
+	uploadResponse      []byte
+	calls               []string
+	filename            string
+	mimetype            string
+	uploadURL           string
+	uploadedBytes       []byte
+	downloadReference   string
 }
 
 func (f *fakeFileUploadClient) CreateFileUploadURL(_ context.Context, filename, mimetype string) (string, error) {
+	f.calls = append(f.calls, "upload-request")
+	f.filename = filename
+	f.mimetype = mimetype
 	return "https://sandbox-files.example.com/files/upload/for-plugin?sign=1", nil
 }
 
 func (f *fakeFileUploadClient) UploadFileToURL(uploadURL, filePath, filename, mimetype string) ([]byte, error) {
-	return []byte(`{"reference":"dify-file-ref:canonical"}`), nil
+	f.calls = append(f.calls, "multipart-upload")
+	f.uploadURL = uploadURL
+	f.filename = filename
+	f.mimetype = mimetype
+	f.uploadedBytes, _ = os.ReadFile(filePath)
+	if f.uploadResponse != nil {
+		return f.uploadResponse, nil
+	}
+	return []byte(`{"reference":"dify-file-ref:eyJyZWNvcmRfaWQiOiJ0b29sLTEifQ=="}`), nil
 }
 
 func (f *fakeFileUploadClient) CreateFileDownloadURL(
 	_ context.Context,
 	_ string,
-	_, _ *string,
+	reference, _ *string,
 	forFrontend bool,
 ) (*FileDownloadResponse, error) {
+	f.calls = append(f.calls, "download-request")
 	f.forFrontend = forFrontend
+	f.downloadRequestCall++
+	if reference != nil {
+		f.downloadReference = *reference
+	}
 	return &FileDownloadResponse{
 		Filename:    "report.pdf",
 		MimeType:    "application/pdf",
@@ -47,20 +71,136 @@ func TestRunFileUploadReturnsFrontendDisplayURL(t *testing.T) {
 
 	client := &fakeFileUploadClient{}
 	var output bytes.Buffer
-	if err := runFileUpload(client, filePath, &output); err != nil {
+	if err := runFileUpload(client, filePath, false, &output); err != nil {
 		t.Fatalf("run file upload: %v", err)
 	}
 
 	if !client.forFrontend {
 		t.Fatal("download request did not select frontend display URL")
 	}
+	if got, want := strings.Join(client.calls, ","), "upload-request,multipart-upload,download-request"; got != want {
+		t.Fatalf("call order = %s, want %s", got, want)
+	}
+	if client.filename != "report.pdf" || client.mimetype != "application/pdf" {
+		t.Fatalf("upload metadata = (%q, %q), want report.pdf/application/pdf", client.filename, client.mimetype)
+	}
+	if client.uploadURL != "https://sandbox-files.example.com/files/upload/for-plugin?sign=1" {
+		t.Fatalf("upload URL = %q", client.uploadURL)
+	}
+	if string(client.uploadedBytes) != "report" {
+		t.Fatalf("uploaded bytes = %q, want report", client.uploadedBytes)
+	}
+	if client.downloadReference != "dify-file-ref:eyJyZWNvcmRfaWQiOiJ0b29sLTEifQ==" {
+		t.Fatalf("download reference = %q", client.downloadReference)
+	}
 	got := strings.TrimSpace(output.String())
-	want := `{"transfer_method":"tool_file","reference":"dify-file-ref:canonical","public_download_url":"/files/tools/report.pdf?sign=2"}`
+	want := `{"transfer_method":"tool_file","reference":"dify-file-ref:eyJyZWNvcmRfaWQiOiJ0b29sLTEifQ==","public_download_url":"/files/tools/report.pdf?sign=2"}`
 	if got != want {
 		t.Fatalf("output = %s, want %s", got, want)
 	}
 }
 
+func TestRunFileUploadWithoutDownloadLinkReturnsOnlyCanonicalMapping(t *testing.T) {
+	filePath := filepath.Join(t.TempDir(), "report.pdf")
+	if err := os.WriteFile(filePath, []byte("report"), 0o600); err != nil {
+		t.Fatalf("write fixture: %v", err)
+	}
+
+	client := &fakeFileUploadClient{}
+	var output bytes.Buffer
+	if err := runFileUpload(client, filePath, true, &output); err != nil {
+		t.Fatalf("run file upload: %v", err)
+	}
+
+	if client.downloadRequestCall != 0 {
+		t.Fatalf("download request calls = %d, want 0", client.downloadRequestCall)
+	}
+	if got, want := strings.Join(client.calls, ","), "upload-request,multipart-upload"; got != want {
+		t.Fatalf("call order = %s, want %s", got, want)
+	}
+	got := strings.TrimSpace(output.String())
+	want := `{"transfer_method":"tool_file","reference":"dify-file-ref:eyJyZWNvcmRfaWQiOiJ0b29sLTEifQ=="}`
+	if got != want {
+		t.Fatalf("output = %s, want %s", got, want)
+	}
+}
+
+func TestRunFileUploadDefaultAcceptsLegacyNonemptyReference(t *testing.T) {
+	filePath := filepath.Join(t.TempDir(), "report.pdf")
+	if err := os.WriteFile(filePath, []byte("report"), 0o600); err != nil {
+		t.Fatalf("write fixture: %v", err)
+	}
+
+	client := &fakeFileUploadClient{uploadResponse: []byte(`{"reference":"raw-id"}`)}
+	var output bytes.Buffer
+	err := runFileUpload(client, filePath, false, &output)
+	if err != nil {
+		t.Fatalf("run file upload: %v", err)
+	}
+	if client.downloadRequestCall != 1 || client.downloadReference != "raw-id" {
+		t.Fatalf("download request = (%d, %q), want legacy reference", client.downloadRequestCall, client.downloadReference)
+	}
+	if !strings.Contains(output.String(), `"reference":"raw-id"`) {
+		t.Fatalf("output = %q, want legacy reference", output.String())
+	}
+}
+
+func TestRunFileUploadWithoutDownloadLinkRejectsNonCanonicalReference(t *testing.T) {
+	filePath := filepath.Join(t.TempDir(), "report.pdf")
+	if err := os.WriteFile(filePath, []byte("report"), 0o600); err != nil {
+		t.Fatalf("write fixture: %v", err)
+	}
+
+	client := &fakeFileUploadClient{uploadResponse: []byte(`{"reference":"raw-id"}`)}
+	var output bytes.Buffer
+	err := runFileUpload(client, filePath, true, &output)
+	if err == nil || !strings.Contains(err.Error(), "invalid reference") {
+		t.Fatalf("error = %v, want invalid reference", err)
+	}
+	if client.downloadRequestCall != 0 {
+		t.Fatalf("download request calls = %d, want 0", client.downloadRequestCall)
+	}
+	if output.Len() != 0 {
+		t.Fatalf("output = %q, want empty", output.String())
+	}
+}
+
+func TestRunFileUploadRejectsMissingReferenceInBothModes(t *testing.T) {
+	filePath := filepath.Join(t.TempDir(), "report.pdf")
+	if err := os.WriteFile(filePath, []byte("report"), 0o600); err != nil {
+		t.Fatalf("write fixture: %v", err)
+	}
+
+	for _, noDownloadLink := range []bool{false, true} {
+		client := &fakeFileUploadClient{uploadResponse: []byte(`{"reference":""}`)}
+		var output bytes.Buffer
+		err := runFileUpload(client, filePath, noDownloadLink, &output)
+		if err == nil || !strings.Contains(err.Error(), "missing reference") {
+			t.Fatalf("noDownloadLink=%t error = %v, want missing reference", noDownloadLink, err)
+		}
+		if client.downloadRequestCall != 0 {
+			t.Fatalf("noDownloadLink=%t download request calls = %d, want 0", noDownloadLink, client.downloadRequestCall)
+		}
+	}
+}
+
+func TestRunFileUploadRejectsInvalidUploadResponseBeforeDownloadRequest(t *testing.T) {
+	filePath := filepath.Join(t.TempDir(), "report.pdf")
+	if err := os.WriteFile(filePath, []byte("report"), 0o600); err != nil {
+		t.Fatalf("write fixture: %v", err)
+	}
+
+	client := &fakeFileUploadClient{uploadResponse: []byte("not-json")}
+	var output bytes.Buffer
+	err := runFileUpload(client, filePath, true, &output)
+	if err == nil || !strings.Contains(err.Error(), "parse upload result") {
+		t.Fatalf("error = %v, want parse upload result failure", err)
+	}
+	if client.downloadRequestCall != 0 {
+		t.Fatalf("download request calls = %d, want 0", client.downloadRequestCall)
+	}
+}
+
 func TestRunFileDownloadRequestsSandboxURLAndWritesFile(t *testing.T) {
 	var requestPayload map[string]json.RawMessage
 	var server *httptest.Server

dify-agent-runtime/internal/agentcli/httpclient.go

@@ -2,7 +2,9 @@ package agentcli
 
 import (
 	"bytes"
+	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"mime/multipart"
@@ -14,29 +16,45 @@ import (
 
 // HTTPClient wraps HTTP interactions with the Agent Stub server.
 type HTTPClient struct {
-	baseURL string
-	authJWE string
-	client  *http.Client
+	baseURL         string
+	authJWE         string
+	client          *http.Client
+	openUploadFile  func(string) (io.ReadCloser, error)
+	doUploadRequest func(*http.Request) (*http.Response, error)
 }
 
+var errUploadRequestAborted = errors.New("upload request aborted")
+
 // NewHTTPClient creates a new HTTP client for the Agent Stub API.
 func NewHTTPClient(env *Environment) *HTTPClient {
 	return &HTTPClient{
-		baseURL: env.URL,
-		authJWE: env.AuthJWE,
-		client:  &http.Client{Timeout: 30 * time.Second},
+		baseURL:         env.URL,
+		authJWE:         env.AuthJWE,
+		client:          &http.Client{Timeout: 30 * time.Second},
+		openUploadFile:  openUploadSource,
+		doUploadRequest: doUploadRequest,
 	}
 }
 
 // NewHTTPClientWithTimeout creates a client with a custom timeout.
 func NewHTTPClientWithTimeout(env *Environment, timeout time.Duration) *HTTPClient {
 	return &HTTPClient{
-		baseURL: env.URL,
-		authJWE: env.AuthJWE,
-		client:  &http.Client{Timeout: timeout},
+		baseURL:         env.URL,
+		authJWE:         env.AuthJWE,
+		client:          &http.Client{Timeout: timeout},
+		openUploadFile:  openUploadSource,
+		doUploadRequest: doUploadRequest,
 	}
 }
 
+func openUploadSource(path string) (io.ReadCloser, error) {
+	return os.Open(path)
+}
+
+func doUploadRequest(req *http.Request) (*http.Response, error) {
+	return (&http.Client{Timeout: 120 * time.Second}).Do(req)
+}
+
 // postJSON sends a POST request with JSON body and returns the response body.
 func (c *HTTPClient) postJSON(path string, payload any) ([]byte, int, error) {
 	body, err := json.Marshal(payload)
@@ -149,49 +167,105 @@ func (c *HTTPClient) putJSON(path string, payload any) ([]byte, int, error) {
 
 // uploadFile uploads a file to a signed URL using multipart form.
 func (c *HTTPClient) uploadFile(uploadURL string, filePath string, filename string, mimetype string) ([]byte, error) {
-	file, err := os.Open(filePath)
+	file, err := c.openUploadFile(filePath)
 	if err != nil {
 		return nil, fmt.Errorf("open file: %w", err)
 	}
-	defer func() { _ = file.Close() }()
+	pipeReader, pipeWriter := io.Pipe()
+
+	req, err := http.NewRequest("POST", uploadURL, pipeReader)
+	if err != nil {
+		_ = pipeReader.Close()
+		_ = pipeWriter.Close()
+		_ = file.Close()
+		return nil, errors.New("create upload request: invalid signed upload URL")
+	}
+	multipartWriter := multipart.NewWriter(pipeWriter)
+	req.Header.Set("Content-Type", multipartWriter.FormDataContentType())
+
+	writerDone := make(chan error, 1)
+	go func() {
+		writeErr := writeMultipartFile(multipartWriter, file, filename, mimetype)
+		if closeErr := file.Close(); writeErr == nil && closeErr != nil {
+			writeErr = fmt.Errorf("close file: %w", closeErr)
+		}
+		if writeErr != nil {
+			_ = pipeWriter.CloseWithError(writeErr)
+		} else {
+			writeErr = pipeWriter.Close()
+		}
+		writerDone <- writeErr
+	}()
+
+	resp, requestErr := c.doUploadRequest(req)
+	_ = pipeReader.CloseWithError(errUploadRequestAborted)
+
+	const maxUploadResponseBytes = 1024 * 1024
+	var respBody []byte
+	var responseErr error
+	if resp != nil {
+		defer func() { _ = resp.Body.Close() }()
+		respBody, responseErr = io.ReadAll(io.LimitReader(resp.Body, maxUploadResponseBytes+1))
+	}
+
+	writerErr := <-writerDone
+	if resp != nil && (resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices) {
+		if len(respBody) > maxUploadResponseBytes {
+			respBody = respBody[:maxUploadResponseBytes]
+		}
+		return nil, fmt.Errorf("upload failed with status %d: %s", resp.StatusCode, string(respBody))
+	}
+	if requestErr != nil {
+		if writerErr != nil && !isUploadWriterAbort(writerErr) {
+			return nil, writerErr
+		}
+		if errors.Is(requestErr, context.DeadlineExceeded) || os.IsTimeout(requestErr) {
+			return nil, errors.New("upload request timed out")
+		}
+		return nil, errors.New("upload request failed")
+	}
+	if writerErr != nil && !isUploadWriterAbort(writerErr) {
+		return nil, writerErr
+	}
+	if isUploadWriterAbort(writerErr) {
+		return nil, errors.New("upload request completed before multipart body was fully written")
+	}
+	if responseErr != nil {
+		return nil, fmt.Errorf("read upload response: %w", responseErr)
+	}
+	if len(respBody) > maxUploadResponseBytes {
+		return nil, fmt.Errorf("upload response exceeds %d bytes", maxUploadResponseBytes)
+	}
+	return respBody, nil
+}
+
+func isUploadWriterAbort(err error) bool {
+	return errors.Is(err, errUploadRequestAborted) || errors.Is(err, io.ErrClosedPipe)
+}
+
+func writeMultipartFile(
+	writer *multipart.Writer,
+	file io.Reader,
+	filename string,
+	mimetype string,
+) (resultErr error) {
+	defer func() {
+		if closeErr := writer.Close(); resultErr == nil && closeErr != nil {
+			resultErr = fmt.Errorf("close multipart writer: %w", closeErr)
+		}
+	}()
 
-	var buf bytes.Buffer
-	writer := multipart.NewWriter(&buf)
 	h := make(textproto.MIMEHeader)
 	h.Set("Content-Disposition", fmt.Sprintf(`form-data; name="file"; filename="%s"`, filename))
 	h.Set("Content-Type", mimetype)
 	part, err := writer.CreatePart(h)
 	if err != nil {
-		return nil, fmt.Errorf("create form file: %w", err)
+		return fmt.Errorf("create form file: %w", err)
 	}
 	if _, err := io.Copy(part, file); err != nil {
-		return nil, fmt.Errorf("copy file content: %w", err)
+		return fmt.Errorf("copy file content: %w", err)
 	}
-	if err := writer.Close(); err != nil {
-		return nil, fmt.Errorf("close multipart writer: %w", err)
-	}
-
-	uploadClient := &http.Client{Timeout: 120 * time.Second}
-	req, err := http.NewRequest("POST", uploadURL, &buf)
-	if err != nil {
-		return nil, fmt.Errorf("create upload request: %w", err)
-	}
-	req.Header.Set("Content-Type", writer.FormDataContentType())
-
-	resp, err := uploadClient.Do(req)
-	if err != nil {
-		return nil, fmt.Errorf("upload request failed: %w", err)
-	}
-	defer func() { _ = resp.Body.Close() }()
-
-	respBody, err := io.ReadAll(resp.Body)
-	if err != nil {
-		return nil, fmt.Errorf("read upload response: %w", err)
-	}
-	if resp.StatusCode >= 400 {
-		return nil, fmt.Errorf("upload failed with status %d: %s", resp.StatusCode, string(respBody))
-	}
-	return respBody, nil
+	return nil
 }
 
 // downloadFromURL downloads bytes from a signed URL.

dify-agent-runtime/internal/agentcli/httpclient_test.go

@@ -0,0 +1,678 @@
+package agentcli
+
+import (
+	"bytes"
+	"errors"
+	"fmt"
+	"io"
+	"mime/multipart"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+)
+
+const fifoTestDeadline = 3 * time.Second
+
+func receiveWithin[T any](ch <-chan T, timeout time.Duration) (T, bool) {
+	timer := time.NewTimer(timeout)
+	defer timer.Stop()
+	select {
+	case value := <-ch:
+		return value, true
+	case <-timer.C:
+		var zero T
+		return zero, false
+	}
+}
+
+type multipartWriteRecorder struct {
+	headerErr        error
+	cleanupErr       error
+	source           *terminalRecordingReader
+	headerAttempted  bool
+	headerFailed     bool
+	cleanupAttempted bool
+}
+
+func (w *multipartWriteRecorder) Write(p []byte) (int, error) {
+	if w.headerFailed || (w.source != nil && w.source.finished) {
+		w.cleanupAttempted = true
+		if w.cleanupErr != nil {
+			return 0, w.cleanupErr
+		}
+		return len(p), nil
+	}
+	if !w.headerAttempted {
+		w.headerAttempted = true
+		if w.headerErr != nil {
+			w.headerFailed = true
+			return 0, w.headerErr
+		}
+	}
+	return len(p), nil
+}
+
+type terminalRecordingReader struct {
+	reader      io.Reader
+	terminalErr error
+	finished    bool
+}
+
+func (r *terminalRecordingReader) Read(p []byte) (int, error) {
+	n, err := r.reader.Read(p)
+	if errors.Is(err, io.EOF) {
+		r.finished = true
+		if r.terminalErr != nil {
+			return n, r.terminalErr
+		}
+	}
+	return n, err
+}
+
+type dataThenErrorReader struct {
+	data      []byte
+	sourceErr error
+	delivered bool
+}
+
+type closeRecordingBody struct {
+	reader io.Reader
+	closed bool
+}
+
+type gatedUploadSource struct {
+	payload     []byte
+	started     chan struct{}
+	release     chan struct{}
+	completed   chan struct{}
+	releaseOnce sync.Once
+	delivered   bool
+	closed      bool
+}
+
+func newGatedUploadSource(payload []byte) *gatedUploadSource {
+	return &gatedUploadSource{
+		payload:   payload,
+		started:   make(chan struct{}),
+		release:   make(chan struct{}),
+		completed: make(chan struct{}),
+	}
+}
+
+func (s *gatedUploadSource) Read(p []byte) (int, error) {
+	if s.delivered {
+		return 0, io.EOF
+	}
+	s.delivered = true
+	close(s.started)
+	<-s.release
+	n := copy(p, s.payload)
+	close(s.completed)
+	return n, nil
+}
+
+func (s *gatedUploadSource) Close() error {
+	s.unblock()
+	s.closed = true
+	return nil
+}
+
+func (s *gatedUploadSource) unblock() {
+	s.releaseOnce.Do(func() { close(s.release) })
+}
+
+type releaseOnReadBody struct {
+	reader  io.Reader
+	release func()
+	closed  bool
+}
+
+func (b *releaseOnReadBody) Read(p []byte) (int, error) {
+	b.release()
+	return b.reader.Read(p)
+}
+
+func (b *releaseOnReadBody) Close() error {
+	b.release()
+	b.closed = true
+	return nil
+}
+
+func (b *closeRecordingBody) Read(p []byte) (int, error) {
+	return b.reader.Read(p)
+}
+
+func (b *closeRecordingBody) Close() error {
+	b.closed = true
+	return nil
+}
+
+func (r *dataThenErrorReader) Read(p []byte) (int, error) {
+	if r.delivered {
+		return 0, r.sourceErr
+	}
+	r.delivered = true
+	return copy(p, r.data), nil
+}
+
+func TestUploadFileStreamsMultipartBody(t *testing.T) {
+	payload := bytes.Repeat([]byte("streamed-payload-"), 128*1024)
+	filePath := filepath.Join(t.TempDir(), "payload.bin")
+	if err := os.WriteFile(filePath, payload, 0o600); err != nil {
+		t.Fatalf("write fixture: %v", err)
+	}
+
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.ContentLength != -1 {
+			t.Errorf("content length = %d, want -1 for streamed request", r.ContentLength)
+		}
+		reader, err := r.MultipartReader()
+		if err != nil {
+			http.Error(w, fmt.Sprintf("create multipart reader: %v", err), http.StatusBadRequest)
+			return
+		}
+		part, err := reader.NextPart()
+		if err != nil {
+			http.Error(w, fmt.Sprintf("read multipart part: %v", err), http.StatusBadRequest)
+			return
+		}
+		defer func() { _ = part.Close() }()
+		if part.FormName() != "file" || part.FileName() != "payload.bin" {
+			http.Error(w, "unexpected multipart metadata", http.StatusBadRequest)
+			return
+		}
+		if got := part.Header.Get("Content-Type"); got != "application/octet-stream" {
+			http.Error(w, "unexpected multipart content type: "+got, http.StatusBadRequest)
+			return
+		}
+		got, err := io.ReadAll(part)
+		if err != nil {
+			http.Error(w, fmt.Sprintf("read multipart content: %v", err), http.StatusBadRequest)
+			return
+		}
+		if !bytes.Equal(got, payload) {
+			http.Error(w, "multipart content mismatch", http.StatusBadRequest)
+			return
+		}
+		w.Header().Set("Content-Type", "application/json")
+		_, _ = w.Write([]byte(`{"reference":"dify-file-ref:canonical"}`))
+	}))
+	defer server.Close()
+
+	client := NewHTTPClient(&Environment{})
+	body, err := client.uploadFile(server.URL, filePath, "payload.bin", "application/octet-stream")
+	if err != nil {
+		t.Fatalf("upload file: %v", err)
+	}
+	if string(body) != `{"reference":"dify-file-ref:canonical"}` {
+		t.Fatalf("response body = %s", body)
+	}
+}
+
+func TestUploadFileStartsRequestBeforeSourceEOF(t *testing.T) {
+	if testing.Short() {
+		t.Skip("uses a local HTTP server and FIFO coordination")
+	}
+	mkfifo, err := exec.LookPath("mkfifo")
+	if err != nil {
+		t.Skip("mkfifo is unavailable")
+	}
+	fifoPath := filepath.Join(t.TempDir(), "stream.bin")
+	if err := exec.Command(mkfifo, "-m", "600", fifoPath).Run(); err != nil {
+		t.Fatalf("create FIFO: %v", err)
+	}
+
+	requestStarted := make(chan struct{})
+	continueSource := make(chan struct{})
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		close(requestStarted)
+		reader, err := r.MultipartReader()
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusBadRequest)
+			return
+		}
+		part, err := reader.NextPart()
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusBadRequest)
+			return
+		}
+		defer func() { _ = part.Close() }()
+		body, err := io.ReadAll(part)
+		if err != nil || string(body) != "prefix-suffix" {
+			http.Error(w, "unexpected streamed body", http.StatusBadRequest)
+			return
+		}
+		_, _ = w.Write([]byte(`{"reference":"dify-file-ref:eyJyZWNvcmRfaWQiOiJ0b29sLTEifQ=="}`))
+	}))
+	defer server.Close()
+
+	writerDone := make(chan error, 1)
+	go func() {
+		file, err := os.OpenFile(fifoPath, os.O_WRONLY, 0)
+		if err != nil {
+			writerDone <- err
+			return
+		}
+		if _, err = file.WriteString("prefix-"); err == nil {
+			<-continueSource
+			_, err = file.WriteString("suffix")
+		}
+		if closeErr := file.Close(); err == nil {
+			err = closeErr
+		}
+		writerDone <- err
+	}()
+
+	uploadDone := make(chan error, 1)
+	go func() {
+		client := NewHTTPClient(&Environment{})
+		_, err := client.uploadFile(server.URL, fifoPath, "stream.bin", "application/octet-stream")
+		uploadDone <- err
+	}()
+
+	releaseSource := sync.OnceFunc(func() { close(continueSource) })
+	writerJoined := false
+	uploadJoined := false
+	defer func() {
+		releaseSource()
+		server.CloseClientConnections()
+		if !writerJoined {
+			if err, ok := receiveWithin(writerDone, fifoTestDeadline); !ok {
+				t.Errorf("cleanup: FIFO writer did not finish within %s", fifoTestDeadline)
+			} else if err != nil {
+				t.Errorf("cleanup: FIFO writer failed: %v", err)
+			}
+		}
+		if !uploadJoined {
+			if err, ok := receiveWithin(uploadDone, fifoTestDeadline); !ok {
+				t.Errorf("cleanup: upload did not finish within %s", fifoTestDeadline)
+			} else if err != nil {
+				t.Errorf("cleanup: upload failed: %v", err)
+			}
+		}
+	}()
+
+	if _, ok := receiveWithin(requestStarted, fifoTestDeadline); !ok {
+		t.Fatalf("HTTP request did not start within %s before the source reached EOF", fifoTestDeadline)
+	}
+	releaseSource()
+
+	writerErr, ok := receiveWithin(writerDone, fifoTestDeadline)
+	if !ok {
+		t.Fatalf("FIFO writer did not finish within %s after source release", fifoTestDeadline)
+	}
+	writerJoined = true
+	if writerErr != nil {
+		t.Fatalf("write FIFO: %v", writerErr)
+	}
+	uploadErr, ok := receiveWithin(uploadDone, fifoTestDeadline)
+	if !ok {
+		t.Fatalf("upload did not finish within %s after source EOF", fifoTestDeadline)
+	}
+	uploadJoined = true
+	if uploadErr != nil {
+		t.Fatalf("upload FIFO: %v", uploadErr)
+	}
+}
+
+func TestUploadFileTransportFailureDoesNotBlockWriter(t *testing.T) {
+	const querySecret = "signed-upload-credential"
+	uploadURL := "https://upload.example/path?X-Amz-Credential=" + querySecret
+	source := &closeRecordingBody{reader: strings.NewReader("payload")}
+	client := NewHTTPClient(&Environment{})
+	client.openUploadFile = func(string) (io.ReadCloser, error) {
+		return source, nil
+	}
+	requestBodyReady := make(chan io.ReadCloser, 1)
+	client.doUploadRequest = func(req *http.Request) (*http.Response, error) {
+		requestBodyReady <- req.Body
+		return nil, errors.New("deterministic transport failure")
+	}
+
+	uploadDone := make(chan error, 1)
+	go func() {
+		_, err := client.uploadFile(uploadURL, "source-path", "payload.bin", "application/octet-stream")
+		uploadDone <- err
+	}()
+
+	var requestBody io.ReadCloser
+	uploadJoined := false
+	defer func() {
+		_ = source.Close()
+		if requestBody == nil {
+			requestBody, _ = receiveWithin(requestBodyReady, fifoTestDeadline)
+		}
+		if requestBody != nil {
+			_ = requestBody.Close()
+		}
+		if !uploadJoined {
+			if _, ok := receiveWithin(uploadDone, fifoTestDeadline); !ok {
+				t.Errorf("cleanup: transport-failure upload did not finish within %s", fifoTestDeadline)
+			}
+		}
+	}()
+
+	var ok bool
+	requestBody, ok = receiveWithin(requestBodyReady, fifoTestDeadline)
+	if !ok {
+		t.Fatalf("upload request did not start within %s", fifoTestDeadline)
+	}
+
+	err, ok := receiveWithin(uploadDone, fifoTestDeadline)
+	if !ok {
+		t.Fatalf("transport-failure upload did not finish within %s", fifoTestDeadline)
+	}
+	uploadJoined = true
+
+	if err == nil || err.Error() != "upload request failed" {
+		t.Fatalf("error = %v, want upload request failure", err)
+	}
+	if !source.closed {
+		t.Fatal("source file was not closed after transport failure")
+	}
+	if strings.Contains(err.Error(), uploadURL) || strings.Contains(err.Error(), querySecret) {
+		t.Fatalf("error leaked signed upload URL credentials: %v", err)
+	}
+}
+
+func TestUploadFileInvalidSignedURLDoesNotLeakCredentials(t *testing.T) {
+	filePath := filepath.Join(t.TempDir(), "payload.txt")
+	if err := os.WriteFile(filePath, []byte("payload"), 0o600); err != nil {
+		t.Fatalf("write fixture: %v", err)
+	}
+	const querySecret = "signed-upload-credential"
+	uploadURL := "http://example.test/upload?X-Amz-Credential=" + querySecret + "\n"
+
+	client := NewHTTPClient(&Environment{})
+	_, err := client.uploadFile(uploadURL, filePath, "payload.txt", "text/plain")
+	if err == nil || err.Error() != "create upload request: invalid signed upload URL" {
+		t.Fatalf("error = %v, want invalid signed upload URL failure", err)
+	}
+	if strings.Contains(err.Error(), uploadURL) || strings.Contains(err.Error(), querySecret) {
+		t.Fatalf("error leaked signed upload URL credentials: %v", err)
+	}
+}
+
+func TestUploadFileRejectsOversizedResponse(t *testing.T) {
+	filePath := filepath.Join(t.TempDir(), "payload.txt")
+	if err := os.WriteFile(filePath, []byte("payload"), 0o600); err != nil {
+		t.Fatalf("write fixture: %v", err)
+	}
+
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		_, _ = io.Copy(io.Discard, r.Body)
+		_, _ = w.Write(bytes.Repeat([]byte("x"), 1024*1024+1))
+	}))
+	defer server.Close()
+
+	client := NewHTTPClient(&Environment{})
+	_, err := client.uploadFile(server.URL, filePath, "payload.txt", "text/plain")
+	if err == nil || !strings.Contains(err.Error(), "upload response exceeds") {
+		t.Fatalf("error = %v, want bounded-response failure", err)
+	}
+}
+
+func TestUploadFileReturnsNonSuccessStatus(t *testing.T) {
+	filePath := filepath.Join(t.TempDir(), "payload.txt")
+	if err := os.WriteFile(filePath, []byte("payload"), 0o600); err != nil {
+		t.Fatalf("write fixture: %v", err)
+	}
+
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		_, _ = io.Copy(io.Discard, r.Body)
+		http.Error(w, "too large", http.StatusRequestEntityTooLarge)
+	}))
+	defer server.Close()
+
+	client := NewHTTPClient(&Environment{})
+	_, err := client.uploadFile(server.URL, filePath, "payload.txt", "text/plain")
+	if err == nil || !strings.Contains(err.Error(), "upload failed with status 413") {
+		t.Fatalf("error = %v, want non-success status", err)
+	}
+}
+
+func uploadFileWithGatedEarlyResponse(t *testing.T, statusCode int, body string) error {
+	t.Helper()
+	source := newGatedUploadSource([]byte("payload"))
+	responseBody := &releaseOnReadBody{reader: strings.NewReader(body), release: source.unblock}
+	requestDone := make(chan error, 1)
+	var orderingErr error
+	client := NewHTTPClient(&Environment{})
+	client.openUploadFile = func(string) (io.ReadCloser, error) {
+		return source, nil
+	}
+	client.doUploadRequest = func(req *http.Request) (*http.Response, error) {
+		go func() {
+			_, err := io.Copy(io.Discard, req.Body)
+			requestDone <- err
+		}()
+		if _, ok := receiveWithin(source.started, fifoTestDeadline); !ok {
+			orderingErr = errors.New("upload source did not start before the early response")
+			source.unblock()
+			return nil, orderingErr
+		}
+		select {
+		case <-source.completed:
+			orderingErr = errors.New("upload source completed before the early response")
+			return nil, orderingErr
+		default:
+		}
+		return &http.Response{StatusCode: statusCode, Body: responseBody}, nil
+	}
+
+	uploadDone := make(chan error, 1)
+	go func() {
+		_, err := client.uploadFile("https://upload.example/path", "source-path", "payload.bin", "application/octet-stream")
+		uploadDone <- err
+	}()
+	uploadJoined := false
+	requestJoined := false
+	defer func() {
+		source.unblock()
+		if !uploadJoined {
+			if _, ok := receiveWithin(uploadDone, fifoTestDeadline); !ok {
+				t.Errorf("cleanup: early-response upload did not finish within %s", fifoTestDeadline)
+			}
+		}
+		if !requestJoined {
+			if _, ok := receiveWithin(requestDone, fifoTestDeadline); !ok {
+				t.Errorf("cleanup: early-response request drain did not finish within %s", fifoTestDeadline)
+			}
+		}
+	}()
+
+	uploadErr, ok := receiveWithin(uploadDone, fifoTestDeadline)
+	if !ok {
+		t.Fatalf("early-response upload did not finish within %s", fifoTestDeadline)
+	}
+	uploadJoined = true
+	if _, ok := receiveWithin(requestDone, fifoTestDeadline); !ok {
+		t.Fatalf("early-response request drain did not finish within %s", fifoTestDeadline)
+	}
+	requestJoined = true
+	if orderingErr != nil {
+		t.Fatalf("invalid early-response ordering: %v", orderingErr)
+	}
+	if _, ok := receiveWithin(source.completed, fifoTestDeadline); !ok {
+		t.Fatalf("upload source did not finish within %s after response processing", fifoTestDeadline)
+	}
+	if !source.closed {
+		t.Fatal("upload source was not closed before uploadFile returned")
+	}
+	if !responseBody.closed {
+		t.Fatal("early response body was not closed before uploadFile returned")
+	}
+	return uploadErr
+}
+
+func TestUploadFileReturnsEarlyNonSuccessStatusInsteadOfWriterAbort(t *testing.T) {
+	err := uploadFileWithGatedEarlyResponse(
+		t,
+		http.StatusRequestEntityTooLarge,
+		"too large without reading body\n",
+	)
+	if err == nil || !strings.Contains(err.Error(), "upload failed with status 413: too large without reading body") {
+		t.Fatalf("error = %v, want early HTTP 413 response", err)
+	}
+	if strings.Contains(err.Error(), "upload request aborted") || strings.Contains(err.Error(), "closed pipe") {
+		t.Fatalf("error exposed internal multipart abort: %v", err)
+	}
+}
+
+func TestUploadFileRejectsEarlySuccessBeforeMultipartCompletes(t *testing.T) {
+	err := uploadFileWithGatedEarlyResponse(t, http.StatusOK, `{"reference":"incomplete"}`)
+	if err == nil || err.Error() != "upload request completed before multipart body was fully written" {
+		t.Fatalf("error = %v, want incomplete multipart failure", err)
+	}
+}
+
+func TestUploadFilePropagatesSourceReadFailure(t *testing.T) {
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		_, _ = io.Copy(io.Discard, r.Body)
+		_, _ = w.Write([]byte(`{"reference":"unused"}`))
+	}))
+	defer server.Close()
+
+	client := NewHTTPClient(&Environment{})
+	_, err := client.uploadFile(server.URL, t.TempDir(), "directory", "application/octet-stream")
+	if err == nil || !strings.Contains(err.Error(), "copy file content") {
+		t.Fatalf("error = %v, want source read failure", err)
+	}
+}
+
+func TestUploadFileClosesResourcesAndJoinsWriterOnResponseOutcomes(t *testing.T) {
+	responseReadErr := errors.New("response read failed")
+	tests := []struct {
+		name           string
+		statusCode     int
+		responseReader io.Reader
+		wantError      string
+	}{
+		{
+			name:           "success",
+			statusCode:     http.StatusOK,
+			responseReader: strings.NewReader(`{"reference":"dify-file-ref:canonical"}`),
+		},
+		{
+			name:           "non-2xx",
+			statusCode:     http.StatusRequestEntityTooLarge,
+			responseReader: strings.NewReader("too large"),
+			wantError:      "upload failed with status 413",
+		},
+		{
+			name:           "response body read error",
+			statusCode:     http.StatusOK,
+			responseReader: &dataThenErrorReader{data: []byte("partial"), sourceErr: responseReadErr},
+			wantError:      "read upload response: response read failed",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			source := &closeRecordingBody{reader: strings.NewReader("payload")}
+			var responseBody *closeRecordingBody
+			client := NewHTTPClient(&Environment{})
+			client.openUploadFile = func(path string) (io.ReadCloser, error) {
+				if path != "source-path" {
+					t.Fatalf("source path = %q", path)
+				}
+				return source, nil
+			}
+			client.doUploadRequest = func(req *http.Request) (*http.Response, error) {
+				if _, err := io.Copy(io.Discard, req.Body); err != nil {
+					t.Fatalf("drain request body: %v", err)
+				}
+				responseBody = &closeRecordingBody{reader: tt.responseReader}
+				return &http.Response{StatusCode: tt.statusCode, Body: responseBody}, nil
+			}
+
+			_, err := client.uploadFile("https://upload.example/path", "source-path", "payload.txt", "text/plain")
+
+			if tt.wantError == "" {
+				if err != nil {
+					t.Fatalf("upload file: %v", err)
+				}
+			} else if err == nil || !strings.Contains(err.Error(), tt.wantError) {
+				t.Fatalf("error = %v, want containing %q", err, tt.wantError)
+			}
+			if !source.closed {
+				t.Fatal("source file was not closed before uploadFile returned")
+			}
+			if responseBody == nil || !responseBody.closed {
+				t.Fatal("response body was not closed before uploadFile returned")
+			}
+		})
+	}
+}
+
+func TestWriteMultipartFileAlwaysAttemptsCloseAndPreservesPrimaryError(t *testing.T) {
+	t.Run("create part failure", func(t *testing.T) {
+		createErr := errors.New("create part failed")
+		closeErr := errors.New("close failed")
+		destination := &multipartWriteRecorder{headerErr: createErr, cleanupErr: closeErr}
+		writer := multipart.NewWriter(destination)
+
+		err := writeMultipartFile(
+			writer,
+			strings.NewReader("payload"),
+			"payload.txt",
+			"text/plain",
+		)
+
+		if !errors.Is(err, createErr) {
+			t.Fatalf("error = %v, want CreatePart failure", err)
+		}
+		if !destination.headerAttempted || !destination.cleanupAttempted {
+			t.Fatal("multipart header and cleanup were not both attempted")
+		}
+	})
+
+	t.Run("copy failure", func(t *testing.T) {
+		sourceErr := errors.New("source read failed")
+		closeErr := errors.New("close failed")
+		source := &terminalRecordingReader{reader: strings.NewReader("payload"), terminalErr: sourceErr}
+		destination := &multipartWriteRecorder{cleanupErr: closeErr, source: source}
+		writer := multipart.NewWriter(destination)
+
+		err := writeMultipartFile(
+			writer,
+			source,
+			"payload.txt",
+			"text/plain",
+		)
+
+		if !errors.Is(err, sourceErr) {
+			t.Fatalf("error = %v, want source copy failure", err)
+		}
+		if !destination.cleanupAttempted {
+			t.Fatal("multipart cleanup was not attempted after the source read failure")
+		}
+	})
+
+	t.Run("close failure", func(t *testing.T) {
+		closeErr := errors.New("close failed")
+		source := &terminalRecordingReader{reader: strings.NewReader("payload")}
+		destination := &multipartWriteRecorder{cleanupErr: closeErr, source: source}
+		writer := multipart.NewWriter(destination)
+
+		err := writeMultipartFile(
+			writer,
+			source,
+			"payload.txt",
+			"text/plain",
+		)
+
+		if !errors.Is(err, closeErr) || !strings.Contains(err.Error(), "close multipart writer") {
+			t.Fatalf("error = %v, want multipart Close failure", err)
+		}
+		if !destination.cleanupAttempted {
+			t.Fatal("multipart cleanup was not attempted after source EOF")
+		}
+	})
+}