diff --git a/src/node/services/taskService.test.ts b/src/node/services/taskService.test.ts index 1e333a2ad5..80ba57c2ef 100644 --- a/src/node/services/taskService.test.ts +++ b/src/node/services/taskService.test.ts @@ -1409,6 +1409,291 @@ describe("TaskService", () => { expect(recovered?.reportMarkdown).toBeUndefined(); }); + test("workspace-turn stale recovery repairs restart-interrupted deferred handles after descendants stop blocking", async () => { + const { config, parentId, projectPath, taskService, historyService, workspaceMocks } = + await startWorkspaceTurnForTest({ disposable: true }); + await config.editConfig((cfg) => { + const project = Array.from(cfg.projects.values())[0]; + assert(project, "test project must exist"); + project.workspaces.push({ + path: path.join(projectPath, "descendant-task"), + id: "descendant-task", + name: "descendant-task", + createdAt: "2026-06-19T00:00:00.000Z", + runtimeConfig: { type: "local" }, + parentWorkspaceId: "childworkspace", + taskStatus: "running", + }); + return cfg; + }); + const muxMetadata = { + type: "workspace-turn-task" as const, + taskHandleId: "wst_handle", + ownerWorkspaceId: parentId, + turnId: "turn", + }; + const appendResult = await historyService.appendToHistory( + "childworkspace", + createMuxMessage("msg_prehandoff", "assistant", "Recovered final text", { + model: "anthropic:claude-opus-4-6", + agentId: "exec", + finishReason: "stop", + muxMetadata, + }) + ); + expect(appendResult.success).toBe(true); + const internal = taskService as unknown as { + activeWorkspaceTurnHandleByWorkspaceId: Map< + string, + { handleId: string; ownerWorkspaceId: string } + >; + handleStreamEnd: (event: StreamEndEvent) => Promise; + }; + + await internal.handleStreamEnd({ + type: "stream-end", + workspaceId: "childworkspace", + messageId: "msg_prehandoff", + metadata: { + model: "anthropic:claude-opus-4-6", + agentId: "exec", + finishReason: "stop", + muxMetadata, + }, + parts: [{ type: "text", text: "Recovered final text" }], + }); + internal.activeWorkspaceTurnHandleByWorkspaceId.clear(); + expect(await taskService.getWorkspaceTurnSnapshot(parentId, "wst_handle")).toMatchObject({ + status: "interrupted", + error: "Workspace turn interrupted after restart", + }); + + await config.editConfig((cfg) => { + const descendant = Array.from(cfg.projects.values()) + .flatMap((project) => project.workspaces) + .find((workspace) => workspace.id === "descendant-task"); + assert(descendant, "descendant task must exist"); + descendant.archivedAt = "2026-06-19T00:01:00.000Z"; + return cfg; + }); + + const repaired = await taskService.getWorkspaceTurnSnapshot(parentId, "wst_handle"); + expect(repaired).toMatchObject({ + status: "completed", + messageId: "msg_prehandoff", + reportMarkdown: "Recovered final text", + }); + expect(repaired?.error).toBeUndefined(); + expect(workspaceMocks.remove).toHaveBeenCalledWith("childworkspace", true); + }); + + test("workspace-turn stale recovery repairs restart-interrupted deferred error handles", async () => { + const { config, parentId, projectPath, taskService, historyService } = + await startWorkspaceTurnForTest(); + await config.editConfig((cfg) => { + const project = Array.from(cfg.projects.values())[0]; + assert(project, "test project must exist"); + project.workspaces.push({ + path: path.join(projectPath, "descendant-task"), + id: "descendant-task", + name: "descendant-task", + createdAt: "2026-06-19T00:00:00.000Z", + runtimeConfig: { type: "local" }, + parentWorkspaceId: "childworkspace", + taskStatus: "running", + archivedAt: "2026-06-19T00:01:00.000Z", + }); + return cfg; + }); + const muxMetadata = { + type: "workspace-turn-task" as const, + taskHandleId: "wst_handle", + ownerWorkspaceId: parentId, + turnId: "turn", + }; + const appendResult = await historyService.appendToHistory( + "childworkspace", + createMuxMessage("msg_truncated", "assistant", "Partial text", { + model: "anthropic:claude-opus-4-6", + agentId: "exec", + finishReason: "length", + muxMetadata, + }) + ); + expect(appendResult.success).toBe(true); + + const internal = taskService as unknown as { + activeWorkspaceTurnHandleByWorkspaceId: Map< + string, + { handleId: string; ownerWorkspaceId: string } + >; + }; + internal.activeWorkspaceTurnHandleByWorkspaceId.clear(); + await new TaskHandleStore(config).upsertWorkspaceTurn({ + kind: "workspace_turn", + handleId: "wst_handle", + ownerWorkspaceId: parentId, + workspaceId: "childworkspace", + turnId: "turn", + status: "interrupted", + createdAt: "2026-06-19T00:00:00.000Z", + updatedAt: "2026-06-19T00:00:01.000Z", + createdWorkspace: true, + disposableWorkspace: false, + deferredMessageIds: ["msg_truncated"], + error: "Workspace turn interrupted after restart", + }); + + const repaired = await taskService.getWorkspaceTurnSnapshot(parentId, "wst_handle"); + expect(repaired).toMatchObject({ + status: "error", + messageId: "msg_truncated", + error: "Workspace turn ended before completion (finishReason: length)", + }); + }); + + test("workspace-turn stale recovery uses deferred history after archived descendants stop blocking", async () => { + const { config, parentId, projectPath, taskService, historyService } = + await startWorkspaceTurnForTest(); + await config.editConfig((cfg) => { + const project = Array.from(cfg.projects.values())[0]; + assert(project, "test project must exist"); + project.workspaces.push({ + path: path.join(projectPath, "descendant-task"), + id: "descendant-task", + name: "descendant-task", + createdAt: "2026-06-19T00:00:00.000Z", + runtimeConfig: { type: "local" }, + parentWorkspaceId: "childworkspace", + taskStatus: "running", + }); + return cfg; + }); + const muxMetadata = { + type: "workspace-turn-task" as const, + taskHandleId: "wst_handle", + ownerWorkspaceId: parentId, + turnId: "turn", + }; + const appendResult = await historyService.appendToHistory( + "childworkspace", + createMuxMessage("msg_prehandoff", "assistant", "Premature final text", { + model: "anthropic:claude-opus-4-6", + agentId: "exec", + finishReason: "stop", + muxMetadata, + }) + ); + expect(appendResult.success).toBe(true); + const internal = taskService as unknown as { + handleStreamEnd: (event: StreamEndEvent) => Promise; + }; + + await internal.handleStreamEnd({ + type: "stream-end", + workspaceId: "childworkspace", + messageId: "msg_prehandoff", + metadata: { + model: "anthropic:claude-opus-4-6", + agentId: "exec", + finishReason: "stop", + muxMetadata, + }, + parts: [{ type: "text", text: "Premature final text" }], + }); + + expect(await taskService.getWorkspaceTurnSnapshot(parentId, "wst_handle")).toMatchObject({ + status: "running", + deferredMessageIds: ["msg_prehandoff"], + }); + + await config.editConfig((cfg) => { + const descendant = Array.from(cfg.projects.values()) + .flatMap((project) => project.workspaces) + .find((workspace) => workspace.id === "descendant-task"); + assert(descendant, "descendant task must exist"); + descendant.archivedAt = "2026-06-19T00:01:00.000Z"; + return cfg; + }); + + const recovered = await taskService.getWorkspaceTurnSnapshot(parentId, "wst_handle"); + expect(recovered).toMatchObject({ + status: "completed", + messageId: "msg_prehandoff", + reportMarkdown: "Premature final text", + }); + expect(recovered?.deferredMessageIds).toBeUndefined(); + }); + + test("workspace-turn deferred recovery waits for active workflow blockers", async () => { + const { config, parentId, taskService, historyService } = await startWorkspaceTurnForTest(); + const runStore = new WorkflowRunStore({ sessionDir: config.getSessionDir("childworkspace") }); + await runStore.createRun({ + id: "wfr_child_background", + workspaceId: "childworkspace", + workflow: { + name: "child-background", + description: "Child background workflow", + scope: "built-in", + executable: true, + }, + source: "export default function workflow() { return { reportMarkdown: 'done' }; }\n", + args: {}, + now: "2026-06-19T00:00:00.000Z", + }); + await runStore.appendStatus("wfr_child_background", "running", "2026-06-19T00:00:01.000Z"); + await recordAgentWorkflowRunReference({ + workspaceSessionDir: config.getSessionDir("childworkspace"), + runId: "wfr_child_background", + createdAtMs: Date.parse("2026-06-19T00:00:01.000Z"), + }); + + const muxMetadata = { + type: "workspace-turn-task" as const, + taskHandleId: "wst_handle", + ownerWorkspaceId: parentId, + turnId: "turn", + }; + const appendResult = await historyService.appendToHistory( + "childworkspace", + createMuxMessage("msg_workflow_blocked", "assistant", "Workflow-blocked final text", { + model: "anthropic:claude-opus-4-6", + agentId: "exec", + finishReason: "stop", + muxMetadata, + }) + ); + expect(appendResult.success).toBe(true); + + await ( + taskService as unknown as { handleStreamEnd: (event: StreamEndEvent) => Promise } + ).handleStreamEnd({ + type: "stream-end", + workspaceId: "childworkspace", + messageId: "msg_workflow_blocked", + metadata: { + model: "anthropic:claude-opus-4-6", + agentId: "exec", + finishReason: "stop", + muxMetadata, + }, + parts: [{ type: "text", text: "Workflow-blocked final text" }], + }); + + expect(await taskService.getWorkspaceTurnSnapshot(parentId, "wst_handle")).toMatchObject({ + status: "running", + deferredMessageIds: ["msg_workflow_blocked"], + }); + + await runStore.appendStatus("wfr_child_background", "completed", "2026-06-19T00:00:02.000Z"); + const recovered = await taskService.getWorkspaceTurnSnapshot(parentId, "wst_handle"); + expect(recovered).toMatchObject({ + status: "completed", + messageId: "msg_workflow_blocked", + reportMarkdown: "Workflow-blocked final text", + }); + }); + test("workspace-turn auto-resume preserves handle metadata", async () => { const { config, parentId, projectPath, taskService, workspaceMocks } = await startWorkspaceTurnForTest(); @@ -7831,6 +8116,59 @@ describe("TaskService", () => { expect(report).toEqual({ reportMarkdown: "persisted report", title: "persisted title" }); }); + test("waitForAgentReport returns persisted artifact for stale running task", async () => { + const config = await createTestConfig(rootDir); + + const projectPath = path.join(rootDir, "repo"); + const parentId = "parent-111"; + const childId = "child-222"; + + await saveWorkspaces( + config, + projectPath, + [ + projectWorkspace(projectPath, "parent", parentId), + projectWorkspace(projectPath, "child", childId, { + name: "agent_explore_child", + parentWorkspaceId: parentId, + agentType: "explore", + taskStatus: "running", + }), + ], + testTaskSettings(1, 3) + ); + + const { taskService } = createTaskServiceHarness(config); + const patchGeneration = spyOn( + ( + taskService as unknown as { + gitPatchArtifactService: { maybeStartGeneration: (...args: unknown[]) => Promise }; + } + ).gitPatchArtifactService, + "maybeStartGeneration" + ).mockResolvedValue(undefined); + + await upsertSubagentReportArtifact({ + workspaceId: parentId, + workspaceSessionDir: config.getSessionDir(parentId), + childTaskId: childId, + parentWorkspaceId: parentId, + ancestorWorkspaceIds: [parentId], + reportMarkdown: "persisted report", + title: "persisted title", + nowMs: Date.now(), + }); + + const report = await taskService.waitForAgentReport(childId, { + timeoutMs: 10, + requestingWorkspaceId: parentId, + }); + + expect(report).toEqual({ reportMarkdown: "persisted report", title: "persisted title" }); + expect(findWorkspaceInConfig(config, childId)?.taskStatus).toBe("reported"); + expect(patchGeneration).toHaveBeenCalledWith(parentId, childId, expect.any(Function)); + }); + test("waitForAgentReport returns persisted report after workspace is removed", async () => { const config = await createTestConfig(rootDir); diff --git a/src/node/services/taskService.ts b/src/node/services/taskService.ts index e6611af141..8ef98cca0e 100644 --- a/src/node/services/taskService.ts +++ b/src/node/services/taskService.ts @@ -496,6 +496,13 @@ export interface DescendantAgentTaskInfo { type AgentTaskWorkspaceEntry = WorkspaceConfigEntry & { projectPath: string }; +const ACTIVE_AGENT_TASK_STATUSES = new Set([ + "queued", + "starting", + "running", + "awaiting_report", +]); + const WORKSPACE_BUSY_IDLE_ONLY_SEND_MESSAGE = "Workspace is busy; idle-only send was skipped."; function isWorkspaceBusyIdleOnlySend(error: unknown): boolean { @@ -4135,6 +4142,23 @@ export class TaskService { }); this.enforceCompletedReportCacheLimit(); + const entry = findWorkspaceEntry(this.config.loadConfigOrDefault(), taskId); + if (entry != null && !hasCompletedAgentReport(entry.workspace)) { + await this.editWorkspaceEntry( + taskId, + (workspace) => { + workspace.taskStatus = "reported"; + workspace.reportedAt = getIsoNow(); + delete workspace.taskRecoveryAttempts; + }, + { allowMissing: true } + ); + await this.maybeStartPatchGenerationForReportedTask(taskId); + await this.emitWorkspaceMetadata(taskId); + await this.maybeStartQueuedTasks(); + await this.cleanupReportedLeafTask(taskId); + } + return { reportMarkdown: artifact.reportMarkdown, title: artifact.title, @@ -4325,6 +4349,12 @@ export class TaskService { this.registerBackgroundableForegroundWaiter(requestingWorkspaceId, entry); } + const persistedAfterRegister = await tryReadPersistedReport(); + if (persistedAfterRegister) { + entry.resolve(persistedAfterRegister); + return; + } + // Don't start the execution timeout while the task is still queued/starting. // The timer starts once the child actually begins running (queued/starting -> running). const initialStatus = taskWorkspaceEntry.workspace.taskStatus; @@ -4511,12 +4541,6 @@ export class TaskService { const cfg = this.config.loadConfigOrDefault(); const index = this.buildAgentTaskIndex(cfg); - const activeStatuses = new Set([ - "queued", - "starting", - "running", - "awaiting_report", - ]); const result: string[] = []; const stack: Array<{ taskId: string; workflowOwned: boolean }> = [ ...(index.childrenByParent.get(workspaceId) ?? []).map((taskId) => ({ @@ -4528,10 +4552,9 @@ export class TaskService { const next = stack.pop()!; const entry = index.byId.get(next.taskId); const workflowOwned = next.workflowOwned || entry?.workflowTask != null; - const status = entry?.taskStatus; if ( - status && - activeStatuses.has(status) && + entry != null && + this.isActiveAgentTaskEntry(entry) && !(options.excludeWorkflowTasks && workflowOwned) ) { result.push(next.taskId); @@ -4554,10 +4577,32 @@ export class TaskService { return null; } const record = await this.taskHandleStore.getWorkspaceTurn(ownerWorkspaceId, handleId); + // Older recovery skipped deferred stream-end history and could mark a completed workspace turn + // interrupted. Re-check the durable child history so affected handles self-heal on next await. + if ( + record?.status === "interrupted" && + record.error === "Workspace turn interrupted after restart" && + (record.deferredMessageIds?.length ?? 0) > 0 + ) { + const recovered = await this.recoverTerminalWorkspaceTurnFromHistory(record); + if (recovered != null) { + await this.taskHandleStore.upsertWorkspaceTurn(recovered); + await this.cleanupDisposableWorkspaceTurn(recovered); + const active = this.activeWorkspaceTurnHandleByWorkspaceId.get(record.workspaceId); + if ( + active?.handleId === record.handleId && + active.ownerWorkspaceId === record.ownerWorkspaceId + ) { + this.activeWorkspaceTurnHandleByWorkspaceId.delete(record.workspaceId); + } + return await this.taskHandleStore.getWorkspaceTurn(ownerWorkspaceId, handleId); + } + } + if ( record != null && (record.status === "starting" || record.status === "running") && - !this.isLiveWorkspaceTurn(record) + !(await this.isLiveWorkspaceTurn(record)) ) { await this.settleStaleWorkspaceTurn(record); return await this.taskHandleStore.getWorkspaceTurn(ownerWorkspaceId, handleId); @@ -4575,7 +4620,7 @@ export class TaskService { for (const record of records) { if ( (record.status === "starting" || record.status === "running") && - !this.isLiveWorkspaceTurn(record) + !(await this.isLiveWorkspaceTurn(record)) ) { await this.settleStaleWorkspaceTurn(record); const latest = await this.taskHandleStore.getWorkspaceTurn( @@ -4997,14 +5042,50 @@ export class TaskService { return record.status === "starting" || record.status === "running"; } - private isLiveWorkspaceTurn(record: WorkspaceTurnTaskHandleRecord): boolean { + private async hasActiveWorkspaceTurnDeferredBlockers( + record: WorkspaceTurnTaskHandleRecord + ): Promise { + if (this.hasActiveDescendantAgentTasks(this.config.loadConfigOrDefault(), record.workspaceId)) { + return true; + } + + const referencedWorkflowRunIds = await this.listAgentReferencedWorkflowRunIds( + record.workspaceId, + [] + ); + if ( + (await this.listActiveBackgroundWorkflowRunIds(record.workspaceId, referencedWorkflowRunIds)) + .length > 0 + ) { + return true; + } + + return (await this.listActiveWorkspaceTurnTaskIdsForOwner(record.workspaceId)).length > 0; + } + + private async isLiveWorkspaceTurn(record: WorkspaceTurnTaskHandleRecord): Promise { const active = this.activeWorkspaceTurnHandleByWorkspaceId.get(record.workspaceId); - return ( - (active?.handleId === record.handleId && - active.ownerWorkspaceId === record.ownerWorkspaceId) || + const hasRuntimeActivity = this.aiService.isStreaming(record.workspaceId) || - this.workspaceService.hasPendingQueuedOrPreparingTurn(record.workspaceId) - ); + this.workspaceService.hasPendingQueuedOrPreparingTurn(record.workspaceId); + if (hasRuntimeActivity) { + return true; + } + + const isActiveHandle = + active?.handleId === record.handleId && active.ownerWorkspaceId === record.ownerWorkspaceId; + if (!isActiveHandle) { + return false; + } + + if ((record.deferredMessageIds?.length ?? 0) === 0) { + return true; + } + + // A deferred workspace-turn stream-end was waiting for background work. Once there is no + // live stream/queued retry and no active descendant/workflow/nested turn left, the in-memory + // handle is stale and should be recovered from the deferred history instead of blocking forever. + return await this.hasActiveWorkspaceTurnDeferredBlockers(record); } private async settleStaleWorkspaceTurn(record: WorkspaceTurnTaskHandleRecord): Promise { @@ -5024,6 +5105,18 @@ export class TaskService { return; } + // Same-process deferred stream-ends can be observed before the final assistant message is + // readable from history. Keep the handle alive in that narrow window; after restart the active + // map is empty, so unrecoverable deferred handles still settle terminally instead of leaking. + const active = this.activeWorkspaceTurnHandleByWorkspaceId.get(record.workspaceId); + if ( + (record.deferredMessageIds?.length ?? 0) > 0 && + active?.handleId === record.handleId && + active.ownerWorkspaceId === record.ownerWorkspaceId + ) { + return; + } + const next: WorkspaceTurnTaskHandleRecord = { ...record, status: "interrupted", @@ -5053,7 +5146,7 @@ export class TaskService { if (!this.isActiveWorkspaceTurn(record)) { continue; } - if (!this.isLiveWorkspaceTurn(record)) { + if (!(await this.isLiveWorkspaceTurn(record))) { await this.settleStaleWorkspaceTurn(record); continue; } @@ -5073,7 +5166,7 @@ export class TaskService { if (record.status !== "starting" && record.status !== "running") { continue; } - if (!this.isLiveWorkspaceTurn(record)) { + if (!(await this.isLiveWorkspaceTurn(record))) { await this.settleStaleWorkspaceTurn(record); continue; } @@ -5082,6 +5175,22 @@ export class TaskService { return taskIds; } + private isActiveAgentTaskEntry(task: AgentTaskWorkspaceEntry): boolean { + const status: AgentTaskStatus = task.taskStatus ?? "running"; + if (!ACTIVE_AGENT_TASK_STATUSES.has(status)) { + return false; + } + + // Archiving a task stops its stream but intentionally leaves taskStatus untouched in + // persisted config. Treat archived, non-streaming tasks as inactive so stale status cannot + // keep ancestors/workspace-turn handles blocked forever. + if (isWorkspaceArchived(task.archivedAt, task.unarchivedAt)) { + return task.id != null && this.aiService.isStreaming(task.id); + } + + return true; + } + private countActiveAgentTasks(config: ReturnType): number { let activeCount = 0; for (const task of this.listAgentTaskWorkspaces(config)) { @@ -5094,7 +5203,7 @@ export class TaskService { if (status === "running" && task.id && this.isForegroundAwaiting(task.id)) { continue; } - if (status === "starting" || status === "running" || status === "awaiting_report") { + if (status !== "queued" && this.isActiveAgentTaskEntry(task)) { activeCount += 1; continue; } @@ -5129,17 +5238,11 @@ export class TaskService { "hasActiveDescendantAgentTasksUsingIndex: workspaceId must be non-empty" ); - const activeStatuses = new Set([ - "queued", - "starting", - "running", - "awaiting_report", - ]); const stack: string[] = [...(index.childrenByParent.get(workspaceId) ?? [])]; while (stack.length > 0) { const next = stack.pop()!; - const status = index.byId.get(next)?.taskStatus; - if (status && activeStatuses.has(status)) { + const entry = index.byId.get(next); + if (entry != null && this.isActiveAgentTaskEntry(entry)) { return true; } const children = index.childrenByParent.get(next); @@ -5895,10 +5998,13 @@ export class TaskService { record: WorkspaceTurnTaskHandleRecord, event: StreamEndEvent ): WorkspaceTurnTaskHandleRecord { + const baseRecord = { ...record }; + delete baseRecord.error; + delete baseRecord.deferredMessageIds; // Truncated/non-stop provider finishes are partial output, not a completed delegated turn. if (event.metadata.finishReason != null && event.metadata.finishReason !== "stop") { return { - ...record, + ...baseRecord, status: "error", updatedAt: getIsoNow(), messageId: event.messageId, @@ -5911,7 +6017,7 @@ export class TaskService { }; } return { - ...record, + ...baseRecord, status: "completed", updatedAt: getIsoNow(), messageId: event.messageId, @@ -5946,8 +6052,10 @@ export class TaskService { }); return null; } + + const allowDeferredMessages = !(await this.hasActiveWorkspaceTurnDeferredBlockers(record)); for (const message of historyResult.data.toReversed()) { - if (this.isDeferredWorkspaceTurnMessage(record, message.id)) { + if (this.isDeferredWorkspaceTurnMessage(record, message.id) && !allowDeferredMessages) { continue; } const event = this.buildWorkspaceTurnStreamEndEventFromHistory(record, message);