diff --git a/src/node/services/agentSession.queueDispatch.test.ts b/src/node/services/agentSession.queueDispatch.test.ts index fa4a827d7b..9b45bda226 100644 --- a/src/node/services/agentSession.queueDispatch.test.ts +++ b/src/node/services/agentSession.queueDispatch.test.ts @@ -514,6 +514,41 @@ describe("AgentSession queued message tool-call dispatch", () => { } }); + test("intentional dedupe cleanup can remove a synthetic entry without canceling its owner", async () => { + const workspaceId = "queue-dispatch-silent-dedupe-removal"; + const { session, cleanup } = await createAgentSessionHarness({ workspaceId }); + + try { + const canceledReasons: string[] = []; + session.queueMessage( + "Incremental report", + { model: TEST_MODEL, agentId: "exec" }, + { + synthetic: true, + agentInitiated: true, + dedupeKey: "agent-report:child:progress", + removableDedupeKey: true, + onCanceled: (reason) => { + canceledReasons.push(reason); + }, + } + ); + + expect( + session.removeQueuedMessagesByDedupeKeyPrefix( + "agent-report:child:", + "Terminal report replaced progress.", + { notifyCancellation: false } + ) + ).toBe(1); + expect(canceledReasons).toEqual([]); + expect(session.hasQueuedMessages()).toBe(false); + } finally { + session.dispose(); + await cleanup(); + } + }); + test("cancel signal retracts a synthetic entry after dequeue while history append is preparing", async () => { const workspaceId = "queue-dispatch-cancel-preparing"; const { session, cleanup, historyService, events } = await createAgentSessionHarness({ diff --git a/src/node/services/agentSession.ts b/src/node/services/agentSession.ts index 2359feab9d..8398ad4382 100644 --- a/src/node/services/agentSession.ts +++ b/src/node/services/agentSession.ts @@ -49,6 +49,7 @@ import { createFileSnapshotMessageId, createAgentSkillSnapshotMessageId, createMcpPromptSnapshotMessageId, + FILE_CHANGE_NOTIFICATION_MESSAGE_ID_PREFIX, } from "@/node/services/utils/messageIds"; import { FileChangeTracker, @@ -346,12 +347,11 @@ function hasSameWorkspaceTurnCorrelation( * and the turn's real outcome can never settle the task handle (see * TaskService.finalizeWorkspaceTurnFromStreamEnd). * - * Scans newest→oldest: interleaved monitor wakes keep the chain open; any - * other user input (manual prompt, new workspace-turn prompt) supersedes the - * turn, and only a correlated assistant message that ended with "tool-calls" - * (a queue-dispatch cut) leaves the turn open. The inherited metadata is - * persisted on each continuation's assistant message, so chains survive - * restarts. + * Scans newest→oldest. Monitor wakes, file-change notices, and correlated + * nested reports keep the chain open. Manual prompts supersede the turn. + * Only a correlated assistant message that ended with "tool-calls" leaves the + * older chain open. Each continuation persists the inherited metadata, so the + * chain survives restarts. */ export function inheritOpenWorkspaceTurnMetadata( messages: readonly MuxMessage[] @@ -384,6 +384,19 @@ export function inheritOpenWorkspaceTurnMetadata( if (muxMetadata?.type === "bash-monitor-wake") { continue; } + // File-change rows are machine context for the pending continuation. + // They do not replace the delegated request that caused the stream. + if ( + message.metadata?.synthetic === true && + message.id.startsWith(FILE_CHANGE_NOTIFICATION_MESSAGE_ID_PREFIX) + ) { + continue; + } + // A correlated nested report can queue before a monitor wake. It continues + // the same delegated turn and is stronger evidence than the older cut. + if (muxMetadata?.type === "workspace-turn-task") { + return muxMetadata; + } return undefined; } } @@ -5517,6 +5530,8 @@ export class AgentSession { onAccepted?: () => Promise | void; onAcceptedPreStreamFailure?: (error: SendMessageError) => Promise | void; onCanceled?: (reason: string) => Promise | void; + onDeliveryAcceptedPreStreamFailure?: (error: SendMessageError) => Promise | void; + onDeliveryCanceled?: (reason: string) => Promise | void; cancelState?: { canceledBeforeAcceptance: boolean }; cancelSignal?: AbortSignal; } @@ -5590,7 +5605,11 @@ export class AgentSession { }); } - removeQueuedMessagesByDedupeKeyPrefix(prefix: string, cancelReason: string): number { + removeQueuedMessagesByDedupeKeyPrefix( + prefix: string, + cancelReason: string, + options?: { notifyCancellation?: boolean } + ): number { this.assertNotDisposed("removeQueuedMessagesByDedupeKeyPrefix"); assert(prefix.length > 0, "removeQueuedMessagesByDedupeKeyPrefix requires prefix"); const removal = this.messageQueue.removeByDedupeKeyPrefix(prefix); @@ -5602,8 +5621,10 @@ export class AgentSession { this.workspaceId, !this.messageQueue.isEmpty() && this.messageQueue.getNextQueueDispatchMode() === "tool-end" ); - for (const callbacks of removal.callbacks) { - this.notifyQueuedMessageCleared(callbacks, cancelReason); + if (options?.notifyCancellation !== false) { + for (const callbacks of removal.callbacks) { + this.notifyQueuedMessageCleared(callbacks, cancelReason); + } } return removal.removedCount; } diff --git a/src/node/services/agentSession.workspaceTurnInheritance.test.ts b/src/node/services/agentSession.workspaceTurnInheritance.test.ts index c19efecd02..28ea263f36 100644 --- a/src/node/services/agentSession.workspaceTurnInheritance.test.ts +++ b/src/node/services/agentSession.workspaceTurnInheritance.test.ts @@ -48,6 +48,30 @@ describe("inheritOpenWorkspaceTurnMetadata", () => { expect(inheritOpenWorkspaceTurnMetadata(messages)).toEqual(correlation); }); + test("a file-change row after a monitor wake keeps the turn open", () => { + const messages = [ + turnPrompt("prompt"), + cutAssistant("cut"), + wake("wake"), + createMuxMessage("file-change-1", "user", "", { + synthetic: true, + }), + ]; + expect(inheritOpenWorkspaceTurnMetadata(messages)).toEqual(correlation); + }); + + test("a correlated nested report before a monitor wake keeps the turn open", () => { + const messages = [ + turnPrompt("prompt"), + cutAssistant("cut"), + createMuxMessage("nested-report", "user", "Nested task completed", { + muxMetadata: correlation, + }), + wake("wake"), + ]; + expect(inheritOpenWorkspaceTurnMetadata(messages)).toEqual(correlation); + }); + test("a correlated assistant that finished with stop closes the turn", () => { const messages = [ turnPrompt("prompt"), diff --git a/src/node/services/messageQueue.test.ts b/src/node/services/messageQueue.test.ts index 0373f37ca5..ffeba8af66 100644 --- a/src/node/services/messageQueue.test.ts +++ b/src/node/services/messageQueue.test.ts @@ -1,6 +1,7 @@ -import { describe, it, expect, beforeEach } from "bun:test"; +import { describe, it, expect, beforeEach, mock } from "bun:test"; import { MessageQueue } from "./messageQueue"; import type { MuxMessageMetadata } from "@/common/types/message"; +import type { SendMessageError } from "@/common/types/errors"; import type { SendMessageOptions } from "@/common/orpc/types"; describe("MessageQueue", () => { @@ -504,6 +505,40 @@ describe("MessageQueue", () => { expect(second.internal?.onAcceptedPreStreamFailure).toBeUndefined(); }); + it("should preserve delivery callbacks when reordering strips correlation", async () => { + const onCanceled = mock(() => undefined); + const onAcceptedPreStreamFailure = mock(() => undefined); + const onDeliveryCanceled = mock(() => undefined); + const onDeliveryAcceptedPreStreamFailure = mock(() => undefined); + queue.add( + "Background report", + { model: "gpt-4", agentId: "exec", muxMetadata: metadata }, + { + synthetic: true, + agentInitiated: true, + workspaceTurnContinuation: true, + onCanceled, + onAcceptedPreStreamFailure, + onDeliveryCanceled, + onDeliveryAcceptedPreStreamFailure, + } + ); + queue.add("User send now", { model: "gpt-4", agentId: "exec" }); + + expect(queue.setVisibleQueueDispatchMode("tool-end")).toBe(true); + queue.dequeueNext(); + const reordered = queue.dequeueNext(); + const error: SendMessageError = { type: "unknown", raw: "startup failed" }; + await reordered.internal?.onCanceled?.("cleared"); + await reordered.internal?.onAcceptedPreStreamFailure?.(error); + + expect(reordered.options?.muxMetadata).toBeUndefined(); + expect(onCanceled).not.toHaveBeenCalled(); + expect(onAcceptedPreStreamFailure).not.toHaveBeenCalled(); + expect(onDeliveryCanceled).toHaveBeenCalledWith("cleared"); + expect(onDeliveryAcceptedPreStreamFailure).toHaveBeenCalledWith(error); + }); + it("should preserve an original queued workspace-turn prompt during reordering", () => { const onAccepted = () => undefined; const onCanceled = () => undefined; @@ -567,6 +602,17 @@ describe("MessageQueue", () => { expect(queue.getMessages()).toEqual(["User message before", "User message after"]); }); + it("removeWorkspaceTurn reports removal when the entry has no callbacks", () => { + queue.add( + "Follow up without callbacks", + { model: "gpt-4", agentId: "exec", muxMetadata: metadata }, + { agentInitiated: true, workspaceTurnContinuation: true } + ); + + expect(queue.removeWorkspaceTurn("wst_followup")).toEqual({}); + expect(queue.hasWorkspaceTurn("wst_followup")).toBe(false); + }); + it("should report clear callbacks for every pending entry", () => { const onCanceledFirst = () => undefined; const onCanceledSecond = () => undefined; diff --git a/src/node/services/messageQueue.ts b/src/node/services/messageQueue.ts index 83091339cf..a01cef9f2f 100644 --- a/src/node/services/messageQueue.ts +++ b/src/node/services/messageQueue.ts @@ -90,8 +90,14 @@ interface QueuedMessageInternalOptions { /** Dedupe-keyed maintenance sends are removable by prefix without changing global queue rules. */ removableDedupeKey?: boolean; onAccepted?: () => Promise | void; + /** Correlation callback. Queue reordering can remove it with workspace-turn metadata. */ onAcceptedPreStreamFailure?: (error: SendMessageError) => Promise | void; + /** Correlation callback. Queue reordering can remove it with workspace-turn metadata. */ onCanceled?: (reason: string) => Promise | void; + /** Delivery callback. It survives workspace-turn correlation removal. */ + onDeliveryAcceptedPreStreamFailure?: (error: SendMessageError) => Promise | void; + /** Delivery callback. It survives workspace-turn correlation removal. */ + onDeliveryCanceled?: (reason: string) => Promise | void; /** Mutable dispatch outcome shared with sendQueuedMessages. */ cancelState?: { canceledBeforeAcceptance: boolean }; /** Cancels a queued entry even after it has been dequeued into PREPARING. */ @@ -136,10 +142,46 @@ interface QueueEntry { onCanceled?: (reason: string) => Promise | void; onAccepted?: () => Promise | void; onAcceptedPreStreamFailure?: (error: SendMessageError) => Promise | void; + onDeliveryCanceled?: (reason: string) => Promise | void; + onDeliveryAcceptedPreStreamFailure?: (error: SendMessageError) => Promise | void; cancelState?: { canceledBeforeAcceptance: boolean }; cancelSignal?: AbortSignal; } +function combineCallbacks( + correlationCallback: ((value: T) => Promise | void) | undefined, + deliveryCallback: ((value: T) => Promise | void) | undefined +): ((value: T) => Promise | void) | undefined { + if (correlationCallback == null) { + return deliveryCallback; + } + if (deliveryCallback == null) { + return correlationCallback; + } + return async (value: T) => { + try { + await correlationCallback?.(value); + } finally { + await deliveryCallback?.(value); + } + }; +} + +function getQueueClearCallbacks(entry: QueueEntry): QueueClearCallbacks | null { + const onCanceled = combineCallbacks(entry.onCanceled, entry.onDeliveryCanceled); + const onAcceptedPreStreamFailure = combineCallbacks( + entry.onAcceptedPreStreamFailure, + entry.onDeliveryAcceptedPreStreamFailure + ); + if (onCanceled == null && onAcceptedPreStreamFailure == null) { + return null; + } + return { + ...(onCanceled != null ? { onCanceled } : {}), + ...(onAcceptedPreStreamFailure != null ? { onAcceptedPreStreamFailure } : {}), + }; +} + /** * FIFO queue of messages sent during active streaming. * @@ -396,6 +438,8 @@ export class MessageQueue { internal?.onAccepted != null || internal?.onAcceptedPreStreamFailure != null || internal?.onCanceled != null || + internal?.onDeliveryAcceptedPreStreamFailure != null || + internal?.onDeliveryCanceled != null || internal?.cancelSignal != null; const incomingIsUserAuthored = internal?.synthetic !== true && internal?.agentInitiated !== true; @@ -479,6 +523,13 @@ export class MessageQueue { entry.onAcceptedPreStreamFailure = internal.onAcceptedPreStreamFailure; } + if (internal?.onDeliveryCanceled != null) { + entry.onDeliveryCanceled = internal.onDeliveryCanceled; + } + if (internal?.onDeliveryAcceptedPreStreamFailure != null) { + entry.onDeliveryAcceptedPreStreamFailure = internal.onDeliveryAcceptedPreStreamFailure; + } + if (internal?.cancelState != null) { entry.cancelState = internal.cancelState; } @@ -590,14 +641,10 @@ export class MessageQueue { * Callers must notify each one when clearing the queue. */ getClearCallbacks(): QueueClearCallbacks[] { - return this.entries - .filter((entry) => entry.onCanceled != null || entry.onAcceptedPreStreamFailure != null) - .map((entry) => ({ - ...(entry.onCanceled != null ? { onCanceled: entry.onCanceled } : {}), - ...(entry.onAcceptedPreStreamFailure != null - ? { onAcceptedPreStreamFailure: entry.onAcceptedPreStreamFailure } - : {}), - })); + return this.entries.flatMap((entry) => { + const callbacks = getQueueClearCallbacks(entry); + return callbacks == null ? [] : [callbacks]; + }); } /** @@ -617,12 +664,9 @@ export class MessageQueue { return null; } const [entry] = this.entries.splice(index, 1); - return { - ...(entry.onCanceled != null ? { onCanceled: entry.onCanceled } : {}), - ...(entry.onAcceptedPreStreamFailure != null - ? { onAcceptedPreStreamFailure: entry.onAcceptedPreStreamFailure } - : {}), - }; + // An empty object still means the entry was removed. Callers must not confuse + // callback absence with a missing queue entry. + return getQueueClearCallbacks(entry) ?? {}; } /** Remove queued entries carrying a dedupe key with the given prefix. */ @@ -661,13 +705,9 @@ export class MessageQueue { entry.agentInitiatedCount = Math.min(entry.agentInitiatedCount, entry.addCount); return [entry]; } - if (entry.onCanceled != null || entry.onAcceptedPreStreamFailure != null) { - removedCallbacks.push({ - ...(entry.onCanceled != null ? { onCanceled: entry.onCanceled } : {}), - ...(entry.onAcceptedPreStreamFailure != null - ? { onAcceptedPreStreamFailure: entry.onAcceptedPreStreamFailure } - : {}), - }); + const callbacks = getQueueClearCallbacks(entry); + if (callbacks != null) { + removedCallbacks.push(callbacks); } return []; }); @@ -727,24 +767,27 @@ export class MessageQueue { const allAddsAreSynthetic = entry.addCount > 0 && entry.syntheticCount === entry.addCount; const allAddsAreAgentInitiated = entry.addCount > 0 && entry.agentInitiatedCount === entry.addCount; + const onCanceled = combineCallbacks(entry.onCanceled, entry.onDeliveryCanceled); + const onAcceptedPreStreamFailure = combineCallbacks( + entry.onAcceptedPreStreamFailure, + entry.onDeliveryAcceptedPreStreamFailure + ); const hasInternalOptions = allAddsAreSynthetic || allAddsAreAgentInitiated || entry.onAccepted != null || - entry.onAcceptedPreStreamFailure != null || - entry.onCanceled != null || + onAcceptedPreStreamFailure != null || + onCanceled != null || entry.cancelSignal != null; const internal = hasInternalOptions ? { ...(allAddsAreSynthetic ? { synthetic: true } : {}), ...(allAddsAreAgentInitiated ? { agentInitiated: true } : {}), - ...(entry.onCanceled != null ? { onCanceled: entry.onCanceled } : {}), + ...(onCanceled != null ? { onCanceled } : {}), ...(entry.cancelState != null ? { cancelState: entry.cancelState } : {}), ...(entry.cancelSignal != null ? { cancelSignal: entry.cancelSignal } : {}), ...(entry.onAccepted != null ? { onAccepted: entry.onAccepted } : {}), - ...(entry.onAcceptedPreStreamFailure != null - ? { onAcceptedPreStreamFailure: entry.onAcceptedPreStreamFailure } - : {}), + ...(onAcceptedPreStreamFailure != null ? { onAcceptedPreStreamFailure } : {}), } : undefined; diff --git a/src/node/services/subagentFailureArtifacts.ts b/src/node/services/subagentFailureArtifacts.ts index 93d6b838f5..6145c1e694 100644 --- a/src/node/services/subagentFailureArtifacts.ts +++ b/src/node/services/subagentFailureArtifacts.ts @@ -30,6 +30,8 @@ export interface SubagentFailureArtifact { parentWorkspaceId: string; createdAtMs: number; updatedAtMs: number; + /** Execution generation that produced this artifact (optional for legacy entries). */ + executionVersion?: string; /** StreamErrorType that terminally failed the task (e.g. "model_refusal"). */ errorType: string; /** Human-readable failure message; waiters reject with exactly this text. */ @@ -112,7 +114,11 @@ export async function readSubagentFailureArtifact( return null; } - return entry; + const executionVersion = + typeof entry.executionVersion === "string" && entry.executionVersion.trim().length > 0 + ? entry.executionVersion.trim() + : undefined; + return { ...entry, executionVersion }; } export async function upsertSubagentFailureArtifact(params: { @@ -124,6 +130,7 @@ export async function upsertSubagentFailureArtifact(params: { ancestorWorkspaceIds: string[]; errorType: string; errorMessage: string; + executionVersion?: string; workflowOwnedAncestorWorkspaceIds?: string[]; model?: string; nowMs?: number; @@ -131,6 +138,11 @@ export async function upsertSubagentFailureArtifact(params: { await workspaceFileLocks.withLock(params.workspaceId, async () => { const nowMs = params.nowMs ?? Date.now(); + const executionVersion = + typeof params.executionVersion === "string" && params.executionVersion.trim().length > 0 + ? params.executionVersion.trim() + : undefined; + const file = await readSubagentFailureArtifactsFile(params.workspaceSessionDir); const existing = file.failuresByChildTaskId[params.childTaskId] ?? null; @@ -139,6 +151,7 @@ export async function upsertSubagentFailureArtifact(params: { parentWorkspaceId: params.parentWorkspaceId, createdAtMs: existing?.createdAtMs ?? nowMs, updatedAtMs: nowMs, + executionVersion, errorType: params.errorType, errorMessage: params.errorMessage, ancestorWorkspaceIds: params.ancestorWorkspaceIds, diff --git a/src/node/services/subagentReportArtifacts.test.ts b/src/node/services/subagentReportArtifacts.test.ts index 7a1fce9a4a..48196a361c 100644 --- a/src/node/services/subagentReportArtifacts.test.ts +++ b/src/node/services/subagentReportArtifacts.test.ts @@ -66,6 +66,29 @@ describe("subagentReportArtifacts", () => { expect(artifact?.planFilePath).toBe(planFilePath); }); + test("upsertSubagentReportArtifact preserves the execution version", async () => { + const workspaceId = "parent-1"; + const childTaskId = "child-versioned"; + const executionVersion = "wst_versioned_report"; + + await upsertSubagentReportArtifact({ + workspaceId, + workspaceSessionDir: testDir, + childTaskId, + parentWorkspaceId: workspaceId, + ancestorWorkspaceIds: [workspaceId], + reportMarkdown: "versioned report", + executionVersion, + nowMs: Date.now(), + }); + + const artifacts = await readSubagentReportArtifactsFile(testDir); + expect(artifacts.artifactsByChildTaskId[childTaskId]?.executionVersion).toBe(executionVersion); + expect((await readSubagentReportArtifact(testDir, childTaskId))?.executionVersion).toBe( + executionVersion + ); + }); + test("upsertSubagentReportArtifact preserves structured output", async () => { const workspaceId = "parent-1"; const childTaskId = "child-structured"; diff --git a/src/node/services/subagentReportArtifacts.ts b/src/node/services/subagentReportArtifacts.ts index 5850bb66f5..1b70771638 100644 --- a/src/node/services/subagentReportArtifacts.ts +++ b/src/node/services/subagentReportArtifacts.ts @@ -19,6 +19,8 @@ export interface SubagentReportArtifactIndexEntry { parentWorkspaceId: string; createdAtMs: number; updatedAtMs: number; + /** Execution generation that produced this artifact (optional for legacy entries). */ + executionVersion?: string; /** Task-level model string used when running the sub-agent (optional for legacy entries). */ model?: string; /** Task-level thinking/reasoning level used when running the sub-agent (optional for legacy entries). */ @@ -135,6 +137,7 @@ export async function readSubagentReportArtifact( parentWorkspaceId?: unknown; createdAtMs?: unknown; updatedAtMs?: unknown; + executionVersion?: unknown; model?: unknown; thinkingLevel?: unknown; title?: unknown; @@ -157,6 +160,11 @@ export async function readSubagentReportArtifact( ? obj.planFilePath.trim() : undefined; + const executionVersion = + typeof obj.executionVersion === "string" && obj.executionVersion.trim().length > 0 + ? obj.executionVersion.trim() + : undefined; + const model = typeof obj.model === "string" && obj.model.trim().length > 0 ? obj.model.trim() : undefined; const thinkingLevel = coerceThinkingLevel(obj.thinkingLevel); @@ -169,6 +177,11 @@ export async function readSubagentReportArtifact( // Trust the index file for metadata (versioned), but allow per-task file to override title. return { ...meta, + executionVersion: + executionVersion ?? + (typeof meta.executionVersion === "string" && meta.executionVersion.trim().length > 0 + ? meta.executionVersion.trim() + : undefined), model: typeof meta.model === "string" && meta.model.trim().length > 0 ? meta.model.trim() @@ -199,6 +212,7 @@ export async function readSubagentReportArtifact( parentWorkspaceId, createdAtMs, updatedAtMs, + executionVersion, model, thinkingLevel, title, @@ -247,6 +261,8 @@ export async function upsertSubagentReportArtifact(params: { parentWorkspaceId: string; ancestorWorkspaceIds: string[]; reportMarkdown: string; + /** Execution generation that produced this artifact. */ + executionVersion?: string; /** Task-level model string used when running the sub-agent (optional for legacy entries). */ model?: string; /** Task-level thinking/reasoning level used when running the sub-agent (optional for legacy entries). */ @@ -262,6 +278,11 @@ export async function upsertSubagentReportArtifact(params: { await workspaceFileLocks.withLock(params.workspaceId, async () => { const nowMs = params.nowMs ?? Date.now(); + const executionVersion = + typeof params.executionVersion === "string" && params.executionVersion.trim().length > 0 + ? params.executionVersion.trim() + : undefined; + const model = typeof params.model === "string" && params.model.trim().length > 0 ? params.model.trim() @@ -292,6 +313,7 @@ export async function upsertSubagentReportArtifact(params: { parentWorkspaceId: params.parentWorkspaceId, createdAtMs, updatedAtMs: nowMs, + executionVersion, model, thinkingLevel, title: params.title, @@ -319,6 +341,7 @@ export async function upsertSubagentReportArtifact(params: { parentWorkspaceId: params.parentWorkspaceId, createdAtMs, updatedAtMs: nowMs, + executionVersion, model, thinkingLevel, title: params.title, diff --git a/src/node/services/taskService.test.ts b/src/node/services/taskService.test.ts index eff2ba2524..ca8f34cf2b 100644 --- a/src/node/services/taskService.test.ts +++ b/src/node/services/taskService.test.ts @@ -56,7 +56,10 @@ import * as forkOrchestrator from "@/node/services/utils/forkOrchestrator"; import { Ok, Err, type Result } from "@/common/types/result"; import { SCRATCH_PROJECT_CONFIG_KEY } from "@/common/constants/scratch"; import { STRUCTURED_WORKFLOW_REPORT_PLACEHOLDER_MARKDOWN } from "@/common/constants/workflowReports"; -import { formatSubagentReportEnvelope } from "@/common/utils/subagentReportEnvelope"; +import { + formatSubagentReportEnvelope, + parseSubagentReportEnvelope, +} from "@/common/utils/subagentReportEnvelope"; import { defaultModel } from "@/common/utils/ai/models"; import { enforceThinkingPolicy } from "@/common/utils/thinking/policy"; import type { AgentAiDefaults, AgentAiSubagentProfile } from "@/common/types/agentAiDefaults"; @@ -64,7 +67,7 @@ import { DEFAULT_TASK_SETTINGS } from "@/common/types/tasks"; import type { ThinkingLevel } from "@/common/types/thinking"; import type { SendMessageError } from "@/common/types/errors"; import type { ErrorEvent, StreamAbortEvent, StreamEndEvent } from "@/common/types/stream"; -import { createMuxMessage, type MuxMessage } from "@/common/types/message"; +import { createMuxMessage, type MuxMessage, type MuxMessageMetadata } from "@/common/types/message"; import { isDynamicToolPart, type DynamicToolPart } from "@/common/types/toolParts"; import { buildWorkflowRunCardMessage, @@ -78,6 +81,8 @@ import type { InitStateManager } from "@/node/services/initStateManager"; import { InitStateManager as RealInitStateManager } from "@/node/services/initStateManager"; import assert from "node:assert"; +type TestWorkspaceTurnMuxMetadata = Extract; + function initGitRepo(projectPath: string): void { execSync("git init -b main", { cwd: projectPath, stdio: "ignore" }); execSync('git config user.email "test@example.com"', { cwd: projectPath, stdio: "ignore" }); @@ -766,6 +771,7 @@ describe("TaskService", () => { stableIds?: string[]; disposable?: boolean; sendMessage?: ReturnType; + resumeStream?: ReturnType; remove?: ReturnType; isStreaming?: ReturnType; hasQueuedMessages?: ReturnType; @@ -773,6 +779,7 @@ describe("TaskService", () => { hasPendingBashMonitorWakeContinuation?: ReturnType; hasPendingWorkspaceTurnContinuation?: ReturnType; hasPendingAutoRetry?: ReturnType; + waitForIdleAndNoQueuedMessages?: ReturnType; waitForPendingStreamErrorRecoveryDecision?: ReturnType; } = {} ) { @@ -802,6 +809,7 @@ describe("TaskService", () => { ); const workspaceMocks = createWorkspaceServiceMocks({ create: createWorkspace, + ...(options.resumeStream != null ? { resumeStream: options.resumeStream } : {}), ...(options.sendMessage != null ? { sendMessage: options.sendMessage } : {}), ...(options.remove != null ? { remove: options.remove } : {}), ...(options.hasQueuedMessages != null @@ -821,6 +829,9 @@ describe("TaskService", () => { ...(options.hasPendingAutoRetry != null ? { hasPendingAutoRetry: options.hasPendingAutoRetry } : {}), + ...(options.waitForIdleAndNoQueuedMessages != null + ? { waitForIdleAndNoQueuedMessages: options.waitForIdleAndNoQueuedMessages } + : {}), ...(options.waitForPendingStreamErrorRecoveryDecision != null ? { waitForPendingStreamErrorRecoveryDecision: @@ -2705,6 +2716,126 @@ describe("TaskService", () => { expect(snapshot).toMatchObject({ status: "running", workspaceId: created.workspaceId }); }); + test("stale stream-end ordering is rechecked when a failed reservation exposes a later turn", async () => { + const { parentId, taskService, historyService, created } = await startWorkspaceTurnForTest(); + const oldAssistant = createMuxMessage( + "old-assistant-before-next-turn", + "assistant", + "Old turn", + { + model: "anthropic:claude-opus-4-6", + finishReason: "stop", + } + ); + expect((await historyService.appendToHistory(created.workspaceId, oldAssistant)).success).toBe( + true + ); + + const internal = taskService as unknown as { + activeWorkspaceTurnHandleByWorkspaceId: Map< + string, + { handleId: string; ownerWorkspaceId: string } + >; + getReservedWorkspaceTurnContinuation: ( + workspaceId: string, + correlation: TestWorkspaceTurnMuxMetadata + ) => { released: Promise } | undefined; + interruptWorkspaceTurnFromUncorrelatedStreamEnd: (event: StreamEndEvent) => Promise; + reserveActiveWorkspaceTurnContinuation: (workspaceId: string) => Promise< + | { + muxMetadata: TestWorkspaceTurnMuxMetadata; + released: Promise; + [Symbol.dispose]: () => void; + } + | undefined + >; + settleWorkspaceTurnContinuationFailure: ( + workspaceId: string, + muxMetadata: TestWorkspaceTurnMuxMetadata, + status: "interrupted" | "error", + error: string + ) => Promise; + taskHandleStore: TaskHandleStore; + }; + const reservation = await internal.reserveActiveWorkspaceTurnContinuation(created.workspaceId); + assert(reservation, "workspace-turn continuation reservation must exist"); + + let markReservationObserved!: () => void; + const reservationObserved = new Promise((resolve) => { + markReservationObserved = resolve; + }); + const getReserved = internal.getReservedWorkspaceTurnContinuation.bind(taskService); + const getReservedSpy = spyOn( + internal, + "getReservedWorkspaceTurnContinuation" + ).mockImplementation((workspaceId, correlation) => { + const current = getReserved(workspaceId, correlation); + if (current === reservation) { + markReservationObserved(); + } + return current; + }); + + try { + const handled = internal.interruptWorkspaceTurnFromUncorrelatedStreamEnd({ + type: "stream-end", + workspaceId: created.workspaceId, + messageId: oldAssistant.id, + metadata: { + model: "anthropic:claude-opus-4-6", + finishReason: "stop", + }, + parts: [], + }); + await reservationObserved; + + await internal.settleWorkspaceTurnContinuationFailure( + created.workspaceId, + reservation.muxMetadata, + "error", + "The reserved continuation failed." + ); + const nextHandleId = "wst_next_turn_after_failed_reservation"; + const nextTurnId = "next-turn-after-failed-reservation"; + await internal.taskHandleStore.upsertWorkspaceTurn({ + kind: "workspace_turn", + handleId: nextHandleId, + ownerWorkspaceId: parentId, + workspaceId: created.workspaceId, + turnId: nextTurnId, + status: "running", + createdAt: "2026-08-21T00:00:01.000Z", + updatedAt: "2026-08-21T00:00:01.000Z", + createdWorkspace: false, + disposableWorkspace: false, + }); + const nextPrompt = createMuxMessage("next-turn-prompt", "user", "Run the next turn", { + muxMetadata: { + type: "workspace-turn-task", + taskHandleId: nextHandleId, + ownerWorkspaceId: parentId, + turnId: nextTurnId, + }, + }); + expect((await historyService.appendToHistory(created.workspaceId, nextPrompt)).success).toBe( + true + ); + internal.activeWorkspaceTurnHandleByWorkspaceId.set(created.workspaceId, { + handleId: nextHandleId, + ownerWorkspaceId: parentId, + }); + reservation[Symbol.dispose](); + + expect(await handled).toBe(true); + expect(await taskService.getWorkspaceTurnSnapshot(parentId, nextHandleId)).toMatchObject({ + status: "running", + }); + } finally { + getReservedSpy.mockRestore(); + reservation[Symbol.dispose](); + } + }); + test("getWorkspaceTurnSnapshot recovers stale completed handles from matching history", async () => { const { parentId, taskService, historyService, created } = await startWorkspaceTurnForTest(); const appendResult = await historyService.appendToHistory( @@ -3114,6 +3245,7 @@ describe("TaskService", () => { // Notification remains pending; once the owner is idle, draining delivers it. hasPendingQueuedOrPreparingTurn.mockImplementation(() => false); await internal.drainTerminalAttention(parentId); + await Promise.all([...internal.pendingTerminalAttentionDrains]); const drained = sendMessage.mock.calls.find( (call) => typeof call[1] === "string" && call[1].includes("wst_handle") ); @@ -3176,6 +3308,67 @@ describe("TaskService", () => { expect(await terminalAttentionStore.listPending(parentId)).toHaveLength(0); }); + test("a response consumes only the included terminal report generation", async () => { + const config = await createTestConfig(rootDir); + const { parentId } = await saveLocalParentWorkspace(config, rootDir); + const taskId = "task-versioned-response-consumption"; + const previousGeneration = "wst_previous_response_generation"; + const pendingGeneration = "wst_pending_response_generation"; + const terminalAttentionStore = new TerminalAttentionStore(config); + const previousNotification = await terminalAttentionStore.enqueueIfAbsent({ + ownerWorkspaceId: parentId, + sourceKind: "agent_task", + sourceId: taskId, + generationId: previousGeneration, + }); + const pendingNotification = await terminalAttentionStore.enqueueIfAbsent({ + ownerWorkspaceId: parentId, + sourceKind: "agent_task", + sourceId: taskId, + generationId: pendingGeneration, + }); + assert(previousNotification, "previous terminal notification must exist"); + assert(pendingNotification, "pending terminal notification must exist"); + + const { historyService, taskService } = createTaskServiceHarness(config); + const reportMessage = createMuxMessage( + "previous-versioned-terminal-report", + "user", + formatSubagentReportEnvelope({ + taskId, + agentType: "explore", + status: "completed", + executionVersion: previousGeneration, + title: "Previous result", + reportMarkdown: "The previous generation completed.", + }), + { timestamp: Date.now(), synthetic: true, uiVisible: true } + ); + await historyService.appendToHistory(parentId, reportMessage); + const reportSequence = reportMessage.metadata?.historySequence; + assert(typeof reportSequence === "number", "report history sequence is required"); + await historyService.appendToHistory( + parentId, + createMuxMessage("versioned-response", "assistant", "Used the previous result.", { + timestamp: Date.now(), + requestHistorySequence: reportSequence, + }) + ); + + await ( + taskService as unknown as { + consumeRespondedAgentTerminalAttention: (ownerWorkspaceId: string) => Promise; + } + ).consumeRespondedAgentTerminalAttention(parentId); + + expect(await terminalAttentionStore.get(parentId, previousNotification.id)).toMatchObject({ + status: "delivered", + }); + expect(await terminalAttentionStore.get(parentId, pendingNotification.id)).toMatchObject({ + status: "pending", + }); + }); + test("late report still resumes an intentionally backgrounded parent once", async () => { const config = await createTestConfig(rootDir); const { parentId } = await saveLocalParentWorkspace(config, rootDir); @@ -3406,80 +3599,418 @@ describe("TaskService", () => { expect(await terminalAttentionStore.listPending(parentId)).toHaveLength(1); }); - test("persistent child reports supersede their private continuation wake prompt", async () => { + test("queued terminal attention fallback reclaims all notifications before retry writes", async () => { const config = await createTestConfig(rootDir); - const projectPath = path.join(rootDir, "repo"); - const parentWorkspaceId = "parent-continuation-report"; - const childTaskId = "child-continuation-report"; - const handleId = "wst_continuation_report"; - await saveWorkspaces( - config, - projectPath, - [ - projectWorkspace(projectPath, "parent", parentWorkspaceId), - projectWorkspace(projectPath, "child", childTaskId, { - parentWorkspaceId, - agentId: "explore", - agentType: "explore", - taskStatus: "reported", - reportedAt: "2026-08-10T00:00:02.000Z", - taskExecutionId: handleId, - taskExecutionStatus: "completed", - }), - ], - testTaskSettings() + const { parentId } = await saveLocalParentWorkspace(config, rootDir); + const handleIds = ["wst_fallback_retry_one", "wst_fallback_retry_two"] as const; + + let releaseFirstPendingWrite!: () => void; + const firstPendingWriteGate = new Promise((resolve) => { + releaseFirstPendingWrite = resolve; + }); + let markFirstPendingWritten!: () => void; + const firstPendingWritten = new Promise((resolve) => { + markFirstPendingWritten = resolve; + }); + let releaseIdleWait!: () => void; + const idleWait = new Promise((resolve) => { + releaseIdleWait = resolve; + }); + const waitForIdleAndNoQueuedMessages = mock(() => idleWait); + + let sendCount = 0; + const sendMessage = mock( + async ( + _workspaceId: string, + _message: string, + _options: unknown, + internal?: { + requireIdle?: boolean; + onAccepted?: () => Promise | void; + onAcceptedPreStreamFailure?: (error: SendMessageError) => Promise | void; + } + ): Promise> => { + sendCount += 1; + if (internal?.requireIdle === true) { + return Err({ type: "unknown", raw: "Workspace is busy; idle-only send was skipped." }); + } + await internal?.onAccepted?.(); + if (sendCount === 2) { + await internal?.onAcceptedPreStreamFailure?.({ + type: "unknown", + raw: "Queued terminal fallback startup failed", + }); + } + return Ok(undefined); + } ); + const workspaceMocks = createWorkspaceServiceMocks({ + sendMessage, + waitForIdleAndNoQueuedMessages, + }); + const { taskService } = createTaskServiceHarness(config, { + workspaceService: workspaceMocks.workspaceService, + }); + const taskHandleStore = (taskService as unknown as { taskHandleStore: TaskHandleStore }) + .taskHandleStore; + const terminalAttentionStore = ( + taskService as unknown as { terminalAttentionStore: TerminalAttentionStore } + ).terminalAttentionStore; - const resumeStream = mock( - (): Promise> => - Promise.resolve(Ok({ started: true })) + for (const [index, handleId] of handleIds.entries()) { + await taskHandleStore.upsertWorkspaceTurn({ + kind: "workspace_turn", + handleId, + ownerWorkspaceId: parentId, + workspaceId: `completed-workspace-${index}`, + turnId: `turn-${index}`, + status: "completed", + createdAt: `2026-08-21T00:00:0${index}.000Z`, + updatedAt: `2026-08-21T00:00:1${index}.000Z`, + createdWorkspace: false, + disposableWorkspace: false, + reportMarkdown: `Completed output ${index}`, + }); + await terminalAttentionStore.enqueueIfAbsent({ + ownerWorkspaceId: parentId, + sourceKind: "workspace_turn", + sourceId: handleId, + }); + } + + const originalMarkPending = terminalAttentionStore.markPending.bind(terminalAttentionStore); + let pendingWriteCount = 0; + spyOn(terminalAttentionStore, "markPending").mockImplementation( + async (ownerWorkspaceId: string, notificationId: string) => { + await originalMarkPending(ownerWorkspaceId, notificationId); + pendingWriteCount += 1; + if (pendingWriteCount === 1) { + markFirstPendingWritten(); + await firstPendingWriteGate; + } + } ); + + const internal = taskService as unknown as { + drainTerminalAttention: (ownerWorkspaceId: string) => Promise; + }; + const failedFallbackDrain = internal.drainTerminalAttention(parentId); + await firstPendingWritten; + + await internal.drainTerminalAttention(parentId); + expect(sendMessage).toHaveBeenCalledTimes(2); + + releaseFirstPendingWrite(); + await failedFallbackDrain; + for (const handleId of handleIds) { + expect( + await terminalAttentionStore.get(parentId, `workspace_turn:${handleId}`) + ).toMatchObject({ status: "pending" }); + } + + releaseIdleWait(); + await flushTerminalAttentionDrains(taskService); + }); + + test("queued terminal attention fallback releases claims after failure before acceptance", async () => { + const config = await createTestConfig(rootDir); + const { parentId } = await saveLocalParentWorkspace(config, rootDir); + const handleId = "wst_fallback_pre_accept_failure"; + + let releaseIdleWait!: () => void; + const idleWait = new Promise((resolve) => { + releaseIdleWait = resolve; + }); + const waitForIdleAndNoQueuedMessages = mock(() => idleWait); + let sendCount = 0; const sendMessage = mock( - (..._args: unknown[]): Promise> => Promise.resolve(Ok(undefined)) + async ( + _workspaceId: string, + _message: string, + _options: unknown, + internal?: { + requireIdle?: boolean; + onAccepted?: () => Promise | void; + onAcceptedPreStreamFailure?: (error: SendMessageError) => Promise | void; + } + ): Promise> => { + sendCount += 1; + if (sendCount === 1 && internal?.requireIdle === true) { + return Err({ type: "unknown", raw: "Workspace is busy; idle-only send was skipped." }); + } + if (sendCount === 2) { + await internal?.onAcceptedPreStreamFailure?.({ + type: "unknown", + raw: "Queued fallback failed before acceptance", + }); + return Ok(undefined); + } + await internal?.onAccepted?.(); + return Ok(undefined); + } ); - const { workspaceService } = createWorkspaceServiceMocks({ resumeStream, sendMessage }); - const { historyService, taskService } = createTaskServiceHarness(config, { workspaceService }); + const workspaceMocks = createWorkspaceServiceMocks({ + sendMessage, + waitForIdleAndNoQueuedMessages, + }); + const { taskService } = createTaskServiceHarness(config, { + workspaceService: workspaceMocks.workspaceService, + }); const taskHandleStore = (taskService as unknown as { taskHandleStore: TaskHandleStore }) .taskHandleStore; + const terminalAttentionStore = ( + taskService as unknown as { terminalAttentionStore: TerminalAttentionStore } + ).terminalAttentionStore; await taskHandleStore.upsertWorkspaceTurn({ kind: "workspace_turn", handleId, - ownerWorkspaceId: parentWorkspaceId, - workspaceId: childTaskId, - turnId: "turn-continuation-report", + ownerWorkspaceId: parentId, + workspaceId: "completed-workspace", + turnId: "turn", status: "completed", - createdAt: "2026-08-10T00:00:01.000Z", - updatedAt: "2026-08-10T00:00:02.000Z", + createdAt: "2026-08-21T00:00:00.000Z", + updatedAt: "2026-08-21T00:00:01.000Z", createdWorkspace: false, disposableWorkspace: false, - reportMarkdown: "Private continuation output", + reportMarkdown: "Completed output", }); - await historyService.appendToHistory( - parentWorkspaceId, - createMuxMessage( - "continuation-report", - "user", - formatSubagentReportEnvelope({ - taskId: childTaskId, - agentType: "explore", - status: "completed", - title: "Tooling Mapper", - reportMarkdown: "Stable child report", - }), - { - timestamp: Date.parse("2026-08-10T00:00:02.000Z"), - synthetic: true, - uiVisible: true, - } - ) - ); - - const terminalAttentionStore = new TerminalAttentionStore(config); await terminalAttentionStore.enqueueIfAbsent({ - ownerWorkspaceId: parentWorkspaceId, - sourceKind: "agent_task", - sourceId: childTaskId, - }); + ownerWorkspaceId: parentId, + sourceKind: "workspace_turn", + sourceId: handleId, + }); + + const internal = taskService as unknown as { + drainTerminalAttention: (ownerWorkspaceId: string) => Promise; + }; + await internal.drainTerminalAttention(parentId); + expect(sendMessage).toHaveBeenCalledTimes(2); + expect(await terminalAttentionStore.get(parentId, `workspace_turn:${handleId}`)).toMatchObject({ + status: "pending", + }); + + await internal.drainTerminalAttention(parentId); + expect(sendMessage).toHaveBeenCalledTimes(3); + expect(await terminalAttentionStore.get(parentId, `workspace_turn:${handleId}`)).toMatchObject({ + status: "delivered", + }); + + releaseIdleWait(); + await flushTerminalAttentionDrains(taskService); + }); + + test("queued terminal attention fallback re-arms an accepted send error", async () => { + const config = await createTestConfig(rootDir); + const { parentId } = await saveLocalParentWorkspace(config, rootDir); + const handleId = "wst_fallback_accepted_error"; + + let releaseIdleWait!: () => void; + const idleWait = new Promise((resolve) => { + releaseIdleWait = resolve; + }); + const waitForIdleAndNoQueuedMessages = mock(() => idleWait); + let sendCount = 0; + const sendMessage = mock( + async ( + _workspaceId: string, + _message: string, + _options: unknown, + internal?: { + requireIdle?: boolean; + onAccepted?: () => Promise | void; + } + ): Promise> => { + sendCount += 1; + if (sendCount === 1 && internal?.requireIdle === true) { + return Err({ type: "unknown", raw: "Workspace is busy; idle-only send was skipped." }); + } + await internal?.onAccepted?.(); + if (sendCount === 2) { + return Err({ type: "unknown", raw: "Accepted fallback startup failed" }); + } + return Ok(undefined); + } + ); + const workspaceMocks = createWorkspaceServiceMocks({ + sendMessage, + waitForIdleAndNoQueuedMessages, + }); + const { taskService } = createTaskServiceHarness(config, { + workspaceService: workspaceMocks.workspaceService, + }); + const taskHandleStore = (taskService as unknown as { taskHandleStore: TaskHandleStore }) + .taskHandleStore; + const terminalAttentionStore = ( + taskService as unknown as { terminalAttentionStore: TerminalAttentionStore } + ).terminalAttentionStore; + await taskHandleStore.upsertWorkspaceTurn({ + kind: "workspace_turn", + handleId, + ownerWorkspaceId: parentId, + workspaceId: "completed-workspace", + turnId: "turn", + status: "completed", + createdAt: "2026-08-21T00:00:00.000Z", + updatedAt: "2026-08-21T00:00:01.000Z", + createdWorkspace: false, + disposableWorkspace: false, + reportMarkdown: "Completed output", + }); + await terminalAttentionStore.enqueueIfAbsent({ + ownerWorkspaceId: parentId, + sourceKind: "workspace_turn", + sourceId: handleId, + }); + + const internal = taskService as unknown as { + drainTerminalAttention: (ownerWorkspaceId: string) => Promise; + }; + await internal.drainTerminalAttention(parentId); + expect(sendMessage).toHaveBeenCalledTimes(2); + expect(await terminalAttentionStore.get(parentId, `workspace_turn:${handleId}`)).toMatchObject({ + status: "pending", + }); + + await internal.drainTerminalAttention(parentId); + expect(sendMessage).toHaveBeenCalledTimes(3); + expect(await terminalAttentionStore.get(parentId, `workspace_turn:${handleId}`)).toMatchObject({ + status: "delivered", + }); + + releaseIdleWait(); + await flushTerminalAttentionDrains(taskService); + }); + + test("queued terminal attention fallback keeps the active workspace-turn correlation", async () => { + let sendCount = 0; + const sendMessage = mock((...args: unknown[]): Promise> => { + sendCount += 1; + const internal = args[3] as { requireIdle?: boolean } | undefined; + if (sendCount === 2 && internal?.requireIdle === true) { + return Promise.resolve( + Err({ type: "unknown", raw: "Workspace is busy; idle-only send was skipped." }) + ); + } + return Promise.resolve(Ok(undefined)); + }); + const { config, parentId, taskService } = await startWorkspaceTurnForTest({ sendMessage }); + const taskHandleStore = (taskService as unknown as { taskHandleStore: TaskHandleStore }) + .taskHandleStore; + const nestedHandleId = "wst_terminal_fallback_continuation"; + await taskHandleStore.upsertWorkspaceTurn({ + kind: "workspace_turn", + handleId: nestedHandleId, + ownerWorkspaceId: "childworkspace", + workspaceId: "completed-nested-workspace", + turnId: "nested-terminal-fallback", + status: "completed", + createdAt: "2026-08-21T00:00:00.000Z", + updatedAt: "2026-08-21T00:00:01.000Z", + createdWorkspace: false, + disposableWorkspace: false, + reportMarkdown: "The nested workspace turn completed.", + }); + const terminalAttentionStore = new TerminalAttentionStore(config); + await terminalAttentionStore.enqueueIfAbsent({ + ownerWorkspaceId: "childworkspace", + sourceKind: "workspace_turn", + sourceId: nestedHandleId, + }); + + await ( + taskService as unknown as { + drainTerminalAttention: (ownerWorkspaceId: string) => Promise; + } + ).drainTerminalAttention("childworkspace"); + + expect(sendMessage).toHaveBeenCalledTimes(3); + expect(sendMessage.mock.calls[2]?.[2]).toMatchObject({ + muxMetadata: { + type: "workspace-turn-task", + taskHandleId: "wst_handle", + ownerWorkspaceId: parentId, + turnId: "turn", + }, + }); + expect(sendMessage.mock.calls[2]?.[3]).toMatchObject({ + workspaceTurnContinuation: true, + }); + }); + + test("persistent child reports supersede their private continuation wake prompt", async () => { + const config = await createTestConfig(rootDir); + const projectPath = path.join(rootDir, "repo"); + const parentWorkspaceId = "parent-continuation-report"; + const childTaskId = "child-continuation-report"; + const handleId = "wst_continuation_report"; + await saveWorkspaces( + config, + projectPath, + [ + projectWorkspace(projectPath, "parent", parentWorkspaceId), + projectWorkspace(projectPath, "child", childTaskId, { + parentWorkspaceId, + agentId: "explore", + agentType: "explore", + taskStatus: "reported", + reportedAt: "2026-08-10T00:00:02.000Z", + taskExecutionId: handleId, + taskExecutionStatus: "completed", + }), + ], + testTaskSettings() + ); + + const resumeStream = mock( + (): Promise> => + Promise.resolve(Ok({ started: true })) + ); + const sendMessage = mock( + (..._args: unknown[]): Promise> => Promise.resolve(Ok(undefined)) + ); + const { workspaceService } = createWorkspaceServiceMocks({ resumeStream, sendMessage }); + const { historyService, taskService } = createTaskServiceHarness(config, { workspaceService }); + const taskHandleStore = (taskService as unknown as { taskHandleStore: TaskHandleStore }) + .taskHandleStore; + await taskHandleStore.upsertWorkspaceTurn({ + kind: "workspace_turn", + handleId, + ownerWorkspaceId: parentWorkspaceId, + workspaceId: childTaskId, + turnId: "turn-continuation-report", + status: "completed", + createdAt: "2026-08-10T00:00:01.000Z", + updatedAt: "2026-08-10T00:00:02.000Z", + createdWorkspace: false, + disposableWorkspace: false, + reportMarkdown: "Private continuation output", + }); + await historyService.appendToHistory( + parentWorkspaceId, + createMuxMessage( + "continuation-report", + "user", + formatSubagentReportEnvelope({ + taskId: childTaskId, + agentType: "explore", + status: "completed", + title: "Tooling Mapper", + reportMarkdown: "Stable child report", + }), + { + timestamp: Date.parse("2026-08-10T00:00:02.000Z"), + synthetic: true, + uiVisible: true, + } + ) + ); + + const terminalAttentionStore = new TerminalAttentionStore(config); + await terminalAttentionStore.enqueueIfAbsent({ + ownerWorkspaceId: parentWorkspaceId, + sourceKind: "agent_task", + sourceId: childTaskId, + }); await terminalAttentionStore.enqueueIfAbsent({ ownerWorkspaceId: parentWorkspaceId, sourceKind: "workspace_turn", @@ -4360,40 +4891,1018 @@ describe("TaskService", () => { | undefined; await internal?.onCanceled?.("Progress wake was canceled"); } - return Ok(undefined); + return Ok(undefined); + } + ); + const { config, parentId, taskService } = await startWorkspaceTurnForTest({ sendMessage }); + + await config.editConfig((cfg) => { + const project = cfg.projects.get(path.join(rootDir, "repo")); + assert(project, "test project must exist"); + project.workspaces.push({ + path: path.join(rootDir, "repo", "nested-progress-canceled"), + id: "nested-progress-canceled", + name: "nested-progress-canceled", + createdAt: "2026-06-19T00:00:00.000Z", + runtimeConfig: { type: "local" }, + parentWorkspaceId: "childworkspace", + taskStatus: "running", + agentType: "explore", + }); + return cfg; + }); + + await taskService.reportAgentProgress("nested-progress-canceled", "progress-call", { + reportMarkdown: "The progress wake was canceled.", + }); + + expect(await taskService.getWorkspaceTurnSnapshot(parentId, "wst_handle")).toMatchObject({ + status: "interrupted", + error: "Progress wake was canceled", + }); + }); + + test("terminal nested report replaces progress before the outer turn settles", async () => { + let progressCancellation: ((reason: string) => Promise | void) | undefined; + let terminalContinuationPending = false; + const handoffEvents: string[] = []; + const sendMessage = mock((...args: unknown[]): Promise> => { + const internal = args[3] as { + onCanceled?: (reason: string) => Promise | void; + queueDedupeKey?: string; + }; + if (internal?.queueDedupeKey?.startsWith("agent-report:") === true) { + progressCancellation = internal.onCanceled; + } + if (internal?.queueDedupeKey?.startsWith("agent-terminal-report:") === true) { + terminalContinuationPending = true; + handoffEvents.push("terminal-accepted"); + } + return Promise.resolve(Ok(undefined)); + }); + const hasPendingWorkspaceTurnContinuation = mock(() => terminalContinuationPending); + const { config, parentId, taskService, workspaceMocks } = await startWorkspaceTurnForTest({ + sendMessage, + hasPendingWorkspaceTurnContinuation, + }); + await config.editConfig((cfg) => { + const project = cfg.projects.get(path.join(rootDir, "repo")); + assert(project, "test project must exist"); + project.workspaces.push({ + path: path.join(rootDir, "repo", "nested-coalesced-report"), + id: "nested-coalesced-report", + name: "nested-coalesced-report", + createdAt: "2026-06-19T00:00:00.000Z", + runtimeConfig: { type: "local" }, + parentWorkspaceId: "childworkspace", + taskStatus: "running", + agentType: "explore", + }); + return cfg; + }); + + await taskService.reportAgentProgress("nested-coalesced-report", "progress-call", { + reportMarkdown: "Progress before completion.", + }); + expect(progressCancellation).toBeDefined(); + + let cancellationPromise: Promise | undefined; + workspaceMocks.workspaceService.removeQueuedMessagesByDedupeKeyPrefix = mock( + (_workspaceId: string, _prefix: string, options?: { notifyCancellation?: boolean }) => { + handoffEvents.push("progress-removed"); + if (options?.notifyCancellation !== false) { + const cancellationResult = progressCancellation?.("Queued progress was canceled"); + cancellationPromise = Promise.resolve(cancellationResult); + } + return Ok(1); + } + ) as WorkspaceService["removeQueuedMessagesByDedupeKeyPrefix"]; + + await handleTaskServiceStreamEndForTest(taskService, { + type: "stream-end", + workspaceId: "nested-coalesced-report", + messageId: "assistant-nested-coalesced-report", + metadata: { model: "anthropic:claude-opus-4-6", finishReason: "stop" }, + parts: [ + { + type: "dynamic-tool", + toolCallId: "nested-report-call", + toolName: "agent_report", + input: { reportMarkdown: "Nested work completed." }, + state: "output-available", + output: { + success: true, + report: { reportMarkdown: "Nested work completed." }, + }, + }, + { type: "text", text: "Nested work completed." }, + ], + }); + + expect(handoffEvents).toEqual(["terminal-accepted", "progress-removed"]); + expect(cancellationPromise).toBeUndefined(); + expect(sendMessage).toHaveBeenCalledWith( + "childworkspace", + expect.stringContaining("Nested work completed."), + expect.objectContaining({ + muxMetadata: { + type: "workspace-turn-task", + taskHandleId: "wst_handle", + ownerWorkspaceId: parentId, + turnId: "turn", + }, + }), + expect.objectContaining({ + queueDedupeKey: "agent-terminal-report:agent_task:nested-coalesced-report", + workspaceTurnContinuation: true, + }) + ); + + const internal = taskService as unknown as { + handleStreamEnd: (event: StreamEndEvent) => Promise; + }; + await internal.handleStreamEnd({ + type: "stream-end", + workspaceId: "childworkspace", + messageId: "outer-uncorrelated-end", + metadata: { + model: "anthropic:claude-opus-4-6", + agentId: "exec", + finishReason: "stop", + }, + parts: [{ type: "text", text: "Outer stream ended before terminal handoff." }], + }); + expect(await taskService.getWorkspaceTurnSnapshot(parentId, "wst_handle")).toMatchObject({ + status: "running", + }); + + terminalContinuationPending = false; + await internal.handleStreamEnd({ + type: "stream-end", + workspaceId: "childworkspace", + messageId: "outer-terminal-continuation", + metadata: { + model: "anthropic:claude-opus-4-6", + agentId: "exec", + finishReason: "stop", + muxMetadata: { + type: "workspace-turn-task", + taskHandleId: "wst_handle", + ownerWorkspaceId: parentId, + turnId: "turn", + }, + }, + parts: [{ type: "text", text: "Outer turn completed." }], + }); + + expect(await taskService.getWorkspaceTurnSnapshot(parentId, "wst_handle")).toMatchObject({ + status: "completed", + reportMarkdown: "Outer turn completed.", + }); + }); + + test("terminal continuation reservation blocks stream-end settlement until acceptance", async () => { + let terminalContinuationPending = false; + let releaseTerminalSend!: () => void; + const terminalSendGate = new Promise((resolve) => { + releaseTerminalSend = resolve; + }); + let markTerminalSendStarted!: () => void; + const terminalSendStarted = new Promise((resolve) => { + markTerminalSendStarted = resolve; + }); + const sendMessage = mock( + async (...args: unknown[]): Promise> => { + const internal = args[3] as { + onAccepted?: () => Promise | void; + queueDedupeKey?: string; + }; + if (internal?.queueDedupeKey?.startsWith("agent-terminal-report:") === true) { + markTerminalSendStarted(); + await terminalSendGate; + terminalContinuationPending = true; + await internal.onAccepted?.(); + } + return Ok(undefined); + } + ); + const hasPendingWorkspaceTurnContinuation = mock(() => terminalContinuationPending); + const { config, parentId, taskService, workspaceMocks } = await startWorkspaceTurnForTest({ + sendMessage, + hasPendingWorkspaceTurnContinuation, + }); + await config.editConfig((cfg) => { + const project = cfg.projects.get(path.join(rootDir, "repo")); + assert(project, "test project must exist"); + project.workspaces.push({ + path: path.join(rootDir, "repo", "nested-fast-terminal-report"), + id: "nested-fast-terminal-report", + name: "nested-fast-terminal-report", + createdAt: "2026-06-19T00:00:00.000Z", + runtimeConfig: { type: "local" }, + parentWorkspaceId: "childworkspace", + taskStatus: "running", + agentType: "explore", + }); + return cfg; + }); + + const completion = handleTaskServiceStreamEndForTest(taskService, { + type: "stream-end", + workspaceId: "nested-fast-terminal-report", + messageId: "assistant-nested-fast-terminal-report", + metadata: { model: "anthropic:claude-opus-4-6", finishReason: "stop" }, + parts: [ + { + type: "dynamic-tool", + toolCallId: "nested-report-call", + toolName: "agent_report", + input: { reportMarkdown: "Nested work completed quickly." }, + state: "output-available", + output: { + success: true, + report: { reportMarkdown: "Nested work completed quickly." }, + }, + }, + { type: "text", text: "Nested work completed quickly." }, + ], + }); + + await terminalSendStarted; + let waiterSettled = false; + const waiter = taskService + .waitForWorkspaceTurn("wst_handle", { + ownerWorkspaceId: parentId, + requestingWorkspaceId: parentId, + backgroundOnMessageQueued: false, + timeoutMs: 10_000, + }) + .then((result) => { + waiterSettled = true; + return result; + }); + const internal = taskService as unknown as { + drainTerminalAttention: (ownerWorkspaceId: string) => Promise; + interruptWorkspaceTurnFromUncorrelatedStreamEnd: (event: StreamEndEvent) => Promise; + handleStreamEnd: (event: StreamEndEvent) => Promise; + }; + const uncorrelatedEnd = internal.interruptWorkspaceTurnFromUncorrelatedStreamEnd({ + type: "stream-end", + workspaceId: "childworkspace", + messageId: "outer-end-before-terminal-acceptance", + metadata: { + model: "anthropic:claude-opus-4-6", + agentId: "exec", + finishReason: "stop", + }, + parts: [{ type: "text", text: "Outer stream ended during terminal handoff." }], + }); + expect(waiterSettled).toBe(false); + expect(await taskService.getWorkspaceTurnSnapshot(parentId, "wst_handle")).toMatchObject({ + status: "running", + }); + + const concurrentDrain = internal.drainTerminalAttention("childworkspace"); + await concurrentDrain; + expect(workspaceMocks.resumeStream).not.toHaveBeenCalled(); + + releaseTerminalSend(); + await Promise.all([completion, uncorrelatedEnd]); + expect(await taskService.getWorkspaceTurnSnapshot(parentId, "wst_handle")).toMatchObject({ + status: "running", + }); + + terminalContinuationPending = false; + await internal.handleStreamEnd({ + type: "stream-end", + workspaceId: "childworkspace", + messageId: "outer-terminal-continuation", + metadata: { + model: "anthropic:claude-opus-4-6", + agentId: "exec", + finishReason: "stop", + muxMetadata: { + type: "workspace-turn-task", + taskHandleId: "wst_handle", + ownerWorkspaceId: parentId, + turnId: "turn", + }, + }, + parts: [{ type: "text", text: "Outer turn completed." }], + }); + expect(await waiter).toMatchObject({ + workspaceId: "childworkspace", + reportMarkdown: "Outer turn completed.", + }); + expect(await taskService.getWorkspaceTurnSnapshot(parentId, "wst_handle")).toMatchObject({ + status: "completed", + reportMarkdown: "Outer turn completed.", + }); + + const attentionStore = new TerminalAttentionStore(config); + const attentionId = TerminalAttentionStore.notificationId( + "agent_task", + "nested-fast-terminal-report" + ); + expect(await attentionStore.get("childworkspace", attentionId)).toMatchObject({ + status: "delivered", + }); + expect(await attentionStore.listPending("childworkspace")).toEqual([]); + expect(workspaceMocks.resumeStream).not.toHaveBeenCalled(); + }); + + test("a live attention drain does not block a new terminal continuation", async () => { + let releaseResume!: () => void; + const resumeGate = new Promise((resolve) => { + releaseResume = resolve; + }); + let markResumeStarted!: () => void; + const resumeStarted = new Promise((resolve) => { + markResumeStarted = resolve; + }); + const resumeStream = mock(async (): Promise> => { + markResumeStarted(); + await resumeGate; + return Ok({ started: true }); + }); + const sendMessage = mock( + async (...args: unknown[]): Promise> => { + const internal = args[3] as { + onAccepted?: () => Promise | void; + queueDedupeKey?: string; + }; + if (internal?.queueDedupeKey?.startsWith("agent-terminal-report:") === true) { + await internal.onAccepted?.(); + } + return Ok(undefined); + } + ); + const { config, taskService } = await startWorkspaceTurnForTest({ + resumeStream, + sendMessage, + }); + + await upsertSubagentReportArtifact({ + workspaceId: "childworkspace", + workspaceSessionDir: config.getSessionDir("childworkspace"), + childTaskId: "drain-owned-report", + parentWorkspaceId: "childworkspace", + ancestorWorkspaceIds: ["childworkspace"], + workflowOwnedAncestorWorkspaceIds: [], + reportMarkdown: "The earlier report needs attention.", + nowMs: Date.now(), + }); + const attentionStore = new TerminalAttentionStore(config); + await attentionStore.enqueueIfAbsent({ + ownerWorkspaceId: "childworkspace", + sourceKind: "agent_task", + sourceId: "drain-owned-report", + }); + + const internal = taskService as unknown as { + drainTerminalAttention: (ownerWorkspaceId: string) => Promise; + }; + const drain = internal.drainTerminalAttention("childworkspace"); + await resumeStarted; + + await config.editConfig((cfg) => { + const project = cfg.projects.get(path.join(rootDir, "repo")); + assert(project, "test project must exist"); + project.workspaces.push({ + path: path.join(rootDir, "repo", "nested-during-attention-drain"), + id: "nested-during-attention-drain", + name: "nested-during-attention-drain", + createdAt: "2026-06-19T00:00:00.000Z", + runtimeConfig: { type: "local" }, + parentWorkspaceId: "childworkspace", + taskStatus: "running", + agentType: "explore", + }); + return cfg; + }); + + await handleTaskServiceStreamEndForTest(taskService, { + type: "stream-end", + workspaceId: "nested-during-attention-drain", + messageId: "assistant-nested-during-attention-drain", + metadata: { model: "anthropic:claude-opus-4-6", finishReason: "stop" }, + parts: [ + { + type: "dynamic-tool", + toolCallId: "nested-report-call", + toolName: "agent_report", + input: { reportMarkdown: "The new nested work completed." }, + state: "output-available", + output: { + success: true, + report: { reportMarkdown: "The new nested work completed." }, + }, + }, + { type: "text", text: "The new nested work completed." }, + ], + }); + + expect( + await attentionStore.get( + "childworkspace", + TerminalAttentionStore.notificationId("agent_task", "nested-during-attention-drain") + ) + ).toMatchObject({ status: "delivered" }); + + releaseResume(); + await drain; + }); + + test("accepted terminal continuation startup failure re-arms terminal attention", async () => { + let releaseIdleWait!: () => void; + const idleWait = new Promise((resolve) => { + releaseIdleWait = resolve; + }); + const idleCheckContext: { taskService?: TaskService; parentId?: string } = {}; + const waitForIdleAndNoQueuedMessages = mock(async () => { + assert(idleCheckContext.taskService, "task service must be available before the idle retry"); + assert(idleCheckContext.parentId, "parent workspace must be available before the idle retry"); + expect( + await idleCheckContext.taskService.getWorkspaceTurnSnapshot( + idleCheckContext.parentId, + "wst_handle" + ) + ).toMatchObject({ status: "error" }); + await idleWait; + }); + const sendMessage = mock( + async (...args: unknown[]): Promise> => { + const internal = args[3] as { + onAccepted?: () => Promise | void; + onAcceptedPreStreamFailure?: (error: SendMessageError) => Promise | void; + onDeliveryAcceptedPreStreamFailure?: (error: SendMessageError) => Promise | void; + queueDedupeKey?: string; + }; + if (internal?.queueDedupeKey?.startsWith("agent-terminal-report:") === true) { + const error: SendMessageError = { + type: "unknown", + raw: "Terminal continuation startup failed", + }; + await internal.onAccepted?.(); + await internal.onAcceptedPreStreamFailure?.(error); + await internal.onDeliveryAcceptedPreStreamFailure?.(error); + } + return Ok(undefined); + } + ); + const { config, parentId, taskService, workspaceMocks } = await startWorkspaceTurnForTest({ + sendMessage, + waitForIdleAndNoQueuedMessages, + }); + idleCheckContext.taskService = taskService; + idleCheckContext.parentId = parentId; + await config.editConfig((cfg) => { + const project = cfg.projects.get(path.join(rootDir, "repo")); + assert(project, "test project must exist"); + project.workspaces.push({ + path: path.join(rootDir, "repo", "nested-terminal-startup-failure"), + id: "nested-terminal-startup-failure", + name: "nested-terminal-startup-failure", + createdAt: "2026-06-19T00:00:00.000Z", + runtimeConfig: { type: "local" }, + parentWorkspaceId: "childworkspace", + taskStatus: "running", + agentType: "explore", + }); + return cfg; + }); + + await handleTaskServiceStreamEndForTest(taskService, { + type: "stream-end", + workspaceId: "nested-terminal-startup-failure", + messageId: "assistant-nested-terminal-startup-failure", + metadata: { model: "anthropic:claude-opus-4-6", finishReason: "stop" }, + parts: [ + { + type: "dynamic-tool", + toolCallId: "nested-report-call", + toolName: "agent_report", + input: { reportMarkdown: "Nested work completed before startup failed." }, + state: "output-available", + output: { + success: true, + report: { reportMarkdown: "Nested work completed before startup failed." }, + }, + }, + { type: "text", text: "Nested work completed before startup failed." }, + ], + }); + + const attentionStore = new TerminalAttentionStore(config); + const attentionId = TerminalAttentionStore.notificationId( + "agent_task", + "nested-terminal-startup-failure" + ); + expect(await attentionStore.get("childworkspace", attentionId)).toMatchObject({ + status: "pending", + }); + expect(waitForIdleAndNoQueuedMessages).toHaveBeenCalledWith("childworkspace"); + expect(workspaceMocks.resumeStream).not.toHaveBeenCalled(); + expect(await taskService.getWorkspaceTurnSnapshot(parentId, "wst_handle")).toMatchObject({ + status: "error", + error: "Terminal continuation startup failed", + }); + + releaseIdleWait(); + await Promise.resolve(); + const internal = taskService as unknown as { + pendingTerminalAttentionDrains: Set>; + }; + await Promise.all([...internal.pendingTerminalAttentionDrains]); + }); + + test("stripped terminal continuation startup failure re-arms attention without settling the turn", async () => { + let releaseIdleWait!: () => void; + const idleWait = new Promise((resolve) => { + releaseIdleWait = resolve; + }); + const waitForIdleAndNoQueuedMessages = mock(() => idleWait); + const sendMessage = mock( + async (...args: unknown[]): Promise> => { + const internal = args[3] as { + onAccepted?: () => Promise | void; + onDeliveryAcceptedPreStreamFailure?: (error: SendMessageError) => Promise | void; + queueDedupeKey?: string; + }; + if (internal?.queueDedupeKey?.startsWith("agent-terminal-report:") === true) { + const error: SendMessageError = { + type: "unknown", + raw: "Stripped terminal continuation startup failed", + }; + await internal.onAccepted?.(); + await internal.onDeliveryAcceptedPreStreamFailure?.(error); + } + return Ok(undefined); + } + ); + const { config, parentId, taskService } = await startWorkspaceTurnForTest({ + sendMessage, + waitForIdleAndNoQueuedMessages, + }); + await config.editConfig((cfg) => { + const project = cfg.projects.get(path.join(rootDir, "repo")); + assert(project, "test project must exist"); + project.workspaces.push({ + path: path.join(rootDir, "repo", "nested-stripped-terminal-failure"), + id: "nested-stripped-terminal-failure", + name: "nested-stripped-terminal-failure", + createdAt: "2026-06-19T00:00:00.000Z", + runtimeConfig: { type: "local" }, + parentWorkspaceId: "childworkspace", + taskStatus: "running", + agentType: "explore", + }); + return cfg; + }); + + await handleTaskServiceStreamEndForTest(taskService, { + type: "stream-end", + workspaceId: "nested-stripped-terminal-failure", + messageId: "assistant-nested-stripped-terminal-failure", + metadata: { model: "anthropic:claude-opus-4-6", finishReason: "stop" }, + parts: [ + { + type: "dynamic-tool", + toolCallId: "nested-report-call", + toolName: "agent_report", + input: { reportMarkdown: "Nested work completed before stripped startup failed." }, + state: "output-available", + output: { + success: true, + report: { + reportMarkdown: "Nested work completed before stripped startup failed.", + }, + }, + }, + { type: "text", text: "Nested work completed before stripped startup failed." }, + ], + }); + + const attentionStore = new TerminalAttentionStore(config); + const attentionId = TerminalAttentionStore.notificationId( + "agent_task", + "nested-stripped-terminal-failure" + ); + expect(await attentionStore.get("childworkspace", attentionId)).toMatchObject({ + status: "pending", + }); + expect(await taskService.getWorkspaceTurnSnapshot(parentId, "wst_handle")).toMatchObject({ + status: "running", + }); + + releaseIdleWait(); + await flushTerminalAttentionDrains(taskService); + }); + + test("stripped terminal continuation cancellation releases its attention claim", async () => { + let terminalCallbacks: + | { + onDeliveryCanceled?: (reason: string) => Promise | void; + queueDedupeKey?: string; + } + | undefined; + const sendMessage = mock((...args: unknown[]): Promise> => { + const internal = args[3] as typeof terminalCallbacks; + if (internal?.queueDedupeKey?.startsWith("agent-terminal-report:") === true) { + terminalCallbacks = internal; + } + return Promise.resolve(Ok(undefined)); + }); + const { config, parentId, taskService } = await startWorkspaceTurnForTest({ sendMessage }); + await config.editConfig((cfg) => { + const project = cfg.projects.get(path.join(rootDir, "repo")); + assert(project, "test project must exist"); + project.workspaces.push({ + path: path.join(rootDir, "repo", "nested-stripped-terminal-cancel"), + id: "nested-stripped-terminal-cancel", + name: "nested-stripped-terminal-cancel", + createdAt: "2026-06-19T00:00:00.000Z", + runtimeConfig: { type: "local" }, + parentWorkspaceId: "childworkspace", + taskStatus: "running", + agentType: "explore", + }); + return cfg; + }); + + await handleTaskServiceStreamEndForTest(taskService, { + type: "stream-end", + workspaceId: "nested-stripped-terminal-cancel", + messageId: "assistant-nested-stripped-terminal-cancel", + metadata: { model: "anthropic:claude-opus-4-6", finishReason: "stop" }, + parts: [ + { + type: "dynamic-tool", + toolCallId: "nested-report-call", + toolName: "agent_report", + input: { reportMarkdown: "Nested work completed before queue cancellation." }, + state: "output-available", + output: { + success: true, + report: { reportMarkdown: "Nested work completed before queue cancellation." }, + }, + }, + { type: "text", text: "Nested work completed before queue cancellation." }, + ], + }); + + await terminalCallbacks?.onDeliveryCanceled?.("Superseded by unrelated input."); + + const attentionStore = new TerminalAttentionStore(config); + const attentionId = TerminalAttentionStore.notificationId( + "agent_task", + "nested-stripped-terminal-cancel" + ); + expect(await attentionStore.get("childworkspace", attentionId)).toMatchObject({ + status: "superseded", + }); + expect(await taskService.getWorkspaceTurnSnapshot(parentId, "wst_handle")).toMatchObject({ + status: "running", + }); + const internal = taskService as unknown as { + claimedTerminalAttentionIdsByOwner: Map>; + }; + expect( + internal.claimedTerminalAttentionIdsByOwner.get("childworkspace")?.has(attentionId) ?? false + ).toBe(false); + }); + + test("failed terminal continuation enqueue settles the deferred stream-end", async () => { + let releaseTerminalSend!: () => void; + const terminalSendGate = new Promise((resolve) => { + releaseTerminalSend = resolve; + }); + let markTerminalSendStarted!: () => void; + const terminalSendStarted = new Promise((resolve) => { + markTerminalSendStarted = resolve; + }); + let terminalContinuationAttempted = false; + const sendMessage = mock( + async (...args: unknown[]): Promise> => { + const internal = args[3] as { queueDedupeKey?: string }; + if (internal?.queueDedupeKey?.startsWith("agent-terminal-report:") === true) { + terminalContinuationAttempted = true; + markTerminalSendStarted(); + await terminalSendGate; + return Err({ type: "unknown", raw: "Terminal enqueue failed" }); + } + return Ok(undefined); + } + ); + const { config, parentId, taskService, workspaceMocks, historyService } = + await startWorkspaceTurnForTest({ sendMessage }); + await config.editConfig((cfg) => { + const project = cfg.projects.get(path.join(rootDir, "repo")); + assert(project, "test project must exist"); + project.workspaces.push({ + path: path.join(rootDir, "repo", "nested-terminal-enqueue-failure"), + id: "nested-terminal-enqueue-failure", + name: "nested-terminal-enqueue-failure", + createdAt: "2026-06-19T00:00:00.000Z", + runtimeConfig: { type: "local" }, + parentWorkspaceId: "childworkspace", + taskStatus: "running", + agentType: "explore", + }); + return cfg; + }); + + await taskService.reportAgentProgress("nested-terminal-enqueue-failure", "progress-call", { + reportMarkdown: "Progress remains queued.", + }); + + const completion = handleTaskServiceStreamEndForTest(taskService, { + type: "stream-end", + workspaceId: "nested-terminal-enqueue-failure", + messageId: "assistant-nested-terminal-enqueue-failure", + metadata: { model: "anthropic:claude-opus-4-6", finishReason: "stop" }, + parts: [ + { + type: "dynamic-tool", + toolCallId: "nested-report-call", + toolName: "agent_report", + input: { reportMarkdown: "Nested work completed." }, + state: "output-available", + output: { + success: true, + report: { reportMarkdown: "Nested work completed." }, + }, + }, + { type: "text", text: "Nested work completed." }, + ], + }); + + await terminalSendStarted; + const internal = taskService as unknown as { + interruptWorkspaceTurnFromUncorrelatedStreamEnd: (event: StreamEndEvent) => Promise; + workspaceTurnContinuationReservationsByWorkspaceId: Map< + string, + Set<{ released: Promise }> + >; + }; + const reservation = Array.from( + internal.workspaceTurnContinuationReservationsByWorkspaceId.get("childworkspace") ?? [] + )[0]; + assert(reservation, "terminal continuation reservation must be published"); + const uncorrelatedEnd = internal.interruptWorkspaceTurnFromUncorrelatedStreamEnd({ + type: "stream-end", + workspaceId: "childworkspace", + messageId: "outer-end-before-failed-terminal-send", + metadata: { + model: "anthropic:claude-opus-4-6", + agentId: "exec", + finishReason: "stop", + }, + parts: [{ type: "text", text: "Outer stream ended during failed handoff." }], + }); + expect(await taskService.getWorkspaceTurnSnapshot(parentId, "wst_handle")).toMatchObject({ + status: "running", + }); + + releaseTerminalSend(); + await Promise.all([completion, uncorrelatedEnd]); + + expect(await reservation.released).toBe(false); + expect(terminalContinuationAttempted).toBe(true); + expect(workspaceMocks.removeQueuedMessagesByDedupeKeyPrefix).not.toHaveBeenCalled(); + expect(await taskService.getWorkspaceTurnSnapshot(parentId, "wst_handle")).toMatchObject({ + status: "interrupted", + messageId: "outer-end-before-failed-terminal-send", + }); + const history = await historyService.getHistoryFromLatestBoundary("childworkspace"); + expect(history.success).toBe(true); + expect(JSON.stringify(history)).toContain("Nested work completed."); + }); + + test("failed terminal continuation enqueue re-evaluates a correlated stream-end", async () => { + let releaseTerminalSend!: () => void; + const terminalSendGate = new Promise((resolve) => { + releaseTerminalSend = resolve; + }); + let markTerminalSendStarted!: () => void; + const terminalSendStarted = new Promise((resolve) => { + markTerminalSendStarted = resolve; + }); + const sendMessage = mock( + async (...args: unknown[]): Promise> => { + const internal = args[3] as { queueDedupeKey?: string }; + if (internal?.queueDedupeKey?.startsWith("agent-terminal-report:") === true) { + markTerminalSendStarted(); + await terminalSendGate; + return Err({ type: "unknown", raw: "Terminal enqueue failed" }); + } + return Ok(undefined); + } + ); + const { config, parentId, taskService } = await startWorkspaceTurnForTest({ sendMessage }); + await config.editConfig((cfg) => { + const project = cfg.projects.get(path.join(rootDir, "repo")); + assert(project, "test project must exist"); + project.workspaces.push({ + path: path.join(rootDir, "repo", "nested-correlated-enqueue-failure"), + id: "nested-correlated-enqueue-failure", + name: "nested-correlated-enqueue-failure", + createdAt: "2026-08-21T00:00:00.000Z", + runtimeConfig: { type: "local" }, + parentWorkspaceId: "childworkspace", + taskStatus: "running", + agentType: "explore", + }); + return cfg; + }); + + const completion = handleTaskServiceStreamEndForTest(taskService, { + type: "stream-end", + workspaceId: "nested-correlated-enqueue-failure", + messageId: "assistant-nested-correlated-enqueue-failure", + metadata: { model: "anthropic:claude-opus-4-6", finishReason: "stop" }, + parts: [ + { + type: "dynamic-tool", + toolCallId: "nested-report-call", + toolName: "agent_report", + input: { reportMarkdown: "Nested work completed." }, + state: "output-available", + output: { + success: true, + report: { reportMarkdown: "Nested work completed." }, + }, + }, + { type: "text", text: "Nested work completed." }, + ], + }); + + await terminalSendStarted; + const internal = taskService as unknown as { + finalizeWorkspaceTurnFromStreamEnd: (event: StreamEndEvent) => Promise; + workspaceTurnContinuationReservationsByWorkspaceId: Map< + string, + Set<{ released: Promise }> + >; + }; + const reservation = Array.from( + internal.workspaceTurnContinuationReservationsByWorkspaceId.get("childworkspace") ?? [] + )[0]; + assert(reservation, "terminal continuation reservation must be published"); + + let correlatedEndSettled = false; + const correlatedEnd = internal + .finalizeWorkspaceTurnFromStreamEnd({ + type: "stream-end", + workspaceId: "childworkspace", + messageId: "outer-correlated-end-before-failed-terminal-send", + metadata: { + model: "anthropic:claude-opus-4-6", + agentId: "exec", + finishReason: "tool-calls", + muxMetadata: { + type: "workspace-turn-task", + taskHandleId: "wst_handle", + ownerWorkspaceId: parentId, + turnId: "turn", + }, + }, + parts: [{ type: "text", text: "Outer stream ended during terminal handoff." }], + }) + .then((handled) => { + correlatedEndSettled = true; + return handled; + }); + await Promise.resolve(); + expect(correlatedEndSettled).toBe(false); + + releaseTerminalSend(); + await Promise.all([completion, correlatedEnd]); + + expect(await reservation.released).toBe(false); + expect(await taskService.getWorkspaceTurnSnapshot(parentId, "wst_handle")).toMatchObject({ + status: "error", + messageId: "outer-correlated-end-before-failed-terminal-send", + }); + }); + + test("correlated settlement rechecks a reservation created after the first continuation check", async () => { + const { parentId, taskService } = await startWorkspaceTurnForTest(); + const internal = taskService as unknown as { + finalizeWorkspaceTurnFromStreamEnd: (event: StreamEndEvent) => Promise; + getReservedWorkspaceTurnContinuation: ( + workspaceId: string, + correlation: TestWorkspaceTurnMuxMetadata + ) => + | { + released: Promise; + markContinuationInstalled: () => void; + [Symbol.dispose]: () => void; + } + | undefined; + hasSameTurnContinuation: ( + event: StreamEndEvent, + correlation: TestWorkspaceTurnMuxMetadata + ) => Promise; + reserveActiveWorkspaceTurnContinuation: (workspaceId: string) => Promise< + | { + markContinuationInstalled: () => void; + [Symbol.dispose]: () => void; + } + | undefined + >; + }; + const originalHasSameTurnContinuation = internal.hasSameTurnContinuation.bind(taskService); + let releaseInitialCheck!: () => void; + const initialCheckGate = new Promise((resolve) => { + releaseInitialCheck = resolve; + }); + let markInitialCheckComplete!: () => void; + const initialCheckComplete = new Promise((resolve) => { + markInitialCheckComplete = resolve; + }); + const hasSameTurnContinuation = spyOn(internal, "hasSameTurnContinuation").mockImplementation( + async (event, correlation) => { + const result = await originalHasSameTurnContinuation(event, correlation); + if (!result) { + markInitialCheckComplete(); + await initialCheckGate; + } + return result; } ); - const { config, parentId, taskService } = await startWorkspaceTurnForTest({ sendMessage }); - await config.editConfig((cfg) => { - const project = cfg.projects.get(path.join(rootDir, "repo")); - assert(project, "test project must exist"); - project.workspaces.push({ - path: path.join(rootDir, "repo", "nested-progress-canceled"), - id: "nested-progress-canceled", - name: "nested-progress-canceled", - createdAt: "2026-06-19T00:00:00.000Z", - runtimeConfig: { type: "local" }, - parentWorkspaceId: "childworkspace", - taskStatus: "running", - agentType: "explore", + const event: StreamEndEvent = { + type: "stream-end", + workspaceId: "childworkspace", + messageId: "outer-correlated-end-before-late-reservation", + metadata: { + model: "anthropic:claude-opus-4-6", + agentId: "exec", + finishReason: "tool-calls", + muxMetadata: { + type: "workspace-turn-task", + taskHandleId: "wst_handle", + ownerWorkspaceId: parentId, + turnId: "turn", + }, + }, + parts: [{ type: "text", text: "Outer stream ended before a late handoff." }], + }; + + try { + let finalizationSettled = false; + const finalization = internal.finalizeWorkspaceTurnFromStreamEnd(event).then((handled) => { + finalizationSettled = true; + return handled; + }); + await initialCheckComplete; + const reservation = await internal.reserveActiveWorkspaceTurnContinuation("childworkspace"); + assert(reservation, "late workspace-turn continuation reservation must exist"); + let markReservationObserved!: () => void; + const reservationObserved = new Promise((resolve) => { + markReservationObserved = resolve; + }); + const originalGetReservedWorkspaceTurnContinuation = + internal.getReservedWorkspaceTurnContinuation.bind(taskService); + const getReservedWorkspaceTurnContinuation = spyOn( + internal, + "getReservedWorkspaceTurnContinuation" + ).mockImplementation((workspaceId, correlation) => { + const current = originalGetReservedWorkspaceTurnContinuation(workspaceId, correlation); + if (current === reservation) { + markReservationObserved(); + } + return current; }); - return cfg; - }); - await taskService.reportAgentProgress("nested-progress-canceled", "progress-call", { - reportMarkdown: "The progress wake was canceled.", - }); + try { + releaseInitialCheck(); + await reservationObserved; + expect(finalizationSettled).toBe(false); + expect(await taskService.getWorkspaceTurnSnapshot(parentId, "wst_handle")).toMatchObject({ + status: "running", + }); - expect(await taskService.getWorkspaceTurnSnapshot(parentId, "wst_handle")).toMatchObject({ - status: "interrupted", - error: "Progress wake was canceled", - }); + reservation.markContinuationInstalled(); + reservation[Symbol.dispose](); + expect(await finalization).toBe(true); + expect(await taskService.getWorkspaceTurnSnapshot(parentId, "wst_handle")).toMatchObject({ + status: "running", + deferredMessageIds: [event.messageId], + }); + } finally { + reservation[Symbol.dispose](); + getReservedWorkspaceTurnContinuation.mockRestore(); + } + } finally { + releaseInitialCheck(); + hasSameTurnContinuation.mockRestore(); + } }); - test("terminal nested agent report resumes a workspace turn with correlation", async () => { - const { config, parentId, taskService, workspaceMocks, historyService } = - await startWorkspaceTurnForTest(); + test("terminal nested agent report queues a correlated workspace turn continuation", async () => { + const { config, parentId, taskService, workspaceMocks } = await startWorkspaceTurnForTest(); await config.editConfig((cfg) => { const project = cfg.projects.get(path.join(rootDir, "repo")); assert(project, "test project must exist"); @@ -4407,9 +5916,25 @@ describe("TaskService", () => { taskStatus: "running", agentType: "explore", taskModelString: "anthropic:claude-opus-4-6", + taskExecutionId: "wst_nested_terminal_generation", + taskExecutionStatus: "running", }); return cfg; }); + const taskHandleStore = (taskService as unknown as { taskHandleStore: TaskHandleStore }) + .taskHandleStore; + await taskHandleStore.upsertWorkspaceTurn({ + kind: "workspace_turn", + handleId: "wst_nested_terminal_generation", + ownerWorkspaceId: "childworkspace", + workspaceId: "nested-terminal-agent", + turnId: "turn-nested-terminal-generation", + status: "running", + createdAt: "2026-06-19T00:00:00.000Z", + updatedAt: "2026-06-19T00:00:00.000Z", + createdWorkspace: false, + disposableWorkspace: false, + }); await handleTaskServiceStreamEndForTest(taskService, { type: "stream-end", @@ -4432,31 +5957,9 @@ describe("TaskService", () => { ], }); - const childHistory = await historyService.getHistoryFromLatestBoundary("childworkspace"); - expect(childHistory.success).toBe(true); - if (!childHistory.success) throw new Error("child history read failed"); - const reportMessage = childHistory.data.find( - (message) => - message.role === "user" && - message.parts.some( - (part) => - part.type === "text" && part.text.includes("The nested terminal report is complete.") - ) - ); - expect(reportMessage?.metadata?.muxMetadata).toEqual({ - type: "workspace-turn-task", - taskHandleId: "wst_handle", - ownerWorkspaceId: parentId, - turnId: "turn", - }); - - await Promise.all([ - ...(taskService as unknown as { pendingTerminalAttentionDrains: Set> }) - .pendingTerminalAttentionDrains, - ]); - - expect(workspaceMocks.resumeStream).toHaveBeenCalledWith( + expect(workspaceMocks.sendMessage).toHaveBeenCalledWith( "childworkspace", + expect.stringContaining("The nested terminal report is complete."), expect.objectContaining({ muxMetadata: { type: "workspace-turn-task", @@ -4465,8 +5968,156 @@ describe("TaskService", () => { turnId: "turn", }, }), - { agentInitiated: true } + expect.objectContaining({ + queueDedupeKey: + "agent-terminal-report:agent_task:nested-terminal-agent:wst_nested_terminal_generation", + workspaceTurnContinuation: true, + }) + ); + + const terminalCall = workspaceMocks.sendMessage.mock.calls.find((call) => { + const internal = call[3] as { queueDedupeKey?: string } | undefined; + return internal?.queueDedupeKey?.startsWith("agent-terminal-report:") === true; + }); + expect(parseSubagentReportEnvelope(String(terminalCall?.[1]))).toMatchObject({ + taskId: "nested-terminal-agent", + executionVersion: "wst_nested_terminal_generation", + reportMarkdown: "The nested terminal report is complete.", + }); + }); + + test("terminal recovery appends the report for the pending execution generation", async () => { + const config = await createTestConfig(rootDir); + const { parentId } = await saveLocalParentWorkspace(config, rootDir); + const childTaskId = "persistent-child-generation-recovery"; + const previousGeneration = "wst_previous_generation"; + const pendingGeneration = "wst_pending_generation"; + const { historyService, taskService } = createTaskServiceHarness(config); + + await historyService.appendToHistory( + parentId, + createMuxMessage( + "previous-generation-report", + "user", + formatSubagentReportEnvelope({ + taskId: childTaskId, + agentType: "explore", + status: "completed", + executionVersion: previousGeneration, + title: "Previous generation", + reportMarkdown: "The previous execution completed.", + }), + { timestamp: Date.parse("2026-08-21T00:00:00.000Z"), synthetic: true, uiVisible: true } + ) ); + await upsertSubagentReportArtifact({ + workspaceId: parentId, + workspaceSessionDir: config.getSessionDir(parentId), + childTaskId, + parentWorkspaceId: parentId, + ancestorWorkspaceIds: [parentId], + executionVersion: pendingGeneration, + reportMarkdown: "The latest execution completed.", + title: "Latest generation", + nowMs: Date.parse("2026-08-21T00:00:01.000Z"), + }); + const terminalAttentionStore = new TerminalAttentionStore(config); + const notification = await terminalAttentionStore.enqueueIfAbsent({ + ownerWorkspaceId: parentId, + sourceKind: "agent_task", + sourceId: childTaskId, + generationId: pendingGeneration, + }); + assert(notification, "terminal attention notification must be created"); + + const ensureResult = await ( + taskService as unknown as { + ensureAgentTerminalMessages: ( + ownerWorkspaceId: string, + notifications: ReadonlyArray + ) => Promise<{ deliverableNotificationIds: Set }>; + } + ).ensureAgentTerminalMessages(parentId, [notification]); + + expect(ensureResult.deliverableNotificationIds.has(notification.id)).toBe(true); + const historyResult = await historyService.getHistoryFromLatestBoundary(parentId); + expect(historyResult.success).toBe(true); + if (!historyResult.success) throw new Error("parent history read failed"); + const terminalReports = historyResult.data.flatMap((message) => { + if (message.role !== "user" || message.metadata?.synthetic !== true) return []; + const content = message.parts + .filter((part): part is Extract => part.type === "text") + .map((part) => part.text) + .join("\n"); + const parsed = parseSubagentReportEnvelope(content); + return parsed?.taskId === childTaskId ? [parsed] : []; + }); + expect(terminalReports).toHaveLength(2); + expect( + terminalReports.some( + (report) => + report.taskId === childTaskId && + report.executionVersion === pendingGeneration && + report.reportMarkdown === "The latest execution completed." + ) + ).toBe(true); + }); + + test("terminal recovery uses the notification outcome when legacy artifacts conflict", async () => { + const config = await createTestConfig(rootDir); + const { parentId } = await saveLocalParentWorkspace(config, rootDir); + const childTaskId = "persistent-child-conflicting-artifacts"; + const pendingGeneration = "wst_pending_failure_generation"; + const sessionDir = config.getSessionDir(parentId); + await upsertSubagentReportArtifact({ + workspaceId: parentId, + workspaceSessionDir: sessionDir, + childTaskId, + parentWorkspaceId: parentId, + ancestorWorkspaceIds: [parentId], + reportMarkdown: "This success belongs to an older legacy execution.", + nowMs: Date.parse("2026-08-21T00:00:00.000Z"), + }); + await upsertSubagentFailureArtifact({ + workspaceId: parentId, + workspaceSessionDir: sessionDir, + childTaskId, + parentWorkspaceId: parentId, + ancestorWorkspaceIds: [parentId], + executionVersion: pendingGeneration, + errorType: "authentication", + errorMessage: "The current execution failed authentication.", + nowMs: Date.parse("2026-08-21T00:00:01.000Z"), + }); + const terminalAttentionStore = new TerminalAttentionStore(config); + const notification = await terminalAttentionStore.enqueueIfAbsent({ + ownerWorkspaceId: parentId, + sourceKind: "agent_task", + sourceId: childTaskId, + generationId: pendingGeneration, + terminalOutcome: "failed", + }); + assert(notification, "terminal attention notification must be created"); + const { historyService, taskService } = createTaskServiceHarness(config); + + const ensureResult = await ( + taskService as unknown as { + ensureAgentTerminalMessages: ( + ownerWorkspaceId: string, + notifications: ReadonlyArray + ) => Promise<{ deliverableNotificationIds: Set }>; + } + ).ensureAgentTerminalMessages(parentId, [notification]); + + expect(ensureResult.deliverableNotificationIds.has(notification.id)).toBe(true); + const historyResult = await historyService.getHistoryFromLatestBoundary(parentId); + expect(historyResult.success).toBe(true); + if (!historyResult.success) throw new Error("parent history read failed"); + const serializedHistory = JSON.stringify(historyResult.data); + expect(serializedHistory).toContain(""); + expect(serializedHistory).toContain(pendingGeneration); + expect(serializedHistory).toContain("The current execution failed authentication."); + expect(serializedHistory).not.toContain("This success belongs to an older legacy execution."); }); test("backfills workspace-turn correlation on an existing terminal report", async () => { @@ -7112,6 +8763,83 @@ describe("TaskService", () => { expect(snapshot?.reportMarkdown).toBeUndefined(); }); + test("an exact continuation keeps an uncorrelated stream-end nonterminal", async () => { + let continuationPending = true; + const hasPendingWorkspaceTurnContinuation = mock( + ( + workspaceId: string, + metadata: { taskHandleId: string; ownerWorkspaceId: string; turnId: string } + ) => + continuationPending && + workspaceId === "childworkspace" && + metadata.taskHandleId === "wst_handle" && + metadata.turnId === "turn" + ); + const { parentId, taskService } = await startWorkspaceTurnForTest({ + hasPendingWorkspaceTurnContinuation, + }); + let waiterSettled = false; + const waiter = taskService + .waitForWorkspaceTurn("wst_handle", { + ownerWorkspaceId: parentId, + requestingWorkspaceId: parentId, + backgroundOnMessageQueued: false, + timeoutMs: 10_000, + }) + .then((result) => { + waiterSettled = true; + return result; + }); + const internal = taskService as unknown as { + handleStreamEnd: (event: StreamEndEvent) => Promise; + }; + + await internal.handleStreamEnd({ + type: "stream-end", + workspaceId: "childworkspace", + messageId: "msg_uncorrelated_continuation", + metadata: { + model: "anthropic:claude-opus-4-6", + agentId: "exec", + finishReason: "stop", + }, + parts: [{ type: "text", text: "Synthetic continuation output" }], + }); + expect(await taskService.getWorkspaceTurnSnapshot(parentId, "wst_handle")).toMatchObject({ + status: "running", + }); + expect(waiterSettled).toBe(false); + + continuationPending = false; + await internal.handleStreamEnd({ + type: "stream-end", + workspaceId: "childworkspace", + messageId: "msg_correlated_final", + metadata: { + model: "anthropic:claude-opus-4-6", + agentId: "exec", + finishReason: "stop", + muxMetadata: { + type: "workspace-turn-task", + taskHandleId: "wst_handle", + ownerWorkspaceId: parentId, + turnId: "turn", + }, + }, + parts: [{ type: "text", text: "The delegated turn completed." }], + }); + + expect(await waiter).toMatchObject({ + workspaceId: "childworkspace", + reportMarkdown: "The delegated turn completed.", + }); + expect(await taskService.getWorkspaceTurnSnapshot(parentId, "wst_handle")).toMatchObject({ + status: "completed", + messageId: "msg_correlated_final", + reportMarkdown: "The delegated turn completed.", + }); + }); + test("workspace-turn stream errors mark the handle failed", async () => { const config = await createTestConfig(rootDir); stubStableIds(config, ["handle", "turn"]); @@ -17633,12 +19361,16 @@ describe("TaskService", () => { await removeWorkspaceFromTestConfig(config, workspaceId); return Ok(undefined); }); - const { workspaceService } = createWorkspaceServiceMocks({ remove }); + const workspaceMocks = createWorkspaceServiceMocks({ remove }); return { config, remove, - ...createTaskServiceHarness(config, { aiService, workspaceService }), + workspaceMocks, + ...createTaskServiceHarness(config, { + aiService, + workspaceService: workspaceMocks.workspaceService, + }), }; } @@ -17817,7 +19549,7 @@ describe("TaskService", () => { const childTwoId = "child-best-of-2"; const bestOf = { groupId: "best-of-group", index: 0, total: 2 } as const; - const { config, historyService, partialService, taskService, remove } = + const { config, historyService, partialService, taskService, remove, workspaceMocks } = await createBestOfTaskServiceTestHarness({ parentId, children: [ @@ -17861,6 +19593,7 @@ describe("TaskService", () => { const afterFirstParentPartial = await partialService.readPartial(parentId); expect(afterFirstParentPartial).not.toBeNull(); expect(getTaskToolPart(afterFirstParentPartial)?.state).toBe("input-available"); + expect(workspaceMocks.removeQueuedMessagesByDedupeKeyPrefix).not.toHaveBeenCalled(); expect(remove).not.toHaveBeenCalled(); await finalizeReportedChildTaskForTest({ @@ -17883,6 +19616,18 @@ describe("TaskService", () => { expect(serializedOutput).toContain("Report from child one"); expect(serializedOutput).toContain("Report from child two"); + expect(workspaceMocks.removeQueuedMessagesByDedupeKeyPrefix).toHaveBeenCalledTimes(2); + expect(workspaceMocks.removeQueuedMessagesByDedupeKeyPrefix).toHaveBeenCalledWith( + parentId, + `agent-report:${childOneId}:`, + expect.objectContaining({ notifyCancellation: false }) + ); + expect(workspaceMocks.removeQueuedMessagesByDedupeKeyPrefix).toHaveBeenCalledWith( + parentId, + `agent-report:${childTwoId}:`, + expect.objectContaining({ notifyCancellation: false }) + ); + const remainingTaskIds = getConfiguredWorkspaceIds(config); expect(remainingTaskIds).toContain(childOneId); expect(remainingTaskIds).toContain(childTwoId); @@ -19994,7 +21739,10 @@ describe("TaskService", () => { expect(removeQueuedMessagesByDedupeKeyPrefix).toHaveBeenCalledWith( parentId, `agent-report:${childId}:`, - { cancelReason: "Incremental sub-agent update superseded by the terminal report." } + { + cancelReason: "Incremental sub-agent update superseded by the terminal report.", + notifyCancellation: false, + } ); }); @@ -21650,10 +23398,10 @@ describe("TaskService", () => { await removeWorkspaceFromTestConfig(config, workspaceId); return Ok(undefined); }); - const { workspaceService } = createWorkspaceServiceMocks({ remove }); + const workspaceMocks = createWorkspaceServiceMocks({ remove }); const { partialService, taskService } = createTaskServiceHarness(config, { aiService, - workspaceService, + workspaceService: workspaceMocks.workspaceService, }); const parentPartial = createMuxMessage( @@ -21733,6 +23481,18 @@ describe("TaskService", () => { expect(outputJson).toContain("Report from child two"); } + expect(workspaceMocks.removeQueuedMessagesByDedupeKeyPrefix).toHaveBeenCalledTimes(2); + expect(workspaceMocks.removeQueuedMessagesByDedupeKeyPrefix).toHaveBeenCalledWith( + parentId, + `agent-report:${childOneId}:`, + expect.objectContaining({ notifyCancellation: false }) + ); + expect(workspaceMocks.removeQueuedMessagesByDedupeKeyPrefix).toHaveBeenCalledWith( + parentId, + `agent-report:${childTwoId}:`, + expect.objectContaining({ notifyCancellation: false }) + ); + const remainingTaskIds = Array.from(config.loadConfigOrDefault().projects.values()) .flatMap((project) => project.workspaces) .map((workspace) => workspace.id) diff --git a/src/node/services/taskService.ts b/src/node/services/taskService.ts index a519ae0993..e1fc6ce9b7 100644 --- a/src/node/services/taskService.ts +++ b/src/node/services/taskService.ts @@ -528,6 +528,13 @@ export interface WorkspaceTurnWaitResult { type WorkspaceTurnMuxMetadata = Extract; +interface WorkspaceTurnContinuationReservation { + muxMetadata: WorkspaceTurnMuxMetadata; + released: Promise; + markContinuationInstalled(): void; + [Symbol.dispose](): void; +} + interface BackgroundableForegroundWaiter { taskId: string; reject: (error: Error) => void; @@ -703,6 +710,10 @@ const WORKSPACE_TURN_RECOVERABLE_STREAM_ERRORS: ReadonlySet = n /** Marker persisted by settleStaleWorkspaceTurn when restart recovery interrupts a handle. */ const WORKSPACE_TURN_STALE_RESTART_ERROR = "Workspace turn interrupted after restart"; +/** Marker for an inferred interruption that later correlated evidence can correct. */ +const WORKSPACE_TURN_UNCORRELATED_STREAM_END_ERROR = + "Workspace turn superseded by an uncorrelated workspace stream-end"; + /** * Reason persisted when other queued input (a manual user message, /compact) * cut a delegated turn at a tool boundary and superseded it. The target @@ -1311,6 +1322,11 @@ export class TaskService { // In-flight durable persistence of notify_on_terminal policy for backgrounded foreground waits. // Awaited at the start of handleStreamEnd so a just-detached wait is treated as non-blocking. private readonly pendingNotifyOnTerminalPersists = new Set>(); + // Serialize notification claims. A drain and a direct continuation must not own the same wake. + private readonly terminalAttentionLocks = new MutexMap(); + // Claims hide notifications while one path owns their delivery. The durable pending record remains + // available after a process restart because claims are intentionally in memory only. + private readonly claimedTerminalAttentionIdsByOwner = new Map>(); // In-flight terminal attention drains (workspace-turn / sub-agent terminal wake-ups). Tracked so // tests and shutdown can await them; drains are idempotent and re-triggered on owner idle events. private readonly pendingTerminalAttentionDrainsByOwner = new Map>(); @@ -1330,6 +1346,12 @@ export class TaskService { string, { handleId: string; ownerWorkspaceId: string } >(); + // Bridge the short gap between selecting an active turn and installing its concrete queued or + // streaming continuation. Settlement checks this exact correlation under the same handle lock. + private readonly workspaceTurnContinuationReservationsByWorkspaceId = new Map< + string, + Set + >(); private readonly taskHandleStore: TaskHandleStore; private readonly terminalAttentionStore: TerminalAttentionStore; private readonly userBackgroundedTaskIds = new Set(); @@ -5894,6 +5916,73 @@ export class TaskService { return execution?.record.handleId; } + private claimTerminalAttention(ownerWorkspaceId: string, notificationId: string): boolean { + const claimedIds = this.claimedTerminalAttentionIdsByOwner.get(ownerWorkspaceId) ?? new Set(); + if (claimedIds.has(notificationId)) { + return false; + } + claimedIds.add(notificationId); + this.claimedTerminalAttentionIdsByOwner.set(ownerWorkspaceId, claimedIds); + return true; + } + + private claimTerminalAttentionNotifications( + ownerWorkspaceId: string, + notifications: readonly TerminalAttentionNotification[] + ): boolean { + const claimedIds = this.claimedTerminalAttentionIdsByOwner.get(ownerWorkspaceId) ?? new Set(); + if (notifications.some((notification) => claimedIds.has(notification.id))) { + return false; + } + for (const notification of notifications) { + claimedIds.add(notification.id); + } + this.claimedTerminalAttentionIdsByOwner.set(ownerWorkspaceId, claimedIds); + return true; + } + + private releaseTerminalAttentionClaim(ownerWorkspaceId: string, notificationId: string): void { + const claimedIds = this.claimedTerminalAttentionIdsByOwner.get(ownerWorkspaceId); + if (claimedIds == null) { + return; + } + claimedIds.delete(notificationId); + if (claimedIds.size === 0) { + this.claimedTerminalAttentionIdsByOwner.delete(ownerWorkspaceId); + } + } + + private async claimPendingTerminalAttention( + ownerWorkspaceId: string + ): Promise { + return this.terminalAttentionLocks.withLock(ownerWorkspaceId, async () => { + const pending = await this.terminalAttentionStore.listPending(ownerWorkspaceId); + return pending.filter((notification) => + this.claimTerminalAttention(ownerWorkspaceId, notification.id) + ); + }); + } + + private releaseTerminalAttentionClaims( + ownerWorkspaceId: string, + notifications: readonly TerminalAttentionNotification[] + ): void { + for (const notification of notifications) { + this.releaseTerminalAttentionClaim(ownerWorkspaceId, notification.id); + } + } + + private async supersedePendingTerminalAttention(ownerWorkspaceId: string): Promise { + const pending = await this.claimPendingTerminalAttention(ownerWorkspaceId); + try { + for (const notification of pending) { + await this.terminalAttentionStore.markSuperseded(ownerWorkspaceId, notification.id); + } + } finally { + this.releaseTerminalAttentionClaims(ownerWorkspaceId, pending); + } + } + private async enqueueTerminalAttention(params: { ownerWorkspaceId: string; sourceKind: TerminalAttentionNotification["sourceKind"]; @@ -5988,32 +6077,42 @@ export class TaskService { notifications: readonly TerminalAttentionNotification[] ): Promise<{ deliverableNotificationIds: Set; - latestMessageTimestampByTaskId: Map; + terminalMessageNotificationIds: Set; + latestLegacyMessageTimestampByTaskId: Map; }> { const deliverableNotificationIds = new Set(); - const latestMessageTimestampByTaskId = new Map(); + const terminalMessageNotificationIds = new Set(); + const latestLegacyMessageTimestampByTaskId = new Map(); const historyResult = await this.historyService.getHistoryFromLatestBoundary(ownerWorkspaceId); if (!historyResult.success) { - return { deliverableNotificationIds, latestMessageTimestampByTaskId }; + return { + deliverableNotificationIds, + terminalMessageNotificationIds, + latestLegacyMessageTimestampByTaskId, + }; } - const existingReportMessages = new Map(); - const existingTaskIds = new Set(); + const existingTerminalMessagesByNotificationId = new Map(); for (const message of historyResult.data) { if (message.role !== "user" || message.metadata?.synthetic !== true) continue; - const taskId = parseTerminalSubagentTaskId( - message.parts - .filter((part): part is Extract => part.type === "text") - .map((part) => part.text) - .join("\n") - ); + const content = message.parts + .filter((part): part is Extract => part.type === "text") + .map((part) => part.text) + .join("\n"); + const taskId = parseTerminalSubagentTaskId(content); if (taskId == null) continue; - existingTaskIds.add(taskId); - existingReportMessages.set(taskId, message); + const executionVersion = parseTerminalSubagentExecutionVersion(content) ?? undefined; + const notificationId = TerminalAttentionStore.notificationId( + "agent_task", + taskId, + executionVersion + ); + terminalMessageNotificationIds.add(notificationId); + existingTerminalMessagesByNotificationId.set(notificationId, message); const timestamp = message.metadata?.timestamp; - if (typeof timestamp === "number" && Number.isFinite(timestamp)) { - latestMessageTimestampByTaskId.set( + if (executionVersion == null && typeof timestamp === "number" && Number.isFinite(timestamp)) { + latestLegacyMessageTimestampByTaskId.set( taskId, - Math.max(latestMessageTimestampByTaskId.get(taskId) ?? 0, timestamp) + Math.max(latestLegacyMessageTimestampByTaskId.get(taskId) ?? 0, timestamp) ); } } @@ -6022,8 +6121,8 @@ export class TaskService { await this.getActiveWorkspaceTurnMuxMetadataForWorkspace(ownerWorkspaceId); const sessionDir = this.config.getSessionDir(ownerWorkspaceId); for (const notification of notifications) { - if (existingTaskIds.has(notification.sourceId)) { - const existingMessage = existingReportMessages.get(notification.sourceId); + if (terminalMessageNotificationIds.has(notification.id)) { + const existingMessage = existingTerminalMessagesByNotificationId.get(notification.id); if (existingMessage != null && workspaceTurnMuxMetadata != null) { const existingCorrelation = this.getWorkspaceTurnMetadataFromValue( existingMessage.metadata?.muxMetadata @@ -6051,7 +6150,7 @@ export class TaskService { updatedMessage ); if (updateResult.success) { - existingReportMessages.set(notification.sourceId, updatedMessage); + existingTerminalMessagesByNotificationId.set(notification.id, updatedMessage); this.workspaceService.emitChatEvent(ownerWorkspaceId, { ...updatedMessage, type: "message", @@ -6073,11 +6172,28 @@ export class TaskService { continue; } - const report = await readSubagentReportArtifact(sessionDir, notification.sourceId); - const failure = - report == null + const storedReport = + notification.terminalOutcome === "completed" + ? await readSubagentReportArtifact(sessionDir, notification.sourceId) + : null; + const report = + storedReport != null && + (storedReport.executionVersion == null || + storedReport.executionVersion === notification.generationId) + ? storedReport + : null; + const storedFailure = + notification.terminalOutcome !== "completed" ? await readSubagentFailureArtifact(sessionDir, notification.sourceId) : null; + const failure = + storedFailure != null && + (storedFailure.executionVersion == null || + storedFailure.executionVersion === notification.generationId) + ? storedFailure + : null; + const executionVersion = + report?.executionVersion ?? failure?.executionVersion ?? notification.generationId; const content = report ? formatSubagentReportUserMessage({ childWorkspaceId: notification.sourceId, @@ -6085,6 +6201,7 @@ export class TaskService { title: report.title ?? "Subagent report", reportMarkdown: report.reportMarkdown, status: "completed", + ...(executionVersion != null ? { executionVersion } : {}), ...(report.model != null ? { model: report.model } : {}), ...(report.thinkingLevel != null ? { thinkingLevel: report.thinkingLevel } : {}), ...(report.structuredOutput !== undefined @@ -6095,6 +6212,7 @@ export class TaskService { ? formatSubagentFailureUserMessage({ childWorkspaceId: notification.sourceId, agentType: "agent", + ...(executionVersion != null ? { executionVersion } : {}), errorType: failure.errorType, errorMessage: failure.errorMessage, }) @@ -6130,11 +6248,15 @@ export class TaskService { continue; } this.workspaceService.emitChatEvent(ownerWorkspaceId, { ...message, type: "message" }); - existingTaskIds.add(notification.sourceId); - latestMessageTimestampByTaskId.set(notification.sourceId, timestamp); + terminalMessageNotificationIds.add(notification.id); + existingTerminalMessagesByNotificationId.set(notification.id, message); deliverableNotificationIds.add(notification.id); } - return { deliverableNotificationIds, latestMessageTimestampByTaskId }; + return { + deliverableNotificationIds, + terminalMessageNotificationIds, + latestLegacyMessageTimestampByTaskId, + }; } private async consumeRespondedAgentTerminalAttention(ownerWorkspaceId: string): Promise { @@ -6143,9 +6265,9 @@ export class TaskService { ); if (pending.length === 0) return; - const pendingIds = new Set(pending.map((notification) => notification.sourceId)); - const terminalSequenceByTaskId = new Map(); - const responded = new Set(); + const pendingNotificationIds = new Set(pending.map((notification) => notification.id)); + const terminalSequenceByNotificationId = new Map(); + const respondedNotificationIds = new Set(); const historyResult = await this.historyService.getHistoryFromLatestBoundary(ownerWorkspaceId); if (!historyResult.success) { log.warn("Failed to inspect terminal sub-agent responses", { @@ -6163,9 +6285,17 @@ export class TaskService { .join("\n"); const taskId = parseTerminalSubagentTaskId(text); const historySequence = message.metadata?.historySequence; - if (taskId != null && pendingIds.has(taskId) && typeof historySequence === "number") { - terminalSequenceByTaskId.set(taskId, historySequence); - responded.delete(taskId); + if (taskId != null && typeof historySequence === "number") { + const executionVersion = parseTerminalSubagentExecutionVersion(text) ?? undefined; + const notificationId = TerminalAttentionStore.notificationId( + "agent_task", + taskId, + executionVersion + ); + if (pendingNotificationIds.has(notificationId)) { + terminalSequenceByNotificationId.set(notificationId, historySequence); + respondedNotificationIds.delete(notificationId); + } } continue; } @@ -6173,14 +6303,16 @@ export class TaskService { if (message.role === "assistant" && message.metadata?.partial !== true) { const requestHistorySequence = message.metadata?.requestHistorySequence; if (typeof requestHistorySequence !== "number") continue; - for (const [taskId, terminalSequence] of terminalSequenceByTaskId) { - if (requestHistorySequence >= terminalSequence) responded.add(taskId); + for (const [notificationId, terminalSequence] of terminalSequenceByNotificationId) { + if (requestHistorySequence >= terminalSequence) { + respondedNotificationIds.add(notificationId); + } } } } for (const notification of pending) { - if (responded.has(notification.sourceId)) { + if (respondedNotificationIds.has(notification.id)) { await this.terminalAttentionStore.markDelivered(ownerWorkspaceId, notification.id); } } @@ -6192,8 +6324,8 @@ export class TaskService { * drained notifications delivered. Stale (deleted-workspace) notifications are marked superseded. */ private async drainTerminalAttention(ownerWorkspaceId: string): Promise { - const pending = await this.terminalAttentionStore.listPending(ownerWorkspaceId); - if (pending.length === 0) { + const pendingAtStart = await this.terminalAttentionStore.listPending(ownerWorkspaceId); + if (pendingAtStart.length === 0) { return; } @@ -6201,16 +6333,12 @@ export class TaskService { const entry = findWorkspaceEntry(cfg, ownerWorkspaceId); if (entry == null) { // Owner workspace no longer exists: the terminal artifacts remain retrievable elsewhere. - for (const notification of pending) { - await this.terminalAttentionStore.markSuperseded(ownerWorkspaceId, notification.id); - } + await this.supersedePendingTerminalAttention(ownerWorkspaceId); return; } if (isWorkspaceArchived(entry.workspace.archivedAt, entry.workspace.unarchivedAt)) { - for (const notification of pending) { - await this.terminalAttentionStore.markSuperseded(ownerWorkspaceId, notification.id); - } + await this.supersedePendingTerminalAttention(ownerWorkspaceId); return; } @@ -6237,12 +6365,32 @@ export class TaskService { return; } + const pending = await this.claimPendingTerminalAttention(ownerWorkspaceId); + if (pending.length === 0) { + return; + } + + let releasePendingClaimsOnReturn = true; + using _terminalAttentionClaims = { + [Symbol.dispose]: () => { + if (releasePendingClaimsOnReturn) { + this.releaseTerminalAttentionClaims(ownerWorkspaceId, pending); + } + }, + }; + + // Publish the exact active-turn handoff before report reconstruction and model-option reads. + // A direct delivery that finds this notification claimed can then release its own reservation. + using initialWorkspaceTurnContinuation = + await this.reserveActiveWorkspaceTurnContinuation(ownerWorkspaceId); + const agentNotifications = pending.filter( (notification) => notification.sourceKind === "agent_task" ); const { deliverableNotificationIds: deliverableAgentNotificationIds, - latestMessageTimestampByTaskId, + terminalMessageNotificationIds, + latestLegacyMessageTimestampByTaskId, } = await this.ensureAgentTerminalMessages(ownerWorkspaceId, agentNotifications); const workspaceTurnNotifications = pending.filter( (notification) => notification.sourceKind === "workspace_turn" @@ -6262,12 +6410,22 @@ export class TaskService { record.workspaceId ); if (isPersistentChildContinuation) { - const latestTerminalMessageAt = latestMessageTimestampByTaskId.get(record.workspaceId); + const deliveryVersion = this.workspaceTurnTerminalAttentionGenerationId(record); + const hasVersionedTerminalMessage = [record.handleId, deliveryVersion].some( + (generationId) => + terminalMessageNotificationIds.has( + TerminalAttentionStore.notificationId("agent_task", record.workspaceId, generationId) + ) + ); + const latestLegacyTerminalMessageAt = latestLegacyMessageTimestampByTaskId.get( + record.workspaceId + ); const continuationCreatedAt = Date.parse(record.createdAt); if ( - latestTerminalMessageAt != null && - Number.isFinite(continuationCreatedAt) && - latestTerminalMessageAt >= continuationCreatedAt + hasVersionedTerminalMessage || + (latestLegacyTerminalMessageAt != null && + Number.isFinite(continuationCreatedAt) && + latestLegacyTerminalMessageAt >= continuationCreatedAt) ) { // A persistent child continuation reports through the stable child transcript row. Once // that report/failure is in parent history, a second task_await wake for the private @@ -6335,8 +6493,13 @@ export class TaskService { entry, defaultModel ); - const workspaceTurnMuxMetadata = - await this.getActiveWorkspaceTurnMuxMetadataForWorkspace(ownerWorkspaceId); + using lateWorkspaceTurnContinuation = + initialWorkspaceTurnContinuation == null + ? await this.reserveActiveWorkspaceTurnContinuation(ownerWorkspaceId) + : undefined; + const workspaceTurnContinuation = + initialWorkspaceTurnContinuation ?? lateWorkspaceTurnContinuation; + const workspaceTurnMuxMetadata = workspaceTurnContinuation?.muxMetadata; const sendOptions = { model: resumeOptions.model, @@ -6363,6 +6526,7 @@ export class TaskService { this.scheduleTerminalAttentionDrainAfterIdle(ownerWorkspaceId); return; } + workspaceTurnContinuation?.markContinuationInstalled(); await markPendingDelivered(); return; } @@ -6386,6 +6550,7 @@ export class TaskService { !(await this.hasBlockingActiveWorkForTerminalDrain(ownerWorkspaceId, latestTaskIndex)) ) { let fallbackAccepted = false; + let fallbackStartupFailed = false; sendResult = await this.workspaceService.sendMessage( ownerWorkspaceId, prompt, @@ -6394,20 +6559,60 @@ export class TaskService { skipAutoResumeReset: true, synthetic: true, agentInitiated: true, + workspaceTurnContinuation: workspaceTurnMuxMetadata != null, onCanceled: () => { + this.releaseTerminalAttentionClaims(ownerWorkspaceId, pending); this.scheduleTerminalAttentionDrainAfterIdle(ownerWorkspaceId); }, onAcceptedPreStreamFailure: async () => { - await markPendingForRetry(); - this.scheduleTerminalAttentionDrainAfterIdle(ownerWorkspaceId); + fallbackStartupFailed = true; + const retryClaimed = + !fallbackAccepted || + (await this.terminalAttentionLocks.withLock(ownerWorkspaceId, () => + Promise.resolve( + this.claimTerminalAttentionNotifications(ownerWorkspaceId, pending) + ) + )); + if (!retryClaimed) { + return; + } + try { + await markPendingForRetry(); + } finally { + this.releaseTerminalAttentionClaims(ownerWorkspaceId, pending); + this.scheduleTerminalAttentionDrainAfterIdle(ownerWorkspaceId); + } }, onAccepted: async () => { fallbackAccepted = true; - await markPendingDelivered(); + try { + await markPendingDelivered(); + } finally { + this.releaseTerminalAttentionClaims(ownerWorkspaceId, pending); + } }, } ); + if (!sendResult.success && fallbackAccepted && !fallbackStartupFailed) { + const retryClaimed = await this.terminalAttentionLocks.withLock(ownerWorkspaceId, () => + Promise.resolve(this.claimTerminalAttentionNotifications(ownerWorkspaceId, pending)) + ); + if (retryClaimed) { + try { + await markPendingForRetry(); + } finally { + this.releaseTerminalAttentionClaims(ownerWorkspaceId, pending); + this.scheduleTerminalAttentionDrainAfterIdle(ownerWorkspaceId); + } + } + } + if (sendResult.success && fallbackStartupFailed) { + return; + } if (sendResult.success && !fallbackAccepted) { + workspaceTurnContinuation?.markContinuationInstalled(); + // The queued callbacks now own these claims until acceptance or cancellation. + releasePendingClaimsOnReturn = false; return; } } @@ -6422,6 +6627,7 @@ export class TaskService { return; } + workspaceTurnContinuation?.markContinuationInstalled(); await markPendingDelivered(); } @@ -6804,6 +7010,8 @@ export class TaskService { * same turn, and the correlated stream-end proves the turn's real outcome. */ allowTerminalResettle?: boolean; + /** Recheck a nonterminal settlement guard while holding the handle lock. */ + shouldSettleCurrent?: (current: WorkspaceTurnTaskHandleRecord) => boolean; }): Promise { assert( params.next.handleId === params.record.handleId, @@ -6884,6 +7092,10 @@ export class TaskService { return { pendingNotify: null, winningStatus: current.status }; } + if (params.shouldSettleCurrent?.(current) === false) { + return null; + } + // 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. @@ -10484,64 +10696,138 @@ export class TaskService { private async interruptWorkspaceTurnFromUncorrelatedStreamEnd( event: StreamEndEvent ): Promise { - const active = this.activeWorkspaceTurnHandleByWorkspaceId.get(event.workspaceId); - if (active == null) { - return false; - } - const record = await this.taskHandleStore.getWorkspaceTurn( - active.ownerWorkspaceId, - active.handleId - ); - if (record == null) { - this.activeWorkspaceTurnHandleByWorkspaceId.delete(event.workspaceId); - log.warn("Ignoring missing uncorrelated workspace turn stream-end handle", { - workspaceId: event.workspaceId, - taskHandleId: active.handleId, - }); - return true; - } - if (record.workspaceId !== event.workspaceId) { - log.warn("Ignoring out-of-scope uncorrelated workspace turn stream-end", { - workspaceId: event.workspaceId, - taskHandleId: record.handleId, + let checkedStreamEndOrderingHandleId: string | undefined; + let matchedActiveTurn = false; + while (true) { + const active = this.activeWorkspaceTurnHandleByWorkspaceId.get(event.workspaceId); + if (active == null) { + return matchedActiveTurn; + } + matchedActiveTurn = true; + + const record = await this.taskHandleStore.getWorkspaceTurn( + active.ownerWorkspaceId, + active.handleId + ); + if (record == null) { + this.activeWorkspaceTurnHandleByWorkspaceId.delete(event.workspaceId); + log.warn("Ignoring missing uncorrelated workspace turn stream-end handle", { + workspaceId: event.workspaceId, + taskHandleId: active.handleId, + }); + return true; + } + if (record.workspaceId !== event.workspaceId) { + log.warn("Ignoring out-of-scope uncorrelated workspace turn stream-end", { + workspaceId: event.workspaceId, + taskHandleId: record.handleId, + }); + return false; + } + if (record.status !== "starting" && record.status !== "running") { + this.activeWorkspaceTurnHandleByWorkspaceId.delete(event.workspaceId); + return true; + } + + if (checkedStreamEndOrderingHandleId !== record.handleId) { + checkedStreamEndOrderingHandleId = record.handleId; + if (await this.isStreamEndBeforeWorkspaceTurnPrompt(record, event)) { + log.debug("Ignoring stale uncorrelated stream-end before queued workspace turn prompt", { + workspaceId: event.workspaceId, + taskHandleId: record.handleId, + streamEndMessageId: event.messageId, + }); + return true; + } + } + + const correlation = this.buildWorkspaceTurnMuxMetadata(record); + if (this.hasConcreteSameTurnContinuation(event, correlation)) { + log.debug("Deferring uncorrelated stream-end to an exact workspace turn continuation", { + workspaceId: event.workspaceId, + taskHandleId: record.handleId, + streamEndMessageId: event.messageId, + }); + return true; + } + + const error = WORKSPACE_TURN_UNCORRELATED_STREAM_END_ERROR; + const next: WorkspaceTurnTaskHandleRecord = { + ...record, + status: "interrupted", + updatedAt: getIsoNow(), + messageId: event.messageId, + error, + }; + let concreteContinuationInstalled = false; + let reservedContinuation: WorkspaceTurnContinuationReservation | undefined; + await this.settleWorkspaceTurn({ + record, + next, + waiterSettlement: { status: "error", error: new Error(error) }, + shouldSettleCurrent: (current) => { + const currentCorrelation = this.buildWorkspaceTurnMuxMetadata(current); + if (this.hasConcreteSameTurnContinuation(event, currentCorrelation)) { + concreteContinuationInstalled = true; + return false; + } + reservedContinuation = this.getReservedWorkspaceTurnContinuation( + event.workspaceId, + currentCorrelation + ); + return reservedContinuation == null; + }, }); - return false; - } - if (record.status !== "starting" && record.status !== "running") { - this.activeWorkspaceTurnHandleByWorkspaceId.delete(event.workspaceId); - return true; - } + if (concreteContinuationInstalled) { + return true; + } + if (reservedContinuation == null) { + return true; + } - if (await this.isStreamEndBeforeWorkspaceTurnPrompt(record, event)) { - log.debug("Ignoring stale uncorrelated stream-end before queued workspace turn prompt", { + log.debug("Waiting for a reserved workspace turn continuation handoff", { workspaceId: event.workspaceId, taskHandleId: record.handleId, streamEndMessageId: event.messageId, }); - return true; + if (await reservedContinuation.released) { + return true; + } + // The handoff failed before it installed a concrete continuation. Re-evaluate the + // original stream-end while the caller still owns the workspace event lock. } - - const error = "Workspace turn superseded by an uncorrelated workspace stream-end"; - const next: WorkspaceTurnTaskHandleRecord = { - ...record, - status: "interrupted", - updatedAt: getIsoNow(), - messageId: event.messageId, - error, - }; - await this.settleWorkspaceTurn({ - record, - next, - waiterSettlement: { status: "error", error: new Error(error) }, - }); - return true; } /** * Whether a continuation of this exact delegated turn is pending or streaming. * Pending entries must carry the same correlation metadata as the ended stream. */ - private hasSameTurnContinuation( + private async hasSameTurnContinuation( + event: StreamEndEvent, + correlation: { taskHandleId: string; ownerWorkspaceId: string; turnId: string } + ): Promise { + while (true) { + if (this.hasConcreteSameTurnContinuation(event, correlation)) { + return true; + } + const reservation = this.getReservedWorkspaceTurnContinuation(event.workspaceId, correlation); + if (reservation == null) { + return false; + } + + log.debug("Waiting for a reserved correlated workspace turn continuation handoff", { + workspaceId: event.workspaceId, + taskHandleId: correlation.taskHandleId, + streamEndMessageId: event.messageId, + }); + if (await reservation.released) { + return true; + } + // A failed handoff does not prove that no other concurrent reservation can continue the turn. + } + } + + private hasConcreteSameTurnContinuation( event: StreamEndEvent, correlation: { taskHandleId: string; ownerWorkspaceId: string; turnId: string } ): boolean { @@ -10599,60 +10885,80 @@ export class TaskService { } return await this.interruptWorkspaceTurnFromUncorrelatedStreamEnd(event); } - const record = await this.taskHandleStore.getWorkspaceTurn( - metadata.ownerWorkspaceId, - metadata.taskHandleId - ); - if (record == null) { - log.warn("Ignoring missing workspace turn stream-end handle", { - workspaceId: event.workspaceId, - taskHandleId: metadata.taskHandleId, + while (true) { + const record = await this.taskHandleStore.getWorkspaceTurn( + metadata.ownerWorkspaceId, + metadata.taskHandleId + ); + if (record == null) { + log.warn("Ignoring missing workspace turn stream-end handle", { + workspaceId: event.workspaceId, + taskHandleId: metadata.taskHandleId, + }); + return true; + } + if (record.workspaceId !== event.workspaceId || record.turnId !== metadata.turnId) { + log.warn("Ignoring out-of-scope workspace turn stream-end", { + workspaceId: event.workspaceId, + taskHandleId: metadata.taskHandleId, + }); + return true; + } + if (this.isDeferredWorkspaceTurnMessage(record, event.messageId)) { + return true; + } + + const isToolBoundary = event.metadata.finishReason === "tool-calls"; + // A queued continuation can stop the in-flight stream at a tool boundary and + // continue the same delegated turn. Wait for any in-progress handoff before + // deciding whether this stream-end is terminal. + if (isToolBoundary && (await this.hasSameTurnContinuation(event, metadata))) { + await this.markWorkspaceTurnStreamEndDeferred(event); + return true; + } + + const next = this.buildTerminalWorkspaceTurnRecordFromEvent(record, event, { + queueCutSupersedeEvidence: this.hasLiveQueueCutSupersedeEvidence(event), }); - return true; - } - if (record.workspaceId !== event.workspaceId || record.turnId !== metadata.turnId) { - log.warn("Ignoring out-of-scope workspace turn stream-end", { - workspaceId: event.workspaceId, - taskHandleId: metadata.taskHandleId, + let concreteContinuationInstalled = false; + let reservedContinuation: WorkspaceTurnContinuationReservation | undefined; + await this.settleWorkspaceTurn({ + record, + next, + waiterSettlement: + 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, + shouldSettleCurrent: isToolBoundary + ? (current) => { + const currentCorrelation = this.buildWorkspaceTurnMuxMetadata(current); + if (this.hasConcreteSameTurnContinuation(event, currentCorrelation)) { + concreteContinuationInstalled = true; + return false; + } + reservedContinuation = this.getReservedWorkspaceTurnContinuation( + event.workspaceId, + currentCorrelation + ); + return reservedContinuation == null; + } + : undefined, }); - return true; - } - if (this.isDeferredWorkspaceTurnMessage(record, event.messageId)) { - return true; - } - - // A queued continuation can stop the in-flight stream at a tool boundary with - // finishReason "tool-calls" and continue the same delegated turn. Report - // wake-ups carry the exact correlation explicitly; bash-monitor wakes inherit - // it from history. Defer settlement until the continuation's terminal - // stream-end instead of reporting a false completion failure to the owner. - // Any other queued input (manual message, /compact) supersedes the turn and - // must settle the old outcome here. - if ( - event.metadata.finishReason === "tool-calls" && - this.hasSameTurnContinuation(event, metadata) - ) { - await this.markWorkspaceTurnStreamEndDeferred(event); - return true; + if (!isToolBoundary || (reservedContinuation == null && !concreteContinuationInstalled)) { + return true; + } + if (concreteContinuationInstalled || (await reservedContinuation?.released) === true) { + await this.markWorkspaceTurnStreamEndDeferred(event); + return true; + } + // A reservation appeared after the first check and then failed. Re-evaluate the + // original correlated stream-end while the caller still owns the workspace event lock. } - - const next = this.buildTerminalWorkspaceTurnRecordFromEvent(record, event, { - queueCutSupersedeEvidence: this.hasLiveQueueCutSupersedeEvidence(event), - }); - await this.settleWorkspaceTurn({ - record, - next, - waiterSettlement: - 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; } private async handleStreamEnd(event: StreamEndEvent): Promise { @@ -11156,6 +11462,69 @@ export class TaskService { }); } + private getReservedWorkspaceTurnContinuation( + workspaceId: string, + correlation: Pick + ): WorkspaceTurnContinuationReservation | undefined { + const reservations = this.workspaceTurnContinuationReservationsByWorkspaceId.get(workspaceId); + return Array.from(reservations ?? []).find((reservation) => { + return ( + reservation.muxMetadata.taskHandleId === correlation.taskHandleId && + reservation.muxMetadata.ownerWorkspaceId === correlation.ownerWorkspaceId && + reservation.muxMetadata.turnId === correlation.turnId + ); + }); + } + + private async reserveActiveWorkspaceTurnContinuation( + workspaceId: string + ): Promise { + const candidate = await this.getActiveWorkspaceTurnRecordForWorkspace(workspaceId); + if (candidate == null) { + return undefined; + } + + return await this.workspaceTurnSettlementLocks.withLock(candidate.handleId, async () => { + const current = await this.taskHandleStore.getWorkspaceTurn( + candidate.ownerWorkspaceId, + candidate.handleId + ); + if ( + current?.workspaceId !== workspaceId || + !isActiveWorkspaceTurnTaskStatus(current.status) + ) { + return undefined; + } + + const reservations = + this.workspaceTurnContinuationReservationsByWorkspaceId.get(workspaceId) ?? new Set(); + const releaseState = Promise.withResolvers(); + let released = false; + let continuationInstalled = false; + const reservation: WorkspaceTurnContinuationReservation = { + muxMetadata: this.buildWorkspaceTurnMuxMetadata(current), + released: releaseState.promise, + markContinuationInstalled: () => { + continuationInstalled = true; + }, + [Symbol.dispose]: () => { + if (released) { + return; + } + released = true; + reservations.delete(reservation); + if (reservations.size === 0) { + this.workspaceTurnContinuationReservationsByWorkspaceId.delete(workspaceId); + } + releaseState.resolve(continuationInstalled); + }, + }; + reservations.add(reservation); + this.workspaceTurnContinuationReservationsByWorkspaceId.set(workspaceId, reservations); + return reservation; + }); + } + // A queued report can defer the preceding stream-end. If dispatch then fails, settle that // exact turn here because no replacement stream-end can arrive. private async settleWorkspaceTurnContinuationFailure( @@ -11394,6 +11763,7 @@ export class TaskService { "failAgentTaskTerminally: errorMessage must be non-empty" ); + const executionVersion = coerceNonEmptyString(entry.workspace.taskExecutionId); let transitionedToInterrupted = false; let parentWorkspaceId = entry.workspace.parentWorkspaceId; await this.editWorkspaceEntry( @@ -11439,6 +11809,7 @@ export class TaskService { workflowOwnedAncestorWorkspaceIds, errorType: failure.errorType, errorMessage: failure.errorMessage, + ...(executionVersion != null ? { executionVersion } : {}), model: entry.workspace.taskModelString, nowMs: persistedAtMs, }); @@ -11495,6 +11866,7 @@ export class TaskService { if (!parentWorkspaceId) { return; } + const executionVersion = coerceNonEmptyString(childEntry.workspace.taskExecutionId); // An active waiter (foreground task tool call or task_await) already // surfaced the rejection to the parent's in-flight turn. if (hadForegroundWaiters) { @@ -11522,6 +11894,7 @@ export class TaskService { formatSubagentFailureUserMessage({ childWorkspaceId, agentType: coerceNonEmptyString(childEntry.workspace.agentType) ?? "agent", + ...(executionVersion != null ? { executionVersion } : {}), errorType: failure.errorType, errorMessage: failure.errorMessage, }), @@ -11559,10 +11932,9 @@ export class TaskService { // The failure message is already injected above. Enqueue even when other children are active: // the drain defers on blocking work, and the later settling child may have a foreground waiter // that suppresses its own terminal wake-up. - const generationId = await this.getAgentTerminalAttentionGenerationId( - parentWorkspaceId, - childWorkspaceId - ); + const generationId = + executionVersion ?? + (await this.getAgentTerminalAttentionGenerationId(parentWorkspaceId, childWorkspaceId)); await this.enqueueTerminalAttention({ ownerWorkspaceId: parentWorkspaceId, sourceKind: "agent_task", @@ -12131,6 +12503,7 @@ export class TaskService { ); if (finalization.kind === "finalized") { for (const taskId of finalization.taskIds) { + this.removeQueuedAgentProgressAfterTerminalDelivery(params.parentWorkspaceId, taskId); cleanupTaskIds.add(taskId); } return; @@ -12425,6 +12798,7 @@ export class TaskService { return { finalized: true }; } + const executionVersion = coerceNonEmptyString(latestChildEntry?.workspace.taskExecutionId); const reportTitle = coerceNonEmptyString(reportArgs.title); this.timelineRecorder.record(parentWorkspaceId, { kind: "task.reported", @@ -12469,6 +12843,7 @@ export class TaskService { ancestorWorkspaceIds, workflowOwnedAncestorWorkspaceIds, reportMarkdown: reportArgs.reportMarkdown, + ...(executionVersion != null ? { executionVersion } : {}), model: latestChildEntry?.workspace.taskModelString, thinkingLevel: latestChildEntry?.workspace.taskThinkingLevel, title: reportArgs.title, @@ -12503,25 +12878,10 @@ export class TaskService { await this.maybeStartPatchGenerationForReportedTask(childWorkspaceId); - const queuedProgressRemoval = this.workspaceService.removeQueuedMessagesByDedupeKeyPrefix( - parentWorkspaceId, - `agent-report:${childWorkspaceId}:`, - { cancelReason: "Incremental sub-agent update superseded by the terminal report." } - ); - if (!queuedProgressRemoval.success) { - log.warn("Failed to remove queued incremental sub-agent reports", { - parentWorkspaceId, - childWorkspaceId, - error: queuedProgressRemoval.error, - }); - } - - await this.deliverReportToParent( - parentWorkspaceId, - childWorkspaceId, - latestChildEntry, - reportArgs - ); + await this.deliverReportToParent(parentWorkspaceId, childWorkspaceId, latestChildEntry, { + ...reportArgs, + ...(executionVersion != null ? { executionVersion } : {}), + }); // Resolve foreground waiters. const hadForegroundWaiters = this.resolveWaiters(childWorkspaceId, { @@ -12573,13 +12933,12 @@ export class TaskService { return { finalized: true }; } - // The report is already injected into parent history above (deliverReportToParent). Enqueue the - // notification even when other children are still active: the drain defers on blocking work and - // a later foreground-awaited sibling may suppress its own wake-up. - const generationId = await this.getAgentTerminalAttentionGenerationId( - parentWorkspaceId, - childWorkspaceId - ); + // The report is already delivered to a foreground waiter, queued as a workspace-turn + // continuation, or appended to parent history. Enqueue the notification even when other + // children are active. The drain defers on blocking work. + const generationId = + executionVersion ?? + (await this.getAgentTerminalAttentionGenerationId(parentWorkspaceId, childWorkspaceId)); await this.enqueueTerminalAttention({ ownerWorkspaceId: parentWorkspaceId, sourceKind: "agent_task", @@ -13074,6 +13433,66 @@ export class TaskService { return { groupId: bestOf.groupId, index: bestOf.index, total: bestOf.total }; } + private async reserveAgentTerminalAttention( + parentWorkspaceId: string, + childWorkspaceId: string, + executionVersion?: string + ): Promise<{ id: string; created: boolean; claimed: boolean }> { + const generationId = + executionVersion ?? + (await this.getAgentTerminalAttentionGenerationId(parentWorkspaceId, childWorkspaceId)); + const id = TerminalAttentionStore.notificationId("agent_task", childWorkspaceId, generationId); + return this.terminalAttentionLocks.withLock(parentWorkspaceId, async () => { + if (!this.claimTerminalAttention(parentWorkspaceId, id)) { + return { id, created: false, claimed: false }; + } + try { + const existing = await this.terminalAttentionStore.get(parentWorkspaceId, id); + if (existing != null && existing.status !== "pending") { + this.releaseTerminalAttentionClaim(parentWorkspaceId, id); + return { id, created: false, claimed: false }; + } + const created = + existing == null + ? await this.terminalAttentionStore.enqueueIfAbsent({ + ownerWorkspaceId: parentWorkspaceId, + sourceKind: "agent_task", + terminalOutcome: "completed", + sourceId: childWorkspaceId, + ...(generationId != null ? { generationId } : {}), + }) + : null; + return { id, created: created != null, claimed: true }; + } catch (error) { + this.releaseTerminalAttentionClaim(parentWorkspaceId, id); + throw error; + } + }); + } + + private removeQueuedAgentProgressAfterTerminalDelivery( + parentWorkspaceId: string, + childWorkspaceId: string + ): void { + const removal = this.workspaceService.removeQueuedMessagesByDedupeKeyPrefix( + parentWorkspaceId, + `agent-report:${childWorkspaceId}:`, + { + cancelReason: "Incremental sub-agent update superseded by the terminal report.", + // The terminal report now owns the continuation. This cleanup must not + // cancel the outer workspace turn. + notifyCancellation: false, + } + ); + if (!removal.success) { + log.warn("Failed to remove queued incremental sub-agent reports", { + parentWorkspaceId, + childWorkspaceId, + error: removal.error, + }); + } + } + private async deliverReportToParent( parentWorkspaceId: string, childWorkspaceId: string, @@ -13083,6 +13502,7 @@ export class TaskService { title?: string; structuredOutput?: unknown; planFilePath?: string; + executionVersion?: string; } ): Promise { assert( @@ -13127,6 +13547,7 @@ export class TaskService { title?: string; structuredOutput?: unknown; planFilePath?: string; + executionVersion?: string; } ): Promise { const agentType = coerceNonEmptyString(childEntry?.workspace.agentType) ?? "agent"; @@ -13168,6 +13589,9 @@ export class TaskService { childEntry ); if (finalization.kind === "finalized") { + for (const finalizedTaskId of finalization.taskIds) { + this.removeQueuedAgentProgressAfterTerminalDelivery(parentWorkspaceId, finalizedTaskId); + } return finalization.taskIds.filter((taskId) => taskId !== childWorkspaceId); } @@ -13202,6 +13626,7 @@ export class TaskService { if (childWorkspaceId) { const waiters = this.pendingWaitersByTaskId.get(childWorkspaceId); if (waiters && waiters.length > 0) { + this.removeQueuedAgentProgressAfterTerminalDelivery(parentWorkspaceId, childWorkspaceId); return []; } } @@ -13218,6 +13643,7 @@ export class TaskService { title: titlePrefix, reportMarkdown: report.reportMarkdown, status: "completed", + ...(report.executionVersion != null ? { executionVersion: report.executionVersion } : {}), ...(childModelString != null ? { model: childModelString } : {}), ...(childThinkingLevel != null ? { thinkingLevel: childThinkingLevel } : {}), ...(report.structuredOutput !== undefined @@ -13225,8 +13651,135 @@ export class TaskService { : {}), }); - const workspaceTurnMuxMetadata = - await this.getActiveWorkspaceTurnMuxMetadataForWorkspace(parentWorkspaceId); + using workspaceTurnContinuation = + await this.reserveActiveWorkspaceTurnContinuation(parentWorkspaceId); + const workspaceTurnMuxMetadata = workspaceTurnContinuation?.muxMetadata; + if (workspaceTurnContinuation != null) { + const activeWorkspaceTurnMuxMetadata = workspaceTurnContinuation.muxMetadata; + const parentEntry = findWorkspaceEntry(this.config.loadConfigOrDefault(), parentWorkspaceId); + if (parentEntry != null) { + const resumeOptions = await this.resolveParentAutoResumeOptions( + parentWorkspaceId, + parentEntry, + defaultModel + ); + const terminalAttention = await this.reserveAgentTerminalAttention( + parentWorkspaceId, + childWorkspaceId, + report.executionVersion + ); + if (!terminalAttention.claimed) { + // Another path owns this durable report, or this generation already completed delivery. + return []; + } + + let terminalContinuationAccepted = false; + const sendResult = await this.workspaceService.sendMessage( + parentWorkspaceId, + reportContent, + { + model: resumeOptions.model, + agentId: resumeOptions.agentId, + thinkingLevel: resumeOptions.thinkingLevel, + reasoningMode: resumeOptions.reasoningMode, + muxMetadata: activeWorkspaceTurnMuxMetadata, + }, + { + skipAutoResumeReset: true, + synthetic: true, + agentInitiated: true, + startStreamInBackground: true, + workspaceTurnContinuation: true, + // A persistent child can report multiple execution generations. Keep each durable + // attention record independent while an earlier generation is still queued. + queueDedupeKey: `agent-terminal-report:${terminalAttention.id}`, + onAccepted: async () => { + // Mark this wake consumed before the accepted continuation can end. + // Its stream-end must not race a later terminal-attention drain. + await this.terminalAttentionStore.markDelivered( + parentWorkspaceId, + terminalAttention.id + ); + workspaceTurnContinuation.markContinuationInstalled(); + terminalContinuationAccepted = true; + this.releaseTerminalAttentionClaim(parentWorkspaceId, terminalAttention.id); + }, + onCanceled: async (reason: string) => { + await this.settleWorkspaceTurnContinuationFailure( + parentWorkspaceId, + activeWorkspaceTurnMuxMetadata, + "interrupted", + reason + ); + }, + onDeliveryCanceled: async () => { + try { + await this.terminalAttentionStore.markSuperseded( + parentWorkspaceId, + terminalAttention.id + ); + } finally { + this.releaseTerminalAttentionClaim(parentWorkspaceId, terminalAttention.id); + } + }, + onAcceptedPreStreamFailure: async (error: SendMessageError) => { + await this.settleWorkspaceTurnContinuationFailure( + parentWorkspaceId, + activeWorkspaceTurnMuxMetadata, + "error", + formatSendMessageError(error).message + ); + }, + onDeliveryAcceptedPreStreamFailure: async () => { + const retryClaimed = + !terminalContinuationAccepted || + (await this.terminalAttentionLocks.withLock(parentWorkspaceId, () => + Promise.resolve( + this.claimTerminalAttention(parentWorkspaceId, terminalAttention.id) + ) + )); + if (!retryClaimed) { + return; + } + try { + // Acceptance consumes the reservation. Re-arm it when startup fails so the + // durable report remains eligible for the normal attention retry path. + await this.terminalAttentionStore.markPending( + parentWorkspaceId, + terminalAttention.id + ); + } finally { + this.releaseTerminalAttentionClaim(parentWorkspaceId, terminalAttention.id); + this.scheduleTerminalAttentionDrainAfterIdle(parentWorkspaceId); + } + }, + } + ); + if (sendResult.success) { + workspaceTurnContinuation.markContinuationInstalled(); + // Install the terminal continuation before removing progress. The + // workspace turn always has a concrete future driver during handoff. + this.removeQueuedAgentProgressAfterTerminalDelivery(parentWorkspaceId, childWorkspaceId); + return []; + } + if (terminalContinuationAccepted) { + workspaceTurnContinuation.markContinuationInstalled(); + return []; + } + this.releaseTerminalAttentionClaim(parentWorkspaceId, terminalAttention.id); + if (terminalAttention.created) { + await this.terminalAttentionStore.delete(parentWorkspaceId, terminalAttention.id); + } else { + this.scheduleTerminalAttentionDrain(parentWorkspaceId); + } + log.warn("Failed to queue terminal sub-agent report continuation", { + parentWorkspaceId, + childWorkspaceId, + error: formatSendMessageError(sendResult.error).message, + }); + } + } + const messageId = createTaskReportMessageId(); const reportMessage = createMuxMessage(messageId, "user", reportContent, { timestamp: Date.now(), @@ -13244,6 +13797,9 @@ export class TaskService { ...reportMessage, type: "message", }); + if (workspaceTurnMuxMetadata == null) { + this.removeQueuedAgentProgressAfterTerminalDelivery(parentWorkspaceId, childWorkspaceId); + } } if (!appendResult.success) { log.error("Failed to append synthetic subagent report to parent history", { diff --git a/src/node/services/utils/messageIds.ts b/src/node/services/utils/messageIds.ts index 51f2206b01..f1c9a2c392 100644 --- a/src/node/services/utils/messageIds.ts +++ b/src/node/services/utils/messageIds.ts @@ -45,6 +45,8 @@ export const createTaskReportMessageId = (): string => export const createTaskFailureMessageId = (): string => `task-failure-${Date.now()}-${randomSuffix(9)}`; +export const FILE_CHANGE_NOTIFICATION_MESSAGE_ID_PREFIX = "file-change-"; + /** External file-change notification message IDs: file-change-{timestamp}-{random} */ export const createFileChangeNotificationMessageId = (): string => - `file-change-${Date.now()}-${randomSuffix(9)}`; + `${FILE_CHANGE_NOTIFICATION_MESSAGE_ID_PREFIX}${Date.now()}-${randomSuffix(9)}`; diff --git a/src/node/services/workspaceService.test.ts b/src/node/services/workspaceService.test.ts index 3e422a4ff1..11cd8a03cd 100644 --- a/src/node/services/workspaceService.test.ts +++ b/src/node/services/workspaceService.test.ts @@ -5688,6 +5688,8 @@ describe("WorkspaceService sendMessage status clearing", () => { fakeSession.hasQueuedOrDispatchingEntry.mockReturnValue(true); const onCanceled = mock(() => undefined); const onAcceptedPreStreamFailure = mock(() => undefined); + const onDeliveryCanceled = mock(() => undefined); + const onDeliveryAcceptedPreStreamFailure = mock(() => undefined); const muxMetadata = { type: "workspace-turn-task" as const, taskHandleId: "wst_stale_progress", @@ -5705,6 +5707,8 @@ describe("WorkspaceService sendMessage status clearing", () => { workspaceTurnContinuation: true, onCanceled, onAcceptedPreStreamFailure, + onDeliveryCanceled, + onDeliveryAcceptedPreStreamFailure, } ); @@ -5715,6 +5719,8 @@ describe("WorkspaceService sendMessage status clearing", () => { expect.objectContaining({ onCanceled: undefined, onAcceptedPreStreamFailure: undefined, + onDeliveryCanceled, + onDeliveryAcceptedPreStreamFailure, }) ); }); diff --git a/src/node/services/workspaceService.ts b/src/node/services/workspaceService.ts index 6bb5215347..78fad0a72e 100644 --- a/src/node/services/workspaceService.ts +++ b/src/node/services/workspaceService.ts @@ -650,6 +650,25 @@ interface QueuedBashMonitorWakeCancellation { matchedOutputByProcess: Map; } +function combineSendCallbacks( + correlationCallback: ((value: T) => Promise | void) | undefined, + deliveryCallback: ((value: T) => Promise | void) | undefined +): ((value: T) => Promise | void) | undefined { + if (correlationCallback == null) { + return deliveryCallback; + } + if (deliveryCallback == null) { + return correlationCallback; + } + return async (value: T) => { + try { + await correlationCallback?.(value); + } finally { + await deliveryCallback?.(value); + } + }; +} + async function waitForAgentSessionIdle(session: AgentSession, signal?: AbortSignal): Promise { assert(session instanceof AgentSession, "waitForAgentSessionIdle requires an AgentSession"); try { @@ -8564,8 +8583,14 @@ export class WorkspaceService extends EventEmitter { /** Force Copilot billing classification to "agent" for internal sends. */ agentInitiated?: boolean; onAccepted?: () => Promise | void; + /** Workspace-turn correlation callback. It is removed when an earlier turn supersedes it. */ onCanceled?: (reason: string) => Promise | void; + /** Workspace-turn correlation callback. It is removed when an earlier turn supersedes it. */ onAcceptedPreStreamFailure?: (error: SendMessageError) => Promise | void; + /** Delivery callback. It runs even when workspace-turn correlation is removed. */ + onDeliveryCanceled?: (reason: string) => Promise | void; + /** Delivery callback. It runs even when workspace-turn correlation is removed. */ + onDeliveryAcceptedPreStreamFailure?: (error: SendMessageError) => Promise | void; cancelState?: { canceledBeforeAcceptance: boolean }; /** Cancels a synthetic send even after it has left MessageQueue for PREPARING. */ cancelSignal?: AbortSignal; @@ -8701,6 +8726,8 @@ export class WorkspaceService extends EventEmitter { onAcceptedPreStreamFailure: preserveCorrelation ? internal?.onAcceptedPreStreamFailure : undefined, + onDeliveryCanceled: internal?.onDeliveryCanceled, + onDeliveryAcceptedPreStreamFailure: internal?.onDeliveryAcceptedPreStreamFailure, }; }; @@ -8715,15 +8742,22 @@ export class WorkspaceService extends EventEmitter { ); if (!pricingGate.success) { if (internal?.synthetic !== true) { - return session.sendMessage(message, normalizedOptions, { + const continuationSendState = getContinuationSendState(); + return session.sendMessage(message, continuationSendState.options, { synthetic: internal?.synthetic, agentInitiated: internal?.agentInitiated, goalKind: internal?.goalKind, cancelState: internal?.cancelState, cancelSignal: internal?.cancelSignal, - onCanceled: internal?.onCanceled, + onCanceled: combineSendCallbacks( + continuationSendState.onCanceled, + continuationSendState.onDeliveryCanceled + ), onAccepted: internal?.onAccepted, - onAcceptedPreStreamFailure: internal?.onAcceptedPreStreamFailure, + onAcceptedPreStreamFailure: combineSendCallbacks( + continuationSendState.onAcceptedPreStreamFailure, + continuationSendState.onDeliveryAcceptedPreStreamFailure + ), startStreamInBackground: internal?.startStreamInBackground, goalContinuation: internal?.goalContinuation, }); @@ -8818,6 +8852,9 @@ export class WorkspaceService extends EventEmitter { cancelState: internal?.cancelState, cancelSignal: internal?.cancelSignal, onCanceled: continuationSendState.onCanceled, + onDeliveryCanceled: continuationSendState.onDeliveryCanceled, + onDeliveryAcceptedPreStreamFailure: + continuationSendState.onDeliveryAcceptedPreStreamFailure, onAccepted: internal?.onAccepted, onAcceptedPreStreamFailure: continuationSendState.onAcceptedPreStreamFailure, } @@ -8861,6 +8898,10 @@ export class WorkspaceService extends EventEmitter { } const continuationSendState = getContinuationSendState(); + const continuationAcceptedPreStreamFailure = combineSendCallbacks( + continuationSendState.onAcceptedPreStreamFailure, + continuationSendState.onDeliveryAcceptedPreStreamFailure + ); const onAcceptedPreStreamFailure = async (error: SendMessageError) => { if (resumedInterruptedTask && normalizedOptions?.editMessageId) { try { @@ -8876,7 +8917,7 @@ export class WorkspaceService extends EventEmitter { ); } } - await continuationSendState.onAcceptedPreStreamFailure?.(error); + await continuationAcceptedPreStreamFailure?.(error); }; const shouldRunPendingAutoTitle = @@ -8897,7 +8938,10 @@ export class WorkspaceService extends EventEmitter { startStreamInBackground: internal?.startStreamInBackground, cancelState: internal?.cancelState, cancelSignal: internal?.cancelSignal, - onCanceled: continuationSendState.onCanceled, + onCanceled: combineSendCallbacks( + continuationSendState.onCanceled, + continuationSendState.onDeliveryCanceled + ), onAccepted: internal?.onAccepted, onAcceptedPreStreamFailure, }); @@ -9486,7 +9530,7 @@ export class WorkspaceService extends EventEmitter { removeQueuedMessagesByDedupeKeyPrefix( workspaceId: string, prefix: string, - options?: { cancelReason?: string } + options?: { cancelReason?: string; notifyCancellation?: boolean } ): Result { try { const session = this.sessions.get(workspaceId.trim()); @@ -9496,7 +9540,8 @@ export class WorkspaceService extends EventEmitter { return Ok( session.removeQueuedMessagesByDedupeKeyPrefix( prefix, - options?.cancelReason ?? "Queued message superseded before dispatch." + options?.cancelReason ?? "Queued message superseded before dispatch.", + { notifyCancellation: options?.notifyCancellation } ) ); } catch (error) {