diff --git a/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.test.ts index f69e3696fc45c..5cf359451ef29 100644 --- a/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.test.ts @@ -707,3 +707,202 @@ describe("AntigravityAdapterV2 html_render pages", () => { }).pipe(Effect.provide(layerSession), Effect.scoped), ); }); + +describe("AntigravityAdapterV2 subagent reads", () => { + it.effect("completes a read T3 served when the agent later fails it with a step error", () => + 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-antigravity-reads-", + }); + const outside = yield* fileSystem.makeTempDirectoryScoped({ + prefix: "t3-antigravity-reads-outside-", + }); + const served = path.join(workspace, "served.ts"); + yield* fileSystem.writeFileString(served, "export {};"); + const denied = path.join(outside, "secret.txt"); + yield* fileSystem.writeFileString(denied, "secret"); + const unseen = path.join(workspace, "unseen.ts"); + const retried = path.join(workspace, "retried.ts"); + const stepError = 'Model produced invalid output. ("model output error")'; + type RuntimeService = AcpSessionRuntime.AcpSessionRuntime["Service"]; + let sessionUpdate: Parameters[0] | undefined; + let readTextFile: Parameters[0] | undefined; + const instanceId = ProviderInstanceId.make("antigravity-subagent-reads-test"); + const threadId = ThreadId.make("thread-antigravity-subagent-reads"); + 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" }, + }, + }).pipe( + Effect.provideService(Crypto.Crypto, crypto), + Effect.map((runtime): RuntimeService => ({ + ...runtime, + handleSessionUpdate: (handler) => + Effect.sync(() => { + sessionUpdate = handler; + }).pipe(Effect.andThen(runtime.handleSessionUpdate(handler))), + handleReadTextFile: (handler) => + Effect.sync(() => { + readTextFile = handler; + }).pipe(Effect.andThen(runtime.handleReadTextFile(handler))), + // agy_acp_server announces a subagent's client_view_file on the parent session, + // never completes it, then fails every open call with the next step error. + prompt: () => + Effect.gen(function* () { + if (sessionUpdate === undefined || readTextFile === undefined) { + return yield* Effect.die("the adapter must handle updates and reads"); + } + const send = sessionUpdate; + const handleRead = readTextFile; + const update = (update: EffectAcpSchema.SessionUpdate) => + send( + normalizeAntigravitySessionUpdate({ sessionId: "mock-session-1", update }), + ); + // The agent's own view_file never reaches T3, so it must not take a served read. + const reads = [ + ["call_native", served, "Running view_file"], + ["call_served", served, "Running client_view_file"], + ["call_missing", retried, "Running client_view_file"], + ["call_retried", retried, "Running client_view_file"], + ["call_denied", denied, "Running client_view_file"], + ["call_unseen", unseen, "Running client_view_file"], + ] as const; + for (const [toolCallId, target, title] of reads) { + yield* update({ + sessionUpdate: "tool_call", + toolCallId, + title, + kind: "read", + status: "in_progress", + rawInput: { absolute_path: target }, + locations: [{ path: target }], + }); + } + const serve = (target: string) => + handleRead( + { sessionId: "subagent-conversation", path: target }, + { requestId: `read-${target}`, method: "fs/read_text_file" }, + ).pipe(Effect.exit); + yield* serve(served); + yield* serve(retried); + yield* fileSystem.writeFileString(retried, "export {};").pipe(Effect.orDie); + yield* serve(retried); + yield* serve(denied); + for (const [toolCallId] of reads) { + yield* update({ + sessionUpdate: "tool_call_update", + toolCallId, + status: "failed", + rawOutput: stepError, + }); + } + return { stopReason: "end_turn" as const }; + }), + })), + ), + withProcess: (_stop, task) => task, + // @effect-diagnostics-next-line effectSucceedWithVoid:off - This optional result requires undefined, not the wider void type. + defaultModel: Effect.succeed(undefined), + }); + 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-subagent-reads"), + modelSelection, + runtimePolicy, + }); + const providerThread = yield* session.ensureThread({ + threadId, + modelSelection, + runtimePolicy, + }); + const now = yield* DateTime.now; + yield* session.startTurn({ + appThread: { + createdBy: "user", + creationSource: "web", + id: threadId, + projectId: ProjectId.make("project-antigravity-subagent-reads"), + title: "Antigravity subagent reads", + 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-subagent-reads"), + runOrdinal: 1, + providerTurnOrdinal: 1, + attemptId: RunAttemptId.make("attempt-antigravity-subagent-reads"), + rootNodeId: NodeId.make("node-antigravity-subagent-reads"), + providerThread, + message: { + createdBy: "user", + creationSource: "web", + messageId: MessageId.make("message-antigravity-subagent-reads"), + text: "review the codebase", + attachments: [], + }, + modelSelection, + runtimePolicy, + }); + const events = Array.from( + yield* session.events.pipe( + Stream.takeUntil((event) => event.type === "turn.terminal"), + Stream.runCollect, + ), + ); + const read = (nativeId: string) => { + const event = events.findLast( + (event) => + event.type === "turn_item.updated" && + event.turnItem.nativeItemRef?.nativeId === nativeId, + ); + return event?.type === "turn_item.updated" && event.turnItem.type === "dynamic_tool" + ? { status: event.turnItem.status, output: event.turnItem.output } + : undefined; + }; + + const failed = { status: "failed" as const, output: stepError }; + assert.deepEqual(read("call_served"), { status: "completed", output: undefined }); + assert.deepEqual(read("call_retried"), { status: "completed", output: undefined }); + assert.deepEqual(read("call_native"), failed); + assert.deepEqual(read("call_missing"), failed); + assert.deepEqual(read("call_denied"), failed); + assert.deepEqual(read("call_unseen"), failed); + }).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 36c58a29ecaee..fbdd5748adcfc 100644 --- a/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.ts @@ -29,6 +29,7 @@ import { antigravitySubagentOutput, classifyAntigravitySubagentToolCall, extractAntigravityUserInputQuestion, + makeAntigravityReadRepair, makeAntigravityUserInputResponse, normalizeAntigravityToolCall, } from "../../provider/acp/AntigravityProtocol.ts"; @@ -128,8 +129,22 @@ export function makeAntigravityAcpAdapterFlavor( }), ) .pipe(Effect.provideService(Scope.Scope, scope)); + // Repair keeps state across updates, so each runtime must have one session-update handler. + const reads = makeAntigravityReadRepair(); return { ...runtime, + handleReadTextFile: (handler) => + runtime.handleReadTextFile((request, context) => + handler(request, context).pipe( + Effect.onExit((exit) => + Effect.sync(() => reads.attempted(request.path, Exit.isSuccess(exit))), + ), + ), + ), + prompt: (payload, promptOptions) => + Effect.sync(reads.reset).pipe(Effect.andThen(runtime.prompt(payload, promptOptions))), + handleSessionUpdate: (handler) => + runtime.handleSessionUpdate((notification) => handler(reads.repair(notification))), start: () => runtime .start() @@ -138,7 +153,7 @@ export function makeAntigravityAcpAdapterFlavor( (started) => options.onSessionStarted?.(started, input.cwd) ?? Effect.void, ), ), - }; + } satisfies typeof runtime; }, Effect.provideService(Crypto.Crypto, options.crypto), Effect.mapError((cause): EffectAcpErrors.AcpError => diff --git a/apps/server/src/provider/acp/AntigravityProtocol.test.ts b/apps/server/src/provider/acp/AntigravityProtocol.test.ts index 66c83e0c48249..0c7937a28ee45 100644 --- a/apps/server/src/provider/acp/AntigravityProtocol.test.ts +++ b/apps/server/src/provider/acp/AntigravityProtocol.test.ts @@ -13,6 +13,7 @@ import { normalizeAntigravitySessionUpdate, normalizeAntigravityToolCall, sanitizeAntigravityToolPayload, + makeAntigravityReadRepair, } from "./AntigravityProtocol.ts"; import { mergeToolCallState, @@ -413,3 +414,81 @@ describe("Antigravity tool results", () => { expect(serialized).not.toContain("inline-data-url"); }); }); + +describe("Antigravity served reads", () => { + const flush = (reads: ReturnType, toolCallId: string) => + reads.repair({ + sessionId: "session-1", + update: { + sessionUpdate: "tool_call_update", + toolCallId, + status: "failed", + rawOutput: "Tool execution failed", + }, + }).update; + + it("credits a read T3 served before the agent announced it", () => { + const reads = makeAntigravityReadRepair(); + reads.attempted("/repo/a.ts", true); + reads.repair({ + sessionId: "session-1", + update: { + sessionUpdate: "tool_call", + toolCallId: "call_late", + title: "Running client_view_file", + kind: "read", + status: "in_progress", + locations: [{ path: "/repo/a.ts" }], + }, + }); + expect(flush(reads, "call_late")).toEqual({ + sessionUpdate: "tool_call_update", + toolCallId: "call_late", + status: "completed", + }); + }); + + it("forgets an earlier turn's unannounced served read", () => { + const reads = makeAntigravityReadRepair(); + reads.attempted("/repo/a.ts", true); + reads.reset(); + reads.repair({ + sessionId: "session-1", + update: { + sessionUpdate: "tool_call", + toolCallId: "call_refused", + title: "Running client_view_file", + kind: "read", + status: "in_progress", + locations: [{ path: "/repo/a.ts" }], + }, + }); + reads.attempted("/repo/a.ts", false); + expect(flush(reads, "call_refused")).toMatchObject({ status: "failed" }); + }); + + it("drops an earlier turn's unanswered read so a new turn's served read gets the credit", () => { + const reads = makeAntigravityReadRepair(); + const announce = (toolCallId: string) => + reads.repair({ + sessionId: "session-1", + update: { + sessionUpdate: "tool_call", + toolCallId, + title: "Running client_view_file", + kind: "read", + status: "in_progress", + locations: [{ path: "/repo/a.ts" }], + }, + }); + announce("call_abandoned"); + reads.reset(); + announce("call_current"); + reads.attempted("/repo/a.ts", true); + expect(flush(reads, "call_current")).toEqual({ + sessionUpdate: "tool_call_update", + toolCallId: "call_current", + status: "completed", + }); + }); +}); diff --git a/apps/server/src/provider/acp/AntigravityProtocol.ts b/apps/server/src/provider/acp/AntigravityProtocol.ts index ac65db6373add..1eeb0cdac31dd 100644 --- a/apps/server/src/provider/acp/AntigravityProtocol.ts +++ b/apps/server/src/provider/acp/AntigravityProtocol.ts @@ -216,6 +216,71 @@ export function sanitizeAntigravityToolPayload(payload: unknown): unknown { } const isJsonRecord = Schema.is(Schema.Record(Schema.String, Schema.Json)); +const ANTIGRAVITY_CLIENT_READ_TITLES = new Set([ + "Running client_view_file", + "Run client_view_file?", +]); + +function antigravityClientReadPath(update: { + readonly title?: string | null; + readonly locations?: ReadonlyArray<{ readonly path: string }> | null; + readonly rawInput?: unknown; +}): string | undefined { + if (!update.title || !ANTIGRAVITY_CLIENT_READ_TITLES.has(update.title)) return undefined; + const target = + update.locations?.[0]?.path ?? + (Predicate.isObject(update.rawInput) ? update.rawInput.absolute_path : undefined); + return typeof target === "string" && target ? target : undefined; +} + +/** + * The agent never completes a subagent's `client_view_file` call, then fails every open call + * with the next unrelated step error. Report reads T3 served as completed instead. + */ +export function makeAntigravityReadRepair() { + type Outcome = "served" | "refused"; + const openReads = new Map(); + // The agent runs a client tool in a background task, so its request can precede the announcement. + const earlyOutcomes: Array<{ readonly path: string; readonly outcome: Outcome }> = []; + return { + /** Earlier turns' rows are already settled, and their unanswered reads must not take credit. */ + reset: () => { + openReads.clear(); + earlyOutcomes.length = 0; + }, + /** Each `fs/read_text_file` request answers the oldest open read of its path, in order. */ + attempted: (path: string, served: boolean) => { + const outcome = served ? "served" : "refused"; + for (const read of openReads.values()) { + if (read.path !== path || read.outcome !== undefined) continue; + read.outcome = outcome; + return; + } + earlyOutcomes.push({ path, outcome }); + }, + repair: ( + notification: EffectAcpSchema.SessionNotification, + ): EffectAcpSchema.SessionNotification => { + const update = notification.update; + if (update.sessionUpdate !== "tool_call" && update.sessionUpdate !== "tool_call_update") { + return notification; + } + const path = antigravityClientReadPath(update); + if (path !== undefined && !openReads.has(update.toolCallId)) { + const early = earlyOutcomes.findIndex((entry) => entry.path === path); + const outcome = early === -1 ? undefined : earlyOutcomes.splice(early, 1)[0]?.outcome; + openReads.set(update.toolCallId, outcome === undefined ? { path } : { path, outcome }); + } + if (update.status !== "completed" && update.status !== "failed") return notification; + const read = openReads.get(update.toolCallId); + openReads.delete(update.toolCallId); + if (update.status === "completed" || read?.outcome !== "served") return notification; + const { rawOutput: _stepError, ...rest } = update; + return { ...notification, update: { ...rest, status: "completed" } }; + }, + }; +} + export function normalizeAntigravitySessionUpdate( notification: EffectAcpSchema.SessionNotification, ): EffectAcpSchema.SessionNotification {