From 6194fcc8d31d74a0aed29cdf05c2bcc936845267 Mon Sep 17 00:00:00 2001 From: wen2zhou Date: Fri, 14 Aug 2026 12:02:29 +0000 Subject: [PATCH] fix(delegation): settle stopped subagents reliably --- scripts/ci/module-impact.json | 3 +- src/main/delegation/acp-execution.test.ts | 226 ++++++- src/main/delegation/acp-execution.ts | 74 ++- .../attempt-runtime-transcript.test.ts | 153 ++++- .../delegation/attempt-runtime-transcript.ts | 139 +++- src/main/delegation/delegated-cleanup.ts | 124 ++++ .../delegation/delegated-turn-lifecycle.ts | 102 +-- .../delegated-user-question-owner.ts | 106 +-- .../delegation/delegated-work-record-types.ts | 11 +- .../durable-delegated-work-contract.ts | 4 + .../delegation/durable-delegated-work.test.ts | 415 +++++++++++- src/main/delegation/durable-delegated-work.ts | 180 ++--- .../in-memory-delegated-work-records.ts | 41 +- .../delegation/production-composition.test.ts | 20 +- src/main/delegation/production-composition.ts | 5 + .../delegation/session-record-adapter.test.ts | 25 +- src/main/delegation/session-record-adapter.ts | 19 +- src/main/delegation/session-records.ts | 8 +- .../local-rpc-server.delegated-work.test.ts | 87 +++ src/main/session-persistence/coordinator.ts | 5 +- .../delegated-work-owner.ts | 200 ++++-- .../delegated-work-records.test.ts | 162 ++++- ...WorkspaceAgentRuntime.architecture.test.ts | 12 +- ...workspace-subagent-runtime-presentation.ts | 132 ++-- .../workspace-subagent-runtime-transcript.ts | 329 ++++++++++ .../src/pages/home/HomePage.render.test.tsx | 45 +- .../ConversationPanel.interaction.test.tsx | 62 +- .../SubagentReleaseSurfaces.render.test.tsx | 620 +++++++++++++++++- .../workspace/SubagentReleaseSurfaces.tsx | 17 +- .../subagent-release-projection.test.ts | 188 ++++++ .../workspace/subagent-release-projection.ts | 76 ++- 31 files changed, 3159 insertions(+), 431 deletions(-) create mode 100644 src/main/delegation/delegated-cleanup.ts create mode 100644 src/renderer/src/lib/acp/workspace-subagent-runtime-transcript.ts diff --git a/scripts/ci/module-impact.json b/scripts/ci/module-impact.json index f4568da49..41e263309 100644 --- a/scripts/ci/module-impact.json +++ b/scripts/ci/module-impact.json @@ -36,7 +36,8 @@ "src/renderer/src/lib/acp/workspace-runtime-command-owner.ts", "src/renderer/src/lib/acp/workspace-runtime-session-lifecycle-owner.ts", "src/renderer/src/lib/acp/workspace-runtime-save-as-skill-owner.ts", - "src/renderer/src/lib/acp/workspace-subagent-runtime-presentation.ts" + "src/renderer/src/lib/acp/workspace-subagent-runtime-presentation.ts", + "src/renderer/src/lib/acp/workspace-subagent-runtime-transcript.ts" ], "interfacePaths": ["src/renderer/src/lib/acp/useWorkspaceAgentRuntime.ts"], "consumerModules": ["workspace_page"], diff --git a/src/main/delegation/acp-execution.test.ts b/src/main/delegation/acp-execution.test.ts index 95d27f916..6eaa062b7 100644 --- a/src/main/delegation/acp-execution.test.ts +++ b/src/main/delegation/acp-execution.test.ts @@ -13,7 +13,7 @@ import { type DelegatedWorkCertificationDriver } from './certification-contract.test' import { delegateExecutionContract } from './execution-contract.test' -import type { DelegateExecutionInput } from './execution-port' +import type { DelegateExecutionEvent, DelegateExecutionInput } from './execution-port' type Deferred = Readonly<{ promise: Promise @@ -631,6 +631,230 @@ describe('ACP delegate execution production adapter', () => { await expect(execution.reserve(1)).resolves.toHaveProperty('slotIds') }) + it('settles cancellation when the terminated provider prompt never returns', async () => { + const prompt = deferred() + const cleanup: string[] = [] + let callbacks!: AcpDelegateExecutionCallbacks + const execution = createAcpDelegateExecution({ + capacity: 1, + prepare: async (input) => ({ + executionId: input.attemptId, + provenance: { + projectId: input.session.projectId, + sessionId: input.session.sessionId, + agentFrameId: input.frameId, + runtimeSegmentId: input.runtimeSegmentId + }, + workspace: { cwd: '/workspace/terminated-provider' }, + runtimeHome: '/runtime/terminated-provider', + frameworkId: 'certified-test', + capability: { + revoke: async () => { + cleanup.push('revoke') + } + }, + disposeResources: async () => { + cleanup.push('resources') + } + }), + assertFrameworkNativeDelegationDisabled: async () => undefined, + createRuntime: (_scope, runtimeCallbacks) => { + callbacks = runtimeCallbacks + return { + createSession: async () => ({ sessionId: 'provider-terminated' }), + sendAppContinuation: () => { + callbacks.onProviderPromptAccepted('provider-terminated') + return prompt.promise + }, + cancelPrompt: async () => undefined, + respondToPermission: async () => undefined, + setPermissionProfile: async () => undefined, + deleteSession: async () => { + cleanup.push('delete') + }, + shutdownForQuit: async () => { + cleanup.push('shutdown') + return { reaped: true } + } + } + } + }) + const reservation = await execution.reserve(1) + const running = execution.run(makeInput('terminated-provider'), reservation.slotIds[0]) + await running.accepted + + await expect(running.cancel()).resolves.toBeUndefined() + await expect(running.completion).resolves.toEqual({ status: 'cancelled' }) + await vi.waitFor(() => expect(cleanup).toEqual(['revoke', 'delete', 'shutdown', 'resources'])) + await expect(execution.reserve(1)).resolves.toHaveProperty('slotIds') + }) + + it('keeps same-Attempt observations open after cancellation until transport shutdown', async () => { + const prompt = deferred() + const shutdown = deferred<{ reaped: boolean }>() + let callbacks!: AcpDelegateExecutionCallbacks + const execution = createAcpDelegateExecution({ + capacity: 1, + prepare: async (input) => ({ + executionId: input.attemptId, + provenance: { + projectId: input.session.projectId, + sessionId: input.session.sessionId, + agentFrameId: input.frameId, + runtimeSegmentId: input.runtimeSegmentId, + promptMessageId: `prompt-${input.attemptId}` + }, + workspace: { cwd: '/workspace/late-observation' }, + runtimeHome: '/runtime/late-observation', + frameworkId: 'certified-test', + capability: { revoke: async () => undefined } + }), + assertFrameworkNativeDelegationDisabled: async () => undefined, + createRuntime: (_scope, runtimeCallbacks) => { + callbacks = runtimeCallbacks + return { + createSession: async () => ({ sessionId: 'provider-late-observation' }), + sendAppContinuation: () => { + callbacks.onProviderPromptAccepted('provider-late-observation') + return prompt.promise + }, + cancelPrompt: async () => undefined, + respondToPermission: async () => undefined, + setPermissionProfile: async () => undefined, + deleteSession: async () => undefined, + shutdownForQuit: () => shutdown.promise + } + } + }) + const reservation = await execution.reserve(1) + const running = execution.run(makeInput('late-observation'), reservation.slotIds[0]) + const events: DelegateExecutionEvent[] = [] + running.subscribe((event) => events.push(event)) + await running.accepted + + const cancelling = running.cancel() + await expect(running.completion).resolves.toEqual({ status: 'cancelled' }) + callbacks.onEvent({ + sessionId: 'provider-late-observation', + id: 'late-message', + timestamp: 20, + kind: 'message', + level: 'info', + role: 'assistant', + text: 'Preserve this tail.' + }) + callbacks.onEvent({ + sessionId: 'provider-late-observation', + id: 'late-message', + timestamp: 20, + kind: 'message', + level: 'info', + role: 'assistant', + text: 'Preserve this tail.' + }) + + expect(events).toEqual([ + expect.objectContaining({ + kind: 'runtime', + update: expect.objectContaining({ + event: expect.objectContaining({ id: 'late-message', text: 'Preserve this tail.' }) + }) + }) + ]) + shutdown.resolve({ reaped: true }) + await cancelling + }) + + it('does not start a provider prompt when cancellation lands during deferred Turn begin', async () => { + const { execution, controls } = makeHarness(1) + const begin = deferred() + const reservation = await execution.reserve(1) + const running = execution.run( + { + ...makeInput('deferred-begin'), + turn: { + promptMessageId: 'prompt-deferred-begin', + messageBranchId: 'branch-deferred-begin', + runtimeSegmentId: 'segment-deferred-begin', + begin: () => begin.promise + } + }, + reservation.slotIds[0] + ) + await vi.waitFor(() => expect(controls.has('deferred-begin')).toBe(true)) + + await running.cancel() + begin.resolve() + + await expect(running.accepted).rejects.toMatchObject({ + name: 'DelegateMessagePreAcceptanceError' + }) + await expect(running.completion).resolves.toEqual({ status: 'cancelled' }) + await vi.waitFor(() => expect(controls.get('deferred-begin')?.prompts).toEqual([])) + }) + + it('reports cleanup failure and retries only the unfinished cleanup step', async () => { + let revokeAttempts = 0 + const cleanup: string[] = [] + let callbacks!: AcpDelegateExecutionCallbacks + const execution = createAcpDelegateExecution({ + capacity: 1, + prepare: async (input) => ({ + executionId: input.attemptId, + provenance: { + projectId: input.session.projectId, + sessionId: input.session.sessionId, + agentFrameId: input.frameId, + runtimeSegmentId: input.runtimeSegmentId + }, + workspace: { cwd: '/workspace/retry-cleanup' }, + runtimeHome: '/runtime/retry-cleanup', + frameworkId: 'certified-test', + capability: { + async revoke() { + revokeAttempts += 1 + if (revokeAttempts === 1) throw new Error('revoke failed once') + cleanup.push('revoke') + } + }, + disposeResources: async () => { + cleanup.push('resources') + } + }), + assertFrameworkNativeDelegationDisabled: async () => undefined, + createRuntime: (_scope, runtimeCallbacks) => { + callbacks = runtimeCallbacks + return { + createSession: async () => ({ sessionId: 'provider-retry-cleanup' }), + sendAppContinuation: () => { + callbacks.onProviderPromptAccepted('provider-retry-cleanup') + return new Promise(() => undefined) + }, + cancelPrompt: async () => undefined, + respondToPermission: async () => undefined, + setPermissionProfile: async () => undefined, + deleteSession: async () => { + cleanup.push('delete') + }, + shutdownForQuit: async () => { + cleanup.push('shutdown') + return { reaped: true } + } + } + } + }) + const reservation = await execution.reserve(1) + const running = execution.run(makeInput('retry-cleanup'), reservation.slotIds[0]) + await running.accepted + + await expect(running.cancel()).rejects.toThrow('revoke failed once') + await expect(running.cancel()).resolves.toBeUndefined() + + expect(revokeAttempts).toBe(2) + expect(cleanup).toEqual(['revoke', 'delete', 'shutdown', 'resources']) + await expect(execution.reserve(1)).resolves.toHaveProperty('slotIds') + }) + it('reserves an entire batch atomically and releases terminal slots', async () => { const { execution, controls } = makeHarness(2) const reservation = await execution.reserve(2) diff --git a/src/main/delegation/acp-execution.ts b/src/main/delegation/acp-execution.ts index b16567b74..535f41ba4 100644 --- a/src/main/delegation/acp-execution.ts +++ b/src/main/delegation/acp-execution.ts @@ -285,6 +285,10 @@ const createAcpDelegateExecution = (options: AcpDelegateExecutionOptions): Deleg let ownsWorkspace = false let writable = true let capabilityRevoked = false + let providerSessionDeleted = false + let runtimeShutdown = false + let resourcesDisposed = false + let slotReleased = false let acceptedSettled = false let terminalSettled = false let cancelRequested = false @@ -295,7 +299,7 @@ const createAcpDelegateExecution = (options: AcpDelegateExecutionOptions): Deleg let lastStopEvent: AcpAgentRuntimeUpdate['event'] | undefined let currentStopEvent: AcpAgentRuntimeUpdate['event'] | undefined // Provider event ids are unique within this Attempt-owned runtime lifetime. - const seenStopEventIds = new Set() + const seenEventIds = new Set() let activeMessage: QueuedPrompt | undefined let activeTurn: QueuedPrompt['turn'] let providerPromptStarted = false @@ -313,6 +317,9 @@ const createAcpDelegateExecution = (options: AcpDelegateExecutionOptions): Deleg if (!writable || terminalSettled) return for (const listener of listeners) listener(event) } + const publishObservation = (event: DelegateExecutionEvent): void => { + for (const listener of listeners) listener(event) + } const callbacks: AcpDelegateExecutionCallbacks = { onProviderPromptAccepted(sessionId) { if (!writable || sessionId !== providerSessionId) return @@ -320,10 +327,9 @@ const createAcpDelegateExecution = (options: AcpDelegateExecutionOptions): Deleg else settleAccepted('provider_prompt_accepted') }, onEvent(event) { - if (!writable || event.sessionId !== providerSessionId) return + if (event.sessionId !== providerSessionId || seenEventIds.has(event.id)) return + seenEventIds.add(event.id) if (event.kind === 'stop') { - if (seenStopEventIds.has(event.id)) return - seenStopEventIds.add(event.id) sawStopEvent = true if (!event.turnUsage || !turnUsageAvailable) { turnUsageAvailable = false @@ -340,7 +346,7 @@ const createAcpDelegateExecution = (options: AcpDelegateExecutionOptions): Deleg const text = getAcpRuntimeEventText(event) if (event.kind === 'message' && event.role === 'assistant' && text) { currentResponse.push(text) - publish({ kind: 'message', text }) + if (writable && !terminalSettled) publish({ kind: 'message', text }) } const promptMessageId = activeTurn?.promptMessageId ?? scope?.provenance.promptMessageId @@ -366,9 +372,10 @@ const createAcpDelegateExecution = (options: AcpDelegateExecutionOptions): Deleg if (event.kind === 'stop') { lastStopEvent = ownedEvent currentStopEvent = ownedEvent + if (cancelRequested || terminalSettled) publishObservation({ kind: 'runtime', update }) return } - publish({ kind: 'runtime', update }) + publishObservation({ kind: 'runtime', update }) }, onPermissionRequest(request) { if (!writable || request.sessionId !== providerSessionId) return @@ -396,27 +403,29 @@ const createAcpDelegateExecution = (options: AcpDelegateExecutionOptions): Deleg activeMessage = undefined for (const pending of queuedPrompts.splice(0)) pending.acceptance.reject(deliveryError) if (scope && !capabilityRevoked) { - capabilityRevoked = true await scope.capability.revoke() + capabilityRevoked = true } } - const cleanup = async (): Promise => { + const cleanupOnce = async (): Promise => { let firstError: unknown try { await revokeWrites() } catch (error) { firstError = error } - if (runtime && providerSessionId) { + if (runtime && providerSessionId && !providerSessionDeleted) { try { await runtime.deleteSession({ sessionId: providerSessionId }) + providerSessionDeleted = true } catch (error) { firstError ??= error } } - if (runtime) { + if (runtime && !runtimeShutdown) { try { await runtime.shutdownForQuit() + runtimeShutdown = true } catch (error) { firstError ??= error } @@ -430,16 +439,28 @@ const createAcpDelegateExecution = (options: AcpDelegateExecutionOptions): Deleg activeWorkspaces.delete(scope.workspace.cwd) ownsWorkspace = false } - try { - await scope.disposeResources?.() - } catch (error) { - firstError ??= error + if (!resourcesDisposed) { + try { + await scope.disposeResources?.() + resourcesDisposed = true + } catch (error) { + firstError ??= error + } } } listeners.clear() - releaseSlot(slotId) + if (!slotReleased) { + slotReleased = true + releaseSlot(slotId) + } if (firstError !== undefined) throw firstError } + let cleanupTail = Promise.resolve() + const cleanup = (): Promise => { + const next = cleanupTail.then(cleanupOnce, cleanupOnce) + cleanupTail = next.catch(() => undefined) + return next + } const promptRequest = ( text: string @@ -533,6 +554,7 @@ const createAcpDelegateExecution = (options: AcpDelegateExecutionOptions): Deleg let response = '' while (!cancelRequested) { await activeTurn?.begin?.() + if (cancelRequested) break currentResponse = [] providerPromptStarted = true const outcome = await runtime.sendAppContinuation(promptRequest(nextPrompt)) @@ -616,7 +638,7 @@ const createAcpDelegateExecution = (options: AcpDelegateExecutionOptions): Deleg accepted: acceptance.promise, completion: terminal.promise, subscribe(listener) { - if (!terminalSettled) listeners.add(listener) + listeners.add(listener) return () => listeners.delete(listener) }, async sendMessage(message, turn) { @@ -653,13 +675,27 @@ const createAcpDelegateExecution = (options: AcpDelegateExecutionOptions): Deleg publish({ kind: 'permission', awaiting: false, requestId: response.requestId }) }, async cancel() { - if (terminalSettled) return + if (terminalSettled && !cancelRequested) return cancelRequested = true - await revokeWrites().catch(() => undefined) + await revokeWrites() if (runtime && providerSessionId) { await runtime.cancelPrompt({ sessionId: providerSessionId }).catch(() => undefined) } - await work.catch(() => undefined) + // A terminated provider may never settle its in-flight prompt. Cancellation owns the + // terminal signal; cleanup is idempotent and remains observed if transport teardown stalls + // or the provider task resumes later. + settleAccepted( + 'provider_prompt_completed', + new DelegateMessagePreAcceptanceError( + 'delegate execution was cancelled before provider acceptance' + ) + ) + if (!terminalSettled) { + terminalSettled = true + terminal.resolve({ status: 'cancelled' }) + } + await cleanup() + void work.catch(() => undefined) } }) } diff --git a/src/main/delegation/attempt-runtime-transcript.test.ts b/src/main/delegation/attempt-runtime-transcript.test.ts index af830bf49..930b831b3 100644 --- a/src/main/delegation/attempt-runtime-transcript.test.ts +++ b/src/main/delegation/attempt-runtime-transcript.test.ts @@ -1,7 +1,11 @@ import { describe, expect, it } from 'vitest' import type { AcpAgentRuntimeUpdate } from '../../shared/acp' -import { projectAttemptRuntimeTranscript } from './attempt-runtime-transcript' +import { + createAttemptRuntimeTranscriptStager, + projectAttemptRuntimeTranscript +} from './attempt-runtime-transcript' +import type { DelegatedWorkDurableRecords, DurableMessage } from './delegated-work-record-types' const scope = { projectId: 'project-1', @@ -15,6 +19,112 @@ const scope = { const update = (event: AcpAgentRuntimeUpdate['event']): AcpAgentRuntimeUpdate => ({ scope, event }) describe('Attempt runtime transcript projection', () => { + it('isolates two Turn lanes and retains one Message identity per provider stream', async () => { + const messages = new Map() + const activityWrites: Array<{ + runtimeSegmentId: string + promptMessageIds: string[] + }> = [] + const records = { + async stageTerminalMessage(_frameId: string, _attemptId: string, message: DurableMessage) { + messages.set(message.id, structuredClone(message)) + }, + async stageTerminalActivities( + _frameId: string, + _attemptId: string, + runtimeSegmentId: string, + activities: Parameters< + NonNullable + >[3] + ) { + activityWrites.push({ + runtimeSegmentId, + promptMessageIds: activities.map((activity) => activity.promptMessageId!) + }) + } + } as unknown as DelegatedWorkDurableRecords + const ids = ['message-lane-1', 'message-lane-2'] + const owner = createAttemptRuntimeTranscriptStager({ + records, + frameId: 'frame-1', + attemptId: 'attempt-1', + createMessageId: () => ids.shift()! + }) + const lane1 = { runtimeSegmentId: 'runtime-1', promptMessageId: 'prompt-1' } + const lane2 = { runtimeSegmentId: 'runtime-2', promptMessageId: 'prompt-2' } + const observe = async ( + lane: typeof lane1, + event: AcpAgentRuntimeUpdate['event'] + ): Promise => owner.observe({ scope: { ...scope, ...lane }, event }) + + await observe(lane1, { + id: 'lane-1-message', + timestamp: 10, + kind: 'message', + level: 'info', + messageId: 'provider-stream', + role: 'assistant', + text: 'First turn' + }) + await observe(lane1, { + id: 'lane-1-tool', + timestamp: 11, + kind: 'tool', + level: 'info', + toolCallId: 'tool-1', + status: 'in_progress' + }) + await owner.settle(lane1, { + terminalStatus: 'completed', + endedAt: 12, + fallbackResponse: 'First turn' + }) + await observe(lane2, { + id: 'lane-2-message', + timestamp: 20, + kind: 'message', + level: 'info', + messageId: 'provider-stream', + role: 'assistant', + text: 'Second turn' + }) + await observe(lane2, { + id: 'lane-2-tool', + timestamp: 21, + kind: 'tool', + level: 'info', + toolCallId: 'tool-2', + status: 'completed' + }) + await owner.settle(lane2, { + terminalStatus: 'completed', + endedAt: 22, + fallbackResponse: 'Second turn' + }) + + expect([...messages.values()]).toEqual([ + expect.objectContaining({ + id: 'message-lane-1', + responseToMessageId: 'prompt-1', + runtimeSegmentId: 'runtime-1', + content: 'First turn' + }), + expect.objectContaining({ + id: 'message-lane-2', + responseToMessageId: 'prompt-2', + runtimeSegmentId: 'runtime-2', + content: 'Second turn' + }) + ]) + expect(activityWrites).toEqual( + expect.arrayContaining([ + { runtimeSegmentId: 'runtime-1', promptMessageIds: ['prompt-1'] }, + { runtimeSegmentId: 'runtime-2', promptMessageIds: ['prompt-2'] } + ]) + ) + expect(ids).toEqual([]) + }) + it('preserves message boundaries, tool history, groups, and terminal usage', () => { let messageId = 0 const transcript = projectAttemptRuntimeTranscript({ @@ -82,6 +192,7 @@ describe('Attempt runtime transcript projection', () => { promptMessageId: 'prompt-1', fallbackResponse: 'Final answer.', endedAt: 20, + terminalStatus: 'completed', turnUsage: { inputTokens: 100, cacheTokens: 20, @@ -145,6 +256,7 @@ describe('Attempt runtime transcript projection', () => { promptMessageId: 'prompt-1', fallbackResponse: 'Fallback response', endedAt: 30, + terminalStatus: 'completed', turnUsageUnavailable: true, createMessageId: () => 'fallback-message' }) @@ -159,6 +271,44 @@ describe('Attempt runtime transcript projection', () => { ]) }) + it('keeps live message and tool evidence open until terminal settlement', () => { + const transcript = projectAttemptRuntimeTranscript({ + updates: [ + update({ + id: 'live-message', + timestamp: 10, + kind: 'message', + level: 'info', + messageId: 'live-stream', + role: 'assistant', + text: 'Still working' + }), + update({ + id: 'live-tool', + timestamp: 11, + kind: 'tool', + level: 'info', + toolCallId: 'tool-live', + title: 'Inspect source', + status: 'in_progress' + }) + ], + frameId: 'frame-1', + promptMessageId: 'prompt-1', + runtimeSegmentId: 'runtime-1', + fallbackResponse: '', + endedAt: 11, + createMessageId: () => 'live-message-id' + }) + + expect(transcript.messages).toEqual([ + expect.objectContaining({ id: 'live-message-id', status: 'streaming' }) + ]) + expect(transcript.messages[0].completedAt).toBeUndefined() + expect(transcript.activities).toEqual([expect.objectContaining({ status: 'in_progress' })]) + expect('completedAt' in transcript.activities[0]).toBe(false) + }) + it('derives graph-unique activity identities from each app-owned Runtime Segment', () => { const project = ( runtimeSegmentId: string @@ -195,6 +345,7 @@ describe('Attempt runtime transcript projection', () => { promptMessageId: 'prompt-1', fallbackResponse: 'done', endedAt: 12, + terminalStatus: 'completed', createMessageId: () => `${runtimeSegmentId}:message` }) diff --git a/src/main/delegation/attempt-runtime-transcript.ts b/src/main/delegation/attempt-runtime-transcript.ts index f393bee36..602ec198a 100644 --- a/src/main/delegation/attempt-runtime-transcript.ts +++ b/src/main/delegation/attempt-runtime-transcript.ts @@ -42,9 +42,19 @@ type StageAttemptRuntimeTranscriptInput = Readonly<{ turnUsageUnavailable?: true }> -type AttemptRuntimeTranscriptStager = ( - input: StageAttemptRuntimeTranscriptInput -) => Promise +type AttemptRuntimeTranscriptLane = Readonly<{ + runtimeSegmentId: string + promptMessageId: string +}> + +type AttemptRuntimeTranscriptStager = Readonly<{ + observe(update: AcpAgentRuntimeUpdate): Promise + settle( + lane: AttemptRuntimeTranscriptLane, + input: StageAttemptRuntimeTranscriptInput + ): Promise + flush(): Promise +}> type AttemptCancellationReason = 'main_agent_stop' | 'session_stop' | 'runtime_interrupted' @@ -107,6 +117,7 @@ const projectAttemptRuntimeTranscript = ( role: 'assistant', content: text, responseToMessageId: promptMessageId, + runtimeSegmentId: scope.runtimeSegmentId, eventIds: [event.id], ...(event.image ? { images: [{ id: event.id, ...event.image }] } : {}), createdAt: event.timestamp, @@ -194,7 +205,7 @@ const projectAttemptRuntimeTranscript = ( } const messages = [...messagesByStream.values()] - const terminalStatus = input.terminalStatus ?? 'completed' + const terminalStatus = input.terminalStatus if (messages.length === 0 && terminalStatus === 'completed') { messages.push({ id: input.createMessageId(), @@ -202,6 +213,7 @@ const projectAttemptRuntimeTranscript = ( role: 'assistant', content: input.fallbackResponse, responseToMessageId: promptMessageId, + runtimeSegmentId: input.runtimeSegmentId, eventIds: [], createdAt: input.endedAt, updatedAt: input.endedAt, @@ -210,6 +222,10 @@ const projectAttemptRuntimeTranscript = ( } const terminalMessage = messages[messages.length - 1] for (const message of messages) { + if (!terminalStatus) { + message.status = 'streaming' + continue + } const isTerminalMessage = message === terminalMessage message.status = isTerminalMessage && terminalStatus !== 'completed' ? 'error' : 'complete' message.completedAt = isTerminalMessage ? input.endedAt : message.updatedAt @@ -221,7 +237,7 @@ const projectAttemptRuntimeTranscript = ( } const activities = [...activitiesById.values()].map((activity) => - isTerminalToolStatus(activity.status) + !terminalStatus || isTerminalToolStatus(activity.status) ? activity : { ...activity, @@ -229,12 +245,16 @@ const projectAttemptRuntimeTranscript = ( updatedAt: input.endedAt } ) - const activityGroups = [...groupsById.values()].map((group) => ({ - ...group, - promptMessageId: group.promptMessageId ?? promptMessageId, - completedAt: input.endedAt, - updatedAt: input.endedAt - })) + const activityGroups = [...groupsById.values()].map((group) => + terminalStatus + ? { + ...group, + promptMessageId: group.promptMessageId ?? promptMessageId, + completedAt: input.endedAt, + updatedAt: input.endedAt + } + : group + ) return { messages, activities, activityGroups, terminalMessage } } @@ -263,19 +283,38 @@ const createAttemptRuntimeTranscriptStager = (options: { records: DelegatedWorkDurableRecords frameId: string attemptId: string - updates: readonly AcpAgentRuntimeUpdate[] - promptMessageId(): string | undefined createMessageId(): string }): AttemptRuntimeTranscriptStager => { - let stagingStarted = false - return async (input) => { - const promptMessageId = options.promptMessageId() - if (!promptMessageId || stagingStarted) return undefined - stagingStarted = true + type LaneState = { + updates: AcpAgentRuntimeUpdate[] + messageIds: string[] + terminalStatus?: StageAttemptRuntimeTranscriptInput['terminalStatus'] + } + const lanes = new Map() + let stagingTail: Promise = Promise.resolve() + const laneKey = (lane: AttemptRuntimeTranscriptLane): string => + `${lane.runtimeSegmentId}\u0000${lane.promptMessageId}` + const laneState = (lane: AttemptRuntimeTranscriptLane): LaneState => { + const key = laneKey(lane) + const existing = lanes.get(key) + if (existing) return existing + const created: LaneState = { updates: [], messageIds: [] } + lanes.set(key, created) + return created + } + const stage = async ( + lane: AttemptRuntimeTranscriptLane, + state: LaneState, + input: Omit & { + terminalStatus?: StageAttemptRuntimeTranscriptInput['terminalStatus'] + } + ): Promise => { + let messageIndex = 0 return stageAttemptRuntimeTranscript(options.records, options.frameId, options.attemptId, { - updates: options.updates, + updates: state.updates, frameId: options.frameId, - promptMessageId, + promptMessageId: lane.promptMessageId, + runtimeSegmentId: lane.runtimeSegmentId, fallbackResponse: input.fallbackResponse ?? '', endedAt: input.endedAt, terminalStatus: input.terminalStatus, @@ -284,8 +323,53 @@ const createAttemptRuntimeTranscriptStager = (options: { : input.turnUsageUnavailable ? { turnUsageUnavailable: true } : {}), - createMessageId: options.createMessageId + createMessageId: () => { + const index = messageIndex++ + return (state.messageIds[index] ??= options.createMessageId()) + } + }) + } + const observe = async (update: AcpAgentRuntimeUpdate): Promise => { + if ( + update.scope.agentFrameId !== options.frameId || + update.scope.attemptId !== options.attemptId + ) { + throw new Error('Runtime evidence does not belong to the transcript Attempt.') + } + const lane = { + runtimeSegmentId: update.scope.runtimeSegmentId, + promptMessageId: update.scope.promptMessageId + } + const state = laneState(lane) + const next = stagingTail.then(async () => { + if (!state.updates.some((candidate) => candidate.event.id === update.event.id)) { + state.updates.push(update) + } + await stage(lane, state, { + ...(state.terminalStatus ? { terminalStatus: state.terminalStatus } : {}), + endedAt: update.event.timestamp, + fallbackResponse: '' + }) }) + stagingTail = next + return next + } + const settle = ( + lane: AttemptRuntimeTranscriptLane, + input: StageAttemptRuntimeTranscriptInput + ): Promise => { + const state = laneState(lane) + state.terminalStatus = input.terminalStatus + const settled = stagingTail.then(() => stage(lane, state, input)) + stagingTail = settled + return settled + } + return { + observe, + settle, + async flush() { + await stagingTail + } } } @@ -297,16 +381,20 @@ const terminalizeUnsuccessfulAttempt = async ( attemptId: string endedAt: number error: unknown + lane?: AttemptRuntimeTranscriptLane cancellationReason?: AttemptCancellationReason }> ): Promise => { let terminalError = input.error try { - await stageTranscript({ - terminalStatus: input.cancellationReason ? 'cancelled' : 'error', - endedAt: input.endedAt - }) + if (input.lane) { + await stageTranscript.settle(input.lane, { + terminalStatus: input.cancellationReason ? 'cancelled' : 'error', + endedAt: input.endedAt + }) + } } catch (stagingError) { + if (input.cancellationReason) throw stagingError terminalError = stagingError } if (input.cancellationReason) { @@ -339,6 +427,7 @@ export { } export type { AttemptRuntimeTranscript, + AttemptRuntimeTranscriptLane, AttemptRuntimeTranscriptStager, ProjectAttemptRuntimeTranscriptInput, StageAttemptRuntimeTranscriptInput diff --git a/src/main/delegation/delegated-cleanup.ts b/src/main/delegation/delegated-cleanup.ts new file mode 100644 index 000000000..9b17987a1 --- /dev/null +++ b/src/main/delegation/delegated-cleanup.ts @@ -0,0 +1,124 @@ +import type { SessionKey } from './durable-delegated-work-contract' + +type DelegatedCleanupScope = Readonly<{ + session: SessionKey + frameId: string + attemptId: string +}> + +type CleanupErrorReporter = (scope: DelegatedCleanupScope, error: unknown) => void + +type DelegatedCleanup = Readonly<{ + report(scope: DelegatedCleanupScope, operation: string, error: unknown): void + start( + scope: DelegatedCleanupScope, + operation: string, + cleanup: () => unknown | Promise, + reportsOwnFailure?: boolean + ): void + retryable( + scope: DelegatedCleanupScope, + operationName: string, + operation: () => Promise + ): () => Promise +}> + +const createDelegatedCleanup = (onCleanupError?: CleanupErrorReporter): DelegatedCleanup => { + const report = (scope: DelegatedCleanupScope, operation: string, error: unknown): void => { + const cleanupError = new AggregateError( + [error], + `Detached Subagent cleanup failed during ${operation}.` + ) + try { + if (onCleanupError) { + onCleanupError(scope, cleanupError) + return + } + console.error('[delegated-work] Detached Subagent cleanup failed.', { + ...scope, + operation, + error + }) + } catch (reportError) { + console.error('[delegated-work] Could not report detached Subagent cleanup failure.', { + ...scope, + operation, + error, + reportError + }) + } + } + + const start = ( + scope: DelegatedCleanupScope, + operation: string, + cleanup: () => unknown | Promise, + reportsOwnFailure = false + ): void => { + let result: unknown | Promise + try { + result = cleanup() + } catch (error) { + report(scope, operation, error) + return + } + void Promise.resolve(result).catch((error) => { + if (!reportsOwnFailure) report(scope, operation, error) + }) + } + + const retryable = ( + scope: DelegatedCleanupScope, + operationName: string, + operation: () => Promise + ): (() => Promise) => { + let completed = false + let inFlight: Promise | undefined + let queuedRetry: Promise | undefined + const runStep = (): Promise => { + if (completed) return Promise.resolve() + if (inFlight) { + if (!queuedRetry) { + const current = inFlight + const retry = current.catch(() => runStep()) + queuedRetry = retry + void retry.then( + () => { + if (queuedRetry === retry) queuedRetry = undefined + }, + () => { + if (queuedRetry === retry) queuedRetry = undefined + } + ) + } + return queuedRetry + } + let started: Promise + try { + started = operation() + } catch (error) { + report(scope, operationName, error) + return Promise.reject(error) + } + const current = started.then( + () => { + completed = true + inFlight = undefined + }, + (error) => { + inFlight = undefined + report(scope, operationName, error) + throw error + } + ) + inFlight = current + return current + } + return runStep + } + + return { report, retryable, start } +} + +export { createDelegatedCleanup } +export type { DelegatedCleanupScope } diff --git a/src/main/delegation/delegated-turn-lifecycle.ts b/src/main/delegation/delegated-turn-lifecycle.ts index a78e96972..7524d99e9 100644 --- a/src/main/delegation/delegated-turn-lifecycle.ts +++ b/src/main/delegation/delegated-turn-lifecycle.ts @@ -1,6 +1,5 @@ -import type { AcpAgentRuntimeUpdate } from '../../shared/acp' import type { ArtifactFile } from '../../shared/artifacts' -import { stageAttemptRuntimeTranscript } from './attempt-runtime-transcript' +import type { AttemptRuntimeTranscriptStager } from './attempt-runtime-transcript' import type { DurableMessage, DelegatedWorkDurableRecords } from './delegated-work-record-types' import type { DelegateExecutionInput } from './execution-port' @@ -51,9 +50,8 @@ const createDelegatedTurnLifecycle = (options: { attemptId: string agentFrameId: string agentName: string - runtimeUpdates: AcpAgentRuntimeUpdate[] + transcript: AttemptRuntimeTranscriptStager now(): number - createMessageId(): string }): Readonly<{ openInitial(context: TurnContext): Promise currentArtifact(): DelegatedArtifactHandle | undefined @@ -64,27 +62,43 @@ const createDelegatedTurnLifecycle = (options: { }> => { let currentArtifact: DelegatedArtifactHandle | undefined const artifactHandles: DelegatedArtifactHandle[] = [] + const disposedArtifacts = new Set() + const pendingArtifactOpens = new Set>() let artifactHandoffFile: string | undefined - let stagedRuntimeUpdateCount = 0 let completedTurnMessage: DurableMessage | undefined + let disposeRequested = false - const openArtifact = async (context: TurnContext, executionId: string): Promise => { - const artifact = await options.artifactEvidence?.open({ - session: options.session, - executionId, - attemptId: options.attemptId, - rootFrameId: context.rootFrameId, - agentFrameId: options.agentFrameId, - messageBranchId: context.messageBranchId, - runtimeSegmentId: context.runtimeSegmentId, - promptMessageId: context.promptMessageId, - agentName: options.agentName - }) - if (!artifact) return - currentArtifact = artifact - artifactHandles.push(artifact) - if (artifactHandoffFile) await artifact.activateAt?.(artifactHandoffFile) - else artifactHandoffFile = artifact.execution?.currentRunFile + const openArtifact = (context: TurnContext, executionId: string): Promise => { + if (disposeRequested) return Promise.resolve() + const opening = (async () => { + const artifact = await options.artifactEvidence?.open({ + session: options.session, + executionId, + attemptId: options.attemptId, + rootFrameId: context.rootFrameId, + agentFrameId: options.agentFrameId, + messageBranchId: context.messageBranchId, + runtimeSegmentId: context.runtimeSegmentId, + promptMessageId: context.promptMessageId, + agentName: options.agentName + }) + if (!artifact) return + artifactHandles.push(artifact) + if (disposeRequested) { + await artifact.dispose() + disposedArtifacts.add(artifact) + return + } + currentArtifact = artifact + if (artifactHandoffFile) await artifact.activateAt?.(artifactHandoffFile) + else artifactHandoffFile = artifact.execution?.currentRunFile + })() + pendingArtifactOpens.add(opening) + void opening.then( + () => pendingArtifactOpens.delete(opening), + () => pendingArtifactOpens.delete(opening) + ) + return opening } return { @@ -102,28 +116,16 @@ const createDelegatedTurnLifecycle = (options: { : {}), async complete(response, turnUsage, turnUsageUnavailable) { const completedAt = options.now() - const turnUpdates = options.runtimeUpdates.slice(stagedRuntimeUpdateCount) - stagedRuntimeUpdateCount = options.runtimeUpdates.length - const transcript = await stageAttemptRuntimeTranscript( - options.records, - options.agentFrameId, - options.attemptId, - { - updates: turnUpdates, - frameId: options.agentFrameId, - promptMessageId: context.promptMessageId, - runtimeSegmentId: context.runtimeSegmentId, - fallbackResponse: response, - endedAt: completedAt, - terminalStatus: 'completed', - ...(turnUsage - ? { turnUsage } - : turnUsageUnavailable - ? { turnUsageUnavailable: true } - : {}), - createMessageId: options.createMessageId - } - ) + const transcript = await options.transcript.settle(context, { + fallbackResponse: response, + endedAt: completedAt, + terminalStatus: 'completed', + ...(turnUsage + ? { turnUsage } + : turnUsageUnavailable + ? { turnUsageUnavailable: true } + : {}) + }) const message = transcript.terminalMessage if (!message) throw new Error('Completed child Turn has no final agent Message.') await currentArtifact?.finalize(message.id) @@ -140,7 +142,17 @@ const createDelegatedTurnLifecycle = (options: { await currentArtifact?.finalize(terminalMessageId) }, async dispose() { - await Promise.allSettled(artifactHandles.map((artifact) => artifact.dispose())) + disposeRequested = true + const openingAtDisposal = [...pendingArtifactOpens] + const disposalAtStart = artifactHandles + .filter((artifact) => !disposedArtifacts.has(artifact)) + .map(async (artifact) => { + await artifact.dispose() + disposedArtifacts.add(artifact) + }) + // An open that was already in flight when disposal began owns disposal of its late handle. + // Await it alongside current handles so failure keeps the outer finalizer retryable. + await Promise.all([...disposalAtStart, ...openingAtDisposal]) } } } diff --git a/src/main/delegation/delegated-user-question-owner.ts b/src/main/delegation/delegated-user-question-owner.ts index 082db958d..66a5b4feb 100644 --- a/src/main/delegation/delegated-user-question-owner.ts +++ b/src/main/delegation/delegated-user-question-owner.ts @@ -97,65 +97,67 @@ class DelegatedUserQuestionOwner { 'delegated user question requires a trusted direct-child capability' ) } - const snapshot = await this.options.records.snapshot() - if (!sameSession(snapshot.session, caller.session)) { - throw new DurableDelegatedWorkError('authorization', 'delegated question Session mismatch') - } - const requestId = explicitRequestId?.trim() || this.options.createId('question') - const canonicalDigest = createHash('sha256') - .update(JSON.stringify(request.questions)) - .digest('hex') - const existing = snapshot.questionRequests.find( - (candidate) => candidate.requestId === requestId - ) - if (existing) { + return this.options.admission(async () => { + const snapshot = await this.options.records.snapshot() + if (!sameSession(snapshot.session, caller.session)) { + throw new DurableDelegatedWorkError('authorization', 'delegated question Session mismatch') + } + const requestId = explicitRequestId?.trim() || this.options.createId('question') + const canonicalDigest = createHash('sha256') + .update(JSON.stringify(request.questions)) + .digest('hex') + const existing = snapshot.questionRequests.find( + (candidate) => candidate.requestId === requestId + ) + if (existing) { + if ( + existing.canonicalDigest !== canonicalDigest || + existing.sourceFrameId !== caller.frameId || + existing.sourceAttemptId !== caller.attemptId + ) { + throw new DurableDelegatedWorkError( + 'conflict', + 'delegated user question request identity was reused with different content or source' + ) + } + return { action: 'pending' } + } + const child = snapshot.records.find((candidate) => candidate.frameId === caller.frameId) + const attempt = child && currentAttempt(child as DurableChild) + const runtimeSegmentId = attempt?.runtimeSegmentIds.at(-1) if ( - existing.canonicalDigest !== canonicalDigest || - existing.sourceFrameId !== caller.frameId || - existing.sourceAttemptId !== caller.attemptId + !child || + child.parentFrameId !== snapshot.rootFrameId || + child.originBindingState !== 'validated' || + !snapshot.originMessageIds.includes(child.originMessageId) || + !attempt || + attempt.id !== caller.attemptId || + attempt.status !== 'running' || + !runtimeSegmentId ) { throw new DurableDelegatedWorkError( - 'conflict', - 'delegated user question request identity was reused with different content or source' + 'authorization', + 'delegated user question source is not the active direct-child Attempt' ) } + await this.options.records.admitQuestion({ + requestId, + canonicalDigest, + sourceFrameId: child.frameId, + sourceAttemptId: attempt.id, + sourceRuntimeSegmentId: runtimeSegmentId, + sourceMessageBranchId: child.messageBranchId, + rootOriginMessageId: child.originMessageId, + rootBranchId: snapshot.rootBranchId, + sourceName: child.title, + questions: structuredClone(request.questions), + askedAt: this.options.now(), + status: 'pending', + draftAnswers: [], + draftQuestionIndex: 0 + }) return { action: 'pending' } - } - const child = snapshot.records.find((candidate) => candidate.frameId === caller.frameId) - const attempt = child && currentAttempt(child as DurableChild) - const runtimeSegmentId = attempt?.runtimeSegmentIds.at(-1) - if ( - !child || - child.parentFrameId !== snapshot.rootFrameId || - child.originBindingState !== 'validated' || - !snapshot.originMessageIds.includes(child.originMessageId) || - !attempt || - attempt.id !== caller.attemptId || - attempt.status !== 'running' || - !runtimeSegmentId - ) { - throw new DurableDelegatedWorkError( - 'authorization', - 'delegated user question source is not the active direct-child Attempt' - ) - } - await this.options.records.admitQuestion({ - requestId, - canonicalDigest, - sourceFrameId: child.frameId, - sourceAttemptId: attempt.id, - sourceRuntimeSegmentId: runtimeSegmentId, - sourceMessageBranchId: child.messageBranchId, - rootOriginMessageId: child.originMessageId, - rootBranchId: snapshot.rootBranchId, - sourceName: child.title, - questions: structuredClone(request.questions), - askedAt: this.options.now(), - status: 'pending', - draftAnswers: [], - draftQuestionIndex: 0 }) - return { action: 'pending' } } async updateDraft(session: SessionKey, input: UpdateQuestionDraftInput): Promise { diff --git a/src/main/delegation/delegated-work-record-types.ts b/src/main/delegation/delegated-work-record-types.ts index c6fff3161..07ee214c0 100644 --- a/src/main/delegation/delegated-work-record-types.ts +++ b/src/main/delegation/delegated-work-record-types.ts @@ -194,7 +194,7 @@ type DurableMessage = { content: string responseToMessageId?: string runtimeSegmentId?: string - status?: 'complete' | 'error' + status?: 'complete' | 'streaming' | 'error' eventIds?: string[] images?: PersistedMessageImage[] turnUsage?: AcpTurnTokenUsage @@ -302,6 +302,15 @@ type DelegatedWorkDurableRecords = Readonly<{ ): Promise confirmQuestion(input: ConfirmQuestionInput): Promise cancelQuestions(frameId: string, endedAt: number, reason: string): Promise + cancelAttempt( + input: Readonly<{ + frameId: string + attemptId: string + endedAt: number + cancellationReason: 'main_agent_stop' | 'session_stop' | 'runtime_interrupted' + questionReason: string + }> + ): Promise<'cancelled' | 'already_terminal'> startRuntime( frameId: string, attemptId: string, diff --git a/src/main/delegation/durable-delegated-work-contract.ts b/src/main/delegation/durable-delegated-work-contract.ts index cc836793a..49690620f 100644 --- a/src/main/delegation/durable-delegated-work-contract.ts +++ b/src/main/delegation/durable-delegated-work-contract.ts @@ -298,6 +298,10 @@ type CreateDurableDelegatedWorkOptions = Readonly<{ reviewEvidence?: DelegatedReviewEvidence onRootPermissionEvent?(event: RootDelegatePermissionEvent): void onAgentRuntimeUpdate?(update: AcpAgentRuntimeUpdate): void + onCleanupError?( + scope: Readonly<{ session: SessionKey; frameId: string; attemptId: string }>, + error: unknown + ): void now?: () => number createId?: (kind: 'frame' | 'attempt' | 'message' | 'runtime' | 'question') => string collectPollIntervalMs?: number diff --git a/src/main/delegation/durable-delegated-work.test.ts b/src/main/delegation/durable-delegated-work.test.ts index 6cf5adf55..b4ee6ca23 100644 --- a/src/main/delegation/durable-delegated-work.test.ts +++ b/src/main/delegation/durable-delegated-work.test.ts @@ -7,14 +7,17 @@ import type { ArtifactFile } from '../../shared/artifacts' import type { ReviewWithChecks } from '../../shared/reviewer' import { createProfileService } from '../specialist/service' import { createDeterministicDelegateExecution } from './deterministic-execution' -import { DelegateMessagePreAcceptanceError } from './execution-port' +import { DelegateMessagePreAcceptanceError, type DelegateExecution } from './execution-port' import { createInMemoryDelegatedWorkRecords, type AuthenticatedDelegateCaller, type DelegatedArtifactEvidence, type DelegatedReviewEvidence } from './durable-delegated-work' -import { createTestDurableDelegatedWork as createDurableDelegatedWork } from './durable-delegated-work-test-fixture' +import { + createTestDurableDelegatedWork as createDurableDelegatedWork, + TEST_EXECUTION_MODEL +} from './durable-delegated-work-test-fixture' const caller: AuthenticatedDelegateCaller = { session: { projectId: 'project-1', sessionId: 'session-1' }, @@ -199,7 +202,32 @@ describe('durable delegated work', () => { 'question-stop' ) - await work.stopSession(caller.session) + execution.controls()[0].complete('Waiting for the pending answer.') + await expect + .poll(async () => (await records.snapshot()).records[0].attempts[0].status) + .toBe('completed') + + await expect(work.stopSession(caller.session)).resolves.toEqual([ + { frameId: 'child-stop', status: 'cancelled' } + ]) + + await expect( + work.requestUserInput( + { + session: caller.session, + frameId: 'child-stop', + role: 'delegate', + attemptId: 'attempt-stop', + originMessageId: 'message-stop', + toolInvocationId: 'ask-after-stop' + }, + { + sessionId: caller.session.sessionId, + questions: [{ question: 'Too late?', options: [{ label: 'Yes' }, { label: 'No' }] }] + }, + 'question-after-stop' + ) + ).rejects.toMatchObject({ code: 'authorization' }) await expect( work.confirmQuestion(caller.session, { @@ -208,9 +236,73 @@ describe('durable delegated work', () => { }) ).rejects.toMatchObject({ code: 'conflict' }) expect((await records.snapshot()).questionRequests[0]).toMatchObject({ status: 'cancelled' }) + expect((await records.snapshot()).records[0].attempts[0]).toMatchObject({ + status: 'completed' + }) + expect((await records.snapshot()).questionRequests).not.toEqual( + expect.arrayContaining([expect.objectContaining({ status: 'pending' })]) + ) expect(execution.controls()).toHaveLength(1) }) + it('does not commit Stop when captured runtime evidence cannot be flushed', async () => { + const execution = createDeterministicDelegateExecution() + const records = createInMemoryDelegatedWorkRecords({ + session: caller.session, + rootFrameId: caller.frameId, + originMessageId: caller.originMessageId + }) + const evidenceFailure = new Error('runtime evidence persistence failed') + const stageTerminalMessage = vi.fn(async () => { + throw evidenceFailure + }) + const cleanupErrors: unknown[] = [] + const work = createDurableDelegatedWork({ + execution, + records: { ...records, stageTerminalMessage }, + onCleanupError: (_scope, error) => cleanupErrors.push(error) + }) + const delegated = await work.delegate( + caller, + { task: 'Preserve evidence before Stop', name: 'Evidence flush' }, + { wait: false } + ) + await expect.poll(() => execution.controls()).toHaveLength(1) + const control = execution.controls()[0] + control.emit({ + kind: 'runtime', + update: { + scope: { + projectId: caller.session.projectId, + sessionId: caller.session.sessionId, + agentFrameId: control.input.frameId, + attemptId: control.input.attemptId, + runtimeSegmentId: control.input.runtimeSegmentId, + promptMessageId: control.input.turn!.promptMessageId + }, + event: { + id: 'captured-before-stop', + timestamp: 20, + kind: 'message', + level: 'info', + messageId: 'provider-message', + role: 'assistant', + text: 'Evidence that must be durable.' + } + } + }) + await expect.poll(() => stageTerminalMessage).toHaveBeenCalled() + + const stopping = await work.stopChildren(caller, [delegated.children[0].frameId]).then( + () => undefined, + (error: unknown) => error + ) + expect(stopping).toBeInstanceOf(AggregateError) + expect((stopping as AggregateError).errors).toContain(evidenceFailure) + expect((await records.snapshot()).records[0].attempts[0]).toMatchObject({ status: 'running' }) + expect(cleanupErrors).not.toHaveLength(0) + }) + it('returns awaiting_user instead of a final result from blocking delegate', async () => { const execution = createDeterministicDelegateExecution() const records = createInMemoryDelegatedWorkRecords({ @@ -686,7 +778,7 @@ describe('durable delegated work', () => { expect(execution.controls()[0].input.attemptId).toBe(childA.children[0].attemptId) }) - it('durably terminalizes every fenced-Turn Attempt when cleanup partially fails', async () => { + it('durably terminalizes every fenced-Turn Attempt and reports detached cleanup failures', async () => { const execution = createDeterministicDelegateExecution() const records = createInMemoryDelegatedWorkRecords({ session: caller.session, @@ -694,9 +786,11 @@ describe('durable delegated work', () => { originMessageId: caller.originMessageId }) let failedOnce = false + const onCleanupError = vi.fn() const work = createDurableDelegatedWork({ execution, records, + onCleanupError, async revokeAttemptWrites() { if (!failedOnce) { failedOnce = true @@ -713,13 +807,17 @@ describe('durable delegated work', () => { { wait: false } ) - await expect(work.cancelTurn(caller.session, caller.originMessageId)).rejects.toThrow( - 'could not be stopped' - ) + await expect(work.cancelTurn(caller.session, caller.originMessageId)).resolves.toHaveLength(2) expect( (await records.snapshot()).records.map((child) => child.attempts.at(-1)!.status) ).toEqual(['cancelled', 'cancelled']) - await expect.poll(() => execution.releasedFrames()).toHaveLength(2) + await expect.poll(() => onCleanupError).toHaveBeenCalledOnce() + expect(onCleanupError).toHaveBeenCalledWith( + expect.objectContaining({ session: caller.session }), + expect.objectContaining({ + message: 'Detached Subagent cleanup failed during attempt write revocation.' + }) + ) }) it('linearizes a Turn fence before an initial admission waiting to commit', async () => { @@ -1035,16 +1133,18 @@ describe('durable delegated work', () => { }) }) - it('restores unresolved permission cards when Stop submission fails', async () => { + it('clears unresolved permission cards while reporting detached Stop cleanup failure', async () => { const execution = createDeterministicDelegateExecution() const records = createInMemoryDelegatedWorkRecords({ session: caller.session, rootFrameId: caller.frameId, originMessageId: caller.originMessageId }) + const onCleanupError = vi.fn() const work = createDurableDelegatedWork({ execution, records, + onCleanupError, revokeAttemptWrites: async () => { throw new Error('stop transport unavailable') } @@ -1063,12 +1163,11 @@ describe('durable delegated work', () => { options: [{ optionId: 'allow', name: 'Allow', kind: 'allow_once' }] }) - await expect(work.stopSession(caller.session)).rejects.toThrow('stop transport unavailable') - await expect(work.rootPermissionRequests(caller.session)).resolves.toMatchObject([ - { requestId: 'permission-retry' } - ]) + await expect(work.stopSession(caller.session)).resolves.toHaveLength(1) + await expect.poll(() => onCleanupError).toHaveBeenCalledOnce() + await expect(work.rootPermissionRequests(caller.session)).resolves.toEqual([]) await expect(work.sessionSummary(caller.session)).resolves.toMatchObject({ - children: [{ status: 'running', awaitingPermission: true }] + children: [{ status: 'cancelled' }] }) }) @@ -2093,8 +2192,22 @@ describe('durable delegated work', () => { ) }) - it('disposes a capability that finishes opening after its Attempt was cancelled', async () => { - const execution = createDeterministicDelegateExecution() + it('reports and retries disposal of a capability that finishes opening after cancellation', async () => { + const deterministicExecution = createDeterministicDelegateExecution() + const reservationRelease = vi.fn() + const execution: DelegateExecution = { + async reserve(count) { + const reservation = await deterministicExecution.reserve(count) + return { + ...reservation, + async release(slotId) { + reservationRelease(slotId) + await reservation.release(slotId) + } + } + }, + run: deterministicExecution.run + } const records = createInMemoryDelegatedWorkRecords({ session: caller.session, rootFrameId: caller.frameId, @@ -2110,10 +2223,23 @@ describe('durable delegated work', () => { }>((resolve) => { resolveOpen = resolve }) - const dispose = vi.fn(async () => undefined) + const dispose = vi + .fn<() => Promise>() + .mockRejectedValueOnce(new Error('late artifact disposal failed')) + .mockResolvedValue(undefined) + const backendClaimRelease = vi.fn(async () => undefined) + const onCleanupError = vi.fn() const work = createDurableDelegatedWork({ execution, records, + onCleanupError, + resolveExecutionModel: () => ({ + snapshot: TEST_EXECUTION_MODEL, + backendLease: { + claim: () => ({ backend: {} as never, release: backendClaimRelease }), + release: async () => undefined + } + }), artifactEvidence: { open: async () => opening, project: async () => [] @@ -2134,8 +2260,17 @@ describe('durable delegated work', () => { resolveOpen({ finalize: async () => undefined, dispose }) await expect(stopping).resolves.toMatchObject([{ status: 'cancelled' }]) - await expect.poll(() => dispose).toHaveBeenCalled() - expect(execution.controls()).toEqual([]) + await expect.poll(() => onCleanupError).toHaveBeenCalledOnce() + expect(onCleanupError).toHaveBeenCalledWith( + expect.objectContaining({ frameId: dispatched.children[0].frameId }), + expect.objectContaining({ + message: 'Detached Subagent cleanup failed during turn artifact disposal.' + }) + ) + await expect.poll(() => dispose).toHaveBeenCalledTimes(2) + expect(backendClaimRelease).toHaveBeenCalledOnce() + expect(reservationRelease).toHaveBeenCalledOnce() + expect(deterministicExecution.controls()).toEqual([]) }) it('defaults an omitted profile to Main Agent without consulting a Specialist resolver', async () => { const execution = createDeterministicDelegateExecution() @@ -3574,6 +3709,248 @@ describe('durable delegated work', () => { }) }) + it('terminalizes Stop when execution cancellation returns but completion never settles', async () => { + let resolveCompletion!: (value: { status: 'completed'; response: string }) => void + const completion = new Promise<{ status: 'completed'; response: string }>((resolve) => { + resolveCompletion = resolve + }) + const cancel = vi.fn(async () => undefined) + const release = vi.fn(async () => undefined) + const execution: DelegateExecution = { + async reserve() { + return { slotIds: ['stalled-slot'], release, releaseAll: release } + }, + run() { + return { + accepted: Promise.resolve('provider_prompt_accepted'), + completion, + subscribe: () => () => undefined, + sendMessage: async () => 'provider_prompt_accepted', + setPermissionProfile: async () => undefined, + respondToPermission: async () => undefined, + cancel + } + } + } + const records = createInMemoryDelegatedWorkRecords({ + session: caller.session, + rootFrameId: caller.frameId, + originMessageId: caller.originMessageId + }) + const work = createDurableDelegatedWork({ execution, records }) + const dispatched = await work.delegate( + caller, + { task: 'Stalled execution', name: 'Stalled execution' }, + { wait: false } + ) + await expect + .poll(async () => (await records.snapshot()).records[0]?.attempts[0]?.runtimeSegmentIds) + .toHaveLength(1) + + await expect(work.stopChildren(caller, [dispatched.children[0].frameId])).resolves.toEqual([ + { frameId: dispatched.children[0].frameId, status: 'cancelled' } + ]) + expect(cancel).toHaveBeenCalledOnce() + await expect(work.sessionSummary(caller.session)).resolves.toMatchObject({ + runningCount: 0, + children: [{ status: 'cancelled' }] + }) + expect(release).toHaveBeenCalledOnce() + + // A provider completion arriving after Stop must not overwrite or duplicate the durable terminal + // transition. + resolveCompletion({ status: 'completed', response: 'too late' }) + await expect + .poll(async () => (await records.snapshot()).records[0]?.attempts[0]?.status) + .toBe('cancelled') + expect(release).toHaveBeenCalledOnce() + }) + + it('returns Stop after terminalizing when cancellation and reservation cleanup never settle', async () => { + const neverSettles = new Promise(() => undefined) + const cancel = vi.fn(() => neverSettles) + const release = vi.fn(() => neverSettles) + const execution: DelegateExecution = { + async reserve() { + return { slotIds: ['hung-cleanup-slot'], release, releaseAll: release } + }, + run() { + return { + accepted: Promise.resolve('provider_prompt_accepted'), + completion: new Promise(() => undefined), + subscribe: () => () => undefined, + sendMessage: async () => 'provider_prompt_accepted', + setPermissionProfile: async () => undefined, + respondToPermission: async () => undefined, + cancel + } + } + } + const records = createInMemoryDelegatedWorkRecords({ + session: caller.session, + rootFrameId: caller.frameId, + originMessageId: caller.originMessageId + }) + const work = createDurableDelegatedWork({ execution, records }) + const dispatched = await work.delegate( + caller, + { task: 'Hung cleanup', name: 'Hung cleanup' }, + { wait: false } + ) + await expect + .poll(async () => (await records.snapshot()).records[0]?.attempts[0]?.runtimeSegmentIds) + .toHaveLength(1) + + const stopped = work.stopChildren(caller, [dispatched.children[0].frameId]) + + await expect.poll(() => cancel).toHaveBeenCalledOnce() + await expect.poll(() => release).toHaveBeenCalledOnce() + await expect(stopped).resolves.toEqual([ + { frameId: dispatched.children[0].frameId, status: 'cancelled' } + ]) + await expect(work.sessionSummary(caller.session)).resolves.toMatchObject({ + runningCount: 0, + children: [{ status: 'cancelled' }] + }) + }) + + it('releases backend and capacity ownership once while artifact and provider drains remain hung', async () => { + const neverSettles = new Promise(() => undefined) + const cancel = vi.fn(() => neverSettles) + const reservationRelease = vi.fn(async () => undefined) + const backendClaimRelease = vi.fn(async () => undefined) + const disposeArtifact = vi.fn(() => neverSettles) + const revokeArtifact = vi.fn(() => neverSettles) + const run = vi.fn(() => ({ + accepted: Promise.resolve('provider_prompt_accepted' as const), + completion: new Promise(() => undefined), + subscribe: () => () => undefined, + sendMessage: async () => 'provider_prompt_accepted' as const, + setPermissionProfile: async () => undefined, + respondToPermission: async () => undefined, + cancel + })) + const execution: DelegateExecution = { + async reserve() { + return { + slotIds: ['independent-cleanup-slot'], + release: reservationRelease, + releaseAll: reservationRelease + } + }, + run + } + const artifactEvidence: DelegatedArtifactEvidence = { + async open() { + return { + finalize: async () => undefined, + dispose: disposeArtifact + } + }, + revoke: revokeArtifact, + project: async () => [] + } + const records = createInMemoryDelegatedWorkRecords({ + session: caller.session, + rootFrameId: caller.frameId, + originMessageId: caller.originMessageId + }) + const work = createDurableDelegatedWork({ + execution, + records, + artifactEvidence, + resolveExecutionModel: () => ({ + snapshot: TEST_EXECUTION_MODEL, + backendLease: { + claim: () => ({ backend: {} as never, release: backendClaimRelease }), + release: async () => undefined + } + }) + }) + const dispatched = await work.delegate( + caller, + { task: 'Independent cleanup', name: 'Independent cleanup' }, + { wait: false } + ) + await expect.poll(() => run).toHaveBeenCalledOnce() + + await expect(work.stopChildren(caller, [dispatched.children[0].frameId])).resolves.toEqual([ + { frameId: dispatched.children[0].frameId, status: 'cancelled' } + ]) + expect(cancel).toHaveBeenCalledOnce() + expect(revokeArtifact).toHaveBeenCalledOnce() + expect(disposeArtifact).toHaveBeenCalledOnce() + expect(backendClaimRelease).toHaveBeenCalledOnce() + expect(reservationRelease).toHaveBeenCalledOnce() + + await expect(work.stopChildren(caller, [dispatched.children[0].frameId])).resolves.toEqual([ + { frameId: dispatched.children[0].frameId, status: 'already_terminal' } + ]) + expect(backendClaimRelease).toHaveBeenCalledOnce() + expect(reservationRelease).toHaveBeenCalledOnce() + }) + + it('reports a failed detached ownership release and retries it after late provider completion', async () => { + let resolveCompletion!: (value: { status: 'completed'; response: string }) => void + const completion = new Promise<{ status: 'completed'; response: string }>((resolve) => { + resolveCompletion = resolve + }) + const release = vi + .fn<() => Promise>() + .mockRejectedValueOnce(new Error('capacity release unavailable')) + .mockResolvedValue(undefined) + const onCleanupError = vi.fn() + const execution: DelegateExecution = { + async reserve() { + return { slotIds: ['retry-cleanup-slot'], release, releaseAll: release } + }, + run() { + return { + accepted: Promise.resolve('provider_prompt_accepted'), + completion, + subscribe: () => () => undefined, + sendMessage: async () => 'provider_prompt_accepted', + setPermissionProfile: async () => undefined, + respondToPermission: async () => undefined, + cancel: async () => undefined + } + } + } + const records = createInMemoryDelegatedWorkRecords({ + session: caller.session, + rootFrameId: caller.frameId, + originMessageId: caller.originMessageId + }) + const work = createDurableDelegatedWork({ execution, records, onCleanupError }) + const dispatched = await work.delegate( + caller, + { task: 'Retry failed cleanup', name: 'Retry failed cleanup' }, + { wait: false } + ) + await expect + .poll(async () => (await records.snapshot()).records[0]?.attempts[0]?.runtimeSegmentIds) + .toHaveLength(1) + + await expect(work.stopChildren(caller, [dispatched.children[0].frameId])).resolves.toEqual([ + { frameId: dispatched.children[0].frameId, status: 'cancelled' } + ]) + await expect.poll(() => onCleanupError).toHaveBeenCalledOnce() + expect(onCleanupError).toHaveBeenCalledWith( + expect.objectContaining({ frameId: dispatched.children[0].frameId }), + expect.objectContaining({ + message: 'Detached Subagent cleanup failed during capacity reservation release.' + }) + ) + expect(release).toHaveBeenCalledOnce() + + resolveCompletion({ status: 'completed', response: 'late completion' }) + await expect.poll(() => release).toHaveBeenCalledTimes(2) + await expect(work.stopChildren(caller, [dispatched.children[0].frameId])).resolves.toEqual([ + { frameId: dispatched.children[0].frameId, status: 'already_terminal' } + ]) + expect(release).toHaveBeenCalledTimes(2) + }) + it('stops the running Session snapshot while preserving terminal history and rejecting new dispatch', async () => { const execution = createDeterministicDelegateExecution() execution.plan({ status: 'completed', response: 'Keep this evidence' }) diff --git a/src/main/delegation/durable-delegated-work.ts b/src/main/delegation/durable-delegated-work.ts index 2263df6c5..f7624c9d7 100644 --- a/src/main/delegation/durable-delegated-work.ts +++ b/src/main/delegation/durable-delegated-work.ts @@ -1,6 +1,5 @@ import { randomUUID } from 'node:crypto' -import type { AcpAgentRuntimeUpdate } from '../../shared/acp' import type { PermissionProfileId } from '../../shared/permission-profiles' import { DelegateExecutionError, @@ -64,6 +63,7 @@ import type { import { submitStructuredOutput } from './structured-output-submission' import { ReliableMessageDeliveryOwner } from './message-delivery-owner' import { DelegatedUserQuestionOwner } from './delegated-user-question-owner' +import { createDelegatedCleanup } from './delegated-cleanup' const createDurableDelegatedWork = ( options: CreateDurableDelegatedWorkOptions @@ -109,6 +109,7 @@ const createDurableDelegatedWork = ( options.resolveSpecialistReference, options.validateInput ) + const cleanup = createDelegatedCleanup(options.onCleanupError) const running = new Map< string, { @@ -117,10 +118,10 @@ const createDurableDelegatedWork = ( deliver(message: DurablePendingMessage): Promise setPermissionProfile(profile: PermissionProfileId): Promise cancel(reason: 'main_agent_stop' | 'session_stop' | 'runtime_interrupted'): Promise - executionStarted(): boolean + flushEvidence(): Promise + finalize(): Promise reservation: DelegateCapacityReservation slotId: string - artifact?: DelegatedArtifactHandle } >() @@ -155,7 +156,12 @@ const createDurableDelegatedWork = ( rejectHandle = reject }) void deliveryHandle.catch(() => undefined) - const runtimeUpdates: AcpAgentRuntimeUpdate[] = [] + const stageRuntimeTranscript = createAttemptRuntimeTranscriptStager({ + records: options.records, + frameId: child.frameId, + attemptId: attempt.id, + createMessageId: () => createId('message') + }) const turnLifecycle = createDelegatedTurnLifecycle({ records: options.records, artifactEvidence: options.artifactEvidence, @@ -166,22 +172,36 @@ const createDurableDelegatedWork = ( attempt.resolvedAgent.kind === 'specialist' ? attempt.resolvedAgent.displayName : 'Main Agent', - runtimeUpdates, - now, - createMessageId: () => createId('message') + transcript: stageRuntimeTranscript, + now }) let cancelRequested = false let cancellationReason: 'main_agent_stop' | 'session_stop' | 'runtime_interrupted' = 'main_agent_stop' let context: Awaited> | undefined - const stageRuntimeTranscript = createAttemptRuntimeTranscriptStager({ - records: options.records, - frameId: child.frameId, - attemptId: attempt.id, - updates: runtimeUpdates, - promptMessageId: () => context?.promptMessageId, - createMessageId: () => createId('message') - }) + const finalizationScope = { session, frameId: child.frameId, attemptId: attempt.id } + const disposeTurn = cleanup.retryable(finalizationScope, 'turn artifact disposal', () => + turnLifecycle.dispose() + ) + const releaseBackendClaim = cleanup.retryable( + finalizationScope, + 'execution backend claim release', + () => (executionBackendClaim ? executionBackendClaim.release() : Promise.resolve()) + ) + const releaseReservation = cleanup.retryable( + finalizationScope, + 'capacity reservation release', + () => reservation.release(slotId) + ) + const finalize = (): Promise => { + permissionOwner.clearAttempt(child.frameId, attempt.id) + if (running.get(child.frameId)?.attemptId === attempt.id) running.delete(child.frameId) + // Start every independent release before awaiting any of them. A provider or artifact that + // never finishes draining must not retain the execution-backend claim or capacity slot. + return Promise.all([disposeTurn(), releaseBackendClaim(), releaseReservation()]).then( + () => undefined + ) + } const completion = (async () => { try { const workspace = await options.workspace?.prepare(session, child.frameId, child.inputs) @@ -197,8 +217,6 @@ const createDurableDelegatedWork = ( } await turnLifecycle.openInitial(startedContext) const artifact = turnLifecycle.currentArtifact() - const runningAttempt = running.get(child.frameId) - if (runningAttempt?.attemptId === attempt.id) runningAttempt.artifact = artifact const ready = await snapshotChild(child.frameId) if (cancelRequested || !ready || currentAttempt(ready).status !== 'running') { throw new Error('delegate execution was cancelled before launch establishment') @@ -228,7 +246,7 @@ const createDurableDelegatedWork = ( handle = options.execution.run(executionInput, slotId) resolveHandle(handle) markEstablished() - const unsubscribe = handle.subscribe((event) => { + handle.subscribe((event) => { permissionOwner.observe(child.frameId, attempt.id, child.title, handle!, event) if (event.kind !== 'runtime') return const { scope: eventScope } = event.update @@ -242,10 +260,11 @@ const createDurableDelegatedWork = ( ) { return } - runtimeUpdates.push(event.update) + void stageRuntimeTranscript + .observe(event.update) + .catch((error) => cleanup.report(finalizationScope, 'runtime evidence append', error)) options.onAgentRuntimeUpdate?.(event.update) }) - void handle.completion.finally(unsubscribe).catch(() => undefined) await Promise.race([handle.accepted, handle.completion.then(() => undefined)]) const outcome = await handle.completion const endedAt = now() @@ -253,16 +272,18 @@ const createDurableDelegatedWork = ( const lastTurnMessage = turnLifecycle.lastTurnMessage() const transcript = lastTurnMessage ? undefined - : await stageRuntimeTranscript({ - terminalStatus: 'completed', - endedAt, - fallbackResponse: outcome.response, - ...(outcome.turnUsage - ? { turnUsage: outcome.turnUsage } - : outcome.turnUsageUnavailable - ? { turnUsageUnavailable: true } - : {}) - }) + : context + ? await stageRuntimeTranscript.settle(context, { + terminalStatus: 'completed', + endedAt, + fallbackResponse: outcome.response, + ...(outcome.turnUsage + ? { turnUsage: outcome.turnUsage } + : outcome.turnUsageUnavailable + ? { turnUsageUnavailable: true } + : {}) + }) + : undefined const terminalMessage = lastTurnMessage ?? transcript?.terminalMessage if (!terminalMessage) throw new Error('Completed delegated runtime has no terminal Message.') @@ -275,7 +296,12 @@ const createDurableDelegatedWork = ( terminalMessage }) } else { - await stageRuntimeTranscript({ terminalStatus: 'cancelled', endedAt }) + if (context) { + await stageRuntimeTranscript.settle(context, { + terminalStatus: 'cancelled', + endedAt + }) + } await options.records.terminalize({ frameId: child.frameId, attemptId: attempt.id, @@ -303,6 +329,7 @@ const createDurableDelegatedWork = ( attemptId: attempt.id, endedAt, error, + ...(context ? { lane: context } : {}), ...(cancelRequested ? { cancellationReason } : {}) }) } catch (terminalizeError) { @@ -314,13 +341,12 @@ const createDurableDelegatedWork = ( markEstablished() } } finally { - permissionOwner.clearAttempt(child.frameId, attempt.id) - await turnLifecycle.dispose() - await executionBackendClaim?.release().catch(() => undefined) - await reservation.release(slotId).catch(() => undefined) - if (running.get(child.frameId)?.attemptId === attempt.id) running.delete(child.frameId) + await finalize() } })() + // A detached delegate can outlive the caller that admitted it. Always observe the task so a + // late cleanup failure after an explicit Stop cannot become an unhandled rejection. + void completion.catch(() => undefined) running.set(child.frameId, { attemptId: attempt.id, completion, @@ -356,7 +382,8 @@ const createDurableDelegatedWork = ( rejectHandle(new Error('delegate execution was cancelled before message delivery')) await handle?.cancel() }, - executionStarted: () => handle !== undefined, + flushEvidence: () => stageRuntimeTranscript.flush(), + finalize, reservation, slotId }) @@ -492,40 +519,43 @@ const createDurableDelegatedWork = ( child: DurableChild, reason: 'main_agent_stop' | 'session_stop' | 'runtime_interrupted' ): Promise => { - await options.records.cancelQuestions(child.frameId, now(), 'Subagent was stopped.') const attempt = currentAttempt(child) - if (attempt.status !== 'running') { - return { frameId: child.frameId, status: 'already_terminal' } - } + const wasRunning = attempt.status === 'running' const snapshot = await options.records.snapshot() const session = snapshot.session const scope = { session, frameId: child.frameId, attemptId: attempt.id } const pendingPermissions = permissionOwner.takeAttempt(child.frameId, attempt.id) const evidenceScope = projectionOwner.attemptScope(snapshot, child, attempt) try { - if (evidenceScope) await options.artifactEvidence?.revoke?.(evidenceScope) const candidate = running.get(child.frameId) const active = candidate?.attemptId === attempt.id ? candidate : undefined - await active?.artifact?.dispose() - await options.revokeAttemptWrites?.(scope) - const executionStarted = active?.executionStarted() === true - await active?.cancel(reason).catch(() => undefined) - await options.settleAttemptCleanup?.(scope) - if (executionStarted) await active?.completion - const latest = await snapshotChild(child.frameId) - if (latest && currentAttempt(latest).status !== 'running') { - return currentAttempt(latest).status === 'cancelled' - ? { frameId: child.frameId, status: 'cancelled' } - : { frameId: child.frameId, status: 'already_terminal' } + // Calling these operations establishes their in-memory cancellation/revocation fences before + // the durable terminal transition. Their physical drains are deliberately detached: Stop is + // an ownership transition and must not wait forever for an external provider or resource. + cleanup.start(scope, 'execution cancellation', () => active?.cancel(reason)) + cleanup.start(scope, 'attempt write revocation', () => options.revokeAttemptWrites?.(scope)) + cleanup.start(scope, 'artifact evidence revocation', () => + evidenceScope ? options.artifactEvidence?.revoke?.(evidenceScope) : undefined + ) + cleanup.start(scope, 'attempt cleanup drain', () => options.settleAttemptCleanup?.(scope)) + cleanup.start(scope, 'attempt ownership release', () => active?.finalize(), true) + await active?.flushEvidence() + const stopped = await withAdmissionLock(() => + options.records.cancelAttempt({ + frameId: child.frameId, + attemptId: attempt.id, + endedAt: now(), + cancellationReason: reason, + questionReason: 'Subagent was stopped.' + }) + ) + if (stopped === 'already_terminal') { + const latest = await snapshotChild(child.frameId) + if (wasRunning && latest && currentAttempt(latest).status === 'cancelled') { + return { frameId: child.frameId, status: 'cancelled' } + } } - await options.records.terminalize({ - frameId: child.frameId, - attemptId: attempt.id, - status: 'cancelled', - endedAt: now(), - cancellationReason: reason - }) - return { frameId: child.frameId, status: 'cancelled' } + return { frameId: child.frameId, status: stopped } } catch (error) { const latest = await snapshotChild(child.frameId) if (latest && currentAttempt(latest).status !== 'running') { @@ -581,28 +611,18 @@ const createDurableDelegatedWork = ( const pinnedAttempt = currentAttempt(child) const candidate = running.get(child.frameId) const active = candidate?.attemptId === pinnedAttempt.id ? candidate : undefined - const settleBestEffort = async ( - operation: () => unknown | Promise - ): Promise => { - try { - await operation() - } catch (cleanupError) { - failures.push(cleanupError) - } - } - await settleBestEffort(() => active?.cancel(reason)) try { const cleanupSnapshot = await options.records.snapshot() const session = cleanupSnapshot.session const scope = { session, frameId: child.frameId, attemptId: pinnedAttempt.id } const evidenceScope = projectionOwner.attemptScope(cleanupSnapshot, child, pinnedAttempt) - await settleBestEffort(() => + cleanup.start(scope, 'execution cancellation', () => active?.cancel(reason)) + cleanup.start(scope, 'artifact evidence revocation', () => evidenceScope ? options.artifactEvidence?.revoke?.(evidenceScope) : undefined ) - await settleBestEffort(() => active?.artifact?.dispose()) - await settleBestEffort(() => options.revokeAttemptWrites?.(scope)) - await settleBestEffort(() => options.settleAttemptCleanup?.(scope)) - await settleBestEffort(() => active?.completion) + cleanup.start(scope, 'attempt write revocation', () => options.revokeAttemptWrites?.(scope)) + cleanup.start(scope, 'attempt cleanup drain', () => options.settleAttemptCleanup?.(scope)) + cleanup.start(scope, 'attempt ownership release', () => active?.finalize(), true) const latest = await snapshotChild(child.frameId) if ( latest && @@ -983,9 +1003,11 @@ const createDurableDelegatedWork = ( if (attempt.status !== 'running') continue const scope = { session: snapshot.session, frameId: child.frameId, attemptId: attempt.id } const evidenceScope = projectionOwner.attemptScope(snapshot, child, attempt) - if (evidenceScope) await options.artifactEvidence?.revoke?.(evidenceScope) - await options.revokeAttemptWrites?.(scope) - await options.settleAttemptCleanup?.(scope) + cleanup.start(scope, 'artifact evidence revocation', () => + evidenceScope ? options.artifactEvidence?.revoke?.(evidenceScope) : undefined + ) + cleanup.start(scope, 'attempt write revocation', () => options.revokeAttemptWrites?.(scope)) + cleanup.start(scope, 'attempt cleanup drain', () => options.settleAttemptCleanup?.(scope)) try { await options.records.terminalize({ frameId: child.frameId, diff --git a/src/main/delegation/in-memory-delegated-work-records.ts b/src/main/delegation/in-memory-delegated-work-records.ts index c2c4cfdfa..4edf818a0 100644 --- a/src/main/delegation/in-memory-delegated-work-records.ts +++ b/src/main/delegation/in-memory-delegated-work-records.ts @@ -303,6 +303,28 @@ const createInMemoryDelegatedWorkRecords = (input: { : request ) }, + async cancelAttempt(input) { + const child = state.records.find((candidate) => candidate.frameId === input.frameId) + const attempt = child && currentAttempt(child) + let cancelledQuestion = false + state.questionRequests = state.questionRequests.map((request) => { + if (request.sourceFrameId !== input.frameId || request.status !== 'pending') return request + cancelledQuestion = true + return { + ...request, + status: 'cancelled' as const, + respondedAt: input.endedAt, + failure: { code: 'cancelled', message: input.questionReason } + } + }) + if (!attempt || attempt.id !== input.attemptId || attempt.status !== 'running') { + return cancelledQuestion ? 'cancelled' : 'already_terminal' + } + attempt.status = 'cancelled' + attempt.endedAt = input.endedAt + attempt.cancellationReason = input.cancellationReason + return 'cancelled' + }, async startRuntime(frameId, attemptId, runtimeSegmentId) { findRunning(frameId, attemptId).runtimeSegmentIds.push(runtimeSegmentId) const child = state.records.find((candidate) => candidate.frameId === frameId)! @@ -319,15 +341,22 @@ const createInMemoryDelegatedWorkRecords = (input: { } }, async stageTerminalMessage(frameId, attemptId, message) { - findRunning(frameId, attemptId) + const child = state.records.find((candidate) => candidate.frameId === frameId) + const attempt = child?.attempts.find((candidate) => candidate.id === attemptId) + if (!attempt) { + throw new Error('Terminal Message provenance is outside the delegated Attempt.') + } if (message.frameId !== frameId || message.role !== 'assistant') { throw new Error('Terminal Message does not belong to the delegated Attempt.') } - const existing = state.messages.find((candidate) => candidate.id === message.id) - if (existing && JSON.stringify(existing) !== JSON.stringify(message)) { - throw new Error('Terminal Message identity is already in use.') - } - if (!existing) state.messages.push({ ...message }) + const existingIndex = state.messages.findIndex((candidate) => candidate.id === message.id) + if (existingIndex >= 0) { + const existing = state.messages[existingIndex] + if (existing.frameId !== frameId || existing.role !== 'assistant') { + throw new Error('Terminal Message identity is already in use.') + } + state.messages[existingIndex] = structuredClone(message) + } else state.messages.push(structuredClone(message)) }, async terminalize(terminal) { const attempt = findRunning(terminal.frameId, terminal.attemptId) diff --git a/src/main/delegation/production-composition.test.ts b/src/main/delegation/production-composition.test.ts index f7aaa8812..57f984d7b 100644 --- a/src/main/delegation/production-composition.test.ts +++ b/src/main/delegation/production-composition.test.ts @@ -100,7 +100,7 @@ const createCompositionHarness = async ( admissionError?: Error, owners: Pick< ProductionDelegatedWorkOptions, - 'artifactEvidence' | 'reviewEvidence' | 'parentMessages' + 'artifactEvidence' | 'reviewEvidence' | 'parentMessages' | 'onCleanupError' > = {}, initialRootInvocations: readonly Readonly<{ rootMessageId: string @@ -2492,9 +2492,10 @@ describe('production delegated-work composition', () => { expect(harness.execution.reservationCounts()).toEqual([]) }) - it('production-composes branch Stop partial failure without rolling back successful targets', async () => { + it('production-composes non-blocking branch Stop while reporting detached cleanup failure', async () => { root = await mkdtemp(join(tmpdir(), 'delegated-production-partial-stop-')) const handles = new Map() + const cleanupErrors: unknown[] = [] let failedOnce = false const harness = await createCompositionHarness(root, 'codex', undefined, undefined, { artifactEvidence: { @@ -2530,7 +2531,8 @@ describe('production delegated-work composition', () => { async project() { return [] } - } + }, + onCleanupError: (_scope, error) => cleanupErrors.push(error) }) const receipt = await harness.composition.host.delegate( harness.caller, @@ -2542,21 +2544,19 @@ describe('production delegated-work composition', () => { ) await expect.poll(() => harness.execution.controls()).toHaveLength(2) - await expect(harness.composition.root.stopActiveBranch?.(harness.session.id)).rejects.toThrow( - 'could not be stopped' - ) - const statusesAfterFailure = harness - .durable() - .runtimeContext!.delegatedWork!.records.map((record) => record.attempts.at(-1)!.status) - expect(statusesAfterFailure.sort()).toEqual(['cancelled', 'running']) await expect( harness.composition.root.stopActiveBranch?.(harness.session.id) ).resolves.toBeUndefined() + await expect.poll(() => cleanupErrors).toHaveLength(1) + expect(cleanupErrors[0]).toBeInstanceOf(AggregateError) expect( harness .durable() .runtimeContext!.delegatedWork!.records.map((record) => record.attempts.at(-1)!.status) ).toEqual(['cancelled', 'cancelled']) + await expect( + harness.composition.root.stopActiveBranch?.(harness.session.id) + ).resolves.toBeUndefined() expect(receipt.children).toHaveLength(2) }) }) diff --git a/src/main/delegation/production-composition.ts b/src/main/delegation/production-composition.ts index c53e0eb6c..26cff8426 100644 --- a/src/main/delegation/production-composition.ts +++ b/src/main/delegation/production-composition.ts @@ -64,6 +64,10 @@ type ProductionDelegatedWorkOptions = Readonly<{ deliver(delivery: ParentMessageDelivery): Promise }> onAgentRuntimeUpdate?(update: AcpAgentRuntimeUpdate): void + onCleanupError?( + scope: Readonly<{ session: SessionKey; frameId: string; attemptId: string }>, + error: unknown + ): void resolveExecutionModel(session: PersistedChatSession): Promise }> @@ -220,6 +224,7 @@ const createProductionDelegatedWorkComposition = ( deliverToParent: options.parentMessages?.deliver, onRootPermissionEvent: (event) => observePermission(key, event), onAgentRuntimeUpdate: options.onAgentRuntimeUpdate, + onCleanupError: options.onCleanupError, assertTurnOpen: (session, messageId) => { if ( cancelledTurns.has(cancelledTurnKey(session, messageId)) || diff --git a/src/main/delegation/session-record-adapter.test.ts b/src/main/delegation/session-record-adapter.test.ts index 65f95c8c1..9e460ccf4 100644 --- a/src/main/delegation/session-record-adapter.test.ts +++ b/src/main/delegation/session-record-adapter.test.ts @@ -816,7 +816,7 @@ describe('Session delegated-work adapter', () => { ).not.toHaveProperty('delegatedContext') }) - it('persists rich terminal transcript evidence without writing each runtime chunk', async () => { + it('uses one durable Message identity from live chunks through normal completion', async () => { const { coordinator, readSession, repository } = createHarness() const execution = createDeterministicDelegateExecution() const rootFrameId = createSession().conversationGraph!.rootFrameId @@ -891,6 +891,24 @@ describe('Session delegated-work adapter', () => { } } }) + await expect + .poll(async () => { + const session = await readSession() + return session.conversationGraph?.messages.find((message) => message.id === 'agent-message') + ?.content + }) + .toBe('Evidence confirmed.') + const live = await readSession() + const liveMessages = live.conversationGraph!.messages.filter( + (message) => message.agentFrameId === 'child-frame' && message.role === 'agent' + ) + expect(liveMessages).toHaveLength(1) + expect(liveMessages[0]).toMatchObject({ + id: 'agent-message', + status: 'streaming', + eventIds: ['message:1', 'message:2'] + }) + expect(liveMessages[0]).not.toHaveProperty('completedAt') control.emit({ kind: 'runtime', update: { @@ -931,6 +949,11 @@ describe('Session delegated-work adapter', () => { await pending const durable = await readSession() + expect( + durable.conversationGraph?.messages.filter( + (message) => message.agentFrameId === 'child-frame' && message.role === 'agent' + ) + ).toHaveLength(1) expect( durable.conversationGraph?.messages.find( (message) => message.id === 'agent-message' && message.role === 'agent' diff --git a/src/main/delegation/session-record-adapter.ts b/src/main/delegation/session-record-adapter.ts index 304933b25..4fda0cdc9 100644 --- a/src/main/delegation/session-record-adapter.ts +++ b/src/main/delegation/session-record-adapter.ts @@ -153,6 +153,16 @@ const createSessionDelegatedWorkRecords = ( }) ) }, + async cancelAttempt(input) { + const outcome = await mutate((expectedRevision) => + options.commands.transitionAttempt(key, { + expectedRevision, + ...input, + status: 'cancelled' + }) + ) + return outcome === 'transitioned' ? 'cancelled' : 'already_terminal' + }, async startRuntime(frameId, attemptId, runtimeSegmentId) { const attempt = (await load()).runtimeContext?.delegatedWork?.records .find((record) => record.agentFrameId === frameId) @@ -202,9 +212,10 @@ const createSessionDelegatedWorkRecords = ( expectedRevision, frameId, attemptId, + allowTerminalEvidence: true, event: { kind: 'message', - runtimeSegmentId: attempt?.runtimeSegmentIds.at(-1) ?? '', + runtimeSegmentId: message.runtimeSegmentId ?? attempt?.runtimeSegmentIds.at(-1) ?? '', message: { id: message.id, role: 'agent', @@ -216,7 +227,7 @@ const createSessionDelegatedWorkRecords = ( turnUsage: message.turnUsage ? { ...message.turnUsage } : undefined, turnUsageUnavailable: message.turnUsageUnavailable, createdAt: message.createdAt, - completedAt: message.completedAt ?? message.updatedAt ?? message.createdAt, + ...(message.completedAt !== undefined ? { completedAt: message.completedAt } : {}), updatedAt: message.updatedAt ?? message.createdAt } } @@ -237,6 +248,7 @@ const createSessionDelegatedWorkRecords = ( expectedRevision, frameId, attemptId, + allowTerminalEvidence: true, event: { kind: 'activity', runtimeSegmentId, @@ -253,8 +265,10 @@ const createSessionDelegatedWorkRecords = ( expectedRevision, frameId, attemptId, + allowTerminalEvidence: true, event: { kind: 'activity-group', + runtimeSegmentId, promptMessageId: activityGroup.promptMessageId!, activityGroup } @@ -277,6 +291,7 @@ const createSessionDelegatedWorkRecords = ( expectedRevision, frameId: input.frameId, attemptId: input.attemptId, + allowTerminalEvidence: true, event: { kind: 'message', runtimeSegmentId: attempt?.runtimeSegmentIds.at(-1) ?? '', diff --git a/src/main/delegation/session-records.ts b/src/main/delegation/session-records.ts index 689658859..1dd6c23da 100644 --- a/src/main/delegation/session-records.ts +++ b/src/main/delegation/session-records.ts @@ -118,6 +118,7 @@ type AttemptAgentEvent = | Readonly<{ kind: 'activity-group' activityGroup: PersistedActivityGroup + runtimeSegmentId?: string promptMessageId: string }> @@ -126,6 +127,7 @@ type AttemptAgentEventInput = Readonly<{ frameId: string attemptId: string event: AttemptAgentEvent + allowTerminalEvidence?: true }> type TransitionAttemptInput = Readonly<{ @@ -136,6 +138,7 @@ type TransitionAttemptInput = Readonly<{ endedAt: number terminalMessageId?: string cancellationReason?: DelegatedWorkCancellationReason + questionReason?: string error?: Readonly<{ code: string; message: string }> }> @@ -216,7 +219,10 @@ type DelegatedWorkRecordCommands = Readonly<{ ): Promise startAttemptRuntime(key: SessionKey, input: StartAttemptRuntimeInput): Promise applyAgentEvent(key: SessionKey, input: AttemptAgentEventInput): Promise - transitionAttempt(key: SessionKey, input: TransitionAttemptInput): Promise + transitionAttempt( + key: SessionKey, + input: TransitionAttemptInput + ): Promise<'transitioned' | 'already_terminal'> admitMessageCommand( key: SessionKey, input: AdmitMessageCommandInput diff --git a/src/main/notebook/local-rpc-server.delegated-work.test.ts b/src/main/notebook/local-rpc-server.delegated-work.test.ts index ad1a27eb3..530031335 100644 --- a/src/main/notebook/local-rpc-server.delegated-work.test.ts +++ b/src/main/notebook/local-rpc-server.delegated-work.test.ts @@ -3,6 +3,7 @@ import { afterEach, describe, expect, it, vi } from 'vitest' import { createDeterministicDelegateExecution } from '../delegation/deterministic-execution' import { createInMemoryDelegatedWorkRecords } from '../delegation/durable-delegated-work' import { createTestDurableDelegatedWork as createDurableDelegatedWork } from '../delegation/durable-delegated-work-test-fixture' +import type { DelegateExecution } from '../delegation/execution-port' import { NotebookLocalRpcServer } from './local-rpc-server' let server: NotebookLocalRpcServer | undefined @@ -509,6 +510,92 @@ describe('authenticated delegatedWorkCall route', () => { connection.release() }) + it('returns a successful child stop response while provider and lease cleanup remain hung', async () => { + const session = { projectId: 'trusted-project', sessionId: 'trusted-session' } + const neverSettles = new Promise(() => undefined) + const cancel = vi.fn(() => neverSettles) + const release = vi.fn(() => neverSettles) + const execution: DelegateExecution = { + async reserve() { + return { slotIds: ['hung-rpc-slot'], release, releaseAll: release } + }, + run() { + return { + accepted: Promise.resolve('provider_prompt_accepted'), + completion: new Promise(() => undefined), + subscribe: () => () => undefined, + sendMessage: async () => 'provider_prompt_accepted', + setPermissionProfile: async () => undefined, + respondToPermission: async () => undefined, + cancel + } + } + } + const records = createInMemoryDelegatedWorkRecords({ + session, + rootFrameId: 'trusted-root-frame', + originMessageId: 'trusted-origin-message' + }) + const work = createDurableDelegatedWork({ execution, records }) + server = new NotebookLocalRpcServer({ execute: async () => ({}) } as never, { + transport: 'tcp', + delegatedWorkService: work + }) + const connection = await server.issueControlConnection( + session.sessionId, + session.projectId, + 'trusted-root-frame' + ) + const endInvocation = connection.beginControlInvocation({ + turnId: 'turn-1', + controlInvocationGeneration: 1, + toolInvocationId: 'trusted-tool-call', + originatingUserMessageId: 'trusted-origin-message' + }) + const receipt = await work.delegate( + { + session, + frameId: 'trusted-root-frame', + role: 'main', + originMessageId: 'trusted-origin-message', + toolInvocationId: 'direct-admission' + }, + { task: 'Stop hung runtime through host', name: 'Stop hung runtime through host' }, + { wait: false } + ) + await expect + .poll(async () => (await records.snapshot()).records[0]?.attempts[0]?.runtimeSegmentIds) + .toHaveLength(1) + + const response = await fetch(connection.endpoint, { + method: 'POST', + headers: { + authorization: `Bearer ${connection.token}`, + 'content-type': 'application/json' + }, + body: JSON.stringify({ + method: 'delegatedWorkCall', + params: { + operation: 'stop_children', + frame_ids: [receipt.children[0].frameId] + } + }) + }) + + expect(response.status).toBe(200) + await expect(response.json()).resolves.toEqual({ + result: [{ frameId: receipt.children[0].frameId, status: 'cancelled' }] + }) + expect(cancel).toHaveBeenCalledOnce() + expect(release).toHaveBeenCalledOnce() + await expect(work.sessionSummary(session)).resolves.toMatchObject({ + runningCount: 0, + children: [{ status: 'cancelled' }] + }) + endInvocation() + connection.release() + }) + it('fails closed before reservation and durable mutation when the framework is unavailable', async () => { const session = { projectId: 'project-1', sessionId: 'session-1' } const execution = createDeterministicDelegateExecution() diff --git a/src/main/session-persistence/coordinator.ts b/src/main/session-persistence/coordinator.ts index 0992d2e03..a4aadf77d 100644 --- a/src/main/session-persistence/coordinator.ts +++ b/src/main/session-persistence/coordinator.ts @@ -578,7 +578,10 @@ class SessionPersistenceCoordinator implements DelegatedWorkRecordCommands { return this.delegatedWorkOwner.applyAgentEvent(key, input) } - transitionAttempt(key: SessionKey, input: TransitionAttemptInput): Promise { + transitionAttempt( + key: SessionKey, + input: TransitionAttemptInput + ): Promise<'transitioned' | 'already_terminal'> { return this.delegatedWorkOwner.transitionAttempt(key, input) } diff --git a/src/main/session-persistence/delegated-work-owner.ts b/src/main/session-persistence/delegated-work-owner.ts index ba92feee3..aec3becbe 100644 --- a/src/main/session-persistence/delegated-work-owner.ts +++ b/src/main/session-persistence/delegated-work-owner.ts @@ -312,17 +312,59 @@ class SessionDelegatedWorkPersistenceOwner implements DelegatedWorkRecordCommand } applyAgentEvent(key: SessionKey, input: AttemptAgentEventInput): Promise { + return this.persistAgentEvent( + key, + input, + input.allowTerminalEvidence ? 'current-attempt-evidence' : 'running' + ) + } + + private persistAgentEvent( + key: SessionKey, + input: AttemptAgentEventInput, + mode: 'running' | 'current-attempt-evidence' + ): Promise { return this.store.mutate(key, input.expectedRevision, (graph, records) => { - assertCurrentRunningAttempt(records, input.frameId, input.attemptId) + const record = records.find((candidate) => candidate.agentFrameId === input.frameId) + const attempt = + mode === 'current-attempt-evidence' + ? record && currentAttempt(record) + : assertCurrentRunningAttempt(records, input.frameId, input.attemptId).attempt + if (!record || !attempt) throw new DelegatedWorkAttemptConflictError() + if (attempt.id !== input.attemptId) throw new DelegatedWorkAttemptConflictError() const frame = graph.frames.find((candidate) => candidate.id === input.frameId) if (!frame) throw new Error(`Delegate Frame not found: ${input.frameId}`) - const branch = graph.branches.find((candidate) => candidate.id === frame.activeBranchId) - if (!branch) throw new Error(`Delegate Branch not found: ${frame.activeBranchId}`) const event = input.event + const runtimeSegmentId = + event.runtimeSegmentId ?? + (mode === 'running' ? attempt.runtimeSegmentIds.at(-1) : undefined) + if (!runtimeSegmentId || !attempt.runtimeSegmentIds.includes(runtimeSegmentId)) { + throw new Error('Agent event Runtime Segment is outside the Attempt.') + } + const promptMessageId = + (event.kind === 'message' ? event.message.responseToMessageId : event.promptMessageId) ?? + (mode === 'running' + ? graph.branches.find((candidate) => candidate.id === frame.activeBranchId)?.headMessageId + : undefined) + const promptMessage = graph.messages.find( + (message) => + message.id === promptMessageId && + message.agentFrameId === input.frameId && + message.runtimeSegmentId === runtimeSegmentId + ) + if (!promptMessage?.introducedOnBranchId) { + throw new Error('Agent event prompt provenance is outside the Attempt Frame.') + } + const branch = graph.branches.find( + (candidate) => + candidate.id === promptMessage.introducedOnBranchId && + candidate.agentFrameId === input.frameId + ) + if (!branch) throw new Error('Agent event Branch is outside the Attempt Frame.') if (event.kind === 'message') { const segment = graph.runtimeSegments.find( (candidate) => - candidate.id === event.runtimeSegmentId && candidate.agentFrameId === input.frameId + candidate.id === runtimeSegmentId && candidate.agentFrameId === input.frameId ) if (!segment) throw new Error('Agent event Runtime Segment is outside the Attempt Frame.') const nextMessage: PersistedConversationGraph['messages'][number] = { @@ -331,12 +373,21 @@ class SessionDelegatedWorkPersistenceOwner implements DelegatedWorkRecordCommand introducedOnBranchId: branch.id, ...(branch.headMessageId ? { parentMessageId: branch.headMessageId } : {}), ...(event.message.role === 'user' ? { revisionRootMessageId: event.message.id } : {}), - runtimeSegmentId: event.runtimeSegmentId + runtimeSegmentId } const existing = graph.messages.find((message) => message.id === event.message.id) if (existing) { if (JSON.stringify(existing) === JSON.stringify(nextMessage)) return - throw new Error(`Message already exists: ${event.message.id}`) + if ( + existing.agentFrameId !== input.frameId || + existing.runtimeSegmentId !== runtimeSegmentId || + existing.role !== nextMessage.role + ) { + throw new Error(`Message already exists: ${event.message.id}`) + } + nextMessage.parentMessageId = existing.parentMessageId + Object.assign(existing, nextMessage) + return } graph.messages.push(nextMessage) branch.headMessageId = event.message.id @@ -348,8 +399,7 @@ class SessionDelegatedWorkPersistenceOwner implements DelegatedWorkRecordCommand message.id === event.promptMessageId && message.agentFrameId === input.frameId ) || !graph.runtimeSegments.some( - (segment) => - segment.id === event.runtimeSegmentId && segment.agentFrameId === input.frameId + (segment) => segment.id === runtimeSegmentId && segment.agentFrameId === input.frameId ) ) { throw new Error('Activity provenance is outside the Attempt Frame.') @@ -359,12 +409,19 @@ class SessionDelegatedWorkPersistenceOwner implements DelegatedWorkRecordCommand agentFrameId: input.frameId, messageBranchId: branch.id, promptMessageId: event.promptMessageId, - runtimeSegmentId: event.runtimeSegmentId + runtimeSegmentId } const existing = graph.activities.find((activity) => activity.id === event.activity.id) if (existing) { if (JSON.stringify(existing) === JSON.stringify(nextActivity)) return - throw new Error(`Activity already exists: ${event.activity.id}`) + if ( + existing.agentFrameId !== input.frameId || + existing.runtimeSegmentId !== runtimeSegmentId + ) { + throw new Error(`Activity already exists: ${event.activity.id}`) + } + Object.assign(existing, nextActivity) + return } graph.activities.push(nextActivity) } else { @@ -385,57 +442,94 @@ class SessionDelegatedWorkPersistenceOwner implements DelegatedWorkRecordCommand const existing = graph.activityGroups.find((group) => group.id === event.activityGroup.id) if (existing) { if (JSON.stringify(existing) === JSON.stringify(nextActivityGroup)) return - throw new Error(`Activity Group already exists: ${event.activityGroup.id}`) + if (existing.agentFrameId !== input.frameId) { + throw new Error(`Activity Group already exists: ${event.activityGroup.id}`) + } + Object.assign(existing, nextActivityGroup) + return } graph.activityGroups.push(nextActivityGroup) } }) } - transitionAttempt(key: SessionKey, input: TransitionAttemptInput): Promise { - return this.store.mutate(key, input.expectedRevision, (graph, records) => { - const { record, attempt } = assertCurrentRunningAttempt( + transitionAttempt( + key: SessionKey, + input: TransitionAttemptInput + ): Promise<'transitioned' | 'already_terminal'> { + return this.store.mutate( + key, + input.expectedRevision, + ( + graph, records, - input.frameId, - input.attemptId - ) - if (input.endedAt < attempt.startedAt) throw new Error('Attempt end precedes its start.') - if (input.status === 'completed' && !input.terminalMessageId) { - throw new Error('A completed Attempt requires a terminal Message.') - } - if (input.status === 'cancelled' && !input.cancellationReason) { - throw new Error('A cancelled Attempt requires a cancellation reason.') - } - if (input.status === 'error' && !input.error) { - throw new Error('An errored Attempt requires error detail.') - } - if ( - input.terminalMessageId && - !graph.messages.some( - (message) => - message.id === input.terminalMessageId && message.agentFrameId === input.frameId - ) - ) { - throw new Error('Terminal Message is outside the Attempt Frame.') - } - const attempts = record.attempts as DelegatedWorkAttemptRecord[] - attempts[attempts.length - 1] = { - ...attempt, - status: input.status, - endedAt: input.endedAt, - ...(input.terminalMessageId ? { terminalMessageId: input.terminalMessageId } : {}), - ...(input.cancellationReason ? { cancellationReason: input.cancellationReason } : {}), - ...(input.error ? { error: input.error } : {}) - } - const frame = graph.frames.find((candidate) => candidate.id === input.frameId) - if (!frame) throw new Error(`Delegate Frame not found: ${input.frameId}`) - frame.status = input.status - frame.completedAt = input.endedAt - for (const segmentId of attempt.runtimeSegmentIds) { - const segment = graph.runtimeSegments.find((candidate) => candidate.id === segmentId) - if (segment && segment.endedAt === undefined) segment.endedAt = input.endedAt + _session, + _commands, + _messagesQuarantined, + questions, + questionsQuarantined + ) => { + let cancelledQuestion = false + if (input.questionReason !== undefined) { + if (questionsQuarantined) throw new Error('Delegated question owner is quarantined.') + for (const [index, request] of questions.entries()) { + if (request.sourceFrameId !== input.frameId || request.status !== 'pending') continue + cancelledQuestion = true + questions[index] = { + ...request, + status: 'cancelled', + respondedAt: input.endedAt, + failure: { code: 'cancelled', message: input.questionReason } + } + } + } + const record = records.find((candidate) => candidate.agentFrameId === input.frameId) + const attempt = record && currentAttempt(record) + if (!record || !attempt || attempt.id !== input.attemptId || attempt.status !== 'running') { + if (input.questionReason !== undefined) { + return cancelledQuestion ? 'transitioned' : 'already_terminal' + } + throw new DelegatedWorkAttemptConflictError() + } + if (input.endedAt < attempt.startedAt) throw new Error('Attempt end precedes its start.') + if (input.status === 'completed' && !input.terminalMessageId) { + throw new Error('A completed Attempt requires a terminal Message.') + } + if (input.status === 'cancelled' && !input.cancellationReason) { + throw new Error('A cancelled Attempt requires a cancellation reason.') + } + if (input.status === 'error' && !input.error) { + throw new Error('An errored Attempt requires error detail.') + } + if ( + input.terminalMessageId && + !graph.messages.some( + (message) => + message.id === input.terminalMessageId && message.agentFrameId === input.frameId + ) + ) { + throw new Error('Terminal Message is outside the Attempt Frame.') + } + const attempts = record.attempts as DelegatedWorkAttemptRecord[] + attempts[attempts.length - 1] = { + ...attempt, + status: input.status, + endedAt: input.endedAt, + ...(input.terminalMessageId ? { terminalMessageId: input.terminalMessageId } : {}), + ...(input.cancellationReason ? { cancellationReason: input.cancellationReason } : {}), + ...(input.error ? { error: input.error } : {}) + } + const frame = graph.frames.find((candidate) => candidate.id === input.frameId) + if (!frame) throw new Error(`Delegate Frame not found: ${input.frameId}`) + frame.status = input.status + frame.completedAt = input.endedAt + for (const segmentId of attempt.runtimeSegmentIds) { + const segment = graph.runtimeSegments.find((candidate) => candidate.id === segmentId) + if (segment && segment.endedAt === undefined) segment.endedAt = input.endedAt + } + return 'transitioned' } - }) + ) } submitStructuredOutput( diff --git a/src/main/session-persistence/delegated-work-records.test.ts b/src/main/session-persistence/delegated-work-records.test.ts index 199d4e232..d30790b9b 100644 --- a/src/main/session-persistence/delegated-work-records.test.ts +++ b/src/main/session-persistence/delegated-work-records.test.ts @@ -596,7 +596,7 @@ describe('delegated-work Session records', () => { }) it('fences terminal Attempts and admits only one same-Frame continuation', async () => { - const { coordinator } = createHarness() + const { coordinator, durable } = createHarness() const rootFrameId = createRootSession().conversationGraph!.rootFrameId await coordinator.createChildren(key, { expectedRevision: 0, @@ -638,6 +638,50 @@ describe('delegated-work Session records', () => { endedAt: 22, terminalMessageId: 'terminal-1' }) + const lateEvidenceEvent = { + kind: 'message' as const, + runtimeSegmentId: 'segment-1', + message: { + id: 'late-evidence', + role: 'agent' as const, + content: 'Captured before transport closed.', + responseToMessageId: 'child-prompt-1', + status: 'complete' as const, + eventIds: ['late-event-1'], + createdAt: 24, + updatedAt: 24 + } + } + await expect( + coordinator.applyAgentEvent(key, { + expectedRevision: 4, + frameId: 'child-frame-1', + attemptId: 'attempt-1', + allowTerminalEvidence: true, + event: { + ...lateEvidenceEvent, + message: { ...lateEvidenceEvent.message, responseToMessageId: rootPrompt.id } + } + }) + ).rejects.toThrow('prompt provenance') + await expect( + coordinator.applyAgentEvent(key, { + expectedRevision: 4, + frameId: 'child-frame-1', + attemptId: 'attempt-1', + allowTerminalEvidence: true, + event: lateEvidenceEvent + }) + ).resolves.toBeUndefined() + await expect( + coordinator.applyAgentEvent(key, { + expectedRevision: 5, + frameId: 'child-frame-1', + attemptId: 'attempt-1', + allowTerminalEvidence: true, + event: lateEvidenceEvent + }) + ).resolves.toBeUndefined() const continuationCommand = ( suffix: string, attemptId: string @@ -664,7 +708,7 @@ describe('delegated-work Session records', () => { const attempts = await Promise.allSettled([ coordinator.startContinuationAttempt(key, { - expectedRevision: 4, + expectedRevision: 6, frameId: 'child-frame-1', previousAttemptId: 'attempt-1', attemptId: 'attempt-2', @@ -677,7 +721,7 @@ describe('delegated-work Session records', () => { messageCommand: continuationCommand('2', 'attempt-2') }), coordinator.startContinuationAttempt(key, { - expectedRevision: 4, + expectedRevision: 6, frameId: 'child-frame-1', previousAttemptId: 'attempt-1', attemptId: 'attempt-3', @@ -694,7 +738,7 @@ describe('delegated-work Session records', () => { expect(attempts.map(({ status }) => status).sort()).toEqual(['fulfilled', 'rejected']) await expect( coordinator.applyAgentEvent(key, { - expectedRevision: 5, + expectedRevision: 7, frameId: 'child-frame-1', attemptId: 'attempt-1', event: { @@ -712,6 +756,40 @@ describe('delegated-work Session records', () => { } }) ).rejects.toMatchObject({ code: 'attempt-conflict' }) + await expect( + coordinator.applyAgentEvent(key, { + expectedRevision: 7, + frameId: 'child-frame-1', + attemptId: 'attempt-1', + allowTerminalEvidence: true, + event: lateEvidenceEvent + }) + ).rejects.toMatchObject({ code: 'attempt-conflict' }) + expect(durable().conversationGraph?.messages).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + id: 'late-evidence', + content: 'Captured before transport closed.', + eventIds: ['late-event-1'] + }) + ]) + ) + expect(durable().runtimeContext?.delegatedWork?.records[0].attempts[0]).toMatchObject({ + id: 'attempt-1', + status: 'completed' + }) + const reopened = createHarness(durable()) + await expect(reopened.coordinator.readChildren(key, rootFrameId)).resolves.toMatchObject([ + { + frameId: 'child-frame-1', + record: { attempts: [{ id: 'attempt-1', status: 'completed' }, { status: 'running' }] } + } + ]) + expect(reopened.durable().conversationGraph?.messages).toEqual( + expect.arrayContaining([ + expect.objectContaining({ id: 'late-evidence', eventIds: ['late-event-1'] }) + ]) + ) await expect(coordinator.readChildren(key, rootFrameId)).resolves.toMatchObject([ { frameId: 'child-frame-1', @@ -721,6 +799,82 @@ describe('delegated-work Session records', () => { ]) }) + it('atomically cancels a pending question without rewriting its completed Attempt', async () => { + const { coordinator, durable } = createHarness() + const rootFrameId = durable().conversationGraph!.rootFrameId + await coordinator.createChildren(key, { + expectedRevision: 0, + parentFrameId: rootFrameId, + originMessageId: rootPrompt.id, + children: [child(1)] + }) + await coordinator.startAttemptRuntime(key, { + expectedRevision: 1, + frameId: 'child-frame-1', + attemptId: 'attempt-1', + runtimeSegmentId: 'segment-1', + frameworkId: 'codex', + startedAt: 20 + }) + const startedGraph = durable().conversationGraph! + const sourceBranchId = startedGraph.frames.find( + ({ id }) => id === 'child-frame-1' + )!.activeBranchId + const rootBranchId = startedGraph.frames.find(({ id }) => id === rootFrameId)!.activeBranchId + await coordinator.admitQuestion(key, { + expectedRevision: 2, + request: { + requestId: 'pending-question', + canonicalDigest: 'a'.repeat(64), + sourceFrameId: 'child-frame-1', + sourceAttemptId: 'attempt-1', + sourceRuntimeSegmentId: 'segment-1', + sourceMessageBranchId: sourceBranchId, + rootOriginMessageId: rootPrompt.id, + rootBranchId, + sourceName: 'Child 1', + questions: [ + { + header: 'Continue', + question: 'Continue?', + options: [ + { label: 'Yes', description: 'Continue the work.' }, + { label: 'No', description: 'Stop the work.' } + ] + } + ], + askedAt: 21, + status: 'pending', + draftAnswers: [], + draftQuestionIndex: 0 + } + }) + await coordinator.transitionAttempt(key, { + expectedRevision: 3, + frameId: 'child-frame-1', + attemptId: 'attempt-1', + status: 'error', + endedAt: 22, + error: { code: 'execution_failure', message: 'Waiting for user input.' } + }) + + await expect( + coordinator.transitionAttempt(key, { + expectedRevision: 4, + frameId: 'child-frame-1', + attemptId: 'attempt-1', + status: 'cancelled', + endedAt: 23, + cancellationReason: 'main_agent_stop', + questionReason: 'Subagent was stopped.' + }) + ).resolves.toBe('transitioned') + expect(durable().runtimeContext?.delegatedWork).toMatchObject({ + records: [{ attempts: [{ id: 'attempt-1', status: 'error' }] }], + questionRequests: [{ requestId: 'pending-question', status: 'cancelled' }] + }) + }) + it('does not publish partial child state when the atomic Session save fails', async () => { const { coordinator, repository, durable } = createHarness() const rootFrameId = durable().conversationGraph!.rootFrameId diff --git a/src/renderer/src/lib/acp/useWorkspaceAgentRuntime.architecture.test.ts b/src/renderer/src/lib/acp/useWorkspaceAgentRuntime.architecture.test.ts index c1dcda296..a983d41ca 100644 --- a/src/renderer/src/lib/acp/useWorkspaceAgentRuntime.architecture.test.ts +++ b/src/renderer/src/lib/acp/useWorkspaceAgentRuntime.architecture.test.ts @@ -82,7 +82,14 @@ const privateOwnerTargets = new Set(ownerTargets.values()) const subagentPresentationTarget = modulePath( resolve(__dirname, 'workspace-subagent-runtime-presentation') ) -const privateRuntimeTargets = new Set([...privateOwnerTargets, subagentPresentationTarget]) +const subagentTranscriptTarget = modulePath( + resolve(__dirname, 'workspace-subagent-runtime-transcript') +) +const privateRuntimeTargets = new Set([ + ...privateOwnerTargets, + subagentPresentationTarget, + subagentTranscriptTarget +]) const facadeTarget = modulePath(facadePath) const workspaceEventsTarget = modulePath(resolve(__dirname, 'workspace-events')) const resolveImportTarget = (sourcePath: string, specifier: string): string | undefined => { @@ -805,7 +812,8 @@ describe('workspace runtime architecture', () => { 'src/renderer/src/lib/acp/useWorkspaceAgentRuntime.ts', 'src/renderer/src/lib/acp/workspace-events.ts', ...ownerNames.map((name) => `src/renderer/src/lib/acp/${name}.ts`), - 'src/renderer/src/lib/acp/workspace-subagent-runtime-presentation.ts' + 'src/renderer/src/lib/acp/workspace-subagent-runtime-presentation.ts', + 'src/renderer/src/lib/acp/workspace-subagent-runtime-transcript.ts' ]) expect(workspaceRuntime.interfacePaths).toEqual([ 'src/renderer/src/lib/acp/useWorkspaceAgentRuntime.ts' diff --git a/src/renderer/src/lib/acp/workspace-subagent-runtime-presentation.ts b/src/renderer/src/lib/acp/workspace-subagent-runtime-presentation.ts index b73dcecc2..2c1b87c16 100644 --- a/src/renderer/src/lib/acp/workspace-subagent-runtime-presentation.ts +++ b/src/renderer/src/lib/acp/workspace-subagent-runtime-presentation.ts @@ -1,74 +1,29 @@ -import { useEffect, useRef, useState } from 'react' +import { useEffect, useLayoutEffect, useRef, useState } from 'react' import { useStore } from 'zustand' import type { AcpAgentRuntimeUpdate, AcpRuntimeEvent } from '../../../../shared/acp' -import type { - DelegatedWorkAttemptRecord, - PersistedChatMessage -} from '../../../../shared/session-persistence' -import { createSessionStore, type ChatSession } from '../../stores/session-store' +import type { ChatSession } from '../../stores/session-store' +import { createSessionStore } from '../../stores/session-store' import { applyRuntimePresentationEvent, createRuntimePresentationContext } from './runtime-event-presentation' - -type WorkspaceSubagentFrameProjection = Readonly<{ - frameId: string - status: 'running' | 'awaiting_user' | 'completed' | 'cancelled' | 'error' - attempt?: DelegatedWorkAttemptRecord - messages: readonly PersistedChatMessage[] -}> +import { + childConversationSession, + fenceTerminalLifecycle, + reconcileDurableChildProjection, + type WorkspaceSubagentFrameProjection +} from './workspace-subagent-runtime-transcript' type SubscribeToSubagentRuntimeUpdates = ( listener: (update: AcpAgentRuntimeUpdate) => void ) => () => void -const childConversationSession = ( - session: ChatSession, - detail: WorkspaceSubagentFrameProjection -): ChatSession => { - const messages = [...detail.messages] - const promptMessage = messages.findLast((message) => message.role === 'user') - const running = detail.status === 'running' && detail.attempt?.status === 'running' - - return { - ...session, - status: detail.status === 'running' ? 'running' : detail.status === 'error' ? 'error' : 'idle', - error: detail.attempt?.error?.message, - activeRun: - running && promptMessage - ? { promptMessageId: promptMessage.id, startedAt: detail.attempt.startedAt } - : undefined, - agentPromptInFlight: running ? true : undefined, - messages, - conversationGraph: session.conversationGraph - ? { ...session.conversationGraph, activeFrameId: detail.frameId } - : undefined, - // This store is an isolated presentation projection. Authority remains in the root Session - // graph; the adapter only lets the existing transcript components render the selected Frame. - activities: session.conversationGraph?.activities - .filter((activity) => activity.agentFrameId === detail.frameId) - .map(({ agentFrameId, messageBranchId, runtimeSegmentId, ...activity }) => { - void agentFrameId - void messageBranchId - void runtimeSegmentId - return activity - }) as ChatSession['activities'], - activityGroups: session.conversationGraph?.activityGroups - .filter((group) => group.agentFrameId === detail.frameId) - .map(({ agentFrameId, messageBranchId, ...group }) => { - void agentFrameId - void messageBranchId - return group - }) - } -} - const isSelectedRuntimeUpdate = ( update: AcpAgentRuntimeUpdate, session: ChatSession, detail: WorkspaceSubagentFrameProjection, - runtimeSegmentId: string | undefined, + runtimeSegmentId: string, promptMessageId: string | undefined ): boolean => update.scope.projectId === session.projectId && @@ -99,21 +54,48 @@ const useSubagentRuntimePresentation = ( const processedEventIds = useRef(new Set()) const runtimeSegmentId = detail.attempt?.runtimeSegmentIds.at(-1) const promptMessageId = detail.messages.findLast((message) => message.role === 'user')?.id + const running = detail.status === 'running' && detail.attempt?.status === 'running' + const runtimeIdentity = + detail.attempt && runtimeSegmentId && promptMessageId + ? [ + session.projectId, + session.id, + detail.frameId, + detail.attempt.id, + runtimeSegmentId, + promptMessageId + ].join('\u0000') + : undefined + const currentRuntimeIdentity = useRef(runtimeIdentity) + const latestLifecycle = useRef({ + running, + terminalProjection: childConversationSession(session, detail) + }) + useLayoutEffect(() => { + currentRuntimeIdentity.current = runtimeIdentity + latestLifecycle.current = { + running, + terminalProjection: childConversationSession(session, detail) + } + }, [detail, running, runtimeIdentity, session]) const liveSession = useStore(store, (state) => state.sessions[0]) - // Runtime updates are ephemeral, so a subscription can miss an event while the selected detail - // is mounting or being replaced. Reconcile every newer durable projection into the isolated - // store; the store's identity merge preserves already-applied live events until durability - // catches up, while a terminal projection advances status and the transcript authoritatively. useEffect(() => { - store.getState().upsertPersistedSession(childConversationSession(session, detail)) - }, [detail, session, store]) + reconcileDurableChildProjection( + store, + childConversationSession(session, detail), + running, + detail.status, + runtimeSegmentId + ) + }, [detail, running, runtimeSegmentId, session, store]) useEffect(() => { - if (!runtimeSegmentId) return + if (!runtimeIdentity || !runtimeSegmentId) return return subscribe((update) => { if ( + currentRuntimeIdentity.current !== runtimeIdentity || !isSelectedRuntimeUpdate(update, session, detail, runtimeSegmentId, promptMessageId) || processedEventIds.current.has(update.event.id) ) { @@ -125,21 +107,37 @@ const useSubagentRuntimePresentation = ( sessionId: session.id, promptMessageId: update.scope.promptMessageId } as AcpRuntimeEvent + const lifecycle = latestLifecycle.current + const appliedPresentation = applyRuntimePresentationEvent(event, store, presentationContext) - if (applyRuntimePresentationEvent(event, store, presentationContext)) return - if (event.kind === 'stop') { + if (!appliedPresentation && event.kind === 'stop') { presentationContext.activityGroupToolCallIdsBySession.delete(session.id) store.getState().finishRun(session.id, event.turnUsage, update.scope.promptMessageId) - } else if (event.kind === 'error') { + } else if (!appliedPresentation && event.kind === 'error') { presentationContext.activityGroupToolCallIdsBySession.delete(session.id) store .getState() .failRun(session.id, event.text?.trim() || event.title?.trim() || 'Agent run failed') - } else if (event.kind === 'system' && event.level === 'warning' && event.text) { + } else if ( + !appliedPresentation && + event.kind === 'system' && + event.level === 'warning' && + event.text + ) { store.getState().setAgentStatus(session.id, event.text) } + if (!lifecycle.running) fenceTerminalLifecycle(store, lifecycle.terminalProjection) }) - }, [detail, presentationContext, promptMessageId, runtimeSegmentId, session, store, subscribe]) + }, [ + detail, + presentationContext, + promptMessageId, + runtimeIdentity, + runtimeSegmentId, + session, + store, + subscribe + ]) return liveSession } diff --git a/src/renderer/src/lib/acp/workspace-subagent-runtime-transcript.ts b/src/renderer/src/lib/acp/workspace-subagent-runtime-transcript.ts new file mode 100644 index 000000000..29632e754 --- /dev/null +++ b/src/renderer/src/lib/acp/workspace-subagent-runtime-transcript.ts @@ -0,0 +1,329 @@ +import type { + DelegatedWorkAttemptRecord, + PersistedChatMessage +} from '../../../../shared/session-persistence' +import { createSessionStore, type ChatSession } from '../../stores/session-store' +type WorkspaceSubagentFrameProjection = Readonly<{ + frameId: string + status: 'running' | 'awaiting_user' | 'completed' | 'cancelled' | 'error' + attempt?: DelegatedWorkAttemptRecord + messages: readonly PersistedChatMessage[] +}> + +type SubagentPresentationStore = ReturnType + +const hasSharedEventIdentity = ( + left: Readonly<{ eventIds: readonly string[] }>, + right: Readonly<{ eventIds: readonly string[] }> +): boolean => left.eventIds.some((eventId) => right.eventIds.includes(eventId)) + +const isSameMessageIdentity = ( + left: ChatSession['messages'][number], + right: ChatSession['messages'][number] +): boolean => + left.id === right.id || + (Boolean(left.streamId) && left.streamId === right.streamId) || + hasSharedEventIdentity(left, right) + +const mergeAcceptedProjectionItems = ( + accepted: readonly Item[], + durable: readonly Item[], + isSameIdentity: (left: Item, right: Item) => boolean, + retainAccepted: (item: Item) => boolean = () => true, + resolveConflict: (accepted: Item, durable: Item) => Item = (_accepted, durableItem) => durableItem +): Item[] => { + const remainingDurable = [...durable] + const merged = accepted.flatMap((item) => { + const durableIndex = remainingDurable.findIndex((candidate) => isSameIdentity(item, candidate)) + if (durableIndex >= 0) { + const [durableItem] = remainingDurable.splice(durableIndex, 1) + return [resolveConflict(item, durableItem)] + } + return retainAccepted(item) ? [item] : [] + }) + return [...merged, ...remainingDurable] +} + +const hasVisibleMessageContent = (message: ChatSession['messages'][number]): boolean => + Boolean( + message.content.trim() || + message.images?.length || + message.artifactIds?.length || + message.uploads?.length + ) + +const appendUnique = (items: readonly Item[], additional: readonly Item[]): Item[] => [ + ...new Set([...items, ...additional]) +] + +const mergeMessageConflict = ( + accepted: ChatSession['messages'][number], + durable: ChatSession['messages'][number] +): ChatSession['messages'][number] => { + const uncoveredEventIds = accepted.eventIds.filter( + (eventId) => !durable.eventIds.includes(eventId) + ) + if (uncoveredEventIds.length === 0) return durable + const acceptedImages = accepted.images?.filter((image) => uncoveredEventIds.includes(image.id)) + return { + ...accepted, + ...durable, + // A runtime stream is cumulative. If durability covers only a prefix of its event identities, + // the accepted candidate still owns the visible tail while durable metadata owns the row. + content: accepted.content, + eventIds: appendUnique(durable.eventIds, uncoveredEventIds), + images: + durable.images || acceptedImages + ? [ + ...(durable.images ?? []), + ...(acceptedImages ?? []).filter( + (image) => !durable.images?.some((candidate) => candidate.id === image.id) + ) + ] + : undefined, + artifactIds: appendUnique(durable.artifactIds ?? [], accepted.artifactIds ?? []), + updatedAt: Math.max(accepted.updatedAt, durable.updatedAt) + } +} + +const settleAcceptedMessages = ( + messages: ChatSession['messages'], + status: WorkspaceSubagentFrameProjection['status'] +): ChatSession['messages'] => { + if (status === 'awaiting_user') return messages + const lastAgentMessage = messages.findLast((message) => message.role === 'agent') + return messages.map((message) => + message.status !== 'streaming' + ? message + : { + ...message, + status: + status !== 'completed' && message === lastAgentMessage ? ('error' as const) : 'complete' + } + ) +} + +const runtimeItemAlias = (id: string, runtimeSegmentId: string | undefined): string => { + if (!runtimeSegmentId) return id + const prefix = `agent-runtime:${encodeURIComponent(runtimeSegmentId)}:` + if (!id.startsWith(prefix)) return id + try { + return decodeURIComponent(id.slice(prefix.length)) + } catch { + return id + } +} + +const mergeActivityConflict = ( + accepted: NonNullable[number], + durable: NonNullable[number] +): NonNullable[number] => { + const uncoveredEventIds = accepted.eventIds.filter( + (eventId) => !durable.eventIds.includes(eventId) + ) + if (uncoveredEventIds.length === 0) return durable + const acceptedIsTerminal = accepted.status === 'completed' || accepted.status === 'failed' + return { + ...durable, + // Uncovered events are a newer accepted tail. Retain their payload while the durable identity + // continues to own the canonical namespaced row and group linkage. + ...accepted, + id: durable.id, + activityGroupId: durable.activityGroupId ?? accepted.activityGroupId, + sortIndex: durable.sortIndex, + eventIds: appendUnique(durable.eventIds, uncoveredEventIds), + status: acceptedIsTerminal ? accepted.status : durable.status, + createdAt: Math.min(accepted.createdAt, durable.createdAt), + updatedAt: Math.max(accepted.updatedAt, durable.updatedAt) + } +} + +const settleAcceptedActivity = ( + activity: NonNullable[number], + status: WorkspaceSubagentFrameProjection['status'] +): NonNullable[number] => { + if ( + status === 'awaiting_user' || + (activity.status !== 'pending' && activity.status !== 'in_progress') + ) { + return activity + } + return { + ...activity, + status: status === 'completed' ? 'completed' : 'failed' + } +} + +const childConversationSession = ( + session: ChatSession, + detail: WorkspaceSubagentFrameProjection +): ChatSession => { + const messages = [...detail.messages] + const promptMessage = messages.findLast((message) => message.role === 'user') + const running = detail.status === 'running' && detail.attempt?.status === 'running' + + return { + ...session, + status: detail.status === 'running' ? 'running' : detail.status === 'error' ? 'error' : 'idle', + error: detail.attempt?.error?.message, + activeRun: + running && promptMessage + ? { promptMessageId: promptMessage.id, startedAt: detail.attempt.startedAt } + : undefined, + agentPromptInFlight: running ? true : undefined, + messages, + conversationGraph: session.conversationGraph + ? { ...session.conversationGraph, activeFrameId: detail.frameId } + : undefined, + // This store is an isolated presentation projection. Authority remains in the root Session + // graph; the adapter only lets the existing transcript components render the selected Frame. + activities: session.conversationGraph?.activities + .filter((activity) => activity.agentFrameId === detail.frameId) + .map(({ agentFrameId, messageBranchId, runtimeSegmentId, ...activity }) => { + void agentFrameId + void messageBranchId + void runtimeSegmentId + return activity + }) as ChatSession['activities'], + activityGroups: session.conversationGraph?.activityGroups + .filter((group) => group.agentFrameId === detail.frameId) + .map(({ agentFrameId, messageBranchId, ...group }) => { + void agentFrameId + void messageBranchId + return group + }) + } +} + +const reconcileDurableChildProjection = ( + store: SubagentPresentationStore, + projection: ChatSession, + running: boolean, + status: WorkspaceSubagentFrameProjection['status'], + runtimeSegmentId: string | undefined +): void => { + if (running) { + store.getState().upsertPersistedSession(projection) + return + } + + // The selected child Frame owns lifecycle state in this isolated store, so the durable projection + // closes the run even when no ephemeral stop event arrived. Transcript rows that were already + // accepted and displayed remain useful evidence, though: merge them by provider identity while + // letting durable rows win conflicts. The subscription remains keyed to this runtime identity so + // events already in transport can still append evidence after the lifecycle becomes terminal. + store.setState((state) => ({ + sessions: state.sessions.map((candidate) => + candidate.id === projection.id + ? (() => { + const sameRuntimeItemIdentity = ( + left: { id: string }, + right: { id: string } + ): boolean => + runtimeItemAlias(left.id, runtimeSegmentId) === + runtimeItemAlias(right.id, runtimeSegmentId) + const messages = settleAcceptedMessages( + mergeAcceptedProjectionItems( + candidate.messages, + projection.messages, + isSameMessageIdentity, + hasVisibleMessageContent, + mergeMessageConflict + ), + status + ) + const mergedActivities = mergeAcceptedProjectionItems( + candidate.activities ?? [], + projection.activities ?? [], + (left, right) => + sameRuntimeItemIdentity(left, right) || hasSharedEventIdentity(left, right), + undefined, + mergeActivityConflict + ) + const mergedGroups = mergeAcceptedProjectionItems( + candidate.activityGroups ?? [], + projection.activityGroups ?? [], + sameRuntimeItemIdentity, + undefined, + (accepted, durable) => ({ + ...accepted, + ...durable, + activityIds: appendUnique(durable.activityIds, accepted.activityIds) + }) + ) + const finalActivityIdByAlias = new Map( + mergedActivities.map((activity) => [ + runtimeItemAlias(activity.id, runtimeSegmentId), + activity.id + ]) + ) + const finalGroupIdByAlias = new Map( + mergedGroups.map((group) => [runtimeItemAlias(group.id, runtimeSegmentId), group.id]) + ) + const resolveActivityId = (id: string): string => + finalActivityIdByAlias.get(runtimeItemAlias(id, runtimeSegmentId)) ?? id + const resolveGroupId = (id: string | undefined): string | undefined => + id + ? (finalGroupIdByAlias.get(runtimeItemAlias(id, runtimeSegmentId)) ?? id) + : undefined + return { + ...projection, + error: projection.error ?? candidate.error, + messages, + activities: mergedActivities.map((activity) => + settleAcceptedActivity( + { + ...activity, + activityGroupId: resolveGroupId(activity.activityGroupId) + }, + status + ) + ), + activityGroups: mergedGroups.map((group) => ({ + ...(status === 'awaiting_user' || group.completedAt !== undefined + ? group + : { + ...group, + completedAt: projection.updatedAt, + updatedAt: projection.updatedAt + }), + activityIds: appendUnique([], group.activityIds.map(resolveActivityId)) + })), + activeRun: undefined, + agentPromptInFlight: undefined, + awaitingFirstAgentOutput: undefined, + agentStatus: undefined, + activeRunRuntimeSegmentId: undefined, + interactionState: undefined + } + })() + : candidate + ) + })) +} + +const fenceTerminalLifecycle = ( + store: SubagentPresentationStore, + projection: ChatSession +): void => { + store.setState((state) => ({ + sessions: state.sessions.map((candidate) => + candidate.id === projection.id + ? { + ...candidate, + status: projection.status, + error: projection.error ?? candidate.error, + activeRun: undefined, + agentPromptInFlight: undefined, + awaitingFirstAgentOutput: undefined, + agentStatus: undefined, + activeRunRuntimeSegmentId: undefined, + interactionState: undefined + } + : candidate + ) + })) +} + +export { childConversationSession, fenceTerminalLifecycle, reconcileDurableChildProjection } +export type { SubagentPresentationStore, WorkspaceSubagentFrameProjection } diff --git a/src/renderer/src/pages/home/HomePage.render.test.tsx b/src/renderer/src/pages/home/HomePage.render.test.tsx index a5ede8211..211f8c0a7 100644 --- a/src/renderer/src/pages/home/HomePage.render.test.tsx +++ b/src/renderer/src/pages/home/HomePage.render.test.tsx @@ -6,6 +6,8 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import type { ProjectFilesChangedEvent } from '../../../../shared/project-files' import type { Project } from '../../../../shared/projects' import type { EnvironmentCheckResult } from '../../../../shared/settings' +import { validateConversationGraph } from '../../../../shared/conversation-graph' +import { sanitizeSessionRuntimeContext } from '../../../../shared/session-persistence' import type { ActivePlanProjection } from '../../../../shared/session-plan/contract' import { EMPTY_SNAPSHOT, useNotificationInboxStore } from '@/stores/notification-inbox-store' import { createInitialProjectState, useProjectStore } from '@/stores/project-store' @@ -109,7 +111,8 @@ const sessionWithPendingDelegatedQuestion = ( delegateName: 'Researcher', status: 'completed', activeBranchId: 'child-branch', - createdAt: 2 + createdAt: 2, + completedAt: 2 } ], branches: [ @@ -148,19 +151,44 @@ const sessionWithPendingDelegatedQuestion = ( eventIds: [], agentFrameId: 'child', introducedOnBranchId: 'child-branch', + runtimeSegmentId: 'runtime-1', createdAt: 2, - updatedAt: 2 + updatedAt: 2, + completedAt: 2 } ], activities: [], activityGroups: [], - runtimeSegments: [] + runtimeSegments: [ + { + id: 'runtime-1', + agentFrameId: 'child', + frameworkId: 'claude-code', + startedAt: 1, + endedAt: 2 + } + ] }, runtimeContext: { version: 1, revision: 1, delegatedWork: { - records: [], + records: [ + { + agentFrameId: 'child', + attempts: [ + { + id: 'attempt-1', + status: 'completed', + resolvedAgent: { kind: 'main' }, + runtimeSegmentIds: ['runtime-1'], + startedAt: 1, + endedAt: 2, + terminalMessageId: 'child-message' + } + ] + } + ], questionRequests: [ { requestId: 'question-1', @@ -545,6 +573,15 @@ describe('HomePage environment repair notice', () => { }) describe('HomePage activity overview', () => { + it('uses a persistence-valid completed Attempt for the pending delegated-question fixture', () => { + const candidate = sessionWithPendingDelegatedQuestion('idle', 600_000) + + expect(sanitizeSessionRuntimeContext(candidate.runtimeContext)).toEqual( + candidate.runtimeContext + ) + expect(() => validateConversationGraph(candidate.conversationGraph!)).not.toThrow() + }) + it('matches the shared session menu and opens Project Settings', async () => { useProjectStore.setState({ ...createInitialProjectState(), diff --git a/src/renderer/src/pages/workspace/ConversationPanel.interaction.test.tsx b/src/renderer/src/pages/workspace/ConversationPanel.interaction.test.tsx index 4b1db3a59..4d662ba38 100644 --- a/src/renderer/src/pages/workspace/ConversationPanel.interaction.test.tsx +++ b/src/renderer/src/pages/workspace/ConversationPanel.interaction.test.tsx @@ -300,7 +300,21 @@ const delegatedQuestionSession = (): ChatSession => ({ version: 1, revision: 1, delegatedWork: { - records: [], + records: [ + { + agentFrameId: 'child', + attempts: [ + { + id: 'attempt-1', + status: 'completed', + resolvedAgent: { kind: 'main' }, + runtimeSegmentIds: ['runtime-1'], + startedAt: 1, + endedAt: 2 + } + ] + } + ], questionRequests: [ { requestId: 'question-1', @@ -656,22 +670,48 @@ describe('ConversationPanel composer intake', () => { expect(document.activeElement).toBe(navigationButton) }) - it('keeps the Main composer available while a delegated question is pending', () => { + it('keeps the Main composer available and confirms a completed Subagent question', async () => { + const terminalSession = delegatedQuestionSession() + const onRespondToElicitation = vi.fn().mockResolvedValue(undefined) + renderPanel({ view: { - activeSession: delegatedQuestionSession() + activeSession: terminalSession }, conversation: { availability: { submit: true } + }, + elicitation: { + respond: onRespondToElicitation } }) expect(container.textContent).toContain('Asked by Researcher') expect(container.textContent).toContain('Which scope?') + expect(container.querySelector('[data-testid="delegated-question-card"]')).not.toBeNull() expect(getComposerEditor().getAttribute('contenteditable')).toBe('true') expect(getComposerForm().contains(getComposerEditor())).toBe(true) + + await act(async () => + container.querySelector('button[aria-label="Narrow"]')?.click() + ) + const finish = Array.from(container.querySelectorAll('button')).find( + (button) => button.textContent?.trim() === 'Finish' + ) + await act(async () => finish?.click()) + expect(onRespondToElicitation).toHaveBeenLastCalledWith( + expect.objectContaining({ + requestId: 'question-1', + delegatedQuestion: { + projectId: 'project-a', + sessionId: 'session-delegated-question', + action: 'confirm', + answers: [{ questionIndex: 0, value: 'Narrow' }] + } + }) + ) }) it('advances to the next Subagent request after Finish and removes an empty queue', async () => { @@ -730,6 +770,22 @@ describe('ConversationPanel composer intake', () => { ...firstSession.runtimeContext!, delegatedWork: { ...firstSession.runtimeContext!.delegatedWork!, + records: [ + ...firstSession.runtimeContext!.delegatedWork!.records, + { + agentFrameId: 'child-two', + attempts: [ + { + id: 'attempt-2', + status: 'completed' as const, + resolvedAgent: { kind: 'main' as const }, + runtimeSegmentIds: ['runtime-2'], + startedAt: 2, + endedAt: 3 + } + ] + } + ], questionRequests: [ ...firstSession.runtimeContext!.delegatedWork!.questionRequests!, secondRequest diff --git a/src/renderer/src/pages/workspace/SubagentReleaseSurfaces.render.test.tsx b/src/renderer/src/pages/workspace/SubagentReleaseSurfaces.render.test.tsx index 80eeebb2e..a1470cbe6 100644 --- a/src/renderer/src/pages/workspace/SubagentReleaseSurfaces.render.test.tsx +++ b/src/renderer/src/pages/workspace/SubagentReleaseSurfaces.render.test.tsx @@ -1,10 +1,12 @@ // @vitest-environment jsdom import { act, cleanup, fireEvent, render, screen, within } from '@testing-library/react' +import { useLayoutEffect } from 'react' import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import type { ChatSession } from '@/stores/session-store' import type { AcpAgentRuntimeUpdate } from '../../../../shared/acp' +import { useSubagentRuntimePresentation } from '@/lib/acp/workspace-subagent-runtime-presentation' const runtimeUpdateHarness = vi.hoisted(() => { const listeners = new Set<(update: AcpAgentRuntimeUpdate) => void>() @@ -48,6 +50,58 @@ import { MobilePreviewSheet } from './MobilePreviewSheet' const renderSurface = (surface: React.ReactNode): ReturnType => render(surface) +const RuntimePresentationProbe = ({ + session, + detail, + publishAfterLayout +}: { + session: ChatSession + detail: Parameters[2] + publishAfterLayout?: () => void +}): React.JSX.Element => { + const projected = useSubagentRuntimePresentation(runtimeUpdateHarness.subscribe, session, detail) + useLayoutEffect(() => publishAfterLayout?.(), [publishAfterLayout]) + return ( + 
+      {JSON.stringify({
+        status: projected.status,
+        error: projected.error,
+        agentStatus: projected.agentStatus,
+        activeRun: Boolean(projected.activeRun),
+        agentPromptInFlight: Boolean(projected.agentPromptInFlight),
+        awaitingFirstAgentOutput: Boolean(projected.awaitingFirstAgentOutput),
+        interactionState: Boolean(projected.interactionState),
+        messages: projected.messages.map(({ content }) => content),
+        activities: projected.activities?.map(
+          ({
+            id,
+            title,
+            status,
+            activityGroupId,
+            eventIds,
+            terminalOutput,
+            rawOutput,
+            providerToolName
+          }) => ({
+            id,
+            title,
+            status,
+            activityGroupId,
+            eventIds,
+            terminalOutput,
+            rawOutput,
+            providerToolName
+          })
+        ),
+        activityGroups: projected.activityGroups?.map(({ id, activityIds }) => ({
+          id,
+          activityIds
+        }))
+      })}
+
+ ) +} + const createSession = (): ChatSession => { const now = 1_700_000_000_000 return { @@ -447,7 +501,7 @@ describe('release-gate Subagent surfaces', () => { ) expect(screen.getByLabelText('Subagent Frame').className).toContain('focus-visible:ring-3') - expect(screen.getByText('error')).toBeTruthy() + expect(screen.getByText('Failed')).toBeTruthy() expect(screen.getByText('Provider turn failed')).toBeTruthy() expect(screen.queryByRole('button', { name: /stop/i })).toBeNull() @@ -627,6 +681,542 @@ describe('release-gate Subagent surfaces', () => { expect(useSessionStore.getState().sessions[0]).toEqual(rootBefore) }) + it.each(['cancelled', 'error', 'awaiting_user'] as const)( + 'clears the isolated running presentation when durable child state becomes %s', + async (status) => { + const running = createSession() + const childBranch = running.conversationGraph?.branches.find( + (branch) => branch.id === 'child-a-branch' + ) + if (childBranch) childBranch.headMessageId = 'child-a-prompt' + if (running.conversationGraph) { + running.conversationGraph.messages = running.conversationGraph.messages.filter( + (message) => message.id !== 'child-a-answer' + ) + } + useSessionStore.setState({ ...createInitialSessionState(), sessions: [running] }) + + renderSurface( + + ) + expect(screen.getByText('Thinking')).toBeTruthy() + await act(async () => { + runtimeUpdateHarness.publish({ + scope: { + projectId: 'project-1', + sessionId: 'session-1', + agentFrameId: 'child-a', + attemptId: 'attempt-a', + runtimeSegmentId: 'runtime-a', + promptMessageId: 'child-a-prompt' + }, + event: { + id: `child-warning-${status}`, + timestamp: running.updatedAt + 1, + kind: 'system', + level: 'warning', + text: 'Retrying child request' + } + }) + }) + expect(screen.getByText('Retrying child request')).toBeTruthy() + + const durable = structuredClone(running) + const runtimeContext = durable.runtimeContext! + durable.runtimeContext = { ...runtimeContext, revision: runtimeContext.revision + 1 } + const frame = durable.conversationGraph?.frames.find(({ id }) => id === 'child-a') + const attempt = durable.runtimeContext.delegatedWork?.records + .find(({ agentFrameId }) => agentFrameId === 'child-a') + ?.attempts.at(-1) + if (!frame || !attempt) throw new Error('Expected child-a durable fixtures') + + if (status === 'awaiting_user') { + Object.assign(durable.runtimeContext.delegatedWork!, { + questionRequests: [ + { + requestId: 'question-a', + canonicalDigest: 'a'.repeat(64), + sourceFrameId: 'child-a', + sourceAttemptId: 'attempt-a', + sourceRuntimeSegmentId: 'runtime-a', + sourceMessageBranchId: 'child-a-branch', + rootOriginMessageId: 'root-prompt', + rootBranchId: 'root-branch', + sourceName: 'Evidence landscape', + questions: [{ question: 'Continue?', options: [{ label: 'Yes' }, { label: 'No' }] }], + askedAt: running.updatedAt, + status: 'pending', + draftAnswers: [], + draftQuestionIndex: 0 + } + ] + }) + } else { + frame.status = status + frame.completedAt = running.updatedAt + Object.assign(attempt, { + status, + endedAt: running.updatedAt, + ...(status === 'cancelled' + ? { cancellationReason: 'main_agent_stop' as const } + : { error: { code: 'provider', message: 'Child failed durably' } }) + }) + } + + await act(async () => { + useSessionStore.getState().upsertPersistedSession(durable) + }) + + expect(document.querySelector(`[data-subagent-status="${status}"]`)).not.toBeNull() + expect(screen.queryByText('Thinking')).toBeNull() + expect(screen.queryByText('Retrying child request')).toBeNull() + if (status === 'error') expect(screen.getByText('Child failed durably')).toBeTruthy() + + await act(async () => { + runtimeUpdateHarness.publish({ + scope: { + projectId: 'project-1', + sessionId: 'session-1', + agentFrameId: 'child-a', + attemptId: 'attempt-a', + runtimeSegmentId: 'runtime-a', + promptMessageId: 'child-a-prompt' + }, + event: { + id: `child-late-warning-${status}`, + timestamp: running.updatedAt + 20, + kind: 'system', + level: 'warning', + text: 'Late child warning' + } + }) + runtimeUpdateHarness.publish({ + scope: { + projectId: 'project-1', + sessionId: 'session-1', + agentFrameId: 'child-a', + attemptId: 'attempt-a', + runtimeSegmentId: 'runtime-a', + promptMessageId: 'child-a-prompt' + }, + event: { + id: `child-late-error-${status}`, + timestamp: running.updatedAt + 21, + kind: 'error', + level: 'error', + text: 'Late runtime error' + } + }) + }) + + expect(document.querySelector(`[data-subagent-status="${status}"]`)).not.toBeNull() + expect(screen.queryByText('Provider notice')).toBeNull() + expect(screen.queryByText('Provider error')).toBeNull() + expect(screen.queryByText('Late child warning')).toBeNull() + expect(screen.queryByText('Late runtime error')).toBeNull() + expect(screen.queryByText('Thinking')).toBeNull() + } + ) + + it('preserves accepted content and applies same-runtime events after durable termination', async () => { + const running = createSession() + const prompt = running.conversationGraph!.messages.find(({ id }) => id === 'child-a-prompt')! + const attempt = running.runtimeContext!.delegatedWork!.records.find( + ({ agentFrameId }) => agentFrameId === 'child-a' + )!.attempts[0] + const runningDetail = { + frameId: 'child-a', + status: 'running' as const, + attempt, + messages: [prompt] + } + const rendered = render() + const scope = { + projectId: 'project-1', + sessionId: 'session-1', + agentFrameId: 'child-a', + attemptId: 'attempt-a', + runtimeSegmentId: 'runtime-a', + promptMessageId: 'child-a-prompt' + } + + await act(async () => { + runtimeUpdateHarness.publish({ + scope, + event: { + id: 'live-group-start', + timestamp: running.updatedAt + 10, + kind: 'tool', + level: 'info', + toolCallId: 'live-group-call', + providerToolName: 'mcp__open-science-activity__begin_activity_group', + rawInput: { title: 'Accepted live group' }, + status: 'in_progress' + } + }) + runtimeUpdateHarness.publish({ + scope, + event: { + id: 'live-ghost-message', + timestamp: running.updatedAt + 11, + kind: 'message', + level: 'info', + role: 'assistant', + messageId: 'live-ghost-stream', + text: 'Live ghost message' + } + }) + runtimeUpdateHarness.publish({ + scope, + event: { + id: 'live-ghost-tail', + timestamp: running.updatedAt + 12, + kind: 'message', + level: 'info', + role: 'assistant', + messageId: 'live-ghost-stream', + text: ' with live tail' + } + }) + runtimeUpdateHarness.publish({ + scope, + event: { + id: 'live-ghost-tool', + timestamp: running.updatedAt + 13, + kind: 'tool', + level: 'info', + toolCallId: 'live-ghost-tool-call', + providerToolName: 'Read', + toolKind: 'read', + title: 'Live ghost tool', + status: 'in_progress' + } + }) + runtimeUpdateHarness.publish({ + scope, + event: { + id: 'live-ghost-tool-complete', + timestamp: running.updatedAt + 14, + kind: 'tool', + level: 'info', + toolCallId: 'live-ghost-tool-call', + providerToolName: 'Read', + toolKind: 'read', + title: 'Live ghost tool', + status: 'completed', + terminalOutput: 'Accepted completion output', + rawOutput: { phase: 'complete' } + } + }) + runtimeUpdateHarness.publish({ + scope, + event: { + id: 'live-only-message', + timestamp: running.updatedAt + 15, + kind: 'message', + level: 'info', + role: 'assistant', + messageId: 'live-only-stream', + text: 'Accepted live-only message' + } + }) + runtimeUpdateHarness.publish({ + scope, + event: { + id: 'live-warning', + timestamp: running.updatedAt + 16, + kind: 'system', + level: 'warning', + text: 'Transient live warning' + } + }) + }) + const probe = screen.getByTestId('runtime-presentation-probe') + expect(probe.textContent).toContain('Live ghost message with live tail') + expect(probe.textContent).toContain('Accepted live-only message') + expect(probe.textContent).toContain('Live ghost tool') + expect(probe.textContent).toContain('Transient live warning') + expect(probe.textContent).toContain('"activeRun":true') + expect(probe.textContent).toContain('"agentPromptInFlight":true') + + const terminal = structuredClone(running) + const terminalFrame = terminal.conversationGraph!.frames.find(({ id }) => id === 'child-a')! + const terminalAttempt = terminal.runtimeContext!.delegatedWork!.records.find( + ({ agentFrameId }) => agentFrameId === 'child-a' + )!.attempts[0] + terminalFrame.status = 'cancelled' + terminalFrame.completedAt = running.updatedAt + 12 + Object.assign(terminalAttempt, { + status: 'cancelled', + endedAt: running.updatedAt + 12, + cancellationReason: 'main_agent_stop' + }) + terminal.conversationGraph!.activities.push({ + id: 'agent-runtime:runtime-a:live-ghost-tool-call', + kind: 'tool', + title: 'Live ghost tool', + activityGroupId: 'agent-runtime:runtime-a:live-group-call', + promptMessageId: 'child-a-prompt', + status: 'failed', + sortIndex: 2, + eventIds: ['live-ghost-tool'], + providerToolName: 'Read', + toolKind: 'read', + terminalOutput: 'Durable stale output', + rawOutput: { phase: 'start' }, + createdAt: running.updatedAt + 13, + updatedAt: running.updatedAt + 16, + agentFrameId: 'child-a', + messageBranchId: 'child-a-branch', + runtimeSegmentId: 'runtime-a' + }) + terminal.conversationGraph!.activityGroups.push({ + id: 'agent-runtime:runtime-a:live-group-call', + title: 'Accepted live group', + sortIndex: 1, + activityIds: ['agent-runtime:runtime-a:live-ghost-tool-call'], + promptMessageId: 'child-a-prompt', + createdAt: running.updatedAt + 10, + updatedAt: running.updatedAt + 16, + completedAt: running.updatedAt + 16, + agentFrameId: 'child-a', + messageBranchId: 'child-a-branch' + }) + const terminalDetail = { + frameId: 'child-a', + status: 'cancelled' as const, + attempt: terminalAttempt, + messages: [ + prompt, + { + id: 'durable-live-message', + role: 'agent' as const, + content: 'Live ghost message', + status: 'complete' as const, + eventIds: ['live-ghost-message'], + responseToMessageId: 'child-a-prompt', + createdAt: running.updatedAt + 10, + updatedAt: running.updatedAt + 12 + } + ] + } + await act(async () => { + rendered.rerender( + { + runtimeUpdateHarness.publish({ + scope, + event: { + id: 'layout-gap-message', + timestamp: running.updatedAt + 17, + kind: 'message', + level: 'info', + role: 'assistant', + messageId: 'layout-gap-stream', + text: 'Evidence from the layout-to-passive cleanup gap' + } + }) + runtimeUpdateHarness.publish({ + scope, + event: { + id: 'layout-gap-error', + timestamp: running.updatedAt + 18, + kind: 'error', + level: 'error', + text: 'Error from the layout-to-passive cleanup gap' + } + }) + }} + /> + ) + }) + + expect(probe.textContent).toContain('"status":"idle"') + expect(probe.textContent?.match(/Live ghost message with live tail/g)).toHaveLength(1) + expect(probe.textContent).toContain('Accepted live-only message') + expect(probe.textContent?.match(/Live ghost tool/g)).toHaveLength(1) + expect(probe.textContent).toContain('agent-runtime:runtime-a:live-ghost-tool-call') + expect(probe.textContent).not.toContain('"id":"live-ghost-tool-call"') + expect(probe.textContent).toContain( + '"activityIds":["agent-runtime:runtime-a:live-ghost-tool-call"]' + ) + expect(probe.textContent).toContain('"eventIds":["live-ghost-tool","live-ghost-tool-complete"]') + expect(probe.textContent).toContain('"status":"completed"') + expect(probe.textContent).toContain('"terminalOutput":"Accepted completion output"') + expect(probe.textContent).toContain('"rawOutput":{"phase":"complete"}') + expect(probe.textContent).not.toContain('Durable stale output') + expect(probe.textContent).not.toContain('Transient live warning') + expect(probe.textContent).toContain('Evidence from the layout-to-passive cleanup gap') + expect(probe.textContent).toContain('Error from the layout-to-passive cleanup gap') + expect(probe.textContent).not.toContain('AgentRuntimeEvidence') + expect(probe.textContent).toContain('"activeRun":false') + expect(probe.textContent).toContain('"agentPromptInFlight":false') + expect(probe.textContent).toContain('"awaitingFirstAgentOutput":false') + expect(probe.textContent).toContain('"interactionState":false') + + await act(async () => { + runtimeUpdateHarness.publish({ + scope, + event: { + id: 'late-terminal-message-prefix', + timestamp: running.updatedAt + 20, + kind: 'message', + level: 'info', + role: 'assistant', + messageId: 'late-terminal-stream', + text: 'Late terminal message' + } + }) + runtimeUpdateHarness.publish({ + scope, + event: { + id: 'late-terminal-message-final', + timestamp: running.updatedAt + 21, + kind: 'message', + level: 'info', + role: 'assistant', + messageId: 'late-terminal-stream', + text: ' final' + } + }) + runtimeUpdateHarness.publish({ + scope, + event: { + id: 'late-terminal-tool', + timestamp: running.updatedAt + 22, + kind: 'tool', + level: 'info', + toolCallId: 'late-terminal-tool-call', + providerToolName: 'Read', + toolKind: 'read', + title: 'Late terminal tool', + status: 'in_progress' + } + }) + runtimeUpdateHarness.publish({ + scope, + event: { + id: 'late-terminal-tool-complete', + timestamp: running.updatedAt + 23, + kind: 'tool', + level: 'info', + toolCallId: 'late-terminal-tool-call', + providerToolName: 'Read', + toolKind: 'read', + title: 'Late terminal tool', + status: 'completed', + terminalOutput: 'Late tool output' + } + }) + runtimeUpdateHarness.publish({ + scope, + event: { + id: 'late-terminal-stop', + timestamp: running.updatedAt + 24, + kind: 'stop', + level: 'info', + turnUsage: { inputTokens: 9, cacheTokens: 0, outputTokens: 4 } + } + }) + runtimeUpdateHarness.publish({ + scope, + event: { + id: 'late-terminal-error', + timestamp: running.updatedAt + 25, + kind: 'error', + level: 'error', + text: 'Late provider error must not replace cancellation' + } + }) + runtimeUpdateHarness.publish({ + scope, + event: { + id: 'late-terminal-warning', + timestamp: running.updatedAt + 26, + kind: 'system', + level: 'warning', + text: 'Late provider warning evidence' + } + }) + runtimeUpdateHarness.publish({ + scope, + event: { + id: 'late-terminal-message-final', + timestamp: running.updatedAt + 27, + kind: 'message', + level: 'info', + role: 'assistant', + messageId: 'late-terminal-stream', + text: ' duplicated final' + } + }) + runtimeUpdateHarness.publish({ + scope: { ...scope, attemptId: 'different-attempt' }, + event: { + id: 'wrong-attempt-message', + timestamp: running.updatedAt + 28, + kind: 'message', + level: 'info', + role: 'assistant', + messageId: 'wrong-attempt-stream', + text: 'Wrong attempt evidence' + } + }) + runtimeUpdateHarness.publish({ + scope: { ...scope, runtimeSegmentId: 'different-runtime' }, + event: { + id: 'wrong-runtime-message', + timestamp: running.updatedAt + 29, + kind: 'message', + level: 'info', + role: 'assistant', + messageId: 'wrong-runtime-stream', + text: 'Wrong runtime evidence' + } + }) + runtimeUpdateHarness.publish({ + scope: { ...scope, promptMessageId: 'different-prompt' }, + event: { + id: 'wrong-prompt-message', + timestamp: running.updatedAt + 30, + kind: 'message', + level: 'info', + role: 'assistant', + messageId: 'wrong-prompt-stream', + text: 'Wrong prompt evidence' + } + }) + }) + + expect(probe.textContent).toContain('"status":"idle"') + expect(probe.textContent).toContain('Late terminal message final') + expect(probe.textContent).toContain('Late terminal tool') + expect(probe.textContent).toContain('Late tool output') + expect(probe.textContent).not.toContain('duplicated final') + expect(probe.textContent).not.toContain('Wrong attempt evidence') + expect(probe.textContent).not.toContain('Wrong runtime evidence') + expect(probe.textContent).not.toContain('Wrong prompt evidence') + expect(probe.textContent).toContain('Late provider error must not replace cancellation') + expect(probe.textContent).not.toContain('Late provider warning evidence') + expect(probe.textContent).not.toContain('AgentRuntimeEvidence') + expect(probe.textContent).toContain('"activeRun":false') + expect(probe.textContent).toContain('"agentPromptInFlight":false') + expect(probe.textContent).toContain('"awaitingFirstAgentOutput":false') + expect(probe.textContent).toContain('"interactionState":false') + }) + it('reconciles a newer durable projection for the same running Attempt', async () => { const running = createSession() const childBranch = running.conversationGraph?.branches.find( @@ -698,7 +1288,7 @@ describe('release-gate Subagent surfaces', () => { expect(screen.queryByText('Thinking')).toBeNull() }) - it('offers Retry when the selected durable Frame cannot be read', () => { + it('falls back to the first existing Frame when the selected Frame was removed', () => { renderSurface( { /> ) - expect(screen.getByRole('alert').textContent).toContain('could not be read') - expect(screen.getByRole('button', { name: 'Retry Subagent preview' }).className).toContain( - 'focus-visible:ring-[3px]' - ) + expect(screen.getByLabelText('Subagent Frame').textContent).toContain('Evidence landscape') + expect(screen.getByText('Fourteen strong studies remain.')).toBeTruthy() + expect(screen.queryByRole('alert')).toBeNull() + }) + + it('uses distinct terminal labels and the warning color for cancellation', () => { + const session = createSession() + const cancelledFrame = session.conversationGraph!.frames.find(({ id }) => id === 'child-a')! + const cancelledAttempt = session.runtimeContext!.delegatedWork!.records.find( + ({ agentFrameId }) => agentFrameId === 'child-a' + )!.attempts[0] + Object.assign(cancelledFrame, { status: 'cancelled' }) + Object.assign(cancelledAttempt, { status: 'cancelled' }) + + renderSurface() + fireEvent.click(screen.getByRole('button', { name: '2 subagents' })) + + const cancelled = screen.getByText('Cancelled') + const failed = screen.getByText('Failed') + expect(cancelled.getAttribute('data-subagent-status')).toBe('cancelled') + expect(cancelled.previousElementSibling?.className).toContain('bg-warning-100') + expect(failed.getAttribute('data-subagent-status')).toBe('error') }) it('shows an actionable unavailable notice and no false support claim', () => { diff --git a/src/renderer/src/pages/workspace/SubagentReleaseSurfaces.tsx b/src/renderer/src/pages/workspace/SubagentReleaseSurfaces.tsx index feffaf2e8..fcabf2abf 100644 --- a/src/renderer/src/pages/workspace/SubagentReleaseSurfaces.tsx +++ b/src/renderer/src/pages/workspace/SubagentReleaseSurfaces.tsx @@ -41,6 +41,14 @@ const statusDotClassName: Record = { error: 'bg-danger-000' } +const statusLabel: Record = { + running: 'Running', + awaiting_user: 'Awaiting user', + completed: 'Completed', + cancelled: 'Cancelled', + error: 'Failed' +} + const SubagentStatus = ({ status, awaitingPermission = false @@ -50,9 +58,7 @@ const SubagentStatus = ({ }): React.JSX.Element => (