From 6c404d6a8a5eb2abe5d50281ad25dd3ea06d7ae7 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 3 Nov 2026 21:57:46 +0000 Subject: [PATCH 1/5] test(server): reproduce #30414, a replayed Antigravity checkpoint failure is resumed forever These tests reproduce #30414. A turn that ends with Antigravity's replayed "context summarization: checkpoint step N failed" error must mark the native thread broken, so the ACP adapter finalizes it as broken (provider thread status "error") and refuses to resume it, letting the next turn start a fresh native session with the thread's context. Co-Authored-By: Claude Opus 5.5 --- .../Adapters/AcpAdapterV2.test.ts | 89 +++++++++++++++++++ .../acp/AntigravityTurnOutcome.test.ts | 31 +++++++ 2 files changed, 120 insertions(+) diff --git a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts index a58a83bffa349..2e4db96635840 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts @@ -824,6 +824,95 @@ describe("AcpAdapterV2", () => { assert.equal(watchedSessions[0], watchedSessions[1]); }).pipe(Effect.provide(layerTest), Effect.scoped), ); + it.live("marks the thread broken when the outcome check reports unusable native history", () => + Effect.gen(function* () { + const path = yield* Path.Path; + const instanceId = ProviderInstanceId.make("acp-test-prompt-outcome-broken"); + const threadId = ThreadId.make("thread-acp-prompt-outcome-broken"); + const failure = { + class: "provider_error", + message: "Antigravity stopped this turn because of an error.", + } as const; + const adapter = yield* makeAcpAdapterV2({ + instanceId, + selfInvocation: yield* resolveSelfInvocation(), + flavor: { + driver: ACP_TEST_DRIVER, + capabilities: AcpProviderCapabilitiesV2, + watchPromptOutcome: () => + Effect.succeed(() => Effect.succeed({ failure, threadDisposition: "broken" as const })), + makeRuntime: makeMockRuntime({ + childProcessSpawner: yield* ChildProcessSpawner.ChildProcessSpawner, + mockAgentPath: yield* path.fromFileUrl( + new URL("../../../scripts/acp-mock-agent.ts", import.meta.url), + ), + wrapRuntime: (runtime) => ({ + ...runtime, + prompt: () => Effect.succeed({ stopReason: "end_turn" as const }), + }), + }), + }, + }); + const runtimePolicy = ProviderAdapter.ProviderAdapterV2RuntimePolicy.make({ + runtimeMode: "full-access", + interactionMode: "default", + cwd: process.cwd(), + }); + const modelSelection = { instanceId, model: "default" } as const; + const runtime = yield* adapter.openSession({ + threadId, + providerSessionId: ProviderSessionId.make("provider-session-acp-prompt-outcome-broken"), + modelSelection, + runtimePolicy, + }); + const providerThread = yield* runtime.ensureThread({ + threadId, + modelSelection, + runtimePolicy, + }); + yield* runtime.startTurn( + makeTurnInput({ + threadId, + providerThread, + instanceId, + runtimePolicy, + now: yield* DateTime.now, + ordinal: 1, + }), + ); + const events = Array.from( + yield* runtime.events.pipe( + Stream.takeUntil((event) => event.type === "turn.terminal"), + Stream.runCollect, + ), + ); + + const terminal = events.at(-1); + if (terminal?.type !== "turn.terminal" || terminal.status !== "failed") { + return assert.fail("expected a failed terminal"); + } + assert.equal(terminal.threadDisposition, "broken"); + assert.deepEqual(terminal.failure, failure); + const updated = events.findLast((event) => event.type === "provider_thread.updated"); + assert.equal( + updated?.type === "provider_thread.updated" ? updated.providerThread.status : null, + "error", + ); + // The turn-start service recovers in a fresh native session when resume refuses. + const resumed = yield* Effect.result( + runtime.resumeThread({ + threadId, + providerThread: { ...providerThread, status: "error" }, + modelSelection, + runtimePolicy, + }), + ); + assert.equal( + resumed._tag === "Failure" ? resumed.failure._tag : null, + "ProviderAdapterResumeThreadError", + ); + }).pipe(Effect.provide(layerTest), Effect.scoped), + ); it.live("retains returned usage when Stop finalizes before the prompt callback", () => Effect.gen(function* () { const path = yield* Path.Path; diff --git a/apps/server/src/provider/acp/AntigravityTurnOutcome.test.ts b/apps/server/src/provider/acp/AntigravityTurnOutcome.test.ts index d3ca431c50794..0c739897b1a0d 100644 --- a/apps/server/src/provider/acp/AntigravityTurnOutcome.test.ts +++ b/apps/server/src/provider/acp/AntigravityTurnOutcome.test.ts @@ -29,6 +29,10 @@ const QUOTA_ERROR = errorPayload( const INVALID_ARGUMENT_ERROR = errorPayload( "Error 400, Message: Request contains an invalid argument., Status: INVALID_ARGUMENT, Details: []", ); +// Context summarization replays a stored failed CHECKPOINT step on every later prompt. +const FAILED_CHECKPOINT_ERROR = errorPayload( + 'context summarization: checkpoint step 841 failed: error executing cascade step: CORTEX_STEP_TYPE_CHECKPOINT: doRequest: error sending request: Post "https://aiplatform.googleapis.com/v1beta1/projects/test-project/locations/global/publishers/google/models/gemini-4-argon:streamGenerateContent?alt=sse": auth: "invalid_grant" "reauth related error (invalid_rapt)"', +); // The model's reply quoting the agent's quota notice is an ordinary planner response. const QUOTED_NOTICE = Buffer.from( "Usage Limit Reached\n\nYou have reached your current quota for this period. Your limit will reset in 1 hour.", @@ -192,4 +196,31 @@ describe("watchAntigravityPromptOutcome", () => { }); }).pipe(Effect.provide(NodeServices.layer), Effect.scoped), ); + + it.effect("marks the native thread broken when a failed checkpoint step is replayed", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const profileDirectory = yield* fs.makeTempDirectoryScoped(); + const sessionId = "e6aba774-5472-4277-b6ee-27b2e05570a7"; + const conversations = path.join(profileDirectory, "antigravity-acp", "conversations"); + yield* fs.makeDirectory(conversations, { recursive: true }); + const check = yield* watchAntigravityPromptOutcome(profileDirectory, sessionId); + const database = new NodeSqlite.DatabaseSync(path.join(conversations, `${sessionId}.db`)); + try { + database.exec( + "CREATE TABLE steps (idx INTEGER PRIMARY KEY, step_type INTEGER, step_payload BLOB)", + ); + const insert = database.prepare("INSERT INTO steps VALUES (?, ?, ?)"); + insert.run(843, USER_INPUT, Buffer.from("Continue.")); + insert.run(844, ERROR_MESSAGE, FAILED_CHECKPOINT_ERROR); + } finally { + database.close(); + } + + const outcome = yield* check({ stopReason: "end_turn" }); + assert.equal(outcome?.threadDisposition, "broken"); + assert.equal(outcome?.failure.class, "provider_error"); + }).pipe(Effect.provide(NodeServices.layer), Effect.scoped), + ); }); From 2a128ff8e0736b0e2f56256a589e69905c733be6 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 3 Nov 2026 22:16:26 +0000 Subject: [PATCH 2/5] fix(server): a failed Antigravity checkpoint no longer bricks the thread Antigravity stores a failed context-summarization checkpoint step and replays it on every later prompt to that native session, so one transient auth failure left the thread failing forever: the outcome check reported a generic reusable failure and T3 resumed the same native session each turn. The outcome check now classifies "context summarization: checkpoint step N failed" as a broken native thread. The ACP adapter carries that disposition through finalizeTurn (terminal threadDisposition "broken", provider thread status "error") and resumeThread refuses an "error" thread, so the turn start recovers in a fresh native session with this thread's context handoff. The "error" status now survives a queued run's dispatch, a session-restarting selection change, and a provider sign-out command until that recovery replaces the row. Co-Authored-By: Claude Opus 5.5 --- .../Adapters/AcpAdapterV2.test.ts | 7 +- .../Orchestrator.control-reads.test.ts | 86 +++++++++++++++++++ .../src/orchestration-v2/Orchestrator.ts | 6 +- .../ProviderTurnStartService.test.ts | 30 ++++++- .../ProviderTurnStartService.ts | 7 +- .../acp/AntigravityTurnOutcome.test.ts | 45 ++++++---- .../provider/acp/AntigravityTurnOutcome.ts | 64 ++++++++++---- packages/provider-acp/src/server/adapter.ts | 33 +++++-- 8 files changed, 227 insertions(+), 51 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts index 2e4db96635840..a949b0d9ed40f 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts @@ -758,7 +758,10 @@ describe("AcpAdapterV2", () => { Effect.sync(() => { watchedSessions.push(sessionId); const failed = watchedSessions.length === 1; - return () => Effect.succeed(failed ? quota : undefined); + return () => + Effect.succeed( + failed ? { failure: quota, threadDisposition: "reusable" as const } : undefined, + ); }), makeRuntime: makeMockRuntime({ childProcessSpawner: yield* ChildProcessSpawner.ChildProcessSpawner, @@ -832,6 +835,8 @@ describe("AcpAdapterV2", () => { const failure = { class: "provider_error", message: "Antigravity stopped this turn because of an error.", + code: null, + retryable: null, } as const; const adapter = yield* makeAcpAdapterV2({ instanceId, diff --git a/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts b/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts index c806794732726..7104b2fbb552b 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts @@ -1430,6 +1430,92 @@ it.effect.each([false, true])("rejects thread.create for an existing id (deleted }).pipe(Effect.provide(layerTest)), ); +it.effect("keeps a broken native thread in error when a queued message starts", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const sink = yield* EventSink.EventSinkV2; + const threadId = ThreadId.make("thread:broken-native-thread"); + const send = (id: string, dispatchMode: "start_immediately" | "queue_after_active") => + orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make(`send:broken-native-thread:${id}`), + threadId, + messageId: MessageId.make(`message:broken-native-thread:${id}`), + text: id, + attachments: [], + dispatchMode: { type: dispatchMode }, + createdBy: "user", + creationSource: "web", + }); + yield* orchestrator.dispatch({ + type: "thread.create", + commandId: CommandId.make("create:broken-native-thread"), + threadId, + projectId: ProjectId.make("project:broken-native-thread"), + title: "Broken", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + createdBy: "user", + creationSource: "web", + }); + yield* send("first", "start_immediately"); + yield* send("queued", "queue_after_active"); + const { runs, providerThreads } = yield* orchestrator.getThreadRecords(threadId, [ + "runs", + "providerThreads", + ]); + const [failed, queued] = runs; + const [providerThread] = providerThreads; + const occurredAt = yield* DateTime.now; + // The first turn ended with its native history unusable. + yield* sink.write({ + events: [ + { + id: EventId.make("event:broken-native-thread:run"), + type: "run.updated", + threadId, + runId: failed!.id, + providerInstanceId: instanceId, + occurredAt, + payload: { ...failed!, status: "failed", completedAt: occurredAt }, + }, + { + id: EventId.make("event:broken-native-thread:provider-thread"), + type: "provider-thread.updated", + threadId, + providerInstanceId: instanceId, + occurredAt, + payload: { + ...providerThread!, + nativeThreadRef: { + driver: providerThread!.driver, + nativeId: "poisoned-native-history", + strength: "strong", + }, + status: "error", + }, + }, + ], + }); + + yield* orchestrator.dispatch({ + type: "queue.resume", + commandId: CommandId.make("resume:broken-native-thread"), + threadId, + }); + const next = yield* orchestrator.getThreadRecords(threadId, ["runs", "providerThreads"]); + const started = next.runs.find((run) => run.id === queued!.id); + assert.equal(started?.status, "starting"); + assert.equal( + next.providerThreads.find((thread) => thread.id === started?.providerThreadId)?.status, + "error", + ); + }).pipe(Effect.provide(layerTest)), +); + it.effect("replays a thread.create receipt without changing the thread", () => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 16256585dc526..897000388cdbc 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -1838,6 +1838,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio nativeThreadRef: activeProviderThread.nativeThreadRef, nativeConversationHeadRef: activeProviderThread.nativeConversationHeadRef, nativeMetadata: activeProviderThread.nativeMetadata, + status: activeProviderThread.status, } : queuedProviderThread; const targetCapabilities = yield* targetAdapter.getCapabilities().pipe( @@ -2021,7 +2022,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const providerThread: OrchestrationV2ProviderThread = { ...deliveryProviderThread, providerSessionId, - status: "not_loaded", + // A broken native history must reach the turn start, which replaces it. + status: deliveryProviderThread.status === "error" ? "error" : "not_loaded", firstRunOrdinal: queuedProviderThread.firstRunOrdinal ?? ordinal, lastRunOrdinal: ordinal, handoffIds: appendContextHandoffId( @@ -5168,7 +5170,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ...providerThread, providerInstanceId: input.modelSelection.instanceId, providerSessionId: nextProviderSessionId, - status: "not_loaded", + status: providerThread.status === "error" ? "error" : "not_loaded", updatedAt: now, }; restartSessionTransition = { diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts index f7f9dc25ddc92..80d14ced8cbaa 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts @@ -1747,11 +1747,32 @@ effectIt.effect("does not mistake a failed state read for a superseded run", () }), ); -effectIt.effect( - "signs out the existing native provider before opening the newly selected provider", - () => +effectIt.effect.each(["not_loaded", "error"] as const)( + "signs out the existing native provider before opening the newly selected provider with status %s", + (providerThreadStatus) => Effect.gen(function* () { const harness = makeLocalCommandHarness({ text: "/logout", previousNativeSession: true }); + const initial = harness.projection(); + const target = initial.providerThreads.find((thread) => thread.id === "new-provider-thread")!; + harness.setProjection({ + ...initial, + providerThreads: initial.providerThreads.map((thread) => + thread.id === target.id + ? { + ...thread, + status: providerThreadStatus, + nativeThreadRef: + providerThreadStatus === "error" + ? { + driver: thread.driver, + nativeId: "poisoned-native-history", + strength: "strong" as const, + } + : thread.nativeThreadRef, + } + : thread, + ), + }); yield* harness.start; yield* harness.start; @@ -1777,6 +1798,9 @@ effectIt.effect( ]); expect(projection.providerTurns).toEqual([]); expect(projection.checkpoints).toEqual([]); + expect(projection.providerThreads.find((thread) => thread.id === target.id)?.status).toBe( + providerThreadStatus, + ); }), ); diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index cd121c7115a2a..b6d5afffb85cd 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -627,7 +627,12 @@ export const layer: Layer.Layer< }, providerThreadUpdate: { ...providerThread, - status: providerThread.nativeThreadRef === null ? "not_loaded" : "idle", + status: + providerThread.status === "error" + ? "error" + : providerThread.nativeThreadRef === null + ? "not_loaded" + : "idle", updatedAt: now, }, }); diff --git a/apps/server/src/provider/acp/AntigravityTurnOutcome.test.ts b/apps/server/src/provider/acp/AntigravityTurnOutcome.test.ts index 0c739897b1a0d..a643a8493a208 100644 --- a/apps/server/src/provider/acp/AntigravityTurnOutcome.test.ts +++ b/apps/server/src/provider/acp/AntigravityTurnOutcome.test.ts @@ -6,6 +6,7 @@ import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; import * as Path from "effect/Path"; import type * as EffectAcpSchema from "effect-acp/compat"; +import type { OrchestrationV2ProviderFailure } from "@t3tools/contracts"; import { watchAntigravityPromptOutcome } from "./AntigravityTurnOutcome.ts"; @@ -37,6 +38,8 @@ const FAILED_CHECKPOINT_ERROR = errorPayload( const QUOTED_NOTICE = Buffer.from( "Usage Limit Reached\n\nYou have reached your current quota for this period. Your limit will reset in 1 hour.", ); +const reusable = (failure: OrchestrationV2ProviderFailure) => + ({ failure, threadDisposition: "reusable" }) as const; describe("watchAntigravityPromptOutcome", () => { it.effect("fails only turns whose own steps end in an upstream error", () => @@ -84,13 +87,13 @@ describe("watchAntigravityPromptOutcome", () => { [USER_INPUT, input], [ERROR_MESSAGE, QUOTA_ERROR], ]), - { + reusable({ class: "usage_limit", message: "You have reached your Antigravity quota for this period.", code: "RESOURCE_EXHAUSTED", retryable: true, resetAt: "1970-01-01T01:00:00.000Z", - }, + }), ); // The earlier error step is outside this turn. assert.isUndefined( @@ -110,12 +113,12 @@ describe("watchAntigravityPromptOutcome", () => { [USER_INPUT, input], [ERROR_MESSAGE, INVALID_ARGUMENT_ERROR], ]), - { + reusable({ class: "provider_error", message: "Request contains an invalid argument.", code: "INVALID_ARGUMENT", retryable: false, - }, + }), ); // A completed tool keeps its usage, and the turn still fails. assert.equal( @@ -130,7 +133,7 @@ describe("watchAntigravityPromptOutcome", () => { stopReason: "end_turn", usage: { inputTokens: 40, outputTokens: 5, totalTokens: 45 }, }, - ))?.class, + ))?.failure.class, "usage_limit", ); assert.isUndefined( @@ -140,12 +143,15 @@ describe("watchAntigravityPromptOutcome", () => { ]), ); // The agent maps the SDK's QUOTA_EXHAUSTED stop reason to `refusal`. - assert.deepEqual(yield* turn([], { stopReason: "refusal" }), { - class: "usage_limit", - message: "You have reached your Antigravity quota for this period.", - code: "QUOTA_EXHAUSTED", - retryable: true, - }); + assert.deepEqual( + yield* turn([], { stopReason: "refusal" }), + reusable({ + class: "usage_limit", + message: "You have reached your Antigravity quota for this period.", + code: "QUOTA_EXHAUSTED", + retryable: true, + }), + ); }).pipe(Effect.provide(NodeServices.layer), Effect.scoped), ); @@ -181,19 +187,22 @@ describe("watchAntigravityPromptOutcome", () => { const linked = yield* watchAntigravityPromptOutcome(profileDirectory, own); append(peer, ERROR_MESSAGE, QUOTA_ERROR); assert.isUndefined(yield* linked({ stopReason: "end_turn" })); - assert.equal((yield* linked({ stopReason: "refusal" }))?.class, "usage_limit"); + assert.equal((yield* linked({ stopReason: "refusal" }))?.failure.class, "usage_limit"); // An unreadable log gives no evidence, but the agent's typed refusal still fails the turn. yield* fs.remove(path.join(conversations, `${own}.db`)); yield* fs.writeFileString(path.join(conversations, `${own}.db`), "not a database"); const corrupt = yield* watchAntigravityPromptOutcome(profileDirectory, own); assert.isUndefined(yield* corrupt({ stopReason: "end_turn" })); - assert.deepEqual(yield* corrupt({ stopReason: "refusal" }), { - class: "usage_limit", - message: "You have reached your Antigravity quota for this period.", - code: "QUOTA_EXHAUSTED", - retryable: true, - }); + assert.deepEqual( + yield* corrupt({ stopReason: "refusal" }), + reusable({ + class: "usage_limit", + message: "You have reached your Antigravity quota for this period.", + code: "QUOTA_EXHAUSTED", + retryable: true, + }), + ); }).pipe(Effect.provide(NodeServices.layer), Effect.scoped), ); diff --git a/apps/server/src/provider/acp/AntigravityTurnOutcome.ts b/apps/server/src/provider/acp/AntigravityTurnOutcome.ts index 874c28eeb1537..882d7a0ba6025 100644 --- a/apps/server/src/provider/acp/AntigravityTurnOutcome.ts +++ b/apps/server/src/provider/acp/AntigravityTurnOutcome.ts @@ -19,6 +19,9 @@ const MAX_ERROR_PAYLOAD_BYTES = 16_384; // The Go harness's rendering of a Google API error, as the agent's api_errors.py reads it. const API_ERROR = /Error (\d{3}), Message: (.+?), Status: ([A-Z_]+)/s; const QUOTA_EXHAUSTED = /\breason:QUOTA_EXHAUSTED\b/; +// The runtime stores a failed context-summarization step and replays it on every later +// prompt to that session, whatever first made it fail. +const FAILED_CHECKPOINT = /\bcontext summarization: checkpoint step \d+ failed\b/; const RESET_TIMESTAMP = /\bquotaResetTimeStamp:([^\s\]]+)/; const RESET_DELAY = /\bquotaResetDelay:((?:\d+(?:\.\d+)?(?:ns|us|µs|μs|ms|h|m|s))+)(?=[\s\]])/; const GO_DURATION_PART = /(\d+(?:\.\d+)?)(ns|us|µs|μs|ms|h|m|s)/g; @@ -89,7 +92,12 @@ function turnFailure( stopReason: EffectAcpSchema.StopReason, step: LastStep | undefined, now: DateTime.Utc, -): OrchestrationV2ProviderFailure | undefined { +): + | { + readonly failure: OrchestrationV2ProviderFailure; + readonly threadDisposition: "reusable" | "broken"; + } + | undefined { const errorText = step?.type === ERROR_STEP_TYPE ? new TextDecoder().decode(step.payload ?? new Uint8Array()) @@ -98,29 +106,49 @@ function turnFailure( if (errorText === undefined && stopReason !== "refusal") return undefined; const text = errorText ?? ""; const apiError = API_ERROR.exec(text); + const reusable = (failure: OrchestrationV2ProviderFailure) => + ({ failure, threadDisposition: "reusable" }) as const; + if (FAILED_CHECKPOINT.test(text)) { + return { + failure: makeProviderFailure({ + class: "provider_error", + message: + "Antigravity could not summarize this conversation, so its saved session can no longer continue. The next message starts a fresh Antigravity session with this thread's context.", + code: "CHECKPOINT_FAILED", + retryable: true, + }), + threadDisposition: "broken", + }; + } if (stopReason === "refusal" || QUOTA_EXHAUSTED.test(text)) { - return makeProviderFailure({ - class: "usage_limit", - message: "You have reached your Antigravity quota for this period.", - code: apiError?.[3] ?? "QUOTA_EXHAUSTED", - retryable: true, - resetAt: resetAtFrom(text, now) ?? null, - }); + return reusable( + makeProviderFailure({ + class: "usage_limit", + message: "You have reached your Antigravity quota for this period.", + code: apiError?.[3] ?? "QUOTA_EXHAUSTED", + retryable: true, + resetAt: resetAtFrom(text, now) ?? null, + }), + ); } if (apiError === null) { - return makeProviderFailure({ - class: "provider_error", - message: "Antigravity stopped this turn because of an error.", - }); + return reusable( + makeProviderFailure({ + class: "provider_error", + message: "Antigravity stopped this turn because of an error.", + }), + ); } const [, httpStatus, message, status] = apiError; const rateLimited = httpStatus === "429" || status === "RESOURCE_EXHAUSTED"; - return makeProviderFailure({ - class: rateLimited ? "usage_limit" : "provider_error", - message: message!.trim(), - code: status ?? null, - retryable: rateLimited || Number(httpStatus) >= 500, - }); + return reusable( + makeProviderFailure({ + class: rateLimited ? "usage_limit" : "provider_error", + message: message!.trim(), + code: status ?? null, + retryable: rateLimited || Number(httpStatus) >= 500, + }), + ); } const conversationSegments = ["antigravity-acp", "conversations"]; diff --git a/packages/provider-acp/src/server/adapter.ts b/packages/provider-acp/src/server/adapter.ts index 985f6c8ee9887..18a821fe17d24 100644 --- a/packages/provider-acp/src/server/adapter.ts +++ b/packages/provider-acp/src/server/adapter.ts @@ -271,6 +271,11 @@ export function acpPerTurnPromptUsage( }; } +interface AcpPromptOutcomeFailure { + readonly failure: OrchestrationV2ProviderFailure; + readonly threadDisposition: "reusable" | "broken"; +} + export interface AcpAdapterV2Flavor { readonly buildPrompt?: (input: { readonly text: string; @@ -285,14 +290,13 @@ export interface AcpAdapterV2Flavor { /** * Runs before `session/prompt`. The returned check sees a prompt the agent ended * without an error or cancel, and reports a failure its stop reason hid. A reported - * failure restarts the runtime before the next turn. + * failure restarts the runtime before the next turn. A `broken` thread's native + * history is unusable, so it is never resumed and the next turn starts fresh. */ readonly watchPromptOutcome?: ( sessionId: string, ) => Effect.Effect< - ( - result: EffectAcpSchema.PromptResponse, - ) => Effect.Effect + (result: EffectAcpSchema.PromptResponse) => Effect.Effect >; readonly requiresSignIn?: (cause: unknown) => boolean; /** @@ -7470,10 +7474,15 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( context: ActiveAcpTurn, status: "completed" | "interrupted" | "failed" | "cancelled", failure?: OrchestrationV2ProviderFailure, - options?: { readonly childTurnsContinue?: boolean }, + options?: { + readonly childTurnsContinue?: boolean; + readonly threadDisposition?: "reusable" | "broken"; + }, ): Effect.fn.Return { if (context.finalized) return; const settledStatus = context.interrupted ? "interrupted" : status; + const threadDisposition = + settledStatus === "failed" ? (options?.threadDisposition ?? "reusable") : "reusable"; context.finalizedStatus = settledStatus; context.finalized = true; const directStopQuarantine = yield* Ref.get(stoppedRunQuarantine); @@ -7552,7 +7561,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( const updatedProviderThread: OrchestrationV2ProviderThread = { ...context.input.providerThread, providerSessionId: input.providerSessionId, - status: "active", + status: threadDisposition === "broken" ? "error" : "active", lastRunOrdinal: context.input.runOrdinal, firstRunOrdinal: context.input.providerThread.firstRunOrdinal ?? context.input.runOrdinal, @@ -7596,7 +7605,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( retry: context.providerRetry.retry, retryStartedAt: context.providerRetry.startedAt, }), - threadDisposition: "reusable", + threadDisposition, } : { type: "turn.terminal", @@ -8190,7 +8199,8 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( yield* Ref.set(runtimeRestartRequired, true); } const failure = - hiddenFailure ?? (idleError ? idleErrorFailure(result.error) : undefined); + hiddenFailure?.failure ?? + (idleError ? idleErrorFailure(result.error) : undefined); const status = failure !== undefined ? "failed" @@ -8213,6 +8223,9 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( // background tools and subagents keep reporting to this run. yield* finalizeTurn(context, "failed", failure, { childTurnsContinue: hasSurvivingBackgroundWork(context), + ...(hiddenFailure === undefined + ? {} + : { threadDisposition: hiddenFailure.threadDisposition }), }); return; } @@ -8477,6 +8490,10 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( readonly modelSelection?: ModelSelection; readonly runtimePolicy?: ProviderAdapter.ProviderAdapterV2RuntimePolicy; }) { + if (threadInput.providerThread.status === "error") { + // A broken turn left this native history unusable; the caller recovers in a fresh session. + return yield* Effect.fail("The native session history is unusable."); + } return yield* runtimeTransitionPermit.withPermit( Effect.gen(function* () { yield* awaitRuntimeTeardown(); From 7ae4c911210750f266656c0dc556eed3d4ddfbda Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 3 Nov 2026 22:28:15 +0000 Subject: [PATCH 3/5] fix(server): a failed Antigravity checkpoint on quota keeps its reset time Co-Authored-By: Claude Opus 5.5 --- .../acp/AntigravityTurnOutcome.test.ts | 23 +++++ .../provider/acp/AntigravityTurnOutcome.ts | 98 ++++++++++--------- 2 files changed, 76 insertions(+), 45 deletions(-) diff --git a/apps/server/src/provider/acp/AntigravityTurnOutcome.test.ts b/apps/server/src/provider/acp/AntigravityTurnOutcome.test.ts index a643a8493a208..298110639aaa8 100644 --- a/apps/server/src/provider/acp/AntigravityTurnOutcome.test.ts +++ b/apps/server/src/provider/acp/AntigravityTurnOutcome.test.ts @@ -230,6 +230,29 @@ describe("watchAntigravityPromptOutcome", () => { const outcome = yield* check({ stopReason: "end_turn" }); assert.equal(outcome?.threadDisposition, "broken"); assert.equal(outcome?.failure.class, "provider_error"); + + // A checkpoint that failed on quota still reports the quota and its reset time. + const quotaCheck = yield* watchAntigravityPromptOutcome(profileDirectory, sessionId); + const quotaDatabase = new NodeSqlite.DatabaseSync( + path.join(conversations, `${sessionId}.db`), + ); + try { + const insert = quotaDatabase.prepare("INSERT INTO steps VALUES (?, ?, ?)"); + insert.run(845, USER_INPUT, Buffer.from("Continue.")); + insert.run( + 846, + ERROR_MESSAGE, + errorPayload( + "context summarization: checkpoint step 841 failed: Error 429, Message: Quota exhausted for test model., Status: RESOURCE_EXHAUSTED, Details: [map[@type:type.googleapis.com/google.rpc.ErrorInfo domain:googleapis.com metadata:map[quotaResetDelay:3600s] reason:QUOTA_EXHAUSTED]]", + ), + ); + } finally { + quotaDatabase.close(); + } + const quotaOutcome = yield* quotaCheck({ stopReason: "end_turn" }); + assert.equal(quotaOutcome?.threadDisposition, "broken"); + assert.equal(quotaOutcome?.failure.class, "usage_limit"); + assert.notEqual(quotaOutcome?.failure.resetAt ?? null, null); }).pipe(Effect.provide(NodeServices.layer), Effect.scoped), ); }); diff --git a/apps/server/src/provider/acp/AntigravityTurnOutcome.ts b/apps/server/src/provider/acp/AntigravityTurnOutcome.ts index 882d7a0ba6025..90087a6c48e3a 100644 --- a/apps/server/src/provider/acp/AntigravityTurnOutcome.ts +++ b/apps/server/src/provider/acp/AntigravityTurnOutcome.ts @@ -92,12 +92,7 @@ function turnFailure( stopReason: EffectAcpSchema.StopReason, step: LastStep | undefined, now: DateTime.Utc, -): - | { - readonly failure: OrchestrationV2ProviderFailure; - readonly threadDisposition: "reusable" | "broken"; - } - | undefined { +): OrchestrationV2ProviderFailure | undefined { const errorText = step?.type === ERROR_STEP_TYPE ? new TextDecoder().decode(step.payload ?? new Uint8Array()) @@ -106,49 +101,62 @@ function turnFailure( if (errorText === undefined && stopReason !== "refusal") return undefined; const text = errorText ?? ""; const apiError = API_ERROR.exec(text); - const reusable = (failure: OrchestrationV2ProviderFailure) => - ({ failure, threadDisposition: "reusable" }) as const; - if (FAILED_CHECKPOINT.test(text)) { - return { - failure: makeProviderFailure({ - class: "provider_error", - message: - "Antigravity could not summarize this conversation, so its saved session can no longer continue. The next message starts a fresh Antigravity session with this thread's context.", - code: "CHECKPOINT_FAILED", - retryable: true, - }), - threadDisposition: "broken", - }; - } if (stopReason === "refusal" || QUOTA_EXHAUSTED.test(text)) { - return reusable( - makeProviderFailure({ - class: "usage_limit", - message: "You have reached your Antigravity quota for this period.", - code: apiError?.[3] ?? "QUOTA_EXHAUSTED", - retryable: true, - resetAt: resetAtFrom(text, now) ?? null, - }), - ); + return makeProviderFailure({ + class: "usage_limit", + message: "You have reached your Antigravity quota for this period.", + code: apiError?.[3] ?? "QUOTA_EXHAUSTED", + retryable: true, + resetAt: resetAtFrom(text, now) ?? null, + }); } if (apiError === null) { - return reusable( - makeProviderFailure({ - class: "provider_error", - message: "Antigravity stopped this turn because of an error.", - }), - ); + return makeProviderFailure({ + class: "provider_error", + message: "Antigravity stopped this turn because of an error.", + }); } const [, httpStatus, message, status] = apiError; const rateLimited = httpStatus === "429" || status === "RESOURCE_EXHAUSTED"; - return reusable( - makeProviderFailure({ - class: rateLimited ? "usage_limit" : "provider_error", - message: message!.trim(), - code: status ?? null, - retryable: rateLimited || Number(httpStatus) >= 500, - }), - ); + return makeProviderFailure({ + class: rateLimited ? "usage_limit" : "provider_error", + message: message!.trim(), + code: status ?? null, + retryable: rateLimited || Number(httpStatus) >= 500, + }); +} + +function turnOutcome( + stopReason: EffectAcpSchema.StopReason, + step: LastStep | undefined, + now: DateTime.Utc, +): + | { + readonly failure: OrchestrationV2ProviderFailure; + readonly threadDisposition: "reusable" | "broken"; + } + | undefined { + const failure = turnFailure(stopReason, step, now); + if (failure === undefined) return undefined; + const text = + step?.type === ERROR_STEP_TYPE + ? new TextDecoder().decode(step.payload ?? new Uint8Array()) + : ""; + if (!FAILED_CHECKPOINT.test(text)) return { failure, threadDisposition: "reusable" }; + return { + // A quota failure keeps its reset time so usage-limit recovery can resume the thread. + failure: + failure.class === "usage_limit" + ? failure + : makeProviderFailure({ + class: "provider_error", + message: + "Antigravity could not summarize this conversation, so its saved session can no longer continue. The next message starts a fresh Antigravity session with this thread's context.", + code: "CHECKPOINT_FAILED", + retryable: true, + }), + threadDisposition: "broken", + }; } const conversationSegments = ["antigravity-acp", "conversations"]; @@ -164,7 +172,7 @@ export const watchAntigravityPromptOutcome = Effect.fn("AntigravityTurnOutcome.w sessionId: string, ) { const typedOnly = (result: EffectAcpSchema.PromptResponse) => - DateTime.now.pipe(Effect.map((now) => turnFailure(result.stopReason, undefined, now))); + DateTime.now.pipe(Effect.map((now) => turnOutcome(result.stopReason, undefined, now))); if (!NATIVE_SESSION_ID.test(sessionId)) return typedOnly; const segments = [...conversationSegments, `${sessionId}.db`]; const before = yield* Effect.tryPromise(async () => { @@ -217,6 +225,6 @@ export const watchAntigravityPromptOutcome = Effect.fn("AntigravityTurnOutcome.w Effect.orElseSucceed(() => undefined), ) : undefined; - return turnFailure(result.stopReason, step, yield* DateTime.now); + return turnOutcome(result.stopReason, step, yield* DateTime.now); }); }); From 8fcf3dcf3ac6841ad2bbcc44585dce67a69b95e9 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 3 Nov 2026 22:42:13 +0000 Subject: [PATCH 4/5] test(server): main's retry-delay case reads the new outcome shape Co-Authored-By: Claude Opus 5.5 --- apps/server/src/provider/acp/AntigravityTurnOutcome.test.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/apps/server/src/provider/acp/AntigravityTurnOutcome.test.ts b/apps/server/src/provider/acp/AntigravityTurnOutcome.test.ts index 72f98539cb12f..244b54ac13fe1 100644 --- a/apps/server/src/provider/acp/AntigravityTurnOutcome.test.ts +++ b/apps/server/src/provider/acp/AntigravityTurnOutcome.test.ts @@ -116,13 +116,13 @@ describe("watchAntigravityPromptOutcome", () => { [USER_INPUT, input], [ERROR_MESSAGE, RATE_LIMIT_ERROR], ]), - { + reusable({ class: "usage_limit", message: "Resource has been exhausted (e.g. check quota).", code: "RESOURCE_EXHAUSTED", retryable: true, resetAt: "1970-01-01T00:00:42.000Z", - }, + }), ); assert.deepEqual( yield* turn([ From b6aceefed2a0909d8cdabb285e554de141ea75e1 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 3 Nov 2026 23:00:48 +0000 Subject: [PATCH 5/5] fix(server): a provider switch keeps a broken Antigravity thread in error Co-Authored-By: Claude Opus 5.5 --- apps/server/src/orchestration-v2/Orchestrator.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 897000388cdbc..6bc2aa2057834 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -7031,7 +7031,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio : null; const providerThread: OrchestrationV2ProviderThread = { ...ensuredProviderThread, - status: "active", + status: ensuredProviderThread.status === "error" ? "error" : "active", firstRunOrdinal: ensuredProviderThread.firstRunOrdinal ?? ordinal, lastRunOrdinal: ordinal, handoffIds: [