diff --git a/src/constants/terminationTimeouts.ts b/src/constants/terminationTimeouts.ts new file mode 100644 index 0000000000..67b298b175 --- /dev/null +++ b/src/constants/terminationTimeouts.ts @@ -0,0 +1,4 @@ +export const TASK_TERMINATION_TOOL_TIMEOUT_MS = 5 * 60 * 1000; +export const TASK_TERMINATION_STOP_STREAM_TIMEOUT_MS = 20 * 1000; +export const TASK_TERMINATION_WORKSPACE_REMOVE_TIMEOUT_MS = 2 * 60 * 1000; +export const WORKTREE_DELETE_GIT_TIMEOUT_MS = 60 * 1000; diff --git a/src/node/git.ts b/src/node/git.ts index 726517e9bd..afcc14b858 100644 --- a/src/node/git.ts +++ b/src/node/git.ts @@ -2,7 +2,7 @@ import * as fs from "fs"; import * as path from "path"; import type { Config } from "@/node/config"; import type { RuntimeConfig } from "@/common/types/runtime"; -import { execFileAsync } from "@/node/utils/disposableExec"; +import { execFileAsync, type ExecFileAsyncOptions } from "@/node/utils/disposableExec"; import { createRuntime } from "./runtime/runtimeFactory"; import { log } from "./services/log"; import { getErrorMessage } from "@/common/utils/errors"; @@ -47,14 +47,15 @@ export interface CreateWorktreeOptions { runtimeConfig?: RuntimeConfig; } -export async function listLocalBranches(projectPath: string): Promise { - using proc = execFileAsync("git", [ - "-C", - projectPath, - "for-each-ref", - "--format=%(refname:short)", - "refs/heads", - ]); +export async function listLocalBranches( + projectPath: string, + options?: ExecFileAsyncOptions +): Promise { + using proc = execFileAsync( + "git", + ["-C", projectPath, "for-each-ref", "--format=%(refname:short)", "refs/heads"], + options + ); const { stdout } = await proc.result; return stdout .split("\n") @@ -63,9 +64,16 @@ export async function listLocalBranches(projectPath: string): Promise .sort((a, b) => a.localeCompare(b)); } -export async function getCurrentBranch(projectPath: string): Promise { +export async function getCurrentBranch( + projectPath: string, + options?: ExecFileAsyncOptions +): Promise { try { - using proc = execFileAsync("git", ["-C", projectPath, "rev-parse", "--abbrev-ref", "HEAD"]); + using proc = execFileAsync( + "git", + ["-C", projectPath, "rev-parse", "--abbrev-ref", "HEAD"], + options + ); const { stdout } = await proc.result; const branch = stdout.trim(); if (!branch || branch === "HEAD") { diff --git a/src/node/services/taskService.test.ts b/src/node/services/taskService.test.ts index 3675651530..4c560f4c7e 100644 --- a/src/node/services/taskService.test.ts +++ b/src/node/services/taskService.test.ts @@ -4,6 +4,10 @@ import * as path from "path"; import * as os from "os"; import { execSync } from "node:child_process"; +import { + TASK_TERMINATION_STOP_STREAM_TIMEOUT_MS, + TASK_TERMINATION_WORKSPACE_REMOVE_TIMEOUT_MS, +} from "@/constants/terminationTimeouts"; import { Config, type ProjectConfig, @@ -9197,6 +9201,180 @@ describe("TaskService", () => { expect(remove).toHaveBeenNthCalledWith(2, parentTaskId, true); }); + test("terminateDescendantAgentTask skips a timed-out stream and continues other descendants", async () => { + const config = await createTestConfig(rootDir); + const projectPath = path.join(rootDir, "repo"); + const rootWorkspaceId = "root-111"; + const parentTaskId = "task-parent"; + const stuckTaskId = "task-stuck"; + const siblingTaskId = "task-sibling"; + + await saveWorkspaces( + config, + projectPath, + [ + projectWorkspace(projectPath, "root", rootWorkspaceId), + projectWorkspace(projectPath, "parent-task", parentTaskId, { + parentWorkspaceId: rootWorkspaceId, + agentType: "exec", + taskStatus: "running", + }), + projectWorkspace(projectPath, "stuck-task", stuckTaskId, { + parentWorkspaceId: parentTaskId, + agentType: "explore", + taskStatus: "running", + }), + projectWorkspace(projectPath, "sibling-task", siblingTaskId, { + parentWorkspaceId: parentTaskId, + agentType: "explore", + taskStatus: "running", + }), + ], + testTaskSettings() + ); + + const stopStream = mock((workspaceId: string): Promise> => { + return workspaceId === stuckTaskId + ? new Promise(() => undefined) + : Promise.resolve(Ok(undefined)); + }); + const { aiService } = createAIServiceMocks(config, { stopStream }); + const { workspaceService, remove } = createWorkspaceServiceMocks(); + const { taskService } = createTaskServiceHarness(config, { aiService, workspaceService }); + const originalSetTimeout = globalThis.setTimeout; + const timeoutSpy = spyOn(globalThis, "setTimeout").mockImplementation((( + handler: () => void, + timeout?: number + ) => { + if (timeout === TASK_TERMINATION_STOP_STREAM_TIMEOUT_MS) { + // Fire on a 0ms macrotask, not a microtask: already-resolved stop + // promises settle through several microtask hops first, so only the + // genuinely stuck stream can lose the race to this timer. + return originalSetTimeout(handler, 0); + } + return originalSetTimeout(handler, timeout); + }) as typeof setTimeout); + + const terminateResult = await taskService.terminateDescendantAgentTask( + rootWorkspaceId, + parentTaskId + ); + timeoutSpy.mockRestore(); + + expect(terminateResult.success).toBe(false); + if (terminateResult.success) return; + expect(terminateResult.error).toContain(`Timed out stopping task stream (${stuckTaskId})`); + expect(terminateResult.error).toContain( + `Skipped removing task workspace (${parentTaskId}): a descendant task workspace was not removed` + ); + expect(remove).not.toHaveBeenCalledWith(stuckTaskId, true); + expect(remove).toHaveBeenCalledWith(siblingTaskId, true); + // The stuck child survives, so its ancestor must survive too: a live child + // must never point at removed parent metadata. + expect(remove).not.toHaveBeenCalledWith(parentTaskId, true); + }); + + test("terminate retry awaits the original in-flight removal instead of trusting a dedup Ok", async () => { + const config = await createTestConfig(rootDir); + const projectPath = path.join(rootDir, "repo"); + const rootWorkspaceId = "root-111"; + const parentTaskId = "task-parent"; + const childTaskId = "task-child"; + + await saveWorkspaces( + config, + projectPath, + [ + projectWorkspace(projectPath, "root", rootWorkspaceId), + projectWorkspace(projectPath, "parent-task", parentTaskId, { + parentWorkspaceId: rootWorkspaceId, + agentType: "exec", + taskStatus: "running", + }), + projectWorkspace(projectPath, "child-task", childTaskId, { + parentWorkspaceId: parentTaskId, + agentType: "explore", + taskStatus: "running", + }), + ], + testTaskSettings() + ); + + // First removal of the child hangs; later calls return Ok, simulating + // WorkspaceService.remove()'s short-circuit for IDs already being removed. + let childRemoveCalls = 0; + const remove = mock((workspaceId: string, _force?: boolean): Promise> => { + if (workspaceId === childTaskId) { + childRemoveCalls += 1; + if (childRemoveCalls === 1) { + return new Promise(() => undefined); + } + } + return Promise.resolve(Ok(undefined)); + }); + const { workspaceService } = createWorkspaceServiceMocks({ remove }); + const { taskService } = createTaskServiceHarness(config, { workspaceService }); + const originalSetTimeout = globalThis.setTimeout; + const timeoutSpy = spyOn(globalThis, "setTimeout").mockImplementation((( + handler: () => void, + timeout?: number + ) => { + if (timeout === TASK_TERMINATION_WORKSPACE_REMOVE_TIMEOUT_MS) { + // 0ms macrotask so pending microtask chains settle first (see above). + return originalSetTimeout(handler, 0); + } + return originalSetTimeout(handler, timeout); + }) as typeof setTimeout); + + const firstAttempt = await taskService.terminateDescendantAgentTask( + rootWorkspaceId, + parentTaskId + ); + const retryAttempt = await taskService.terminateDescendantAgentTask( + rootWorkspaceId, + parentTaskId + ); + timeoutSpy.mockRestore(); + + expect(firstAttempt.success).toBe(false); + expect(retryAttempt.success).toBe(false); + if (retryAttempt.success) return; + expect(retryAttempt.error).toContain(`Timed out removing task workspace (${childTaskId})`); + // The retry must not re-call remove for the child (a fresh call would + // return the dedup Ok) and must keep the parent blocked. + expect(childRemoveCalls).toBe(1); + expect(remove).not.toHaveBeenCalledWith(parentTaskId, true); + }); + + test("descendant traversal terminates when task metadata contains a cycle", () => { + const config = new Config(rootDir); + const { taskService } = createTaskServiceHarness(config); + const traversal = taskService as unknown as { + listDescendantAgentTaskIdsFromIndex: ( + index: { + byId: Map; + childrenByParent: Map; + parentById: Map; + }, + workspaceId: string + ) => string[]; + }; + const index = { + byId: new Map(), + childrenByParent: new Map([ + ["root", ["child"]], + ["child", ["grandchild"]], + ["grandchild", ["child", "root"]], + ]), + parentById: new Map(), + }; + + expect(traversal.listDescendantAgentTaskIdsFromIndex(index, "root")).toEqual([ + "child", + "grandchild", + ]); + }); + test("terminateAllDescendantAgentTasks interrupts entire subtree leaf-first", async () => { const config = await createTestConfig(rootDir); diff --git a/src/node/services/taskService.ts b/src/node/services/taskService.ts index cbd904ceb4..e55f7e00ca 100644 --- a/src/node/services/taskService.ts +++ b/src/node/services/taskService.ts @@ -3,6 +3,11 @@ import * as fsPromises from "fs/promises"; import type { z } from "zod"; +import { + TASK_TERMINATION_STOP_STREAM_TIMEOUT_MS, + TASK_TERMINATION_WORKSPACE_REMOVE_TIMEOUT_MS, +} from "@/constants/terminationTimeouts"; +import { raceWithAbortAndTimeout } from "@/node/utils/concurrency/withTimeout"; import { MutexMap } from "@/node/utils/concurrency/mutexMap"; import { AsyncMutex } from "@/node/utils/concurrency/asyncMutex"; import type { Config, ProjectsConfig, Workspace as WorkspaceConfigEntry } from "@/node/config"; @@ -1135,6 +1140,12 @@ export class TaskService { // Cache completed reports so callers can retrieve them without re-reading disk. // Bounded by max entries; disk persistence is the source of truth for restart-safety. private readonly completedReportsByTaskId = new Map(); + + // Task workspace removals that outlived their termination timeout. Retries must + // await the ORIGINAL removal outcome: WorkspaceService.remove() short-circuits Ok + // for IDs already being removed, so re-calling it would count a still-in-flight + // (possibly failing) removal as success and let ancestor deletion orphan the child. + private readonly pendingTaskWorkspaceRemovals = new Map>>(); private readonly gitPatchArtifactService: GitPatchArtifactService; private readonly handoffInProgress = new Set(); /** @@ -3832,6 +3843,7 @@ export class TaskService { assert(taskId.length > 0, "terminateDescendantAgentTask: taskId must be non-empty"); const terminatedTaskIds: string[] = []; + const terminationErrors: string[] = []; { await using _lock = await this.mutex.acquire(); @@ -3863,23 +3875,103 @@ export class TaskService { const terminationError = new Error("Task terminated"); + // When a descendant workspace could not be removed, keep every ancestor of it + // so the surviving child never points at removed parent metadata. + const ancestorsBlockedByFailedChild = new Set(); + const blockAncestorsOf = (id: string) => { + for ( + let cur = parentById.get(id); + cur != null && !ancestorsBlockedByFailedChild.has(cur); + cur = parentById.get(cur) + ) { + ancestorsBlockedByFailedChild.add(cur); + } + }; + for (const id of toTerminate) { // Best-effort: stop any active stream immediately to avoid further token usage. try { - const stopResult = await this.aiService.stopStream(id, { abandonPartial: true }); - if (!stopResult.success) { + const stopPromise = this.aiService.stopStream(id, { abandonPartial: true }); + const stopOutcome = await raceWithAbortAndTimeout(stopPromise, { + timeoutMs: TASK_TERMINATION_STOP_STREAM_TIMEOUT_MS, + }); + if (stopOutcome.kind !== "ok") { + void stopPromise.catch((error: unknown) => { + log.debug("terminateDescendantAgentTask: timed-out stopStream later threw", { + taskId: id, + error, + }); + }); + terminationErrors.push(`Timed out stopping task stream (${id})`); + blockAncestorsOf(id); + continue; + } + if (!stopOutcome.value.success) { log.debug("terminateDescendantAgentTask: stopStream failed", { taskId: id }); } } catch (error: unknown) { log.debug("terminateDescendantAgentTask: stopStream threw", { taskId: id, error }); } + if (ancestorsBlockedByFailedChild.has(id)) { + terminationErrors.push( + `Skipped removing task workspace (${id}): a descendant task workspace was not removed` + ); + continue; + } + this.completedReportsByTaskId.delete(id); this.rejectWaiters(id, terminationError); - const removeResult = await this.workspaceService.remove(id, true); - if (!removeResult.success) { - return Err(`Failed to remove task workspace (${id}): ${removeResult.error}`); + try { + let removePromise = this.pendingTaskWorkspaceRemovals.get(id); + if (!removePromise) { + removePromise = this.workspaceService.remove(id, true); + this.pendingTaskWorkspaceRemovals.set(id, removePromise); + const trackedPromise = removePromise; + void trackedPromise + .then( + (result) => result.success, + (error: unknown) => { + log.debug("terminateDescendantAgentTask: workspace removal threw", { + taskId: id, + error, + }); + return false; + } + ) + .then(async (removed) => { + if (this.pendingTaskWorkspaceRemovals.get(id) === trackedPromise) { + this.pendingTaskWorkspaceRemovals.delete(id); + } + // A removal that outlived its termination timeout frees the task slot + // only when it settles, so kick the scheduler for queued tasks then. + if (removed) { + await this.maybeStartQueuedTasks(); + } + }); + } + const removeOutcome = await raceWithAbortAndTimeout(removePromise, { + timeoutMs: TASK_TERMINATION_WORKSPACE_REMOVE_TIMEOUT_MS, + }); + if (removeOutcome.kind !== "ok") { + terminationErrors.push(`Timed out removing task workspace (${id})`); + blockAncestorsOf(id); + continue; + } + if (!removeOutcome.value.success) { + terminationErrors.push( + `Failed to remove task workspace (${id}): ${removeOutcome.value.error}` + ); + blockAncestorsOf(id); + continue; + } + } catch (error: unknown) { + terminationErrors.push( + `Failed to remove task workspace (${id}): ${getErrorMessage(error)}` + ); + blockAncestorsOf(id); + continue; } terminatedTaskIds.push(id); @@ -3889,6 +3981,9 @@ export class TaskService { // Free slots and start any queued tasks (best-effort). await this.maybeStartQueuedTasks(); + if (terminationErrors.length > 0) { + return Err(terminationErrors.join("; ")); + } return Ok({ terminatedTaskIds }); } @@ -6585,9 +6680,14 @@ export class TaskService { ); const result: string[] = []; + const visited = new Set([workspaceId]); const stack: string[] = [...(index.childrenByParent.get(workspaceId) ?? [])]; while (stack.length > 0) { const next = stack.pop()!; + if (visited.has(next)) { + continue; + } + visited.add(next); result.push(next); const children = index.childrenByParent.get(next); if (children) { diff --git a/src/node/services/tools/task_terminate.test.ts b/src/node/services/tools/task_terminate.test.ts index 545306f42d..fbc3ffadb2 100644 --- a/src/node/services/tools/task_terminate.test.ts +++ b/src/node/services/tools/task_terminate.test.ts @@ -60,6 +60,32 @@ describe("task_terminate tool", () => { }); }); + it("reports aggregated cleanup failures as error, not invalid_scope", async () => { + using tempDir = new TestTempDir("test-task-terminate-cleanup-error"); + const baseConfig = createTestToolConfig(tempDir.path, { workspaceId: "root-workspace" }); + const cleanupError = + "Timed out stopping task stream (child-task); " + + "Skipped removing task workspace (parent-task): a descendant task workspace was not removed"; + + const taskService = { + terminateDescendantAgentTask: mock( + (): Promise> => + Promise.resolve(Err(cleanupError)) + ), + listActiveDescendantAgentTaskIds: mock(() => []), + } as unknown as TaskService; + + const tool = createTaskTerminateTool({ ...baseConfig, taskService }); + + const result: unknown = await Promise.resolve( + tool.execute!({ task_ids: ["parent-task"] }, mockToolCallOptions) + ); + + expect(result).toEqual({ + results: [{ status: "error", taskId: "parent-task", error: cleanupError }], + }); + }); + it("returns terminated with terminatedTaskIds on success", async () => { using tempDir = new TestTempDir("test-task-terminate-ok"); const baseConfig = createTestToolConfig(tempDir.path, { workspaceId: "root-workspace" }); @@ -88,6 +114,52 @@ describe("task_terminate tool", () => { }); }); + it("returns an interrupted error promptly while completed task IDs still resolve", async () => { + using tempDir = new TestTempDir("test-task-terminate-abort"); + const baseConfig = createTestToolConfig(tempDir.path, { workspaceId: "root-workspace" }); + const controller = new AbortController(); + + const taskService = { + terminateDescendantAgentTask: mock( + ( + _workspaceId: string, + taskId: string + ): Promise> => { + if (taskId === "stuck-task") { + return new Promise(() => undefined); + } + return Promise.resolve(Ok({ terminatedTaskIds: [taskId] })); + } + ), + } as unknown as TaskService; + const tool = createTaskTerminateTool({ ...baseConfig, taskService }); + + const resultPromise = Promise.resolve( + tool.execute!( + { task_ids: ["stuck-task", "finished-task"] }, + { ...mockToolCallOptions, abortSignal: controller.signal } + ) + ); + await Promise.resolve(); + await Promise.resolve(); + controller.abort(); + + expect(await resultPromise).toEqual({ + results: [ + { + status: "error", + taskId: "stuck-task", + error: "Termination interrupted; cleanup continues in the background", + }, + { + status: "terminated", + taskId: "finished-task", + terminatedTaskIds: ["finished-task"], + }, + ], + }); + }); + it("interrupts a workspace turn without deleting the workspace", async () => { using tempDir = new TestTempDir("test-task-terminate-workspace-turn"); const baseConfig = createTestToolConfig(tempDir.path, { workspaceId: "root-workspace" }); @@ -181,6 +253,62 @@ describe("task_terminate tool", () => { }); }); + it("does not start termination when the signal is already aborted", async () => { + using tempDir = new TestTempDir("test-task-terminate-preaborted"); + const baseConfig = createTestToolConfig(tempDir.path, { workspaceId: "root-workspace" }); + const controller = new AbortController(); + controller.abort(); + + const terminateDescendantAgentTask = mock( + (): Promise> => + Promise.resolve(Ok({ terminatedTaskIds: ["child-task"] })) + ); + const tool = createTaskTerminateTool({ + ...baseConfig, + taskService: { terminateDescendantAgentTask } as unknown as TaskService, + }); + + const result: unknown = await Promise.resolve( + tool.execute!( + { task_ids: ["child-task"] }, + { ...mockToolCallOptions, abortSignal: controller.signal } + ) + ); + + expect(terminateDescendantAgentTask).not.toHaveBeenCalled(); + expect(result).toEqual({ + results: [ + { + status: "error", + taskId: "child-task", + error: "Termination interrupted before it started", + }, + ], + }); + }); + + it("returns a per-task error when a workflow branch throws", async () => { + using tempDir = new TestTempDir("test-task-terminate-workflow-throws"); + const baseConfig = createTestToolConfig(tempDir.path, { workspaceId: "root-workspace" }); + + const tool = createTaskTerminateTool({ + ...baseConfig, + taskService: {} as unknown as TaskService, + workflowService: { + getRun: mock(() => Promise.reject(new Error("workflow lookup failed"))), + interruptRun: mock(() => Promise.reject(new Error("unused"))), + }, + }); + + const result: unknown = await Promise.resolve( + tool.execute!({ task_ids: ["wfr_run_1"] }, mockToolCallOptions) + ); + + expect(result).toEqual({ + results: [{ status: "error", taskId: "wfr_run_1", error: "workflow lookup failed" }], + }); + }); + it("treats interrupting an already-interrupted workflow run as idempotent success", async () => { using tempDir = new TestTempDir("test-task-terminate-workflow-idempotent"); const baseConfig = createTestToolConfig(tempDir.path, { workspaceId: "root-workspace" }); diff --git a/src/node/services/tools/task_terminate.ts b/src/node/services/tools/task_terminate.ts index ba7c2b2e4f..54018f472c 100644 --- a/src/node/services/tools/task_terminate.ts +++ b/src/node/services/tools/task_terminate.ts @@ -8,6 +8,9 @@ import { TOOL_DEFINITIONS, } from "@/common/utils/tools/toolDefinitions"; +import { TASK_TERMINATION_TOOL_TIMEOUT_MS } from "@/constants/terminationTimeouts"; +import { log } from "@/node/services/log"; +import { raceWithAbortAndTimeout } from "@/node/utils/concurrency/withTimeout"; import { isWorkspaceTurnTaskId } from "@/node/services/taskHandleStore"; import { fromBashTaskId, isWorkflowRunTaskId } from "./taskId"; import { @@ -80,7 +83,7 @@ export const createTaskTerminateTool: ToolFactory = (config: ToolConfiguration) return tool({ description: TOOL_DEFINITIONS.task_terminate.description, inputSchema: TOOL_DEFINITIONS.task_terminate.schema, - execute: async (args): Promise => { + execute: async (args, { abortSignal }): Promise => { const workspaceId = requireWorkspaceId(config, "task_terminate"); const taskService = requireTaskService(config, "task_terminate"); @@ -88,85 +91,127 @@ export const createTaskTerminateTool: ToolFactory = (config: ToolConfiguration) const results = await Promise.all( uniqueTaskIds.map(async (taskId) => { - if (isWorkflowRunTaskId(taskId)) { - return await interruptWorkflowRun(config, workspaceId, taskId); - } - - if (isWorkspaceTurnTaskId(taskId)) { - const interruptResult = await taskService.interruptWorkspaceTurn(workspaceId, taskId); - if (!interruptResult.success) { - const msg = interruptResult.error; - if (/not found/i.test(msg) || /scope/i.test(msg)) { - return { status: "invalid_scope" as const, taskId }; - } - return { status: "error" as const, taskId, error: msg }; - } + // A pre-aborted call must not start destructive termination work at all. + if (abortSignal?.aborted) { return { - status: "interrupted" as const, + status: "error" as const, taskId, - note: "Workspace turn interrupted. The full workspace is preserved for inspection and future prompts.", + error: "Termination interrupted before it started", }; } + const terminationPromise = (async () => { + try { + if (isWorkflowRunTaskId(taskId)) { + return await interruptWorkflowRun(config, workspaceId, taskId); + } - const maybeProcessId = fromBashTaskId(taskId); - if (taskId.startsWith("bash:") && !maybeProcessId) { - return { status: "error" as const, taskId, error: "Invalid bash taskId." }; - } - - if (maybeProcessId) { - if (!config.backgroundProcessManager) { - return { - status: "error" as const, - taskId, - error: "Background process manager not available", - }; - } - - const proc = await config.backgroundProcessManager.getProcess(maybeProcessId); - if (!proc) { - return { status: "not_found" as const, taskId }; - } + if (isWorkspaceTurnTaskId(taskId)) { + const interruptResult = await taskService.interruptWorkspaceTurn( + workspaceId, + taskId + ); + if (!interruptResult.success) { + const msg = interruptResult.error; + if (/not found/i.test(msg) || /scope/i.test(msg)) { + return { status: "invalid_scope" as const, taskId }; + } + return { status: "error" as const, taskId, error: msg }; + } + return { + status: "interrupted" as const, + taskId, + note: "Workspace turn interrupted. The full workspace is preserved for inspection and future prompts.", + }; + } - const inScope = - proc.workspaceId === workspaceId || - (await taskService.isDescendantAgentTask(workspaceId, proc.workspaceId)); - if (!inScope) { - return { status: "invalid_scope" as const, taskId }; - } + const maybeProcessId = fromBashTaskId(taskId); + if (taskId.startsWith("bash:") && !maybeProcessId) { + return { status: "error" as const, taskId, error: "Invalid bash taskId." }; + } - const terminateResult = await config.backgroundProcessManager.terminate(maybeProcessId); - if (!terminateResult.success) { - return { status: "error" as const, taskId, error: terminateResult.error }; - } + if (maybeProcessId) { + if (!config.backgroundProcessManager) { + return { + status: "error" as const, + taskId, + error: "Background process manager not available", + }; + } + + const proc = await config.backgroundProcessManager.getProcess(maybeProcessId); + if (!proc) { + return { status: "not_found" as const, taskId }; + } + + const inScope = + proc.workspaceId === workspaceId || + (await taskService.isDescendantAgentTask(workspaceId, proc.workspaceId)); + if (!inScope) { + return { status: "invalid_scope" as const, taskId }; + } + + const terminateResult = + await config.backgroundProcessManager.terminate(maybeProcessId); + if (!terminateResult.success) { + return { status: "error" as const, taskId, error: terminateResult.error }; + } + + return { + status: "terminated" as const, + taskId, + terminatedTaskIds: [taskId], + }; + } - return { - status: "terminated" as const, - taskId, - terminatedTaskIds: [taskId], - }; - } + const terminateResult = await taskService.terminateDescendantAgentTask( + workspaceId, + taskId + ); + if (!terminateResult.success) { + const msg = terminateResult.error; + const activeDescendantIds = + taskService.listActiveDescendantAgentTaskIds(workspaceId); + const activeTaskIds = + activeDescendantIds.length > 0 ? activeDescendantIds : undefined; + // Exact-match the canonical scope errors: aggregated cleanup failures + // may mention "descendant" or "not found" and must stay actionable errors. + if (msg === "Task not found") { + return { status: "not_found" as const, taskId, activeTaskIds }; + } + if (msg === "Task is not a descendant of this workspace") { + return { status: "invalid_scope" as const, taskId, activeTaskIds }; + } + return { status: "error" as const, taskId, error: msg }; + } - const terminateResult = await taskService.terminateDescendantAgentTask( - workspaceId, - taskId - ); - if (!terminateResult.success) { - const msg = terminateResult.error; - const activeDescendantIds = taskService.listActiveDescendantAgentTaskIds(workspaceId); - const activeTaskIds = activeDescendantIds.length > 0 ? activeDescendantIds : undefined; - if (/not found/i.test(msg)) { - return { status: "not_found" as const, taskId, activeTaskIds }; - } - if (/descendant/i.test(msg) || /scope/i.test(msg)) { - return { status: "invalid_scope" as const, taskId, activeTaskIds }; + return { + status: "terminated" as const, + taskId, + terminatedTaskIds: terminateResult.data.terminatedTaskIds, + }; + } catch (error: unknown) { + return { status: "error" as const, taskId, error: getErrorMessage(error) }; } - return { status: "error" as const, taskId, error: msg }; + })(); + + const outcome = await raceWithAbortAndTimeout(terminationPromise, { + signal: abortSignal, + timeoutMs: TASK_TERMINATION_TOOL_TIMEOUT_MS, + }); + if (outcome.kind === "ok") { + return outcome.value; } + void terminationPromise.catch((error: unknown) => { + log.debug("task_terminate cleanup failed after tool returned", { taskId, error }); + }); return { - status: "terminated" as const, + status: "error" as const, taskId, - terminatedTaskIds: terminateResult.data.terminatedTaskIds, + error: + outcome.kind === "aborted" + ? "Termination interrupted; cleanup continues in the background" + : "Termination timed out; cleanup continues in the background", }; }) ); diff --git a/src/node/services/workspaceService.ts b/src/node/services/workspaceService.ts index 5557d90015..5bd71edc1c 100644 --- a/src/node/services/workspaceService.ts +++ b/src/node/services/workspaceService.ts @@ -1,3 +1,5 @@ +import { TASK_TERMINATION_STOP_STREAM_TIMEOUT_MS } from "@/constants/terminationTimeouts"; +import { raceWithAbortAndTimeout } from "@/node/utils/concurrency/withTimeout"; import { EventEmitter } from "events"; import * as path from "path"; import * as fsPromises from "fs/promises"; @@ -4122,11 +4124,23 @@ export class WorkspaceService extends EventEmitter { : undefined; try { - const stopResult = await this.aiService.stopStream(workspaceId, { abandonPartial: true }); - if (!stopResult.success) { + const stopPromise = this.aiService.stopStream(workspaceId, { abandonPartial: true }); + const stopOutcome = await raceWithAbortAndTimeout(stopPromise, { + timeoutMs: TASK_TERMINATION_STOP_STREAM_TIMEOUT_MS, + }); + if (stopOutcome.kind !== "ok") { + void stopPromise.catch((error: unknown) => { + log.debug("Timed-out workspace removal stopStream later threw", { + workspaceId, + error, + }); + }); + return Err("Timed out stopping workspace stream; workspace was not removed"); + } + if (!stopOutcome.value.success) { log.debug("Failed to stop stream during workspace removal", { workspaceId, - error: stopResult.error, + error: stopOutcome.value.error, }); } } catch (error: unknown) { diff --git a/src/node/utils/concurrency/withTimeout.test.ts b/src/node/utils/concurrency/withTimeout.test.ts new file mode 100644 index 0000000000..c78ba35bfc --- /dev/null +++ b/src/node/utils/concurrency/withTimeout.test.ts @@ -0,0 +1,38 @@ +import { afterEach, describe, expect, it, vi } from "bun:test"; + +import { raceWithAbortAndTimeout } from "./withTimeout"; + +afterEach(() => { + vi.restoreAllMocks(); +}); + +describe("raceWithAbortAndTimeout", () => { + it("returns a resolved value and clears the timeout", async () => { + const clearTimeoutSpy = vi.spyOn(globalThis, "clearTimeout"); + + const result = await raceWithAbortAndTimeout(Promise.resolve("done"), { timeoutMs: 1000 }); + + expect(result).toEqual({ kind: "ok", value: "done" }); + expect(clearTimeoutSpy).toHaveBeenCalledTimes(1); + }); + + it("returns timeout when the deadline expires", async () => { + const result = await raceWithAbortAndTimeout(new Promise(() => undefined), { + timeoutMs: 1, + }); + + expect(result).toEqual({ kind: "timeout" }); + }); + + it("returns aborted when the signal aborts", async () => { + const controller = new AbortController(); + const resultPromise = raceWithAbortAndTimeout(new Promise(() => undefined), { + signal: controller.signal, + timeoutMs: 1000, + }); + + controller.abort(); + + expect(await resultPromise).toEqual({ kind: "aborted" }); + }); +}); diff --git a/src/node/utils/concurrency/withTimeout.ts b/src/node/utils/concurrency/withTimeout.ts new file mode 100644 index 0000000000..e884ec74f7 --- /dev/null +++ b/src/node/utils/concurrency/withTimeout.ts @@ -0,0 +1,53 @@ +export type AbortAndTimeoutResult = + | { kind: "ok"; value: T } + | { kind: "timeout" } + | { kind: "aborted" }; + +export async function raceWithAbortAndTimeout( + promise: Promise, + options: { signal?: AbortSignal; timeoutMs?: number } +): Promise> { + if (options.signal?.aborted) { + return { kind: "aborted" }; + } + + return await new Promise>((resolve, reject) => { + let settled = false; + let timer: ReturnType | undefined; + + const cleanup = () => { + if (timer != null) { + clearTimeout(timer); + } + options.signal?.removeEventListener("abort", onAbort); + }; + const settle = (result: AbortAndTimeoutResult) => { + if (settled) { + return; + } + settled = true; + cleanup(); + resolve(result); + }; + const onAbort = () => settle({ kind: "aborted" }); + + options.signal?.addEventListener("abort", onAbort, { once: true }); + // Register settlement before arming the timer so an already-settled + // promise wins over a simultaneous timeout. + promise.then( + (value) => settle({ kind: "ok", value }), + (error: unknown) => { + if (settled) { + return; + } + settled = true; + cleanup(); + reject(error instanceof Error ? error : new Error(String(error))); + } + ); + if (options.timeoutMs != null) { + timer = setTimeout(() => settle({ kind: "timeout" }), options.timeoutMs); + timer.unref?.(); + } + }); +} diff --git a/src/node/worktree/WorktreeManager.ts b/src/node/worktree/WorktreeManager.ts index 498e4eb3c6..be5ff40beb 100644 --- a/src/node/worktree/WorktreeManager.ts +++ b/src/node/worktree/WorktreeManager.ts @@ -7,7 +7,7 @@ import type { InitLogger, } from "@/node/runtime/Runtime"; import { listLocalBranches, cleanStaleLock, getCurrentBranch } from "@/node/git"; -import { execAsync, execFileAsync } from "@/node/utils/disposableExec"; +import { execAsync, execFileAsync, type ExecFileAsyncOptions } from "@/node/utils/disposableExec"; import { getBashPath } from "@/node/utils/main/bashPath"; import { getProjectName } from "@/node/utils/runtime/helpers"; import { getErrorMessage } from "@/common/utils/errors"; @@ -16,10 +16,11 @@ import { expandTilde } from "@/node/runtime/tildeExpansion"; import { toPosixPath } from "@/node/utils/paths"; import { log } from "@/node/services/log"; import { GIT_NO_HOOKS_ENV } from "@/node/utils/gitNoHooksEnv"; +import { WORKTREE_DELETE_GIT_TIMEOUT_MS } from "@/constants/terminationTimeouts"; import { syncLocalGitSubmodules } from "@/node/runtime/submoduleSync"; import { syncMuxignoreFiles } from "./muxignore"; -type GitExecOptions = { env?: Record; signal?: AbortSignal } | undefined; +type GitExecOptions = Pick | undefined; function isAbortError(_error: unknown, signal?: AbortSignal): boolean { return signal?.aborted ?? false; @@ -498,8 +499,11 @@ export class WorktreeManager { // Clean up stale lock before git operations on main repo cleanStaleLock(projectPath); - // Disable git hooks for untrusted projects - const noHooksEnv = this.getGitExecOptions(trusted); + // Disable git hooks for untrusted projects and bound every git cleanup command. + const noHooksEnv: GitExecOptions = { + ...(this.getGitExecOptions(trusted) ?? {}), + timeoutMs: WORKTREE_DELETE_GIT_TIMEOUT_MS, + }; // In-place workspaces are identified by projectPath === workspaceName. // These are direct workspace directories (e.g., CLI/benchmark sessions), not git worktrees. @@ -749,7 +753,7 @@ export class WorktreeManager { let localBranches: string[]; try { - localBranches = await listLocalBranches(args.projectPath); + localBranches = await listLocalBranches(args.projectPath, args.noHooksEnv); } catch (error) { log.debug("Failed to list local branches; skipping branch deletion", { projectPath: args.projectPath, @@ -820,7 +824,7 @@ export class WorktreeManager { protectedBranches.add(localBranches[0]); } - const currentBranch = await getCurrentBranch(projectPath); + const currentBranch = await getCurrentBranch(projectPath, noHooksEnv); if (currentBranch) { protectedBranches.add(currentBranch); }