From fc1e95f3119d45c40c536efc9560d78024b8ae00 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 3 Nov 2026 19:12:49 +0000 Subject: [PATCH 1/6] test(server): reproduce #30304 Grok monitor completed mid-turn wedging the root turn These tests reproduce #30304: a non-persistent Grok monitor whose _x.ai/task_completed arrives before session/prompt resolves stays running, and the root turn never reaches turn.terminal after end_turn. Co-Authored-By: Claude Opus 5.5 --- .../Adapters/AcpAdapterV2.test.ts | 195 ++++++++++++++++++ 1 file changed, 195 insertions(+) diff --git a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts index a58a83bffa349..230032b20c661 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts @@ -8569,6 +8569,201 @@ describe("AcpAdapterV2", () => { }).pipe(Effect.provide(layerTest), Effect.scoped), ); + it.effect("finalizes the root turn when a Grok monitor completes before the prompt settles", () => + 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), + ); + 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!, + 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 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.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 }, + }); + yield* ext("_x.ai/task_completed", { + sessionUpdate: "task_completed", + task_snapshot: { + task_id: taskId, + command, + display_command: "[monitor] Watch job1 done file", + output: "job1 finished\n", + }, + }); + 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); + 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.live("durably poisons start and resume when required hard teardown is unavailable", () => Effect.gen(function* () { const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; From 7bdf76cfb2d1584e34184c6cf448ca198d2f486b Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 3 Nov 2026 19:24:18 +0000 Subject: [PATCH 2/6] fix(provider-acp): Grok turns finalize when a monitor ends before the prompt settles A non-persistent Grok monitor whose _x.ai/task_completed arrived before session/prompt resolved was dropped: only command reports were held for replay at prompt settlement, so the monitor row stayed running and the root turn never finalized. Hold every background terminal that arrives during the turn, so monitors settle exactly like background commands. Fixes pingdotgg/t3code#30304 Co-Authored-By: Claude Opus 5.5 --- packages/provider-acp/src/server/adapter.ts | 8 +------- 1 file changed, 1 insertion(+), 7 deletions(-) diff --git a/packages/provider-acp/src/server/adapter.ts b/packages/provider-acp/src/server/adapter.ts index 985f6c8ee9887..c58949e96ebcd 100644 --- a/packages/provider-acp/src/server/adapter.ts +++ b/packages/provider-acp/src/server/adapter.ts @@ -6688,13 +6688,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 = From 2ded4d2a4e5939c1ee1348a458f1240edf1ab683 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 3 Nov 2026 19:40:09 +0000 Subject: [PATCH 3/6] fix(provider-acp): a mid-turn TaskOutput result survives prompt settlement Holding every early background terminal for settlement regressed one order: an outputless task_completed, then a TaskOutput result, then the prompt response. Settlement replayed the stale cached completion over the hydrated row, so a failed monitor flipped to completed and lost its output. Hydration now refreshes any cached completion, keeping a known failure and the cached output when the hydration carries none. Co-Authored-By: Claude Opus 5.5 --- .../Adapters/AcpAdapterV2.test.ts | 398 ++++++++++-------- packages/provider-acp/src/server/adapter.ts | 17 +- 2 files changed, 227 insertions(+), 188 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts index 230032b20c661..2722c5a15b8d2 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts @@ -610,6 +610,211 @@ function assertReinjectedInstructions( assert.include(text, `\n${request}\n`); } +// With `hydration`, an outputless task_completed is followed by a TaskOutput result. +const runGrokMonitorEndingMidTurn = Effect.fnUntraced(function* (input: { + readonly hydration?: "completed" | "failed"; +}) { + 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 }, + }); + yield* 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" } : {}), + }, + }); + if (input.hydration !== undefined) { + yield* update({ + sessionUpdate: "tool_call", + toolCallId: "call-fetch-job1", + title: "get_command_or_subagent_output", + status: "completed", + rawInput: { task_id: taskId }, + rawOutput: { + type: "TaskOutput", + Result: { task_id: taskId, status: input.hydration, output: "HYDRATED_RESULT\n" }, + }, + }); + } + 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 }; +}); + describe("AcpAdapterV2", () => { it.live("reports per-turn usage unavailable on a continuation turn that never prompts", () => Effect.gen(function* () { @@ -8571,186 +8776,7 @@ describe("AcpAdapterV2", () => { it.effect("finalizes the root turn when a Grok monitor completes before the prompt settles", () => 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), - ); - 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!, - 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 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.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 }, - }); - yield* ext("_x.ai/task_completed", { - sessionUpdate: "task_completed", - task_snapshot: { - task_id: taskId, - command, - display_command: "[monitor] Watch job1 done file", - output: "job1 finished\n", - }, - }); - 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); + const { monitorStatuses, terminalStatus } = yield* runGrokMonitorEndingMidTurn({}); assert.equal( monitorStatuses.at(-1), "completed", @@ -8764,6 +8790,18 @@ describe("AcpAdapterV2", () => { }).pipe(Effect.provide(layerTest), Effect.scoped), ); + it.effect.each(["completed", "failed"] as const)( + "keeps a mid-turn %s TaskOutput hydration over an earlier outputless monitor end", + (hydration) => + Effect.gen(function* () { + const { monitorStatuses, monitorOutputs, terminalStatus } = + yield* runGrokMonitorEndingMidTurn({ hydration }); + 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 c58949e96ebcd..c773cd454c456 100644 --- a/packages/provider-acp/src/server/adapter.ts +++ b/packages/provider-acp/src/server/adapter.ts @@ -3836,15 +3836,16 @@ 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" && cached !== undefined) { context.backgroundCommands.set(backgroundCompletion.taskId, { - status: backgroundCompletion.status, - output: toolOutputText(hydrated), + status: cached?.status === "failed" ? "failed" : backgroundCompletion.status, + output: + backgroundCompletion.appendOutput.length === 0 && cached?.output !== undefined + ? cached.output + : toolOutputText(hydrated), }); } yield* emitTool(context, hydrated); From af23452d0fe6749c9adb15f060e0d7b01808d44c Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 3 Nov 2026 19:51:23 +0000 Subject: [PATCH 4/6] fix(provider-acp): a TaskOutput failure before task_completed survives settlement Cache every terminal hydration of a registered background task, so an outputless task_completed that follows cannot replay `completed` over it. Co-Authored-By: Claude Opus 5.5 --- .../Adapters/AcpAdapterV2.test.ts | 50 ++++++++++++------- packages/provider-acp/src/server/adapter.ts | 2 +- 2 files changed, 33 insertions(+), 19 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts index 2722c5a15b8d2..000214804860e 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts @@ -610,9 +610,10 @@ function assertReinjectedInstructions( assert.include(text, `\n${request}\n`); } -// With `hydration`, an outputless task_completed is followed by a TaskOutput result. +// 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"; }) { const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; const path = yield* Path.Path; @@ -777,7 +778,7 @@ const runGrokMonitorEndingMidTurn = Effect.fnUntraced(function* (input: { status: "completed", rawOutput: { type: "Monitor", taskId, timeoutMs: 36000000, persistent: false }, }); - yield* ext("_x.ai/task_completed", { + const taskCompleted = ext("_x.ai/task_completed", { sessionUpdate: "task_completed", task_snapshot: { task_id: taskId, @@ -786,18 +787,26 @@ const runGrokMonitorEndingMidTurn = Effect.fnUntraced(function* (input: { ...(input.hydration === undefined ? { output: "job1 finished\n" } : {}), }, }); - if (input.hydration !== undefined) { - yield* update({ - sessionUpdate: "tool_call", - toolCallId: "call-fetch-job1", - title: "get_command_or_subagent_output", - status: "completed", - rawInput: { task_id: taskId }, - rawOutput: { - type: "TaskOutput", - Result: { task_id: taskId, status: input.hydration, output: "HYDRATED_RESULT\n" }, - }, - }); + const hydrate = + input.hydration === undefined + ? Effect.void + : update({ + sessionUpdate: "tool_call", + toolCallId: "call-fetch-job1", + title: "get_command_or_subagent_output", + status: "completed", + rawInput: { task_id: taskId }, + rawOutput: { + type: "TaskOutput", + Result: { task_id: taskId, status: input.hydration, output: "HYDRATED_RESULT\n" }, + }, + }); + if (input.order === "hydration-first") { + yield* hydrate; + yield* taskCompleted; + } else { + yield* taskCompleted; + yield* hydrate; } yield* update({ sessionUpdate: "agent_message_chunk", @@ -8790,12 +8799,17 @@ describe("AcpAdapterV2", () => { }).pipe(Effect.provide(layerTest), Effect.scoped), ); - it.effect.each(["completed", "failed"] as const)( - "keeps a mid-turn %s TaskOutput hydration over an earlier outputless monitor end", - (hydration) => + it.effect.each([ + { hydration: "completed", order: "hydration-last" }, + { hydration: "failed", order: "hydration-last" }, + { hydration: "completed", order: "hydration-first" }, + { hydration: "failed", order: "hydration-first" }, + ] as const)( + "keeps a mid-turn $hydration TaskOutput hydration ($order) over an outputless monitor end", + ({ hydration, order }) => Effect.gen(function* () { const { monitorStatuses, monitorOutputs, terminalStatus } = - yield* runGrokMonitorEndingMidTurn({ hydration }); + yield* runGrokMonitorEndingMidTurn({ hydration, order }); assert.equal(monitorStatuses.at(-1), hydration); assert.equal(monitorOutputs.at(-1), "HYDRATED_RESULT"); assert.deepEqual(terminalStatus, Option.some("completed")); diff --git a/packages/provider-acp/src/server/adapter.ts b/packages/provider-acp/src/server/adapter.ts index c773cd454c456..dbeeba6efcb61 100644 --- a/packages/provider-acp/src/server/adapter.ts +++ b/packages/provider-acp/src/server/adapter.ts @@ -3839,7 +3839,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( // 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" && cached !== undefined) { + if (backgroundCompletion.status !== "running") { context.backgroundCommands.set(backgroundCompletion.taskId, { status: cached?.status === "failed" ? "failed" : backgroundCompletion.status, output: From 173566087384fb65abb758f19488ce7e119184f1 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 3 Nov 2026 20:08:32 +0000 Subject: [PATCH 5/6] fix(provider-acp): settlement replays the TaskOutput snapshot, not the accumulated row text A repeated TaskOutput or one after streamed progress no longer survives settlement as duplicated output. Co-Authored-By: Claude Opus 5.5 --- .../Adapters/AcpAdapterV2.test.ts | 47 +++++++++++-------- packages/provider-acp/src/server/adapter.ts | 9 ++-- 2 files changed, 33 insertions(+), 23 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts index 000214804860e..cf3ce26014091 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts @@ -614,6 +614,7 @@ function assertReinjectedInstructions( 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; @@ -787,20 +788,26 @@ const runGrokMonitorEndingMidTurn = Effect.fnUntraced(function* (input: { ...(input.hydration === undefined ? { output: "job1 finished\n" } : {}), }, }); + const hydration = input.hydration; const hydrate = - input.hydration === undefined + hydration === undefined ? Effect.void - : update({ - sessionUpdate: "tool_call", - toolCallId: "call-fetch-job1", - title: "get_command_or_subagent_output", - status: "completed", - rawInput: { task_id: taskId }, - rawOutput: { - type: "TaskOutput", - Result: { task_id: taskId, status: input.hydration, output: "HYDRATED_RESULT\n" }, - }, - }); + : 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; @@ -8800,16 +8807,18 @@ describe("AcpAdapterV2", () => { ); it.effect.each([ - { hydration: "completed", order: "hydration-last" }, - { hydration: "failed", order: "hydration-last" }, - { hydration: "completed", order: "hydration-first" }, - { hydration: "failed", order: "hydration-first" }, + { hydration: "completed", order: "hydration-last", hydrations: 1 }, + { hydration: "failed", order: "hydration-last", hydrations: 1 }, + { hydration: "completed", order: "hydration-first", hydrations: 1 }, + { hydration: "failed", order: "hydration-first", hydrations: 1 }, + { hydration: "completed", order: "hydration-last", hydrations: 2 }, ] as const)( - "keeps a mid-turn $hydration TaskOutput hydration ($order) over an outputless monitor end", - ({ hydration, order }) => + "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({ hydration, order }); + yield* runGrokMonitorEndingMidTurn(input); assert.equal(monitorStatuses.at(-1), hydration); assert.equal(monitorOutputs.at(-1), "HYDRATED_RESULT"); assert.deepEqual(terminalStatus, Option.some("completed")); diff --git a/packages/provider-acp/src/server/adapter.ts b/packages/provider-acp/src/server/adapter.ts index dbeeba6efcb61..069d23b189a40 100644 --- a/packages/provider-acp/src/server/adapter.ts +++ b/packages/provider-acp/src/server/adapter.ts @@ -3842,10 +3842,11 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( if (backgroundCompletion.status !== "running") { context.backgroundCommands.set(backgroundCompletion.taskId, { status: cached?.status === "failed" ? "failed" : backgroundCompletion.status, - output: - backgroundCompletion.appendOutput.length === 0 && cached?.output !== undefined - ? cached.output - : toolOutputText(hydrated), + output: !cached + ? toolOutputText(hydrated) + : backgroundCompletion.appendOutput.length > 0 + ? backgroundCompletion.appendOutput + : (cached.output ?? toolOutputText(hydrated)), }); } yield* emitTool(context, hydrated); From e066f47ee24f1d1aa88df47810e4a17fa318f0ec Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 10 Nov 2026 23:33:13 +0000 Subject: [PATCH 6/6] test(server): drop Grok monitor hydration rows other cases already cover AI-assisted; verified with the tests and tools listed above. --- apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts | 2 -- 1 file changed, 2 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts index a890544775d00..488dc4189000b 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts @@ -9615,9 +9615,7 @@ describe("AcpAdapterV2", () => { ); it.effect.each([ - { hydration: "completed", order: "hydration-last", hydrations: 1 }, { hydration: "failed", order: "hydration-last", hydrations: 1 }, - { hydration: "completed", order: "hydration-first", hydrations: 1 }, { hydration: "failed", order: "hydration-first", hydrations: 1 }, { hydration: "completed", order: "hydration-last", hydrations: 2 }, ] as const)(