diff --git a/src/main/providers/filesystem-path-listing-provider.ts b/src/main/providers/filesystem-path-listing-provider.ts
new file mode 100644
index 00000000000..1cd5c500c8e
--- /dev/null
+++ b/src/main/providers/filesystem-path-listing-provider.ts
@@ -0,0 +1,6 @@
+export type FilesystemPathListingProvider = {
+  listFiles(
+    rootPath: string,
+    options?: { excludePaths?: string[]; signal?: AbortSignal; maxResults?: number }
+  ): Promise<string[]>
+}
diff --git a/src/main/providers/ssh-filesystem-directory-reader.ts b/src/main/providers/ssh-filesystem-directory-reader.ts
new file mode 100644
index 00000000000..5b25a08610f
--- /dev/null
+++ b/src/main/providers/ssh-filesystem-directory-reader.ts
@@ -0,0 +1,59 @@
+import type { DirEntry } from '../../shared/types'
+import {
+  assertFilesystemDirectoryWithinLimit,
+  resolveFilesystemDirectoryListingLimits
+} from '../../shared/filesystem-directory-listing-limit'
+import {
+  assertMobileFileDirectoryWithinLimit,
+  MOBILE_FILE_DIRECTORY_MAX_ENTRIES,
+  MOBILE_FILE_DIRECTORY_MAX_RETAINED_BYTES
+} from '../../shared/mobile-file-directory-limit'
+import { isMethodNotFoundError } from '../ssh/ssh-filesystem-stream-reader'
+import type { SshChannelMultiplexer } from '../ssh/ssh-channel-multiplexer'
+
+export const SSH_BOUNDED_READ_DIR_UNAVAILABLE_MESSAGE =
+  'Safe remote directory browsing requires an updated relay. Reconnect the SSH target before retrying.'
+
+export class SshFilesystemDirectoryReader {
+  private boundedMethodUnavailable = false
+
+  constructor(private readonly mux: SshChannelMultiplexer) {}
+
+  async readDir(
+    dirPath: string,
+    options?: { maxEntries?: number; maxRetainedBytes?: number }
+  ): Promise<DirEntry[]> {
+    if (this.boundedMethodUnavailable) {
+      throw new Error(SSH_BOUNDED_READ_DIR_UNAVAILABLE_MESSAGE)
+    }
+    const mobileLimits =
+      options?.maxEntries === MOBILE_FILE_DIRECTORY_MAX_ENTRIES &&
+      options.maxRetainedBytes === MOBILE_FILE_DIRECTORY_MAX_RETAINED_BYTES
+    const limits = mobileLimits
+      ? {
+          maxEntries: MOBILE_FILE_DIRECTORY_MAX_ENTRIES,
+          maxRetainedBytes: MOBILE_FILE_DIRECTORY_MAX_RETAINED_BYTES
+        }
+      : resolveFilesystemDirectoryListingLimits(options)
+    let entries: DirEntry[]
+    try {
+      entries = (await this.mux.request('fs.readDirBounded', {
+        dirPath,
+        ...limits
+      })) as DirEntry[]
+    } catch (error) {
+      if (!isMethodNotFoundError(error)) {
+        throw error
+      }
+      // Why: fallback would restore remote-side unbounded enumeration.
+      this.boundedMethodUnavailable = true
+      throw new Error(SSH_BOUNDED_READ_DIR_UNAVAILABLE_MESSAGE)
+    }
+    if (mobileLimits) {
+      assertMobileFileDirectoryWithinLimit(entries)
+    } else {
+      assertFilesystemDirectoryWithinLimit(entries, limits)
+    }
+    return entries
+  }
+}
diff --git a/src/main/providers/ssh-filesystem-download.test.ts b/src/main/providers/ssh-filesystem-download.test.ts
index d65735cb806..23af1490e7f 100644
--- a/src/main/providers/ssh-filesystem-download.test.ts
+++ b/src/main/providers/ssh-filesystem-download.test.ts
@@ -4,6 +4,7 @@ import { tmpdir } from 'node:os'
 import { join } from 'node:path'
 
 import { downloadFolderViaSftp } from './ssh-filesystem-download'
+import { SSH_DIRECTORY_TRANSFER_LIMITS } from '../ssh/ssh-directory-transfer-budget'
 
 type SftpEntryKind = 'directory' | 'file' | 'symlink' | 'fifo'
 
@@ -29,6 +30,31 @@ function sftpEntry(filename: string, kind: SftpEntryKind) {
   return { filename, longname: filename, attrs: sftpStats(kind) }
 }
 
+function withDirectoryHandles(sftp: Record<string, unknown>): unknown {
+  if (typeof sftp.opendir === 'function' && typeof sftp.close === 'function') {
+    return sftp
+  }
+  const readDirectory = sftp.readdir as (
+    path: string,
+    callback: (error?: Error, value?: unknown) => void
+  ) => void
+  const completed = new Set<string>()
+  return Object.assign(sftp, {
+    opendir: (path: string, callback: (error: undefined, handle: Buffer) => void) =>
+      callback(undefined, Buffer.from(path)),
+    readdir: (handle: Buffer, callback: (error?: Error, value?: unknown) => void) => {
+      const path = handle.toString('utf8')
+      if (completed.has(path)) {
+        callback(undefined, false)
+        return
+      }
+      completed.add(path)
+      readDirectory(path, callback)
+    },
+    close: (_handle: Buffer, callback: (error?: Error) => void) => callback()
+  })
+}
+
 describe('downloadFolderViaSftp', () => {
   const roots: string[] = []
 
@@ -64,7 +90,11 @@ describe('downloadFolderViaSftp', () => {
     }
 
     await expect(
-      downloadFolderViaSftp(async () => sftp as never, '/remote/src', destination)
+      downloadFolderViaSftp(
+        async () => withDirectoryHandles(sftp) as never,
+        '/remote/src',
+        destination
+      )
     ).rejects.toThrow("Remote entries map to the same local name 'a.txt'")
     expect(sftp.fastGet).toHaveBeenCalledTimes(1)
   })
@@ -83,7 +113,11 @@ describe('downloadFolderViaSftp', () => {
     }
 
     await expect(
-      downloadFolderViaSftp(async () => sftp as never, '/remote/src', destination)
+      downloadFolderViaSftp(
+        async () => withDirectoryHandles(sftp) as never,
+        '/remote/src',
+        destination
+      )
     ).rejects.toThrow("Cannot download unsupported remote entry 'build.pipe'")
     expect(sftp.fastGet).not.toHaveBeenCalled()
   })
@@ -103,7 +137,11 @@ describe('downloadFolderViaSftp', () => {
     }
 
     await expect(
-      downloadFolderViaSftp(async () => sftp as never, '/remote/src', destination)
+      downloadFolderViaSftp(
+        async () => withDirectoryHandles(sftp) as never,
+        '/remote/src',
+        destination
+      )
     ).rejects.toThrow("Cannot download symbolic link 'creds'")
     // The link target could be /etc/passwd; rejecting from directory-entry
     // metadata means it is never followed with stat or opened by fastGet.
@@ -125,7 +163,11 @@ describe('downloadFolderViaSftp', () => {
     }
 
     await expect(
-      downloadFolderViaSftp(async () => sftp as never, '/remote/src', destination)
+      downloadFolderViaSftp(
+        async () => withDirectoryHandles(sftp) as never,
+        '/remote/src',
+        destination
+      )
     ).rejects.toThrow("Remote entries map to the same local name 'download'")
     expect(sftp.fastGet).not.toHaveBeenCalled()
   })
@@ -146,9 +188,12 @@ describe('downloadFolderViaSftp', () => {
       end: vi.fn()
     }
 
-    await downloadFolderViaSftp(async () => sftp as never, sourcePath, destination, {
-      windowsRemotePaths: false
-    })
+    await downloadFolderViaSftp(
+      async () => withDirectoryHandles(sftp) as never,
+      sourcePath,
+      destination,
+      { windowsRemotePaths: false }
+    )
 
     expect(sftp.fastGet).toHaveBeenCalledWith(
       '/remote/parent\\literal/..\\secret.txt',
@@ -171,9 +216,12 @@ describe('downloadFolderViaSftp', () => {
     }
 
     await expect(
-      downloadFolderViaSftp(async () => sftp as never, 'C:/remote/src', destination, {
-        windowsRemotePaths: true
-      })
+      downloadFolderViaSftp(
+        async () => withDirectoryHandles(sftp) as never,
+        'C:/remote/src',
+        destination,
+        { windowsRemotePaths: true }
+      )
     ).rejects.toThrow("Invalid remote directory entry '..\\secret.txt'")
     expect(sftp.fastGet).not.toHaveBeenCalled()
   })
@@ -195,9 +243,12 @@ describe('downloadFolderViaSftp', () => {
     }
     const controller = new AbortController()
 
-    const result = downloadFolderViaSftp(async () => sftp as never, '/remote/src', destination, {
-      signal: controller.signal
-    })
+    const result = downloadFolderViaSftp(
+      async () => withDirectoryHandles(sftp) as never,
+      '/remote/src',
+      destination,
+      { signal: controller.signal }
+    )
     await vi.waitFor(() => expect(sftp.fastGet).toHaveBeenCalledTimes(1))
     controller.abort(new Error('renderer closed'))
 
@@ -222,22 +273,26 @@ describe('downloadFolderViaSftp', () => {
   it('cancels a pending SFTP directory read', async () => {
     const destination = await createDestination()
     let readDirCallback: ((error?: Error) => void) | undefined
+    const readdir = vi.fn((_path: string, callback: (error?: Error) => void) => {
+      readDirCallback = callback
+    })
     const sftp = {
       stat: vi.fn((_path: string, callback: (err: Error | undefined, value: unknown) => void) =>
         callback(undefined, sftpStats('directory'))
       ),
-      readdir: vi.fn((_path: string, callback: (error?: Error) => void) => {
-        readDirCallback = callback
-      }),
+      readdir,
       fastGet: vi.fn(),
       end: vi.fn()
     }
     const controller = new AbortController()
 
-    const result = downloadFolderViaSftp(async () => sftp as never, '/remote/src', destination, {
-      signal: controller.signal
-    })
-    await vi.waitFor(() => expect(sftp.readdir).toHaveBeenCalledTimes(1))
+    const result = downloadFolderViaSftp(
+      async () => withDirectoryHandles(sftp) as never,
+      '/remote/src',
+      destination,
+      { signal: controller.signal }
+    )
+    await vi.waitFor(() => expect(readdir).toHaveBeenCalledTimes(1))
     controller.abort(new Error('renderer closed'))
     readDirCallback?.(new Error('channel closed'))
 
@@ -245,4 +300,44 @@ describe('downloadFolderViaSftp', () => {
     expect(sftp.fastGet).not.toHaveBeenCalled()
     expect(sftp.end).toHaveBeenCalledTimes(1)
   })
+
+  it('streams directory chunks and stops at the retained entry budget', async () => {
+    const destination = await createDestination()
+    const handle = Buffer.from('handle')
+    const chunks = [
+      [sftpEntry('one.txt', 'file'), sftpEntry('two.txt', 'file')],
+      [sftpEntry('three.txt', 'file')],
+      false
+    ]
+    const sftp = {
+      stat: vi.fn((_path: string, callback: (err: Error | undefined, value: unknown) => void) =>
+        callback(undefined, sftpStats('directory'))
+      ),
+      opendir: vi.fn((_path: string, callback: (err: Error | undefined, value: Buffer) => void) =>
+        callback(undefined, handle)
+      ),
+      readdir: vi.fn(
+        (_handle: Buffer, callback: (err: Error | undefined, value: unknown) => void) =>
+          callback(undefined, chunks.shift())
+      ),
+      close: vi.fn((_handle: Buffer, callback: (err?: Error) => void) => callback()),
+      fastGet: vi.fn(),
+      end: vi.fn()
+    }
+
+    await expect(
+      downloadFolderViaSftp(
+        async () => withDirectoryHandles(sftp) as never,
+        '/remote/src',
+        destination,
+        {
+          limits: { ...SSH_DIRECTORY_TRANSFER_LIMITS, maximumEntries: 2 }
+        }
+      )
+    ).rejects.toThrow('SSH directory transfer exceeds the entries limit')
+
+    expect(sftp.readdir).toHaveBeenCalledTimes(2)
+    expect(sftp.close).toHaveBeenCalledWith(handle, expect.any(Function))
+    expect(sftp.fastGet).not.toHaveBeenCalled()
+  })
 })
diff --git a/src/main/providers/ssh-filesystem-download.ts b/src/main/providers/ssh-filesystem-download.ts
index f73ef4c655f..53136cc49d7 100644
--- a/src/main/providers/ssh-filesystem-download.ts
+++ b/src/main/providers/ssh-filesystem-download.ts
@@ -7,12 +7,25 @@ import {
   normalizeRuntimePathSeparators
 } from '../../shared/cross-platform-path'
 import { sanitizeLocalDownloadFilename } from '../local-download-filename'
-import { fastGetViaSftp, readDirViaSftp, statViaSftp } from './ssh-filesystem-provider-sftp'
+import {
+  fastGetViaSftp,
+  readDirectoryEntriesViaSftp,
+  statViaSftp
+} from './ssh-filesystem-provider-sftp'
+import {
+  SSH_FOLDER_DOWNLOAD_LIMITS,
+  SshDirectoryTransferBudget,
+  type SshDirectoryTransferLimits
+} from '../ssh/ssh-directory-transfer-budget'
 
 export type SftpFactory = (options?: { signal?: AbortSignal }) => Promise<SFTPWrapper>
 
 /** When known, windowsRemotePaths drives remote path joining; omit uses path-shape heuristics. */
-export type FolderDownloadOptions = { signal?: AbortSignal; windowsRemotePaths?: boolean }
+export type FolderDownloadOptions = {
+  signal?: AbortSignal
+  windowsRemotePaths?: boolean
+  limits?: SshDirectoryTransferLimits
+}
 
 const DOWNLOAD_UNAVAILABLE_MESSAGE =
   'Remote folder download is unavailable. Reconnect the SSH target and retry.'
@@ -82,13 +95,11 @@ async function downloadDirectoryTree(
   sftp: SFTPWrapper,
   sourceDir: string,
   destinationDir: string,
+  budget: SshDirectoryTransferBudget,
+  depth: number,
   signal?: AbortSignal,
   windowsRemotePaths?: boolean
 ): Promise<void> {
-  signal?.throwIfAborted()
-  const entries = (await readDirViaSftp(sftp, sourceDir, { signal })).filter(
-    (entry) => entry.filename !== '.' && entry.filename !== '..'
-  )
   signal?.throwIfAborted()
   const usedLocalNames = new Set<string>()
   const plannedEntries: {
@@ -96,7 +107,13 @@ async function downloadDirectoryTree(
     kind: 'directory' | 'file'
     localName: string
   }[] = []
-  for (const entry of entries) {
+  for await (const entry of readDirectoryEntriesViaSftp(sftp, sourceDir, { signal })) {
+    signal?.throwIfAborted()
+    if (entry.filename === '.' || entry.filename === '..') {
+      continue
+    }
+    const remotePath = joinSftpChildPath(sourceDir, entry.filename, windowsRemotePaths)
+    budget.recordPath(remotePath, depth + 1)
     const localName = sanitizeLocalDownloadFilename(entry.filename)
     if (usedLocalNames.has(localName)) {
       throw new Error(`Remote entries map to the same local name '${localName}'`)
@@ -115,7 +132,15 @@ async function downloadDirectoryTree(
     const remotePath = joinSftpChildPath(sourceDir, entry.filename, windowsRemotePaths)
     const localPath = join(destinationDir, localName)
     if (kind === 'directory') {
-      await downloadDirectoryTree(sftp, remotePath, localPath, signal, windowsRemotePaths)
+      await downloadDirectoryTree(
+        sftp,
+        remotePath,
+        localPath,
+        budget,
+        depth + 1,
+        signal,
+        windowsRemotePaths
+      )
       continue
     }
     // Why: filesystem semantics belong to the selected volume, not the host OS;
@@ -170,10 +195,14 @@ export async function downloadFolderViaSftp(
     if (!rootStats.isDirectory()) {
       throw new Error('Cannot download a file as a folder')
     }
+    const budget = new SshDirectoryTransferBudget(options?.limits ?? SSH_FOLDER_DOWNLOAD_LIMITS)
+    budget.recordPath(sourcePath, 0, { countEntry: false })
     await downloadDirectoryTree(
       sftp,
       sourcePath,
       destinationPath,
+      budget,
+      0,
       signal,
       options?.windowsRemotePaths
     )
diff --git a/src/main/providers/ssh-filesystem-provider-directory-limits.test.ts b/src/main/providers/ssh-filesystem-provider-directory-limits.test.ts
new file mode 100644
index 00000000000..0fc0e56705f
--- /dev/null
+++ b/src/main/providers/ssh-filesystem-provider-directory-limits.test.ts
@@ -0,0 +1,104 @@
+import { describe, expect, it, vi } from 'vitest'
+import {
+  FILESYSTEM_DIRECTORY_MAX_ENTRIES,
+  FILESYSTEM_DIRECTORY_MAX_RETAINED_BYTES
+} from '../../shared/filesystem-directory-listing-limit'
+import {
+  MOBILE_FILE_DIRECTORY_MAX_ENTRIES,
+  MOBILE_FILE_DIRECTORY_MAX_RETAINED_BYTES
+} from '../../shared/mobile-file-directory-limit'
+import { JsonRpcErrorCode } from '../ssh/relay-protocol'
+import { SSH_BOUNDED_READ_DIR_UNAVAILABLE_MESSAGE } from './ssh-filesystem-directory-reader'
+import { SshFilesystemProvider } from './ssh-filesystem-provider'
+
+function createProvider(request: ReturnType<typeof vi.fn>): SshFilesystemProvider {
+  return new SshFilesystemProvider('conn-1', {
+    request,
+    onNotification: vi.fn(() => () => {})
+  } as never)
+}
+
+describe('SshFilesystemProvider directory limits', () => {
+  it('routes normal listings through the bounded method without changing results', async () => {
+    const entries = [
+      { name: 'src', isDirectory: true, isSymlink: false },
+      { name: 'README.md', isDirectory: false, isSymlink: false }
+    ]
+    const request = vi.fn().mockResolvedValue(entries)
+    const provider = createProvider(request)
+
+    await expect(provider.readDir('/home/user/project')).resolves.toEqual(entries)
+    expect(request).toHaveBeenCalledWith('fs.readDirBounded', {
+      dirPath: '/home/user/project',
+      maxEntries: FILESYSTEM_DIRECTORY_MAX_ENTRIES,
+      maxRetainedBytes: FILESYSTEM_DIRECTORY_MAX_RETAINED_BYTES
+    })
+  })
+
+  it('forwards stricter mobile directory retention limits', async () => {
+    const request = vi.fn().mockResolvedValue([])
+    const provider = createProvider(request)
+
+    await provider.readDir('/home/user/project', {
+      maxEntries: 10_000,
+      maxRetainedBytes: 4 * 1024 * 1024
+    })
+
+    expect(request).toHaveBeenCalledWith('fs.readDirBounded', {
+      dirPath: '/home/user/project',
+      maxEntries: 10_000,
+      maxRetainedBytes: 4 * 1024 * 1024
+    })
+  })
+
+  it('validates bounded relay results before returning them', async () => {
+    const request = vi.fn().mockResolvedValue([
+      { name: 'one', isDirectory: false, isSymlink: false },
+      { name: 'two', isDirectory: false, isSymlink: false },
+      { name: 'three', isDirectory: false, isSymlink: false }
+    ])
+    const provider = createProvider(request)
+
+    await expect(
+      provider.readDir('/home/user/project', { maxEntries: 2, maxRetainedBytes: 1024 })
+    ).rejects.toThrow('This folder is too large to list safely')
+  })
+
+  it('validates bounded relay results on the mobile path too', async () => {
+    // Why: the mobile branch asserts against its own tighter ceiling, so a relay that ignores the
+    // forwarded limits has to be caught here rather than trusted because the request looked right.
+    const request = vi.fn().mockResolvedValue(
+      Array.from({ length: MOBILE_FILE_DIRECTORY_MAX_ENTRIES + 1 }, (_unused, index) => ({
+        name: `entry-${index}`,
+        isDirectory: false,
+        isSymlink: false
+      }))
+    )
+    const provider = createProvider(request)
+
+    await expect(
+      provider.readDir('/home/user/project', {
+        maxEntries: MOBILE_FILE_DIRECTORY_MAX_ENTRIES,
+        maxRetainedBytes: MOBILE_FILE_DIRECTORY_MAX_RETAINED_BYTES
+      })
+    ).rejects.toThrow(/too large/i)
+  })
+
+  it('requires reconnect instead of falling back to an unbounded old relay', async () => {
+    const request = vi
+      .fn()
+      .mockRejectedValue(
+        Object.assign(new Error('Method not found'), { code: JsonRpcErrorCode.MethodNotFound })
+      )
+    const provider = createProvider(request)
+
+    await expect(provider.readDir('/home/user/project')).rejects.toThrow(
+      SSH_BOUNDED_READ_DIR_UNAVAILABLE_MESSAGE
+    )
+    await expect(provider.readDir('/home/user/project')).rejects.toThrow(
+      SSH_BOUNDED_READ_DIR_UNAVAILABLE_MESSAGE
+    )
+    expect(request).toHaveBeenCalledTimes(1)
+    expect(request).not.toHaveBeenCalledWith('fs.readDir', expect.anything())
+  })
+})
diff --git a/src/main/providers/ssh-filesystem-provider-download-folder.test.ts b/src/main/providers/ssh-filesystem-provider-download-folder.test.ts
index c875301883d..e5b2979f0f5 100644
--- a/src/main/providers/ssh-filesystem-provider-download-folder.test.ts
+++ b/src/main/providers/ssh-filesystem-provider-download-folder.test.ts
@@ -29,6 +29,29 @@ function sftpEntry(filename: string, kind: SftpEntryKind) {
   return { filename, longname: filename, attrs: sftpStats(kind) }
 }
 
+function createSftpDirectoryHandleMethods(
+  readEntries: (remotePath: string) => ReturnType<typeof sftpEntry>[]
+) {
+  const completedPaths = new Set<string>()
+  return {
+    opendir: vi.fn((remotePath: string, callback: (error: undefined, handle: Buffer) => void) =>
+      callback(undefined, Buffer.from(remotePath))
+    ),
+    readdir: vi.fn(
+      (handle: Buffer, callback: (error?: Error & { code?: number }, value?: unknown) => void) => {
+        const remotePath = handle.toString('utf8')
+        if (completedPaths.has(remotePath)) {
+          callback(Object.assign(new Error('EOF'), { code: 1 }))
+          return
+        }
+        completedPaths.add(remotePath)
+        callback(undefined, readEntries(remotePath))
+      }
+    ),
+    close: vi.fn((_handle: Buffer, callback: (error?: Error) => void) => callback())
+  }
+}
+
 function createMockMux() {
   return {
     request: vi.fn(),
@@ -69,14 +92,10 @@ describe('SshFilesystemProvider downloadFolder', () => {
         (remotePath: string, callback: (err: Error | undefined, value: unknown) => void) =>
           callback(undefined, sftpStats(remotePath === '/remote/src' ? 'directory' : 'file'))
       ),
-      readdir: vi.fn(
-        (remotePath: string, callback: (err: Error | undefined, value: unknown) => void) =>
-          callback(
-            undefined,
-            remotePath === '/remote/src'
-              ? [sftpEntry('index.ts', 'file'), sftpEntry('lib', 'directory')]
-              : [sftpEntry('a.ts', 'file')]
-          )
+      ...createSftpDirectoryHandleMethods((remotePath) =>
+        remotePath === '/remote/src'
+          ? [sftpEntry('index.ts', 'file'), sftpEntry('lib', 'directory')]
+          : [sftpEntry('a.ts', 'file')]
       ),
       fastGet: vi.fn((_source: string, _destination: string, callback: (err?: Error) => void) =>
         callback()
@@ -95,6 +114,7 @@ describe('SshFilesystemProvider downloadFolder', () => {
       ['/remote/src/lib/a.ts', join(destination, 'lib', 'a.ts')]
     ])
     await expect(stat(join(destination, 'lib'))).resolves.toMatchObject({})
+    expect(sftp.close).toHaveBeenCalledTimes(2)
     expect(sftp.end).toHaveBeenCalledTimes(1)
     expect(mux.request).not.toHaveBeenCalled()
   })
@@ -107,9 +127,7 @@ describe('SshFilesystemProvider downloadFolder', () => {
         (remotePath: string, callback: (err: Error | undefined, value: unknown) => void) =>
           callback(undefined, sftpStats(remotePath === '/remote/src' ? 'directory' : 'directory'))
       ),
-      readdir: vi.fn((_path: string, callback: (err: Error | undefined, value: unknown) => void) =>
-        callback(undefined, [sftpEntry('linked-dir', 'symlink')])
-      ),
+      ...createSftpDirectoryHandleMethods(() => [sftpEntry('linked-dir', 'symlink')]),
       fastGet: vi.fn(),
       end: vi.fn()
     }
@@ -120,6 +138,7 @@ describe('SshFilesystemProvider downloadFolder', () => {
     )
 
     expect(sftp.fastGet).not.toHaveBeenCalled()
+    expect(sftp.close).toHaveBeenCalledTimes(1)
     expect(sftp.end).toHaveBeenCalledTimes(1)
   })
 
@@ -130,9 +149,10 @@ describe('SshFilesystemProvider downloadFolder', () => {
       stat: vi.fn((_path: string, callback: (err: Error | undefined, value: unknown) => void) =>
         callback(undefined, sftpStats('directory'))
       ),
-      readdir: vi.fn((_path: string, callback: (err: Error | undefined, value: unknown) => void) =>
-        callback(undefined, [sftpEntry('a:b.txt', 'file'), sftpEntry('a?b.txt', 'file')])
-      ),
+      ...createSftpDirectoryHandleMethods(() => [
+        sftpEntry('a:b.txt', 'file'),
+        sftpEntry('a?b.txt', 'file')
+      ]),
       fastGet: vi.fn(),
       end: vi.fn()
     }
@@ -143,6 +163,7 @@ describe('SshFilesystemProvider downloadFolder', () => {
     )
 
     expect(sftp.fastGet).not.toHaveBeenCalled()
+    expect(sftp.close).toHaveBeenCalledTimes(1)
     expect(sftp.end).toHaveBeenCalledTimes(1)
   })
 
diff --git a/src/main/providers/ssh-filesystem-provider-sftp.ts b/src/main/providers/ssh-filesystem-provider-sftp.ts
index a3b1b840be3..af1af6c2ac2 100644
--- a/src/main/providers/ssh-filesystem-provider-sftp.ts
+++ b/src/main/providers/ssh-filesystem-provider-sftp.ts
@@ -95,15 +95,38 @@ export function fastGetViaSftp(
   )
 }
 
-export function readDirViaSftp(
+export async function* readDirectoryEntriesViaSftp(
   sftp: SFTPWrapper,
   dirPath: string,
   options?: { signal?: AbortSignal }
-): Promise<FileEntryWithStats[]> {
-  return waitForSftpCallback<FileEntryWithStats[]>(
-    (callback) => sftp.readdir(dirPath, callback),
+): AsyncGenerator<FileEntryWithStats> {
+  const handle = await waitForSftpCallback<Buffer>(
+    (callback) => sftp.opendir(dirPath, callback),
     options
   )
+  try {
+    for (;;) {
+      const entries = await waitForSftpCallback<FileEntryWithStats[] | false>(
+        (callback) =>
+          sftp.readdir(handle, (error, value) => {
+            if ((error as { code?: number } | undefined)?.code === 1) {
+              callback(null, false)
+              return
+            }
+            callback(error, value && value.length > 0 ? value : false)
+          }),
+        options
+      )
+      if (entries === false) {
+        break
+      }
+      yield* entries
+    }
+  } finally {
+    await waitForSftpCallback<void>((callback) => sftp.close(handle, callback), options).catch(
+      () => undefined
+    )
+  }
 }
 
 export function statViaSftp(
diff --git a/src/main/providers/ssh-filesystem-provider.test.ts b/src/main/providers/ssh-filesystem-provider.test.ts
index a7061616c4a..e8409249c0c 100644
--- a/src/main/providers/ssh-filesystem-provider.test.ts
+++ b/src/main/providers/ssh-filesystem-provider.test.ts
@@ -63,20 +63,6 @@ describe('SshFilesystemProvider', () => {
     expect(provider.getConnectionId()).toBe('conn-1')
   })
 
-  describe('readDir', () => {
-    it('sends fs.readDir request', async () => {
-      const entries = [
-        { name: 'src', isDirectory: true, isSymlink: false },
-        { name: 'README.md', isDirectory: false, isSymlink: false }
-      ]
-      mux.request.mockResolvedValue(entries)
-
-      const result = await provider.readDir('/home/user/project')
-      expect(mux.request).toHaveBeenCalledWith('fs.readDir', { dirPath: '/home/user/project' })
-      expect(result).toEqual(entries)
-    })
-  })
-
   describe('readFile', () => {
     it('short-circuits on empty:true metadata without subscribing to chunks', async () => {
       mux.request.mockResolvedValue({ totalSize: 0, isBinary: false, empty: true })
diff --git a/src/main/providers/ssh-filesystem-provider.ts b/src/main/providers/ssh-filesystem-provider.ts
index 74f31b8e6bf..d9a304af305 100644
--- a/src/main/providers/ssh-filesystem-provider.ts
+++ b/src/main/providers/ssh-filesystem-provider.ts
@@ -25,6 +25,7 @@ import type { DirEntry, FsChangeEvent, SearchOptions, SearchResult } from '../..
 import { routeSshFilesystemWatchNotification } from './ssh-filesystem-watch-notifications'
 import type { WorkspaceSpaceDirectoryScanResult } from '../../shared/workspace-space-types'
 import { isWindowsRemoteHost, type RemoteHostPlatform } from '../ssh/ssh-remote-platform'
+import { SshFilesystemDirectoryReader } from './ssh-filesystem-directory-reader'
 const WORKSPACE_SPACE_SCAN_TIMEOUT_MS = 130_000
 
 export class SshFilesystemProvider implements IFilesystemProvider {
@@ -35,6 +36,7 @@ export class SshFilesystemProvider implements IFilesystemProvider {
   private tempDirPromise: Promise<string> | null = null
   private disposed = false
   private loggedStreamFallback = false
+  private readonly directoryReader: SshFilesystemDirectoryReader
   readonly downloadFolder?: IFilesystemProvider['downloadFolder']
 
   constructor(
@@ -46,6 +48,7 @@ export class SshFilesystemProvider implements IFilesystemProvider {
   ) {
     this.connectionId = connectionId
     this.mux = mux
+    this.directoryReader = new SshFilesystemDirectoryReader(mux)
 
     if (createSftp) {
       // Why: system SSH has raw single-file transfer but no ssh2 SFTP channel;
@@ -83,8 +86,11 @@ export class SshFilesystemProvider implements IFilesystemProvider {
     return this.connectionId
   }
 
-  async readDir(dirPath: string): Promise<DirEntry[]> {
-    return (await this.mux.request('fs.readDir', { dirPath })) as DirEntry[]
+  async readDir(
+    dirPath: string,
+    options?: { maxEntries?: number; maxRetainedBytes?: number }
+  ): Promise<DirEntry[]> {
+    return this.directoryReader.readDir(dirPath, options)
   }
 
   async readFile(filePath: string): Promise<FileReadResult> {
diff --git a/src/main/providers/types.ts b/src/main/providers/types.ts
index 6fe63f70cc1..822f540bba0 100644
--- a/src/main/providers/types.ts
+++ b/src/main/providers/types.ts
@@ -21,6 +21,7 @@ import type { GitHistoryOptions, GitHistoryResult } from '../../shared/git-histo
 import type { PtyStartupIngressIntent } from '../../shared/pty-startup-ingress'
 import type { CommitMessageDraftContext } from '../../shared/commit-message-generation'
 import type { WorkspaceSpaceDirectoryScanResult } from '../../shared/workspace-space-types'
+import type { FilesystemPathListingProvider } from './filesystem-path-listing-provider'
 import type { StartupCommandDelivery } from '../../shared/codex-startup-delivery'
 import type { TerminalOscLinkRange } from '../../shared/terminal-osc-link-ranges'
 import type { GitProviderStatusOptions } from './git-provider-status-options'
@@ -228,8 +229,10 @@ export type FileReadResult = {
   imageDimensions?: { width: number; height: number }
 }
 
-export type IFilesystemProvider = {
-  readDir(dirPath: string): Promise<DirEntry[]>
+type FilesystemDirectoryReadOptions = { maxEntries?: number; maxRetainedBytes?: number }
+
+export type IFilesystemProvider = FilesystemPathListingProvider & {
+  readDir(dirPath: string, options?: FilesystemDirectoryReadOptions): Promise<DirEntry[]>
   readFile(filePath: string): Promise<FileReadResult>
   readTerminalArtifact?(
     filePath: string,
@@ -258,10 +261,6 @@ export type IFilesystemProvider = {
   copy(source: string, destination: string): Promise<void>
   realpath(filePath: string): Promise<string>
   search(opts: SearchOptions): Promise<SearchResult>
-  listFiles(
-    rootPath: string,
-    options?: { excludePaths?: string[]; signal?: AbortSignal; maxResults?: number }
-  ): Promise<string[]>
   scanWorkspaceSpace?(
     rootPath: string,
     options?: { signal?: AbortSignal }
diff --git a/src/main/ssh/ssh-directory-transfer-budget.test.ts b/src/main/ssh/ssh-directory-transfer-budget.test.ts
new file mode 100644
index 00000000000..0e2c94594f2
--- /dev/null
+++ b/src/main/ssh/ssh-directory-transfer-budget.test.ts
@@ -0,0 +1,51 @@
+import { describe, expect, it } from 'vitest'
+import {
+  SshDirectoryTransferBudget,
+  type SshDirectoryTransferLimits
+} from './ssh-directory-transfer-budget'
+
+const limits: SshDirectoryTransferLimits = {
+  maximumEntries: 2,
+  maximumDepth: 2,
+  maximumPathBytes: 4,
+  maximumRetainedPathBytes: 8,
+  maximumFileBytes: 4,
+  maximumTotalFileBytes: 8
+}
+
+describe('SshDirectoryTransferBudget', () => {
+  it('accepts every exact boundary', () => {
+    const budget = new SshDirectoryTransferBudget(limits)
+
+    budget.recordPath('1234', 2)
+    budget.recordPath('5678', 2)
+    budget.recordFile(4)
+    budget.recordFile(4)
+  })
+
+  it.each([
+    ['entries', (budget: SshDirectoryTransferBudget) => budget.recordPath('', 0)],
+    ['depth', (budget: SshDirectoryTransferBudget) => budget.recordPath('', 3)],
+    ['path', (budget: SshDirectoryTransferBudget) => budget.recordPath('12345', 0)],
+    [
+      'paths',
+      (budget: SshDirectoryTransferBudget) => budget.recordPath('1', 0, { countEntry: false })
+    ],
+    ['file', (budget: SshDirectoryTransferBudget) => budget.recordFile(5)],
+    ['files', (budget: SshDirectoryTransferBudget) => budget.recordFile(1)]
+  ] as const)('rejects one unit beyond the %s boundary', (reason, exceed) => {
+    const budget = new SshDirectoryTransferBudget(limits)
+    if (reason === 'entries') {
+      budget.recordPath('', 0)
+      budget.recordPath('', 0)
+    } else if (reason === 'paths') {
+      budget.recordPath('1234', 0, { countEntry: false })
+      budget.recordPath('5678', 0, { countEntry: false })
+    } else if (reason === 'files') {
+      budget.recordFile(4)
+      budget.recordFile(4)
+    }
+
+    expect(() => exceed(budget)).toThrow(expect.objectContaining({ reason }))
+  })
+})
diff --git a/src/main/ssh/ssh-directory-transfer-budget.ts b/src/main/ssh/ssh-directory-transfer-budget.ts
new file mode 100644
index 00000000000..65c609ae518
--- /dev/null
+++ b/src/main/ssh/ssh-directory-transfer-budget.ts
@@ -0,0 +1,78 @@
+export type SshDirectoryTransferLimits = {
+  maximumEntries: number
+  maximumDepth: number
+  maximumPathBytes: number
+  maximumRetainedPathBytes: number
+  maximumFileBytes: number
+  maximumTotalFileBytes: number
+}
+
+export const SSH_DIRECTORY_TRANSFER_LIMITS: SshDirectoryTransferLimits = {
+  maximumEntries: 4_096,
+  maximumDepth: 64,
+  maximumPathBytes: 16 * 1024,
+  maximumRetainedPathBytes: 4 * 1024 * 1024,
+  maximumFileBytes: 16 * 1024 * 1024,
+  maximumTotalFileBytes: 32 * 1024 * 1024
+}
+
+// Why: folder DOWNLOAD streams each file straight to disk and only holds one directory's entry list
+// in memory at a time, so its real cost is bounded by depth plus that directory's size — far below
+// the whole-tree entry count. The limits above are sized for an in-memory upload package, and their
+// 4,096-entry ceiling rejects an ordinary checkout containing node_modules. Downloads get a much
+// higher whole-tree backstop while keeping the depth and per-path guards.
+export const SSH_FOLDER_DOWNLOAD_LIMITS: SshDirectoryTransferLimits = {
+  maximumEntries: 1_000_000,
+  maximumDepth: 64,
+  maximumPathBytes: 16 * 1024,
+  maximumRetainedPathBytes: 256 * 1024 * 1024,
+  maximumFileBytes: SSH_DIRECTORY_TRANSFER_LIMITS.maximumFileBytes,
+  maximumTotalFileBytes: Number.MAX_SAFE_INTEGER
+}
+
+export const WINDOWS_SSH_UPLOAD_PACKAGE_MAX_BYTES = 48 * 1024 * 1024
+
+export class SshDirectoryTransferCapacityError extends Error {
+  constructor(readonly reason: 'entries' | 'depth' | 'path' | 'paths' | 'file' | 'files') {
+    super(`SSH directory transfer exceeds the ${reason} limit`)
+    this.name = 'SshDirectoryTransferCapacityError'
+  }
+}
+
+export class SshDirectoryTransferBudget {
+  private entries = 0
+  private retainedPathBytes = 0
+  private totalFileBytes = 0
+
+  constructor(private readonly limits = SSH_DIRECTORY_TRANSFER_LIMITS) {}
+
+  recordPath(path: string, depth: number, options?: { countEntry?: boolean }): void {
+    if (depth > this.limits.maximumDepth) {
+      throw new SshDirectoryTransferCapacityError('depth')
+    }
+    if (options?.countEntry !== false) {
+      this.entries += 1
+      if (this.entries > this.limits.maximumEntries) {
+        throw new SshDirectoryTransferCapacityError('entries')
+      }
+    }
+    const pathBytes = Buffer.byteLength(path, 'utf8')
+    if (pathBytes > this.limits.maximumPathBytes) {
+      throw new SshDirectoryTransferCapacityError('path')
+    }
+    this.retainedPathBytes += pathBytes
+    if (this.retainedPathBytes > this.limits.maximumRetainedPathBytes) {
+      throw new SshDirectoryTransferCapacityError('paths')
+    }
+  }
+
+  recordFile(fileBytes: number): void {
+    if (fileBytes > this.limits.maximumFileBytes) {
+      throw new SshDirectoryTransferCapacityError('file')
+    }
+    this.totalFileBytes += fileBytes
+    if (this.totalFileBytes > this.limits.maximumTotalFileBytes) {
+      throw new SshDirectoryTransferCapacityError('files')
+    }
+  }
+}