diff --git a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts index f63a220fb5191..4f6991e3a2abb 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts @@ -9328,6 +9328,15 @@ describe("AcpAdapterV2", () => { assert.equal(backgroundItem.turnItem.output ?? "", output); } }); + // A foreground tool still open at end_turn must not hold the deferred finalize. + yield* update({ + sessionUpdate: "tool_call", + toolCallId: "stale-foreground", + title: "Read file", + kind: "read", + status: "in_progress", + rawInput: { path: "a.ts" }, + }); yield* Deferred.succeed(promptGate, { stopReason: "end_turn" }); yield* Deferred.await(promptHeld); yield* assertCommand(status, "EARLY_BG_DONE\n"); diff --git a/apps/server/src/orchestration-v2/Adapters/AcpRegistryAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AcpRegistryAdapterV2.test.ts index 1767c0dfceca4..3cd728d4f7d94 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpRegistryAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpRegistryAdapterV2.test.ts @@ -17,6 +17,7 @@ import { pendingBackgroundTurnItems } from "@t3tools/shared/orchestrationV2Pendi import { resolveSelfInvocation, selfInvocationArgs } from "@t3tools/shared/nodeRuntime"; import * as Exit from "effect/Exit"; import * as Fiber from "effect/Fiber"; +import * as TestClock from "effect/testing/TestClock"; import * as DateTime from "effect/DateTime"; import * as Queue from "effect/Queue"; import * as Stream from "effect/Stream"; @@ -275,7 +276,7 @@ const openClaudeReplaySession = Effect.fn("openClaudeReplaySession")(function* ( runtimePolicy, }); yield* Effect.addFinalizer(() => Effect.sync(() => replayGate?.releaseAll())); - if (continuationEnabled) { + if (continuationEnabled && agentId === "claude-acp") { assert.isDefined( runtime.hasPendingBackgroundWork, "Claude ACP v2 must report background work after foreground idle", @@ -379,6 +380,29 @@ const claudeMarker = (title: string) => sessionUpdate: "session_info_update", title, }); +// Factory Droid speaks ACP v1, so its prompt ends with a `session/prompt` response. +const droidFirstTurn = (...frames: ReadonlyArray) => [ + ...claudeFrames.open.map((entry) => + entry.frame.method === "initialize" && entry.type === "emit_inbound" + ? { + ...entry, + frame: { + ...entry.frame, + result: { protocolVersion: 1, agentCapabilities: {}, authMethods: [] }, + }, + } + : entry, + ), + { + type: "expect_outbound", + frame: { kind: "request", method: "session/prompt", params: "" }, + }, + ...frames, + { + type: "emit_inbound", + frame: { kind: "response", method: "session/prompt", result: { stopReason: "end_turn" } }, + }, +]; const claudeChild = [ claudeFrames.update(CLAUDE_ROOT_SESSION, { sessionUpdate: "tool_call_update", @@ -2093,6 +2117,385 @@ describe("AcpRegistryAdapterV2", () => { }).pipe(Effect.provide(layerTest), Effect.scoped), ); + it.effect("continues Factory Droid when its background Task finishes after end_turn", () => + Effect.gen(function* () { + const { update } = claudeFrames; + const gate = makeProviderReplayGate(["task-completed", "answer"]); + const launchTask = (toolCallId: string) => + update(CLAUDE_ROOT_SESSION, { + sessionUpdate: "tool_call", + toolCallId, + title: "Task", + kind: "other", + status: "completed", + rawInput: { subagent_type: "explorer", prompt: `sleep 20; echo ${toolCallId}` }, + rawOutput: { type: "Text", text: "Task launched in background" }, + }); + // A Task the previous run never saw must still merge with the run that launched it. + const assertTaskCard = ( + events: ReadonlyArray, + toolCallId: string, + ) => { + const task = events.findLast( + (event) => + event.type === "turn_item.updated" && + event.turnItem.nativeItemRef?.nativeId === toolCallId, + ); + if (task?.type !== "turn_item.updated") throw new Error(`Missing ${toolCallId} update`); + assert.include(task.turnItem, { title: "Task", status: "completed" }); + assert.deepInclude(task.turnItem, { + input: { subagent_type: "explorer", prompt: `sleep 20; echo ${toolCallId}` }, + }); + }; + const session = yield* openClaudeReplaySession( + [ + ...droidFirstTurn( + launchTask("task-1"), + launchTask("task-2"), + update(CLAUDE_ROOT_SESSION, { + sessionUpdate: "agent_message_chunk", + content: { type: "text", text: "waiting" }, + }), + ), + { + ...update(CLAUDE_ROOT_SESSION, { + sessionUpdate: "tool_call_update", + toolCallId: "task-1", + status: "completed", + rawOutput: { type: "Text", text: "The command ran and exited with code 0." }, + }), + label: "task-completed", + }, + claudeMarker("task done"), + { + ...update(CLAUDE_ROOT_SESSION, { + sessionUpdate: "tool_call_update", + toolCallId: "task-2", + status: "completed", + rawOutput: { type: "Text", text: "The second task exited with code 0." }, + }), + label: "answer", + }, + update(CLAUDE_ROOT_SESSION, { + sessionUpdate: "usage_update", + used: 2_000, + size: 10_000, + }), + update(CLAUDE_ROOT_SESSION, { + sessionUpdate: "agent_message_chunk", + content: { type: "text", text: "The background task exited with code 0." }, + }), + claudeMarker("late answer"), + claudeFrames.closeRequest, + claudeFrames.closed, + ], + gate, + true, + "factory-droid", + ); + yield* session.runtime.startTurn(session.turnInput(1)); + assert.equal((yield* session.terminal).status, "completed"); + gate.release("task-completed"); + yield* session.throughMarker("task done"); + assert.lengthOf(session.offers, 1, "Droid's Task completion must offer a continuation run"); + yield* session.offers[0]!.dispatchIfCurrent!(Effect.void); + yield* session.runtime.startTurn(session.turnInput(2, { continuation: true })); + const taskSettled = yield* Effect.forkScoped( + session.until((event) => event.type === "turn.terminal"), + ); + yield* TestClock.adjust("3 seconds"); + const taskRun = yield* Fiber.join(taskSettled); + assert.include(taskRun.at(-1), { runOrdinal: 2, status: "completed" }); + assertTaskCard(taskRun, "task-1"); + // Droid answers ~4 s after the Task result, past the quiet window. + gate.release("answer"); + yield* session.throughMarker("late answer"); + assert.lengthOf(session.offers, 2, "Droid's late answer must offer another continuation"); + yield* session.offers[1]!.dispatchIfCurrent!(Effect.void); + yield* session.runtime.startTurn(session.turnInput(3, { continuation: true })); + const answerRun = yield* session.until( + (event) => + event.type === "message.updated" && + event.message.text === "The background task exited with code 0.", + ); + assertTaskCard(answerRun, "task-2"); + const answer = answerRun.at(-1); + if (answer?.type !== "message.updated") throw new Error("Missing late answer"); + assert.equal(answer.message.runId, session.turnInput(3).runId); + const answerSettled = yield* Effect.forkScoped(session.terminal); + yield* TestClock.adjust("3 seconds"); + assert.include(yield* Fiber.join(answerSettled), { runOrdinal: 3, status: "completed" }); + }).pipe(Effect.provide(layerTest), Effect.scoped), + ); + + it.effect.each([ + { + blocker: "an unanswered question", + frames: [ + { + type: "emit_inbound", + label: "blocker", + frame: { + kind: "request", + method: "elicitation/create", + params: { + sessionId: CLAUDE_ROOT_SESSION, + mode: "form", + message: "Which file?", + requestedSchema: { + type: "object", + properties: { + question_1: { + type: "string", + title: "File", + description: "Which file should I inspect?", + oneOf: [ + { const: "option_0", title: "a.ts" }, + { const: "other", title: "Other" }, + ], + }, + question_1_other: { type: "string" }, + }, + required: ["question_1"], + }, + }, + }, + }, + { ...claudeMarker("still waiting"), label: "check" }, + { + type: "expect_outbound", + frame: { + kind: "response", + method: "elicitation/create", + result: { action: "accept", content: { question_1: "option_0" } }, + }, + }, + ], + response: { answers: { question_1: "option_0" } }, + }, + { + blocker: "an unanswered approval", + frames: [ + { + type: "emit_inbound", + label: "blocker", + frame: { + kind: "request", + method: "session/request_permission", + params: { + sessionId: CLAUDE_ROOT_SESSION, + toolCall: { + toolCallId: "approve-command", + title: "Run tests", + kind: "execute", + status: "pending", + rawInput: { command: "vp test run" }, + }, + options: [{ optionId: "allow", name: "Allow", kind: "allow_once" }], + }, + }, + }, + { ...claudeMarker("still waiting"), label: "check" }, + { + type: "expect_outbound", + frame: { + kind: "response", + method: "session/request_permission", + result: { outcome: { outcome: "selected", optionId: "allow" } }, + }, + }, + ], + response: { decision: "accept" as const }, + }, + { + blocker: "a running tool", + frames: [ + { + ...claudeFrames.update(CLAUDE_ROOT_SESSION, { + sessionUpdate: "tool_call", + toolCallId: "slow-tool", + title: "Execute", + kind: "execute", + status: "in_progress", + rawInput: { command: "sleep 20; echo done" }, + }), + label: "blocker", + }, + claudeMarker("blocked"), + { ...claudeMarker("still waiting"), label: "check" }, + { + ...claudeFrames.update(CLAUDE_ROOT_SESSION, { + sessionUpdate: "tool_call_update", + toolCallId: "slow-tool", + status: "completed", + rawOutput: { type: "Text", text: "done" }, + }), + label: "resolve", + }, + claudeMarker("resolved"), + ], + response: undefined, + }, + ])( + "keeps a Factory Droid continuation open past the quiet window for $blocker", + ({ frames, response }) => + Effect.gen(function* () { + const gate = makeProviderReplayGate(["wake", "blocker", "check", "resolve"]); + const session = yield* openClaudeReplaySession( + [ + ...droidFirstTurn(), + { + ...claudeFrames.update(CLAUDE_ROOT_SESSION, { + sessionUpdate: "agent_message_chunk", + content: { type: "text", text: "Checking the background result" }, + }), + label: "wake", + }, + claudeMarker("woke"), + ...frames, + claudeFrames.closeRequest, + claudeFrames.closed, + ], + gate, + true, + "factory-droid", + ); + yield* session.runtime.startTurn(session.turnInput(1)); + assert.equal((yield* session.terminal).status, "completed"); + gate.release("wake"); + yield* session.throughMarker("woke"); + yield* session.offers[0]!.dispatchIfCurrent!(Effect.void); + yield* session.runtime.startTurn(session.turnInput(2, { continuation: true })); + gate.release("blocker"); + const blocked = (yield* session.until( + (event) => + (event.type === "runtime_request.updated" && + event.runtimeRequest.status === "pending") || + (event.type === "provider_thread.updated" && + event.providerThread.nativeMetadata?.title === "blocked"), + )).at(-1)!; + yield* TestClock.adjust("3 seconds"); + gate.release("check"); + assert.notInclude( + (yield* session.throughMarker("still waiting")).map((event) => event.type), + "turn.terminal", + "the continuation must not settle while Droid still waits", + ); + // Requests are answered with no later update, so only their resolution rearms the finalize. + if (blocked.type === "runtime_request.updated" && response !== undefined) { + yield* session.runtime.respondToRuntimeRequest({ + requestId: blocked.runtimeRequest.id, + ...response, + }); + } else { + gate.release("resolve"); + yield* session.throughMarker("resolved"); + } + const settled = yield* Effect.forkScoped(session.terminal); + yield* TestClock.adjust("3 seconds"); + assert.include(yield* Fiber.join(settled), { runOrdinal: 2, status: "completed" }); + }).pipe(Effect.provide(layerTest), Effect.scoped), + ); + + const permissionRequest = (toolCallId: string, kind: string) => ({ + kind: "request", + method: "session/request_permission", + params: { + sessionId: CLAUDE_ROOT_SESSION, + toolCall: { toolCallId, title: toolCallId, kind, status: "pending", rawInput: {} }, + options: [{ optionId: "allow", name: "Allow", kind: "allow_once" }], + }, + }); + it.effect.each([ + { + request: "a question", + frame: { + kind: "request", + method: "elicitation/create", + params: { + sessionId: CLAUDE_ROOT_SESSION, + mode: "form", + message: "Which file?", + requestedSchema: { + type: "object", + properties: { + question_1: { + type: "string", + title: "File", + oneOf: [{ const: "option_0", title: "a.ts" }], + }, + }, + required: ["question_1"], + }, + }, + }, + reply: { + method: "elicitation/create", + result: { action: "accept", content: { question_1: "option_0" } }, + }, + response: { answers: { question_1: "option_0" } }, + }, + { + request: "an approval", + frame: permissionRequest("run-tests", "execute"), + reply: { + method: "session/request_permission", + result: { outcome: { outcome: "selected", optionId: "allow" } }, + }, + response: { decision: "accept" as const }, + }, + { + request: "an auto-approved read", + frame: permissionRequest("read-file", "read"), + reply: { + method: "session/request_permission", + result: { outcome: { outcome: "selected", optionId: "allow" } }, + }, + response: undefined, + }, + ])("admits $request Factory Droid sends after its run settled", ({ frame, reply, response }) => + Effect.gen(function* () { + const gate = makeProviderReplayGate(["request"]); + const session = yield* openClaudeReplaySession( + [ + ...droidFirstTurn(), + { type: "emit_inbound", label: "request", frame }, + { type: "expect_outbound", frame: { kind: "response", ...reply } }, + claudeMarker("answered"), + claudeFrames.closeRequest, + claudeFrames.closed, + ], + gate, + true, + "factory-droid", + ); + yield* session.runtime.startTurn(session.turnInput(1)); + assert.equal((yield* session.terminal).status, "completed"); + gate.release("request"); + if (response === undefined) { + yield* session.throughMarker("answered"); + assert.lengthOf(session.offers, 0, "an automatic decision needs no continuation"); + return; + } + yield* Queue.take(session.offered); + yield* session.offers[0]!.dispatchIfCurrent!(Effect.void); + yield* session.runtime.startTurn(session.turnInput(2, { continuation: true })); + const pending = (yield* session.until( + (event) => + event.type === "runtime_request.updated" && event.runtimeRequest.status === "pending", + )).at(-1); + if (pending?.type !== "runtime_request.updated") throw new Error("Missing request"); + yield* session.runtime.respondToRuntimeRequest({ + requestId: pending.runtimeRequest.id, + ...response, + }); + yield* session.throughMarker("answered"); + const settled = yield* Effect.forkScoped(session.terminal); + yield* TestClock.adjust("3 seconds"); + assert.include(yield* Fiber.join(settled), { runOrdinal: 2, status: "completed" }); + }).pipe(Effect.provide(layerTest), Effect.scoped), + ); + it.effect.each(["error", "end_turn"])( "hands live Claude child tools to the next run after %s and keeps nested output in the child", (stopReason) => diff --git a/packages/provider-acp-registry/src/server/adapter.ts b/packages/provider-acp-registry/src/server/adapter.ts index 1b8b4240fd00e..8210f634c82d6 100644 --- a/packages/provider-acp-registry/src/server/adapter.ts +++ b/packages/provider-acp-registry/src/server/adapter.ts @@ -221,7 +221,13 @@ export const makeAcpRegistryAdapterV2 = Effect.fn("makeAcpRegistryAdapterV2")(fu customModels: readCustomModelEntries(options.settings.customModels).map((entry) => entry.slug), promptFailure: (cause) => acpRegistryPromptFailure(registryAgentId, cause), ...(registryAgentId === "factory-droid" - ? { extractElicitationForm: extractDroidElicitationForm } + ? { + extractElicitationForm: extractDroidElicitationForm, + // Droid wakes itself with root-session output after a background Task ends; with + // no prompt in flight, that continuation settles after the quiet window. + enablePostSettleContinuation: true, + deferFinalizeForBackgroundWork: true, + } : {}), // glm-acp-agent is the one agent whose per-turn usage a live probe has verified. ...(registryAgentId === "glm-acp-agent" ? { reportsPerTurnPromptUsage: true } : {}), diff --git a/packages/provider-acp/src/server/adapter.ts b/packages/provider-acp/src/server/adapter.ts index a2c2c429fa88c..9425df837216c 100644 --- a/packages/provider-acp/src/server/adapter.ts +++ b/packages/provider-acp/src/server/adapter.ts @@ -1369,6 +1369,8 @@ interface ActiveAcpTurn { readonly tools: Map; readonly toolUpdatesSkipped: Map; readonly toolStartedAt: Map; + /** A continuation's sparse tool updates merge from the finalized run before it. */ + readonly precedingRunTools: AcpRunTools | null; readonly subagents: Map; readonly subagentsBySessionId: Map; readonly pendingSubagentNotifications: Map>; @@ -1612,6 +1614,8 @@ function acpSubagentHasPendingBackgroundWork(subagent: ActiveAcpSubagent): boole ); } +type AcpRunTools = Pick; + type AcpCarryoverSubagents = { readonly sessionId: string; /** The root's subscriber keeps ingesting, so late child terminals can still project. */ @@ -2208,6 +2212,9 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( const stoppedRunQuarantine = yield* Ref.make(false); // Native subagents can outlive interrupts; carry lineage forward to finish their original turn items. const carryoverSubagents = yield* Ref.make(null); + const finalizedRunTools = yield* Ref.make< + (AcpRunTools & { readonly sessionId: string }) | null + >(null); const handledBackgroundTaskIdsInActiveTurn = yield* Ref.make>( new Set(), ); @@ -3771,6 +3778,31 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( acpSubagentStatusBlocksTurnSettlement(subagent.task.status), ); + const awaitsRequestOrTool = Effect.fnUntraced(function* ( + context: ActiveAcpTurn, + countsTool: (tool: AcpToolCallState) => boolean, + ) { + if (openAgentRequests.size > 0) return true; + if ((yield* Ref.get(pendingRuntimeRequests)).size > 0) return true; + for (const tool of context.tools.values()) { + const status = toolStatus(tool.status); + if (status !== "pending" && status !== "running" && status !== "waiting") continue; + if (countsTool(tool)) return true; + } + return false; + }); + + // Background tools and subagents are `hasDeferredBackgroundWork`'s, which skips + // persistent tasks that never end. + const isForegroundTool = (tool: AcpToolCallState) => + flavor.extractBackgroundTaskId?.(tool) === undefined && + extractSubagentUpdate(tool) === undefined; + + const blocksDeferredFinalize = Effect.fnUntraced(function* (context: ActiveAcpTurn) { + if (hasDeferredBackgroundWork(context)) return true; + return yield* awaitsRequestOrTool(context, isForegroundTool); + }); + const rearmDeferredFinalize = /* Provider events include streaming deltas; avoid a span for each update. */ Effect.fnUntraced(function* (context: ActiveAcpTurn) { @@ -3779,6 +3811,17 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( if (hasDeferredBackgroundWork(context)) return; yield* scheduleDeferredFinalize(context); }); + const rearmActiveDeferredFinalize = Ref.get(activeTurn).pipe( + Effect.flatMap((context) => + context === null ? Effect.void : rearmDeferredFinalize(context), + ), + ); + const forgetPendingRuntimeRequest = (requestId: string) => + Ref.update(pendingRuntimeRequests, (current) => { + const updated = new Map(current); + updated.delete(requestId); + return updated; + }).pipe(Effect.andThen(rearmActiveDeferredFinalize)); // let breaks circular inference from monitor hydration re-entry. let emitTool: ( @@ -3843,7 +3886,9 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( if (context.reasoning.current?.sourceMessageId == null) yield* closeTextStream(context, "reasoning"); yield* closeTextStream(context, "user"); - const previous = context.tools.get(incoming.toolCallId); + const previous = + context.tools.get(incoming.toolCallId) ?? + context.precedingRunTools?.tools.get(incoming.toolCallId); const merged = mergeToolCallState(previous, incoming); const normalized = flavor.normalizeToolCall?.(merged) ?? merged; const backgroundTaskId = flavor.extractBackgroundTaskId?.(normalized); @@ -4013,7 +4058,10 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( nativeId: toolCall.toolCallId, strength: "strong" as const, }; - const startedAt = context.toolStartedAt.get(toolCall.toolCallId) ?? now; + const startedAt = + context.toolStartedAt.get(toolCall.toolCallId) ?? + context.precedingRunTools?.toolStartedAt.get(toolCall.toolCallId) ?? + now; context.toolStartedAt.set(toolCall.toolCallId, startedAt); const completedAt = completedAtForStatus(status, now); const title = toolCall.title ?? null; @@ -4921,8 +4969,12 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( ); }); + // A post-settle agent can ask after its run settled; the request opens a continuation. + const admitsRequestsOutsideTurn = () => + followsAgentWorkState || postSettleContinuationEnabled; + const awaitRequestTurn = Effect.fnUntraced(function* (generation: number) { - if (!followsAgentWorkState) return true; + if (!admitsRequestsOutsideTurn()) return true; while (true) { const waiting = yield* runRuntimeCallbackAtGeneration( generation, @@ -5868,11 +5920,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( Effect.ensuring( runRuntimeCallbackAtGeneration( generation, - Ref.update(pendingRuntimeRequests, (current) => { - const updated = new Map(current); - updated.delete(String(requestId)); - return updated; - }), + forgetPendingRuntimeRequest(String(requestId)), ).pipe(Effect.asVoid), ), ); @@ -6524,14 +6572,14 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( policy: ProviderAdapter.ProviderAdapterV2RuntimePolicy, request: EffectAcpSchema.RequestPermissionRequest = params, ) => (flavor.permissionDisposition ?? acpPermissionDisposition)(policy, request); - if (followsAgentWorkState && dispositionFor(latestRuntimePolicy) === "ask") { + if (admitsRequestsOutsideTurn() && dispositionFor(latestRuntimePolicy) === "ask") { if (!(yield* awaitRequestTurn(handlerGeneration))) return { outcome: { outcome: "cancelled" } } as const; } const admitted = yield* runRuntimeCallbackAtGeneration( handlerGeneration, Effect.gen(function* () { - const context = followsAgentWorkState + const context = admitsRequestsOutsideTurn() ? yield* Ref.get(activeTurn) : yield* activeContext; // Kiro sends the tool's kind on its `tool_call` and leaves it @@ -6609,11 +6657,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( Effect.ensuring( runRuntimeCallbackAtGeneration( handlerGeneration, - Ref.update(pendingRuntimeRequests, (current) => { - const updated = new Map(current); - updated.delete(String(requestId)); - return updated; - }), + forgetPendingRuntimeRequest(String(requestId)), ).pipe(Effect.asVoid), ), ); @@ -7559,7 +7603,8 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( for (const tool of context.tools.values()) { if (!acpCompletedTurnShouldTerminalizeTool(tool, { ...flavor, extractSubagentUpdate })) continue; - yield* emitTool(context, tool, status); + // Store the status too, so the deferred finalize stops counting the tool as open. + yield* emitTool(context, { ...tool, status }, status); } }); @@ -7764,6 +7809,16 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( }); } } + if (postSettleContinuationEnabled && settledStatus === "completed") { + yield* Ref.set(finalizedRunTools, { + sessionId: context.nativeThreadId, + tools: new Map([...(context.precedingRunTools?.tools ?? []), ...context.tools]), + toolStartedAt: new Map([ + ...(context.precedingRunTools?.toolStartedAt ?? []), + ...context.toolStartedAt, + ]), + }); + } yield* Ref.set(activeTurn, null); steerInjections.clear(); if ( @@ -7791,7 +7846,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( scheduleDeferredFinalize = Effect.fn("scheduleDeferredFinalize")(function* (context) { if (!flavor.deferFinalizeForBackgroundWork) return; if (!context.promptSettled || context.finalized || context.interrupted) return; - if (hasDeferredBackgroundWork(context)) return; + if (yield* blocksDeferredFinalize(context)) return; context.backgroundFinalizeGeneration += 1; const generation = context.backgroundFinalizeGeneration; yield* Effect.gen(function* () { @@ -7803,7 +7858,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( ) { return; } - if (hasDeferredBackgroundWork(context)) return; + if (yield* blocksDeferredFinalize(context)) return; const status = context.promptSettledStatus ?? "completed"; yield* finalizeTurn(context, status); }).pipe(Effect.forkIn(sessionScope), Effect.asVoid); @@ -7816,12 +7871,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( const promptInactivityPaused = Effect.fn("AcpAdapterV2.promptInactivityPaused")(function* ( context: ActiveAcpTurn, ) { - if (openAgentRequests.size > 0) return true; - if ((yield* Ref.get(pendingRuntimeRequests)).size > 0) return true; - for (const tool of context.tools.values()) { - const status = toolStatus(tool.status); - if (status === "pending" || status === "running" || status === "waiting") return true; - } + if (yield* awaitsRequestOrTool(context, () => true)) return true; for (const subagent of context.subagents.values()) { if (acpSubagentStatusBlocksTurnSettlement(subagent.task.status)) return true; } @@ -8171,6 +8221,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( new Map(current).set(requestedSessionId, initialNativeMetadata), ); } + const precedingRunTools = yield* Ref.getAndSet(finalizedRunTools, null); const context: ActiveAcpTurn = { input: turnInput, providerTurnId, @@ -8191,6 +8242,10 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( tools: new Map(), toolUpdatesSkipped: new Map(), toolStartedAt: new Map(), + precedingRunTools: + isContinuationTurn && precedingRunTools?.sessionId === requestedSessionId + ? precedingRunTools + : null, subagents: new Map(), subagentsBySessionId: new Map(), pendingSubagentNotifications: new Map(), @@ -8463,6 +8518,9 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( !context.interrupted && hasDeferredBackgroundWork(context) ) { + // A foreground tool still open at end_turn is stale; left open it would + // hold the deferred finalize until Stop. + yield* terminalizeOpenForegroundTools(context); yield* closeTextStreams(context); yield* ( options.testHooks?.afterPromptSettledWithBackgroundWork?.() ?? Effect.void