diff --git a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts index f63a220fb5191..ee7803a6492a7 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts @@ -4028,79 +4028,301 @@ describe("AcpAdapterV2", () => { }).pipe(Effect.provide(layerTest), Effect.scoped), ); - it.live("replaces an unexpectedly terminated ACP runtime before the next turn", () => - Effect.gen(function* () { - const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; - const path = yield* Path.Path; - const selfInvocation = yield* resolveSelfInvocation(); - const mockAgentPath = yield* path.fromFileUrl( - new URL("../../../scripts/acp-mock-agent.ts", import.meta.url), - ); - const runtimeInputs: AcpAdapterV2RuntimeInput[] = []; - let runtimeOrdinalSeen = 0; - const baseMakeRuntime = makeMockRuntime({ - childProcessSpawner, - mockAgentPath, - environment: (runtimeOrdinal) => { - runtimeOrdinalSeen = runtimeOrdinal; - return {}; - }, - }); - const instanceId = ProviderInstanceId.make("acp-test-unexpected-termination"); - const adapter = yield* makeAcpAdapterV2({ - instanceId, - flavor: { - driver: ACP_TEST_DRIVER, - capabilities: AcpProviderCapabilitiesV2, - makeRuntime: (runtimeInput) => - Effect.sync(() => { - runtimeInputs.push(runtimeInput); - }).pipe(Effect.andThen(baseMakeRuntime(runtimeInput))), - }, - selfInvocation, - }); - const threadId = ThreadId.make("thread-acp-unexpected-termination"); - const runtimePolicy = ProviderAdapter.ProviderAdapterV2RuntimePolicy.make({ - runtimeMode: "full-access", - interactionMode: "default", - cwd: process.cwd(), - }); - const modelSelection = { instanceId, model: "default" } as const; - const runtime = yield* adapter.openSession({ - threadId, - providerSessionId: ProviderSessionId.make("provider-session-acp-unexpected-termination"), - modelSelection, - runtimePolicy, - }); - const providerThread = yield* runtime.ensureThread({ - threadId, - modelSelection, - runtimePolicy, - }); - assert.equal(runtimeOrdinalSeen, 1); - yield* runtimeInputs[0]!.onTermination!( - new EffectAcpErrors.AcpTransportError({ - detail: "Injected unexpected writer termination", - cause: "test", - }), - ); - - yield* runtime.startTurn( - makeTurnInput({ - threadId, - providerThread, + it.live.each(["termination", "active termination", "unowned termination", "rollback"] as const)( + "handles Copilot runtime tasks on %s", + (replacement) => + Effect.gen(function* () { + const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const path = yield* Path.Path; + const selfInvocation = yield* resolveSelfInvocation(); + const mockAgentPath = yield* path.fromFileUrl( + new URL("../../../scripts/acp-mock-agent.ts", import.meta.url), + ); + const runtimeInputs: AcpAdapterV2RuntimeInput[] = []; + let runtimeOrdinalSeen = 0; + const owned = replacement !== "unowned termination"; + const activeTermination = replacement === "active termination" || !owned; + let prompts = 0; + const activePrompt = yield* Deferred.make(); + const releaseActivePrompt = yield* Deferred.make(); + const handlers: Array< + Parameters[0] + > = []; + const routing = makeCopilotSubagentRouting(); + const runtimes: Array = []; + const rawHandlers: Array< + (event: unknown) => Effect.Effect + > = []; + const baseMakeRuntime = makeMockRuntime({ + childProcessSpawner, + mockAgentPath, + wrapRuntime: (runtime) => { + const wrapped: AcpSessionRuntime.AcpSessionRuntime["Service"] = { + ...runtime, + handleSessionUpdate: (handler) => + Effect.sync(() => { + handlers.push(handler); + }).pipe(Effect.andThen(runtime.handleSessionUpdate(handler))), + handleExtNotification: (method, schema, handler) => + Effect.sync(() => { + if (method === "github.com/copilot/sessionEvent") + rawHandlers.push(handler as (typeof rawHandlers)[number]); + }).pipe(Effect.andThen(runtime.handleExtNotification(method, schema, handler))), + prompt: () => + Effect.gen(function* () { + prompts += 1; + if (activeTermination && prompts === 2) { + yield* handlers[0]!({ + sessionId: "mock-session-1", + update: { + sessionUpdate: "tool_call", + toolCallId: "root-command", + title: "Detached root command", + kind: "execute", + status: "in_progress", + rawInput: { command: "sleep 80" }, + }, + }); + yield* Deferred.succeed(activePrompt, undefined); + yield* Deferred.await(releaseActivePrompt); + return { stopReason: "end_turn" as const }; + } + if (runtimeOrdinalSeen === 1) { + yield* handlers[0]!({ + sessionId: "mock-session-1", + update: { + sessionUpdate: "tool_call", + toolCallId: "task-bg", + title: "Background task", + kind: "other", + status: "pending", + rawInput: { + description: "Background task", + prompt: "Do work", + agent_type: "explore", + mode: "background", + }, + }, + }); + yield* handlers[0]!({ + sessionId: "mock-session-1", + update: { + sessionUpdate: "tool_call_update", + toolCallId: "task-bg", + status: "completed", + }, + }); + } else { + yield* handlers[0]!({ + sessionId: "mock-session-1", + update: { + sessionUpdate: "agent_message_chunk", + content: { type: "text", text: "stale runtime output" }, + }, + }); + yield* handlers[1]!({ + sessionId: "mock-session-1", + update: { + sessionUpdate: "agent_message_chunk", + content: { type: "text", text: "replacement output" }, + }, + }); + } + return { stopReason: "end_turn" as const }; + }), + }; + runtimes.push(wrapped); + return wrapped; + }, + environment: (runtimeOrdinal) => { + runtimeOrdinalSeen = runtimeOrdinal; + return {}; + }, + }); + const instanceId = ProviderInstanceId.make("acp-test-unexpected-termination"); + const adapter = yield* makeAcpAdapterV2({ instanceId, + flavor: { + driver: ACP_TEST_DRIVER, + capabilities: AcpProviderCapabilitiesV2, + registerExtensions: routing.registerExtensions, + normalizeSessionUpdate: routing.normalizeSessionUpdate, + extractSubagentUpdate: extractCopilotSubagentUpdate, + subagentsOwnedByRuntime: owned, + enablePostSettleContinuation: true, + makeRuntime: (runtimeInput) => + Effect.sync(() => { + runtimeInputs.push(runtimeInput); + }).pipe(Effect.andThen(baseMakeRuntime(runtimeInput))), + }, + selfInvocation, + continuationRequests: { offer: () => Effect.void }, + }); + const threadId = ThreadId.make("thread-acp-unexpected-termination"); + const runtimePolicy = ProviderAdapter.ProviderAdapterV2RuntimePolicy.make({ + runtimeMode: "full-access", + interactionMode: "default", + cwd: process.cwd(), + }); + const modelSelection = { instanceId, model: "default" } as const; + const runtime = yield* adapter.openSession({ + threadId, + providerSessionId: ProviderSessionId.make("provider-session-acp-unexpected-termination"), + modelSelection, runtimePolicy, - now: yield* DateTime.now, - }), - ); - yield* runtime.events.pipe( - Stream.filter((event) => event.type === "turn.terminal"), - Stream.runHead, - ); - assert.equal(runtimeOrdinalSeen, 2); - assert.lengthOf(runtimeInputs, 2); - }).pipe(Effect.provide(layerTest), Effect.scoped), + }); + let providerThread = yield* runtime.ensureThread({ + threadId, + modelSelection, + runtimePolicy, + }); + const events = yield* Queue.unbounded(); + yield* runtime.events.pipe( + Stream.runForEach((event) => Queue.offer(events, event)), + Effect.forkScoped, + ); + yield* runtime.startTurn( + makeTurnInput({ + threadId, + providerThread, + instanceId, + runtimePolicy, + now: yield* DateTime.now, + }), + ); + const initialEvents: ProviderAdapter.ProviderAdapterV2Event[] = []; + while (true) { + const event = yield* Queue.take(events); + initialEvents.push(event); + if (event.type === "turn.terminal") break; + } + assert.isTrue( + initialEvents.some( + (event) => + event.type === "subagent.updated" && + event.subagent.nativeTaskRef?.nativeId === "task-bg" && + event.subagent.status === "running", + ), + ); + assert.equal(runtimeOrdinalSeen, 1); + assert.isTrue(yield* runtime.hasPendingBackgroundWork!); + if (activeTermination) { + yield* runtime.startTurn( + makeTurnInput({ + threadId, + providerThread, + instanceId, + runtimePolicy, + now: yield* DateTime.now, + ordinal: 2, + }), + ); + yield* Deferred.await(activePrompt); + } + if (replacement !== "rollback") { + yield* runtimeInputs[0]!.onTermination!( + new EffectAcpErrors.AcpTransportError({ + detail: "Injected unexpected writer termination", + cause: "test", + }), + ); + } else { + const rolledBack = yield* runtime.rollbackThread({ + providerThread, + providerThreadTurns: [], + target: { + type: "thread_start", + checkpointId: CheckpointId.make("checkpoint-copilot-runtime-tasks"), + appRunOrdinal: 0, + }, + }); + providerThread = rolledBack.providerThread; + } + assert.equal( + yield* runtime.hasPendingBackgroundWork!, + !owned, + "termination must retire only runtime-owned tasks rehydrated into the active turn", + ); + if (activeTermination) { + yield* Deferred.succeed(releaseActivePrompt, undefined); + const activeEvents: ProviderAdapter.ProviderAdapterV2Event[] = []; + while (true) { + const event = yield* Queue.take(events); + activeEvents.push(event); + if (event.type === "turn.terminal") break; + } + assert.deepEqual( + activeEvents.flatMap((event) => + event.type === "subagent.updated" && + event.subagent.nativeTaskRef?.nativeId === "task-bg" + ? [event.subagent.status] + : [], + ), + owned ? ["interrupted"] : [], + ); + assert.isFalse( + activeEvents.some( + (event) => + event.type === "turn_item.updated" && + event.turnItem.nativeItemRef?.nativeId === "root-command" && + event.turnItem.status === "interrupted", + ), + "runtime-owned retirement must leave root commands alone", + ); + } + + yield* runtime.startTurn( + makeTurnInput({ + threadId, + providerThread, + instanceId, + runtimePolicy, + now: yield* DateTime.now, + ordinal: 3, + }), + ); + const observed: ProviderAdapter.ProviderAdapterV2Event[] = []; + while (true) { + const event = yield* Queue.take(events); + observed.push(event); + if (event.type === "turn.terminal") break; + } + const statuses = observed.flatMap((event) => + event.type === "subagent.updated" && event.subagent.nativeTaskRef?.nativeId === "task-bg" + ? [event.subagent.status] + : [], + ); + assert.deepEqual(statuses, activeTermination ? [] : ["interrupted"]); + const replacementMessages = observed.flatMap((event) => + event.type === "message.updated" && event.message.role === "assistant" + ? [event.message.text] + : [], + ); + assert.include(replacementMessages, "replacement output"); + assert.isFalse(replacementMessages.some((text) => text.includes("stale runtime output"))); + yield* rawHandlers[0]!({ + sessionId: "mock-session-1", + type: "subagent.started", + agentId: "stale-agent", + data: { toolCallId: "task-bg" }, + }); + const update = { + sessionId: "mock-session-1", + update: { + sessionUpdate: "tool_call" as const, + toolCallId: "new-tool", + title: "New", + _meta: { "github.com/copilot": { agentId: "stale-agent" } }, + }, + }; + assert.equal( + routing.normalizeSessionUpdate(update, runtimes[1]!).sessionId, + "mock-session-1", + ); + + assert.equal(runtimeOrdinalSeen, 2); + assert.lengthOf(runtimeInputs, 2); + }).pipe(Effect.provide(layerTest), Effect.scoped), ); it.effect("fails a prompt its agent leaves silent past the inactivity timeout", () => diff --git a/packages/provider-acp-registry/src/server/adapter.ts b/packages/provider-acp-registry/src/server/adapter.ts index 1b8b4240fd00e..101a12b45912f 100644 --- a/packages/provider-acp-registry/src/server/adapter.ts +++ b/packages/provider-acp-registry/src/server/adapter.ts @@ -256,6 +256,7 @@ export const makeAcpRegistryAdapterV2 = Effect.fn("makeAcpRegistryAdapterV2")(fu registerExtensions: copilotRouting.registerExtensions, normalizeSessionUpdate: copilotRouting.normalizeSessionUpdate, extractSubagentUpdate: extractCopilotSubagentUpdate, + subagentsOwnedByRuntime: true, restartRuntimeAfterPromptError: isCopilotLostSessionError, }), makeRuntime: options.makeRuntime ?? makeAcpRegistryRuntime(runtimeOptions, catalog), diff --git a/packages/provider-acp-registry/src/server/copilotAcp.ts b/packages/provider-acp-registry/src/server/copilotAcp.ts index 5f29c930e9a91..299b1dd9b311a 100644 --- a/packages/provider-acp-registry/src/server/copilotAcp.ts +++ b/packages/provider-acp-registry/src/server/copilotAcp.ts @@ -66,8 +66,8 @@ interface CopilotSessionAttribution { } export function makeCopilotSubagentRouting() { - const sessions = new Map(); - const session = (sessionId: string) => { + const sessionsByRuntime = new WeakMap>(); + const session = (sessions: Map, sessionId: string) => { let current = sessions.get(sessionId); if (current === undefined) { current = { delta: undefined, taskByAgentId: new Map() }; @@ -75,59 +75,75 @@ export function makeCopilotSubagentRouting() { } return current; }; - const release = (sessionId: string, current: CopilotSessionAttribution) => { + const release = ( + sessions: Map, + sessionId: string, + current: CopilotSessionAttribution, + ) => { if (current.delta === undefined && current.taskByAgentId.size === 0) { sessions.delete(sessionId); } }; - const registerExtensions = (context: AcpAdapterV2ExtensionContext) => - context.runtime.handleExtNotification(COPILOT_SESSION_EVENT, CopilotSessionEvent, (event) => { - const data = event.data ?? {}; - const task = typeof data.parentToolCallId === "string" ? data.parentToolCallId : undefined; - switch (event.type) { - case "assistant.message_delta": - case "assistant.reasoning_delta": { - if (typeof data.deltaContent !== "string") return Effect.void; - session(event.sessionId).delta = { - kind: event.type, - text: data.deltaContent, - ...(task === undefined ? {} : { task }), - }; - return Effect.void; - } - case "subagent.started": { - if (event.agentId !== undefined && typeof data.toolCallId === "string") { - session(event.sessionId).taskByAgentId.set(event.agentId, data.toolCallId); + const registerExtensions = (context: AcpAdapterV2ExtensionContext) => { + const runtimeSessions = new Map(); + sessionsByRuntime.set(context.runtime, runtimeSessions); + return context.runtime.handleExtNotification( + COPILOT_SESSION_EVENT, + CopilotSessionEvent, + (event) => { + const data = event.data ?? {}; + const task = typeof data.parentToolCallId === "string" ? data.parentToolCallId : undefined; + switch (event.type) { + case "assistant.message_delta": + case "assistant.reasoning_delta": { + if (typeof data.deltaContent !== "string") return Effect.void; + session(runtimeSessions, event.sessionId).delta = { + kind: event.type, + text: data.deltaContent, + ...(task === undefined ? {} : { task }), + }; + return Effect.void; } - return Effect.void; - } - case "subagent.completed": - case "subagent.failed": { - const current = sessions.get(event.sessionId); - if (current !== undefined && event.agentId !== undefined) { - current.taskByAgentId.delete(event.agentId); - release(event.sessionId, current); + case "subagent.started": { + if (event.agentId !== undefined && typeof data.toolCallId === "string") { + session(runtimeSessions, event.sessionId).taskByAgentId.set( + event.agentId, + data.toolCallId, + ); + } + return Effect.void; + } + case "subagent.completed": + case "subagent.failed": { + const current = runtimeSessions.get(event.sessionId); + if (current !== undefined && event.agentId !== undefined) { + current.taskByAgentId.delete(event.agentId); + release(runtimeSessions, event.sessionId, current); + } + if (typeof data.toolCallId !== "string") return Effect.void; + // Background tasks end only here; their tool call settles at launch. + return context.finishSubagent({ + sessionId: event.sessionId, + childSessionId: childSessionId(data.toolCallId), + status: event.type === "subagent.failed" ? "failed" : "completed", + result: typeof data.error === "string" ? data.error : null, + }); } - if (typeof data.toolCallId !== "string") return Effect.void; - // Background tasks end only here; their tool call settles at launch. - return context.finishSubagent({ - sessionId: event.sessionId, - childSessionId: childSessionId(data.toolCallId), - status: event.type === "subagent.failed" ? "failed" : "completed", - result: typeof data.error === "string" ? data.error : null, - }); + default: + return Effect.void; } - default: - return Effect.void; - } - }); + }, + ); + }; const normalizeSessionUpdate = ( notification: AcpSchema.SessionNotification, + runtime: object, ): AcpSchema.SessionNotification => { const update = notification.update; - const current = sessions.get(notification.sessionId); + const sessions = sessionsByRuntime.get(runtime); + const current = sessions?.get(notification.sessionId); if (current === undefined) return notification; let task: string | undefined; if ( @@ -136,7 +152,7 @@ export function makeCopilotSubagentRouting() { ) { const delta = current.delta; current.delta = undefined; - release(notification.sessionId, current); + release(sessions!, notification.sessionId, current); const kind = update.sessionUpdate === "agent_message_chunk" ? "assistant.message_delta" diff --git a/packages/provider-acp/src/server/adapter.ts b/packages/provider-acp/src/server/adapter.ts index c4fc2f092644a..67a0b55b5a700 100644 --- a/packages/provider-acp/src/server/adapter.ts +++ b/packages/provider-acp/src/server/adapter.ts @@ -335,6 +335,7 @@ export interface AcpAdapterV2Flavor { readonly mcpOverHttp?: boolean; readonly normalizeSessionUpdate?: ( notification: EffectAcpSchema.SessionNotification, + runtime: AcpSessionRuntime.AcpSessionRuntime["Service"], ) => EffectAcpSchema.SessionNotification; /** * Agent updates that belong to no turn (Kiro's own cloud-config fetch). They @@ -483,6 +484,7 @@ export interface AcpAdapterV2Flavor { readonly extractSubagentSessionEvent?: ( notification: EffectAcpSchema.SessionNotification, ) => AcpSubagentSessionEvent | undefined; + readonly subagentsOwnedByRuntime?: boolean; readonly extractSubagentUpdate?: ( toolCall: AcpToolCallState, ) => AcpAdapterV2SubagentUpdate | undefined; @@ -2317,6 +2319,17 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( runtimeGeneration, Effect.gen(function* () { yield* Ref.set(runtimeRestartRequired, true); + yield* retireRuntimeSubagents(); + if (flavor.subagentsOwnedByRuntime === true) { + const context = yield* Ref.get(activeTurn); + if (context !== null) { + yield* terminalizeCarryoverSubagents({ + sessionId: context.nativeThreadId, + projectsLateTerminals: false, + subagents: [...context.subagents.values()], + }); + } + } if (!followsAgentWorkState) return; agentWork = { sessionUpdate: "state_update", state: "idle", stopReason: "error" }; for (const child of nativeChildren.values()) { @@ -6271,6 +6284,14 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( } }); + const retireRuntimeSubagents = Effect.fn("AcpAdapterV2.retireRuntimeSubagents")( + function* () { + if (flavor.subagentsOwnedByRuntime !== true) return; + const carryover = yield* Ref.getAndSet(carryoverSubagents, null); + yield* terminalizeCarryoverSubagents(carryover); + }, + ); + // Only the flavor's own start marker counts: agents also send late // updates for the previous prompt (Kiro's context usage) and echoes // that arrive before they would honor a cancel. @@ -6288,7 +6309,10 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( const projectAcpRuntimeSessionUpdateEffect = /* Provider events include streaming deltas; avoid a span for each update. */ Effect.fnUntraced( - function* (rawNotification: EffectAcpSchema.SessionNotification) { + function* ( + rawNotification: EffectAcpSchema.SessionNotification, + sourceRuntime: AcpSessionRuntime.AcpSessionRuntime["Service"], + ) { if (flavor.ignoreSessionUpdate?.(rawNotification) === true) return; yield* notePromptProgress(rawNotification); const nativePromptId = flavor.forkFromTurn?.nativeTurnId(rawNotification); @@ -6336,7 +6360,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( ); } yield* handleSessionUpdate( - flavor.normalizeSessionUpdate?.(notification) ?? notification, + flavor.normalizeSessionUpdate?.(notification, sourceRuntime) ?? notification, ); }, Effect.mapError( @@ -6350,10 +6374,11 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( const projectAcpRuntimeSessionUpdate = ( handlerGeneration: number, rawNotification: EffectAcpSchema.SessionNotification, + sourceRuntime: AcpSessionRuntime.AcpSessionRuntime["Service"], ) => runRuntimeCallbackAtGeneration( handlerGeneration, - projectAcpRuntimeSessionUpdateEffect(rawNotification), + projectAcpRuntimeSessionUpdateEffect(rawNotification, sourceRuntime), ).pipe(Effect.asVoid); // Post-settle background work retains the latest turn policy. @@ -6486,7 +6511,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( } if (handlerOptions.sessionUpdates !== false) { yield* targetRuntime.handleSessionUpdate((rawNotification) => - projectAcpRuntimeSessionUpdate(handlerGeneration, rawNotification), + projectAcpRuntimeSessionUpdate(handlerGeneration, rawNotification, targetRuntime), ); } yield* targetRuntime.handleRequestPermission((nativeParams, requestContext) => @@ -6882,6 +6907,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( resumeSessionId?: string, ) { if (runtimeScope !== undefined) { + yield* runtimeCallbackPermit.withPermit(retireRuntimeSubagents()); yield* Scope.close(runtimeScope, Exit.void); } runtimeScope = yield* Scope.make(); @@ -7029,6 +7055,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( return yield* lifecycle.error; } yield* quarantineNativeTransportAtGeneration(previousGeneration); + yield* retireRuntimeSubagents(); runtime = replacementExit.value.replacementRuntime; runtimeScope = replacementScope; runtimeMcpBridge = replacementMcpBridge; @@ -7044,13 +7071,20 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( const buffered = stagedSessionUpdates.splice(0, stagedSessionUpdates.length); if (buffered.length === 0) { handleCandidateSessionUpdate = (notification) => - projectAcpRuntimeSessionUpdate(replacementGeneration, notification); + projectAcpRuntimeSessionUpdate( + replacementGeneration, + notification, + replacementExit.value.replacementRuntime, + ); break; } yield* Effect.forEach( buffered, (notification) => - projectAcpRuntimeSessionUpdateEffect(notification).pipe( + projectAcpRuntimeSessionUpdateEffect( + notification, + replacementExit.value.replacementRuntime, + ).pipe( Effect.catchCause((cause) => Effect.logError("failed to replay staged ACP session update", { driver, @@ -8043,6 +8077,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( ) { const restartRequired = yield* Ref.get(runtimeRestartRequired); if (!restartRequired) return false; + yield* runtimeCallbackPermit.withPermit(advanceRuntimeCallbackGeneration); yield* restartAcpRuntime(threadId); yield* Ref.set(runtimeRestartRequired, false); yield* Ref.set(activeSessionId, null);