异步grep

Author: linyuan.yang

packages/sbot/src/Tools/FileSystem/content/grep.ts

@@ -1,4 +1,5 @@
 import fs from 'fs';
+import fsAsync, { type FileHandle } from 'fs/promises';
 import path from 'path';
 import { spawn } from 'child_process';
 import { DynamicStructuredTool, type StructuredToolInterface } from '@langchain/core/tools';
@@ -13,13 +14,17 @@ const logger = LoggerService.getLogger('Tools/FileSystem/content/grep.ts');
 const MAX_LINE_LENGTH = 2000;
 const DEFAULT_MAX_MATCHES = 100;
 const MAX_FILE_SIZE = 1 * 1024 * 1024; // 1MB - skip large files in Node.js fallback
+const MAX_COLUMNS = 4096;
 
 // ─── 类型 ─────────────────────────────────────────────────────────────────────
 interface MatchLine { lineNum: number; text: string; }
 interface FileMatches { filePath: string; mtime: number; matches: MatchLine[]; }
 interface SearchResult { results: FileMatches[]; reachedLimit: boolean; }
 
-// ─── ripgrep 搜索（--json 模式）──────────────────────────────────────────────
+// rg 默认排除目录（gitignore 语法，无 / 时匹配任意层级）
+const RG_EXCLUDE_ARGS = [...EXCLUDE_DIRS].map(d => `--glob=!${d}`);
+
+// ─── ripgrep 搜索（--json 流式 + 达到上限即终止）────────────────────────────
 function searchWithRg(
     dir: string,
     pattern: string,
@@ -29,114 +34,135 @@ function searchWithRg(
     maxMatches: number,
 ): Promise<SearchResult> {
     return new Promise((resolve, reject) => {
-        const args = ['--json', '--glob=!.git/*'];
+        const args = ['--json', `--max-columns=${MAX_COLUMNS}`, ...RG_EXCLUDE_ARGS];
         if (!useRegex) args.push('--fixed-strings');
         if (includeHidden) args.push('--hidden');
         if (fileGlob) args.push(`--iglob=${fileGlob}`);
         args.push('--', pattern, dir);
 
         const proc = spawn('rg', args);
-        let stdout = '';
+        const byFile = new Map<string, FileMatches>();
+        const fileOrder: string[] = [];
+        let totalMatches = 0;
+        let reachedLimit = false;
+        let killed = false;
+        let buffer = '';
         let stderr = '';
-        proc.stdout.on('data', (d: Buffer) => { stdout += d.toString(); });
+
+        const stop = (limit: boolean) => {
+            if (killed) return;
+            killed = true;
+            if (limit) reachedLimit = true;
+            try { proc.kill('SIGTERM'); } catch { /* ignore */ }
+        };
+
+        const processLine = (line: string): boolean => {
+            if (!line) return true;
+            let parsed: any;
+            try { parsed = JSON.parse(line); } catch { return true; }
+            if (parsed.type !== 'match') return true;
+
+            const fp: string = parsed.data.path.text;
+            const lineNum: number = parsed.data.line_number;
+            const raw: string = (parsed.data.lines?.text ?? '').replace(/\r?\n$/, '');
+            const text = raw.length > MAX_LINE_LENGTH ? raw.slice(0, MAX_LINE_LENGTH) + '...' : raw;
+
+            if (!byFile.has(fp)) {
+                let mtime = 0;
+                try { mtime = fs.statSync(fp).mtimeMs; } catch { /* */ }
+                byFile.set(fp, { filePath: fp, mtime, matches: [] });
+                fileOrder.push(fp);
+            }
+            byFile.get(fp)!.matches.push({ lineNum, text });
+            totalMatches++;
+            return totalMatches < maxMatches;
+        };
+
+        proc.stdout.on('data', (d: Buffer) => {
+            if (killed) return;
+            buffer += d.toString();
+            let idx;
+            while ((idx = buffer.indexOf('\n')) !== -1) {
+                const line = buffer.slice(0, idx).replace(/\r$/, '');
+                buffer = buffer.slice(idx + 1);
+                if (!processLine(line)) { stop(true); return; }
+            }
+            // 防御：单行过长时丢弃
+            if (buffer.length > 10 * MAX_COLUMNS) buffer = '';
+        });
         proc.stderr.on('data', (d: Buffer) => { stderr += d.toString(); });
-        proc.on('error', reject);
+        proc.on('error', e => { reject(e); });
         proc.on('close', code => {
-            if (code === 1) return resolve({ results: [], reachedLimit: false });
-            if (code === 2 && !stdout.trim()) return reject(new Error(`ripgrep: ${stderr.trim()}`));
-
-            const byFile = new Map<string, FileMatches>();
-            const fileOrder: string[] = [];
-            let totalMatches = 0;
-            let reachedLimit = false;
-
-            for (const line of stdout.trim().split(/\r?\n/)) {
-                if (!line) continue;
-                let parsed: any;
-                try { parsed = JSON.parse(line); } catch { continue; }
-                if (parsed.type !== 'match') continue;
-                if (totalMatches >= maxMatches) { reachedLimit = true; break; }
-
-                const fp: string = parsed.data.path.text;
-                const lineNum: number = parsed.data.line_number;
-                const raw: string = parsed.data.lines.text.replace(/\r?\n$/, '');
-                const text = raw.length > MAX_LINE_LENGTH ? raw.slice(0, MAX_LINE_LENGTH) + '...' : raw;
-
-                if (!byFile.has(fp)) {
-                    let mtime = 0;
-                    try { mtime = fs.statSync(fp).mtimeMs; } catch { /* */ }
-                    byFile.set(fp, { filePath: fp, mtime, matches: [] });
-                    fileOrder.push(fp);
-                }
-                byFile.get(fp)!.matches.push({ lineNum, text });
-                totalMatches++;
+            if (!killed && buffer) processLine(buffer.replace(/\r$/, ''));
+            // code 1 = 无匹配；被 kill 时 code/signal 不可靠，按 totalMatches 判断
+            if (!killed && code !== 0 && code !== 1 && totalMatches === 0) {
+                return reject(new Error(`ripgrep: ${stderr.trim() || `exit ${code}`}`));
             }
-
             const results = fileOrder.map(fp => byFile.get(fp)!);
             results.sort((a, b) => b.mtime - a.mtime);
             resolve({ results, reachedLimit });
         });
     });
 }
 
-// ─── Node.js fallback ─────────────────────────────────────────────────────────
-function isBinary(fp: string): boolean {
+// ─── Node.js fallback（全异步，避免阻塞事件循环）────────────────────────────
+async function isBinaryAsync(fp: string): Promise<boolean> {
+    let fh: FileHandle | undefined;
     try {
+        fh = await fsAsync.open(fp, 'r');
         const buf = Buffer.alloc(512);
-        const fd = fs.openSync(fp, 'r');
-        const read = fs.readSync(fd, buf, 0, 512, 0);
-        fs.closeSync(fd);
-        return buf.subarray(0, read).includes(0);
+        const { bytesRead } = await fh.read(buf, 0, 512, 0);
+        return buf.subarray(0, bytesRead).includes(0);
     } catch { return true; }
+    finally { try { await fh?.close(); } catch { /* ignore */ } }
 }
 
-function searchWithNodeJs(
+async function searchWithNodeJs(
     dir: string,
     pattern: string,
     fileRegex: RegExp,
     useRegex: boolean,
     includeHidden: boolean,
     maxFileSize: number,
     maxMatches: number,
-): SearchResult {
+): Promise<SearchResult> {
     const searchRegex = useRegex ? new RegExp(pattern) : null;
-    const searchLower = pattern;
 
     const allFiles: Array<{ path: string; mtime: number }> = [];
-    function walk(d: string) {
-        try {
-            for (const entry of fs.readdirSync(d, { withFileTypes: true })) {
-                if (!includeHidden && entry.name.startsWith('.')) continue;
-                const full = path.join(d, entry.name);
-                if (entry.isDirectory()) {
-                    if (EXCLUDE_DIRS.has(entry.name)) continue;
-                    walk(full);
-                } else if (entry.isFile() && fileRegex.test(entry.name)) {
-                    try {
-                        const stat = fs.statSync(full);
-                        if (stat.size <= maxFileSize) allFiles.push({ path: full, mtime: stat.mtimeMs });
-                    } catch { /* skip */ }
-                }
+    async function walk(d: string): Promise<void> {
+        let entries;
+        try { entries = await fsAsync.readdir(d, { withFileTypes: true }); }
+        catch (e: any) { logger.warn(`Cannot access ${d}: ${e.message}`); return; }
+        for (const entry of entries) {
+            if (!includeHidden && entry.name.startsWith('.')) continue;
+            const full = path.join(d, entry.name);
+            if (entry.isDirectory()) {
+                if (EXCLUDE_DIRS.has(entry.name)) continue;
+                await walk(full);
+            } else if (entry.isFile() && fileRegex.test(entry.name)) {
+                try {
+                    const stat = await fsAsync.stat(full);
+                    if (stat.size <= maxFileSize) allFiles.push({ path: full, mtime: stat.mtimeMs });
+                } catch { /* skip */ }
             }
-        } catch (e: any) { logger.warn(`Cannot access ${d}: ${e.message}`); }
+        }
     }
-    walk(dir);
+    await walk(dir);
     allFiles.sort((a, b) => b.mtime - a.mtime);
 
     const results: FileMatches[] = [];
     let totalMatches = 0;
     let reachedLimit = false;
 
     outer: for (const { path: fp, mtime } of allFiles) {
-        if (isBinary(fp)) continue;
+        if (await isBinaryAsync(fp)) continue;
         try {
             const fileMatches: MatchLine[] = [];
-            const lines = fs.readFileSync(fp, 'utf8').split('\n');
+            const content = await fsAsync.readFile(fp, 'utf8');
+            const lines = content.split('\n');
             for (let i = 0; i < lines.length; i++) {
                 const line = lines[i];
-                const hit = searchRegex
-                    ? searchRegex.test(line)
-                    : line.includes(searchLower);
+                const hit = searchRegex ? searchRegex.test(line) : line.includes(pattern);
                 if (!hit) continue;
                 const text = line.length > MAX_LINE_LENGTH ? line.slice(0, MAX_LINE_LENGTH) + '...' : line;
                 fileMatches.push({ lineNum: i + 1, text });
@@ -195,7 +221,7 @@ export function createGrepFilesTool(): StructuredToolInterface {
                     result = await searchWithRg(abs, pattern, useRegex, includeHidden, glob, maxMatches);
                 } else {
                     const fileRegex = globToRegex(glob ?? '*');
-                    result = searchWithNodeJs(abs, pattern, fileRegex, useRegex, includeHidden, MAX_FILE_SIZE, maxMatches);
+                    result = await searchWithNodeJs(abs, pattern, fileRegex, useRegex, includeHidden, MAX_FILE_SIZE, maxMatches);
                 }
 
                 if (result.results.length === 0) return createSuccessResult(createTextContent('No matches found'));