import { createLogger } from '@shared/utils/logger'; import { createReadStream } from 'fs'; import { stat } from 'fs/promises'; import * as readline from 'readline'; import type { TaskBoundariesResult, TaskBoundary, TaskChangeScope, TaskScopeConfidence, } from '@shared/types'; const logger = createLogger('Service:TaskBoundaryParser'); type TaskBoundaryEvent = 'start' | 'complete' | null; /** Файл-модифицирующие инструменты, которые включаем в scope.toolUseIds */ const FILE_MODIFYING_TOOLS = new Set(['Edit', 'Write', 'MultiEdit', 'NotebookEdit']); /** Кеш-запись: данные + mtime файла + время протухания */ interface BoundaryCacheEntry { data: TaskBoundariesResult; mtime: number; expiresAt: number; } /** Информация о tool_use блоке собранная при парсинге */ interface ToolUseInfo { toolUseId: string; toolName: string; filePath?: string; } /** Regex для teamctl task команд */ const TEAMCTL_TASK_REGEX = /task\s+(start|complete|set-status)\s+(\d+)/; const MCP_TASK_BOUNDARY_TOOLS = new Set(['task_start', 'task_complete', 'task_set_status']); export class TaskBoundaryParser { private cache = new Map(); private readonly cacheTtl = 60 * 1000; // 60s /** Парсинг JSONL файла для обнаружения границ задач */ async parseBoundaries(filePath: string): Promise { // 1. Проверяем кеш (TTL + mtime) let fileStat; try { fileStat = await stat(filePath); } catch (err) { logger.debug(`Cannot stat file ${filePath}: ${String(err)}`); return { boundaries: [], scopes: [], isSingleTaskSession: true, detectedMechanism: 'none' }; } const cached = this.cache.get(filePath); if (cached?.mtime === fileStat.mtimeMs && cached.expiresAt > Date.now()) { return cached.data; } // 2. Стриминг JSONL const boundaries: TaskBoundary[] = []; const allToolUsesByLine = new Map(); let lineNumber = 0; let detectedMechanism: 'TaskUpdate' | 'teamctl' | 'mcp' | 'none' = 'none'; try { const stream = createReadStream(filePath, { encoding: 'utf8' }); const rl = readline.createInterface({ input: stream, crlfDelay: Infinity }); for await (const line of rl) { lineNumber++; const trimmed = line.trim(); if (!trimmed) continue; try { const entry = JSON.parse(trimmed) as Record; const timestamp = typeof entry.timestamp === 'string' ? entry.timestamp : ''; const content = this.extractContent(entry); if (!Array.isArray(content)) continue; // Собираем ВСЕ tool_use блоки для scope tracking for (const block of content) { if (!block || typeof block !== 'object') continue; const b = block as Record; if (b.type !== 'tool_use') continue; const rawName = typeof b.name === 'string' ? b.name : ''; const toolName = rawName.replace(/^proxy_/, ''); const toolUseId = typeof b.id === 'string' ? b.id : ''; const input = b.input as Record | undefined; const fp = typeof input?.file_path === 'string' ? input.file_path : undefined; if (!allToolUsesByLine.has(lineNumber)) allToolUsesByLine.set(lineNumber, []); allToolUsesByLine.get(lineNumber)!.push({ toolUseId, toolName, filePath: fp }); } // Пробуем TaskUpdate const taskUpdateBounds = this.extractTaskUpdateBoundaries(content, lineNumber, timestamp); if (taskUpdateBounds.length > 0) { detectedMechanism = 'TaskUpdate'; boundaries.push(...taskUpdateBounds); continue; } const mcpBounds = this.extractMcpTaskBoundaries(content, lineNumber, timestamp); if (mcpBounds.length > 0) { detectedMechanism = 'mcp'; boundaries.push(...mcpBounds); continue; } // Пробуем teamctl const teamctlBounds = this.extractTeamctlBoundaries(content, lineNumber, timestamp); if (teamctlBounds.length > 0) { detectedMechanism = 'teamctl'; boundaries.push(...teamctlBounds); } } catch { // Пропускаем невалидные строки } } rl.close(); stream.destroy(); } catch (err) { logger.debug(`Error reading file ${filePath}: ${String(err)}`); } // 3. Вычисляем scopes const scopes = this.computeScopes(boundaries, allToolUsesByLine, lineNumber); const uniqueTaskIds = new Set(boundaries.map((b) => b.taskId)); const isSingleTaskSession = uniqueTaskIds.size <= 1; const result: TaskBoundariesResult = { boundaries, scopes, isSingleTaskSession, detectedMechanism, }; this.cache.set(filePath, { data: result, mtime: fileStat.mtimeMs, expiresAt: Date.now() + this.cacheTtl, }); return result; } /** Получить scope для конкретной задачи */ async getTaskScope(filePath: string, taskId: string): Promise { const result = await this.parseBoundaries(filePath); return result.scopes.find((s) => s.taskId === taskId) ?? null; } /** Очистить кеш (для тестов) */ clearCache(): void { this.cache.clear(); } // ── Приватные методы ── /** Извлечь content array из JSONL entry (оба формата: subagent и main) */ private extractContent(entry: Record): unknown[] | null { const message = entry.message as Record | undefined; if (message && Array.isArray(message.content)) return message.content as unknown[]; if (Array.isArray(entry.content)) return entry.content as unknown[]; return null; } /** * Найти TaskUpdate/proxy_TaskUpdate tool_use блоки. * status: in_progress → start, completed → complete */ private extractTaskUpdateBoundaries( content: unknown[], lineNumber: number, timestamp: string ): TaskBoundary[] { const results: TaskBoundary[] = []; for (const block of content) { if (!block || typeof block !== 'object') continue; const b = block as Record; if (b.type !== 'tool_use') continue; const rawName = typeof b.name === 'string' ? b.name : ''; const toolName = rawName.replace(/^proxy_/, ''); if (toolName !== 'TaskUpdate') continue; const input = b.input as Record | undefined; if (!input) continue; const rawTaskId = input.taskId; const taskId = typeof rawTaskId === 'string' ? rawTaskId : typeof rawTaskId === 'number' ? String(rawTaskId) : ''; if (!taskId) continue; const status = typeof input.status === 'string' ? input.status : ''; let event: TaskBoundaryEvent = null; if (status === 'in_progress') event = 'start'; else if (status === 'completed') event = 'complete'; if (event) { const toolUseId = typeof b.id === 'string' ? b.id : undefined; results.push({ taskId, event, lineNumber, timestamp, mechanism: 'TaskUpdate', toolUseId, }); } } return results; } /** * Find MCP task tools that mark task boundaries. */ private extractMcpTaskBoundaries( content: unknown[], lineNumber: number, timestamp: string ): TaskBoundary[] { const results: TaskBoundary[] = []; for (const block of content) { if (!block || typeof block !== 'object') continue; const b = block as Record; if (b.type !== 'tool_use') continue; const rawName = typeof b.name === 'string' ? b.name : ''; const toolName = rawName.replace(/^proxy_/, ''); if (!MCP_TASK_BOUNDARY_TOOLS.has(toolName)) continue; const input = b.input as Record | undefined; if (!input) continue; const rawTaskId = input.taskId; const taskId = typeof rawTaskId === 'string' ? rawTaskId : typeof rawTaskId === 'number' ? String(rawTaskId) : ''; if (!taskId) continue; let event: TaskBoundaryEvent = null; if (toolName === 'task_start') event = 'start'; else if (toolName === 'task_complete') event = 'complete'; else { const status = typeof input.status === 'string' ? input.status : ''; if (status === 'in_progress') event = 'start'; else if (status === 'completed') event = 'complete'; } if (event) { const toolUseId = typeof b.id === 'string' ? b.id : undefined; results.push({ taskId, event, lineNumber, timestamp, mechanism: 'mcp', toolUseId, }); } } return results; } /** * Найти teamctl task start/complete/set-status команды в Bash tool_use блоках. * Regex: /task\s+(start|complete|set-status)\s+(\d+)/ */ private extractTeamctlBoundaries( content: unknown[], lineNumber: number, timestamp: string ): TaskBoundary[] { const results: TaskBoundary[] = []; for (const block of content) { if (!block || typeof block !== 'object') continue; const b = block as Record; if (b.type !== 'tool_use') continue; const rawName = typeof b.name === 'string' ? b.name : ''; const toolName = rawName.replace(/^proxy_/, ''); if (toolName !== 'Bash') continue; const input = b.input as Record | undefined; if (!input) continue; const command = typeof input.command === 'string' ? input.command : ''; if (!command.includes('teamctl')) continue; const match = TEAMCTL_TASK_REGEX.exec(command); if (!match) continue; const action = match[1]; // start | complete | set-status const taskId = match[2]; let event: TaskBoundaryEvent = null; if (action === 'start') event = 'start'; else if (action === 'complete') event = 'complete'; else if (action === 'set-status') { // set-status может быть start или complete — определяем по аргументам if (command.includes('in_progress') || command.includes('in-progress')) event = 'start'; else if (command.includes('completed') || command.includes('done')) event = 'complete'; } if (event) { const toolUseId = typeof b.id === 'string' ? b.id : undefined; results.push({ taskId, event, lineNumber, timestamp, mechanism: 'teamctl', toolUseId, }); } } return results; } /** * Вычислить scopes для каждой задачи на основе границ. * * Tier 1 (high): обе границы (start + complete) * Tier 2 (medium): только start (end = конец файла) * Tier 3 (low): только complete (start = начало файла) * Tier 4 (fallback): нет границ (весь файл) */ private computeScopes( boundaries: TaskBoundary[], allToolUsesByLine: Map, totalLines: number ): TaskChangeScope[] { // Группируем по taskId const byTask = new Map(); for (const b of boundaries) { if (!byTask.has(b.taskId)) byTask.set(b.taskId, []); byTask.get(b.taskId)!.push(b); } const scopes: TaskChangeScope[] = []; for (const [taskId, taskBoundaries] of byTask) { const starts = taskBoundaries.filter((b) => b.event === 'start'); const completes = taskBoundaries.filter((b) => b.event === 'complete'); const hasStart = starts.length > 0; const hasComplete = completes.length > 0; // Определяем границы строк let startLine: number; let endLine: number; let startTimestamp: string; let endTimestamp: string; let confidence: TaskScopeConfidence; if (hasStart && hasComplete) { // Tier 1: обе границы const firstStart = starts.reduce( (a, b) => (a.lineNumber < b.lineNumber ? a : b), starts[0] ); const lastComplete = completes.reduce( (a, b) => (a.lineNumber > b.lineNumber ? a : b), completes[0] ); startLine = firstStart.lineNumber; endLine = lastComplete.lineNumber; startTimestamp = firstStart.timestamp; endTimestamp = lastComplete.timestamp; confidence = { tier: 1, label: 'high', reason: 'Both start and complete markers found' }; } else if (hasStart) { // Tier 2: только start const firstStart = starts.reduce( (a, b) => (a.lineNumber < b.lineNumber ? a : b), starts[0] ); startLine = firstStart.lineNumber; endLine = totalLines; startTimestamp = firstStart.timestamp; endTimestamp = ''; confidence = { tier: 2, label: 'medium', reason: 'Only start marker found, end assumed at file end', }; } else { // Tier 3: только complete const lastComplete = completes.reduce( (a, b) => (a.lineNumber > b.lineNumber ? a : b), completes[0] ); startLine = 1; endLine = lastComplete.lineNumber; startTimestamp = ''; endTimestamp = lastComplete.timestamp; confidence = { tier: 3, label: 'low', reason: 'Only complete marker found, start assumed at file beginning', }; } // Собираем tool_use IDs в диапазоне [startLine, endLine], только файл-модифицирующие const toolUseIds: string[] = []; const filePaths = new Set(); for (const [line, tools] of allToolUsesByLine) { if (line < startLine || line > endLine) continue; for (const tool of tools) { if (FILE_MODIFYING_TOOLS.has(tool.toolName) && tool.toolUseId) { toolUseIds.push(tool.toolUseId); if (tool.filePath) filePaths.add(tool.filePath); } } } scopes.push({ taskId, memberName: '', // будет заполнен вызывающим кодом startLine, endLine, startTimestamp, endTimestamp, toolUseIds, filePaths: [...filePaths], confidence, }); } return scopes; } }