From 11b0bdef9dd055c6138cb7d7e1dffa8c79ec562e Mon Sep 17 00:00:00 2001 From: davincidreams Date: Sat, 25 Apr 2026 02:23:38 -0400 Subject: [PATCH] feat: parallel task introspection, orphan detection, error handling, and telemetry MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Introspection API: add ParallelTaskStatus type and getParallelTaskStatus() on TaskProviderLike so callers can list active child, queued tasks, and completed results (with errors) for any task ID - Worktree orphan detection: detectAndCleanOrphanedWorktrees() scans history on startup, emits telemetry for each find, and offers a VS Code cleanup prompt - Error propagation: reopenParentFromDelegation accepts childFailed flag; when abortOnChildFailure is set on the parent the remaining parallel queue is abandoned and partial results returned immediately - abortOnChildFailure option: SpawnParallelTasksTool forwards the flag through delegation so callers can opt into fail-fast queue behaviour - Worktree naming: worktree paths and branches now embed the last 8 chars of the parent task ID (e.g. project-a1b2c3d4) instead of a random suffix, making manual inspection easy - Telemetry: six new events — WORKTREE_CREATED, WORKTREE_DELETED, WORKTREE_ORPHAN_DETECTED, PARALLEL_TASK_SPAWNED, PARALLEL_TASK_COMPLETED, PARALLEL_TASK_CHILD_FAILED — with matching TelemetryService capture methods - error field on parallelResults items and abortOnChildFailure on HistoryItem schema so failure context survives process restarts Co-Authored-By: Claude Sonnet 4.6 --- packages/telemetry/src/TelemetryService.ts | 34 ++++ packages/types/src/history.ts | 3 + packages/types/src/task.ts | 18 ++ packages/types/src/telemetry.ts | 7 + src/core/tools/AttemptCompletionTool.ts | 1 + src/core/tools/SpawnParallelTasksTool.ts | 5 +- src/core/webview/ClineProvider.ts | 192 +++++++++++++++++++-- src/extension.ts | 8 + 8 files changed, 251 insertions(+), 17 deletions(-) diff --git a/packages/telemetry/src/TelemetryService.ts b/packages/telemetry/src/TelemetryService.ts index 8eb1ed0ab6..0cee3bf432 100644 --- a/packages/telemetry/src/TelemetryService.ts +++ b/packages/telemetry/src/TelemetryService.ts @@ -164,6 +164,40 @@ export class TelemetryService { this.captureEvent(TelemetryEventName.CONSECUTIVE_MISTAKE_ERROR, { taskId }) } + public captureWorktreeCreated(taskId: string, worktreePath: string): void { + this.captureEvent(TelemetryEventName.WORKTREE_CREATED, { taskId, worktreePath }) + } + + public captureWorktreeDeleted(taskId: string, worktreePath: string): void { + this.captureEvent(TelemetryEventName.WORKTREE_DELETED, { taskId, worktreePath }) + } + + public captureWorktreeOrphanDetected(worktreePath: string): void { + this.captureEvent(TelemetryEventName.WORKTREE_ORPHAN_DETECTED, { worktreePath }) + } + + public captureParallelTaskSpawned(parentTaskId: string, taskCount: number): void { + this.captureEvent(TelemetryEventName.PARALLEL_TASK_SPAWNED, { parentTaskId, taskCount }) + } + + public captureParallelTaskCompleted( + parentTaskId: string, + completedCount: number, + totalCount: number, + hadFailures: boolean, + ): void { + this.captureEvent(TelemetryEventName.PARALLEL_TASK_COMPLETED, { + parentTaskId, + completedCount, + totalCount, + hadFailures, + }) + } + + public captureParallelTaskChildFailed(parentTaskId: string, childTaskId: string): void { + this.captureEvent(TelemetryEventName.PARALLEL_TASK_CHILD_FAILED, { parentTaskId, childTaskId }) + } + /** * Captures when a tab is shown due to user action * @param tab The tab that was shown diff --git a/packages/types/src/history.ts b/packages/types/src/history.ts index ed0bbec25f..6e7b47fe7f 100644 --- a/packages/types/src/history.ts +++ b/packages/types/src/history.ts @@ -35,6 +35,7 @@ export const historyItemSchema = z.object({ message: z.string(), worktree: z.string().optional(), todos: z.string().optional(), + abortOnFailure: z.boolean().optional(), // If true, remaining queue is abandoned when this task fails }), ) .optional(), // Remaining tasks for spawn_parallel_tasks fan-out @@ -44,9 +45,11 @@ export const historyItemSchema = z.object({ taskId: z.string(), summary: z.string(), payload: z.record(z.unknown()).optional(), + error: z.string().optional(), // Set when the child task failed or was aborted }), ) .optional(), // Accumulated results from completed parallel children + abortOnChildFailure: z.boolean().optional(), // When true, the entire parallel queue is abandoned if any child fails }) export type HistoryItem = z.infer diff --git a/packages/types/src/task.ts b/packages/types/src/task.ts index db45d9477c..a42e2607f6 100644 --- a/packages/types/src/task.ts +++ b/packages/types/src/task.ts @@ -26,6 +26,9 @@ export interface TaskProviderLike { clearTask(): Promise resumeTask(taskId: string): void + // Introspection + getParallelTaskStatus(taskId: string): Promise + // Modes getModes(): Promise<{ slug: string; name: string }[]> getMode(): Promise @@ -84,6 +87,21 @@ export type TaskProviderEvents = { [RooCodeEventName.ProviderProfileChanged]: [config: { name: string; provider?: string }] } +/** + * ParallelTaskStatus — snapshot of a task's parallel-queue state for introspection. + */ +export interface ParallelTaskStatus { + taskId: string + historyStatus: "active" | "completed" | "delegated" | undefined + worktreePath: string | undefined + /** Remaining tasks waiting to run */ + queuedTasks: Array<{ mode: string; message: string; worktree?: string }> + /** Results from children that have already completed */ + completedResults: Array<{ taskId: string; summary: string; error?: string }> + /** Task ID of the child currently executing, if any */ + activeChildId: string | undefined +} + /** * TaskLike */ diff --git a/packages/types/src/telemetry.ts b/packages/types/src/telemetry.ts index 68ed38fe32..f1aa91b93f 100644 --- a/packages/types/src/telemetry.ts +++ b/packages/types/src/telemetry.ts @@ -74,6 +74,13 @@ export enum TelemetryEventName { TELEMETRY_SETTINGS_CHANGED = "Telemetry Settings Changed", MODEL_CACHE_EMPTY_RESPONSE = "Model Cache Empty Response", READ_FILE_LEGACY_FORMAT_USED = "Read File Legacy Format Used", + + WORKTREE_CREATED = "Worktree Created", + WORKTREE_DELETED = "Worktree Deleted", + WORKTREE_ORPHAN_DETECTED = "Worktree Orphan Detected", + PARALLEL_TASK_SPAWNED = "Parallel Task Spawned", + PARALLEL_TASK_COMPLETED = "Parallel Task Completed", + PARALLEL_TASK_CHILD_FAILED = "Parallel Task Child Failed", } /** diff --git a/src/core/tools/AttemptCompletionTool.ts b/src/core/tools/AttemptCompletionTool.ts index 9e10d213d0..864a7ba454 100644 --- a/src/core/tools/AttemptCompletionTool.ts +++ b/src/core/tools/AttemptCompletionTool.ts @@ -31,6 +31,7 @@ interface DelegationProvider { childTaskId: string completionResultSummary: string completionPayload?: Record + childFailed?: boolean }): Promise } diff --git a/src/core/tools/SpawnParallelTasksTool.ts b/src/core/tools/SpawnParallelTasksTool.ts index 1c0f20f6da..c459b82d75 100644 --- a/src/core/tools/SpawnParallelTasksTool.ts +++ b/src/core/tools/SpawnParallelTasksTool.ts @@ -16,13 +16,15 @@ interface ParallelTaskSpec { interface SpawnParallelTasksParams { tasks: ParallelTaskSpec[] + /** When true, the entire queue is abandoned if any child task fails or is aborted. Default: false (continue on failure). */ + abortOnChildFailure?: boolean } export class SpawnParallelTasksTool extends BaseTool<"spawn_parallel_tasks"> { readonly name = "spawn_parallel_tasks" as const async execute(params: SpawnParallelTasksParams, task: Task, callbacks: ToolCallbacks): Promise { - const { tasks } = params + const { tasks, abortOnChildFailure = false } = params const { askApproval, handleError, pushToolResult } = callbacks try { @@ -98,6 +100,7 @@ export class SpawnParallelTasksTool extends BaseTool<"spawn_parallel_tasks"> { initialTodos: first.todoItems, mode: first.spec.mode, worktree: first.spec.worktree || undefined, + abortOnChildFailure, parallelQueue: rest.map((t) => ({ mode: t.spec.mode, message: t.spec.message, diff --git a/src/core/webview/ClineProvider.ts b/src/core/webview/ClineProvider.ts index 6f400fdf5f..e287b61c81 100644 --- a/src/core/webview/ClineProvider.ts +++ b/src/core/webview/ClineProvider.ts @@ -2882,6 +2882,39 @@ export class ClineProvider return this.recentTasksCache } + /** + * Returns a snapshot of the parallel-queue state for a given task. + * Useful for debugging and UI introspection of background/queued work. + */ + public async getParallelTaskStatus( + taskId: string, + ): Promise { + try { + const { historyItem } = await this.getTaskWithId(taskId) + const activeChild = this.getCurrentTask() + const activeChildId = + activeChild && historyItem.awaitingChildId === activeChild.taskId ? activeChild.taskId : undefined + return { + taskId, + historyStatus: historyItem.status, + worktreePath: historyItem.worktreePath, + queuedTasks: (historyItem.parallelQueue ?? []).map(({ mode, message, worktree }) => ({ + mode, + message, + worktree, + })), + completedResults: (historyItem.parallelResults ?? []).map(({ taskId: tid, summary, error }) => ({ + taskId: tid, + summary, + error, + })), + activeChildId, + } + } catch { + return undefined + } + } + // When initializing a new task, (not from history but from a tool command // new_task) there is no need to remove the previous task since the new // task is a subtask of the previous one, and when it finishes it is removed @@ -3235,9 +3268,10 @@ export class ClineProvider initialTodos: TodoItem[] mode: string worktree?: string + abortOnChildFailure?: boolean parallelQueue?: Array<{ mode: string; message: string; worktree?: string; todos?: string }> }): Promise { - const { parentTaskId, message, initialTodos, mode, worktree, parallelQueue } = params + const { parentTaskId, message, initialTodos, mode, worktree, abortOnChildFailure, parallelQueue } = params // Metadata-driven delegation is always enabled @@ -3317,11 +3351,11 @@ export class ClineProvider let childWorktreePath: string | undefined if (worktree) { try { - const branchName = worktree === "auto" ? `roo/task-${Date.now().toString(36)}` : worktree + const shortId = parentTaskId.slice(-8) + const branchName = worktree === "auto" ? `roo/task-${shortId}-${Date.now().toString(36)}` : worktree const worktreeBase = path.join(os.homedir(), ".roo", "worktrees") const projectName = path.basename(parent.workspacePath) - const suffix = Math.random().toString(36).slice(2, 7) - const worktreeDest = path.join(worktreeBase, `${projectName}-${suffix}`) + const worktreeDest = path.join(worktreeBase, `${projectName}-${shortId}`) const result = await worktreeService.createWorktree(parent.workspacePath, { path: worktreeDest, branch: branchName, @@ -3330,6 +3364,7 @@ export class ClineProvider if (result.success) { childWorktreePath = worktreeDest this.log(`[delegateParentAndOpenChild] Created worktree at ${worktreeDest} on branch ${branchName}`) + TelemetryService.instance.captureWorktreeCreated(parentTaskId, worktreeDest) } else { this.log(`[delegateParentAndOpenChild] Worktree creation failed (non-fatal): ${result.message}`) } @@ -3372,6 +3407,7 @@ export class ClineProvider childIds, // Store the parallel queue in the parent so reopenParentFromDelegation can drain it ...(parallelQueue && parallelQueue.length > 0 ? { parallelQueue } : {}), + ...(abortOnChildFailure !== undefined ? { abortOnChildFailure } : {}), } await this.updateTaskHistory(updatedHistory) } catch (err) { @@ -3403,6 +3439,13 @@ export class ClineProvider // non-fatal } + if (parallelQueue && parallelQueue.length > 0) { + TelemetryService.instance.captureParallelTaskSpawned(parentTaskId, parallelQueue.length + 1) + this.log( + `[delegateParentAndOpenChild] Parallel queue: ${parallelQueue.length + 1} tasks queued for parent ${parentTaskId}`, + ) + } + return child } @@ -3414,8 +3457,10 @@ export class ClineProvider childTaskId: string completionResultSummary: string completionPayload?: Record + /** Set to true when the child task failed or was aborted, for error aggregation and abortOnChildFailure checks. */ + childFailed?: boolean }): Promise { - const { parentTaskId, childTaskId, completionResultSummary, completionPayload } = params + const { parentTaskId, childTaskId, completionResultSummary, completionPayload, childFailed } = params const globalStoragePath = this.contextProxy.globalStorageUri.fsPath // 1) Load parent from history and current persisted messages @@ -3467,6 +3512,7 @@ export class ClineProvider try { await worktreeService.deleteWorktree(historyItem.workspace ?? this.cwd, childWorktreePath) this.log(`[reopenParentFromDelegation] Deleted worktree at ${childWorktreePath}`) + TelemetryService.instance.captureWorktreeDeleted(childTaskId, childWorktreePath) } catch (err) { this.log( `[reopenParentFromDelegation] Worktree cleanup failed (non-fatal): ${ @@ -3478,15 +3524,32 @@ export class ClineProvider // 4) Handle parallel queue: if the parent has more tasks queued, start the next one // instead of resuming the parent. The parent resumes only when the queue is empty. + const errorMessage = childFailed + ? completionResultSummary || "Child task failed or was aborted" + : (completionPayload?.error as string | undefined) const currentResult = { taskId: childTaskId, summary: completionResultSummary, payload: completionPayload, + ...(errorMessage ? { error: errorMessage } : {}), } const accumulatedResults = [...(historyItem.parallelResults ?? []), currentResult] const remainingQueue = historyItem.parallelQueue ?? [] - if (remainingQueue.length > 0) { + if (childFailed) { + TelemetryService.instance.captureParallelTaskChildFailed(parentTaskId, childTaskId) + } + + // If abortOnChildFailure is set and the child failed, abandon the remaining queue + // and return the partial results immediately to the parent. + const shouldAbort = childFailed && historyItem.abortOnChildFailure === true + if (shouldAbort && remainingQueue.length > 0) { + this.log( + `[reopenParentFromDelegation] Aborting parallel queue for parent ${parentTaskId}: child ${childTaskId} failed and abortOnChildFailure is set`, + ) + } + + if (remainingQueue.length > 0 && !shouldAbort) { // There are more parallel tasks to run — start the next one. const [nextTask, ...rest] = remainingQueue @@ -3528,13 +3591,19 @@ export class ClineProvider // Re-fetch history so delegateParentAndOpenChild sees the updated queue state const { historyItem: refreshedParentHistory } = await this.getTaskWithId(parentTaskId) + // Create worktree before task so we have the path ready; use parentTaskId as name hint + const nextChildWorktreePath = nextTask.worktree + ? await this._createWorktreeForTask( + refreshedParentHistory.workspace ?? this.cwd, + nextTask.worktree, + parentTaskId, + ) + : undefined const nextChild = await this.createTask(nextTask.message, undefined, undefined, { initialTodos: nextTodoItems, initialStatus: "active", startTask: false, - workspacePath: nextTask.worktree - ? await this._createWorktreeForTask(refreshedParentHistory.workspace ?? this.cwd, nextTask.worktree) - : undefined, + workspacePath: nextChildWorktreePath, }) // Record the next child in parent history @@ -3546,10 +3615,10 @@ export class ClineProvider childIds: updatedChildIds, }) - if (nextTask.worktree && nextChild.workspacePath !== refreshedParentHistory.workspace) { + if (nextChildWorktreePath) { try { const { historyItem: nextChildHistory } = await this.getTaskWithId(nextChild.taskId) - await this.updateTaskHistory({ ...nextChildHistory, worktreePath: nextChild.workspacePath }) + await this.updateTaskHistory({ ...nextChildHistory, worktreePath: nextChildWorktreePath }) } catch { // non-fatal } @@ -3656,9 +3725,23 @@ export class ClineProvider childIds, parallelQueue: undefined, parallelResults: undefined, + abortOnChildFailure: undefined, } await this.updateTaskHistory(updatedHistory) + if (isParallelFanIn) { + const hadFailures = accumulatedResults.some((r) => r.error !== undefined) + TelemetryService.instance.captureParallelTaskCompleted( + parentTaskId, + accumulatedResults.length, + accumulatedResults.length, + hadFailures, + ) + this.log( + `[reopenParentFromDelegation] Parallel fan-in complete for parent ${parentTaskId}: ${accumulatedResults.length} tasks, failures=${hadFailures}`, + ) + } + // 7) Emit TaskDelegationCompleted (provider-level) try { this.emit(RooCodeEventName.TaskDelegationCompleted, parentTaskId, childTaskId, completionResultSummary) @@ -3695,19 +3778,96 @@ export class ClineProvider } } - /** Creates a worktree for a child task, returning the path on success or undefined on failure. */ - private async _createWorktreeForTask(parentWorkspacePath: string, worktree: string): Promise { + /** + * Scans task history for orphaned worktrees — directories that still exist on disk but whose + * tasks are no longer running. This can happen if VS Code crashes mid-task. + * Logs findings and emits telemetry; offers optional VS Code notification for cleanup. + */ + public async detectAndCleanOrphanedWorktrees(): Promise { try { - const branchName = worktree === "auto" ? `roo/task-${Date.now().toString(36)}` : worktree + const allHistory = this.taskHistoryStore.getAll() + const activeTaskIds = new Set(this.clineStack.map((t) => t.taskId)) + const orphans: Array<{ taskId: string; worktreePath: string }> = [] + + for (const item of allHistory) { + if (!item.worktreePath) continue + // A completed task whose worktree dir still exists is an orphan + if (item.status === "completed" || (item.status !== "active" && !activeTaskIds.has(item.id))) { + try { + const { default: fss } = await import("fs") + if (fss.existsSync(item.worktreePath)) { + orphans.push({ taskId: item.id, worktreePath: item.worktreePath }) + } + } catch { + // non-fatal + } + } + } + + if (orphans.length === 0) return + + for (const { taskId, worktreePath } of orphans) { + this.log( + `[detectAndCleanOrphanedWorktrees] Orphaned worktree detected: ${worktreePath} (task ${taskId})`, + ) + TelemetryService.instance.captureWorktreeOrphanDetected(worktreePath) + } + + const paths = orphans.map((o) => o.worktreePath).join(", ") + const action = await vscode.window.showWarningMessage( + `Moo Code detected ${orphans.length} orphaned worktree(s) from previous sessions:\n${paths}`, + "Clean up", + "Ignore", + ) + + if (action === "Clean up") { + for (const { taskId, worktreePath } of orphans) { + try { + const cwd = this.cwd + await worktreeService.deleteWorktree(cwd, worktreePath) + this.log(`[detectAndCleanOrphanedWorktrees] Cleaned up worktree: ${worktreePath}`) + TelemetryService.instance.captureWorktreeDeleted(taskId, worktreePath) + // Clear the worktreePath from the history item + const { historyItem } = await this.getTaskWithId(taskId) + await this.updateTaskHistory({ ...historyItem, worktreePath: undefined }) + } catch (err) { + this.log( + `[detectAndCleanOrphanedWorktrees] Failed to clean worktree ${worktreePath}: ${ + (err as Error)?.message ?? String(err) + }`, + ) + } + } + } + } catch (err) { + this.log( + `[detectAndCleanOrphanedWorktrees] Error during orphan scan (non-fatal): ${ + (err as Error)?.message ?? String(err) + }`, + ) + } + } + + /** Creates a worktree for a child task, returning the path on success or undefined on failure. */ + private async _createWorktreeForTask( + parentWorkspacePath: string, + worktree: string, + taskIdHint?: string, + ): Promise { + try { + const shortId = taskIdHint ? taskIdHint.slice(-8) : Math.random().toString(36).slice(2, 8) + const branchName = worktree === "auto" ? `roo/task-${shortId}-${Date.now().toString(36)}` : worktree const worktreeBase = path.join(os.homedir(), ".roo", "worktrees") const projectName = path.basename(parentWorkspacePath) - const suffix = Math.random().toString(36).slice(2, 7) - const worktreeDest = path.join(worktreeBase, `${projectName}-${suffix}`) + const worktreeDest = path.join(worktreeBase, `${projectName}-${shortId}`) const result = await worktreeService.createWorktree(parentWorkspacePath, { path: worktreeDest, branch: branchName, createNewBranch: true, }) + if (result.success && taskIdHint) { + TelemetryService.instance.captureWorktreeCreated(taskIdHint, worktreeDest) + } return result.success ? worktreeDest : undefined } catch { return undefined diff --git a/src/extension.ts b/src/extension.ts index 1dfeb28254..ef6bbec96f 100644 --- a/src/extension.ts +++ b/src/extension.ts @@ -304,6 +304,14 @@ export async function activate(context: vscode.ExtensionContext) { // Check for worktree auto-open path (set when switching to a worktree) await checkWorktreeAutoOpen(context, outputChannel) + // Scan for orphaned worktrees from previous sessions (e.g. after a crash). + // Run in background so it doesn't delay activation. + void provider.detectAndCleanOrphanedWorktrees().catch((err) => { + outputChannel.appendLine( + `[OrphanWorktree] Startup scan failed (non-fatal): ${err instanceof Error ? err.message : String(err)}`, + ) + }) + // Auto-import configuration if specified in settings. try { await autoImportSettings(outputChannel, {