diff --git a/src/node/services/taskService.test.ts b/src/node/services/taskService.test.ts index 1263de0ac35..fa02527e8fb 100644 --- a/src/node/services/taskService.test.ts +++ b/src/node/services/taskService.test.ts @@ -402,6 +402,8 @@ function createWorkspaceServiceMocks( emitChatEvent: ReturnType; isWorkflowInvocationCurrent: ReturnType; create: ReturnType; + on: ReturnType; + off: ReturnType; }> ): { workspaceService: WorkspaceService; @@ -429,6 +431,10 @@ function createWorkspaceServiceMocks( emitChatEvent: ReturnType; isWorkflowInvocationCurrent: ReturnType; create: ReturnType; + on: ReturnType; + off: ReturnType; + /** Emit a chat event to listeners registered via workspaceService.on("chat"). */ + emitChatToListeners: (event: { workspaceId: string; message: WorkspaceChatMessage }) => void; } { const sendMessage = overrides?.sendMessage ?? mock((): Promise> => Promise.resolve(Ok(undefined))); @@ -478,6 +484,35 @@ function createWorkspaceServiceMocks( Promise.resolve(Err("workspaceService.create not mocked")) ); + const chatListeners = new Set< + (event: { workspaceId: string; message: WorkspaceChatMessage }) => void + >(); + const on = + overrides?.on ?? + mock((event: string, listener: (...args: unknown[]) => void) => { + if (event === "chat") { + chatListeners.add( + listener as (event: { workspaceId: string; message: WorkspaceChatMessage }) => void + ); + } + return undefined; + }); + const off = + overrides?.off ?? + mock((event: string, listener: (...args: unknown[]) => void) => { + if (event === "chat") { + chatListeners.delete( + listener as (event: { workspaceId: string; message: WorkspaceChatMessage }) => void + ); + } + return undefined; + }); + const emitChatToListeners = (event: { workspaceId: string; message: WorkspaceChatMessage }) => { + for (const listener of chatListeners) { + listener(event); + } + }; + return { workspaceService: { create, @@ -504,6 +539,8 @@ function createWorkspaceServiceMocks( isExperimentEnabled, emitChatEvent, isWorkflowInvocationCurrent, + on, + off, } as unknown as WorkspaceService, create, sendMessage, @@ -529,6 +566,9 @@ function createWorkspaceServiceMocks( isExperimentEnabled, emitChatEvent, isWorkflowInvocationCurrent, + on, + off, + emitChatToListeners, }; } @@ -20134,7 +20174,7 @@ describe("TaskService", () => { expect(childWorkspace?.taskLaunchError).toBe(refusalMessage); }); - test("running tasks are NOT settled by aborted, context_exceeded, or retryable stream errors", async () => { + test("running tasks are NOT settled by aborted, context_exceeded, or in-flight retryable stream errors", async () => { const config = await createTestConfig(rootDir); const projectPath = path.join(rootDir, "repo"); @@ -20157,7 +20197,12 @@ describe("TaskService", () => { testTaskSettings(1, 3) ); - const { workspaceService, sendMessage } = createWorkspaceServiceMocks(); + // Simulate an in-flight auto-retry so retryable transport errors stay owned + // by the session retry loop rather than terminal-settling immediately. + const hasPendingAutoRetry = mock((workspaceId: string) => workspaceId === childId); + const { workspaceService, sendMessage } = createWorkspaceServiceMocks({ + hasPendingAutoRetry, + }); const { taskService } = createTaskServiceHarness(config, { workspaceService }); const internal = taskService as unknown as { @@ -20185,7 +20230,8 @@ describe("TaskService", () => { errorType: "context_exceeded", }); - // Retryable transport errors stay owned by the agent session's retry loop. + // Retryable transport errors with a pending auto-retry stay owned by the + // agent session's retry loop (settlement happens on abandon if that fails). await internal.handleTaskStreamError({ type: "error", workspaceId: childId, @@ -20204,6 +20250,244 @@ describe("TaskService", () => { expect(childWorkspace?.taskLaunchError).toBeUndefined(); }); + test("running tasks settle immediately when a retryable stream error has no in-flight recovery", async () => { + const config = await createTestConfig(rootDir); + + const projectPath = path.join(rootDir, "repo"); + const parentId = "parent-111"; + const childId = "child-222"; + const networkError = "fetch failed"; + + await saveWorkspaces( + config, + projectPath, + [ + projectWorkspace(projectPath, "parent", parentId), + projectWorkspace(projectPath, "child", childId, { + name: "agent_explore_child", + parentWorkspaceId: parentId, + agentType: "explore", + taskStatus: "running", + taskModelString: "openai:gpt-5.5-pro", + }), + ], + testTaskSettings(1, 3) + ); + + // Default mocks: no pending auto-retry and not streaming — e.g. auto-retry + // disabled, so the stream-error itself must settle the parent waiter. + const { workspaceService } = createWorkspaceServiceMocks(); + const { taskService } = createTaskServiceHarness(config, { workspaceService }); + + const internal = taskService as unknown as { + handleTaskStreamError: (event: ErrorEvent) => Promise; + }; + + const waiterOutcome = taskService + .waitForAgentReport(childId, { timeoutMs: 10_000, requestingWorkspaceId: parentId }) + .then( + () => null, + (error: unknown) => error + ); + + await internal.handleTaskStreamError({ + type: "error", + workspaceId: childId, + messageId: "assistant-error-network", + error: networkError, + errorType: "network", + }); + + const rejection = await waiterOutcome; + expect(rejection).toBeInstanceOf(Error); + expect((rejection as Error).message).toBe(networkError); + + const childWorkspace = findWorkspaceInConfig(config, childId); + expect(childWorkspace?.taskStatus).toBe("interrupted"); + expect(childWorkspace?.taskLaunchError).toBe(networkError); + }); + + // Regression: a retryable stream error schedules in-session auto-retry, so + // handleTaskStreamError deliberately leaves the parent waiter blocked. If that + // retry is later abandoned (missing resume options, retry callback failed) without + // a new settling stream-error, the parent would wait until the 10-minute timeout. + // auto-retry-abandoned must settle the child and unblock the parent promptly. + test("auto-retry abandoned after a retryable stream error settles the child and unblocks the parent", async () => { + const config = await createTestConfig(rootDir); + + const projectPath = path.join(rootDir, "repo"); + const parentId = "parent-111"; + const childId = "child-222"; + const networkError = "fetch failed"; + + await saveWorkspaces( + config, + projectPath, + [ + projectWorkspace(projectPath, "parent", parentId), + projectWorkspace(projectPath, "child", childId, { + name: "agent_explore_child", + parentWorkspaceId: parentId, + agentType: "explore", + taskStatus: "running", + taskModelString: "openai:gpt-5.5-pro", + }), + ], + testTaskSettings(1, 3) + ); + + // Pending auto-retry keeps the task running after the stream error; abandonment + // is what must settle the parent (the no-pending path is covered separately). + let retryPending = true; + const hasPendingAutoRetry = mock( + (workspaceId: string) => retryPending && workspaceId === childId + ); + const workspaceMocks = createWorkspaceServiceMocks({ hasPendingAutoRetry }); + const { taskService } = createTaskServiceHarness(config, { + workspaceService: workspaceMocks.workspaceService, + }); + + const internal = taskService as unknown as { + handleTaskStreamError: (event: ErrorEvent) => Promise; + workspaceEventLocks: { withLock(key: string, fn: () => Promise): Promise }; + }; + + // Parent is blocked on the child when the retryable failure lands. + const waiterOutcome = taskService + .waitForAgentReport(childId, { timeoutMs: 10_000, requestingWorkspaceId: parentId }) + .then( + () => null, + (error: unknown) => error + ); + + await internal.handleTaskStreamError({ + type: "error", + workspaceId: childId, + messageId: "assistant-error-network", + error: networkError, + errorType: "network", + }); + + // Still running after the retryable error — session owns the retry loop. + expect(findWorkspaceInConfig(config, childId)?.taskStatus).toBe("running"); + + // Auto-retry abandons without a further stream-error (e.g. missing_retry_options). + retryPending = false; + workspaceMocks.emitChatToListeners({ + workspaceId: childId, + message: { type: "auto-retry-abandoned", reason: "missing_retry_options" }, + }); + // Drain the chat-listener lock so settlement finishes before assertions. + await internal.workspaceEventLocks.withLock(childId, () => Promise.resolve()); + + const rejection = await waiterOutcome; + expect(rejection).toBeInstanceOf(Error); + expect((rejection as Error).message).toBe(networkError); + + const childWorkspace = findWorkspaceInConfig(config, childId); + expect(childWorkspace?.taskStatus).toBe("interrupted"); + expect(childWorkspace?.taskLaunchError).toBe(networkError); + + const failure = await readSubagentFailureArtifact(config.getSessionDir(parentId), childId); + expect(failure).not.toBeNull(); + expect(failure?.errorType).toBe("network"); + expect(failure?.errorMessage).toBe(networkError); + }); + + test("workspace-turn auto-retry abandoned after pending retry settles the handle as failed", async () => { + let retryPending = true; + const hasPendingAutoRetry = mock( + (workspaceId: string) => retryPending && workspaceId === "childworkspace" + ); + const waitForPendingStreamErrorRecoveryDecision = mock((): Promise => Promise.resolve()); + const { parentId, taskService } = await startWorkspaceTurnForTest({ + hasPendingAutoRetry, + waitForPendingStreamErrorRecoveryDecision, + }); + + const internal = taskService as unknown as { + handleTaskStreamError: (event: ErrorEvent) => Promise; + handleTaskAutoRetryAbandoned: (workspaceId: string, reason: string) => Promise; + }; + + await internal.handleTaskStreamError({ + type: "error", + workspaceId: "childworkspace", + messageId: "msg_truncated_pending", + error: "Anthropic stream closed unexpectedly before the response completed.", + errorType: "stream_truncated", + }); + + expect(await taskService.getWorkspaceTurnSnapshot(parentId, "wst_handle")).toMatchObject({ + status: "running", + workspaceId: "childworkspace", + }); + + // Retry later abandons (callback failed / missing options). Clear pending flag first. + retryPending = false; + await internal.handleTaskAutoRetryAbandoned("childworkspace", "missing_retry_options"); + + expect(await taskService.getWorkspaceTurnSnapshot(parentId, "wst_handle")).toMatchObject({ + status: "error", + workspaceId: "childworkspace", + error: "Anthropic stream closed unexpectedly before the response completed.", + }); + }); + + test("auto-retry abandoned with disabled_by_user does not terminal-fail a 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", + taskModelString: "openai:gpt-5.5-pro", + }), + ], + testTaskSettings(1, 3) + ); + + // Pending retry during the stream-error path keeps the task running; clear it + // before abandon so we exercise the disabled_by_user early-return itself + // (not the hasPendingAutoRetry live-check). + let retryPending = true; + const hasPendingAutoRetry = mock( + (workspaceId: string) => retryPending && workspaceId === childId + ); + const { workspaceService } = createWorkspaceServiceMocks({ hasPendingAutoRetry }); + const { taskService } = createTaskServiceHarness(config, { workspaceService }); + + const internal = taskService as unknown as { + handleTaskStreamError: (event: ErrorEvent) => Promise; + handleTaskAutoRetryAbandoned: (workspaceId: string, reason: string) => Promise; + }; + + await internal.handleTaskStreamError({ + type: "error", + workspaceId: childId, + messageId: "assistant-error-network", + error: "fetch failed", + errorType: "network", + }); + + retryPending = false; + await internal.handleTaskAutoRetryAbandoned(childId, "disabled_by_user"); + + const childWorkspace = findWorkspaceInConfig(config, childId); + expect(childWorkspace?.taskStatus).toBe("running"); + expect(childWorkspace?.taskLaunchError).toBeUndefined(); + }); + test("background task refusal stays observable after cleanup and restart via failure artifact", async () => { const config = await createTestConfig(rootDir); diff --git a/src/node/services/taskService.ts b/src/node/services/taskService.ts index 33d593d33a1..ba6037b62fd 100644 --- a/src/node/services/taskService.ts +++ b/src/node/services/taskService.ts @@ -101,6 +101,7 @@ import { } from "@/common/types/thinking"; import { snapshotTranscriptAnchor } from "@/node/services/transcriptAnchor"; import type { ErrorEvent, StreamAbortEvent, StreamEndEvent } from "@/common/types/stream"; +import type { WorkspaceChatMessage } from "@/common/orpc/types"; import { isActiveWorkflowRunStatus, isTerminalWorkflowRunStatus, @@ -844,6 +845,36 @@ function isErrorEvent(value: unknown): value is ErrorEvent { return isTypedWorkspaceEvent(value, "error"); } +// Exhaustive map so adding a StreamErrorType to the schema fails typecheck until +// this lookup is updated — otherwise a new retryable type would not be cached and +// auto-retry-abandoned settlement would fall back to a generic abandon reason. +const STREAM_ERROR_TYPE_LOOKUP = { + authentication: true, + rate_limit: true, + server_error: true, + api: true, + retry_failed: true, + aborted: true, + network: true, + context_exceeded: true, + quota: true, + model_not_found: true, + runtime_not_ready: true, + runtime_start_failed: true, + empty_output: true, + stream_truncated: true, + max_output_tokens: true, + model_refusal: true, + unknown: true, +} as const satisfies Record; + +function isStreamErrorType(value: unknown): value is StreamErrorType { + return ( + typeof value === "string" && + Object.prototype.hasOwnProperty.call(STREAM_ERROR_TYPE_LOOKUP, value) + ); +} + function hasAncestorWorkspaceId( entry: { ancestorWorkspaceIds?: unknown } | null | undefined, ancestorWorkspaceId: string @@ -1261,6 +1292,17 @@ export class TaskService { // Bounded by max entries; disk persistence is the source of truth for restart-safety. private readonly completedReportsByTaskId = new Map(); + /** + * Last observed stream-error for a child workspace. Used when auto-retry is later + * abandoned (retry callback failed, user disabled retries, missing resume options): + * the original error event may have kept the parent handle alive because a retry was + * pending, and abandonment itself carries only a reason string, not the stream error. + */ + private readonly lastStreamErrorByWorkspaceId = new Map< + string, + { error: string; errorType: StreamErrorType } + >(); + // 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 @@ -1720,6 +1762,32 @@ export class TaskService { log.error("TaskService.handleTaskStreamError failed", { error }); }); }); + + // Auto-retry lifecycle is a chat event (not an AIService stream event). When a child + // schedules a retry, stream-error handling leaves parent waiters blocked; if that retry + // is later abandoned without a new stream-error, nothing else settles the handle and + // the parent blocks until timeout. Listen here so abandonment fails the task gracefully. + this.workspaceService.on( + "chat", + (event: { workspaceId: string; message: WorkspaceChatMessage }) => { + if (event.message.type !== "auto-retry-abandoned") { + return; + } + const workspaceId = event.workspaceId; + const reason = event.message.reason; + void this.workspaceEventLocks + .withLock(workspaceId, async () => { + await this.handleTaskAutoRetryAbandoned(workspaceId, reason); + }) + .catch((error: unknown) => { + log.error("TaskService.handleTaskAutoRetryAbandoned failed", { + workspaceId, + reason, + error, + }); + }); + } + ); } setTimelineRecorder(recorder: TimelineRecorder): void { @@ -9349,6 +9417,8 @@ export class TaskService { private async handleStreamEnd(event: StreamEndEvent): Promise { const workspaceId = event.workspaceId; + // A successful stream-end supersedes any cached stream-error used for abandon settlement. + this.lastStreamErrorByWorkspaceId.delete(workspaceId); // Ensure any in-flight notify_on_terminal persistence (from a just-detached foreground wait) // has settled so the config we read below reflects the durable non-blocking policy. @@ -9851,13 +9921,29 @@ export class TaskService { } private async handleTaskStreamError(event: ErrorEvent): Promise { + // Remember the latest stream failure so a later auto-retry-abandoned event can + // surface the original error text when settling parent waiters. + if (event.errorType != null && isStreamErrorType(event.errorType) && event.error.length > 0) { + this.lastStreamErrorByWorkspaceId.set(event.workspaceId, { + error: event.error, + errorType: event.errorType, + }); + } + if (await this.finalizeWorkspaceTurnFromStreamError(event)) { + // Workspace-turn path settled (or kept running for a pending retry). Clear the + // cache only on true terminal settlement — pending-retry keeps the handle live. + const active = await this.getActiveWorkspaceTurnRecordForWorkspace(event.workspaceId); + if (active == null) { + this.lastStreamErrorByWorkspaceId.delete(event.workspaceId); + } return; } const workspaceId = event.workspaceId; const cfg = this.config.loadConfigOrDefault(); const entry = findWorkspaceEntry(cfg, workspaceId); if (!entry?.workspace.parentWorkspaceId) { + this.lastStreamErrorByWorkspaceId.delete(workspaceId); return; } @@ -9865,6 +9951,7 @@ export class TaskService { // Stream errors only need settlement handling while the task is mid-run // (running) or waiting on its completion tool (awaiting_report). if (status !== "running" && status !== "awaiting_report") { + this.lastStreamErrorByWorkspaceId.delete(workspaceId); return; } const taskIndex = this.buildAgentTaskIndex(cfg); @@ -9877,6 +9964,7 @@ export class TaskService { taskIndex ) ) { + this.lastStreamErrorByWorkspaceId.delete(workspaceId); return; } @@ -9910,12 +9998,47 @@ export class TaskService { errorType: event.errorType ?? "unknown", errorMessage: event.error, }); + this.lastStreamErrorByWorkspaceId.delete(workspaceId); return; } if (status !== "awaiting_report") { - // Retryable errors during `running` are handled by the agent session's - // retry loop; TaskService only intervenes once the task owes its report. + // Retryable (or non-settling non-retryable) errors during `running` are owned by + // the agent session's recovery loop. Wait for that decision, then: + // - if a retry/stream is in flight, keep the parent blocked and rely on + // auto-retry-abandoned / a later stream-end to settle; + // - if nothing is recovering (auto-retry disabled, exhausted immediately), + // settle now so the parent is not blocked until waitForAgentReport times out. + // Keep lastStreamError cached when a retry is pending so abandon can reuse it. + await this.workspaceService.waitForPendingStreamErrorRecoveryDecision(workspaceId); + if ( + this.aiService.isStreaming(workspaceId) || + this.workspaceService.hasPendingAutoRetry(workspaceId) + ) { + return; + } + // context_exceeded / aborted have non-auto-retry in-session recovery paths + // (compaction, user follow-up); do not terminal-fail those here. + if ( + event.errorType != null && + WORKSPACE_TURN_RECOVERABLE_STREAM_ERRORS.has(event.errorType) + ) { + return; + } + log.error( + "Task stream error left no in-flight recovery; interrupting task so the parent is not blocked", + { + workspaceId, + taskStatus: status, + errorType: event.errorType, + error: event.error, + } + ); + await this.failAgentTaskTerminally(workspaceId, entry, { + errorType: event.errorType ?? "unknown", + errorMessage: event.error, + }); + this.lastStreamErrorByWorkspaceId.delete(workspaceId); return; } @@ -9934,6 +10057,112 @@ export class TaskService { }); } + /** + * When a child session abandons auto-retry without emitting a new stream-error + * that settles the task (retry callback failed, missing resume options, user + * disabled retries after a scheduled attempt), parent waiters would otherwise + * stay blocked until waitForAgentReport's timeout. Settle workspace-turn handles + * and agent tasks terminally using the last known stream error when available. + * + * `disabled_by_user` is excluded: the user may still manually continue the child, + * and treating that as a hard parent-visible failure would be surprising. + */ + private async handleTaskAutoRetryAbandoned(workspaceId: string, reason: string): Promise { + assert(workspaceId.length > 0, "handleTaskAutoRetryAbandoned: workspaceId must be non-empty"); + if (reason === "disabled_by_user") { + return; + } + + // If a retry was re-scheduled (or a stream restarted) between the abandon + // emission and this locked handler, leave the handle running. + if ( + this.aiService.isStreaming(workspaceId) || + this.workspaceService.hasPendingAutoRetry(workspaceId) + ) { + return; + } + + const cached = this.lastStreamErrorByWorkspaceId.get(workspaceId); + const errorMessage = + cached?.error ?? + (reason.length > 0 + ? `Auto-retry abandoned (${reason})` + : "Auto-retry abandoned without completing the turn"); + const errorType = cached?.errorType ?? "unknown"; + + // Prefer the workspace-turn settlement path when an active handle exists. + const turnRecord = await this.getActiveWorkspaceTurnRecordForWorkspace(workspaceId); + if (turnRecord != null) { + // Mirror finalizeWorkspaceTurnFromStreamError's recoverable gate: if an + // in-session recovery path is still in flight, do not settle. + if ( + WORKSPACE_TURN_RECOVERABLE_STREAM_ERRORS.has(errorType) && + (await this.hasRecoverableWorkspaceTurnRetryInFlight(workspaceId, { + requireAutoRetry: false, + })) + ) { + return; + } + log.error("Workspace-turn auto-retry abandoned; settling handle as failed", { + workspaceId, + handleId: turnRecord.handleId, + reason, + errorType, + error: errorMessage, + }); + const next: WorkspaceTurnTaskHandleRecord = { + ...turnRecord, + status: "error", + updatedAt: getIsoNow(), + error: errorMessage, + }; + await this.settleWorkspaceTurn({ + record: turnRecord, + next, + waiterSettlement: { status: "error", error: new Error(errorMessage) }, + }); + this.lastStreamErrorByWorkspaceId.delete(workspaceId); + return; + } + + const cfg = this.config.loadConfigOrDefault(); + const entry = findWorkspaceEntry(cfg, workspaceId); + if (!entry?.workspace.parentWorkspaceId) { + this.lastStreamErrorByWorkspaceId.delete(workspaceId); + return; + } + + const status = entry.workspace.taskStatus; + if (status !== "running" && status !== "awaiting_report") { + this.lastStreamErrorByWorkspaceId.delete(workspaceId); + return; + } + + const taskIndex = this.buildAgentTaskIndex(cfg); + if (await this.hasActiveTaskOwnedWork(workspaceId, taskIndex)) { + return; + } + + // User-steerable pause: do not terminal-fail on abandon when the last error was + // an explicit abort (the user may still send a follow-up). + if (errorType === "aborted") { + return; + } + + log.error("Task auto-retry abandoned; interrupting task so the parent is not blocked", { + workspaceId, + taskStatus: status, + reason, + errorType, + error: errorMessage, + }); + await this.failAgentTaskTerminally(workspaceId, entry, { + errorType, + errorMessage, + }); + this.lastStreamErrorByWorkspaceId.delete(workspaceId); + } + /** * Terminal settlement for a child task whose stream failed with a * non-retryable error: mark interrupted with a descriptive launch error,