diff --git a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts index 02d6f64c0be66..cf0975b725f32 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts @@ -891,6 +891,227 @@ function assertReinjectedInstructions( assert.include(blocks[1]?.text ?? "", ""); } +// With `hydration`, an outputless task_completed meets a TaskOutput result in `order`. +const runGrokMonitorEndingMidTurn = Effect.fnUntraced(function* (input: { + readonly hydration?: "completed" | "failed"; + readonly order?: "hydration-last" | "hydration-first"; + readonly hydrations?: number; +}) { + 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), + ); + type RuntimeService = AcpSessionRuntime.AcpSessionRuntime["Service"]; + let sessionUpdateHandler: Parameters[0] | undefined; + const extNotifications = new Map< + string, + (payload: unknown) => Effect.Effect + >(); + const promptStarted = yield* Deferred.make(); + const promptGate = yield* Deferred.make(); + const promptHeld = yield* Deferred.make(); + const instanceId = ProviderInstanceId.make("acp-test"); + const grokFlavor = makeGrokAcpAdapterFlavor( + { + instanceId, + settings: DEFAULT_GROK_SETTINGS, + environment: {}, + hostPlatform: yield* HostProcess.Platform, + selfInvocation, + }, + yield* HostProcess.HomeDirectory, + ); + const adapter = yield* makeAcpAdapterV2({ + instanceId, + selfInvocation, + testHooks: { + afterPromptSettledWithBackgroundWork: () => + Deferred.succeed(promptHeld, undefined).pipe(Effect.asVoid), + }, + flavor: { + driver: ACP_TEST_DRIVER, + capabilities: AcpProviderCapabilitiesV2, + enablePostSettleContinuation: true, + deferFinalizeForBackgroundWork: true, + normalizeToolCall: grokFlavor.normalizeToolCall!, + extractBackgroundTaskId: grokFlavor.extractBackgroundTaskId!, + isBackgroundCommandTool: grokFlavor.isBackgroundCommandTool!, + isPersistentBackgroundTool: grokFlavor.isPersistentBackgroundTool!, + extractBackgroundTaskCompletion: grokFlavor.extractBackgroundTaskCompletion!, + registerExtensions: ({ runtime, applyBackgroundTaskMutation }) => + registerXAiBackgroundTaskTracking(runtime, applyBackgroundTaskMutation), + makeRuntime: makeMockRuntime({ + childProcessSpawner, + mockAgentPath, + wrapRuntime: (runtime) => ({ + ...runtime, + handleSessionUpdate: (handler) => + Effect.sync(() => { + sessionUpdateHandler = handler; + }).pipe(Effect.andThen(runtime.handleSessionUpdate(handler))), + handleExtNotification: (method, schema, handler) => + Effect.sync(() => { + extNotifications.set(method, (payload) => + handler(Schema.decodeUnknownSync(schema)(payload)), + ); + }).pipe(Effect.andThen(runtime.handleExtNotification(method, schema, handler))), + prompt: () => + Deferred.succeed(promptStarted, undefined).pipe( + Effect.andThen(Deferred.await(promptGate)), + ), + }), + }), + }, + continuationRequests: { offer: () => Effect.void }, + }); + const threadId = ThreadId.make("thread-acp-grok-monitor-completed-mid-turn"); + 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-grok-monitor-mid-turn"), + modelSelection, + runtimePolicy, + }); + const monitorToolCallId = "call-monitor-job1"; + const monitorStatuses: Array = []; + const monitorOutputs: Array = []; + const terminal = yield* Deferred.make(); + yield* runtime.events.pipe( + Stream.runForEach((event) => + Effect.gen(function* () { + if ( + event.type === "turn_item.updated" && + event.turnItem.nativeItemRef?.nativeId === monitorToolCallId + ) { + monitorStatuses.push(event.turnItem.status); + if (event.turnItem.type === "command_execution") { + monitorOutputs.push(event.turnItem.output ?? ""); + } + } + if (event.type === "turn.terminal") { + yield* Deferred.succeed(terminal, event.status); + } + }), + ), + Effect.forkScoped, + ); + const providerThread = yield* runtime.ensureThread({ + threadId, + modelSelection, + runtimePolicy, + }); + const now = yield* DateTime.now; + yield* runtime + .startTurn( + makeTurnInput({ threadId, providerThread, instanceId, runtimePolicy, now, ordinal: 1 }), + ) + .pipe(Effect.forkScoped); + yield* Deferred.await(promptStarted); + const update = (update: EffectAcpSchema.SessionUpdate) => + sessionUpdateHandler!({ sessionId: "mock-session-1", update }); + const ext = (method: string, update: Record) => + extNotifications.get(method)!({ sessionId: "mock-session-1", update }); + + // Frame order from Grok CLI 1.0.70; every monitor frame precedes the prompt response. + const taskId = "a9b6879f-acea-447f-8d25-e0d4bbb2f2a9"; + const command = "while [ ! -f job1.done ]; do sleep 2; done; echo job1 finished"; + const xAiMonitorMeta = { + "x.ai/tool": { version: 1, name: "monitor", kind: "monitor", label: "Monitor" }, + }; + yield* update({ + sessionUpdate: "tool_call", + toolCallId: monitorToolCallId, + title: "monitor", + rawInput: { command, description: "Watch job1 done file" }, + _meta: xAiMonitorMeta, + }); + yield* update({ + sessionUpdate: "tool_call_update", + toolCallId: monitorToolCallId, + kind: "other", + title: "Start monitor: Watch job1 done file", + rawInput: { + variant: "Monitor", + command, + description: "Watch job1 done file", + timeout_ms: 36000000, + persistent: false, + }, + _meta: xAiMonitorMeta, + }); + yield* ext("_x.ai/task_backgrounded", { + sessionUpdate: "task_backgrounded", + tool_call_id: monitorToolCallId, + task_id: taskId, + command, + monitor_description: "Watch job1 done file", + description: "Watch job1 done file", + }); + yield* update({ + sessionUpdate: "tool_call_update", + toolCallId: monitorToolCallId, + status: "completed", + rawOutput: { type: "Monitor", taskId, timeoutMs: 36000000, persistent: false }, + }); + const taskCompleted = ext("_x.ai/task_completed", { + sessionUpdate: "task_completed", + task_snapshot: { + task_id: taskId, + command, + display_command: "[monitor] Watch job1 done file", + ...(input.hydration === undefined ? { output: "job1 finished\n" } : {}), + }, + }); + const hydration = input.hydration; + const hydrate = + hydration === undefined + ? Effect.void + : Effect.forEach( + Array.from({ length: input.hydrations ?? 1 }, (_, index) => index), + (index) => + update({ + sessionUpdate: "tool_call", + toolCallId: `call-fetch-job1-${index}`, + title: "get_command_or_subagent_output", + status: "completed", + rawInput: { task_id: taskId }, + rawOutput: { + type: "TaskOutput", + Result: { task_id: taskId, status: hydration, output: "HYDRATED_RESULT\n" }, + }, + }), + { discard: true }, + ); + if (input.order === "hydration-first") { + yield* hydrate; + yield* taskCompleted; + } else { + yield* taskCompleted; + yield* hydrate; + } + yield* update({ + sessionUpdate: "agent_message_chunk", + content: { type: "text", text: "ALL_JOBS_DONE" }, + }); + yield* Deferred.succeed(promptGate, { stopReason: "end_turn" }); + + yield* Effect.raceFirst(Deferred.await(promptHeld), Deferred.await(terminal)); + const settled = yield* Deferred.await(terminal).pipe( + Effect.timeoutOption("1 minute"), + Effect.forkScoped, + ); + yield* TestClock.adjust("1 minute"); + const terminalStatus = yield* Fiber.join(settled); + return { monitorStatuses, monitorOutputs, terminalStatus }; +}); + const makeAdvertisedCompactionHarness = Effect.fnUntraced(function* ( environment: Readonly>, ) { @@ -10006,6 +10227,39 @@ describe("AcpAdapterV2", () => { }).pipe(Effect.provide(layerTest)), ); + it.effect("finalizes the root turn when a Grok monitor completes before the prompt settles", () => + Effect.gen(function* () { + const { monitorStatuses, terminalStatus } = yield* runGrokMonitorEndingMidTurn({}); + assert.equal( + monitorStatuses.at(-1), + "completed", + "a monitor whose task_completed arrived before the prompt settled must not stay running", + ); + assert.deepEqual( + terminalStatus, + Option.some("completed"), + "the root turn must finalize after end_turn once its only monitor has completed", + ); + }).pipe(Effect.provide(layerTest), Effect.scoped), + ); + + it.effect.each([ + { hydration: "failed", order: "hydration-last", hydrations: 1 }, + { hydration: "failed", order: "hydration-first", hydrations: 1 }, + { hydration: "completed", order: "hydration-last", hydrations: 2 }, + ] as const)( + "keeps $hydrations mid-turn $hydration TaskOutput hydration ($order) over an outputless monitor end", + (input) => + Effect.gen(function* () { + const { hydration } = input; + const { monitorStatuses, monitorOutputs, terminalStatus } = + yield* runGrokMonitorEndingMidTurn(input); + assert.equal(monitorStatuses.at(-1), hydration); + assert.equal(monitorOutputs.at(-1), "HYDRATED_RESULT"); + assert.deepEqual(terminalStatus, Option.some("completed")); + }).pipe(Effect.provide(layerTest), Effect.scoped), + ); + it.live("durably poisons start and resume when required hard teardown is unavailable", () => Effect.gen(function* () { const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; diff --git a/packages/provider-acp/src/server/adapter.ts b/packages/provider-acp/src/server/adapter.ts index 24cde9fd967ba..ea129fdf3e997 100644 --- a/packages/provider-acp/src/server/adapter.ts +++ b/packages/provider-acp/src/server/adapter.ts @@ -4093,15 +4093,17 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( backgroundCompletion.appendOutput, ) : { ...target, status: nextStatus }; - // BackgroundTaskStarted normalization can reopen this row, so keep the - // completion for settlement when no task_completed recorded one. - if ( - backgroundCompletion.status !== "running" && - context.backgroundCommands.get(backgroundCompletion.taskId) === null - ) { + // BackgroundTaskStarted normalization can reopen this row, so settlement + // replays the newest completion; a hydration never undoes a known failure. + const cached = context.backgroundCommands.get(backgroundCompletion.taskId); + if (backgroundCompletion.status !== "running") { context.backgroundCommands.set(backgroundCompletion.taskId, { - status: backgroundCompletion.status, - output: toolOutputText(hydrated), + status: cached?.status === "failed" ? "failed" : backgroundCompletion.status, + output: !cached + ? toolOutputText(hydrated) + : backgroundCompletion.appendOutput.length > 0 + ? backgroundCompletion.appendOutput + : (cached.output ?? toolOutputText(hydrated)), }); } yield* emitTool(context, hydrated); @@ -7006,13 +7008,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( if ((yield* Ref.get(activeSessionId)) !== incoming.sessionId) return; const context = yield* Ref.get(activeTurn); let mutation = incoming; - if ( - incoming.status !== "running" && - context !== null && - !context.finalized && - (incoming.report?.kind === "command" || - context.backgroundCommands.has(incoming.taskId)) - ) { + if (incoming.status !== "running" && context !== null && !context.finalized) { const previous = context.backgroundCommands.get(incoming.taskId); const toolCallId = context.toolCallIdsByBackgroundTaskId.get(incoming.taskId); const tool =