perf: batch remote filesystem walks

.changeset/batch-remote-filesystem-walks.md

@@ -0,0 +1,5 @@
+---
+"@moonshot-ai/kimi-code": patch
+---
+
+Speed up file completion in remote environments without ripgrep.

packages/acp-server/src/acp-fs/acpFsService.ts

@@ -20,6 +20,8 @@ import {
   HostFileSystem,
   type HostDirEntry,
   type HostFileStat,
+  type HostWalkOptions,
+  type HostWalkResult,
   IHostFileSystem,
   ISessionContext,
   LifecycleScope,
@@ -151,6 +153,10 @@ export class AcpHostFileSystem implements IHostFileSystem {
     return this.inner.readdir(path);
   }
 
+  walk(path: string, options: HostWalkOptions): Promise<HostWalkResult> {
+    return this.inner.walk(path, options);
+  }
+
   mkdir(path: string, options?: { readonly recursive?: boolean }): Promise<void> {
     return this.inner.mkdir(path, options);
   }

packages/agent-core-v2/src/os/backends/node-local/hostFsService.ts

@@ -14,8 +14,15 @@ import {
 } from 'node:fs/promises';
 import { decodeTextWithErrors, readUtf8Lines, type TextDecodeErrors } from '#/_base/execEnv/decodeText';
 
-import { type HostDirEntry, type HostFileStat, IHostFileSystem } from '#/os/interface/hostFileSystem';
+import {
+  type HostDirEntry,
+  type HostFileStat,
+  type HostWalkOptions,
+  type HostWalkResult,
+  IHostFileSystem,
+} from '#/os/interface/hostFileSystem';
 import { toHostFsError } from '#/os/interface/hostFsErrors';
+import { walkHostDirectory } from '#/os/interface/hostFsWalk';
 
 const READ_CHUNK_SIZE = 64 * 1024;
 
@@ -192,6 +199,10 @@ export class HostFileSystem implements IHostFileSystem {
     }
   }
 
+  async walk(path: string, options: HostWalkOptions): Promise<HostWalkResult> {
+    return walkHostDirectory(this, path, options);
+  }
+
   async mkdir(path: string, options?: { readonly recursive?: boolean }): Promise<void> {
     try {
       await mkdir(path, { recursive: options?.recursive ?? false });

packages/agent-core-v2/src/os/backends/remote/remoteFileSystem.ts

@@ -1,3 +1,5 @@
+import * as posixPath from 'node:path/posix';
+
 import {
   decodeTextWithErrors,
   readUtf8Lines,
@@ -7,8 +9,11 @@ import { splitLinesKeepingTerminator } from '#/_base/text/line-endings';
 import type {
   HostDirEntry,
   HostFileStat,
+  HostWalkOptions,
+  HostWalkResult,
   IHostFileSystem,
 } from '#/os/interface/hostFileSystem';
+import { EMPTY_HOST_WALK_RESULT, HOST_WALK_MAX_IGNORE_BYTES, walkHostDirectory } from '#/os/interface/hostFsWalk';
 import {
   HostFsError,
   OsFsErrors,
@@ -28,16 +33,23 @@ import {
   FS_READ_FILE_METHOD,
   FS_REMOVE_METHOD,
   FS_RENAME_METHOD,
+  FS_WALK_METHOD,
   FS_WRITE_FILE_CHUNK_BYTES,
   FS_WRITE_FILE_METHOD,
   type FsGetMetadataResult,
   type FsReadDirectoryResult,
   type FsReadFileResult,
+  type FsWalkResult,
   type FsWriteMode,
 } from '#/remote/protocol/methods';
 import { rpcDomainError } from './rpcDomainError';
 
 const HOST_FS_CODES: ReadonlySet<string> = new Set(Object.values(OsFsErrors.codes));
+const REMOTE_WALK_PATH = {
+  join: (...paths: readonly string[]) => posixPath.join(...paths),
+  relative: (from: string, to: string) => posixPath.relative(from, to),
+  separator: posixPath.sep,
+};
 
 export function toRemoteFsError(error: unknown): Error {
   if (!(error instanceof RpcError)) {
@@ -62,6 +74,7 @@ function isUtf8Encoding(encoding: BufferEncoding): boolean {
 
 export class RemoteFileSystem implements IHostFileSystem {
   declare readonly _serviceBrand: undefined;
+  private walkSupported: boolean | undefined;
 
   constructor(private readonly connection: ExecChannel) {}
 
@@ -232,6 +245,67 @@ export class RemoteFileSystem implements IHostFileSystem {
     }));
   }
 
+  async walk(path: string, options: HostWalkOptions): Promise<HostWalkResult> {
+    if (options.signal?.aborted === true) return EMPTY_HOST_WALK_RESULT;
+    if (this.walkSupported === false) return this.walkFallback(path, options);
+    if (options.ignore !== undefined && Buffer.byteLength(JSON.stringify(options.ignore)) > HOST_WALK_MAX_IGNORE_BYTES) {
+      return this.walkFallback(path, options);
+    }
+    try {
+      const call = this.connection.call(FS_WALK_METHOD, {
+        path,
+        maxDepth: options.maxDepth,
+        maxDirectories: options.maxDirectories,
+        maxEntries: options.maxEntries,
+        maxResponseBytes: options.maxResponseBytes,
+        ignore: options.ignore,
+        timeoutMs: options.timeoutMs,
+      }) as Promise<FsWalkResult>;
+      const result = options.signal === undefined ? await call : await this.raceWalkAbort(call, options.signal);
+      if (result === undefined) return EMPTY_HOST_WALK_RESULT;
+      this.walkSupported = true;
+      return {
+        entries: result.entries,
+        errors: result.errors.map(({ path: errorPath, error }) => ({
+          path: errorPath,
+          error: toRemoteFsError(RpcError.fromErrorBody(error)),
+        })),
+        truncated: result.truncated,
+        responseBytes: result.responseBytes,
+      };
+    } catch (error) {
+      if (error instanceof RpcError && error.code === RpcErrorCode.MethodNotFound) {
+        this.walkSupported = false;
+        return this.walkFallback(path, options);
+      }
+      throw toRemoteFsError(error);
+    }
+  }
+
+  private async walkFallback(path: string, options: HostWalkOptions): Promise<HostWalkResult> {
+    return walkHostDirectory(this, path, options, { path: REMOTE_WALK_PATH });
+  }
+
+  private async raceWalkAbort(
+    call: Promise<FsWalkResult>,
+    signal: AbortSignal,
+  ): Promise<FsWalkResult | undefined> {
+    let onAbort: () => void;
+    const aborted = new Promise<undefined>((resolve) => {
+      onAbort = () => resolve(undefined);
+      signal.addEventListener('abort', onAbort, { once: true });
+    });
+    try {
+      return await Promise.race([call, aborted]);
+    } catch (error) {
+      if (signal.aborted) return undefined;
+      throw error;
+    } finally {
+      signal.removeEventListener('abort', onAbort!);
+      call.catch(() => {});
+    }
+  }
+
   async mkdir(path: string, options?: { readonly recursive?: boolean }): Promise<void> {
     await this.call(FS_CREATE_DIRECTORY_METHOD, {
       path,

packages/agent-core-v2/src/os/interface/hostFileSystem.ts

@@ -18,6 +18,28 @@ export interface HostDirEntry {
   readonly isSymbolicLink?: boolean;
 }
 
+export interface HostWalkOptions {
+  readonly maxDepth: number;
+  readonly maxDirectories: number;
+  readonly maxEntries: number;
+  readonly maxResponseBytes: number;
+  readonly ignore?: readonly string[];
+  readonly timeoutMs?: number;
+  readonly signal?: AbortSignal;
+}
+
+export interface HostWalkError {
+  readonly path: string;
+  readonly error: Error;
+}
+
+export interface HostWalkResult {
+  readonly entries: readonly HostDirEntry[];
+  readonly errors: readonly HostWalkError[];
+  readonly truncated: boolean;
+  readonly responseBytes: number;
+}
+
 export interface IHostFileSystem {
   readonly _serviceBrand: undefined;
 
@@ -37,6 +59,7 @@ export interface IHostFileSystem {
   stat(path: string): Promise<HostFileStat>;
   lstat(path: string): Promise<HostFileStat>;
   readdir(path: string): Promise<readonly HostDirEntry[]>;
+  walk(path: string, options: HostWalkOptions): Promise<HostWalkResult>;
   mkdir(path: string, options?: { readonly recursive?: boolean }): Promise<void>;
   remove(path: string): Promise<void>;
   rename?(from: string, to: string): Promise<void>;

packages/agent-core-v2/src/os/interface/hostFsWalk.ts

@@ -0,0 +1,133 @@
+import { join, relative, sep } from 'node:path';
+
+import ignore from 'ignore';
+
+import type { HostDirEntry, HostWalkOptions, HostWalkResult } from './hostFileSystem';
+import { HostFsError, OsFsErrors } from './hostFsErrors';
+
+export const HOST_WALK_MAX_DEPTH = 64;
+export const HOST_WALK_MAX_DIRECTORIES = 10_000;
+export const HOST_WALK_MAX_ENTRIES = 100_000;
+export const HOST_WALK_MAX_RESPONSE_BYTES = 8 * 1024 * 1024;
+export const HOST_WALK_MAX_IGNORE_BYTES = 1024 * 1024;
+export const HOST_WALK_DEFAULT_TIMEOUT_MS = 55_000;
+export const HOST_WALK_MAX_TIMEOUT_MS = 60_000;
+export const HOST_WALK_MAX_DIRECTORY_ENTRIES = 50_000;
+
+export const EMPTY_HOST_WALK_RESULT: HostWalkResult = {
+  entries: [],
+  errors: [],
+  truncated: true,
+  responseBytes: 0,
+};
+
+interface HostWalkerFs {
+  readdir(path: string): Promise<readonly HostDirEntry[]>;
+}
+
+interface HostWalkPath {
+  readonly separator: string;
+  join(...paths: readonly string[]): string;
+  relative(from: string, to: string): string;
+}
+
+interface HostWalkHooks {
+  readonly path?: HostWalkPath;
+  readonly serializeError?: (path: string, error: Error) => unknown;
+}
+
+export async function walkHostDirectory(
+  fs: HostWalkerFs,
+  root: string,
+  options: HostWalkOptions,
+  hooks: HostWalkHooks = {},
+  readDirectory: (path: string) => Promise<readonly HostDirEntry[]> = fs.readdir.bind(fs),
+): Promise<HostWalkResult> {
+  validateLimit('maxDepth', options.maxDepth, 0, HOST_WALK_MAX_DEPTH);
+  validateLimit('maxDirectories', options.maxDirectories, 1, HOST_WALK_MAX_DIRECTORIES);
+  validateLimit('maxEntries', options.maxEntries, 1, HOST_WALK_MAX_ENTRIES);
+  validateLimit('maxResponseBytes', options.maxResponseBytes, 1, HOST_WALK_MAX_RESPONSE_BYTES);
+  if (options.timeoutMs !== undefined) {
+    validateLimit('timeoutMs', options.timeoutMs, 1, HOST_WALK_MAX_TIMEOUT_MS);
+  }
+  const matcher = options.ignore === undefined ? undefined : ignore().add(options.ignore);
+  const serializeError = hooks.serializeError ?? ((_path, error) => ({
+    name: error.name,
+    message: error.message,
+  }));
+  const pathApi = hooks.path ?? { join, relative, separator: sep };
+  const timeout = AbortSignal.timeout(options.timeoutMs ?? HOST_WALK_DEFAULT_TIMEOUT_MS);
+  const signal = options.signal === undefined
+    ? timeout
+    : AbortSignal.any([options.signal, timeout]);
+  const entries: HostDirEntry[] = [];
+  const errors: { path: string; error: Error }[] = [];
+  const pending = [{ path: root, depth: 0 }];
+  let responseBytes = 64;
+  let truncated = false;
+  for (let index = 0; index < pending.length && !truncated; index += 1) {
+    if (signal?.aborted === true) {
+      truncated = true;
+      break;
+    }
+    const { path: directory, depth } = pending[index]!;
+    let children: readonly HostDirEntry[];
+    try {
+      children = await readDirectory(directory);
+      if (children.length > HOST_WALK_MAX_DIRECTORY_ENTRIES) {
+        throw new HostFsError(
+          OsFsErrors.codes.OS_FS_DIRECTORY_TOO_LARGE,
+          `readdir ${directory} exceeds the ${HOST_WALK_MAX_DIRECTORY_ENTRIES}-entry limit`,
+          { details: { path: directory, op: 'readdir', limit: HOST_WALK_MAX_DIRECTORY_ENTRIES } },
+        );
+      }
+    } catch (error) {
+      const errorBytes = Buffer.byteLength(JSON.stringify({
+        path: directory,
+        error: serializeError(directory, asError(error)),
+      })) + 1;
+      if (responseBytes + errorBytes > options.maxResponseBytes) {
+        truncated = true;
+        break;
+      }
+      responseBytes += errorBytes;
+      errors.push({ path: directory, error: asError(error) });
+      continue;
+    }
+    for (const child of children) {
+      if (signal?.aborted) {
+        truncated = true;
+        break;
+      }
+      const path = pathApi.join(directory, child.name);
+      const isDir = child.isDirectory && child.isSymbolicLink !== true;
+      const relPath = pathApi.relative(root, path).split(pathApi.separator).join('/');
+      if (matcher?.ignores(isDir ? `${relPath}/` : relPath) === true) continue;
+      const entryBytes = Buffer.byteLength(JSON.stringify({ ...child, name: path })) + 1;
+      if (entries.length >= options.maxEntries || responseBytes + entryBytes > options.maxResponseBytes) {
+        truncated = true;
+        break;
+      }
+      responseBytes += entryBytes;
+      entries.push({ ...child, name: path });
+      if (isDir && depth < options.maxDepth) {
+        if (pending.length >= options.maxDirectories) {
+          truncated = true;
+          break;
+        }
+        pending.push({ path, depth: depth + 1 });
+      }
+    }
+  }
+  return { entries, errors, truncated, responseBytes };
+}
+
+function validateLimit(name: string, value: number, min: number, max: number): void {
+  if (!Number.isInteger(value) || value < min || value > max) {
+    throw new RangeError(`${name} must be an integer in [${min}, ${max}]`);
+  }
+}
+
+function asError(error: unknown): Error {
+  return error instanceof Error ? error : new Error(String(error));
+}

packages/agent-core-v2/src/remote/README.md

@@ -67,6 +67,12 @@ deviations per the design spec, plus one forced addition:
 - `fs/readDirectory` preserves the symlink marker on each entry. The client
   rejects a truncated listing with `os.fs.directory_too_large` rather than
   exposing an incomplete directory; the current limit is 50,000 entries.
+- `fs/walk` performs a bounded recursive listing in one call. It accepts the
+  root ignore rules needed to prune ignored directories before traversal and
+  reports entries, recoverable directory errors, and truncation. Clients fall
+  back to per-directory reads when an older executor does not implement it or
+  when the JSON-encoded ignore rules exceed 1MiB. Walks default to a 55-second
+  budget so they can return a truncated result before the 60-second RPC deadline.
 - Added `fs/rename` (no codex counterpart).
 - `fs/remove` takes only `path` and always removes recursively and ignores a
   missing path, matching `IHostFileSystem.remove` (codex has `recursive`/`force`).
@@ -89,7 +95,7 @@ deviations per the design spec, plus one forced addition:
 - `environment` adds `osArch`/`osVersion`/`pathClass` over codex (rg artifact
   selection needs arch), plus `cwd`/`tempDir`.
 - No ws transport, no `resumeSessionId`, no sandbox parameter family, no
-  `fs/walk`/`fs/copy`/`capabilityRoots/discoverV1`/`environmentConfig/read`.
+  `fs/copy`/`capabilityRoots/discoverV1`/`environmentConfig/read`.
 
 Process output is delivered through notifications (`process/output`,
 `process/exited`, `process/closed`). The executor retains process handles until

packages/agent-core-v2/src/remote/client/remoteEnvironmentProvider.ts

@@ -61,6 +61,7 @@ class DeclaredRemoteEnvironment implements Environment {
     stat: async (path) => this.connected().fs.stat(path),
     lstat: async (path) => this.connected().fs.lstat(path),
     readdir: async (path) => this.connected().fs.readdir(path),
+    walk: async (path, options) => this.connected().fs.walk(path, options),
     mkdir: async (path, options) => this.connected().fs.mkdir(path, options),
     remove: async (path) => this.connected().fs.remove(path),
     rename: async (from, to) => this.connected().fs.rename(from, to),