From 6884fe8f5291f10a0ae327eb0532f0b65b318842 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Thu, 18 Jun 2026 14:44:19 +0000 Subject: [PATCH 1/7] =?UTF-8?q?=F0=9F=A4=96=20fix:=20prevent=20workflow=20?= =?UTF-8?q?task=20recovery=20after=20interrupts?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Workflow-owned task recovery now checks the owning workflow run status before startup, queued-launch, stream-end, or stream-error recovery can restart the task. Inactive workflow owners interrupt child tasks instead, preserving queued prompts for later inspection or manual workflow resume.\n\n---\n\n_Generated with mux • Model: `openai:gpt-5.5` • Thinking: `xhigh` • Cost: `$17.65`_\n\n --- src/node/services/taskService.test.ts | 146 +++++++++++++++++++++ src/node/services/taskService.ts | 180 +++++++++++++++++++++++++- 2 files changed, 324 insertions(+), 2 deletions(-) diff --git a/src/node/services/taskService.test.ts b/src/node/services/taskService.test.ts index e795c81aa7..9334250d7f 100644 --- a/src/node/services/taskService.test.ts +++ b/src/node/services/taskService.test.ts @@ -5599,6 +5599,96 @@ describe("TaskService", () => { expect(childTask?.reportedAt).toBeUndefined(); }); + test("initialize interrupts workflow-owned tasks instead of recovering them after owning workflow interrupt", async () => { + const config = await createTestConfig(rootDir); + + const projectPath = path.join(rootDir, "repo"); + const runtimeConfig = { type: "local" as const }; + const parentId = "parent-workflow-interrupted"; + const queuedChildId = "child-workflow-queued"; + const runningChildId = "child-workflow-running"; + const awaitingChildId = "child-workflow-awaiting"; + const nestedRunningChildId = "child-workflow-nested-running"; + const workflowRunId = "wfr_interrupted_owner"; + + await saveWorkspaces( + config, + projectPath, + [ + projectWorkspace(projectPath, "parent", parentId, { runtimeConfig }), + projectWorkspace(projectPath, "queued", queuedChildId, { + name: "agent_explore_queued", + parentWorkspaceId: parentId, + agentId: "explore", + agentType: "explore", + taskStatus: "queued", + taskPrompt: "queued work", + taskModelString: defaultModel, + runtimeConfig, + workflowTask: { runId: workflowRunId, stepId: "queued" }, + }), + projectWorkspace(projectPath, "running", runningChildId, { + name: "agent_explore_running", + parentWorkspaceId: parentId, + agentId: "explore", + agentType: "explore", + taskStatus: "running", + taskModelString: defaultModel, + runtimeConfig, + workflowTask: { runId: workflowRunId, stepId: "running" }, + }), + projectWorkspace(projectPath, "awaiting", awaitingChildId, { + name: "agent_explore_awaiting", + parentWorkspaceId: parentId, + agentId: "explore", + agentType: "explore", + taskStatus: "awaiting_report", + taskModelString: defaultModel, + runtimeConfig, + workflowTask: { runId: workflowRunId, stepId: "awaiting" }, + }), + projectWorkspace(projectPath, "nested-running", nestedRunningChildId, { + name: "agent_explore_nested_running", + parentWorkspaceId: runningChildId, + agentId: "explore", + agentType: "explore", + taskStatus: "running", + taskModelString: defaultModel, + runtimeConfig, + }), + ], + testTaskSettings(10, 3) + ); + const runStore = new WorkflowRunStore({ sessionDir: config.getSessionDir(parentId) }); + await runStore.createRun({ + id: workflowRunId, + workspaceId: parentId, + definition: { + name: "interrupted", + description: "Interrupted", + scope: "built-in", + executable: true, + }, + definitionSource: "export default function workflow() { return {}; }\n", + args: {}, + now: "2026-05-29T00:00:00.000Z", + }); + await runStore.appendStatus(workflowRunId, "interrupted", "2026-05-29T00:00:01.000Z"); + + const { aiService } = createAIServiceMocks(config, { isStreaming: mock(() => false) }); + const { workspaceService, sendMessage, resumeStream } = createWorkspaceServiceMocks(); + const { taskService } = createTaskServiceHarness(config, { aiService, workspaceService }); + + await taskService.initialize(); + + expect(sendMessage).not.toHaveBeenCalled(); + expect(resumeStream).not.toHaveBeenCalled(); + for (const taskId of [queuedChildId, runningChildId, awaitingChildId, nestedRunningChildId]) { + expect(findWorkspaceInConfig(config, taskId)?.taskStatus).toBe("interrupted"); + } + expect(findWorkspaceInConfig(config, queuedChildId)?.taskPrompt).toBe("queued work"); + }); + test("initialize resumes awaiting_report tasks after restart", async () => { const config = await createTestConfig(rootDir); @@ -8728,6 +8818,62 @@ describe("TaskService", () => { } }); + test("handleStreamEnd interrupts workflow-owned tasks when owning workflow is already interrupted", async () => { + const config = await createTestConfig(rootDir); + + const projectPath = path.join(rootDir, "repo"); + const parentId = "parent-stream-workflow-interrupted"; + const childId = "child-stream-workflow-interrupted"; + const workflowRunId = "wfr_stream_interrupted_owner"; + + await saveWorkspaces( + config, + projectPath, + [ + projectWorkspace(projectPath, "parent", parentId), + projectWorkspace(projectPath, "child", childId, { + name: "agent_explore_child", + parentWorkspaceId: parentId, + agentId: "explore", + agentType: "explore", + taskStatus: "running", + taskModelString: "openai:gpt-4o-mini", + workflowTask: { runId: workflowRunId, stepId: "slow-step" }, + }), + ], + testTaskSettings() + ); + const runStore = new WorkflowRunStore({ sessionDir: config.getSessionDir(parentId) }); + await runStore.createRun({ + id: workflowRunId, + workspaceId: parentId, + definition: { + name: "interrupted", + description: "Interrupted", + scope: "built-in", + executable: true, + }, + definitionSource: "export default function workflow() { return {}; }\n", + args: {}, + now: "2026-05-29T00:00:00.000Z", + }); + await runStore.appendStatus(workflowRunId, "interrupted", "2026-05-29T00:00:01.000Z"); + + const { workspaceService, sendMessage } = createWorkspaceServiceMocks(); + const { taskService } = createTaskServiceHarness(config, { workspaceService }); + + await handleTaskServiceStreamEndForTest(taskService, { + type: "stream-end", + workspaceId: childId, + messageId: "assistant-child-output", + metadata: { model: "openai:gpt-4o-mini" }, + parts: [], + }); + + expect(sendMessage).not.toHaveBeenCalled(); + expect(findWorkspaceInConfig(config, childId)?.taskStatus).toBe("interrupted"); + }); + test("non-plan subagent stream-end with final assistant text finalizes an implicit report", async () => { const config = await createTestConfig(rootDir); diff --git a/src/node/services/taskService.ts b/src/node/services/taskService.ts index 5e5f4b62a0..7e838009ab 100644 --- a/src/node/services/taskService.ts +++ b/src/node/services/taskService.ts @@ -5,7 +5,7 @@ import type { z } from "zod"; import { MutexMap } from "@/node/utils/concurrency/mutexMap"; import { AsyncMutex } from "@/node/utils/concurrency/asyncMutex"; -import type { Config, Workspace as WorkspaceConfigEntry } from "@/node/config"; +import type { Config, ProjectsConfig, Workspace as WorkspaceConfigEntry } from "@/node/config"; import type { AIService } from "@/node/services/aiService"; import type { WorkspaceService } from "@/node/services/workspaceService"; import type { HistoryService } from "@/node/services/historyService"; @@ -74,7 +74,11 @@ import type { WorkspaceGoalService } from "@/node/services/workspaceGoalService" import { getTotalCost, sumUsageHistory } from "@/common/utils/tokens/usageAggregator"; import type { ParsedThinkingInput, ThinkingLevel } from "@/common/types/thinking"; import type { ErrorEvent, StreamEndEvent } from "@/common/types/stream"; -import { isActiveWorkflowRunStatus, isTerminalWorkflowRunStatus } from "@/common/types/workflow"; +import { + isActiveWorkflowRunStatus, + isTerminalWorkflowRunStatus, + type WorkflowRunStatus, +} from "@/common/types/workflow"; import { isDynamicToolPart, type DynamicToolPart } from "@/common/types/toolParts"; import { isWorkflowDisplayOnlyMessage, @@ -460,6 +464,13 @@ interface AgentTaskIndex { parentById: Map; } +interface InactiveWorkflowTaskOwner { + ownerTaskId: string; + runId: string; + status?: WorkflowRunStatus; + reason: string; +} + interface PendingTaskWaiter { taskId: string; resolve: (report: { reportMarkdown: string; title?: string; structuredOutput?: unknown }) => void; @@ -1029,6 +1040,127 @@ export class TaskService { } } + // Workflow abort/interrupt is the source of truth: task-level restart and stream-end + // recovery must not resurrect a workflow child (or its descendants) once the owning run + // is no longer active. Manual workflow_resume replays from the workflow journal instead. + private findWorkflowTaskOwnerInAncestry( + index: AgentTaskIndex, + taskId: string + ): { taskId: string; workspace: AgentTaskWorkspaceEntry } | null { + assert(taskId.length > 0, "findWorkflowTaskOwnerInAncestry requires taskId"); + let current: string | undefined = taskId; + for (let depth = 0; current != null; depth += 1) { + assert( + depth < 32, + `findWorkflowTaskOwnerInAncestry: possible parentWorkspaceId cycle starting at ${taskId}` + ); + const entry = index.byId.get(current); + if (entry?.workflowTask != null) { + return { taskId: current, workspace: entry }; + } + current = index.parentById.get(current); + } + return null; + } + + private async getInactiveWorkflowTaskOwnerForRecovery( + taskId: string, + config: ProjectsConfig + ): Promise { + assert(taskId.length > 0, "getInactiveWorkflowTaskOwnerForRecovery requires taskId"); + const owner = this.findWorkflowTaskOwnerInAncestry(this.buildAgentTaskIndex(config), taskId); + if (owner == null) { + return null; + } + + const workflowTask = owner.workspace.workflowTask; + assert(workflowTask != null, "workflow task owner must include workflowTask metadata"); + const parentWorkspaceId = coerceNonEmptyString(owner.workspace.parentWorkspaceId); + if (!parentWorkspaceId) { + return { + ownerTaskId: owner.taskId, + runId: workflowTask.runId, + reason: "workflow-owned task is missing its parent workspace", + }; + } + + try { + const runStore = new WorkflowRunStore({ + sessionDir: this.config.getSessionDir(parentWorkspaceId), + }); + const run = await runStore.getRun(workflowTask.runId); + if (run.workspaceId !== parentWorkspaceId) { + return { + ownerTaskId: owner.taskId, + runId: workflowTask.runId, + status: run.status, + reason: `workflow run belongs to ${run.workspaceId}, not ${parentWorkspaceId}`, + }; + } + if (isActiveWorkflowRunStatus(run.status)) { + return null; + } + return { + ownerTaskId: owner.taskId, + runId: workflowTask.runId, + status: run.status, + reason: `workflow run is ${run.status}`, + }; + } catch (error: unknown) { + return { + ownerTaskId: owner.taskId, + runId: workflowTask.runId, + reason: `workflow run is unavailable: ${getErrorMessage(error)}`, + }; + } + } + + private async interruptTaskRecoveryForInactiveWorkflowOwner( + taskId: string, + config: ProjectsConfig, + trigger: string + ): Promise { + assert(taskId.length > 0, "interruptTaskRecoveryForInactiveWorkflowOwner requires taskId"); + assert(trigger.length > 0, "interruptTaskRecoveryForInactiveWorkflowOwner requires trigger"); + const inactiveOwner = await this.getInactiveWorkflowTaskOwnerForRecovery(taskId, config); + if (inactiveOwner == null) { + return false; + } + + let updated = false; + await this.editWorkspaceEntry( + taskId, + (ws) => { + if (hasCompletedAgentReport(ws)) { + return; + } + const previousStatus = ws.taskStatus; + const persistedQueuedPrompt = coerceNonEmptyString(ws.taskPrompt); + ws.taskStatus = "interrupted"; + ws.reportedAt = undefined; + if (previousStatus !== "queued" && !persistedQueuedPrompt) { + ws.taskPrompt = undefined; + } + updated = true; + }, + { allowMissing: true } + ); + + log.debug("Skipping workflow-owned task recovery after inactive workflow owner", { + taskId, + trigger, + ownerTaskId: inactiveOwner.ownerTaskId, + workflowRunId: inactiveOwner.runId, + workflowRunStatus: inactiveOwner.status, + reason: inactiveOwner.reason, + }); + if (updated) { + this.rejectWaiters(taskId, new Error("Task interrupted")); + await this.emitWorkspaceMetadata(taskId); + } + return true; + } + private markTaskQueueBackgrounded(taskId: string): void { this.userBackgroundedTaskIds.add(taskId); } @@ -1422,6 +1554,16 @@ export class TaskService { for (const task of awaitingReportTasks) { if (!task.id) continue; + if ( + await this.interruptTaskRecoveryForInactiveWorkflowOwner( + task.id, + config, + "startup-awaiting-report" + ) + ) { + continue; + } + // Avoid resuming a task while it still has active descendants (it shouldn't report yet). const hasActiveDescendants = this.hasActiveDescendantAgentTasks(config, task.id); if (hasActiveDescendants) { @@ -1446,6 +1588,12 @@ export class TaskService { for (const task of runningTasks) { if (!task.id) continue; + if ( + await this.interruptTaskRecoveryForInactiveWorkflowOwner(task.id, config, "startup-running") + ) { + continue; + } + // Best-effort: if mux restarted mid-stream, nudge the agent to continue and report. // Only do this when the task has no running descendants, to avoid duplicate spawns. const hasActiveDescendants = this.hasActiveDescendantAgentTasks(config, task.id); @@ -4301,6 +4449,12 @@ export class TaskService { } const taskId = task.id; assert(taskId != null && taskId.length > 0, "queued task id is required"); + if ( + await this.interruptTaskRecoveryForInactiveWorkflowOwner(taskId, config, "queued-launch") + ) { + continue; + } + if (this.aiService.isStreaming(taskId)) { await this.setTaskStatus(taskId, "running"); reservedSlots += 1; @@ -4649,6 +4803,15 @@ export class TaskService { if (entry.workspace.taskStatus !== "awaiting_report") { return false; } + if ( + await this.interruptTaskRecoveryForInactiveWorkflowOwner( + workspaceId, + cfg, + `completion-tool-${options?.reason ?? "unknown"}` + ) + ) { + return false; + } if (this.hasActiveDescendantAgentTasks(cfg, workspaceId)) { return false; } @@ -4998,6 +5161,13 @@ export class TaskService { return; } + if ( + reportArgs == null && + (await this.interruptTaskRecoveryForInactiveWorkflowOwner(workspaceId, cfg, "stream-end")) + ) { + return; + } + const isPlanLike = await this.isPlanLikeTaskWorkspace(entry); // Never allow a task to finish/report while it still has active descendant tasks. @@ -5078,6 +5248,12 @@ export class TaskService { return; } + if ( + await this.interruptTaskRecoveryForInactiveWorkflowOwner(workspaceId, cfg, "stream-error") + ) { + return; + } + if (this.hasActiveDescendantAgentTasks(cfg, workspaceId)) { return; } From e0ab3551b352d2aec134d484d02f6274d3ec3888 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Thu, 18 Jun 2026 15:01:46 +0000 Subject: [PATCH 2/7] =?UTF-8?q?=F0=9F=A4=96=20refactor:=20simplify=20workf?= =?UTF-8?q?low=20task=20recovery=20gating?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Thomas Kosiewski --- src/node/services/taskService.ts | 188 +++++++++++++++++++------------ 1 file changed, 114 insertions(+), 74 deletions(-) diff --git a/src/node/services/taskService.ts b/src/node/services/taskService.ts index 7e838009ab..462ca4231c 100644 --- a/src/node/services/taskService.ts +++ b/src/node/services/taskService.ts @@ -464,6 +464,14 @@ interface AgentTaskIndex { parentById: Map; } +type WorkflowTaskConfig = NonNullable; + +interface WorkflowTaskOwner { + taskId: string; + workspace: AgentTaskWorkspaceEntry; + workflowTask: WorkflowTaskConfig; +} + interface InactiveWorkflowTaskOwner { ownerTaskId: string; runId: string; @@ -471,6 +479,8 @@ interface InactiveWorkflowTaskOwner { reason: string; } +type InterruptedTaskStatusMutation = "interrupted" | "preserved-completed-report"; + interface PendingTaskWaiter { taskId: string; resolve: (report: { reportMarkdown: string; title?: string; structuredOutput?: unknown }) => void; @@ -1046,7 +1056,7 @@ export class TaskService { private findWorkflowTaskOwnerInAncestry( index: AgentTaskIndex, taskId: string - ): { taskId: string; workspace: AgentTaskWorkspaceEntry } | null { + ): WorkflowTaskOwner | null { assert(taskId.length > 0, "findWorkflowTaskOwnerInAncestry requires taskId"); let current: string | undefined = taskId; for (let depth = 0; current != null; depth += 1) { @@ -1055,8 +1065,13 @@ export class TaskService { `findWorkflowTaskOwnerInAncestry: possible parentWorkspaceId cycle starting at ${taskId}` ); const entry = index.byId.get(current); - if (entry?.workflowTask != null) { - return { taskId: current, workspace: entry }; + if (entry == null) { + current = index.parentById.get(current); + continue; + } + const workflowTask = entry.workflowTask; + if (workflowTask != null) { + return { taskId: current, workspace: entry, workflowTask }; } current = index.parentById.get(current); } @@ -1065,16 +1080,19 @@ export class TaskService { private async getInactiveWorkflowTaskOwnerForRecovery( taskId: string, - config: ProjectsConfig + config: ProjectsConfig, + index?: AgentTaskIndex ): Promise { assert(taskId.length > 0, "getInactiveWorkflowTaskOwnerForRecovery requires taskId"); - const owner = this.findWorkflowTaskOwnerInAncestry(this.buildAgentTaskIndex(config), taskId); + const owner = this.findWorkflowTaskOwnerInAncestry( + index ?? this.buildAgentTaskIndex(config), + taskId + ); if (owner == null) { return null; } - const workflowTask = owner.workspace.workflowTask; - assert(workflowTask != null, "workflow task owner must include workflowTask metadata"); + const workflowTask = owner.workflowTask; const parentWorkspaceId = coerceNonEmptyString(owner.workspace.parentWorkspaceId); if (!parentWorkspaceId) { return { @@ -1115,33 +1133,46 @@ export class TaskService { } } + private applyInterruptedTaskStatus( + workspace: WorkspaceConfigEntry + ): InterruptedTaskStatusMutation { + if (hasCompletedAgentReport(workspace)) { + // Preserve completed report evidence so already-finished tasks stay inspectable + // and collapse-eligible after a later interrupt/recovery pass. + return "preserved-completed-report"; + } + + const previousStatus = workspace.taskStatus; + const persistedQueuedPrompt = coerceNonEmptyString(workspace.taskPrompt); + workspace.taskStatus = "interrupted"; + workspace.reportedAt = undefined; + + // Queued tasks persist their initial prompt in config until first start. Preserve that + // intent across interrupts, including repeated interrupts after the status is no longer queued. + if (previousStatus !== "queued" && !persistedQueuedPrompt) { + workspace.taskPrompt = undefined; + } + return "interrupted"; + } + private async interruptTaskRecoveryForInactiveWorkflowOwner( taskId: string, config: ProjectsConfig, - trigger: string + trigger: string, + index?: AgentTaskIndex ): Promise { assert(taskId.length > 0, "interruptTaskRecoveryForInactiveWorkflowOwner requires taskId"); assert(trigger.length > 0, "interruptTaskRecoveryForInactiveWorkflowOwner requires trigger"); - const inactiveOwner = await this.getInactiveWorkflowTaskOwnerForRecovery(taskId, config); + const inactiveOwner = await this.getInactiveWorkflowTaskOwnerForRecovery(taskId, config, index); if (inactiveOwner == null) { return false; } - let updated = false; + let interrupted = false; await this.editWorkspaceEntry( taskId, (ws) => { - if (hasCompletedAgentReport(ws)) { - return; - } - const previousStatus = ws.taskStatus; - const persistedQueuedPrompt = coerceNonEmptyString(ws.taskPrompt); - ws.taskStatus = "interrupted"; - ws.reportedAt = undefined; - if (previousStatus !== "queued" && !persistedQueuedPrompt) { - ws.taskPrompt = undefined; - } - updated = true; + interrupted = this.applyInterruptedTaskStatus(ws) === "interrupted"; }, { allowMissing: true } ); @@ -1154,7 +1185,7 @@ export class TaskService { workflowRunStatus: inactiveOwner.status, reason: inactiveOwner.reason, }); - if (updated) { + if (interrupted) { this.rejectWaiters(taskId, new Error("Task interrupted")); await this.emitWorkspaceMetadata(taskId); } @@ -1540,6 +1571,7 @@ export class TaskService { const maybeStartQueuedTasksMs = Date.now() - maybeStartQueuedTasksStartedAt; const config = this.config.loadConfigOrDefault(); + const taskIndex = this.buildAgentTaskIndex(config); const awaitingReportTasks = this.listAgentTaskWorkspaces(config).filter( (t) => t.taskStatus === "awaiting_report" ); @@ -1558,14 +1590,15 @@ export class TaskService { await this.interruptTaskRecoveryForInactiveWorkflowOwner( task.id, config, - "startup-awaiting-report" + "startup-awaiting-report", + taskIndex ) ) { continue; } // Avoid resuming a task while it still has active descendants (it shouldn't report yet). - const hasActiveDescendants = this.hasActiveDescendantAgentTasks(config, task.id); + const hasActiveDescendants = this.hasActiveDescendantAgentTasksUsingIndex(taskIndex, task.id); if (hasActiveDescendants) { skippedAwaitingReportDueToActiveDescendants += 1; continue; @@ -1589,14 +1622,19 @@ export class TaskService { for (const task of runningTasks) { if (!task.id) continue; if ( - await this.interruptTaskRecoveryForInactiveWorkflowOwner(task.id, config, "startup-running") + await this.interruptTaskRecoveryForInactiveWorkflowOwner( + task.id, + config, + "startup-running", + taskIndex + ) ) { continue; } // Best-effort: if mux restarted mid-stream, nudge the agent to continue and report. // Only do this when the task has no running descendants, to avoid duplicate spawns. - const hasActiveDescendants = this.hasActiveDescendantAgentTasks(config, task.id); + const hasActiveDescendants = this.hasActiveDescendantAgentTasksUsingIndex(taskIndex, task.id); if (hasActiveDescendants) { skippedRunningDueToActiveDescendants += 1; continue; @@ -3203,28 +3241,8 @@ export class TaskService { const updated = await this.editWorkspaceEntry( id, (ws) => { - if (hasCompletedAgentReport(ws)) { - // Preserve completed report evidence so already-finished descendants stay - // collapse-eligible after a later parent hard interrupt. - preservedCompletedDescendant = true; - return; - } - - const previousStatus = ws.taskStatus; - const persistedQueuedPrompt = coerceNonEmptyString(ws.taskPrompt); - ws.taskStatus = "interrupted"; - ws.reportedAt = undefined; - - // Queued tasks persist their initial prompt in config until first start. - // Preserve that prompt when interrupting queued descendants so users can - // still inspect/resume the preserved workspace intent. - // - // Also preserve across repeated hard interrupts: once a never-started task - // is first interrupted, its status becomes "interrupted". Later cascades - // must not clear the same persisted prompt. - if (previousStatus !== "queued" && !persistedQueuedPrompt) { - ws.taskPrompt = undefined; - } + preservedCompletedDescendant = + this.applyInterruptedTaskStatus(ws) === "preserved-completed-report"; }, { allowMissing: true } ); @@ -4264,22 +4282,7 @@ export class TaskService { private isWorkflowOwnedTaskUsingIndex(index: AgentTaskIndex, taskId: string): boolean { assert(taskId.length > 0, "isWorkflowOwnedTaskUsingIndex: taskId must be non-empty"); - - let current: string | undefined = taskId; - for (let depth = 0; current != null && depth < 32; depth++) { - const entry = index.byId.get(current); - if (entry?.workflowTask != null) { - return true; - } - current = index.parentById.get(current); - } - - if (current != null) { - throw new Error( - `isWorkflowOwnedTaskUsingIndex: possible parentWorkspaceId cycle starting at ${taskId}` - ); - } - return false; + return this.findWorkflowTaskOwnerInAncestry(index, taskId) != null; } private countActiveAgentTasks(config: ReturnType): number { @@ -4314,9 +4317,20 @@ export class TaskService { config: ReturnType, workspaceId: string ): boolean { - assert(workspaceId.length > 0, "hasActiveDescendantAgentTasks: workspaceId must be non-empty"); + return this.hasActiveDescendantAgentTasksUsingIndex( + this.buildAgentTaskIndex(config), + workspaceId + ); + } - const index = this.buildAgentTaskIndex(config); + private hasActiveDescendantAgentTasksUsingIndex( + index: AgentTaskIndex, + workspaceId: string + ): boolean { + assert( + workspaceId.length > 0, + "hasActiveDescendantAgentTasksUsingIndex: workspaceId must be non-empty" + ); const activeStatuses = new Set([ "queued", @@ -4434,6 +4448,7 @@ export class TaskService { }); if (availableSlots === 0) return; + const taskIndex = this.buildAgentTaskIndex(config); const queuedTasks = this.listAgentTaskWorkspaces(config) .filter((task) => task.taskStatus === "queued" && typeof task.id === "string") .sort((a, b) => { @@ -4450,7 +4465,12 @@ export class TaskService { const taskId = task.id; assert(taskId != null && taskId.length > 0, "queued task id is required"); if ( - await this.interruptTaskRecoveryForInactiveWorkflowOwner(taskId, config, "queued-launch") + await this.interruptTaskRecoveryForInactiveWorkflowOwner( + taskId, + config, + "queued-launch", + taskIndex + ) ) { continue; } @@ -4803,16 +4823,18 @@ export class TaskService { if (entry.workspace.taskStatus !== "awaiting_report") { return false; } + const taskIndex = this.buildAgentTaskIndex(cfg); if ( await this.interruptTaskRecoveryForInactiveWorkflowOwner( workspaceId, cfg, - `completion-tool-${options?.reason ?? "unknown"}` + `completion-tool-${options?.reason ?? "unknown"}`, + taskIndex ) ) { return false; } - if (this.hasActiveDescendantAgentTasks(cfg, workspaceId)) { + if (this.hasActiveDescendantAgentTasksUsingIndex(taskIndex, workspaceId)) { return false; } if (this.aiService.isStreaming(workspaceId)) { @@ -4952,11 +4974,15 @@ export class TaskService { const cfg = this.config.loadConfigOrDefault(); const entry = findWorkspaceEntry(cfg, workspaceId); if (!entry) return; + const taskIndex = this.buildAgentTaskIndex(cfg); // Parent workspaces must not end while they have active background tasks/workflows. // Enforce by auto-resuming the stream with a directive to await outstanding work. if (!entry.workspace.parentWorkspaceId) { - const hasActiveDescendants = this.hasActiveDescendantAgentTasks(cfg, workspaceId); + const hasActiveDescendants = this.hasActiveDescendantAgentTasksUsingIndex( + taskIndex, + workspaceId + ); const referencedWorkflowRunIds = await this.listAgentReferencedWorkflowRunIds( workspaceId, event.parts, @@ -5163,7 +5189,12 @@ export class TaskService { if ( reportArgs == null && - (await this.interruptTaskRecoveryForInactiveWorkflowOwner(workspaceId, cfg, "stream-end")) + (await this.interruptTaskRecoveryForInactiveWorkflowOwner( + workspaceId, + cfg, + "stream-end", + taskIndex + )) ) { return; } @@ -5172,7 +5203,10 @@ export class TaskService { // Never allow a task to finish/report while it still has active descendant tasks. // We'll auto-resume this task once the last descendant reports. - const hasActiveDescendants = this.hasActiveDescendantAgentTasks(cfg, workspaceId); + const hasActiveDescendants = this.hasActiveDescendantAgentTasksUsingIndex( + taskIndex, + workspaceId + ); if (hasActiveDescendants) { if (status === "awaiting_report") { await this.setTaskStatus(workspaceId, "running"); @@ -5247,14 +5281,20 @@ export class TaskService { if (status !== "running" && status !== "awaiting_report") { return; } + const taskIndex = this.buildAgentTaskIndex(cfg); if ( - await this.interruptTaskRecoveryForInactiveWorkflowOwner(workspaceId, cfg, "stream-error") + await this.interruptTaskRecoveryForInactiveWorkflowOwner( + workspaceId, + cfg, + "stream-error", + taskIndex + ) ) { return; } - if (this.hasActiveDescendantAgentTasks(cfg, workspaceId)) { + if (this.hasActiveDescendantAgentTasksUsingIndex(taskIndex, workspaceId)) { return; } From 70a0989963f3dc278e64951d9a9e28e5212eb4cd Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Thu, 18 Jun 2026 15:13:25 +0000 Subject: [PATCH 3/7] =?UTF-8?q?=F0=9F=A4=96=20fix:=20refresh=20startup=20r?= =?UTF-8?q?ecovery=20after=20child=20interrupts?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Startup recovery now interrupts inactive workflow-owned tasks before checking whether parent tasks still have active descendants, then reloads the task index so parent awaiting_report recovery can proceed in the same startup pass.\n\n---\n\n_Generated with mux • Model: `openai:gpt-5.5` • Thinking: `xhigh` • Cost: `$17.65`_\n\n --- src/node/services/taskService.test.ts | 67 +++++++++++++++++++++++++++ src/node/services/taskService.ts | 33 +++++++++++-- 2 files changed, 96 insertions(+), 4 deletions(-) diff --git a/src/node/services/taskService.test.ts b/src/node/services/taskService.test.ts index 9334250d7f..57f8400706 100644 --- a/src/node/services/taskService.test.ts +++ b/src/node/services/taskService.test.ts @@ -5689,6 +5689,73 @@ describe("TaskService", () => { expect(findWorkspaceInConfig(config, queuedChildId)?.taskPrompt).toBe("queued work"); }); + test("initialize recovers parent tasks after interrupting inactive workflow children", async () => { + const config = await createTestConfig(rootDir); + + const projectPath = path.join(rootDir, "repo"); + const rootId = "root-workflow-child-interrupted"; + const parentTaskId = "parent-awaiting-after-child-interrupt"; + const childTaskId = "child-running-inactive-workflow"; + const workflowRunId = "wfr_child_inactive_owner"; + + await saveWorkspaces( + config, + projectPath, + [ + projectWorkspace(projectPath, "root", rootId), + projectWorkspace(projectPath, "parent", parentTaskId, { + name: "agent_explore_parent", + parentWorkspaceId: rootId, + agentId: "explore", + agentType: "explore", + taskStatus: "awaiting_report", + taskModelString: defaultModel, + }), + projectWorkspace(projectPath, "child", childTaskId, { + name: "agent_explore_child", + parentWorkspaceId: parentTaskId, + agentId: "explore", + agentType: "explore", + taskStatus: "running", + taskModelString: defaultModel, + workflowTask: { runId: workflowRunId, stepId: "child" }, + }), + ], + testTaskSettings(10, 3) + ); + const runStore = new WorkflowRunStore({ sessionDir: config.getSessionDir(parentTaskId) }); + await runStore.createRun({ + id: workflowRunId, + workspaceId: parentTaskId, + definition: { + name: "interrupted", + description: "Interrupted", + scope: "built-in", + executable: true, + }, + definitionSource: "export default function workflow() { return {}; }\n", + args: {}, + now: "2026-05-29T00:00:00.000Z", + }); + await runStore.appendStatus(workflowRunId, "interrupted", "2026-05-29T00:00:01.000Z"); + + const { aiService } = createAIServiceMocks(config, { isStreaming: mock(() => false) }); + const { workspaceService, sendMessage } = createWorkspaceServiceMocks(); + const { taskService } = createTaskServiceHarness(config, { aiService, workspaceService }); + + await taskService.initialize(); + + expect(findWorkspaceInConfig(config, childTaskId)?.taskStatus).toBe("interrupted"); + expect(sendMessage).toHaveBeenCalledWith( + parentTaskId, + expect.stringContaining("awaiting its final agent_report"), + expect.objectContaining({ + toolPolicy: [{ regex_match: "^agent_report$", action: "require" }], + }), + expect.objectContaining({ synthetic: true }) + ); + }); + test("initialize resumes awaiting_report tasks after restart", async () => { const config = await createTestConfig(rootDir); diff --git a/src/node/services/taskService.ts b/src/node/services/taskService.ts index 462ca4231c..6dfc657a8c 100644 --- a/src/node/services/taskService.ts +++ b/src/node/services/taskService.ts @@ -1570,15 +1570,40 @@ export class TaskService { await this.maybeStartQueuedTasks(); const maybeStartQueuedTasksMs = Date.now() - maybeStartQueuedTasksStartedAt; - const config = this.config.loadConfigOrDefault(); - const taskIndex = this.buildAgentTaskIndex(config); - const awaitingReportTasks = this.listAgentTaskWorkspaces(config).filter( + let config = this.config.loadConfigOrDefault(); + let taskIndex = this.buildAgentTaskIndex(config); + let awaitingReportTasks = this.listAgentTaskWorkspaces(config).filter( (t) => t.taskStatus === "awaiting_report" ); - const runningTasks = this.listAgentTaskWorkspaces(config).filter( + let runningTasks = this.listAgentTaskWorkspaces(config).filter( (t) => t.taskStatus === "running" ); + let interruptedInactiveWorkflowOwnerAtStartup = false; + for (const task of [...awaitingReportTasks, ...runningTasks]) { + if (!task.id) continue; + if ( + await this.interruptTaskRecoveryForInactiveWorkflowOwner( + task.id, + config, + "startup-inactive-workflow-owner-prepass", + taskIndex + ) + ) { + interruptedInactiveWorkflowOwnerAtStartup = true; + } + } + if (interruptedInactiveWorkflowOwnerAtStartup) { + // Refresh before descendant checks so a parent awaiting_report task does not stay + // blocked by a child that this startup pass just interrupted. + config = this.config.loadConfigOrDefault(); + taskIndex = this.buildAgentTaskIndex(config); + awaitingReportTasks = this.listAgentTaskWorkspaces(config).filter( + (t) => t.taskStatus === "awaiting_report" + ); + runningTasks = this.listAgentTaskWorkspaces(config).filter((t) => t.taskStatus === "running"); + } + let resumedAwaitingReportCount = 0; let skippedAwaitingReportDueToActiveDescendants = 0; let failedAwaitingReportCount = 0; From 5279449f86501951411270de6624df89cd3de14f Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Thu, 18 Jun 2026 15:24:01 +0000 Subject: [PATCH 4/7] =?UTF-8?q?=F0=9F=A4=96=20fix:=20drain=20queue=20after?= =?UTF-8?q?=20startup=20workflow=20interrupts?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When startup recovery interrupts inactive workflow-owned tasks, run the queue scheduler again after recovery prompts so newly freed task slots can start unrelated queued work immediately.\n\n---\n\n_Generated with mux • Model: `openai:gpt-5.5` • Thinking: `xhigh` • Cost: `$17.65`_\n\n --- src/node/services/taskService.test.ts | 86 +++++++++++++++++++++++++++ src/node/services/taskService.ts | 7 +++ 2 files changed, 93 insertions(+) diff --git a/src/node/services/taskService.test.ts b/src/node/services/taskService.test.ts index 57f8400706..f0bfa8999d 100644 --- a/src/node/services/taskService.test.ts +++ b/src/node/services/taskService.test.ts @@ -5756,6 +5756,92 @@ describe("TaskService", () => { ); }); + test("initialize drains queued tasks after interrupting inactive workflow children", async () => { + const config = await createTestConfig(rootDir); + + const projectPath = await createTestProject(rootDir, "repo-queue-after-interrupt"); + const runtimeConfig = { type: "worktree" as const, srcBaseDir: config.srcDir }; + const runtime = createRuntime(runtimeConfig, { projectPath }); + const rootName = "root"; + await runtime.createWorkspace({ + projectPath, + branchName: rootName, + trunkBranch: "main", + directoryName: rootName, + initLogger: createNullInitLogger(), + }); + + const rootId = "root-queue-after-interrupt"; + const runningTaskId = "running-inactive-workflow-occupies-slot"; + const queuedTaskId = "queued-starts-after-interrupt"; + const workflowRunId = "wfr_queue_after_interrupt"; + + await saveWorkspaces( + config, + projectPath, + [ + { + path: runtime.getWorkspacePath(projectPath, rootName), + id: rootId, + name: rootName, + createdAt: new Date().toISOString(), + runtimeConfig, + }, + projectWorkspace(projectPath, "running", runningTaskId, { + name: "agent_explore_running", + parentWorkspaceId: rootId, + agentId: "explore", + agentType: "explore", + taskStatus: "running", + taskModelString: defaultModel, + runtimeConfig, + workflowTask: { runId: workflowRunId, stepId: "running" }, + }), + projectWorkspace(projectPath, "queued", queuedTaskId, { + name: "agent_explore_queued", + parentWorkspaceId: rootId, + agentId: "explore", + agentType: "explore", + taskStatus: "queued", + taskPrompt: "queued work", + taskModelString: defaultModel, + runtimeConfig, + }), + ], + testTaskSettings(1, 3) + ); + const runStore = new WorkflowRunStore({ sessionDir: config.getSessionDir(rootId) }); + await runStore.createRun({ + id: workflowRunId, + workspaceId: rootId, + definition: { + name: "interrupted", + description: "Interrupted", + scope: "built-in", + executable: true, + }, + definitionSource: "export default function workflow() { return {}; }\n", + args: {}, + now: "2026-05-29T00:00:00.000Z", + }); + await runStore.appendStatus(workflowRunId, "interrupted", "2026-05-29T00:00:01.000Z"); + + const { aiService } = createAIServiceMocks(config, { isStreaming: mock(() => false) }); + const { workspaceService, sendMessage } = createWorkspaceServiceMocks(); + const { taskService } = createTaskServiceHarness(config, { aiService, workspaceService }); + + await taskService.initialize(); + + expect(findWorkspaceInConfig(config, runningTaskId)?.taskStatus).toBe("interrupted"); + expect(findWorkspaceInConfig(config, queuedTaskId)?.taskStatus).toBe("running"); + expect(sendMessage).toHaveBeenCalledWith( + queuedTaskId, + "queued work", + expect.objectContaining({ agentId: "explore" }), + expect.objectContaining({ allowQueuedAgentTask: true }) + ); + }); + test("initialize resumes awaiting_report tasks after restart", async () => { const config = await createTestConfig(rootDir); diff --git a/src/node/services/taskService.ts b/src/node/services/taskService.ts index 6dfc657a8c..c8ba12e3c8 100644 --- a/src/node/services/taskService.ts +++ b/src/node/services/taskService.ts @@ -1726,6 +1726,13 @@ export class TaskService { }); } + if (interruptedInactiveWorkflowOwnerAtStartup) { + // Startup queue draining already ran before these interruptions freed slots. + // Run it once more after recovery prompts so unrelated queued work is not stranded. + await this.maybeStartQueuedTasks(); + config = this.config.loadConfigOrDefault(); + } + // Restart-safety for git patch artifacts: // - If mux crashed mid-generation, patch artifacts can be left "pending". // - Completed tasks can be stranded in config until cleanup runs again, so restart should From aa98ca60a81c84762499a7167134201883438ac3 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Thu, 18 Jun 2026 15:32:26 +0000 Subject: [PATCH 5/7] =?UTF-8?q?=F0=9F=A4=96=20fix:=20interrupt=20queued=20?= =?UTF-8?q?workflow=20tasks=20before=20slot=20checks?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Queued workflow-owned tasks from inactive runs are now interrupted before capacity gating, so abandoned workflow steps do not remain live when all parallel task slots are occupied.\n\n---\n\n_Generated with mux • Model: `openai:gpt-5.5` • Thinking: `xhigh` • Cost: `$17.65`_\n\n --- src/node/services/taskService.test.ts | 13 ++++++++ src/node/services/taskService.ts | 43 ++++++++++++++++++++------- 2 files changed, 46 insertions(+), 10 deletions(-) diff --git a/src/node/services/taskService.test.ts b/src/node/services/taskService.test.ts index f0bfa8999d..a26bf80113 100644 --- a/src/node/services/taskService.test.ts +++ b/src/node/services/taskService.test.ts @@ -5773,6 +5773,7 @@ describe("TaskService", () => { const rootId = "root-queue-after-interrupt"; const runningTaskId = "running-inactive-workflow-occupies-slot"; + const queuedWorkflowTaskId = "queued-inactive-workflow"; const queuedTaskId = "queued-starts-after-interrupt"; const workflowRunId = "wfr_queue_after_interrupt"; @@ -5797,6 +5798,17 @@ describe("TaskService", () => { runtimeConfig, workflowTask: { runId: workflowRunId, stepId: "running" }, }), + projectWorkspace(projectPath, "queued-workflow", queuedWorkflowTaskId, { + name: "agent_explore_queued_workflow", + parentWorkspaceId: rootId, + agentId: "explore", + agentType: "explore", + taskStatus: "queued", + taskPrompt: "abandoned workflow queued work", + taskModelString: defaultModel, + runtimeConfig, + workflowTask: { runId: workflowRunId, stepId: "queued" }, + }), projectWorkspace(projectPath, "queued", queuedTaskId, { name: "agent_explore_queued", parentWorkspaceId: rootId, @@ -5833,6 +5845,7 @@ describe("TaskService", () => { await taskService.initialize(); expect(findWorkspaceInConfig(config, runningTaskId)?.taskStatus).toBe("interrupted"); + expect(findWorkspaceInConfig(config, queuedWorkflowTaskId)?.taskStatus).toBe("interrupted"); expect(findWorkspaceInConfig(config, queuedTaskId)?.taskStatus).toBe("running"); expect(sendMessage).toHaveBeenCalledWith( queuedTaskId, diff --git a/src/node/services/taskService.ts b/src/node/services/taskService.ts index c8ba12e3c8..4b414ac564 100644 --- a/src/node/services/taskService.ts +++ b/src/node/services/taskService.ts @@ -4468,8 +4468,40 @@ export class TaskService { { await using _lock = await this.mutex.acquire(); - const config = this.config.loadConfigOrDefault(); + let config = this.config.loadConfigOrDefault(); const taskSettings: TaskSettings = config.taskSettings ?? DEFAULT_TASK_SETTINGS; + const listQueuedTasks = (sourceConfig: ProjectsConfig): AgentTaskWorkspaceEntry[] => + this.listAgentTaskWorkspaces(sourceConfig) + .filter((task) => task.taskStatus === "queued" && typeof task.id === "string") + .sort((a, b) => { + const aTime = a.createdAt ? Date.parse(a.createdAt) : 0; + const bTime = b.createdAt ? Date.parse(b.createdAt) : 0; + return aTime - bTime; + }); + let taskIndex = this.buildAgentTaskIndex(config); + let queuedTasks = listQueuedTasks(config); + + let interruptedInactiveWorkflowQueuedTask = false; + for (const task of queuedTasks) { + const taskId = task.id; + assert(taskId != null && taskId.length > 0, "queued task id is required"); + if ( + await this.interruptTaskRecoveryForInactiveWorkflowOwner( + taskId, + config, + "queued-inactive-workflow-owner-prepass", + taskIndex + ) + ) { + interruptedInactiveWorkflowQueuedTask = true; + } + } + if (interruptedInactiveWorkflowQueuedTask) { + config = this.config.loadConfigOrDefault(); + taskIndex = this.buildAgentTaskIndex(config); + queuedTasks = listQueuedTasks(config); + } + const availableSlots = Math.max( 0, taskSettings.maxParallelAgentTasks - this.countActiveAgentTasks(config) @@ -4480,15 +4512,6 @@ export class TaskService { }); if (availableSlots === 0) return; - const taskIndex = this.buildAgentTaskIndex(config); - const queuedTasks = this.listAgentTaskWorkspaces(config) - .filter((task) => task.taskStatus === "queued" && typeof task.id === "string") - .sort((a, b) => { - const aTime = a.createdAt ? Date.parse(a.createdAt) : 0; - const bTime = b.createdAt ? Date.parse(b.createdAt) : 0; - return aTime - bTime; - }); - let reservedSlots = 0; for (const task of queuedTasks) { if (reservedSlots >= availableSlots) { From 2f1c0627d7c3a7debf4b4221648907515fcffb51 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Thu, 18 Jun 2026 15:41:20 +0000 Subject: [PATCH 6/7] =?UTF-8?q?=F0=9F=A4=96=20fix:=20schedule=20queue=20dr?= =?UTF-8?q?ain=20after=20workflow=20recovery=20interrupts?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Inactive workflow-owner interruptions now schedule queued-task draining outside the scheduler path so freed task slots do not leave unrelated queued work stranded. Scheduler callers opt out to avoid recursive queue scans.\n\n---\n\n_Generated with mux • Model: `openai:gpt-5.5` • Thinking: `xhigh` • Cost: `$17.65`_\n\n --- src/node/services/taskService.test.ts | 5 +++++ src/node/services/taskService.ts | 15 +++++++++++---- 2 files changed, 16 insertions(+), 4 deletions(-) diff --git a/src/node/services/taskService.test.ts b/src/node/services/taskService.test.ts index a26bf80113..20456a9ff1 100644 --- a/src/node/services/taskService.test.ts +++ b/src/node/services/taskService.test.ts @@ -9027,6 +9027,10 @@ describe("TaskService", () => { const { workspaceService, sendMessage } = createWorkspaceServiceMocks(); const { taskService } = createTaskServiceHarness(config, { workspaceService }); + const maybeStartQueuedTasks = spyOn( + taskService as unknown as { maybeStartQueuedTasks: () => Promise }, + "maybeStartQueuedTasks" + ).mockResolvedValue(undefined); await handleTaskServiceStreamEndForTest(taskService, { type: "stream-end", @@ -9038,6 +9042,7 @@ describe("TaskService", () => { expect(sendMessage).not.toHaveBeenCalled(); expect(findWorkspaceInConfig(config, childId)?.taskStatus).toBe("interrupted"); + expect(maybeStartQueuedTasks).toHaveBeenCalledTimes(1); }); test("non-plan subagent stream-end with final assistant text finalizes an implicit report", async () => { diff --git a/src/node/services/taskService.ts b/src/node/services/taskService.ts index 4b414ac564..4fc254c359 100644 --- a/src/node/services/taskService.ts +++ b/src/node/services/taskService.ts @@ -1159,7 +1159,8 @@ export class TaskService { taskId: string, config: ProjectsConfig, trigger: string, - index?: AgentTaskIndex + index?: AgentTaskIndex, + options?: { scheduleQueueDrain?: boolean } ): Promise { assert(taskId.length > 0, "interruptTaskRecoveryForInactiveWorkflowOwner requires taskId"); assert(trigger.length > 0, "interruptTaskRecoveryForInactiveWorkflowOwner requires trigger"); @@ -1188,6 +1189,9 @@ export class TaskService { if (interrupted) { this.rejectWaiters(taskId, new Error("Task interrupted")); await this.emitWorkspaceMetadata(taskId); + if (options?.scheduleQueueDrain !== false) { + this.scheduleMaybeStartQueuedTasks(); + } } return true; } @@ -1587,7 +1591,8 @@ export class TaskService { task.id, config, "startup-inactive-workflow-owner-prepass", - taskIndex + taskIndex, + { scheduleQueueDrain: false } ) ) { interruptedInactiveWorkflowOwnerAtStartup = true; @@ -4490,7 +4495,8 @@ export class TaskService { taskId, config, "queued-inactive-workflow-owner-prepass", - taskIndex + taskIndex, + { scheduleQueueDrain: false } ) ) { interruptedInactiveWorkflowQueuedTask = true; @@ -4524,7 +4530,8 @@ export class TaskService { taskId, config, "queued-launch", - taskIndex + taskIndex, + { scheduleQueueDrain: false } ) ) { continue; From 14842dcce6ccc2bcb6bb512c8fd2664bd5dcf10b Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Thu, 18 Jun 2026 15:51:41 +0000 Subject: [PATCH 7/7] =?UTF-8?q?=F0=9F=A4=96=20fix:=20honor=20inactive=20wo?= =?UTF-8?q?rkflow=20owners=20across=20ancestry?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Workflow recovery now checks every workflow owner in a task ancestry chain so an active nested workflow cannot mask an interrupted outer workflow owner.\n\n---\n\n_Generated with mux • Model: `openai:gpt-5.5` • Thinking: `xhigh` • Cost: `$17.65`_\n\n --- src/node/services/taskService.test.ts | 19 ++++++++++ src/node/services/taskService.ts | 54 +++++++++++++++++---------- 2 files changed, 54 insertions(+), 19 deletions(-) diff --git a/src/node/services/taskService.test.ts b/src/node/services/taskService.test.ts index 20456a9ff1..4ab5e32102 100644 --- a/src/node/services/taskService.test.ts +++ b/src/node/services/taskService.test.ts @@ -5610,6 +5610,7 @@ describe("TaskService", () => { const awaitingChildId = "child-workflow-awaiting"; const nestedRunningChildId = "child-workflow-nested-running"; const workflowRunId = "wfr_interrupted_owner"; + const innerWorkflowRunId = "wfr_active_inner_owner"; await saveWorkspaces( config, @@ -5655,6 +5656,7 @@ describe("TaskService", () => { taskStatus: "running", taskModelString: defaultModel, runtimeConfig, + workflowTask: { runId: innerWorkflowRunId, stepId: "nested" }, }), ], testTaskSettings(10, 3) @@ -5674,6 +5676,23 @@ describe("TaskService", () => { now: "2026-05-29T00:00:00.000Z", }); await runStore.appendStatus(workflowRunId, "interrupted", "2026-05-29T00:00:01.000Z"); + const innerRunStore = new WorkflowRunStore({ + sessionDir: config.getSessionDir(runningChildId), + }); + await innerRunStore.createRun({ + id: innerWorkflowRunId, + workspaceId: runningChildId, + definition: { + name: "inner-running", + description: "Inner running", + scope: "built-in", + executable: true, + }, + definitionSource: "export default function workflow() { return {}; }\n", + args: {}, + now: "2026-05-29T00:00:00.000Z", + }); + await innerRunStore.appendStatus(innerWorkflowRunId, "running", "2026-05-29T00:00:01.000Z"); const { aiService } = createAIServiceMocks(config, { isStreaming: mock(() => false) }); const { workspaceService, sendMessage, resumeStream } = createWorkspaceServiceMocks(); diff --git a/src/node/services/taskService.ts b/src/node/services/taskService.ts index 4fc254c359..e8c92cf911 100644 --- a/src/node/services/taskService.ts +++ b/src/node/services/taskService.ts @@ -1053,16 +1053,17 @@ export class TaskService { // Workflow abort/interrupt is the source of truth: task-level restart and stream-end // recovery must not resurrect a workflow child (or its descendants) once the owning run // is no longer active. Manual workflow_resume replays from the workflow journal instead. - private findWorkflowTaskOwnerInAncestry( + private findWorkflowTaskOwnersInAncestry( index: AgentTaskIndex, taskId: string - ): WorkflowTaskOwner | null { - assert(taskId.length > 0, "findWorkflowTaskOwnerInAncestry requires taskId"); + ): WorkflowTaskOwner[] { + assert(taskId.length > 0, "findWorkflowTaskOwnersInAncestry requires taskId"); + const owners: WorkflowTaskOwner[] = []; let current: string | undefined = taskId; for (let depth = 0; current != null; depth += 1) { assert( depth < 32, - `findWorkflowTaskOwnerInAncestry: possible parentWorkspaceId cycle starting at ${taskId}` + `findWorkflowTaskOwnersInAncestry: possible parentWorkspaceId cycle starting at ${taskId}` ); const entry = index.byId.get(current); if (entry == null) { @@ -1071,27 +1072,23 @@ export class TaskService { } const workflowTask = entry.workflowTask; if (workflowTask != null) { - return { taskId: current, workspace: entry, workflowTask }; + owners.push({ taskId: current, workspace: entry, workflowTask }); } current = index.parentById.get(current); } - return null; + return owners; } - private async getInactiveWorkflowTaskOwnerForRecovery( - taskId: string, - config: ProjectsConfig, - index?: AgentTaskIndex - ): Promise { - assert(taskId.length > 0, "getInactiveWorkflowTaskOwnerForRecovery requires taskId"); - const owner = this.findWorkflowTaskOwnerInAncestry( - index ?? this.buildAgentTaskIndex(config), - taskId - ); - if (owner == null) { - return null; - } + private findWorkflowTaskOwnerInAncestry( + index: AgentTaskIndex, + taskId: string + ): WorkflowTaskOwner | null { + return this.findWorkflowTaskOwnersInAncestry(index, taskId)[0] ?? null; + } + private async getInactiveWorkflowTaskOwner( + owner: WorkflowTaskOwner + ): Promise { const workflowTask = owner.workflowTask; const parentWorkspaceId = coerceNonEmptyString(owner.workspace.parentWorkspaceId); if (!parentWorkspaceId) { @@ -1133,6 +1130,25 @@ export class TaskService { } } + private async getInactiveWorkflowTaskOwnerForRecovery( + taskId: string, + config: ProjectsConfig, + index?: AgentTaskIndex + ): Promise { + assert(taskId.length > 0, "getInactiveWorkflowTaskOwnerForRecovery requires taskId"); + const owners = this.findWorkflowTaskOwnersInAncestry( + index ?? this.buildAgentTaskIndex(config), + taskId + ); + for (const owner of owners) { + const inactiveOwner = await this.getInactiveWorkflowTaskOwner(owner); + if (inactiveOwner != null) { + return inactiveOwner; + } + } + return null; + } + private applyInterruptedTaskStatus( workspace: WorkspaceConfigEntry ): InterruptedTaskStatusMutation {