From c52a2983d0991e9254f4af9db8388a7a18be43ab Mon Sep 17 00:00:00 2001 From: Yaacov Date: Tue, 6 Oct 2026 13:13:05 +0300 Subject: [PATCH 01/13] fix(antigravity): keep turns open for running commands so Stop can end them Antigravity can answer end_turn while a command it started is still running. T3 Code then marked the turn and the command completed, and a later Stop had no turn to act on, so the command kept running. The Antigravity flavor now treats running execute tools as background work, so the turn stays open until the command's terminal update or an explicit Stop. Tools still awaiting approval do not hold it. The agent runs in its own process group, and Stop tears that group down, ending the commands in it; the next turn respawns the agent. Subagent batches still go idle when the root turn ends instead of holding it open. Fixes #16442 Co-Authored-By: Codex (gpt-6.1-sol) Co-Authored-By: Claude Opus 5.5 (1M context) --- apps/server/scripts/acp-mock-agent.ts | 92 +++++++ .../orchestration-v2/Adapters/AcpAdapterV2.ts | 10 +- .../Adapters/AntigravityAdapterV2.test.ts | 251 +++++++++++++++++- .../Adapters/AntigravityAdapterV2.ts | 17 ++ 4 files changed, 366 insertions(+), 4 deletions(-) diff --git a/apps/server/scripts/acp-mock-agent.ts b/apps/server/scripts/acp-mock-agent.ts index 5548bebac84c0..2f136f245901a 100644 --- a/apps/server/scripts/acp-mock-agent.ts +++ b/apps/server/scripts/acp-mock-agent.ts @@ -71,6 +71,9 @@ const exitAfterResidualCallbacks = process.env.T3_ACP_EXIT_AFTER_RESIDUAL_CALLBA const emitRunningCommandThenHang = process.env.T3_ACP_EMIT_RUNNING_COMMAND_THEN_HANG === "1"; const emitRunningCommandThenHangOnFirstPrompt = process.env.T3_ACP_EMIT_RUNNING_COMMAND_THEN_HANG_FIRST_PROMPT === "1"; +const emitCommandOutlivingPrompt = process.env.T3_ACP_EMIT_COMMAND_OUTLIVING_PROMPT === "1"; +const reportOutlivingCommandExit = process.env.T3_ACP_REPORT_OUTLIVING_COMMAND_EXIT === "1"; +const emitSubagentLaunchThenEndTurn = process.env.T3_ACP_EMIT_SUBAGENT_LAUNCH_THEN_END_TURN === "1"; const emitEmptySuccessfulBash = process.env.T3_ACP_EMIT_EMPTY_SUCCESSFUL_BASH === "1"; const emitEmptySuccessfulBashThenHang = process.env.T3_ACP_EMIT_EMPTY_SUCCESSFUL_BASH_THEN_HANG === "1"; @@ -1344,6 +1347,95 @@ const program = Effect.gen(function* () { return yield* Effect.never; } + if (emitCommandOutlivingPrompt) { + // Antigravity can answer end_turn while a command it started keeps running. + // A long-lived Node child stands in for the command on every platform. + const command = NodeChildProcess.spawn( + process.execPath, + ["-e", "setInterval(() => {}, 1 << 30)"], + { stdio: "ignore" }, + ); + yield* Effect.callback((resume) => { + command.once("spawn", () => resume(Effect.void)); + command.once("error", (error) => resume(Effect.die(error))); + }); + if (runningCommandPidPath !== undefined) { + NodeFS.writeFileSync(runningCommandPidPath, String(command.pid)); + } + if (reportOutlivingCommandExit) { + yield* Effect.callback((resume) => { + command.once("exit", () => resume(Effect.void)); + }).pipe( + Effect.andThen( + agent.client.sessionUpdate({ + sessionId: requestedSessionId, + update: { + sessionUpdate: "tool_call_update", + toolCallId: "command-outliving-prompt", + status: "completed", + }, + }), + ), + Effect.ignore, + Effect.forkDetach, + ); + } + // Awaiting approval: not started, so it must not hold the turn open. + yield* agent.client.sessionUpdate({ + sessionId: requestedSessionId, + update: { + sessionUpdate: "tool_call_update", + toolCallId: "command-awaiting-approval", + title: "run_command", + kind: "execute", + status: "pending", + rawInput: { command: "npm test" }, + }, + }); + yield* agent.client.sessionUpdate({ + sessionId: requestedSessionId, + update: { + sessionUpdate: "tool_call_update", + toolCallId: "command-outliving-prompt", + title: "run_command", + kind: "execute", + status: "in_progress", + rawInput: { command: "npm run dev" }, + }, + }); + yield* agent.client.sessionUpdate({ + sessionId: requestedSessionId, + update: { + sessionUpdate: "agent_message_chunk", + content: { type: "text", text: "Started the command." }, + }, + }); + return yield* finishPrompt(requestedSessionId, "end_turn"); + } + + if (emitSubagentLaunchThenEndTurn) { + // ACP 1.1.1 batch launch: completes at once and never reports the children. + yield* agent.client.sessionUpdate({ + sessionId: requestedSessionId, + update: { + sessionUpdate: "tool_call_update", + toolCallId: "subagent-launch", + title: "Running start_subagent", + kind: "other", + status: "completed", + rawOutput: "Started two agents.", + }, + }); + yield* agent.client.sessionUpdate({ + sessionId: requestedSessionId, + update: { + sessionUpdate: "agent_message_chunk", + content: { type: "text", text: "Started the agents." }, + }, + }); + return yield* finishPrompt(requestedSessionId, "end_turn"); + } + if ( emitRunningCommandThenHang || (emitRunningCommandThenHangOnFirstPrompt && promptCount === 1) diff --git a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.ts index 6b5b1b91fb012..fd31663786d6e 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.ts @@ -2982,9 +2982,13 @@ export function makeAcpAdapterV2( const status = toolStatus(tool.status); if (status === "pending" || status === "running") return true; } - for (const subagent of context.subagents.values()) { - if (acpSubagentStatusBlocksTurnSettlement(subagent.task.status)) { - return true; + // Batch launches without child completion signals go idle when the + // root turn ends, so they must not hold it open. + if (flavor.subagentsIdleOnTurnCompletion !== true) { + for (const subagent of context.subagents.values()) { + if (acpSubagentStatusBlocksTurnSettlement(subagent.task.status)) { + return true; + } } } return false; diff --git a/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.test.ts index 6905e72fafb56..b541e06138bee 100644 --- a/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.test.ts @@ -13,12 +13,16 @@ import { import { resolveSelfInvocation } from "@t3tools/shared/nodeRuntime"; import * as Crypto from "effect/Crypto"; import * as DateTime from "effect/DateTime"; +import * as Deferred from "effect/Deferred"; +import type * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Path from "effect/Path"; +import * as Queue from "effect/Queue"; import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; import { ChildProcessSpawner } from "effect/process"; import type * as EffectAcpSchema from "effect-acp/compat"; @@ -26,7 +30,7 @@ import * as ServerConfig from "../../config.ts"; import type * as AcpSessionRuntime from "../../provider/acp/AcpSessionRuntime.ts"; import { makeAntigravityAcpRuntime } from "../../provider/acp/AntigravityAcpSupport.ts"; import * as IdAllocator from "../IdAllocator.ts"; -import { ProviderAdapterV2RuntimePolicy } from "../ProviderAdapter.ts"; +import { ProviderAdapterV2RuntimePolicy, type ProviderAdapterV2Event } from "../ProviderAdapter.ts"; import { makeAntigravityAcpAdapterFlavor, makeAntigravityAdapterV2, @@ -535,3 +539,248 @@ describe("AntigravityAdapterV2 client file system under restrictive policies", ( }).pipe(Effect.provide(layerSession), Effect.scoped), ); }); + +describe("AntigravityAdapterV2 commands that outlive the prompt", () => { + const pidIsRunning = (pid: number) => { + try { + process.kill(pid, 0); + return true; + } catch { + return false; + } + }; + + const openMockSession = (label: string, env: Record) => + Effect.gen(function* () { + const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const crypto = yield* Crypto.Crypto; + const mockAgentPath = yield* path.fromFileUrl( + new URL("../../../scripts/acp-mock-agent.ts", import.meta.url), + ); + const workspace = yield* fileSystem.makeTempDirectoryScoped({ prefix: `t3-ag-${label}-` }); + const commandPidPath = path.join(workspace, "command.pid"); + const commandPids: Array = []; + yield* Effect.addFinalizer(() => + Effect.sync(() => { + for (const pid of commandPids) if (pidIsRunning(pid)) process.kill(pid, "SIGKILL"); + }), + ); + // The mock publishes the PID once the command has spawned, before it answers. + const readCommandPid = Effect.gen(function* () { + const pid = Number(yield* fileSystem.readFileString(commandPidPath)); + commandPids.push(pid); + return pid; + }); + // The turn finishes after a debounce once its background work ends; + // the hook tells the test when to advance the clock past it. + const finishArmed = yield* Deferred.make(); + const instanceId = ProviderInstanceId.make(`antigravity-${label}`); + const adapter = makeAntigravityAdapterV2({ + instanceId, + testHooks: { + onDeferredFinalizeScheduled: (debounce) => + Deferred.succeed(finishArmed, debounce).pipe(Effect.asVoid), + }, + crypto, + selfInvocation: yield* resolveSelfInvocation(), + fileSystem, + path, + idAllocator: yield* IdAllocator.IdAllocatorV2, + serverConfig: yield* ServerConfig.ServerConfig, + makeRuntime: (input) => + makeAntigravityAcpRuntime({ + ...input, + childProcessSpawner, + spawn: { + command: process.execPath, + args: [mockAgentPath], + cwd: input.cwd, + env: { + T3_ACP_ANTIGRAVITY: "1", + T3_ACP_RUNNING_COMMAND_PID_PATH: commandPidPath, + ...env, + }, + }, + }).pipe(Effect.provideService(Crypto.Crypto, crypto)), + withProcess: (_stop, task) => task, + defaultModel: Effect.succeed(undefined), + }); + const threadId = ThreadId.make(`thread-antigravity-${label}`); + const runtimePolicy = ProviderAdapterV2RuntimePolicy.make({ + runtimeMode: "full-access", + interactionMode: "default", + cwd: workspace, + }); + const modelSelection = { instanceId, model: "gemini-test-low" } as const; + const session = yield* adapter.openSession({ + threadId, + providerSessionId: ProviderSessionId.make(`provider-session-antigravity-${label}`), + modelSelection, + runtimePolicy, + }); + const providerThread = yield* session.ensureThread({ + threadId, + modelSelection, + runtimePolicy, + }); + const events = yield* Queue.unbounded(); + yield* session.events.pipe( + Stream.runForEach((event) => Queue.offer(events, event)), + Effect.forkScoped({ startImmediately: true }), + ); + const seen: Array = []; + const waitFor = (predicate: (event: ProviderAdapterV2Event) => boolean) => + Queue.take(events).pipe( + Effect.tap((event) => Effect.sync(() => seen.push(event))), + Effect.repeat({ until: predicate }), + ); + const waitForReply = waitFor( + (event) => + event.type === "message.updated" && + event.message.role === "assistant" && + !event.message.streaming, + ); + const waitForTerminal = waitFor((event) => event.type === "turn.terminal").pipe( + Effect.map((event) => (event.type === "turn.terminal" ? event.status : undefined)), + ); + const commandItem = (toolCallId: string) => { + const command = seen.findLast( + (event) => + event.type === "turn_item.updated" && + event.turnItem.type === "command_execution" && + event.turnItem.nativeItemRef?.nativeId?.includes(toolCallId) === true, + ); + return command?.type === "turn_item.updated" ? command.turnItem : undefined; + }; + + const startTurn = (ordinal: number) => + Effect.gen(function* () { + const now = yield* DateTime.now; + yield* session.startTurn({ + appThread: { + createdBy: "user", + creationSource: "web", + id: threadId, + projectId: ProjectId.make(`project-antigravity-${label}`), + title: `Antigravity ${label}`, + providerInstanceId: instanceId, + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: workspace, + activeProviderThreadId: providerThread.id, + lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId }, + forkedFrom: null, + createdAt: now, + updatedAt: now, + archivedAt: null, + settledOverride: null, + settledAt: null, + lastVisitedAt: null, + deletedAt: null, + }, + threadId, + runId: RunId.make(`run-antigravity-${label}-${ordinal}`), + runOrdinal: ordinal, + providerTurnOrdinal: ordinal, + attemptId: RunAttemptId.make(`attempt-antigravity-${label}-${ordinal}`), + rootNodeId: NodeId.make(`node-antigravity-${label}-${ordinal}`), + providerThread, + message: { + createdBy: "user", + creationSource: "web", + messageId: MessageId.make(`message-antigravity-${label}-${ordinal}`), + text: "start a long command", + attachments: [], + }, + modelSelection, + runtimePolicy, + }); + }); + + return { + session, + providerThread, + startTurn, + waitForReply, + waitForTerminal, + readCommandPid, + commandItem, + finishArmed, + }; + }); + + it.effect("keeps the turn open until the command's terminal update, then completes it", () => + Effect.gen(function* () { + const mock = yield* openMockSession("command-finishes", { + T3_ACP_EMIT_COMMAND_OUTLIVING_PROMPT: "1", + T3_ACP_REPORT_OUTLIVING_COMMAND_EXIT: "1", + }); + yield* mock.startTurn(1); + + // The agent has answered end_turn; the command it started is still running. + yield* mock.waitForReply; + const commandPid = yield* mock.readCommandPid; + assert.isTrue(pidIsRunning(commandPid)); + + // The command finishes; the mock reports its terminal update on exit. + process.kill(commandPid); + yield* TestClock.adjust(yield* Deferred.await(mock.finishArmed)); + const status = yield* mock.waitForTerminal; + // The turn can only have settled after the command's own update: the + // command was gone by then, and a tool awaiting approval did not hold it. + assert.isFalse(pidIsRunning(commandPid), "the turn waits for the command"); + assert.equal(status, "completed"); + assert.equal(mock.commandItem("command-outliving-prompt")?.status, "completed"); + }).pipe(Effect.provide(layerSession), Effect.scoped), + ); + + it.effect("lets Stop end the command and respawns the agent for the next turn", () => + Effect.gen(function* () { + const mock = yield* openMockSession("command-stopped", { + T3_ACP_EMIT_COMMAND_OUTLIVING_PROMPT: "1", + }); + yield* mock.startTurn(1); + + // The agent has answered end_turn; the command it started is still running. + yield* mock.waitForReply; + const commandPid = yield* mock.readCommandPid; + assert.isTrue(pidIsRunning(commandPid)); + const providerTurnId = mock.commandItem("command-outliving-prompt")?.providerTurnId; + if (providerTurnId == null) return yield* Effect.die("missing command item"); + + yield* mock.session.interruptTurn({ + providerThread: mock.providerThread, + providerTurnId, + requestRuntimeRestart: true, + }); + assert.equal( + yield* mock.waitForTerminal, + "interrupted", + "the turn stays open for the command until Stop", + ); + assert.equal(mock.commandItem("command-outliving-prompt")?.status, "interrupted"); + assert.isFalse(pidIsRunning(commandPid), "Stop ends the command"); + + // The next turn runs on a respawned agent, which starts a new command. + yield* mock.startTurn(2); + yield* mock.waitForReply; + const nextCommandPid = yield* mock.readCommandPid; + assert.notEqual(nextCommandPid, commandPid); + assert.isTrue(pidIsRunning(nextCommandPid)); + }).pipe(Effect.provide(layerSession), Effect.scoped), + ); + + it.effect("completes a turn whose only open work is a subagent batch", () => + Effect.gen(function* () { + const mock = yield* openMockSession("subagent-batch", { + T3_ACP_EMIT_SUBAGENT_LAUNCH_THEN_END_TURN: "1", + }); + yield* mock.startTurn(1); + assert.equal(yield* mock.waitForTerminal, "completed"); + }).pipe(Effect.provide(layerSession), Effect.scoped), + ); +}); diff --git a/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.ts index 1ee304556cc16..750e648911aa8 100644 --- a/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.ts @@ -5,6 +5,7 @@ import { type OrchestrationV2ProviderCapabilities, type ProviderSetupError, } from "@t3tools/contracts"; +import { HostProcessPlatform } from "@t3tools/shared/hostProcess"; import type { SelfInvocation } from "@t3tools/shared/nodeRuntime"; import * as Crypto from "effect/Crypto"; import * as Effect from "effect/Effect"; @@ -88,6 +89,7 @@ export interface AntigravityAdapterV2Options { readonly onSessionEvent?: AcpAdapterV2Flavor["onSessionEvent"]; readonly nativeLogging?: Parameters[0]["nativeLogging"]; readonly continuationRequests?: Parameters[0]["continuationRequests"]; + readonly testHooks?: Parameters[0]["testHooks"]; } /** @@ -133,12 +135,18 @@ export function makeAntigravityAcpAdapterFlavor( // AcpAdapterV2 owns the runtime scope; sign-in and sign-out stop the // process by closing a child of it, and the adapter respawns on the next turn. const scope = yield* Scope.fork(yield* Effect.scope); + const platform = yield* HostProcessPlatform; const runtime = yield* options .withProcess( Scope.close(scope, Exit.void), options.makeRuntime({ ...input, clientFileSystem: true, + // Own the agent's process group so Stop also ends the commands it + // started, including ones still running after the prompt returned. + ownDetachedProcessGroup: true, + ownDescendantProcessGroups: platform === "linux", + processGroupPlatform: platform, additionalDirectories: [options.serverConfig.attachmentsDir], }), ) @@ -172,6 +180,14 @@ export function makeAntigravityAcpAdapterFlavor( // without the replay and is what the official client does. preferResumeSession: true, subagentsIdleOnTurnCompletion: true, + terminateRuntimeProcessGroupOnInterrupt: true, + // A command can keep running after Antigravity returns end_turn. Keep the + // turn open until the command's terminal update or an explicit Stop. + deferFinalizeForBackgroundWork: true, + extractBackgroundTaskId: (toolCall) => + toolCall.kind === "execute" && toolCall.status !== undefined && toolCall.status !== "pending" + ? toolCall.toolCallId + : undefined, ...(options.onSessionEvent === undefined ? {} : { onSessionEvent: options.onSessionEvent }), applyModelSelection: ({ runtime, modelSelection }) => Effect.gen(function* () { @@ -239,5 +255,6 @@ export function makeAntigravityAdapterV2(options: AntigravityAdapterV2Options) { ...(options.continuationRequests === undefined ? {} : { continuationRequests: options.continuationRequests }), + ...(options.testHooks === undefined ? {} : { testHooks: options.testHooks }), }); } From ad45816fb4e9f9d7302c0e0d2ed6569e9de095dc Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 01:40:42 +0000 Subject: [PATCH 02/13] test(antigravity): cover commands that outlive the prompt Co-Authored-By: Claude Fable 5.1 --- .../Adapters/AntigravityAdapterV2.test.ts | 241 ++++++++++++++++++ .../Adapters/AntigravityAdapterV2.ts | 2 + 2 files changed, 243 insertions(+) diff --git a/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.test.ts index 2ded59be7fc38..aa44c9fefdedc 100644 --- a/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.test.ts @@ -13,12 +13,16 @@ import { import { resolveSelfInvocation } from "@t3tools/shared/nodeRuntime"; import * as Crypto from "effect/Crypto"; import * as DateTime from "effect/DateTime"; +import * as Deferred from "effect/Deferred"; +import type * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Path from "effect/Path"; +import * as Queue from "effect/Queue"; import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; import { ChildProcessSpawner } from "effect/process"; import type * as EffectAcpSchema from "effect-acp/compat"; @@ -510,3 +514,240 @@ describe("AntigravityAdapterV2 client file system under restrictive policies", ( }).pipe(Effect.provide(layerSession), Effect.scoped), ); }); + +describe("AntigravityAdapterV2 commands that outlive the prompt", () => { + const pidIsRunning = (pid: number) => { + try { + process.kill(pid, 0); + return true; + } catch { + return false; + } + }; + + const openMockSession = (label: string, env: Record) => + Effect.gen(function* () { + const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const crypto = yield* Crypto.Crypto; + const mockAgentPath = yield* path.fromFileUrl( + new URL("../../../scripts/acp-mock-agent.ts", import.meta.url), + ); + const workspace = yield* fileSystem.makeTempDirectoryScoped({ prefix: `t3-ag-${label}-` }); + const commandPidPath = path.join(workspace, "command.pid"); + const commandPids: Array = []; + yield* Effect.addFinalizer(() => + Effect.sync(() => { + for (const pid of commandPids) if (pidIsRunning(pid)) process.kill(pid, "SIGKILL"); + }), + ); + // The mock publishes the PID once the command has spawned, before it answers. + const readCommandPid = Effect.gen(function* () { + const pid = Number(yield* fileSystem.readFileString(commandPidPath)); + commandPids.push(pid); + return pid; + }); + // A deferred finish waits out a debounce; the hook says how far to move the clock. + const finishArmed = yield* Deferred.make(); + const instanceId = ProviderInstanceId.make(`antigravity-${label}`); + const adapter = yield* makeAntigravityAdapterV2({ + instanceId, + selfInvocation: yield* resolveSelfInvocation(), + testHooks: { + onDeferredFinalizeScheduled: (debounce) => + Deferred.succeed(finishArmed, debounce).pipe(Effect.asVoid), + }, + makeRuntime: (input) => + makeAntigravityAcpRuntime({ + ...input, + childProcessSpawner, + spawn: { + command: process.execPath, + args: [mockAgentPath], + cwd: input.cwd, + env: { + T3_ACP_ANTIGRAVITY: "1", + T3_ACP_RUNNING_COMMAND_PID_PATH: commandPidPath, + ...env, + }, + }, + }).pipe(Effect.provideService(Crypto.Crypto, crypto)), + withProcess: (_stop, task) => task, + defaultModel: Effect.succeed(undefined), + }); + const threadId = ThreadId.make(`thread-antigravity-${label}`); + const runtimePolicy = ProviderAdapter.ProviderAdapterV2RuntimePolicy.make({ + runtimeMode: "full-access", + interactionMode: "default", + cwd: workspace, + }); + const modelSelection = { instanceId, model: "gemini-test-low" } as const; + const session = yield* adapter.openSession({ + threadId, + providerSessionId: ProviderSessionId.make(`provider-session-antigravity-${label}`), + modelSelection, + runtimePolicy, + }); + const providerThread = yield* session.ensureThread({ + threadId, + modelSelection, + runtimePolicy, + }); + const events = yield* Queue.unbounded(); + yield* session.events.pipe( + Stream.runForEach((event) => Queue.offer(events, event)), + Effect.forkScoped({ startImmediately: true }), + ); + const seen: Array = []; + const waitFor = (predicate: (event: ProviderAdapter.ProviderAdapterV2Event) => boolean) => + Queue.take(events).pipe( + Effect.tap((event) => Effect.sync(() => seen.push(event))), + Effect.repeat({ until: predicate }), + ); + const waitForReply = waitFor( + (event) => + event.type === "message.updated" && + event.message.role === "assistant" && + !event.message.streaming, + ); + const waitForTerminal = waitFor((event) => event.type === "turn.terminal").pipe( + Effect.map((event) => (event.type === "turn.terminal" ? event.status : undefined)), + ); + const commandItem = (toolCallId: string) => { + const command = seen.findLast( + (event) => + event.type === "turn_item.updated" && + event.turnItem.type === "command_execution" && + event.turnItem.nativeItemRef?.nativeId?.includes(toolCallId) === true, + ); + return command?.type === "turn_item.updated" ? command.turnItem : undefined; + }; + + const startTurn = (ordinal: number) => + Effect.gen(function* () { + const now = yield* DateTime.now; + yield* session.startTurn({ + appThread: { + createdBy: "user", + creationSource: "web", + id: threadId, + projectId: ProjectId.make(`project-antigravity-${label}`), + title: `Antigravity ${label}`, + providerInstanceId: instanceId, + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: workspace, + activeProviderThreadId: providerThread.id, + lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId }, + forkedFrom: null, + createdAt: now, + updatedAt: now, + archivedAt: null, + settledOverride: null, + settledAt: null, + lastVisitedAt: null, + deletedAt: null, + }, + threadId, + runId: RunId.make(`run-antigravity-${label}-${ordinal}`), + runOrdinal: ordinal, + providerTurnOrdinal: ordinal, + attemptId: RunAttemptId.make(`attempt-antigravity-${label}-${ordinal}`), + rootNodeId: NodeId.make(`node-antigravity-${label}-${ordinal}`), + providerThread, + message: { + createdBy: "user", + creationSource: "web", + messageId: MessageId.make(`message-antigravity-${label}-${ordinal}`), + text: "start a long command", + attachments: [], + }, + modelSelection, + runtimePolicy, + }); + }); + + return { + session, + providerThread, + startTurn, + waitForReply, + waitForTerminal, + readCommandPid, + commandItem, + finishArmed, + }; + }); + + it.effect("keeps the turn open until the running command reports its end", () => + Effect.gen(function* () { + const mock = yield* openMockSession("command-finishes", { + T3_ACP_EMIT_COMMAND_OUTLIVING_PROMPT: "1", + T3_ACP_REPORT_OUTLIVING_COMMAND_EXIT: "1", + }); + yield* mock.startTurn(1); + + // The agent has answered end_turn; the command it started is still running, + // next to a second command still awaiting approval. + yield* mock.waitForReply; + const commandPid = yield* mock.readCommandPid; + assert.isTrue(pidIsRunning(commandPid)); + + // The mock reports the command's terminal update when it exits. Only then + // may the turn arm its finish; ending before it means it never waited. + process.kill(commandPid); + const turnEndedEarly = mock.waitForTerminal.pipe( + Effect.flatMap((status) => + Effect.sync(() => + assert.fail(`the turn ended (${status}) before its command reported its end`), + ), + ), + ); + const debounce = yield* Effect.raceFirst(Deferred.await(mock.finishArmed), turnEndedEarly); + yield* TestClock.adjust(debounce); + assert.equal(yield* mock.waitForTerminal, "completed"); + assert.equal(mock.commandItem("command-outliving-prompt")?.status, "completed"); + }).pipe(Effect.provide(layerSession), Effect.scoped), + ); + + it.effect("lets Stop end the running command and respawns the agent for the next turn", () => + Effect.gen(function* () { + const mock = yield* openMockSession("command-stopped", { + T3_ACP_EMIT_COMMAND_OUTLIVING_PROMPT: "1", + }); + yield* mock.startTurn(1); + + yield* mock.waitForReply; + const commandPid = yield* mock.readCommandPid; + assert.isTrue(pidIsRunning(commandPid)); + const providerTurnId = mock.commandItem("command-outliving-prompt")?.providerTurnId; + if (providerTurnId == null) return yield* Effect.die("missing command item"); + + yield* mock.session.interruptTurn({ + providerThread: mock.providerThread, + providerTurnId, + requestRuntimeRestart: true, + }); + assert.equal(yield* mock.waitForTerminal, "interrupted"); + assert.equal(mock.commandItem("command-outliving-prompt")?.status, "interrupted"); + assert.isFalse(pidIsRunning(commandPid), "Stop ends the command"); + + yield* mock.startTurn(2); + yield* mock.waitForReply; + assert.isTrue(pidIsRunning(yield* mock.readCommandPid), "a respawned agent runs turn 2"); + }).pipe(Effect.provide(layerSession), Effect.scoped), + ); + + it.effect("completes a turn whose only open work is a subagent batch", () => + Effect.gen(function* () { + const mock = yield* openMockSession("subagent-batch", { + T3_ACP_EMIT_SUBAGENT_LAUNCH_THEN_END_TURN: "1", + }); + yield* mock.startTurn(1); + assert.equal(yield* mock.waitForTerminal, "completed"); + }).pipe(Effect.provide(layerSession), Effect.scoped), + ); +}); diff --git a/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.ts index 62b2ec6eac11f..ae03fb4ad0c0e 100644 --- a/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.ts @@ -79,6 +79,7 @@ export interface AntigravityAdapterV2Options { readonly onSessionEvent?: AcpAdapterV2Flavor["onSessionEvent"]; readonly nativeLogging?: Parameters[0]["nativeLogging"]; readonly continuationRequests?: Parameters[0]["continuationRequests"]; + readonly testHooks?: Parameters[0]["testHooks"]; } /** @@ -233,5 +234,6 @@ export const makeAntigravityAdapterV2 = Effect.fn("makeAntigravityAdapterV2")(fu ...(options.continuationRequests === undefined ? {} : { continuationRequests: options.continuationRequests }), + ...(options.testHooks === undefined ? {} : { testHooks: options.testHooks }), }); }); From d5f72a80f91f51b69b36a8141852a7dddcd2ff0a Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 02:12:52 +0000 Subject: [PATCH 03/13] fix(antigravity): keep Stop available for running commands --- .../Adapters/AntigravityAdapterV2.ts | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.ts index ae03fb4ad0c0e..eebbcdc8c8d47 100644 --- a/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.ts @@ -6,6 +6,7 @@ import { type ProviderSetupError, } from "@t3tools/contracts"; import type { SelfInvocation } from "@t3tools/shared/nodeRuntime"; +import * as HostProcess from "@t3tools/shared/HostProcess"; import * as Crypto from "effect/Crypto"; import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; @@ -124,12 +125,16 @@ export function makeAntigravityAcpAdapterFlavor( const makeRuntime = (input: AcpAdapterV2RuntimeInput) => Effect.gen(function* () { const scope = yield* Scope.fork(yield* Effect.scope); + const platform = yield* HostProcess.Platform; const runtime = yield* options .withProcess( Scope.close(scope, Exit.void), options.makeRuntime({ ...input, clientFileSystem: true, + ownDetachedProcessGroup: true, + ownDescendantProcessGroups: platform === "linux", + processGroupPlatform: platform, additionalDirectories: [options.host.paths.attachmentsDir], }), ) @@ -163,6 +168,13 @@ export function makeAntigravityAcpAdapterFlavor( // without the replay and is what the official client does. preferResumeSession: true, subagentsIdleOnTurnCompletion: true, + terminateRuntimeProcessGroupOnInterrupt: true, + // Commands can outlive end_turn; retain Stop until their terminal updates arrive. + deferFinalizeForBackgroundWork: true, + extractBackgroundTaskId: (toolCall) => + toolCall.kind === "execute" && toolCall.status !== undefined && toolCall.status !== "pending" + ? toolCall.toolCallId + : undefined, ...(options.onSessionEvent === undefined ? {} : { onSessionEvent: options.onSessionEvent }), applyModelSelection: ({ runtime, modelSelection }) => Effect.gen(function* () { From 65bb7ab6f05cfdf0321125de0998532154156823 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 02:13:32 +0000 Subject: [PATCH 04/13] test(antigravity): verify Stop replaces the ACP process --- apps/server/scripts/acp-mock-agent.ts | 4 ++++ .../Adapters/AntigravityAdapterV2.test.ts | 11 ++++++++++- 2 files changed, 14 insertions(+), 1 deletion(-) diff --git a/apps/server/scripts/acp-mock-agent.ts b/apps/server/scripts/acp-mock-agent.ts index 4f33eccc73218..05966f28e44d8 100644 --- a/apps/server/scripts/acp-mock-agent.ts +++ b/apps/server/scripts/acp-mock-agent.ts @@ -77,6 +77,7 @@ const emitEmptySuccessfulBashThenHang = const exitOnCancel = process.env.T3_ACP_EXIT_ON_CANCEL === "1"; const runningCommandIgnoresTerm = process.env.T3_ACP_RUNNING_COMMAND_IGNORE_TERM === "1"; const runningCommandPidPath = process.env.T3_ACP_RUNNING_COMMAND_PID_PATH; +const agentPidPath = process.env.T3_ACP_AGENT_PID_PATH; const runningCommandSeparateSession = process.env.T3_ACP_RUNNING_COMMAND_SEPARATE_SESSION === "1"; const exitAfterRunningCommandLaunch = process.env.T3_ACP_EXIT_AFTER_RUNNING_COMMAND_LAUNCH === "1"; const omitXAiPromptCompleteStopReason = @@ -1379,6 +1380,9 @@ const program = Effect.gen(function* () { if (runningCommandPidPath !== undefined) { NodeFS.writeFileSync(runningCommandPidPath, String(command.pid)); } + if (agentPidPath !== undefined) { + NodeFS.writeFileSync(agentPidPath, String(process.pid)); + } if (reportOutlivingCommandExit) { yield* Effect.callback((resume) => { command.once("exit", () => resume(Effect.void)); diff --git a/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.test.ts index aa44c9fefdedc..50fd629f416bf 100644 --- a/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.test.ts @@ -536,6 +536,7 @@ describe("AntigravityAdapterV2 commands that outlive the prompt", () => { ); const workspace = yield* fileSystem.makeTempDirectoryScoped({ prefix: `t3-ag-${label}-` }); const commandPidPath = path.join(workspace, "command.pid"); + const agentPidPath = path.join(workspace, "agent.pid"); const commandPids: Array = []; yield* Effect.addFinalizer(() => Effect.sync(() => { @@ -569,6 +570,7 @@ describe("AntigravityAdapterV2 commands that outlive the prompt", () => { env: { T3_ACP_ANTIGRAVITY: "1", T3_ACP_RUNNING_COMMAND_PID_PATH: commandPidPath, + T3_ACP_AGENT_PID_PATH: agentPidPath, ...env, }, }, @@ -677,6 +679,7 @@ describe("AntigravityAdapterV2 commands that outlive the prompt", () => { waitForReply, waitForTerminal, readCommandPid, + readAgentPid: fileSystem.readFileString(agentPidPath).pipe(Effect.map(Number)), commandItem, finishArmed, }; @@ -723,6 +726,8 @@ describe("AntigravityAdapterV2 commands that outlive the prompt", () => { yield* mock.waitForReply; const commandPid = yield* mock.readCommandPid; assert.isTrue(pidIsRunning(commandPid)); + const agentPid = yield* mock.readAgentPid; + assert.isTrue(pidIsRunning(agentPid)); const providerTurnId = mock.commandItem("command-outliving-prompt")?.providerTurnId; if (providerTurnId == null) return yield* Effect.die("missing command item"); @@ -734,10 +739,14 @@ describe("AntigravityAdapterV2 commands that outlive the prompt", () => { assert.equal(yield* mock.waitForTerminal, "interrupted"); assert.equal(mock.commandItem("command-outliving-prompt")?.status, "interrupted"); assert.isFalse(pidIsRunning(commandPid), "Stop ends the command"); + assert.isFalse(pidIsRunning(agentPid), "Stop ends the ACP agent"); yield* mock.startTurn(2); yield* mock.waitForReply; - assert.isTrue(pidIsRunning(yield* mock.readCommandPid), "a respawned agent runs turn 2"); + assert.isTrue(pidIsRunning(yield* mock.readCommandPid), "turn 2 starts a command"); + const nextAgentPid = yield* mock.readAgentPid; + assert.notEqual(nextAgentPid, agentPid, "turn 2 respawns the ACP agent"); + assert.isTrue(pidIsRunning(nextAgentPid)); }).pipe(Effect.provide(layerSession), Effect.scoped), ); From c9edfe2c2abd87279b3d3270c1dd50c1a7a4ce4a Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 02:21:07 +0000 Subject: [PATCH 05/13] test(server): cover Antigravity fallback restart cleanup --- .../Adapters/AntigravityAdapterV2.test.ts | 60 ++++++++++--------- 1 file changed, 31 insertions(+), 29 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.test.ts index 50fd629f416bf..2cab61cefd2a2 100644 --- a/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.test.ts @@ -716,38 +716,40 @@ describe("AntigravityAdapterV2 commands that outlive the prompt", () => { }).pipe(Effect.provide(layerSession), Effect.scoped), ); - it.effect("lets Stop end the running command and respawns the agent for the next turn", () => - Effect.gen(function* () { - const mock = yield* openMockSession("command-stopped", { - T3_ACP_EMIT_COMMAND_OUTLIVING_PROMPT: "1", - }); - yield* mock.startTurn(1); + it.effect.each(["Stop", "fallback restart"])( + "lets %s end the running command and respawns the agent for the next turn", + (action) => + Effect.gen(function* () { + const mock = yield* openMockSession("command-stopped", { + T3_ACP_EMIT_COMMAND_OUTLIVING_PROMPT: "1", + }); + yield* mock.startTurn(1); - yield* mock.waitForReply; - const commandPid = yield* mock.readCommandPid; - assert.isTrue(pidIsRunning(commandPid)); - const agentPid = yield* mock.readAgentPid; - assert.isTrue(pidIsRunning(agentPid)); - const providerTurnId = mock.commandItem("command-outliving-prompt")?.providerTurnId; - if (providerTurnId == null) return yield* Effect.die("missing command item"); + yield* mock.waitForReply; + const commandPid = yield* mock.readCommandPid; + assert.isTrue(pidIsRunning(commandPid)); + const agentPid = yield* mock.readAgentPid; + assert.isTrue(pidIsRunning(agentPid)); + const providerTurnId = mock.commandItem("command-outliving-prompt")?.providerTurnId; + if (providerTurnId == null) return yield* Effect.die("missing command item"); - yield* mock.session.interruptTurn({ - providerThread: mock.providerThread, - providerTurnId, - requestRuntimeRestart: true, - }); - assert.equal(yield* mock.waitForTerminal, "interrupted"); - assert.equal(mock.commandItem("command-outliving-prompt")?.status, "interrupted"); - assert.isFalse(pidIsRunning(commandPid), "Stop ends the command"); - assert.isFalse(pidIsRunning(agentPid), "Stop ends the ACP agent"); + yield* mock.session.interruptTurn({ + providerThread: mock.providerThread, + providerTurnId, + ...(action === "Stop" ? { requestRuntimeRestart: true } : {}), + }); + assert.equal(yield* mock.waitForTerminal, "interrupted"); + assert.equal(mock.commandItem("command-outliving-prompt")?.status, "interrupted"); + assert.isFalse(pidIsRunning(commandPid), `${action} ends the command`); + assert.isFalse(pidIsRunning(agentPid), `${action} ends the ACP agent`); - yield* mock.startTurn(2); - yield* mock.waitForReply; - assert.isTrue(pidIsRunning(yield* mock.readCommandPid), "turn 2 starts a command"); - const nextAgentPid = yield* mock.readAgentPid; - assert.notEqual(nextAgentPid, agentPid, "turn 2 respawns the ACP agent"); - assert.isTrue(pidIsRunning(nextAgentPid)); - }).pipe(Effect.provide(layerSession), Effect.scoped), + yield* mock.startTurn(2); + yield* mock.waitForReply; + assert.isTrue(pidIsRunning(yield* mock.readCommandPid), "turn 2 starts a command"); + const nextAgentPid = yield* mock.readAgentPid; + assert.notEqual(nextAgentPid, agentPid, "turn 2 respawns the ACP agent"); + assert.isTrue(pidIsRunning(nextAgentPid)); + }).pipe(Effect.provide(layerSession), Effect.scoped), ); it.effect("completes a turn whose only open work is a subagent batch", () => From c0c6e1f6db79a8580d58c61c61e3c1e64890dc43 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 02:26:42 +0000 Subject: [PATCH 06/13] fix(server): restart Antigravity runtime on every interrupt --- .../server/src/orchestration-v2/Adapters/AntigravityAdapterV2.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.ts index eebbcdc8c8d47..86266aadc47b1 100644 --- a/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.ts @@ -168,6 +168,7 @@ export function makeAntigravityAcpAdapterFlavor( // without the replay and is what the official client does. preferResumeSession: true, subagentsIdleOnTurnCompletion: true, + restartRuntimeOnEveryInterrupt: true, terminateRuntimeProcessGroupOnInterrupt: true, // Commands can outlive end_turn; retain Stop until their terminal updates arrive. deferFinalizeForBackgroundWork: true, From 7787310856e345c9dd6c3aa02fa28f6313e8a7ec Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 02:39:35 +0000 Subject: [PATCH 07/13] test(server): cover Antigravity crash after prompt settlement --- .../Adapters/AntigravityAdapterV2.test.ts | 28 +++++++++++++------ 1 file changed, 20 insertions(+), 8 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.test.ts index 2cab61cefd2a2..0ad078d1cd5a6 100644 --- a/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.test.ts @@ -716,7 +716,7 @@ describe("AntigravityAdapterV2 commands that outlive the prompt", () => { }).pipe(Effect.provide(layerSession), Effect.scoped), ); - it.effect.each(["Stop", "fallback restart"])( + it.effect.each(["Stop", "fallback restart", "agent crash"])( "lets %s end the running command and respawns the agent for the next turn", (action) => Effect.gen(function* () { @@ -733,13 +733,25 @@ describe("AntigravityAdapterV2 commands that outlive the prompt", () => { const providerTurnId = mock.commandItem("command-outliving-prompt")?.providerTurnId; if (providerTurnId == null) return yield* Effect.die("missing command item"); - yield* mock.session.interruptTurn({ - providerThread: mock.providerThread, - providerTurnId, - ...(action === "Stop" ? { requestRuntimeRestart: true } : {}), - }); - assert.equal(yield* mock.waitForTerminal, "interrupted"); - assert.equal(mock.commandItem("command-outliving-prompt")?.status, "interrupted"); + if (action === "agent crash") { + process.kill(agentPid, "SIGKILL"); + } else { + yield* mock.session.interruptTurn({ + providerThread: mock.providerThread, + providerTurnId, + ...(action === "Stop" ? { requestRuntimeRestart: true } : {}), + }); + } + assert.equal( + yield* mock.waitForTerminal, + action === "agent crash" ? "failed" : "interrupted", + ); + const commandStatus = mock.commandItem("command-outliving-prompt")?.status; + if (action === "agent crash") { + assert.oneOf(commandStatus, ["failed", "interrupted"]); + } else { + assert.equal(commandStatus, "interrupted"); + } assert.isFalse(pidIsRunning(commandPid), `${action} ends the command`); assert.isFalse(pidIsRunning(agentPid), `${action} ends the ACP agent`); From 97b2e32bf7028e9487bc0b9e1bbd0b343b139231 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 02:46:34 +0000 Subject: [PATCH 08/13] fix(acp): settle deferred turns after runtime termination --- packages/provider-acp/src/server/adapter.ts | 53 +++++++++++++++++++-- 1 file changed, 48 insertions(+), 5 deletions(-) diff --git a/packages/provider-acp/src/server/adapter.ts b/packages/provider-acp/src/server/adapter.ts index 70b5038eca9c8..cf86ab78f7a0a 100644 --- a/packages/provider-acp/src/server/adapter.ts +++ b/packages/provider-acp/src/server/adapter.ts @@ -1857,17 +1857,47 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( Effect.void; const nativeLogging = options.nativeLogging?.(input.threadId); - const handleRuntimeTerminationAtGeneration = (runtimeGeneration: number) => + const handleRuntimeTerminationAtGeneration = ( + runtimeGeneration: number, + error: EffectAcpErrors.AcpError, + ) => runRuntimeCallbackAtGeneration( runtimeGeneration, - Ref.set(runtimeRestartRequired, true), + Effect.gen(function* () { + yield* Ref.set(runtimeRestartRequired, true); + const context = yield* Ref.get(activeTurn); + // In-flight prompts report termination through their existing failure path. + if (context === null || !context.promptSettled || context.finalized) return; + yield* quarantineStoppedRun(); + yield* cancelPendingRuntimeRequests(); + const terminatedScope = runtimeScope; + // Return from the transport callback before closing its own scope. + yield* Effect.gen(function* () { + if (terminatedScope !== undefined) { + yield* Scope.close(terminatedScope, Exit.fail(error)); + } + yield* runRuntimeCallbackAtGeneration( + runtimeGeneration, + Effect.gen(function* () { + if (context.finalized) return; + yield* terminalizeOpenRunOwnedItems(context, { terminalizeSubagents: true }); + yield* finalizeTurn( + context, + "failed", + flavor.promptFailure?.(error) ?? + makeProviderFailure({ cause: error, class: "provider_error" }), + ); + }), + ); + }).pipe(Effect.forkIn(sessionScope), Effect.asVoid); + }), ).pipe(Effect.asVoid); const makeRuntimeInput = ( runtimeGeneration: number, mcpContext: AcpMcpContext, resumeSessionId?: string, - onTermination: AcpAdapterV2RuntimeInput["onTermination"] = () => - handleRuntimeTerminationAtGeneration(runtimeGeneration), + onTermination: AcpAdapterV2RuntimeInput["onTermination"] = (error) => + handleRuntimeTerminationAtGeneration(runtimeGeneration, error), ): AcpAdapterV2RuntimeInput => { return { cwd: input.runtimePolicy.cwd ?? process.cwd(), @@ -5629,7 +5659,10 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( candidateLifecycle, (current): readonly [Effect.Effect, CandidateLifecycle] => { if (current._tag === "Committed") { - return [handleRuntimeTerminationAtGeneration(replacementGeneration), current]; + return [ + handleRuntimeTerminationAtGeneration(replacementGeneration, error), + current, + ]; } return [ Effect.void, @@ -6391,6 +6424,16 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( ) { const restartRequired = yield* Ref.get(runtimeRestartRequired); if (!restartRequired) return false; + yield* runtimeCallbackPermit.withPermit( + Effect.gen(function* () { + yield* quarantineNativeTransportAtGeneration( + yield* Ref.get(runtimeCallbackGeneration), + ); + yield* advanceRuntimeCallbackGeneration; + yield* quarantineStoppedRun(); + yield* Ref.set(endedBackgroundTaskIds, new Set()); + }), + ); yield* restartAcpRuntime(threadId); yield* Ref.set(runtimeRestartRequired, false); yield* Ref.set(activeSessionId, null); From 59d114c5352c3443b9b23b908b3d507dcbe45753 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 02:56:01 +0000 Subject: [PATCH 09/13] test(server): reproduce Antigravity crash before prompt settlement --- .../Adapters/AntigravityAdapterV2.test.ts | 107 +++++++++++++----- 1 file changed, 81 insertions(+), 26 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.test.ts index 0ad078d1cd5a6..ac2c28d7a8868 100644 --- a/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.test.ts @@ -525,7 +525,11 @@ describe("AntigravityAdapterV2 commands that outlive the prompt", () => { } }; - const openMockSession = (label: string, env: Record) => + const openMockSession = ( + label: string, + env: Record, + crashBeforeSettlement = false, + ) => Effect.gen(function* () { const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; const fileSystem = yield* FileSystem.FileSystem; @@ -551,6 +555,8 @@ describe("AntigravityAdapterV2 commands that outlive the prompt", () => { }); // A deferred finish waits out a debounce; the hook says how far to move the clock. const finishArmed = yield* Deferred.make(); + const terminationHandled = yield* Deferred.make(); + const promptSettledWithBackgroundWork = yield* Deferred.make(); const instanceId = ProviderInstanceId.make(`antigravity-${label}`); const adapter = yield* makeAntigravityAdapterV2({ instanceId, @@ -558,10 +564,17 @@ describe("AntigravityAdapterV2 commands that outlive the prompt", () => { testHooks: { onDeferredFinalizeScheduled: (debounce) => Deferred.succeed(finishArmed, debounce).pipe(Effect.asVoid), + afterPromptSettledWithBackgroundWork: () => + Deferred.succeed(promptSettledWithBackgroundWork, undefined).pipe(Effect.asVoid), }, makeRuntime: (input) => makeAntigravityAcpRuntime({ ...input, + onTermination: (error) => + (input.onTermination?.(error) ?? Effect.void).pipe( + Effect.andThen(Deferred.succeed(terminationHandled, undefined)), + Effect.asVoid, + ), childProcessSpawner, spawn: { command: process.execPath, @@ -574,7 +587,32 @@ describe("AntigravityAdapterV2 commands that outlive the prompt", () => { ...env, }, }, - }).pipe(Effect.provideService(Crypto.Crypto, crypto)), + }).pipe( + Effect.provideService(Crypto.Crypto, crypto), + Effect.map((runtime) => { + if (!crashBeforeSettlement) return runtime; + return { + ...runtime, + prompt: (...args: Parameters) => + runtime.prompt(...args).pipe( + Effect.tap(() => + Effect.gen(function* () { + if (!crashBeforeSettlement) return; + crashBeforeSettlement = false; + // Hold a genuine successful response while termination wins the + // adapter callback permit. The prompt consumer lives in session scope. + const commandPid = yield* readCommandPid; + const agentPid = Number(yield* fileSystem.readFileString(agentPidPath)); + assert.isTrue(pidIsRunning(commandPid)); + assert.isTrue(pidIsRunning(agentPid)); + process.kill(agentPid, "SIGKILL"); + yield* Deferred.await(terminationHandled); + }).pipe(Effect.orDie), + ), + ), + }; + }), + ), withProcess: (_stop, task) => task, defaultModel: Effect.succeed(undefined), }); @@ -682,6 +720,7 @@ describe("AntigravityAdapterV2 commands that outlive the prompt", () => { readAgentPid: fileSystem.readFileString(agentPidPath).pipe(Effect.map(Number)), commandItem, finishArmed, + promptSettledWithBackgroundWork, }; }); @@ -716,38 +755,54 @@ describe("AntigravityAdapterV2 commands that outlive the prompt", () => { }).pipe(Effect.provide(layerSession), Effect.scoped), ); - it.effect.each(["Stop", "fallback restart", "agent crash"])( + it.effect.each(["Stop", "fallback restart", "agent crash", "agent crash before settlement"])( "lets %s end the running command and respawns the agent for the next turn", (action) => Effect.gen(function* () { - const mock = yield* openMockSession("command-stopped", { - T3_ACP_EMIT_COMMAND_OUTLIVING_PROMPT: "1", - }); + const crashBeforeSettlement = action === "agent crash before settlement"; + const crashed = action === "agent crash" || crashBeforeSettlement; + const mock = yield* openMockSession( + "command-stopped", + { T3_ACP_EMIT_COMMAND_OUTLIVING_PROMPT: "1" }, + crashBeforeSettlement, + ); yield* mock.startTurn(1); - yield* mock.waitForReply; - const commandPid = yield* mock.readCommandPid; - assert.isTrue(pidIsRunning(commandPid)); - const agentPid = yield* mock.readAgentPid; - assert.isTrue(pidIsRunning(agentPid)); - const providerTurnId = mock.commandItem("command-outliving-prompt")?.providerTurnId; - if (providerTurnId == null) return yield* Effect.die("missing command item"); + if (!crashBeforeSettlement) { + yield* mock.waitForReply; + assert.isTrue(pidIsRunning(yield* mock.readCommandPid)); + const agentPid = yield* mock.readAgentPid; + assert.isTrue(pidIsRunning(agentPid)); + const providerTurnId = mock.commandItem("command-outliving-prompt")?.providerTurnId; + if (providerTurnId == null) return yield* Effect.die("missing command item"); - if (action === "agent crash") { - process.kill(agentPid, "SIGKILL"); - } else { - yield* mock.session.interruptTurn({ - providerThread: mock.providerThread, - providerTurnId, - ...(action === "Stop" ? { requestRuntimeRestart: true } : {}), - }); + if (action === "agent crash") { + process.kill(agentPid, "SIGKILL"); + } else { + yield* mock.session.interruptTurn({ + providerThread: mock.providerThread, + providerTurnId, + ...(action === "Stop" ? { requestRuntimeRestart: true } : {}), + }); + } } - assert.equal( - yield* mock.waitForTerminal, - action === "agent crash" ? "failed" : "interrupted", - ); + const terminal = crashBeforeSettlement + ? Effect.raceFirst( + mock.waitForTerminal, + Deferred.await(mock.promptSettledWithBackgroundWork).pipe( + Effect.andThen( + Effect.sync(() => + assert.fail("the terminated agent left the turn waiting for background work"), + ), + ), + ), + ) + : mock.waitForTerminal; + assert.equal(yield* terminal, crashed ? "failed" : "interrupted"); + const commandPid = yield* mock.readCommandPid; + const agentPid = yield* mock.readAgentPid; const commandStatus = mock.commandItem("command-outliving-prompt")?.status; - if (action === "agent crash") { + if (crashed) { assert.oneOf(commandStatus, ["failed", "interrupted"]); } else { assert.equal(commandStatus, "interrupted"); From 5a411e9d3871fc4b56cbc12c2b89ce41ba18f3b5 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 03:00:34 +0000 Subject: [PATCH 10/13] fix(acp): fail held prompt results after runtime termination --- packages/provider-acp/src/server/adapter.ts | 68 ++++++++++++++------- 1 file changed, 46 insertions(+), 22 deletions(-) diff --git a/packages/provider-acp/src/server/adapter.ts b/packages/provider-acp/src/server/adapter.ts index cf86ab78f7a0a..83ec2d403f077 100644 --- a/packages/provider-acp/src/server/adapter.ts +++ b/packages/provider-acp/src/server/adapter.ts @@ -1498,6 +1498,10 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( const activeInteractionMode = yield* Ref.make(null); const promptInstructionStates = yield* Ref.make(new Map()); const runtimeRestartRequired = yield* Ref.make(false); + const runtimeTermination = yield* Ref.make<{ + readonly generation: number; + readonly error: EffectAcpErrors.AcpError; + } | null>(null); const runtimeTeardownState = yield* Ref.make({ _tag: "Idle" }); const runtimeCallbackGeneration = yield* Ref.make(0); const runtimeCallbackGenerationCounter = yield* Ref.make(0); @@ -1857,6 +1861,34 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( Effect.void; const nativeLogging = options.nativeLogging?.(input.threadId); + const failTurnAfterRuntimeTermination = Effect.fnUntraced(function* ( + context: ActiveAcpTurn, + runtimeGeneration: number, + error: EffectAcpErrors.AcpError, + ) { + yield* quarantineStoppedRun(); + yield* cancelPendingRuntimeRequests(); + const terminatedScope = runtimeScope; + // Return from the transport callback before closing its own scope. + yield* Effect.gen(function* () { + if (terminatedScope !== undefined) { + yield* Scope.close(terminatedScope, Exit.fail(error)); + } + yield* runRuntimeCallbackAtGeneration( + runtimeGeneration, + Effect.gen(function* () { + if (context.finalized) return; + yield* terminalizeOpenRunOwnedItems(context, { terminalizeSubagents: true }); + yield* finalizeTurn( + context, + "failed", + flavor.promptFailure?.(error) ?? + makeProviderFailure({ cause: error, class: "provider_error" }), + ); + }), + ); + }).pipe(Effect.forkIn(sessionScope), Effect.asVoid); + }); const handleRuntimeTerminationAtGeneration = ( runtimeGeneration: number, error: EffectAcpErrors.AcpError, @@ -1865,31 +1897,11 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( runtimeGeneration, Effect.gen(function* () { yield* Ref.set(runtimeRestartRequired, true); + yield* Ref.set(runtimeTermination, { generation: runtimeGeneration, error }); const context = yield* Ref.get(activeTurn); // In-flight prompts report termination through their existing failure path. if (context === null || !context.promptSettled || context.finalized) return; - yield* quarantineStoppedRun(); - yield* cancelPendingRuntimeRequests(); - const terminatedScope = runtimeScope; - // Return from the transport callback before closing its own scope. - yield* Effect.gen(function* () { - if (terminatedScope !== undefined) { - yield* Scope.close(terminatedScope, Exit.fail(error)); - } - yield* runRuntimeCallbackAtGeneration( - runtimeGeneration, - Effect.gen(function* () { - if (context.finalized) return; - yield* terminalizeOpenRunOwnedItems(context, { terminalizeSubagents: true }); - yield* finalizeTurn( - context, - "failed", - flavor.promptFailure?.(error) ?? - makeProviderFailure({ cause: error, class: "provider_error" }), - ); - }), - ); - }).pipe(Effect.forkIn(sessionScope), Effect.asVoid); + yield* failTurnAfterRuntimeTermination(context, runtimeGeneration, error); }), ).pipe(Effect.asVoid); const makeRuntimeInput = ( @@ -6430,7 +6442,9 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( yield* Ref.get(runtimeCallbackGeneration), ); yield* advanceRuntimeCallbackGeneration; + const carryover = yield* Ref.getAndSet(carryoverSubagents, null); yield* quarantineStoppedRun(); + yield* terminalizeCarryoverSubagents(carryover); yield* Ref.set(endedBackgroundTaskIds, new Set()); }), ); @@ -6728,6 +6742,16 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( promptGeneration, Effect.gen(function* () { if (context.finalized) return; + // A session-scoped consumer can hold a successful result past runtime exit. + const termination = yield* Ref.get(runtimeTermination); + if (termination?.generation === promptGeneration) { + yield* failTurnAfterRuntimeTermination( + context, + promptGeneration, + termination.error, + ); + return; + } const status = result.stopReason === "cancelled" ? context.interrupted From 0ec8b18145df0540e593fc9ddf1e285b4f0a48a3 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 03:10:38 +0000 Subject: [PATCH 11/13] fix(acp): scope runtime exit check to deferred background work --- packages/provider-acp/src/server/adapter.ts | 20 ++++++++++---------- 1 file changed, 10 insertions(+), 10 deletions(-) diff --git a/packages/provider-acp/src/server/adapter.ts b/packages/provider-acp/src/server/adapter.ts index 83ec2d403f077..dc952ca978d1d 100644 --- a/packages/provider-acp/src/server/adapter.ts +++ b/packages/provider-acp/src/server/adapter.ts @@ -6742,16 +6742,6 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( promptGeneration, Effect.gen(function* () { if (context.finalized) return; - // A session-scoped consumer can hold a successful result past runtime exit. - const termination = yield* Ref.get(runtimeTermination); - if (termination?.generation === promptGeneration) { - yield* failTurnAfterRuntimeTermination( - context, - promptGeneration, - termination.error, - ); - return; - } const status = result.stopReason === "cancelled" ? context.interrupted @@ -6763,6 +6753,16 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( !context.interrupted && hasDeferredBackgroundWork(context) ) { + // A session-scoped consumer can hold a successful result past runtime exit. + const termination = yield* Ref.get(runtimeTermination); + if (termination?.generation === promptGeneration) { + yield* failTurnAfterRuntimeTermination( + context, + promptGeneration, + termination.error, + ); + return; + } context.promptSettled = true; context.promptSettledStatus = status; yield* closeTextStreams(context); From 6cad6106c16397ffefb881e5b1b5f5d2b886f543 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 11 Nov 2026 07:25:12 +0000 Subject: [PATCH 12/13] fix(antigravity): end turns at end_turn and keep Stop ending started commands Real Antigravity 1.4.5 never reports a command's end once its prompt has returned, so holding the turn open for running commands left it active until Stop and queued follow-up messages behind long-running commands such as dev servers. Antigravity turns end at end_turn again, as on main. The agent still runs in its own process group and every interrupt restarts it, so Stop during an active turn ends the commands the agent started. The shared ACP termination handling and subagent gate only served turns held open past end_turn, so they are reverted. AI-assisted; verified with the tests and tools listed above. Co-Authored-By: Claude Opus 5.5 --- apps/server/scripts/acp-mock-agent.ts | 73 +------- .../Adapters/AntigravityAdapterV2.test.ts | 171 +++--------------- .../Adapters/AntigravityAdapterV2.ts | 8 - packages/provider-acp/src/server/adapter.ts | 87 +-------- 4 files changed, 38 insertions(+), 301 deletions(-) diff --git a/apps/server/scripts/acp-mock-agent.ts b/apps/server/scripts/acp-mock-agent.ts index 080b970f59ba2..5d61073a6885a 100644 --- a/apps/server/scripts/acp-mock-agent.ts +++ b/apps/server/scripts/acp-mock-agent.ts @@ -57,9 +57,7 @@ const emitTrackedToolCallKindThenHang = process.env.T3_ACP_EMIT_TRACKED_TOOL_CAL const clientFsProbePath = process.env.T3_ACP_CLIENT_FS_PROBE_PATH; const clientFsProbeLogPath = process.env.T3_ACP_CLIENT_FS_PROBE_LOG_PATH; const encodeJson = Schema.encodeSync(Schema.fromJsonString(Schema.Unknown)); -const emitCommandOutlivingPrompt = process.env.T3_ACP_EMIT_COMMAND_OUTLIVING_PROMPT === "1"; -const reportOutlivingCommandExit = process.env.T3_ACP_REPORT_OUTLIVING_COMMAND_EXIT === "1"; -const emitSubagentLaunchThenEndTurn = process.env.T3_ACP_EMIT_SUBAGENT_LAUNCH_THEN_END_TURN === "1"; +const emitRunningCommandThenHang = process.env.T3_ACP_EMIT_RUNNING_COMMAND_THEN_HANG === "1"; const runningCommandPidPath = process.env.T3_ACP_RUNNING_COMMAND_PID_PATH; const agentPidPath = process.env.T3_ACP_AGENT_PID_PATH; const emitEmptySuccessfulBash = process.env.T3_ACP_EMIT_EMPTY_SUCCESSFUL_BASH === "1"; @@ -1357,9 +1355,8 @@ const program = Effect.gen(function* () { return yield* AcpError.AcpRequestError.internalError("Mock prompt failure"); } - if (emitCommandOutlivingPrompt) { - // Antigravity can answer end_turn while a command it started keeps running. - // A long-lived Node child stands in for the command on every platform. + if (emitRunningCommandThenHang) { + // The command is a long-lived Node child so the scenario runs on every platform. const command = NodeChildProcess.spawn( process.execPath, ["-e", "setInterval(() => {}, 1 << 30)"], @@ -1375,78 +1372,18 @@ const program = Effect.gen(function* () { if (agentPidPath !== undefined) { NodeFS.writeFileSync(agentPidPath, String(process.pid)); } - if (reportOutlivingCommandExit) { - yield* Effect.callback((resume) => { - command.once("exit", () => resume(Effect.void)); - }).pipe( - Effect.andThen( - agent.client.sessionUpdate({ - sessionId: requestedSessionId, - update: { - sessionUpdate: "tool_call_update", - toolCallId: "command-outliving-prompt", - status: "completed", - }, - }), - ), - Effect.ignore, - Effect.forkDetach, - ); - } - // Awaiting approval: not started, so it must not hold the turn open. - yield* agent.client.sessionUpdate({ - sessionId: requestedSessionId, - update: { - sessionUpdate: "tool_call_update", - toolCallId: "command-awaiting-approval", - title: "run_command", - kind: "execute", - status: "pending", - rawInput: { command: "npm test" }, - }, - }); yield* agent.client.sessionUpdate({ sessionId: requestedSessionId, update: { sessionUpdate: "tool_call_update", - toolCallId: "command-outliving-prompt", + toolCallId: "running-command", title: "run_command", kind: "execute", status: "in_progress", rawInput: { command: "npm run dev" }, }, }); - yield* agent.client.sessionUpdate({ - sessionId: requestedSessionId, - update: { - sessionUpdate: "agent_message_chunk", - content: { type: "text", text: "Started the command." }, - }, - }); - return yield* finishPrompt(requestedSessionId, "end_turn"); - } - - if (emitSubagentLaunchThenEndTurn) { - // ACP 1.1.1 batch launch: completes at once and never reports the children. - yield* agent.client.sessionUpdate({ - sessionId: requestedSessionId, - update: { - sessionUpdate: "tool_call_update", - toolCallId: "subagent-launch", - title: "Running start_subagent", - kind: "other", - status: "completed", - rawOutput: "Started two agents.", - }, - }); - yield* agent.client.sessionUpdate({ - sessionId: requestedSessionId, - update: { - sessionUpdate: "agent_message_chunk", - content: { type: "text", text: "Started the agents." }, - }, - }); - return yield* finishPrompt(requestedSessionId, "end_turn"); + return yield* Effect.never; } if (emitTrackedToolCallKindThenHang !== undefined) { diff --git a/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.test.ts index fdf36ff0ea78b..a5be9f7126040 100644 --- a/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.test.ts @@ -17,7 +17,6 @@ import { htmlRenderFromToolItem } from "@t3tools/shared/toolOutput"; import * as SecureRandom from "effect/crypto/SecureRandom"; import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; -import type * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; import * as FileSystem from "effect/FileSystem"; @@ -25,7 +24,6 @@ import * as Layer from "effect/Layer"; import * as Path from "effect/Path"; import * as Queue from "effect/Queue"; import * as Stream from "effect/Stream"; -import * as TestClock from "effect/testing/TestClock"; import { ChildProcessSpawner } from "effect/process"; import type * as EffectAcpSchema from "effect-acp/compat"; import * as EffectAcpErrors from "effect-acp/errors"; @@ -1247,7 +1245,7 @@ describe("AntigravityAdapterV2 session system prompt", () => { ); }); -describe("AntigravityAdapterV2 commands that outlive the prompt", () => { +describe("AntigravityAdapterV2 commands started during a turn", () => { const pidIsRunning = (pid: number) => { try { process.kill(pid, 0); @@ -1257,11 +1255,7 @@ describe("AntigravityAdapterV2 commands that outlive the prompt", () => { } }; - const openMockSession = ( - label: string, - env: Record, - crashBeforeSettlement = false, - ) => + const openMockSession = (label: string) => Effect.gen(function* () { const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; const fileSystem = yield* FileSystem.FileSystem; @@ -1279,34 +1273,19 @@ describe("AntigravityAdapterV2 commands that outlive the prompt", () => { for (const pid of commandPids) if (pidIsRunning(pid)) process.kill(pid, "SIGKILL"); }), ); - // The mock publishes the PID once the command has spawned, before it answers. + // The mock publishes the PID once the command has spawned, before it reports the command. const readCommandPid = Effect.gen(function* () { const pid = Number(yield* fileSystem.readFileString(commandPidPath)); commandPids.push(pid); return pid; }); - // A deferred finish waits out a debounce; the hook says how far to move the clock. - const finishArmed = yield* Deferred.make(); - const terminationHandled = yield* Deferred.make(); - const promptSettledWithBackgroundWork = yield* Deferred.make(); const instanceId = ProviderInstanceId.make(`antigravity-${label}`); const adapter = yield* makeAntigravityAdapterV2({ instanceId, selfInvocation: yield* resolveSelfInvocation(), - testHooks: { - onDeferredFinalizeScheduled: (debounce) => - Deferred.succeed(finishArmed, debounce).pipe(Effect.asVoid), - afterPromptSettledWithBackgroundWork: () => - Deferred.succeed(promptSettledWithBackgroundWork, undefined).pipe(Effect.asVoid), - }, makeRuntime: (input) => makeAntigravityAcpRuntime({ ...input, - onTermination: (error) => - (input.onTermination?.(error) ?? Effect.void).pipe( - Effect.andThen(Deferred.succeed(terminationHandled, undefined)), - Effect.asVoid, - ), childProcessSpawner, spawn: { command: process.execPath, @@ -1316,35 +1295,10 @@ describe("AntigravityAdapterV2 commands that outlive the prompt", () => { T3_ACP_ANTIGRAVITY: "1", T3_ACP_RUNNING_COMMAND_PID_PATH: commandPidPath, T3_ACP_AGENT_PID_PATH: agentPidPath, - ...env, + T3_ACP_EMIT_RUNNING_COMMAND_THEN_HANG: "1", }, }, - }).pipe( - Effect.provideService(SecureRandom.SecureRandom, crypto), - Effect.map((runtime) => { - if (!crashBeforeSettlement) return runtime; - return { - ...runtime, - prompt: (...args: Parameters) => - runtime.prompt(...args).pipe( - Effect.tap(() => - Effect.gen(function* () { - if (!crashBeforeSettlement) return; - crashBeforeSettlement = false; - // Hold a genuine successful response while termination wins the - // adapter callback permit. The prompt consumer lives in session scope. - const commandPid = yield* readCommandPid; - const agentPid = Number(yield* fileSystem.readFileString(agentPidPath)); - assert.isTrue(pidIsRunning(commandPid)); - assert.isTrue(pidIsRunning(agentPid)); - process.kill(agentPid, "SIGKILL"); - yield* Deferred.await(terminationHandled); - }).pipe(Effect.orDie), - ), - ), - }; - }), - ), + }).pipe(Effect.provideService(SecureRandom.SecureRandom, crypto)), withProcess: (_stop, task) => task, defaultModel: Effect.undefined, }); @@ -1377,11 +1331,11 @@ describe("AntigravityAdapterV2 commands that outlive the prompt", () => { Effect.tap((event) => Effect.sync(() => seen.push(event))), Effect.repeat({ until: predicate }), ); - const waitForReply = waitFor( + const waitForCommand = waitFor( (event) => - event.type === "message.updated" && - event.message.role === "assistant" && - !event.message.streaming, + event.type === "turn_item.updated" && + event.turnItem.type === "command_execution" && + event.turnItem.nativeItemRef?.nativeId?.includes("running-command") === true, ); const waitForTerminal = waitFor((event) => event.type === "turn.terminal").pipe( Effect.map((event) => (event.type === "turn.terminal" ? event.status : undefined)), @@ -1446,118 +1400,43 @@ describe("AntigravityAdapterV2 commands that outlive the prompt", () => { session, providerThread, startTurn, - waitForReply, + waitForCommand, waitForTerminal, readCommandPid, readAgentPid: fileSystem.readFileString(agentPidPath).pipe(Effect.map(Number)), commandItem, - finishArmed, - promptSettledWithBackgroundWork, }; }); - it.effect("keeps the turn open until the running command reports its end", () => - Effect.gen(function* () { - const mock = yield* openMockSession("command-finishes", { - T3_ACP_EMIT_COMMAND_OUTLIVING_PROMPT: "1", - T3_ACP_REPORT_OUTLIVING_COMMAND_EXIT: "1", - }); - yield* mock.startTurn(1); - - // The agent has answered end_turn; the command it started is still running, - // next to a second command still awaiting approval. - yield* mock.waitForReply; - const commandPid = yield* mock.readCommandPid; - assert.isTrue(pidIsRunning(commandPid)); - - // The mock reports the command's terminal update when it exits. Only then - // may the turn arm its finish; ending before it means it never waited. - process.kill(commandPid); - const turnEndedEarly = mock.waitForTerminal.pipe( - Effect.flatMap((status) => - Effect.sync(() => - assert.fail(`the turn ended (${status}) before its command reported its end`), - ), - ), - ); - const debounce = yield* Effect.raceFirst(Deferred.await(mock.finishArmed), turnEndedEarly); - yield* TestClock.adjust(debounce); - assert.equal(yield* mock.waitForTerminal, "completed"); - assert.equal(mock.commandItem("command-outliving-prompt")?.status, "completed"); - }).pipe(Effect.provide(layerSession), Effect.scoped), - ); - - it.effect.each(["Stop", "fallback restart", "agent crash", "agent crash before settlement"])( + it.effect.each(["Stop", "fallback restart"])( "lets %s end the running command and respawns the agent for the next turn", (action) => Effect.gen(function* () { - const crashBeforeSettlement = action === "agent crash before settlement"; - const crashed = action === "agent crash" || crashBeforeSettlement; - const mock = yield* openMockSession( - "command-stopped", - { T3_ACP_EMIT_COMMAND_OUTLIVING_PROMPT: "1" }, - crashBeforeSettlement, - ); + const mock = yield* openMockSession("command-stopped"); yield* mock.startTurn(1); - - if (!crashBeforeSettlement) { - yield* mock.waitForReply; - assert.isTrue(pidIsRunning(yield* mock.readCommandPid)); - const agentPid = yield* mock.readAgentPid; - assert.isTrue(pidIsRunning(agentPid)); - const providerTurnId = mock.commandItem("command-outliving-prompt")?.providerTurnId; - if (providerTurnId == null) return yield* Effect.die("missing command item"); - - if (action === "agent crash") { - process.kill(agentPid, "SIGKILL"); - } else { - yield* mock.session.interruptTurn({ - providerThread: mock.providerThread, - providerTurnId, - ...(action === "Stop" ? { requestRuntimeRestart: true } : {}), - }); - } - } - const terminal = crashBeforeSettlement - ? Effect.raceFirst( - mock.waitForTerminal, - Deferred.await(mock.promptSettledWithBackgroundWork).pipe( - Effect.andThen( - Effect.sync(() => - assert.fail("the terminated agent left the turn waiting for background work"), - ), - ), - ), - ) - : mock.waitForTerminal; - assert.equal(yield* terminal, crashed ? "failed" : "interrupted"); + yield* mock.waitForCommand; const commandPid = yield* mock.readCommandPid; + assert.isTrue(pidIsRunning(commandPid)); const agentPid = yield* mock.readAgentPid; - const commandStatus = mock.commandItem("command-outliving-prompt")?.status; - if (crashed) { - assert.oneOf(commandStatus, ["failed", "interrupted"]); - } else { - assert.equal(commandStatus, "interrupted"); - } + assert.isTrue(pidIsRunning(agentPid)); + const providerTurnId = mock.commandItem("running-command")?.providerTurnId; + if (providerTurnId == null) return yield* Effect.die("missing command item"); + + yield* mock.session.interruptTurn({ + providerThread: mock.providerThread, + providerTurnId, + ...(action === "Stop" ? { requestRuntimeRestart: true } : {}), + }); + assert.equal(yield* mock.waitForTerminal, "interrupted"); assert.isFalse(pidIsRunning(commandPid), `${action} ends the command`); assert.isFalse(pidIsRunning(agentPid), `${action} ends the ACP agent`); yield* mock.startTurn(2); - yield* mock.waitForReply; + yield* mock.waitForCommand; assert.isTrue(pidIsRunning(yield* mock.readCommandPid), "turn 2 starts a command"); const nextAgentPid = yield* mock.readAgentPid; assert.notEqual(nextAgentPid, agentPid, "turn 2 respawns the ACP agent"); assert.isTrue(pidIsRunning(nextAgentPid)); }).pipe(Effect.provide(layerSession), Effect.scoped), ); - - it.effect("completes a turn whose only open work is a subagent batch", () => - Effect.gen(function* () { - const mock = yield* openMockSession("subagent-batch", { - T3_ACP_EMIT_SUBAGENT_LAUNCH_THEN_END_TURN: "1", - }); - yield* mock.startTurn(1); - assert.equal(yield* mock.waitForTerminal, "completed"); - }).pipe(Effect.provide(layerSession), Effect.scoped), - ); }); diff --git a/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.ts index ba834fb606e3e..24cb141aaca3c 100644 --- a/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.ts @@ -90,7 +90,6 @@ export interface AntigravityAdapterV2Options { readonly onSessionEvent?: AcpAdapterV2Flavor["onSessionEvent"]; readonly nativeLogging?: Parameters[0]["nativeLogging"]; readonly continuationRequests?: Parameters[0]["continuationRequests"]; - readonly testHooks?: Parameters[0]["testHooks"]; /** The agent's GEMINI_HOME. Its session logs show turns that ended on an upstream error. */ readonly profileDirectory?: string; } @@ -241,12 +240,6 @@ export function makeAntigravityAcpAdapterFlavor( subagentsIdleOnTurnCompletion: true, restartRuntimeOnEveryInterrupt: true, terminateRuntimeProcessGroupOnInterrupt: true, - // Commands can outlive end_turn; retain Stop until their terminal updates arrive. - deferFinalizeForBackgroundWork: true, - extractBackgroundTaskId: (toolCall) => - toolCall.kind === "execute" && toolCall.status !== undefined && toolCall.status !== "pending" - ? toolCall.toolCallId - : undefined, ...(options.onSessionEvent === undefined ? {} : { onSessionEvent: options.onSessionEvent }), applyModelSelection: Effect.fn("AntigravityAdapterV2.applyModelSelection")(function* ({ runtime, @@ -318,6 +311,5 @@ export const makeAntigravityAdapterV2 = Effect.fn("makeAntigravityAdapterV2")(fu ...(options.continuationRequests === undefined ? {} : { continuationRequests: options.continuationRequests }), - ...(options.testHooks === undefined ? {} : { testHooks: options.testHooks }), }); }); diff --git a/packages/provider-acp/src/server/adapter.ts b/packages/provider-acp/src/server/adapter.ts index ea7e04c1d2cdc..a2c2c429fa88c 100644 --- a/packages/provider-acp/src/server/adapter.ts +++ b/packages/provider-acp/src/server/adapter.ts @@ -1936,10 +1936,6 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( // Unique markers keep delayed prompt completions from clearing later compactions. const compactionMarkers = yield* Ref.make(new Map()); const runtimeRestartRequired = yield* Ref.make(false); - const runtimeTermination = yield* Ref.make<{ - readonly generation: number; - readonly error: EffectAcpErrors.AcpError; - } | null>(null); const runtimeTeardownState = yield* Ref.make({ _tag: "Idle" }); const runtimeCallbackGeneration = yield* Ref.make(0); const runtimeCallbackGenerationCounter = yield* Ref.make(0); @@ -2316,50 +2312,12 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( ? notePromptProgress() : Effect.void, ); - const failTurnAfterRuntimeTermination = Effect.fnUntraced(function* ( - context: ActiveAcpTurn, - runtimeGeneration: number, - error: EffectAcpErrors.AcpError, - ) { - yield* quarantineStoppedRun(); - yield* cancelPendingRuntimeRequests(); - const terminatedScope = runtimeScope; - // Return from the transport callback before closing its own scope. - yield* Effect.gen(function* () { - if (terminatedScope !== undefined) { - yield* Scope.close(terminatedScope, Exit.fail(error)); - } - yield* runRuntimeCallbackAtGeneration( - runtimeGeneration, - Effect.gen(function* () { - if (context.finalized) return; - yield* terminalizeOpenRunOwnedItems(context, { terminalizeSubagents: true }); - yield* finalizeTurn( - context, - "failed", - flavor.promptFailure?.(error) ?? - makeProviderFailure({ cause: error, class: "provider_error" }), - ); - }), - ); - }).pipe(Effect.forkIn(sessionScope), Effect.asVoid); - }); - const handleRuntimeTerminationAtGeneration = ( - runtimeGeneration: number, - error: EffectAcpErrors.AcpError, - ) => + const handleRuntimeTerminationAtGeneration = (runtimeGeneration: number) => runRuntimeCallbackAtGeneration( runtimeGeneration, Effect.gen(function* () { yield* Ref.set(runtimeRestartRequired, true); - yield* Ref.set(runtimeTermination, { generation: runtimeGeneration, error }); - if (!followsAgentWorkState) { - const context = yield* Ref.get(activeTurn); - // In-flight prompts report termination through their existing failure path. - if (context === null || !context.promptSettled || context.finalized) return; - yield* failTurnAfterRuntimeTermination(context, runtimeGeneration, error); - return; - } + if (!followsAgentWorkState) return; agentWork = { sessionUpdate: "state_update", state: "idle", stopReason: "error" }; for (const child of nativeChildren.values()) { if (child.status !== "running") continue; @@ -2393,8 +2351,8 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( agentRequests: Set, mcpContext: AcpMcpContext, resumeSessionId?: string, - onTermination: AcpAdapterV2RuntimeInput["onTermination"] = (error) => - handleRuntimeTerminationAtGeneration(runtimeGeneration, error), + onTermination: AcpAdapterV2RuntimeInput["onTermination"] = () => + handleRuntimeTerminationAtGeneration(runtimeGeneration), ): AcpAdapterV2RuntimeInput => { return { cwd: input.runtimePolicy.cwd ?? process.cwd(), @@ -3795,13 +3753,9 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( const status = toolStatus(tool.status); if (status === "pending" || status === "running") return true; } - // Batch launches without child completion signals go idle when the - // root turn ends, so they must not hold it open. - if (flavor.subagentsIdleOnTurnCompletion !== true) { - for (const subagent of context.subagents.values()) { - if (acpSubagentStatusBlocksTurnSettlement(subagent.task.status)) { - return true; - } + for (const subagent of context.subagents.values()) { + if (acpSubagentStatusBlocksTurnSettlement(subagent.task.status)) { + return true; } } return false; @@ -6989,10 +6943,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( candidateLifecycle, (current): readonly [Effect.Effect, CandidateLifecycle] => { if (current._tag === "Committed") { - return [ - handleRuntimeTerminationAtGeneration(replacementGeneration, error), - current, - ]; + return [handleRuntimeTerminationAtGeneration(replacementGeneration), current]; } return [ Effect.void, @@ -8092,18 +8043,6 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( ) { const restartRequired = yield* Ref.get(runtimeRestartRequired); if (!restartRequired) return false; - yield* runtimeCallbackPermit.withPermit( - Effect.gen(function* () { - yield* quarantineNativeTransportAtGeneration( - yield* Ref.get(runtimeCallbackGeneration), - ); - yield* advanceRuntimeCallbackGeneration; - const carryover = yield* Ref.getAndSet(carryoverSubagents, null); - yield* quarantineStoppedRun(); - yield* terminalizeCarryoverSubagents(carryover); - yield* Ref.set(endedBackgroundTaskIds, new Set()); - }), - ); yield* restartAcpRuntime(threadId); yield* Ref.set(runtimeRestartRequired, false); yield* Ref.set(activeSessionId, null); @@ -8524,16 +8463,6 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( !context.interrupted && hasDeferredBackgroundWork(context) ) { - // A session-scoped consumer can hold a successful result past runtime exit. - const termination = yield* Ref.get(runtimeTermination); - if (termination?.generation === promptGeneration) { - yield* failTurnAfterRuntimeTermination( - context, - promptGeneration, - termination.error, - ); - return; - } yield* closeTextStreams(context); yield* ( options.testHooks?.afterPromptSettledWithBackgroundWork?.() ?? Effect.void From 946ba01ea15df8e348b154083fd0b0a27e0d09a1 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 11 Nov 2026 07:46:33 +0000 Subject: [PATCH 13/13] fix(antigravity): commands still running after end_turn stay stoppable When Antigravity answered end_turn while a command it started was still running, the turn completed and the command item was marked completed, so Stop had no running work to target and the command kept running. Antigravity's started execute tools are now tracked as background tasks without deferring the turn. The turn still completes at end_turn and follow-ups run at once, while the command item stays running. Stop routes to that run as pending background work, and the interrupt tears down the agent's process group, ending the command. AI-assisted; verified with the tests and tools listed above. Co-Authored-By: Claude Opus 5.5 --- apps/server/scripts/acp-mock-agent.ts | 7 +++-- .../Adapters/AntigravityAdapterV2.test.ts | 26 ++++++++++++++++--- .../Adapters/AntigravityAdapterV2.ts | 6 +++++ 3 files changed, 34 insertions(+), 5 deletions(-) diff --git a/apps/server/scripts/acp-mock-agent.ts b/apps/server/scripts/acp-mock-agent.ts index 5d61073a6885a..d53d0eaccc2f7 100644 --- a/apps/server/scripts/acp-mock-agent.ts +++ b/apps/server/scripts/acp-mock-agent.ts @@ -57,7 +57,8 @@ const emitTrackedToolCallKindThenHang = process.env.T3_ACP_EMIT_TRACKED_TOOL_CAL const clientFsProbePath = process.env.T3_ACP_CLIENT_FS_PROBE_PATH; const clientFsProbeLogPath = process.env.T3_ACP_CLIENT_FS_PROBE_LOG_PATH; const encodeJson = Schema.encodeSync(Schema.fromJsonString(Schema.Unknown)); -const emitRunningCommandThenHang = process.env.T3_ACP_EMIT_RUNNING_COMMAND_THEN_HANG === "1"; +// "hang" keeps the prompt running after the command starts; "end_turn" answers while it runs. +const emitRunningCommand = process.env.T3_ACP_EMIT_RUNNING_COMMAND; const runningCommandPidPath = process.env.T3_ACP_RUNNING_COMMAND_PID_PATH; const agentPidPath = process.env.T3_ACP_AGENT_PID_PATH; const emitEmptySuccessfulBash = process.env.T3_ACP_EMIT_EMPTY_SUCCESSFUL_BASH === "1"; @@ -1355,7 +1356,7 @@ const program = Effect.gen(function* () { return yield* AcpError.AcpRequestError.internalError("Mock prompt failure"); } - if (emitRunningCommandThenHang) { + if (emitRunningCommand !== undefined) { // The command is a long-lived Node child so the scenario runs on every platform. const command = NodeChildProcess.spawn( process.execPath, @@ -1383,6 +1384,8 @@ const program = Effect.gen(function* () { rawInput: { command: "npm run dev" }, }, }); + if (emitRunningCommand === "end_turn") + return yield* finishPrompt(requestedSessionId, "end_turn"); return yield* Effect.never; } diff --git a/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.test.ts index a5be9f7126040..744c72668a8f7 100644 --- a/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.test.ts @@ -1255,7 +1255,7 @@ describe("AntigravityAdapterV2 commands started during a turn", () => { } }; - const openMockSession = (label: string) => + const openMockSession = (label: string, runningCommand: "hang" | "end_turn") => Effect.gen(function* () { const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; const fileSystem = yield* FileSystem.FileSystem; @@ -1295,7 +1295,7 @@ describe("AntigravityAdapterV2 commands started during a turn", () => { T3_ACP_ANTIGRAVITY: "1", T3_ACP_RUNNING_COMMAND_PID_PATH: commandPidPath, T3_ACP_AGENT_PID_PATH: agentPidPath, - T3_ACP_EMIT_RUNNING_COMMAND_THEN_HANG: "1", + T3_ACP_EMIT_RUNNING_COMMAND: runningCommand, }, }, }).pipe(Effect.provideService(SecureRandom.SecureRandom, crypto)), @@ -1408,11 +1408,31 @@ describe("AntigravityAdapterV2 commands started during a turn", () => { }; }); + it.effect("keeps a command running past end_turn reachable by Stop", () => + Effect.gen(function* () { + const mock = yield* openMockSession("command-outlives-turn", "end_turn"); + yield* mock.startTurn(1); + assert.equal(yield* mock.waitForTerminal, "completed"); + const commandPid = yield* mock.readCommandPid; + assert.isTrue(pidIsRunning(commandPid)); + const command = mock.commandItem("running-command"); + assert.equal(command?.status, "running"); + if (command?.providerTurnId == null) return yield* Effect.die("missing command item"); + + yield* mock.session.interruptTurn({ + providerThread: mock.providerThread, + providerTurnId: command.providerTurnId, + requestRuntimeRestart: true, + }); + assert.isFalse(pidIsRunning(commandPid), "Stop ends the command"); + }).pipe(Effect.provide(layerSession), Effect.scoped), + ); + it.effect.each(["Stop", "fallback restart"])( "lets %s end the running command and respawns the agent for the next turn", (action) => Effect.gen(function* () { - const mock = yield* openMockSession("command-stopped"); + const mock = yield* openMockSession("command-stopped", "hang"); yield* mock.startTurn(1); yield* mock.waitForCommand; const commandPid = yield* mock.readCommandPid; diff --git a/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.ts index 24cb141aaca3c..9d927e168634e 100644 --- a/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.ts @@ -240,6 +240,12 @@ export function makeAntigravityAcpAdapterFlavor( subagentsIdleOnTurnCompletion: true, restartRuntimeOnEveryInterrupt: true, terminateRuntimeProcessGroupOnInterrupt: true, + // A started command can outlive end_turn. Its item stays running after the turn + // completes so Stop can still reach it; the turn itself does not wait for it. + extractBackgroundTaskId: (toolCall) => + toolCall.kind === "execute" && toolCall.status !== undefined && toolCall.status !== "pending" + ? toolCall.toolCallId + : undefined, ...(options.onSessionEvent === undefined ? {} : { onSessionEvent: options.onSessionEvent }), applyModelSelection: Effect.fn("AntigravityAdapterV2.applyModelSelection")(function* ({ runtime,