diff --git a/src/node/services/taskService.test.ts b/src/node/services/taskService.test.ts index 3b34aadd8e..9cfd3ecb83 100644 --- a/src/node/services/taskService.test.ts +++ b/src/node/services/taskService.test.ts @@ -3724,6 +3724,854 @@ describe("TaskService", () => { }); }); + test("correlated stream-end corrects a stale error settlement after self-healed retry", async () => { + const { config, parentId, taskService } = await startWorkspaceTurnForTest(); + await new TaskHandleStore(config).upsertWorkspaceTurn({ + kind: "workspace_turn", + handleId: "wst_handle", + ownerWorkspaceId: parentId, + workspaceId: "childworkspace", + turnId: "turn", + status: "error", + createdAt: "2026-06-19T00:00:00.000Z", + updatedAt: "2026-06-19T00:00:01.000Z", + createdWorkspace: true, + disposableWorkspace: false, + error: "Stream error: provider overloaded", + terminalAttentionNotifiedAt: "2026-06-19T00:00:02.000Z", + }); + const muxMetadata = { + type: "workspace-turn-task" as const, + taskHandleId: "wst_handle", + ownerWorkspaceId: parentId, + turnId: "turn", + }; + const internal = taskService as unknown as { + handleStreamEnd: (event: StreamEndEvent) => Promise; + }; + + await internal.handleStreamEnd({ + type: "stream-end", + workspaceId: "childworkspace", + messageId: "msg_retry_final", + metadata: { + model: "anthropic:claude-opus-4-6", + agentId: "exec", + finishReason: "stop", + muxMetadata, + }, + parts: [{ type: "text", text: "Recovered after retry" }], + }); + + const snapshot = await taskService.getWorkspaceTurnSnapshot(parentId, "wst_handle"); + expect(snapshot).toMatchObject({ + status: "completed", + messageId: "msg_retry_final", + reportMarkdown: "Recovered after retry", + }); + expect(snapshot?.error).toBeUndefined(); + expect(snapshot?.terminalAttentionNotifiedAt).toBeUndefined(); + }); + + test("resettled workspace turn re-arms a consumed notify_on_terminal wake-up", async () => { + const { config, parentId, taskService } = await startWorkspaceTurnForTest(); + await new TaskHandleStore(config).upsertWorkspaceTurn({ + kind: "workspace_turn", + handleId: "wst_handle", + ownerWorkspaceId: parentId, + workspaceId: "childworkspace", + turnId: "turn", + status: "error", + createdAt: "2026-06-19T00:00:00.000Z", + updatedAt: "2026-06-19T00:00:01.000Z", + createdWorkspace: true, + disposableWorkspace: false, + error: "Stream error: provider overloaded", + attentionPolicy: "notify_on_terminal", + terminalAttentionNotifiedAt: "2026-06-19T00:00:02.000Z", + }); + // The stale error's wake-up was already delivered; without the tombstone reset, + // enqueueIfAbsent would swallow the corrected outcome's notification. + const terminalAttentionStore = new TerminalAttentionStore(config); + await terminalAttentionStore.enqueueIfAbsent({ + ownerWorkspaceId: parentId, + sourceKind: "workspace_turn", + sourceId: "wst_handle", + outputDelivery: "requires_task_await", + terminalOutcome: "error", + }); + await terminalAttentionStore.markDelivered(parentId, "workspace_turn:wst_handle"); + const muxMetadata = { + type: "workspace-turn-task" as const, + taskHandleId: "wst_handle", + ownerWorkspaceId: parentId, + turnId: "turn", + }; + const internal = taskService as unknown as { + handleStreamEnd: (event: StreamEndEvent) => Promise; + }; + + await internal.handleStreamEnd({ + type: "stream-end", + workspaceId: "childworkspace", + messageId: "msg_retry_final", + metadata: { + model: "anthropic:claude-opus-4-6", + agentId: "exec", + finishReason: "stop", + muxMetadata, + }, + parts: [{ type: "text", text: "Recovered after retry" }], + }); + + const notification = await terminalAttentionStore.get(parentId, "workspace_turn:wst_handle"); + expect(notification).toMatchObject({ terminalOutcome: "completed" }); + expect(await taskService.getWorkspaceTurnSnapshot(parentId, "wst_handle")).toMatchObject({ + status: "completed", + reportMarkdown: "Recovered after retry", + }); + }); + + test("duplicate correlated stream-end replay keeps a settled error handle unchanged", async () => { + const { config, parentId, taskService } = await startWorkspaceTurnForTest(); + await new TaskHandleStore(config).upsertWorkspaceTurn({ + kind: "workspace_turn", + handleId: "wst_handle", + ownerWorkspaceId: parentId, + workspaceId: "childworkspace", + turnId: "turn", + status: "error", + createdAt: "2026-06-19T00:00:00.000Z", + updatedAt: "2026-06-19T00:00:01.000Z", + createdWorkspace: true, + disposableWorkspace: false, + messageId: "msg_truncated_replay", + error: "Workspace turn ended before completion (finishReason: length)", + terminalAttentionNotifiedAt: "2026-06-19T00:00:02.000Z", + }); + const muxMetadata = { + type: "workspace-turn-task" as const, + taskHandleId: "wst_handle", + ownerWorkspaceId: parentId, + turnId: "turn", + }; + const internal = taskService as unknown as { + handleStreamEnd: (event: StreamEndEvent) => Promise; + }; + + await internal.handleStreamEnd({ + type: "stream-end", + workspaceId: "childworkspace", + messageId: "msg_truncated_replay", + metadata: { + model: "anthropic:claude-opus-4-6", + agentId: "exec", + finishReason: "length", + muxMetadata, + }, + parts: [{ type: "text", text: "Partial text" }], + }); + + const snapshot = await taskService.getWorkspaceTurnSnapshot(parentId, "wst_handle"); + expect(snapshot).toMatchObject({ + status: "error", + messageId: "msg_truncated_replay", + updatedAt: "2026-06-19T00:00:01.000Z", + terminalAttentionNotifiedAt: "2026-06-19T00:00:02.000Z", + }); + }); + + test("late correlated stream-end does not resettle an explicitly interrupted workspace turn", async () => { + const { config, parentId, taskService } = await startWorkspaceTurnForTest(); + // Explicit interrupt (user Esc / task_terminate): status interrupted WITHOUT the + // stale-restart marker. An in-flight stream-end completing after the cancel must not + // make the canceled turn appear completed. + 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, + }); + const muxMetadata = { + type: "workspace-turn-task" as const, + taskHandleId: "wst_handle", + ownerWorkspaceId: parentId, + turnId: "turn", + }; + const internal = taskService as unknown as { + handleStreamEnd: (event: StreamEndEvent) => Promise; + }; + + await internal.handleStreamEnd({ + type: "stream-end", + workspaceId: "childworkspace", + messageId: "msg_late_final", + metadata: { + model: "anthropic:claude-opus-4-6", + agentId: "exec", + finishReason: "stop", + muxMetadata, + }, + parts: [{ type: "text", text: "Late final text" }], + }); + + expect(await taskService.getWorkspaceTurnSnapshot(parentId, "wst_handle")).toMatchObject({ + status: "interrupted", + updatedAt: "2026-06-19T00:00:01.000Z", + }); + }); + + test("explicitly interrupted workspace turns are not revived by same-turn retry evidence", async () => { + const isStreaming = mock((workspaceId: string) => workspaceId === "childworkspace"); + const { config, parentId, taskService, historyService } = await startWorkspaceTurnForTest({ + isStreaming, + }); + const muxMetadata = { + type: "workspace-turn-task" as const, + taskHandleId: "wst_handle", + ownerWorkspaceId: parentId, + turnId: "turn", + }; + expect( + ( + await historyService.appendToHistory( + "childworkspace", + createMuxMessage("msg_prompt", "user", "Summarize", { muxMetadata }) + ) + ).success + ).toBe(true); + 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, + }); + const internal = taskService as unknown as { + activeWorkspaceTurnHandleByWorkspaceId: Map< + string, + { handleId: string; ownerWorkspaceId: string } + >; + }; + internal.activeWorkspaceTurnHandleByWorkspaceId.clear(); + + expect(await taskService.getWorkspaceTurnSnapshot(parentId, "wst_handle")).toMatchObject({ + status: "interrupted", + updatedAt: "2026-06-19T00:00:01.000Z", + }); + }); + + test("correlated stream-end never overwrites a completed workspace turn", async () => { + const { config, parentId, taskService } = await startWorkspaceTurnForTest(); + await new TaskHandleStore(config).upsertWorkspaceTurn({ + kind: "workspace_turn", + handleId: "wst_handle", + ownerWorkspaceId: parentId, + workspaceId: "childworkspace", + turnId: "turn", + status: "completed", + createdAt: "2026-06-19T00:00:00.000Z", + updatedAt: "2026-06-19T00:00:01.000Z", + createdWorkspace: true, + disposableWorkspace: false, + messageId: "msg_first", + reportMarkdown: "First result", + }); + const muxMetadata = { + type: "workspace-turn-task" as const, + taskHandleId: "wst_handle", + ownerWorkspaceId: parentId, + turnId: "turn", + }; + const internal = taskService as unknown as { + handleStreamEnd: (event: StreamEndEvent) => Promise; + }; + + await internal.handleStreamEnd({ + type: "stream-end", + workspaceId: "childworkspace", + messageId: "msg_second", + metadata: { + model: "anthropic:claude-opus-4-6", + agentId: "exec", + finishReason: "stop", + muxMetadata, + }, + parts: [{ type: "text", text: "Second result" }], + }); + + expect(await taskService.getWorkspaceTurnSnapshot(parentId, "wst_handle")).toMatchObject({ + status: "completed", + messageId: "msg_first", + reportMarkdown: "First result", + }); + }); + + test("getWorkspaceTurnSnapshot repairs a stale error handle from self-healed history", async () => { + const { config, parentId, taskService, historyService } = await startWorkspaceTurnForTest(); + const muxMetadata = { + type: "workspace-turn-task" as const, + taskHandleId: "wst_handle", + ownerWorkspaceId: parentId, + turnId: "turn", + }; + const appendResult = await historyService.appendToHistory( + "childworkspace", + createMuxMessage("msg_selfhealed", "assistant", "Self-healed final text", { + model: "anthropic:claude-opus-4-6", + agentId: "exec", + finishReason: "stop", + muxMetadata, + }) + ); + expect(appendResult.success).toBe(true); + await new TaskHandleStore(config).upsertWorkspaceTurn({ + kind: "workspace_turn", + handleId: "wst_handle", + ownerWorkspaceId: parentId, + workspaceId: "childworkspace", + turnId: "turn", + status: "error", + createdAt: "2026-06-19T00:00:00.000Z", + updatedAt: "2026-06-19T00:00:01.000Z", + createdWorkspace: true, + disposableWorkspace: false, + error: "Stream error: provider overloaded", + }); + + // List paths skip history repair for settled handles (no runtime activity), so the + // stale record stays visible there until a snapshot read reconciles it. + const listed = await taskService.listWorkspaceTurnTasks(parentId, { statuses: ["error"] }); + expect(listed.map((record) => record.handleId)).toContain("wst_handle"); + + const snapshot = await taskService.getWorkspaceTurnSnapshot(parentId, "wst_handle"); + expect(snapshot).toMatchObject({ + status: "completed", + messageId: "msg_selfhealed", + reportMarkdown: "Self-healed final text", + }); + expect(snapshot?.error).toBeUndefined(); + }); + + test("getWorkspaceTurnSnapshot revives an interrupted handle while the child retries the same turn", async () => { + const hasPendingAutoRetry = mock((workspaceId: string) => workspaceId === "childworkspace"); + const { config, parentId, taskService, historyService } = await startWorkspaceTurnForTest({ + hasPendingAutoRetry, + }); + const muxMetadata = { + type: "workspace-turn-task" as const, + taskHandleId: "wst_handle", + ownerWorkspaceId: parentId, + turnId: "turn", + }; + const appendResult = await historyService.appendToHistory( + "childworkspace", + createMuxMessage("msg_prompt", "user", "Summarize", { muxMetadata }) + ); + expect(appendResult.success).toBe(true); + 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, + error: "Workspace turn interrupted after restart", + terminalAttentionNotifiedAt: "2026-06-19T00:00:02.000Z", + }); + // Delivered tombstone from the stale settlement; revive must clear it so the revived + // turn's eventual real settlement can enqueue a fresh wake-up. + const terminalAttentionStore = new TerminalAttentionStore(config); + await terminalAttentionStore.enqueueIfAbsent({ + ownerWorkspaceId: parentId, + sourceKind: "workspace_turn", + sourceId: "wst_handle", + outputDelivery: "requires_task_await", + terminalOutcome: "interrupted", + }); + await terminalAttentionStore.markDelivered(parentId, "workspace_turn:wst_handle"); + const internal = taskService as unknown as { + activeWorkspaceTurnHandleByWorkspaceId: Map< + string, + { handleId: string; ownerWorkspaceId: string } + >; + }; + internal.activeWorkspaceTurnHandleByWorkspaceId.clear(); + + const snapshot = await taskService.getWorkspaceTurnSnapshot(parentId, "wst_handle"); + expect(snapshot).toMatchObject({ status: "running", workspaceId: "childworkspace" }); + expect(snapshot?.error).toBeUndefined(); + expect(snapshot?.terminalAttentionNotifiedAt).toBeUndefined(); + expect(await terminalAttentionStore.get(parentId, "workspace_turn:wst_handle")).toBeNull(); + expect(internal.activeWorkspaceTurnHandleByWorkspaceId.get("childworkspace")).toEqual({ + handleId: "wst_handle", + ownerWorkspaceId: parentId, + }); + }); + + test("history repair of a stale error handle waits for active child background work", async () => { + const { config, parentId, projectPath, taskService, historyService } = + await startWorkspaceTurnForTest(); + // The retried turn emitted its correlated final while a descendant task was still + // running; reporting completed before it finishes would hand the parent an + // incomplete result that active handles avoid via deferred stream-ends. Instead the + // handle is revived with the final recorded as deferred, then settles once the + // blocker is gone. + 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", + }; + expect( + ( + await historyService.appendToHistory( + "childworkspace", + createMuxMessage("msg_blocked_final", "assistant", "Blocked final text", { + model: "anthropic:claude-opus-4-6", + agentId: "exec", + finishReason: "stop", + muxMetadata, + }) + ) + ).success + ).toBe(true); + await new TaskHandleStore(config).upsertWorkspaceTurn({ + kind: "workspace_turn", + handleId: "wst_handle", + ownerWorkspaceId: parentId, + workspaceId: "childworkspace", + turnId: "turn", + status: "error", + createdAt: "2026-06-19T00:00:00.000Z", + updatedAt: "2026-06-19T00:00:01.000Z", + createdWorkspace: true, + disposableWorkspace: false, + error: "Stream error: provider overloaded", + }); + + const blocked = await taskService.getWorkspaceTurnSnapshot(parentId, "wst_handle"); + expect(blocked).toMatchObject({ + status: "running", + deferredMessageIds: ["msg_blocked_final"], + }); + expect(blocked?.error).toBeUndefined(); + expect(blocked?.reportMarkdown).toBeUndefined(); + + 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; + }); + + expect(await taskService.getWorkspaceTurnSnapshot(parentId, "wst_handle")).toMatchObject({ + status: "completed", + messageId: "msg_blocked_final", + reportMarkdown: "Blocked final text", + }); + }); + + test("turn-end blocker scan keeps a stale handle live between retry streams via child blockers", async () => { + const { config, parentId, projectPath, taskService, historyService } = + await startWorkspaceTurnForTest(); + // Codex handoff gap: the retried child's stream ended, no auto-retry is pending, but + // its descendant work is still running. The blocker scan must still treat the turn as + // live so the parent cannot end its turn during that window. + 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", + }; + expect( + ( + await historyService.appendToHistory( + "childworkspace", + createMuxMessage("msg_prompt", "user", "Summarize", { muxMetadata }) + ) + ).success + ).toBe(true); + await new TaskHandleStore(config).upsertWorkspaceTurn({ + kind: "workspace_turn", + handleId: "wst_handle", + ownerWorkspaceId: parentId, + workspaceId: "childworkspace", + turnId: "turn", + status: "error", + createdAt: "2026-06-19T00:00:00.000Z", + updatedAt: "2026-06-19T00:00:01.000Z", + createdWorkspace: true, + disposableWorkspace: false, + error: "Stream error: provider overloaded", + }); + const internal = taskService as unknown as { + activeWorkspaceTurnHandleByWorkspaceId: Map< + string, + { handleId: string; ownerWorkspaceId: string } + >; + listActiveWorkspaceTurnTaskIdsForOwner: (ownerWorkspaceId: string) => Promise; + }; + internal.activeWorkspaceTurnHandleByWorkspaceId.clear(); + + expect(await internal.listActiveWorkspaceTurnTaskIdsForOwner(parentId)).toContain("wst_handle"); + expect(await taskService.getWorkspaceTurnSnapshot(parentId, "wst_handle")).toMatchObject({ + status: "running", + workspaceId: "childworkspace", + }); + }); + + test("turn-end blocker scan revives and includes a stale retrying handle", async () => { + const hasPendingAutoRetry = mock((workspaceId: string) => workspaceId === "childworkspace"); + const { config, parentId, taskService, historyService } = await startWorkspaceTurnForTest({ + hasPendingAutoRetry, + }); + const muxMetadata = { + type: "workspace-turn-task" as const, + taskHandleId: "wst_handle", + ownerWorkspaceId: parentId, + turnId: "turn", + }; + expect( + ( + await historyService.appendToHistory( + "childworkspace", + createMuxMessage("msg_prompt", "user", "Summarize", { muxMetadata }) + ) + ).success + ).toBe(true); + await new TaskHandleStore(config).upsertWorkspaceTurn({ + kind: "workspace_turn", + handleId: "wst_handle", + ownerWorkspaceId: parentId, + workspaceId: "childworkspace", + turnId: "turn", + status: "error", + createdAt: "2026-06-19T00:00:00.000Z", + updatedAt: "2026-06-19T00:00:01.000Z", + createdWorkspace: true, + disposableWorkspace: false, + error: "Stream error: provider overloaded", + }); + const internal = taskService as unknown as { + activeWorkspaceTurnHandleByWorkspaceId: Map< + string, + { handleId: string; ownerWorkspaceId: string } + >; + listActiveWorkspaceTurnTaskIdsForOwner: (ownerWorkspaceId: string) => Promise; + }; + internal.activeWorkspaceTurnHandleByWorkspaceId.clear(); + + // The parent turn-end path must treat the stale-but-retrying handle as live work so + // the parent cannot end its turn while the child is still running. + expect(await internal.listActiveWorkspaceTurnTaskIdsForOwner(parentId)).toContain("wst_handle"); + expect(await taskService.getWorkspaceTurnSnapshot(parentId, "wst_handle")).toMatchObject({ + status: "running", + workspaceId: "childworkspace", + }); + }); + + test("active-only listWorkspaceTurnTasks revives and includes a stale retrying handle", async () => { + const hasPendingAutoRetry = mock((workspaceId: string) => workspaceId === "childworkspace"); + const { config, parentId, taskService, historyService } = await startWorkspaceTurnForTest({ + hasPendingAutoRetry, + }); + const muxMetadata = { + type: "workspace-turn-task" as const, + taskHandleId: "wst_handle", + ownerWorkspaceId: parentId, + turnId: "turn", + }; + expect( + ( + await historyService.appendToHistory( + "childworkspace", + createMuxMessage("msg_prompt", "user", "Summarize", { muxMetadata }) + ) + ).success + ).toBe(true); + await new TaskHandleStore(config).upsertWorkspaceTurn({ + kind: "workspace_turn", + handleId: "wst_handle", + ownerWorkspaceId: parentId, + workspaceId: "childworkspace", + turnId: "turn", + status: "error", + createdAt: "2026-06-19T00:00:00.000Z", + updatedAt: "2026-06-19T00:00:01.000Z", + createdWorkspace: true, + disposableWorkspace: false, + error: "Stream error: provider overloaded", + }); + const internal = taskService as unknown as { + activeWorkspaceTurnHandleByWorkspaceId: Map< + string, + { handleId: string; ownerWorkspaceId: string } + >; + }; + internal.activeWorkspaceTurnHandleByWorkspaceId.clear(); + + // task_list defaults to active statuses; the status filter applies AFTER + // normalization, so the stale-but-retrying handle is revived and reported as + // running instead of silently disappearing from the active view. + const listed = await taskService.listWorkspaceTurnTasks(parentId, { + statuses: ["queued", "starting", "running"], + }); + expect(listed.map((record) => record.handleId)).toContain("wst_handle"); + expect(listed.find((record) => record.handleId === "wst_handle")?.status).toBe("running"); + }); + + test("queued manual input does not revive a settled workspace turn", async () => { + // Ordinary queued input is not yet in history, so the newest-correlated-prompt guard + // cannot see it; the liveness gate must not treat it as a same-turn continuation. + const hasQueuedMessages = mock((workspaceId: string) => workspaceId === "childworkspace"); + const hasPendingQueuedOrPreparingTurn = mock( + (workspaceId: string) => workspaceId === "childworkspace" + ); + const { config, parentId, taskService, historyService } = await startWorkspaceTurnForTest({ + hasQueuedMessages, + hasPendingQueuedOrPreparingTurn, + }); + const muxMetadata = { + type: "workspace-turn-task" as const, + taskHandleId: "wst_handle", + ownerWorkspaceId: parentId, + turnId: "turn", + }; + expect( + ( + await historyService.appendToHistory( + "childworkspace", + createMuxMessage("msg_prompt", "user", "Summarize", { muxMetadata }) + ) + ).success + ).toBe(true); + await new TaskHandleStore(config).upsertWorkspaceTurn({ + kind: "workspace_turn", + handleId: "wst_handle", + ownerWorkspaceId: parentId, + workspaceId: "childworkspace", + turnId: "turn", + status: "error", + createdAt: "2026-06-19T00:00:00.000Z", + updatedAt: "2026-06-19T00:00:01.000Z", + createdWorkspace: true, + disposableWorkspace: false, + error: "Stream error: provider overloaded", + }); + const internal = taskService as unknown as { + activeWorkspaceTurnHandleByWorkspaceId: Map< + string, + { handleId: string; ownerWorkspaceId: string } + >; + listActiveWorkspaceTurnTaskIdsForOwner: (ownerWorkspaceId: string) => Promise; + }; + internal.activeWorkspaceTurnHandleByWorkspaceId.clear(); + + expect(await internal.listActiveWorkspaceTurnTaskIdsForOwner(parentId)).not.toContain( + "wst_handle" + ); + expect(await taskService.getWorkspaceTurnSnapshot(parentId, "wst_handle")).toMatchObject({ + status: "error", + error: "Stream error: provider overloaded", + }); + }); + + test("a newer unrelated child prompt does not revive a settled workspace turn", async () => { + const isStreaming = mock((workspaceId: string) => workspaceId === "childworkspace"); + const { config, parentId, taskService, historyService } = await startWorkspaceTurnForTest({ + isStreaming, + }); + const muxMetadata = { + type: "workspace-turn-task" as const, + taskHandleId: "wst_handle", + ownerWorkspaceId: parentId, + turnId: "turn", + }; + expect( + ( + await historyService.appendToHistory( + "childworkspace", + createMuxMessage("msg_prompt", "user", "Summarize", { muxMetadata }) + ) + ).success + ).toBe(true); + expect( + ( + await historyService.appendToHistory( + "childworkspace", + createMuxMessage("msg_manual", "user", "Manual follow-up", {}) + ) + ).success + ).toBe(true); + await new TaskHandleStore(config).upsertWorkspaceTurn({ + kind: "workspace_turn", + handleId: "wst_handle", + ownerWorkspaceId: parentId, + workspaceId: "childworkspace", + turnId: "turn", + status: "error", + createdAt: "2026-06-19T00:00:00.000Z", + updatedAt: "2026-06-19T00:00:01.000Z", + createdWorkspace: true, + disposableWorkspace: false, + error: "Stream error: provider overloaded", + }); + const internal = taskService as unknown as { + activeWorkspaceTurnHandleByWorkspaceId: Map< + string, + { handleId: string; ownerWorkspaceId: string } + >; + }; + internal.activeWorkspaceTurnHandleByWorkspaceId.clear(); + + expect(await taskService.getWorkspaceTurnSnapshot(parentId, "wst_handle")).toMatchObject({ + status: "error", + error: "Stream error: provider overloaded", + }); + }); + + test("revive does not clobber a newer same-status settlement written after the stale read", async () => { + const { config, parentId, taskService } = await startWorkspaceTurnForTest(); + const taskHandleStore = new TaskHandleStore(config); + // The record currently on disk: a FRESH error settled by the live retry itself. + const freshError = { + kind: "workspace_turn" as const, + handleId: "wst_handle", + ownerWorkspaceId: parentId, + workspaceId: "childworkspace", + turnId: "turn", + status: "error" as const, + createdAt: "2026-06-19T00:00:00.000Z", + updatedAt: "2026-06-19T00:05:00.000Z", + createdWorkspace: true, + disposableWorkspace: false, + error: "Stream error: retry also failed", + }; + await taskHandleStore.upsertWorkspaceTurn(freshError); + const internal = taskService as unknown as { + reviveRetryingWorkspaceTurn: (record: typeof freshError) => Promise; + }; + + // Reconcile observed an OLDER error record (same status, earlier updatedAt) before the + // retry failed; the revive must notice the newer settlement and leave it untouched. + const revived = await internal.reviveRetryingWorkspaceTurn({ + ...freshError, + updatedAt: "2026-06-19T00:00:01.000Z", + error: "Stream error: provider overloaded", + }); + + expect(revived).toMatchObject({ + status: "error", + updatedAt: "2026-06-19T00:05:00.000Z", + error: "Stream error: retry also failed", + }); + expect(await taskHandleStore.getWorkspaceTurn(parentId, "wst_handle")).toMatchObject({ + status: "error", + updatedAt: "2026-06-19T00:05:00.000Z", + error: "Stream error: retry also failed", + }); + }); + + test("history repair scans past newer unrelated prompts to a correlated final message", async () => { + const { config, parentId, taskService, historyService } = await startWorkspaceTurnForTest(); + const muxMetadata = { + type: "workspace-turn-task" as const, + taskHandleId: "wst_handle", + ownerWorkspaceId: parentId, + turnId: "turn", + }; + // The turn self-healed and finished, THEN the child received an unrelated manual + // prompt before the parent ever called task_await. + expect( + ( + await historyService.appendToHistory( + "childworkspace", + createMuxMessage("msg_selfhealed_final", "assistant", "Self-healed final text", { + model: "anthropic:claude-opus-4-6", + agentId: "exec", + finishReason: "stop", + muxMetadata, + }) + ) + ).success + ).toBe(true); + expect( + ( + await historyService.appendToHistory( + "childworkspace", + createMuxMessage("msg_manual_later", "user", "Manual follow-up", {}) + ) + ).success + ).toBe(true); + await new TaskHandleStore(config).upsertWorkspaceTurn({ + kind: "workspace_turn", + handleId: "wst_handle", + ownerWorkspaceId: parentId, + workspaceId: "childworkspace", + turnId: "turn", + status: "error", + createdAt: "2026-06-19T00:00:00.000Z", + updatedAt: "2026-06-19T00:00:01.000Z", + createdWorkspace: true, + disposableWorkspace: false, + error: "Stream error: provider overloaded", + }); + + const snapshot = await taskService.getWorkspaceTurnSnapshot(parentId, "wst_handle"); + expect(snapshot).toMatchObject({ + status: "completed", + messageId: "msg_selfhealed_final", + reportMarkdown: "Self-healed final text", + }); + expect(snapshot?.error).toBeUndefined(); + }); + test("workspace-turn stale recovery uses deferred history after archived descendants stop blocking", async () => { const { config, parentId, projectPath, taskService, historyService } = await startWorkspaceTurnForTest(); diff --git a/src/node/services/taskService.ts b/src/node/services/taskService.ts index a202597ece..52f84e69c1 100644 --- a/src/node/services/taskService.ts +++ b/src/node/services/taskService.ts @@ -689,6 +689,25 @@ const WORKSPACE_TURN_RECOVERABLE_STREAM_ERRORS: ReadonlySet = n "context_exceeded", ]); +/** Marker persisted by settleStaleWorkspaceTurn when restart recovery interrupts a handle. */ +const WORKSPACE_TURN_STALE_RESTART_ERROR = "Workspace turn interrupted after restart"; + +/** + * Settled workspace-turn records eligible for self-heal correction (resettle from a + * correlated stream-end, or read-time repair/revive): transient stream-error settlements + * (status "error") and stale restart-recovery interrupts. Explicit interrupts — user Esc, + * task_terminate, cancel reasons — must stay terminal even if a late correlated stream-end + * or same-turn retry evidence arrives, so canceled work never resurfaces as completed. + */ +function isSelfHealEligibleSettledWorkspaceTurn( + record: Pick +): boolean { + if (record.status === "error") { + return true; + } + return record.status === "interrupted" && record.error === WORKSPACE_TURN_STALE_RESTART_ERROR; +} + /** * A workspace-turn stream error may resolve without parent intervention when * the child can still make progress on its own. The caller must still confirm @@ -5227,6 +5246,13 @@ export class TaskService { waiterSettlement: | { status: "completed"; result: WorkspaceTurnWaitResult } | { status: "error"; error: Error }; + /** + * Allow replacing an already-settled interrupted/error record (never completed). + * Only the strictly turn-correlated stream-end path may set this: a handle that + * settled from a transient failure can self-heal when the child auto-retries the + * same turn, and the correlated stream-end proves the turn's real outcome. + */ + allowTerminalResettle?: boolean; }): Promise { assert( params.next.handleId === params.record.handleId, @@ -5242,7 +5268,7 @@ export class TaskService { const pendingNotify = await this.workspaceTurnSettlementLocks.withLock( params.record.handleId, async (): Promise< - | { kind: "notify"; outcome: TerminalAttentionOutcome; title?: string } + | { kind: "notify"; outcome: TerminalAttentionOutcome; title?: string; resettled: boolean } | { kind: "drain_pending" } | null > => { @@ -5258,7 +5284,17 @@ export class TaskService { "settleWorkspaceTurn requires current record to match workspaceId" ); - if (this.isTerminalWorkspaceTurnStatus(current.status)) { + // A completed record is immutable; a self-heal-eligible settled record (transient + // error / stale restart interrupt — never an explicit user interrupt) may be + // corrected once by an explicitly allowed resettle, but only when the new settlement + // actually changes the outcome (duplicate stream-end replays must stay idempotent). + const resettleStaleTerminal = + params.allowTerminalResettle === true && + this.isTerminalWorkspaceTurnStatus(current.status) && + current.status !== "completed" && + isSelfHealEligibleSettledWorkspaceTurn(current) && + (params.next.status !== current.status || params.next.messageId !== current.messageId); + if (this.isTerminalWorkspaceTurnStatus(current.status) && !resettleStaleTerminal) { const active = this.activeWorkspaceTurnHandleByWorkspaceId.get(params.record.workspaceId); if ( active?.handleId === params.record.handleId && @@ -5285,11 +5321,24 @@ export class TaskService { } // Decide the terminal wake-up using persisted policy + the restart-safe dedupe marker. + // A resettle corrects a previously reported outcome, so it re-arms the wake-up even if + // the stale settlement was already notified/consumed. const policy = resolveBackgroundWorkAttentionPolicy(current.attentionPolicy); const shouldNotify = - policy === "notify_on_terminal" && current.terminalAttentionNotifiedAt == null; - - await this.taskHandleStore.upsertWorkspaceTurn(params.next); + policy === "notify_on_terminal" && + (current.terminalAttentionNotifiedAt == null || resettleStaleTerminal); + + const nextRecord = { ...params.next }; + if (resettleStaleTerminal) { + log.debug("Workspace turn resettled from stale terminal status", { + handleId: current.handleId, + workspaceId: current.workspaceId, + staleStatus: current.status, + nextStatus: nextRecord.status, + }); + delete nextRecord.terminalAttentionNotifiedAt; + } + await this.taskHandleStore.upsertWorkspaceTurn(nextRecord); const active = this.activeWorkspaceTurnHandleByWorkspaceId.get(params.record.workspaceId); if ( active?.handleId === params.record.handleId && @@ -5302,7 +5351,7 @@ export class TaskService { params.waiterSettlement ); this.markTaskForegroundRelevant(params.record.handleId); - await this.cleanupDisposableWorkspaceTurn(params.next); + await this.cleanupDisposableWorkspaceTurn(nextRecord); this.scheduleMaybeStartQueuedTasks(); // A foreground waiter that received this terminal result already integrates it, so suppress @@ -5316,8 +5365,9 @@ export class TaskService { } return { kind: "notify", - outcome: workspaceTurnTerminalOutcome(params.next.status), - ...(params.next.title != null ? { title: params.next.title } : {}), + outcome: workspaceTurnTerminalOutcome(nextRecord.status), + resettled: resettleStaleTerminal, + ...(nextRecord.title != null ? { title: nextRecord.title } : {}), }; } ); @@ -5333,6 +5383,14 @@ export class TaskService { // Enqueue the terminal wake-up outside the lock. The persisted notification is the restart-safe // record of intent; only after it is accepted do we set terminalAttentionNotifiedAt on the // handle so a duplicate settlement / stale recovery cannot double-wake. + if (pendingNotify.resettled) { + // The stale settlement's wake-up may already be delivered/consumed; enqueueIfAbsent + // treats that tombstone as "already notified" and would swallow the corrected outcome. + await this.terminalAttentionStore.delete( + params.record.ownerWorkspaceId, + TerminalAttentionStore.notificationId("workspace_turn", params.record.handleId) + ); + } await this.enqueueTerminalAttention({ ownerWorkspaceId: params.record.ownerWorkspaceId, sourceKind: "workspace_turn", @@ -6161,7 +6219,16 @@ export class TaskService { } private async normalizeWorkspaceTurnRecord( - record: WorkspaceTurnTaskHandleRecord + record: WorkspaceTurnTaskHandleRecord, + options: { + /** + * Also reconcile settled interrupted/error records against the child's durable + * history (one history read per record). Enabled for single-handle snapshot reads + * (task_await) but not for list paths, which would pay that read for every + * historical terminal handle on each call. + */ + repairSettledTurnsFromHistory?: boolean; + } = {} ): Promise { assert(record.ownerWorkspaceId.length > 0, "normalizeWorkspaceTurnRecord requires owner id"); assert(record.handleId.length > 0, "normalizeWorkspaceTurnRecord requires handle id"); @@ -6171,7 +6238,7 @@ export class TaskService { // task_await agree on the self-healed terminal status. if ( record.status === "interrupted" && - record.error === "Workspace turn interrupted after restart" && + record.error === WORKSPACE_TURN_STALE_RESTART_ERROR && (record.deferredMessageIds?.length ?? 0) > 0 ) { const recovered = await this.recoverTerminalWorkspaceTurnFromHistory(record); @@ -6200,9 +6267,226 @@ export class TaskService { return await this.taskHandleStore.getWorkspaceTurn(record.ownerWorkspaceId, record.handleId); } + if ( + (record.status === "interrupted" || record.status === "error") && + isSelfHealEligibleSettledWorkspaceTurn(record) + ) { + return await this.reconcileSettledWorkspaceTurn(record, { + repairFromHistory: options.repairSettledTurnsFromHistory === true, + }); + } + + return record; + } + + /** + * Settled interrupted/error handles can go stale: a stream error or restart can settle + * the handle even though the child workspace self-heals by auto-retrying the same turn + * (retries replay the turn's synthetic prompt, so their output still correlates via + * muxMetadata). Reconcile against the child's live state at read time so task_await and + * task_list report reality instead of the stale settlement. + */ + private async reconcileSettledWorkspaceTurn( + record: WorkspaceTurnTaskHandleRecord, + options: { repairFromHistory: boolean } + ): Promise { + assert( + record.status === "interrupted" || record.status === "error", + "reconcileSettledWorkspaceTurn requires a settled interrupted/error record" + ); + // Lazily computed and cached: whether the child still has active descendant/workflow/ + // nested-turn work. A settled record cannot accumulate deferredMessageIds (the deferral + // path skips inactive handles), so the repair/deferred decisions below must consult + // these blockers — otherwise a retried turn could be reported completed while its + // background work is still running. + let blockersActive: boolean | null = null; + const deferredBlockersActive = async (): Promise => { + blockersActive ??= await this.hasActiveWorkspaceTurnDeferredBlockers(record); + return blockersActive; + }; + // Evidence the turn is still live. All checks are in-memory so historical terminal + // handles stay cheap on list/blocker-scan paths (no history reads). Queued/preparing + // turns are deliberately excluded: ordinary queued manual input is not yet in history, + // so it would defeat the newest-correlated-prompt guard below and revive unrelated + // work. The synthetic background-await continuation between retry streams is instead + // covered by the descendant hint — it is only queued while such blockers are active. + const turnLive = + this.aiService.isStreaming(record.workspaceId) || + this.workspaceService.hasPendingAutoRetry(record.workspaceId) || + this.hasActiveDescendantAgentTasks(this.config.loadConfigOrDefault(), record.workspaceId); + if (!turnLive && !options.repairFromHistory) { + return record; + } + + const historyResult = await this.historyService.getHistoryFromLatestBoundary( + record.workspaceId + ); + if (!historyResult.success) { + return record; + } + // Auto-retry replays the newest accepted user message, so live child activity belongs + // to this turn only while its prompt is still that newest message. Any newer user + // message disables the revive — but only the revive: a correlated final assistant + // message older than that unrelated prompt still proves the turn completed, so the + // durable-history repair scan must continue past it. + let reviveAllowed = turnLive; + for (const message of historyResult.data.toReversed()) { + if ( + this.isDeferredWorkspaceTurnMessage(record, message.id) && + (await deferredBlockersActive()) + ) { + continue; + } + const event = this.buildWorkspaceTurnStreamEndEventFromHistory(record, message); + if (event != null) { + // The turn produced a correlated final assistant message after settlement, so the + // child self-healed. While its background work is still running, revive the handle + // with the final recorded as deferred: the parent stays blocked, and the standard + // deferred-recovery machinery settles the true outcome once blockers finish. + if (reviveAllowed && (await deferredBlockersActive())) { + return await this.reviveRetryingWorkspaceTurn(record, { + deferredMessageIds: [event.messageId], + }); + } + const recovered = this.buildTerminalWorkspaceTurnRecordFromEvent(record, event); + if ( + !options.repairFromHistory || + (recovered.status === record.status && recovered.messageId === record.messageId) + ) { + return record; + } + if (await deferredBlockersActive()) { + return record; + } + return await this.persistRepairedSettledWorkspaceTurn(record, recovered); + } + if (message.role !== "user") { + continue; + } + const metadata = this.getWorkspaceTurnMetadataFromValue(message.metadata?.muxMetadata); + const correlatedPrompt = + metadata != null && + metadata.taskHandleId === record.handleId && + metadata.ownerWorkspaceId === record.ownerWorkspaceId && + metadata.turnId === record.turnId; + if (correlatedPrompt) { + if (reviveAllowed) { + return await this.reviveRetryingWorkspaceTurn(record); + } + // A correlated final must be newer than the prompt; nothing older can repair. + return record; + } + reviveAllowed = false; + if (!options.repairFromHistory) { + return record; + } + } return record; } + private async persistRepairedSettledWorkspaceTurn( + record: WorkspaceTurnTaskHandleRecord, + recovered: WorkspaceTurnTaskHandleRecord + ): Promise { + const next = await this.workspaceTurnSettlementLocks.withLock(record.handleId, async () => { + const current = await this.taskHandleStore.getWorkspaceTurn( + record.ownerWorkspaceId, + record.handleId + ); + // A concurrent settlement/repair wins; only replace the exact record we reconciled. + // Comparing updatedAt (not just status) matters: a concurrent settlement can produce + // a NEWER record with the same status that must not be clobbered by our stale read. + if ( + current == null || + current.status !== record.status || + current.updatedAt !== record.updatedAt + ) { + return current; + } + log.debug("Workspace turn repaired from self-healed child history", { + handleId: record.handleId, + workspaceId: record.workspaceId, + staleStatus: record.status, + nextStatus: recovered.status, + }); + await this.taskHandleStore.upsertWorkspaceTurn(recovered); + return recovered; + }); + if (next === 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 next; + } + + private async reviveRetryingWorkspaceTurn( + record: WorkspaceTurnTaskHandleRecord, + options: { deferredMessageIds?: string[] } = {} + ): Promise { + return await this.workspaceTurnSettlementLocks.withLock(record.handleId, async () => { + const current = await this.taskHandleStore.getWorkspaceTurn( + record.ownerWorkspaceId, + record.handleId + ); + // A concurrent transition wins; only revive the exact record we reconciled against. + // Comparing updatedAt (not just status) matters: the live retry itself can fail and + // settle a NEWER record with the same status (e.g. error → error) between our read + // and this lock — reviving that fresh terminal failure would strand task_await. + if ( + current == null || + current.status !== record.status || + current.updatedAt !== record.updatedAt + ) { + return current; + } + // Another turn already owns the child workspace; the activity is not this turn's retry. + const active = this.activeWorkspaceTurnHandleByWorkspaceId.get(record.workspaceId); + if ( + active != null && + (active.handleId !== record.handleId || active.ownerWorkspaceId !== record.ownerWorkspaceId) + ) { + return current; + } + const next: WorkspaceTurnTaskHandleRecord = { + ...current, + status: "running", + updatedAt: getIsoNow(), + // An already-observed correlated final (blocked on child background work) rides + // along as deferred so standard deferred recovery settles it once blockers finish. + ...(options.deferredMessageIds != null + ? { deferredMessageIds: options.deferredMessageIds } + : {}), + }; + delete next.error; + // The revived turn's next terminal transition is a new outcome; re-arm its wake-up. + // The notification tombstone must go too: enqueueIfAbsent would otherwise treat the + // stale settlement's delivered wake-up as "already notified" and swallow the new one. + delete next.terminalAttentionNotifiedAt; + await this.terminalAttentionStore.delete( + record.ownerWorkspaceId, + TerminalAttentionStore.notificationId("workspace_turn", record.handleId) + ); + await this.taskHandleStore.upsertWorkspaceTurn(next); + // Re-register so stream-end/abort/error settlement paths own the handle again. + this.activeWorkspaceTurnHandleByWorkspaceId.set(record.workspaceId, { + handleId: record.handleId, + ownerWorkspaceId: record.ownerWorkspaceId, + }); + log.debug("Workspace turn revived: child is retrying the same turn", { + handleId: record.handleId, + workspaceId: record.workspaceId, + staleStatus: record.status, + }); + return next; + }); + } + async getWorkspaceTurnSnapshot( ownerWorkspaceId: string, handleId: string @@ -6214,7 +6498,11 @@ export class TaskService { if (record == null) { return null; } - return await this.normalizeWorkspaceTurnRecord(record); + // Snapshot reads back task_await, which must report the child's live state even when + // a stale settlement (interrupted/error) was later corrected by a self-healed retry. + return await this.normalizeWorkspaceTurnRecord(record, { + repairSettledTurnsFromHistory: true, + }); } async listWorkspaceTurnTasks( @@ -7113,14 +7401,14 @@ export class TaskService { ...record, status: "interrupted", updatedAt: getIsoNow(), - error: "Workspace turn interrupted after restart", + error: WORKSPACE_TURN_STALE_RESTART_ERROR, }; await this.settleWorkspaceTurn({ record, next, waiterSettlement: { status: "error", - error: new Error("Workspace turn interrupted after restart"), + error: new Error(WORKSPACE_TURN_STALE_RESTART_ERROR), }, }); } @@ -7164,23 +7452,41 @@ export class TaskService { private async listActiveWorkspaceTurnTaskIdsForOwner( ownerWorkspaceId: string ): Promise { - const records = await this.taskHandleStore.listWorkspaceTurns(ownerWorkspaceId, { - statuses: ["queued", "starting", "running"], - }); + const records = await this.taskHandleStore.listWorkspaceTurns(ownerWorkspaceId); const taskIds: string[] = []; for (const record of records) { if ( - record.status !== "queued" && - record.status !== "starting" && - record.status !== "running" + record.status === "queued" || + record.status === "starting" || + record.status === "running" ) { + if (!(await this.isLiveWorkspaceTurn(record))) { + await this.settleStaleWorkspaceTurn(record); + continue; + } + taskIds.push(record.handleId); continue; } - if (!(await this.isLiveWorkspaceTurn(record))) { - await this.settleStaleWorkspaceTurn(record); - continue; + // A stale settled handle whose child is actively retrying the same turn is live + // delegated work: revive it here too, so the parent's turn-end blocker scan cannot + // finish while the child is still running. Cheap for historical handles — reconcile + // short-circuits on in-memory runtime checks before touching history. + if ( + (record.status === "interrupted" || record.status === "error") && + isSelfHealEligibleSettledWorkspaceTurn(record) + ) { + const latest = await this.reconcileSettledWorkspaceTurn(record, { + repairFromHistory: false, + }); + if ( + latest != null && + (latest.status === "queued" || + latest.status === "starting" || + latest.status === "running") + ) { + taskIds.push(latest.handleId); + } } - taskIds.push(record.handleId); } return taskIds; } @@ -8405,6 +8711,11 @@ export class TaskService { next.status === "completed" ? { status: "completed", result: this.buildWorkspaceTurnWaitResult(next) } : { status: "error", error: new Error(next.error ?? "Workspace turn failed") }, + // This stream-end is strictly correlated (workspaceId + turnId), so it proves the + // delegated turn's real outcome. A handle that settled interrupted/error from a + // transient failure (provider error, restart) may have self-healed via auto-retry + // of the same turn; let this settlement correct that stale record. + allowTerminalResettle: true, }); return true; }