diff --git a/acceptance/cmd/fs/cp/file-to-nonexistent-dir/foo.txt b/acceptance/cmd/fs/cp/file-to-nonexistent-dir/foo.txt
new file mode 100644
index 00000000000..8baef1b4abc
--- /dev/null
+++ b/acceptance/cmd/fs/cp/file-to-nonexistent-dir/foo.txt
@@ -0,0 +1 @@
+abc
diff --git a/acceptance/cmd/fs/cp/file-to-nonexistent-dir/out.test.toml b/acceptance/cmd/fs/cp/file-to-nonexistent-dir/out.test.toml
new file mode 100644
index 00000000000..c7c6f58ed6e
--- /dev/null
+++ b/acceptance/cmd/fs/cp/file-to-nonexistent-dir/out.test.toml
@@ -0,0 +1 @@
+Cloud = true
diff --git a/acceptance/cmd/fs/cp/file-to-nonexistent-dir/output.txt b/acceptance/cmd/fs/cp/file-to-nonexistent-dir/output.txt
new file mode 100644
index 00000000000..9122e25d50e
--- /dev/null
+++ b/acceptance/cmd/fs/cp/file-to-nonexistent-dir/output.txt
@@ -0,0 +1,109 @@
+
+=== foo.txt -> local
+>>> [CLI] fs cp foo.txt local/existing-from-local/
+foo.txt -> local/existing-from-local/foo.txt
+
+>>> [CLI] fs cp foo.txt local/nonexistent/mydir/
+Error: no such directory: local/nonexistent/mydir
+
+Exit code: 1
+
+=== foo.txt -> dbfs:/tmp/fs-cp-test-[UNIQUE_NAME]
+>>> [CLI] fs cp foo.txt dbfs:/tmp/fs-cp-test-[UNIQUE_NAME]/existing-from-local/
+foo.txt -> dbfs:/tmp/fs-cp-test-[UNIQUE_NAME]/existing-from-local/foo.txt
+
+>>> [CLI] fs cp foo.txt dbfs:/tmp/fs-cp-test-[UNIQUE_NAME]/nonexistent/mydir/
+Error: no such directory: /tmp/fs-cp-test-[UNIQUE_NAME]/nonexistent
+
+Exit code: 1
+
+=== foo.txt -> dbfs:/Volumes/main/default/fs-cp-test-[UNIQUE_NAME]
+>>> [CLI] fs cp foo.txt dbfs:/Volumes/main/default/fs-cp-test-[UNIQUE_NAME]/existing-from-local/
+foo.txt -> dbfs:/Volumes/main/default/fs-cp-test-[UNIQUE_NAME]/existing-from-local/foo.txt
+
+>>> [CLI] fs cp foo.txt dbfs:/Volumes/main/default/fs-cp-test-[UNIQUE_NAME]/nonexistent/mydir/
+Error: no such directory: /Volumes/main/default/fs-cp-test-[UNIQUE_NAME]/nonexistent
+
+Exit code: 1
+
+=== dbfs:/tmp/fs-cp-test-[UNIQUE_NAME]/src/foo.txt -> local
+>>> [CLI] fs cp dbfs:/tmp/fs-cp-test-[UNIQUE_NAME]/src/foo.txt local/existing-from-dbfs/
+dbfs:/tmp/fs-cp-test-[UNIQUE_NAME]/src/foo.txt -> local/existing-from-dbfs/foo.txt
+
+>>> [CLI] fs cp dbfs:/tmp/fs-cp-test-[UNIQUE_NAME]/src/foo.txt local/nonexistent/mydir/
+Error: no such directory: local/nonexistent/mydir
+
+Exit code: 1
+
+=== dbfs:/tmp/fs-cp-test-[UNIQUE_NAME]/src/foo.txt -> dbfs:/tmp/fs-cp-test-[UNIQUE_NAME]
+>>> [CLI] fs cp dbfs:/tmp/fs-cp-test-[UNIQUE_NAME]/src/foo.txt dbfs:/tmp/fs-cp-test-[UNIQUE_NAME]/existing-from-dbfs/
+dbfs:/tmp/fs-cp-test-[UNIQUE_NAME]/src/foo.txt -> dbfs:/tmp/fs-cp-test-[UNIQUE_NAME]/existing-from-dbfs/foo.txt
+
+>>> [CLI] fs cp dbfs:/tmp/fs-cp-test-[UNIQUE_NAME]/src/foo.txt dbfs:/tmp/fs-cp-test-[UNIQUE_NAME]/nonexistent/mydir/
+Error: no such directory: /tmp/fs-cp-test-[UNIQUE_NAME]/nonexistent
+
+Exit code: 1
+
+=== dbfs:/tmp/fs-cp-test-[UNIQUE_NAME]/src/foo.txt -> dbfs:/Volumes/main/default/fs-cp-test-[UNIQUE_NAME]
+>>> [CLI] fs cp dbfs:/tmp/fs-cp-test-[UNIQUE_NAME]/src/foo.txt dbfs:/Volumes/main/default/fs-cp-test-[UNIQUE_NAME]/existing-from-dbfs/
+dbfs:/tmp/fs-cp-test-[UNIQUE_NAME]/src/foo.txt -> dbfs:/Volumes/main/default/fs-cp-test-[UNIQUE_NAME]/existing-from-dbfs/foo.txt
+
+>>> [CLI] fs cp dbfs:/tmp/fs-cp-test-[UNIQUE_NAME]/src/foo.txt dbfs:/Volumes/main/default/fs-cp-test-[UNIQUE_NAME]/nonexistent/mydir/
+Error: no such directory: /Volumes/main/default/fs-cp-test-[UNIQUE_NAME]/nonexistent
+
+Exit code: 1
+
+=== dbfs:/Volumes/main/default/fs-cp-test-[UNIQUE_NAME]/src/foo.txt -> local
+>>> [CLI] fs cp dbfs:/Volumes/main/default/fs-cp-test-[UNIQUE_NAME]/src/foo.txt local/existing-from-volume/
+dbfs:/Volumes/main/default/fs-cp-test-[UNIQUE_NAME]/src/foo.txt -> local/existing-from-volume/foo.txt
+
+>>> [CLI] fs cp dbfs:/Volumes/main/default/fs-cp-test-[UNIQUE_NAME]/src/foo.txt local/nonexistent/mydir/
+Error: no such directory: local/nonexistent/mydir
+
+Exit code: 1
+
+=== dbfs:/Volumes/main/default/fs-cp-test-[UNIQUE_NAME]/src/foo.txt -> dbfs:/tmp/fs-cp-test-[UNIQUE_NAME]
+>>> [CLI] fs cp dbfs:/Volumes/main/default/fs-cp-test-[UNIQUE_NAME]/src/foo.txt dbfs:/tmp/fs-cp-test-[UNIQUE_NAME]/existing-from-volume/
+dbfs:/Volumes/main/default/fs-cp-test-[UNIQUE_NAME]/src/foo.txt -> dbfs:/tmp/fs-cp-test-[UNIQUE_NAME]/existing-from-volume/foo.txt
+
+>>> [CLI] fs cp dbfs:/Volumes/main/default/fs-cp-test-[UNIQUE_NAME]/src/foo.txt dbfs:/tmp/fs-cp-test-[UNIQUE_NAME]/nonexistent/mydir/
+Error: no such directory: /tmp/fs-cp-test-[UNIQUE_NAME]/nonexistent
+
+Exit code: 1
+
+=== dbfs:/Volumes/main/default/fs-cp-test-[UNIQUE_NAME]/src/foo.txt -> dbfs:/Volumes/main/default/fs-cp-test-[UNIQUE_NAME]
+>>> [CLI] fs cp dbfs:/Volumes/main/default/fs-cp-test-[UNIQUE_NAME]/src/foo.txt dbfs:/Volumes/main/default/fs-cp-test-[UNIQUE_NAME]/existing-from-volume/
+dbfs:/Volumes/main/default/fs-cp-test-[UNIQUE_NAME]/src/foo.txt -> dbfs:/Volumes/main/default/fs-cp-test-[UNIQUE_NAME]/existing-from-volume/foo.txt
+
+>>> [CLI] fs cp dbfs:/Volumes/main/default/fs-cp-test-[UNIQUE_NAME]/src/foo.txt dbfs:/Volumes/main/default/fs-cp-test-[UNIQUE_NAME]/nonexistent/mydir/
+Error: no such directory: /Volumes/main/default/fs-cp-test-[UNIQUE_NAME]/nonexistent
+
+Exit code: 1
+
+=== Copies landed in the existing directories
+>>> cat local/existing-from-local/foo.txt
+abc
+
+>>> [CLI] fs cat dbfs:/tmp/fs-cp-test-[UNIQUE_NAME]/existing-from-local/foo.txt
+abc
+
+>>> [CLI] fs cat dbfs:/Volumes/main/default/fs-cp-test-[UNIQUE_NAME]/existing-from-local/foo.txt
+abc
+
+>>> cat local/existing-from-dbfs/foo.txt
+abc
+
+>>> [CLI] fs cat dbfs:/tmp/fs-cp-test-[UNIQUE_NAME]/existing-from-dbfs/foo.txt
+abc
+
+>>> [CLI] fs cat dbfs:/Volumes/main/default/fs-cp-test-[UNIQUE_NAME]/existing-from-dbfs/foo.txt
+abc
+
+>>> cat local/existing-from-volume/foo.txt
+abc
+
+>>> [CLI] fs cat dbfs:/tmp/fs-cp-test-[UNIQUE_NAME]/existing-from-volume/foo.txt
+abc
+
+>>> [CLI] fs cat dbfs:/Volumes/main/default/fs-cp-test-[UNIQUE_NAME]/existing-from-volume/foo.txt
+abc
diff --git a/acceptance/cmd/fs/cp/file-to-nonexistent-dir/script b/acceptance/cmd/fs/cp/file-to-nonexistent-dir/script
new file mode 100644
index 00000000000..cd942e0d8b0
--- /dev/null
+++ b/acceptance/cmd/fs/cp/file-to-nonexistent-dir/script
@@ -0,0 +1,46 @@
+# A trailing slash on the target means "copy into this directory": it works when the directory
+# exists and fails with "no such directory" when it does not. Covers local, DBFS and UC Volume
+# sources and targets.
+CATALOG_NAME="main"
+SCHEMA_NAME="default"
+VOLUME_NAME="fs-cp-test-${UNIQUE_NAME}"
+VOLUME="dbfs:/Volumes/${CATALOG_NAME}/${SCHEMA_NAME}/${VOLUME_NAME}"
+DBFS="dbfs:/tmp/fs-cp-test-${UNIQUE_NAME}"
+
+cleanup() {
+  $CLI volumes delete "${CATALOG_NAME}.${SCHEMA_NAME}.${VOLUME_NAME}" 2>/dev/null || true
+  $CLI fs rm -r ${DBFS} 2>/dev/null || true
+}
+trap cleanup EXIT
+
+$CLI volumes create "${CATALOG_NAME}" "${SCHEMA_NAME}" "${VOLUME_NAME}" MANAGED >/dev/null
+for root in ${VOLUME} ${DBFS}; do
+  $CLI fs mkdir $root/src
+  $CLI fs cp foo.txt $root/src/foo.txt >/dev/null
+done
+
+for src_kind in local dbfs volume; do
+  case $src_kind in
+    local) src=foo.txt ;;
+    dbfs) src=${DBFS}/src/foo.txt ;;
+    volume) src=${VOLUME}/src/foo.txt ;;
+  esac
+  # Separate existing directory per source, so later copies are not skipped as already existing.
+  existing="existing-from-$src_kind"
+  mkdir -p local/$existing
+  $CLI fs mkdir ${DBFS}/$existing
+  $CLI fs mkdir ${VOLUME}/$existing
+  for dst in local ${DBFS} ${VOLUME}; do
+    title "$src -> $dst"
+    trace $CLI fs cp $src $dst/$existing/
+    errcode trace $CLI fs cp $src $dst/nonexistent/mydir/
+  done
+done
+
+title "Copies landed in the existing directories"
+for existing in existing-from-local existing-from-dbfs existing-from-volume; do
+  trace cat local/$existing/foo.txt
+  trace $CLI fs cat ${DBFS}/$existing/foo.txt
+  trace $CLI fs cat ${VOLUME}/$existing/foo.txt
+done
+rm -rf local
diff --git a/acceptance/cmd/fs/cp/file-to-nonexistent-dir/test.toml b/acceptance/cmd/fs/cp/file-to-nonexistent-dir/test.toml
new file mode 100644
index 00000000000..ab359727a00
--- /dev/null
+++ b/acceptance/cmd/fs/cp/file-to-nonexistent-dir/test.toml
@@ -0,0 +1,6 @@
+Cloud = true
+
+# Local paths in errors use the OS separator.
+[[Repls]]
+Old = 'local\\nonexistent\\mydir'
+New = 'local/nonexistent/mydir'
diff --git a/acceptance/cmd/fs/dbfs/out.test.toml b/acceptance/cmd/fs/dbfs/out.test.toml
new file mode 100644
index 00000000000..c7c6f58ed6e
--- /dev/null
+++ b/acceptance/cmd/fs/dbfs/out.test.toml
@@ -0,0 +1 @@
+Cloud = true
diff --git a/acceptance/cmd/fs/dbfs/output.txt b/acceptance/cmd/fs/dbfs/output.txt
new file mode 100644
index 00000000000..da4966b8648
--- /dev/null
+++ b/acceptance/cmd/fs/dbfs/output.txt
@@ -0,0 +1,51 @@
+
+=== Create a directory and write a file in it
+>>> [CLI] fs mkdir dbfs:/tmp/fs-dbfs-[UNIQUE_NAME]/dir
+
+>>> [CLI] fs cp big.bin dbfs:/tmp/fs-dbfs-[UNIQUE_NAME]/dir/big.bin
+big.bin -> dbfs:/tmp/fs-dbfs-[UNIQUE_NAME]/dir/big.bin
+
+=== Read the file back
+>>> cmp big.bin big.back
+contents match
+
+=== Writing over an existing file needs --overwrite
+>>> [CLI] fs cp big.bin dbfs:/tmp/fs-dbfs-[UNIQUE_NAME]/dir/big.bin
+big.bin -> dbfs:/tmp/fs-dbfs-[UNIQUE_NAME]/dir/big.bin (skipped; already exists)
+
+>>> [CLI] fs cp --overwrite big.bin dbfs:/tmp/fs-dbfs-[UNIQUE_NAME]/dir/big.bin
+big.bin -> dbfs:/tmp/fs-dbfs-[UNIQUE_NAME]/dir/big.bin
+
+=== Writing into a directory that does not exist fails
+>>> [CLI] fs cp big.bin dbfs:/tmp/fs-dbfs-[UNIQUE_NAME]/missing/big.bin
+Error: no such directory: /tmp/fs-dbfs-[UNIQUE_NAME]/missing
+
+Exit code: 1
+
+=== List a directory, then a file
+>>> [CLI] fs ls dbfs:/tmp/fs-dbfs-[UNIQUE_NAME]/dir
+big.bin
+
+>>> [CLI] fs ls dbfs:/tmp/fs-dbfs-[UNIQUE_NAME]/dir/big.bin
+Error: not a directory: /tmp/fs-dbfs-[UNIQUE_NAME]/dir/big.bin
+
+Exit code: 1
+
+=== cat a directory fails
+>>> [CLI] fs cat dbfs:/tmp/fs-dbfs-[UNIQUE_NAME]/dir
+Error: not a file: /tmp/fs-dbfs-[UNIQUE_NAME]/dir
+
+Exit code: 1
+
+=== rm a non-empty directory needs -r
+>>> [CLI] fs rm dbfs:/tmp/fs-dbfs-[UNIQUE_NAME]/dir
+Error: directory not empty: /tmp/fs-dbfs-[UNIQUE_NAME]/dir
+
+Exit code: 1
+
+>>> [CLI] fs rm -r dbfs:/tmp/fs-dbfs-[UNIQUE_NAME]/dir
+
+>>> [CLI] fs ls dbfs:/tmp/fs-dbfs-[UNIQUE_NAME]/dir
+Error: no such directory: /tmp/fs-dbfs-[UNIQUE_NAME]/dir
+
+Exit code: 1
diff --git a/acceptance/cmd/fs/dbfs/script b/acceptance/cmd/fs/dbfs/script
new file mode 100644
index 00000000000..7fde1aedafa
--- /dev/null
+++ b/acceptance/cmd/fs/dbfs/script
@@ -0,0 +1,38 @@
+# fs commands against DBFS: the filer's error mapping for DBFS-specific responses, and a file over
+# the 1 MB DBFS block size so writes and reads are split into several API calls.
+DBFS="dbfs:/tmp/fs-dbfs-${UNIQUE_NAME}"
+
+cleanup() {
+  $CLI fs rm -r ${DBFS} 2>/dev/null || true
+  rm -f big.bin big.back
+}
+trap cleanup EXIT
+
+python3 -c 'import sys; sys.stdout.buffer.write(bytes(range(256)) * 6000)' > big.bin
+
+title "Create a directory and write a file in it"
+trace $CLI fs mkdir ${DBFS}/dir
+trace $CLI fs cp big.bin ${DBFS}/dir/big.bin
+
+title "Read the file back"
+$CLI fs cp ${DBFS}/dir/big.bin big.back > /dev/null
+trace cmp big.bin big.back && echo "contents match"
+
+title "Writing over an existing file needs --overwrite"
+trace $CLI fs cp big.bin ${DBFS}/dir/big.bin
+trace $CLI fs cp --overwrite big.bin ${DBFS}/dir/big.bin
+
+title "Writing into a directory that does not exist fails"
+errcode trace $CLI fs cp big.bin ${DBFS}/missing/big.bin
+
+title "List a directory, then a file"
+trace $CLI fs ls ${DBFS}/dir
+errcode trace $CLI fs ls ${DBFS}/dir/big.bin
+
+title "cat a directory fails"
+errcode trace $CLI fs cat ${DBFS}/dir
+
+title "rm a non-empty directory needs -r"
+errcode trace $CLI fs rm ${DBFS}/dir
+trace $CLI fs rm -r ${DBFS}/dir
+errcode trace $CLI fs ls ${DBFS}/dir
diff --git a/acceptance/cmd/fs/dbfs/test.toml b/acceptance/cmd/fs/dbfs/test.toml
new file mode 100644
index 00000000000..c7c6f58ed6e
--- /dev/null
+++ b/acceptance/cmd/fs/dbfs/test.toml
@@ -0,0 +1 @@
+Cloud = true
diff --git a/integration/cmd/fs/cp_test.go b/integration/cmd/fs/cp_test.go
index cea277f105b..5633bb31ffc 100644
--- a/integration/cmd/fs/cp_test.go
+++ b/integration/cmd/fs/cp_test.go
@@ -369,37 +369,6 @@ func TestFsCpSourceIsDirectoryButTargetIsFile(t *testing.T) {
 	}
 }
 
-func TestFsCpFileToNonExistentDir(t *testing.T) {
-	t.Parallel()
-
-	for _, testCase := range copyTests() {
-		t.Run(testCase.name, func(t *testing.T) {
-			t.Parallel()
-
-			ctx := t.Context()
-			sourceFiler, sourceDir := testCase.setupSource(t)
-			targetFiler, targetDir := testCase.setupTarget(t)
-			setupSourceFile(t, ctx, sourceFiler)
-
-			// Create a directory in the target
-			err := targetFiler.Mkdir(ctx, "existingdir")
-			require.NoError(t, err)
-
-			// Copy file to existing directory with trailing slash - should succeed
-			existingDir := path.Join(targetDir, "existingdir") + "/"
-			testcli.RequireSuccessfulRun(t, ctx, "fs", "cp", path.Join(sourceDir, "foo.txt"), existingDir)
-			assertTargetFile(t, ctx, targetFiler, "existingdir/foo.txt")
-
-			// Try to copy file to a non-existent directory with trailing slash - should fail
-			nonExistentDir := path.Join(targetDir, "nonexistent", "mydir") + "/"
-			_, _, err = testcli.RequireErrorRun(t, ctx, "fs", "cp", path.Join(sourceDir, "foo.txt"), nonExistentDir)
-			assert.Error(t, err)
-			assert.Contains(t, err.Error(), "directory")
-			assert.Contains(t, err.Error(), "no such directory")
-		})
-	}
-}
-
 func TestFsCpFileToNonExistentDirWindowsPaths(t *testing.T) {
 	if runtime.GOOS != "windows" {
 		t.Skip("Skipping test on non-windows OS")
diff --git a/libs/testserver/dbfs.go b/libs/testserver/dbfs.go
new file mode 100644
index 00000000000..712db80920e
--- /dev/null
+++ b/libs/testserver/dbfs.go
@@ -0,0 +1,294 @@
+package testserver
+
+import (
+	"encoding/base64"
+	"encoding/json"
+	"fmt"
+	"path"
+	"slices"
+	"strconv"
+	"strings"
+
+	"github.com/databricks/databricks-sdk-go/service/files"
+)
+
+// fakeDbfs is the DBFS namespace (dbfs:/...), separate from workspace files: a path written
+// through the DBFS API is not visible through the workspace or Files APIs, as on a real workspace.
+type fakeDbfs struct {
+	files      map[string]dbfsFile
+	dirs       map[string]int64 // path -> modification time (ms)
+	handles    map[int64]*dbfsHandle
+	nextHandle int64
+}
+
+// The real API's per-call limits, enforced so that broken chunking in the CLI fails locally too.
+const (
+	dbfsMaxBlockSize   = 1 << 20
+	dbfsMaxReadSize    = 1 << 20
+	dbfsDefaultReadLen = 1 << 19
+)
+
+type dbfsFile struct {
+	data    []byte
+	modTime int64
+}
+
+// dbfsHandle is an open streaming write: blocks accumulate until close commits the file.
+type dbfsHandle struct {
+	path string
+	data []byte
+}
+
+func (s *FakeWorkspace) dbfsState() *fakeDbfs {
+	if s.dbfs == nil {
+		s.dbfs = &fakeDbfs{
+			files:   map[string]dbfsFile{},
+			dirs:    map[string]int64{"/": nowMilli()},
+			handles: map[int64]*dbfsHandle{},
+		}
+	}
+	return s.dbfs
+}
+
+func dbfsError(statusCode int, errorCode, message string) Response {
+	return Response{
+		StatusCode: statusCode,
+		Body:       map[string]string{"error_code": errorCode, "message": message},
+	}
+}
+
+func dbfsNotFound(p string) Response {
+	return dbfsError(404, "RESOURCE_DOES_NOT_EXIST", fmt.Sprintf("No file or directory exists on path %s.", p))
+}
+
+func dbfsCleanPath(p string) string {
+	p = strings.TrimPrefix(p, "dbfs:")
+	if !strings.HasPrefix(p, "/") {
+		p = "/" + p
+	}
+	return path.Clean(p)
+}
+
+// mkdirs creates p and its parents. It fails if any of them is a file.
+func (d *fakeDbfs) mkdirs(p string) *Response {
+	var missing []string
+	for dir := p; ; dir = path.Dir(dir) {
+		if dir == "/" {
+			break
+		}
+		if _, isFile := d.files[dir]; isFile {
+			resp := dbfsError(400, "RESOURCE_ALREADY_EXISTS", fmt.Sprintf("A file or directory already exists at the input path %s.", dir))
+			return &resp
+		}
+		if _, isDir := d.dirs[dir]; isDir {
+			break
+		}
+		missing = append(missing, dir)
+	}
+	now := nowMilli()
+	for _, dir := range missing {
+		d.dirs[dir] = now
+	}
+	return nil
+}
+
+func (s *FakeWorkspace) DbfsGetStatus(req Request) Response {
+	defer s.LockUnlock()()
+	d := s.dbfsState()
+	p := dbfsCleanPath(req.URL.Query().Get("path"))
+
+	if f, ok := d.files[p]; ok {
+		return Response{Body: files.FileInfo{Path: p, FileSize: int64(len(f.data)), ModificationTime: f.modTime}}
+	}
+	if modTime, ok := d.dirs[p]; ok {
+		return Response{Body: files.FileInfo{Path: p, IsDir: true, ModificationTime: modTime}}
+	}
+	return dbfsNotFound(p)
+}
+
+func (s *FakeWorkspace) DbfsList(req Request) Response {
+	defer s.LockUnlock()()
+	d := s.dbfsState()
+	p := dbfsCleanPath(req.URL.Query().Get("path"))
+
+	// Listing a file returns the file itself, which the CLI uses to detect "not a directory".
+	if f, ok := d.files[p]; ok {
+		return Response{Body: files.ListStatusResponse{Files: []files.FileInfo{
+			{Path: p, FileSize: int64(len(f.data)), ModificationTime: f.modTime},
+		}}}
+	}
+	if _, ok := d.dirs[p]; !ok {
+		return dbfsNotFound(p)
+	}
+
+	var entries []files.FileInfo
+	for filePath, f := range d.files {
+		if path.Dir(filePath) == p {
+			entries = append(entries, files.FileInfo{Path: filePath, FileSize: int64(len(f.data)), ModificationTime: f.modTime})
+		}
+	}
+	for dirPath, modTime := range d.dirs {
+		if dirPath != p && path.Dir(dirPath) == p {
+			entries = append(entries, files.FileInfo{Path: dirPath, IsDir: true, ModificationTime: modTime})
+		}
+	}
+	slices.SortFunc(entries, func(a, b files.FileInfo) int { return strings.Compare(a.Path, b.Path) })
+	return Response{Body: files.ListStatusResponse{Files: entries}}
+}
+
+func (s *FakeWorkspace) DbfsMkdirs(req Request) Response {
+	var request files.MkDirs
+	if err := json.Unmarshal(req.Body, &request); err != nil {
+		return dbfsError(400, "INVALID_PARAMETER_VALUE", err.Error())
+	}
+	defer s.LockUnlock()()
+	if resp := s.dbfsState().mkdirs(dbfsCleanPath(request.Path)); resp != nil {
+		return *resp
+	}
+	return Response{Body: map[string]any{}}
+}
+
+func (s *FakeWorkspace) DbfsCreate(req Request) Response {
+	var request files.Create
+	if err := json.Unmarshal(req.Body, &request); err != nil {
+		return dbfsError(400, "INVALID_PARAMETER_VALUE", err.Error())
+	}
+	defer s.LockUnlock()()
+	d := s.dbfsState()
+	p := dbfsCleanPath(request.Path)
+
+	if _, isDir := d.dirs[p]; isDir {
+		return dbfsError(400, "RESOURCE_ALREADY_EXISTS", fmt.Sprintf("A directory already exists at the input path %s.", p))
+	}
+	if _, isFile := d.files[p]; isFile && !request.Overwrite {
+		return dbfsError(400, "RESOURCE_ALREADY_EXISTS", fmt.Sprintf("A file or directory already exists at the input path %s.", p))
+	}
+	// Like the real API, creating a file creates its missing parent directories.
+	if resp := d.mkdirs(path.Dir(p)); resp != nil {
+		return *resp
+	}
+
+	d.nextHandle++
+	d.handles[d.nextHandle] = &dbfsHandle{path: p}
+	return Response{Body: files.CreateResponse{Handle: d.nextHandle}}
+}
+
+func (s *FakeWorkspace) DbfsAddBlock(req Request) Response {
+	var request files.AddBlock
+	if err := json.Unmarshal(req.Body, &request); err != nil {
+		return dbfsError(400, "INVALID_PARAMETER_VALUE", err.Error())
+	}
+	data, err := base64.StdEncoding.DecodeString(request.Data)
+	if err != nil {
+		return dbfsError(400, "INVALID_PARAMETER_VALUE", "data is not valid base64: "+err.Error())
+	}
+	defer s.LockUnlock()()
+	h, ok := s.dbfsState().handles[request.Handle]
+	if !ok {
+		return dbfsError(404, "RESOURCE_DOES_NOT_EXIST", fmt.Sprintf("No handle %d.", request.Handle))
+	}
+	if len(data) > dbfsMaxBlockSize {
+		return dbfsError(400, "MAX_BLOCK_SIZE_EXCEEDED", fmt.Sprintf("Block size %d exceeds the maximum of %d bytes.", len(data), dbfsMaxBlockSize))
+	}
+	h.data = append(h.data, data...)
+	return Response{Body: map[string]any{}}
+}
+
+func (s *FakeWorkspace) DbfsClose(req Request) Response {
+	var request files.Close
+	if err := json.Unmarshal(req.Body, &request); err != nil {
+		return dbfsError(400, "INVALID_PARAMETER_VALUE", err.Error())
+	}
+	defer s.LockUnlock()()
+	d := s.dbfsState()
+	h, ok := d.handles[request.Handle]
+	if !ok {
+		return dbfsError(404, "RESOURCE_DOES_NOT_EXIST", fmt.Sprintf("No handle %d.", request.Handle))
+	}
+	delete(d.handles, request.Handle)
+	d.files[h.path] = dbfsFile{data: h.data, modTime: nowMilli()}
+	return Response{Body: map[string]any{}}
+}
+
+func (s *FakeWorkspace) DbfsRead(req Request) Response {
+	defer s.LockUnlock()()
+	d := s.dbfsState()
+	q := req.URL.Query()
+	p := dbfsCleanPath(q.Get("path"))
+
+	if _, isDir := d.dirs[p]; isDir {
+		return dbfsError(400, "INVALID_PARAMETER_VALUE", fmt.Sprintf("The path %s is a directory.", p))
+	}
+	f, ok := d.files[p]
+	if !ok {
+		return dbfsNotFound(p)
+	}
+
+	offset := int64(0)
+	if v := q.Get("offset"); v != "" {
+		var err error
+		if offset, err = strconv.ParseInt(v, 10, 64); err != nil || offset < 0 {
+			return dbfsError(400, "INVALID_PARAMETER_VALUE", "offset must be a non-negative integer")
+		}
+	}
+	length := int64(dbfsDefaultReadLen)
+	if v := q.Get("length"); v != "" {
+		var err error
+		if length, err = strconv.ParseInt(v, 10, 64); err != nil || length < 0 {
+			return dbfsError(400, "INVALID_PARAMETER_VALUE", "length must be a non-negative integer")
+		}
+	}
+	if length > dbfsMaxReadSize {
+		return dbfsError(400, "MAX_READ_SIZE_EXCEEDED", fmt.Sprintf("Read length %d exceeds the maximum of %d bytes.", length, dbfsMaxReadSize))
+	}
+	start := min(offset, int64(len(f.data)))
+	end := min(start+length, int64(len(f.data)))
+	chunk := f.data[start:end]
+	return Response{Body: files.ReadResponse{
+		BytesRead: int64(len(chunk)),
+		Data:      base64.StdEncoding.EncodeToString(chunk),
+	}}
+}
+
+func (s *FakeWorkspace) DbfsDelete(req Request) Response {
+	var request files.Delete
+	if err := json.Unmarshal(req.Body, &request); err != nil {
+		return dbfsError(400, "INVALID_PARAMETER_VALUE", err.Error())
+	}
+	defer s.LockUnlock()()
+	d := s.dbfsState()
+	p := dbfsCleanPath(request.Path)
+
+	if _, isFile := d.files[p]; isFile {
+		delete(d.files, p)
+		return Response{Body: map[string]any{}}
+	}
+	if _, isDir := d.dirs[p]; !isDir {
+		// The real API succeeds when the path does not exist.
+		return Response{Body: map[string]any{}}
+	}
+
+	prefix := strings.TrimSuffix(p, "/") + "/"
+	var children []string
+	for filePath := range d.files {
+		if strings.HasPrefix(filePath, prefix) {
+			children = append(children, filePath)
+		}
+	}
+	for dirPath := range d.dirs {
+		if dirPath != p && strings.HasPrefix(dirPath, prefix) {
+			children = append(children, dirPath)
+		}
+	}
+	if len(children) > 0 && !request.Recursive {
+		return dbfsError(400, "IO_ERROR", "Directory is not empty. Use recursive delete or remove contents first.")
+	}
+	for _, child := range children {
+		delete(d.files, child)
+		delete(d.dirs, child)
+	}
+	if p != "/" {
+		delete(d.dirs, p)
+	}
+	return Response{Body: map[string]any{}}
+}
diff --git a/libs/testserver/dbfs_test.go b/libs/testserver/dbfs_test.go
new file mode 100644
index 00000000000..c70ebb2e9e6
--- /dev/null
+++ b/libs/testserver/dbfs_test.go
@@ -0,0 +1,67 @@
+package testserver_test
+
+import (
+	"encoding/base64"
+	"errors"
+	"testing"
+
+	"github.com/databricks/cli/libs/testserver"
+	"github.com/databricks/databricks-sdk-go"
+	"github.com/databricks/databricks-sdk-go/apierr"
+	"github.com/databricks/databricks-sdk-go/service/files"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func newDbfsTestClient(t *testing.T) *databricks.WorkspaceClient {
+	server := testserver.New(t)
+	testserver.AddDefaultHandlers(server)
+	w, err := databricks.NewWorkspaceClient(&databricks.Config{Host: server.URL, Token: "testtoken"})
+	require.NoError(t, err)
+	return w
+}
+
+func dbfsErrorCode(err error) string {
+	if aerr, ok := errors.AsType[*apierr.APIError](err); ok {
+		return aerr.ErrorCode
+	}
+	return ""
+}
+
+func TestDbfsEnforcesBlockAndReadLimits(t *testing.T) {
+	ctx := t.Context()
+	w := newDbfsTestClient(t)
+
+	created, err := w.Dbfs.Create(ctx, files.Create{Path: "/tmp/f"})
+	require.NoError(t, err)
+	err = w.Dbfs.AddBlock(ctx, files.AddBlock{Handle: created.Handle, Data: base64.StdEncoding.EncodeToString(make([]byte, 1<<20+1))})
+	assert.Equal(t, "MAX_BLOCK_SIZE_EXCEEDED", dbfsErrorCode(err))
+	require.NoError(t, w.Dbfs.AddBlock(ctx, files.AddBlock{Handle: created.Handle, Data: base64.StdEncoding.EncodeToString([]byte("abc"))}))
+	require.NoError(t, w.Dbfs.CloseByHandle(ctx, created.Handle))
+
+	_, err = w.Dbfs.Read(ctx, files.ReadDbfsRequest{Path: "/tmp/f", Length: 1<<20 + 1})
+	assert.Equal(t, "MAX_READ_SIZE_EXCEEDED", dbfsErrorCode(err))
+	_, err = w.Dbfs.Read(ctx, files.ReadDbfsRequest{Path: "/tmp/f", Offset: -1, Length: 1})
+	assert.Equal(t, "INVALID_PARAMETER_VALUE", dbfsErrorCode(err))
+
+	res, err := w.Dbfs.Read(ctx, files.ReadDbfsRequest{Path: "/tmp/f", Offset: 1, Length: 10})
+	require.NoError(t, err)
+	assert.Equal(t, int64(2), res.BytesRead)
+}
+
+func TestDbfsRecursiveDeleteOfRootKeepsRoot(t *testing.T) {
+	ctx := t.Context()
+	w := newDbfsTestClient(t)
+
+	require.NoError(t, w.Dbfs.MkdirsByPath(ctx, "/a/b"))
+	require.NoError(t, w.Dbfs.Delete(ctx, files.Delete{Path: "/", Recursive: true}))
+
+	info, err := w.Dbfs.GetStatusByPath(ctx, "/")
+	require.NoError(t, err)
+	assert.True(t, info.IsDir)
+	_, err = w.Dbfs.GetStatusByPath(ctx, "/a")
+	assert.Equal(t, "RESOURCE_DOES_NOT_EXIST", dbfsErrorCode(err))
+
+	// Used to loop forever once the root itself had been deleted.
+	require.NoError(t, w.Dbfs.MkdirsByPath(ctx, "/c"))
+}
diff --git a/libs/testserver/fake_workspace.go b/libs/testserver/fake_workspace.go
index dd5050f6f49..e2097048aeb 100644
--- a/libs/testserver/fake_workspace.go
+++ b/libs/testserver/fake_workspace.go
@@ -207,6 +207,9 @@ type FakeWorkspace struct {
 	files        map[string]FileEntry
 	repoIdByPath map[string]int64
 
+	// dbfs is the DBFS namespace, created on first use (see dbfs.go).
+	dbfs *fakeDbfs
+
 	// snapshots records immutable-snapshot content paths. Breaking the glass on a
 	// recorded path flips it dirty for good, as in the real API.
 	snapshots map[string]*snapshotRecord
diff --git a/libs/testserver/handlers.go b/libs/testserver/handlers.go
index 50ff904a573..06c18785480 100644
--- a/libs/testserver/handlers.go
+++ b/libs/testserver/handlers.go
@@ -465,6 +465,32 @@ func AddDefaultHandlers(server *Server) {
 		return data
 	})
 
+	// DBFS (dbfs:/ paths):
+	server.Handle("GET", "/api/2.0/dbfs/get-status", func(req Request) any {
+		return req.Workspace.DbfsGetStatus(req)
+	})
+	server.Handle("GET", "/api/2.0/dbfs/list", func(req Request) any {
+		return req.Workspace.DbfsList(req)
+	})
+	server.Handle("POST", "/api/2.0/dbfs/mkdirs", func(req Request) any {
+		return req.Workspace.DbfsMkdirs(req)
+	})
+	server.Handle("POST", "/api/2.0/dbfs/create", func(req Request) any {
+		return req.Workspace.DbfsCreate(req)
+	})
+	server.Handle("POST", "/api/2.0/dbfs/add-block", func(req Request) any {
+		return req.Workspace.DbfsAddBlock(req)
+	})
+	server.Handle("POST", "/api/2.0/dbfs/close", func(req Request) any {
+		return req.Workspace.DbfsClose(req)
+	})
+	server.Handle("GET", "/api/2.0/dbfs/read", func(req Request) any {
+		return req.Workspace.DbfsRead(req)
+	})
+	server.Handle("POST", "/api/2.0/dbfs/delete", func(req Request) any {
+		return req.Workspace.DbfsDelete(req)
+	})
+
 	server.Handle("GET", "/api/2.1/unity-catalog/current-metastore-assignment", func(req Request) any {
 		return TestMetastore
 	})