diff --git a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts index 3f5423a848e16..3f3bb07ea44db 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts @@ -614,29 +614,34 @@ const makePromptInactivityHarness = Effect.fnUntraced(function* (input: { }) { const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; const natives: Array = []; + const runtimeInputs: AcpAdapterV2RuntimeInput[] = []; const instanceId = ProviderInstanceId.make("acp-prompt-inactivity"); + const baseMakeRuntime = makeMockRuntime({ + childProcessSpawner, + mockAgentPath: yield* bundledMockAgentPath, + environment: input.environment, + wrapRuntime: (runtime) => { + natives.push(runtime); + return { + ...runtime, + prompt: input.wrapPrompt?.(runtime.prompt) ?? runtime.prompt, + cancel: + input.cancelled === undefined + ? runtime.cancel + : Deferred.succeed(input.cancelled, undefined).pipe(Effect.andThen(runtime.cancel)), + }; + }, + }); const adapter = yield* makeAcpAdapterV2({ instanceId, flavor: { driver: ACP_TEST_DRIVER, capabilities: AcpProviderCapabilitiesV2, promptInactivityTimeout: "15 minutes", - makeRuntime: makeMockRuntime({ - childProcessSpawner, - mockAgentPath: yield* bundledMockAgentPath, - environment: input.environment, - wrapRuntime: (runtime) => { - natives.push(runtime); - return { - ...runtime, - prompt: input.wrapPrompt?.(runtime.prompt) ?? runtime.prompt, - cancel: - input.cancelled === undefined - ? runtime.cancel - : Deferred.succeed(input.cancelled, undefined).pipe(Effect.andThen(runtime.cancel)), - }; - }, - }), + makeRuntime: (runtimeInput) => + Effect.sync(() => { + runtimeInputs.push(runtimeInput); + }).pipe(Effect.andThen(baseMakeRuntime(runtimeInput))), }, selfInvocation: yield* resolveSelfInvocation(), }); @@ -680,6 +685,13 @@ const makePromptInactivityHarness = Effect.fnUntraced(function* (input: { ), emitUpdate: (notification: EffectAcpSchema.SessionNotification) => natives.at(-1)!.request("_test/emit-update", notification), + /** Reports the latest runtime's transport as lost while its process keeps running. */ + terminate: () => + runtimeInputs + .at(-1)! + .onTermination( + new EffectAcpErrors.AcpTransportError({ detail: "Injected termination", cause: "test" }), + ), terminals: (count: number) => runtime.events.pipe( Stream.filter((event) => event.type === "turn.terminal"), @@ -4868,76 +4880,298 @@ describe("AcpAdapterV2", () => { }).pipe(Effect.provide(layerTest)), ); - it.live("replaces an unexpectedly terminated ACP runtime before the next turn", () => - Effect.gen(function* () { - const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; - const selfInvocation = yield* resolveSelfInvocation(); - const mockAgentPath = yield* bundledMockAgentPath; - 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 selfInvocation = yield* resolveSelfInvocation(); + const mockAgentPath = yield* bundledMockAgentPath; + 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)), + }); + 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)), ); it.effect("fails a prompt its agent leaves silent past the inactivity timeout", () => @@ -5175,6 +5409,48 @@ describe("AcpAdapterV2", () => { }).pipe(Effect.provide(layerTest)), ); + it.effect("drops an approval its finished turn left open when the runtime is replaced", () => + Effect.gen(function* () { + const approvalSeen = yield* Deferred.make(); + let prompts = 0; + const harness = yield* makePromptInactivityHarness({ + environment: (runtimeOrdinal) => + runtimeOrdinal === 1 + ? { T3_ACP_EMIT_TOOL_CALLS: "1" } + : { T3_ACP_HANG_PROMPT_FOREVER: "1" }, + runtimeMode: "approval-required", + // The first prompt returns while its approval is still unanswered. + wrapPrompt: (prompt) => (request, options) => + (prompts += 1) === 1 + ? Effect.raceFirst( + prompt(request, options), + Deferred.await(approvalSeen).pipe( + Effect.as({ stopReason: "end_turn" } satisfies EffectAcpSchema.PromptResponse), + ), + ) + : prompt(request, options), + }); + yield* harness.startTurn(1); + yield* harness.runtime.events.pipe( + Stream.filter( + (event) => + event.type === "runtime_request.updated" && event.runtimeRequest.status === "pending", + ), + Stream.runHead, + ); + yield* Deferred.succeed(approvalSeen, undefined); + yield* harness.terminals(1); + yield* harness.terminate(); + + // A leftover approval would pause the replacement's inactivity deadline. + yield* harness.startTurn(2); + yield* TestClock.adjust("15 minutes"); + assert.isFalse(yield* harness.nextTurnRejected(3)); + const [silent] = yield* harness.terminals(1); + assert.equal(silent?.status, "failed"); + }).pipe(Effect.provide(layerTest)), + ); + it.live("does not poll the process ledger while a cgroup lease owns teardown", () => Effect.gen(function* () { if ((yield* HostProcess.Platform) !== "linux") return; diff --git a/packages/provider-acp-registry/src/server/adapter.ts b/packages/provider-acp-registry/src/server/adapter.ts index 360ba35e8cc5c..e874bb9d64ccc 100644 --- a/packages/provider-acp-registry/src/server/adapter.ts +++ b/packages/provider-acp-registry/src/server/adapter.ts @@ -280,6 +280,7 @@ export const makeAcpRegistryAdapterV2 = Effect.fn("makeAcpRegistryAdapterV2")(fu registerExtensions: copilotRouting.registerExtensions, normalizeSessionUpdate: copilotRouting.normalizeSessionUpdate, extractSubagentUpdate: extractCopilotSubagentUpdate, + subagentsOwnedByRuntime: true, restartRuntimeAfterPromptError: isCopilotLostSessionError, // session/new reports a built-in default, not the model Copilot runs (github/copilot-cli#5363). alwaysSetModel: true, diff --git a/packages/provider-acp-registry/src/server/copilotAcp.ts b/packages/provider-acp-registry/src/server/copilotAcp.ts index 8482886892985..a2031a144bc41 100644 --- a/packages/provider-acp-registry/src/server/copilotAcp.ts +++ b/packages/provider-acp-registry/src/server/copilotAcp.ts @@ -69,8 +69,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() }; @@ -78,80 +78,96 @@ 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); - } - 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); + 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; } - 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, - }); - } - // Copilot sends no ACP compaction_update; agentId marks a subagent's own compaction. - // Older Copilot builds report a successful compaction without `outcome`. - case "session.compaction_complete": { - if ( - event.agentId !== undefined || - data.success !== true || - (data.outcome !== undefined && data.outcome !== "applied") - ) { + case "subagent.started": { + if (event.agentId !== undefined && typeof data.toolCallId === "string") { + session(runtimeSessions, event.sessionId).taskByAgentId.set( + event.agentId, + data.toolCallId, + ); + } return Effect.void; } - return context.markSessionCompacted(event.sessionId); - } - // Copilot holds session/prompt while an attached async shell runs (github/copilot-cli#5293). - case "assistant.turn_start": - case "assistant.idle": { - if (event.agentId !== undefined || data.aborted === true) return Effect.void; - return context.reportForegroundWork({ - sessionId: event.sessionId, - state: event.type === "assistant.idle" ? "idle" : "running", - }); + 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, + }); + } + // Copilot sends no ACP compaction_update; agentId marks a subagent's own compaction. + // Older Copilot builds report a successful compaction without `outcome`. + case "session.compaction_complete": { + if ( + event.agentId !== undefined || + data.success !== true || + (data.outcome !== undefined && data.outcome !== "applied") + ) { + return Effect.void; + } + return context.markSessionCompacted(event.sessionId); + } + // Copilot holds session/prompt while an attached async shell runs (github/copilot-cli#5293). + case "assistant.turn_start": + case "assistant.idle": { + if (event.agentId !== undefined || data.aborted === true) return Effect.void; + return context.reportForegroundWork({ + sessionId: event.sessionId, + state: event.type === "assistant.idle" ? "idle" : "running", + }); + } + 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 ( @@ -160,7 +176,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 dc12bd7aa1e14..411c6951d2637 100644 --- a/packages/provider-acp/src/server/adapter.ts +++ b/packages/provider-acp/src/server/adapter.ts @@ -382,6 +382,7 @@ export interface AcpAdapterV2Flavor { readonly deferMcpTools?: 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 @@ -548,6 +549,8 @@ export interface AcpAdapterV2Flavor { readonly extractSubagentSessionEvent?: ( notification: EffectAcpSchema.SessionNotification, ) => AcpSubagentSessionEvent | undefined; + /** Native subagents run inside the agent process, so they end when its runtime does. */ + readonly subagentsOwnedByRuntime?: boolean; readonly extractSubagentUpdate?: ( toolCall: AcpToolCallState, ) => AcpAdapterV2SubagentUpdate | undefined; @@ -2524,6 +2527,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()) { @@ -6604,6 +6618,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. @@ -6621,7 +6643,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); @@ -6675,7 +6700,8 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( ) ?? Effect.void ); } - const normalized = flavor.normalizeSessionUpdate?.(notification) ?? notification; + const normalized = + flavor.normalizeSessionUpdate?.(notification, sourceRuntime) ?? notification; // Turn projection misses compactions that finish while idle or after settlement. if ( normalized.update.sessionUpdate === "compaction_update" && @@ -6696,10 +6722,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. @@ -6832,7 +6859,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) => @@ -7254,6 +7281,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(); @@ -7401,6 +7429,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; @@ -7416,13 +7445,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, @@ -8459,6 +8495,8 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( ) { const restartRequired = yield* Ref.get(runtimeRestartRequired); if (!restartRequired) return false; + yield* runtimeCallbackPermit.withPermit(advanceRuntimeCallbackGeneration); + yield* cancelPendingRuntimeRequests(); yield* restartAcpRuntime(threadId); yield* Ref.set(runtimeRestartRequired, false); yield* Ref.set(activeSessionId, null);