diff --git a/apps/server/scripts/acp-mock-agent.ts b/apps/server/scripts/acp-mock-agent.ts index 44011bc7f2d4b..d53d0eaccc2f7 100644 --- a/apps/server/scripts/acp-mock-agent.ts +++ b/apps/server/scripts/acp-mock-agent.ts @@ -1,5 +1,6 @@ #!/usr/bin/env node // @effect-diagnostics nodeBuiltinImport:off +import * as NodeChildProcess from "node:child_process"; import * as NodeFS from "node:fs"; import * as Effect from "effect/Effect"; @@ -56,6 +57,10 @@ 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)); +// "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"; const emitEmptySuccessfulBashThenHang = process.env.T3_ACP_EMIT_EMPTY_SUCCESSFUL_BASH_THEN_HANG === "1"; @@ -1351,6 +1356,39 @@ const program = Effect.gen(function* () { return yield* AcpError.AcpRequestError.internalError("Mock prompt failure"); } + if (emitRunningCommand !== undefined) { + // 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)"], + { 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 (agentPidPath !== undefined) { + NodeFS.writeFileSync(agentPidPath, String(process.pid)); + } + yield* agent.client.sessionUpdate({ + sessionId: requestedSessionId, + update: { + sessionUpdate: "tool_call_update", + toolCallId: "running-command", + title: "run_command", + kind: "execute", + status: "in_progress", + rawInput: { command: "npm run dev" }, + }, + }); + if (emitRunningCommand === "end_turn") + return yield* finishPrompt(requestedSessionId, "end_turn"); + return yield* Effect.never; + } + if (emitTrackedToolCallKindThenHang !== undefined) { yield* agent.client.sessionUpdate({ sessionId: requestedSessionId, diff --git a/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.test.ts index 0945a282cee2c..744c72668a8f7 100644 --- a/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.test.ts @@ -22,6 +22,7 @@ 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 { ChildProcessSpawner } from "effect/process"; import type * as EffectAcpSchema from "effect-acp/compat"; @@ -1243,3 +1244,219 @@ describe("AntigravityAdapterV2 session system prompt", () => { }).pipe(Effect.provide(layerSession), Effect.scoped), ); }); + +describe("AntigravityAdapterV2 commands started during a turn", () => { + const pidIsRunning = (pid: number) => { + try { + process.kill(pid, 0); + return true; + } catch { + return false; + } + }; + + const openMockSession = (label: string, runningCommand: "hang" | "end_turn") => + Effect.gen(function* () { + const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const crypto = yield* SecureRandom.SecureRandom; + 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 agentPidPath = path.join(workspace, "agent.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 reports the command. + const readCommandPid = Effect.gen(function* () { + const pid = Number(yield* fileSystem.readFileString(commandPidPath)); + commandPids.push(pid); + return pid; + }); + const instanceId = ProviderInstanceId.make(`antigravity-${label}`); + const adapter = yield* makeAntigravityAdapterV2({ + instanceId, + selfInvocation: yield* resolveSelfInvocation(), + 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, + T3_ACP_AGENT_PID_PATH: agentPidPath, + T3_ACP_EMIT_RUNNING_COMMAND: runningCommand, + }, + }, + }).pipe(Effect.provideService(SecureRandom.SecureRandom, crypto)), + withProcess: (_stop, task) => task, + defaultModel: Effect.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 waitForCommand = waitFor( + (event) => + 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)), + ); + 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, + waitForCommand, + waitForTerminal, + readCommandPid, + readAgentPid: fileSystem.readFileString(agentPidPath).pipe(Effect.map(Number)), + commandItem, + }; + }); + + 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", "hang"); + yield* mock.startTurn(1); + yield* mock.waitForCommand; + const commandPid = yield* mock.readCommandPid; + assert.isTrue(pidIsRunning(commandPid)); + const agentPid = yield* mock.readAgentPid; + 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.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), + ); +}); diff --git a/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.ts index 1490140c1e6e2..9d927e168634e 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 SecureRandom from "effect/crypto/SecureRandom"; import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; @@ -139,6 +140,7 @@ export function makeAntigravityAcpAdapterFlavor( const makeRuntime = Effect.fn("AntigravityAdapterV2.makeRuntime")( function* (input: AcpAdapterV2RuntimeInput) { const scope = yield* Scope.fork(yield* Effect.scope); + const platform = yield* HostProcess.Platform; const runtime = yield* options .withProcess( Scope.close(scope, Exit.void), @@ -146,6 +148,9 @@ export function makeAntigravityAcpAdapterFlavor( ...input, clientFileReads: true, sessionSystemPromptAppend: PULL_REQUEST_LINKING_INSTRUCTIONS, + ownDetachedProcessGroup: true, + ownDescendantProcessGroups: platform === "linux", + processGroupPlatform: platform, additionalDirectories: [options.host.paths.attachmentsDir], }), ) @@ -233,6 +238,14 @@ export function makeAntigravityAcpAdapterFlavor( ), // Loading history replays tool calls; resume avoids replay. 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,