From 98b3cb3008e2e8681181b7c4a4ae62e119b20f0f Mon Sep 17 00:00:00 2001 From: PR Batch Tester Date: Sun, 1 Nov 2026 20:59:39 +0000 Subject: [PATCH 01/16] fix(antigravity): start fresh after a lost thinking signature When Claude rejects Antigravity's saved history because a thinking block lost its signature, every later prompt to that native session fails the same way. The native error step is now read as a broken-thread failure: the turn fails with INVALID_THINKING_SIGNATURE and threadDisposition broken, which persists the provider thread as error. The ACP adapter refuses to resume an error row, so the next turn takes the existing resume-failure path into a fresh native session with a context handoff. Quota and other API errors keep the native session. Co-Authored-By: Claude Opus 5.5 --- .../Adapters/AcpAdapterV2.test.ts | 59 ++++++++++-- .../acp/AntigravityTurnOutcome.test.ts | 91 +++++++++++++++---- .../provider/acp/AntigravityTurnOutcome.ts | 67 ++++++++++---- packages/provider-acp/src/server/adapter.ts | 33 +++++-- 4 files changed, 195 insertions(+), 55 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts index a58a83bffa349..1da87fea517ec 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts @@ -745,6 +745,12 @@ describe("AcpAdapterV2", () => { retryable: true, resetAt: "2030-01-01T00:00:00.000Z", } as const; + const poisoned = { + class: "provider_error", + message: "Saved history is unusable.", + code: "INVALID_THINKING_SIGNATURE", + retryable: true, + } as const; // The runtime ordinal of each prompt, and the session each check watched. const promptRuntimes: Array = []; const watchedSessions: Array = []; @@ -757,8 +763,13 @@ describe("AcpAdapterV2", () => { watchPromptOutcome: (sessionId) => Effect.sync(() => { watchedSessions.push(sessionId); - const failed = watchedSessions.length === 1; - return () => Effect.succeed(failed ? quota : undefined); + const outcome = + watchedSessions.length === 1 + ? { failure: quota, threadDisposition: "reusable" as const } + : watchedSessions.length === 3 + ? { failure: poisoned, threadDisposition: "broken" as const } + : undefined; + return () => Effect.succeed(outcome); }), makeRuntime: makeMockRuntime({ childProcessSpawner: yield* ChildProcessSpawner.ChildProcessSpawner, @@ -805,23 +816,55 @@ describe("AcpAdapterV2", () => { ordinal, }), ); - const events = yield* runtime.events.pipe( - Stream.takeUntil((event) => event.type === "turn.terminal"), - Stream.runCollect, + return Array.from( + yield* runtime.events.pipe( + Stream.takeUntil((event) => event.type === "turn.terminal"), + Stream.runCollect, + ), ); - return events.at(-1); }); - const failed = yield* runTurn(1); + const failed = (yield* runTurn(1)).at(-1); if (failed?.type !== "turn.terminal" || failed.status !== "failed") { return assert.fail("expected a failed terminal"); } assert.deepEqual(failed.failure, quota); - const completed = yield* runTurn(2); + assert.equal(failed.threadDisposition, "reusable"); + const completed = (yield* runTurn(2)).at(-1); assert.equal(completed?.type === "turn.terminal" ? completed.status : null, "completed"); assert.deepEqual(promptRuntimes, [1, 2], "the next turn runs on a fresh runtime"); assert.equal(watchedSessions.length, 2); assert.equal(watchedSessions[0], watchedSessions[1]); + + // Poisoned native history marks the thread broken; its persisted error row is never resumed. + const poisonedEvents = yield* runTurn(3); + const broken = poisonedEvents.at(-1); + if (broken?.type !== "turn.terminal" || broken.status !== "failed") { + return assert.fail("expected a failed terminal"); + } + assert.deepEqual(broken.failure, poisoned); + assert.equal(broken.threadDisposition, "broken"); + const brokenThread = poisonedEvents.findLast( + (event) => event.type === "provider_thread.updated", + ); + assert.equal( + brokenThread?.type === "provider_thread.updated" + ? brokenThread.providerThread.status + : null, + "error", + ); + 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", () => diff --git a/apps/server/src/provider/acp/AntigravityTurnOutcome.test.ts b/apps/server/src/provider/acp/AntigravityTurnOutcome.test.ts index 09b8730bd6001..6665f6de1ff53 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"; @@ -32,10 +33,32 @@ const RATE_LIMIT_ERROR = errorPayload( const INVALID_ARGUMENT_ERROR = errorPayload( "Error 400, Message: Request contains an invalid argument., Status: INVALID_ARGUMENT, Details: []", ); +// Vertex Claude rejects a resumed history whose saved thinking block lost its signature. +const MISSING_SIGNATURE_ERROR = errorPayload( + 'Error 400, Message: {"type":"error","error":{"type":"invalid_request_error","message":"messages.1.content.0.thinking.signature: Field required"},"request_id":"req_vrtx_test"}, Status: INVALID_ARGUMENT, Details: []', +); +const INVALID_SIGNATURE_ERROR = errorPayload( + 'Error 400, Message: {"type":"error","error":{"type":"invalid_request_error","message":"messages.3.content.0: Invalid `signature` in `thinking` block"}}, Status: INVALID_ARGUMENT, Details: []', +); // 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.", ); +const QUOTED_SIGNATURE_ERROR = Buffer.from( + "The API said: messages.1.content.0.thinking.signature: Field required", +); +const reusable = (failure: OrchestrationV2ProviderFailure) => + ({ failure, threadDisposition: "reusable" }) as const; +const POISONED_HISTORY = { + failure: { + class: "provider_error", + message: + "Antigravity's saved conversation can no longer be sent to the model. The next message starts a fresh Antigravity session with this thread's context.", + code: "INVALID_THINKING_SIGNATURE", + retryable: true, + }, + threadDisposition: "broken", +} as const; describe("watchAntigravityPromptOutcome", () => { it.effect("fails only turns whose own steps end in an upstream error", () => @@ -83,13 +106,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( @@ -109,25 +132,47 @@ 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([ [USER_INPUT, input], [ERROR_MESSAGE, INVALID_ARGUMENT_ERROR], ]), - { + reusable({ class: "provider_error", message: "Request contains an invalid argument.", code: "INVALID_ARGUMENT", retryable: false, - }, + }), + ); + // A saved thinking block without a valid signature fails every later prompt. + assert.deepEqual( + yield* turn([ + [USER_INPUT, input], + [ERROR_MESSAGE, MISSING_SIGNATURE_ERROR], + ]), + POISONED_HISTORY, + ); + assert.deepEqual( + yield* turn([ + [USER_INPUT, input], + [ERROR_MESSAGE, INVALID_SIGNATURE_ERROR], + ]), + POISONED_HISTORY, + ); + // Reply text is never evidence, even when it quotes the error. + assert.isUndefined( + yield* turn([ + [USER_INPUT, input], + [PLANNER_RESPONSE, QUOTED_SIGNATURE_ERROR], + ]), ); // A completed tool keeps its usage, and the turn still fails. assert.equal( @@ -142,7 +187,7 @@ describe("watchAntigravityPromptOutcome", () => { stopReason: "end_turn", usage: { inputTokens: 40, outputTokens: 5, totalTokens: 45 }, }, - ))?.class, + ))?.failure.class, "usage_limit", ); assert.isUndefined( @@ -152,12 +197,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), ); @@ -193,19 +241,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 00de1e836296e..3474a07a781fe 100644 --- a/apps/server/src/provider/acp/AntigravityTurnOutcome.ts +++ b/apps/server/src/provider/acp/AntigravityTurnOutcome.ts @@ -19,6 +19,10 @@ 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/; +// Claude rejects a saved thinking block without a valid signature. It sits in the native +// history, so every later prompt to that session fails the same way. +const INVALID_THINKING_SIGNATURE = + /\bmessages\.\d+\.content\.\d+(?:\.thinking\.signature: Field required|: Invalid `signature` in `thinking` block)/; const RESET_TIMESTAMP = /\bquotaResetTimeStamp:([^\s\]]+)/; const RESET_DELAY = /\bquotaResetDelay:((?:\d+(?:\.\d+)?(?:ns|us|µs|μs|ms|h|m|s))+)(?=[\s\]])/; // google.rpc.RetryInfo, which plain rate limits carry instead of a quota reset. @@ -91,7 +95,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()) @@ -100,30 +109,50 @@ 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 (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 (INVALID_THINKING_SIGNATURE.test(text)) { + return { + failure: makeProviderFailure({ + class: "provider_error", + message: + "Antigravity's saved conversation can no longer be sent to the model. The next message starts a fresh Antigravity session with this thread's context.", + code: "INVALID_THINKING_SIGNATURE", + retryable: true, + }), + threadDisposition: "broken", + }; } 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, - resetAt: resetAtFrom(text, now) ?? null, - }); + return reusable( + makeProviderFailure({ + class: rateLimited ? "usage_limit" : "provider_error", + message: message!.trim(), + code: status ?? null, + retryable: rateLimited || Number(httpStatus) >= 500, + resetAt: resetAtFrom(text, now) ?? null, + }), + ); } 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 833ae1ce7241912562f64a5d6c7e34e6eecadda9 Mon Sep 17 00:00:00 2001 From: PR Batch Tester Date: Sun, 1 Nov 2026 21:06:44 +0000 Subject: [PATCH 02/16] fix(server): keep broken history invalid after sign-out --- .../ProviderTurnStartService.test.ts | 30 +++++++++++++++++-- .../ProviderTurnStartService.ts | 7 ++++- 2 files changed, 33 insertions(+), 4 deletions(-) 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, }, }); From af463680f8cc6be2e455276df68489d442c3eeb6 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Tue, 3 Nov 2026 23:09:46 +0000 Subject: [PATCH 03/16] fix(antigravity): preserve broken history across dispatch paths --- .../src/orchestration-v2/Orchestrator.ts | 4 +- .../ProviderSwitch.integration.test.ts | 79 +++++++++++++++---- 2 files changed, 66 insertions(+), 17 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 16256585dc526..cc2465129a69f 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -2021,7 +2021,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const providerThread: OrchestrationV2ProviderThread = { ...deliveryProviderThread, providerSessionId, - status: "not_loaded", + status: deliveryProviderThread.status === "error" ? "error" : "not_loaded", firstRunOrdinal: queuedProviderThread.firstRunOrdinal ?? ordinal, lastRunOrdinal: ordinal, handoffIds: appendContextHandoffId( @@ -7029,7 +7029,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: [ diff --git a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts index 2570c28e36b77..88b050ea0fa17 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts @@ -113,6 +113,7 @@ function makeTestAdapter(input: { >; readonly failure?: ReturnType; readonly failedRunOrdinals?: ReadonlySet; + readonly brokenRunOrdinals?: ReadonlySet; readonly interruptedRunOrdinals?: ReadonlySet; readonly holdRunOrdinal?: number; readonly pendingBackgroundWork?: Ref.Ref; @@ -212,6 +213,7 @@ function makeTestAdapter(input: { resumeThread: ({ providerThread }) => Effect.gen(function* () { if ( + (input.brokenRunOrdinals !== undefined && providerThread.status === "error") || input.failResume || (input.failResumeOnce !== undefined && (yield* Ref.getAndSet(input.failResumeOnce, false))) @@ -367,7 +369,9 @@ function makeTestAdapter(input: { }), } : { status: terminalStatus, failure: null }), - threadDisposition: "reusable", + threadDisposition: input.brokenRunOrdinals?.has(turnInput.runOrdinal) + ? "broken" + : "reusable", }, ]; const reportedUsage = input.tokenUsageByRunOrdinal?.[turnInput.runOrdinal]; @@ -383,6 +387,17 @@ function makeTestAdapter(input: { }, }); } + if (input.brokenRunOrdinals?.has(turnInput.runOrdinal)) { + yield* PubSub.publish(events, { + type: "provider_thread.updated", + driver: input.driver, + providerThread: { + ...turnInput.providerThread, + status: "error", + updatedAt: eventTime, + }, + }); + } for (const event of providerEvents) { yield* PubSub.publish(events, event); } @@ -1583,19 +1598,20 @@ describe("orchestration v2 provider switching", () => { it.live.each( ( [ - ["failed", false, false, false], + ["failed", false, false, false, false], + ["failed", false, false, false, true], ["failed", true, true, true], ["interrupted", false, true, false], ["interrupted", true, false, true], ] as const ).map( - ([status, queued, returning, native]) => + ([status, queued, returning, native, broken]) => [ - `${status} context ${queued ? "through the queue" : "immediately"} to ${returning ? "a returning" : "a new"} provider via ${native ? "native history" : "text"}`, - { status, queued, returning, native }, + `${status}${broken ? " broken" : ""} context ${queued ? "through the queue" : "immediately"} to ${returning ? "a returning" : "a new"} provider via ${native ? "native history" : "text"}`, + { status, queued, returning, native, broken }, ] as const, ), - )("hands off %s", ([, { status, queued, returning, native }]) => + )("hands off %s", ([, { status, queued, returning, native, broken }]) => Effect.scoped( Effect.gen(function* () { const cwd = yield* checkpointWorkspace(`handoff-${status}-${queued}-${returning}`); @@ -1606,6 +1622,7 @@ describe("orchestration v2 provider switching", () => { const sourceOrdinal = returning ? 2 : 1; const sourceSelection = returning ? CLAUDE_MODEL_SELECTION : CODEX_MODEL_SELECTION; const targetSelection = returning ? CODEX_MODEL_SELECTION : CLAUDE_MODEL_SELECTION; + const generation = yield* Ref.make(0); const originalPrompt = "Keep the release marker violet and preserve the existing API."; const partialResponse = "I checked the API and found the release configuration."; const layerRegistry = ProviderAdapterRegistry.layerFromAdapters( @@ -1628,6 +1645,12 @@ describe("orchestration v2 provider switching", () => { ...(modelSelection.instanceId === sourceSelection.instanceId ? { failedRunOrdinals: new Set(status === "failed" ? [sourceOrdinal] : []), + ...(broken + ? { + brokenRunOrdinals: new Set([sourceOrdinal]), + nativeThreadGeneration: generation, + } + : {}), interruptedRunOrdinals: new Set( status === "interrupted" ? [sourceOrdinal] : [], ), @@ -1742,8 +1765,17 @@ describe("orchestration v2 provider switching", () => { yield* dispatch("back", "Finish the remaining work", sourceSelection); yield* waitForRun(sourceOrdinal + 2, "completed"); const back = (yield* Ref.get(capturedTurns)).at(-1)!; - assert.notInclude(back.text, originalPrompt); - assert.notInclude(back.text, partialResponse); + if (broken) { + const source = (yield* Ref.get(capturedTurns)).find( + (turn) => turn.modelSelection.instanceId === sourceSelection.instanceId, + )!; + assert.notEqual(back.nativeThreadId, source.nativeThreadId); + assert.include(back.text, originalPrompt); + assert.include(back.text, partialResponse); + } else { + assert.notInclude(back.text, originalPrompt); + assert.notInclude(back.text, partialResponse); + } }).pipe( Effect.provide( ProviderReplayHarness.layerWithRegistry( @@ -2666,6 +2698,7 @@ describe("orchestration v2 provider switching", () => { it.live.each([ ["automatic", "Claude"], ["queue.resume", "compatible Codex"], + ["queue.resume", "broken current provider"], ] as const)( "delivers a queued PR update on the current provider through %s (%s)", ([delivery, target]) => @@ -2676,10 +2709,17 @@ describe("orchestration v2 provider switching", () => { const started = yield* Deferred.make(); const release = yield* Deferred.make(); const compatible = target === "compatible Codex"; + const broken = target === "broken current provider"; + const generation = yield* Ref.make(0); const selection = { - ...(compatible - ? { ...CODEX_MODEL_SELECTION, instanceId: ProviderInstanceId.make("codex-alternate") } - : CLAUDE_MODEL_SELECTION), + ...(broken + ? CODEX_MODEL_SELECTION + : compatible + ? { + ...CODEX_MODEL_SELECTION, + instanceId: ProviderInstanceId.make("codex-alternate"), + } + : CLAUDE_MODEL_SELECTION), options: [{ id: "reasoningEffort", value: "low" }], }; const targetCapabilities = compatible @@ -2702,10 +2742,13 @@ describe("orchestration v2 provider switching", () => { holdFirstTurn: started, releaseFirstTurn: release, failedRunOrdinals: new Set([1]), + ...(broken + ? { brokenRunOrdinals: new Set([1]), nativeThreadGeneration: generation } + : {}), failure: makeProviderFailure({ - message: "Usage limit reached", - class: "usage_limit", - code: "usage_limit", + message: broken ? "Saved history is unusable" : "Usage limit reached", + class: broken ? "provider_error" : "usage_limit", + code: broken ? "INVALID_THINKING_SIGNATURE" : "usage_limit", }), }), makeTestAdapter({ @@ -2717,6 +2760,7 @@ describe("orchestration v2 provider switching", () => { capturedTurns, }), ]; + if (broken) adapters.pop(); const layerRegistry = compatible ? Layer.succeed( ProviderAdapterRegistry.ProviderAdapterRegistryV2, @@ -2840,7 +2884,7 @@ describe("orchestration v2 provider switching", () => { const turns = yield* Ref.get(capturedTurns); assert.lengthOf(turns, delivery === "automatic" ? 3 : 2); const executed = turns.at(-1)!; - assert.equal(executed.driver, compatible ? CODEX_DRIVER : CLAUDE_DRIVER); + assert.equal(executed.driver, compatible || broken ? CODEX_DRIVER : CLAUDE_DRIVER); assert.include(executed.text, "Review the new PR comments"); assert.deepEqual(executed.modelSelection, selection); assert.deepEqual(after.thread.modelSelection, selection); @@ -2854,6 +2898,11 @@ describe("orchestration v2 provider switching", () => { assert.equal(executed.providerThreadId, turns[1]!.providerThreadId); assert.equal(executed.nativeThreadId, turns[1]!.nativeThreadId); } + if (broken) { + assert.notEqual(executed.nativeThreadId, turns[0]!.nativeThreadId); + assert.isFalse(executed.nativeThreadHasTurns); + assert.include(executed.text, "Original work"); + } if (compatible) { assert.equal(executed.nativeThreadId, turns[0]!.nativeThreadId); assert.isEmpty(after.contextHandoffs); From 13c48d4798642d97ceea59f07d2c38cb6e046efd Mon Sep 17 00:00:00 2001 From: PR Batch Tester Date: Sun, 1 Nov 2026 20:59:39 +0000 Subject: [PATCH 04/16] fix(antigravity): start fresh after a lost thinking signature When Claude rejects Antigravity's saved history because a thinking block lost its signature, every later prompt to that native session fails the same way. The native error step is now read as a broken-thread failure: the turn fails with INVALID_THINKING_SIGNATURE and threadDisposition broken, which persists the provider thread as error. The ACP adapter refuses to resume an error row, so the next turn takes the existing resume-failure path into a fresh native session with a context handoff. Quota and other API errors keep the native session. Co-Authored-By: Claude Opus 5.5 --- .../Adapters/AcpAdapterV2.test.ts | 59 ++++++++++-- .../acp/AntigravityTurnOutcome.test.ts | 91 +++++++++++++++---- .../provider/acp/AntigravityTurnOutcome.ts | 67 ++++++++++---- packages/provider-acp/src/server/adapter.ts | 33 +++++-- 4 files changed, 195 insertions(+), 55 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts index 3efa9047d5817..e533be18a0b68 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts @@ -745,6 +745,12 @@ describe("AcpAdapterV2", () => { retryable: true, resetAt: "2030-01-01T00:00:00.000Z", } as const; + const poisoned = { + class: "provider_error", + message: "Saved history is unusable.", + code: "INVALID_THINKING_SIGNATURE", + retryable: true, + } as const; // The runtime ordinal of each prompt, and the session each check watched. const promptRuntimes: Array = []; const watchedSessions: Array = []; @@ -757,8 +763,13 @@ describe("AcpAdapterV2", () => { watchPromptOutcome: (sessionId) => Effect.sync(() => { watchedSessions.push(sessionId); - const failed = watchedSessions.length === 1; - return () => Effect.succeed(failed ? quota : undefined); + const outcome = + watchedSessions.length === 1 + ? { failure: quota, threadDisposition: "reusable" as const } + : watchedSessions.length === 3 + ? { failure: poisoned, threadDisposition: "broken" as const } + : undefined; + return () => Effect.succeed(outcome); }), makeRuntime: makeMockRuntime({ childProcessSpawner: yield* ChildProcessSpawner.ChildProcessSpawner, @@ -805,23 +816,55 @@ describe("AcpAdapterV2", () => { ordinal, }), ); - const events = yield* runtime.events.pipe( - Stream.takeUntil((event) => event.type === "turn.terminal"), - Stream.runCollect, + return Array.from( + yield* runtime.events.pipe( + Stream.takeUntil((event) => event.type === "turn.terminal"), + Stream.runCollect, + ), ); - return events.at(-1); }); - const failed = yield* runTurn(1); + const failed = (yield* runTurn(1)).at(-1); if (failed?.type !== "turn.terminal" || failed.status !== "failed") { return assert.fail("expected a failed terminal"); } assert.deepEqual(failed.failure, quota); - const completed = yield* runTurn(2); + assert.equal(failed.threadDisposition, "reusable"); + const completed = (yield* runTurn(2)).at(-1); assert.equal(completed?.type === "turn.terminal" ? completed.status : null, "completed"); assert.deepEqual(promptRuntimes, [1, 2], "the next turn runs on a fresh runtime"); assert.equal(watchedSessions.length, 2); assert.equal(watchedSessions[0], watchedSessions[1]); + + // Poisoned native history marks the thread broken; its persisted error row is never resumed. + const poisonedEvents = yield* runTurn(3); + const broken = poisonedEvents.at(-1); + if (broken?.type !== "turn.terminal" || broken.status !== "failed") { + return assert.fail("expected a failed terminal"); + } + assert.deepEqual(broken.failure, poisoned); + assert.equal(broken.threadDisposition, "broken"); + const brokenThread = poisonedEvents.findLast( + (event) => event.type === "provider_thread.updated", + ); + assert.equal( + brokenThread?.type === "provider_thread.updated" + ? brokenThread.providerThread.status + : null, + "error", + ); + 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", () => diff --git a/apps/server/src/provider/acp/AntigravityTurnOutcome.test.ts b/apps/server/src/provider/acp/AntigravityTurnOutcome.test.ts index 09b8730bd6001..6665f6de1ff53 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"; @@ -32,10 +33,32 @@ const RATE_LIMIT_ERROR = errorPayload( const INVALID_ARGUMENT_ERROR = errorPayload( "Error 400, Message: Request contains an invalid argument., Status: INVALID_ARGUMENT, Details: []", ); +// Vertex Claude rejects a resumed history whose saved thinking block lost its signature. +const MISSING_SIGNATURE_ERROR = errorPayload( + 'Error 400, Message: {"type":"error","error":{"type":"invalid_request_error","message":"messages.1.content.0.thinking.signature: Field required"},"request_id":"req_vrtx_test"}, Status: INVALID_ARGUMENT, Details: []', +); +const INVALID_SIGNATURE_ERROR = errorPayload( + 'Error 400, Message: {"type":"error","error":{"type":"invalid_request_error","message":"messages.3.content.0: Invalid `signature` in `thinking` block"}}, Status: INVALID_ARGUMENT, Details: []', +); // 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.", ); +const QUOTED_SIGNATURE_ERROR = Buffer.from( + "The API said: messages.1.content.0.thinking.signature: Field required", +); +const reusable = (failure: OrchestrationV2ProviderFailure) => + ({ failure, threadDisposition: "reusable" }) as const; +const POISONED_HISTORY = { + failure: { + class: "provider_error", + message: + "Antigravity's saved conversation can no longer be sent to the model. The next message starts a fresh Antigravity session with this thread's context.", + code: "INVALID_THINKING_SIGNATURE", + retryable: true, + }, + threadDisposition: "broken", +} as const; describe("watchAntigravityPromptOutcome", () => { it.effect("fails only turns whose own steps end in an upstream error", () => @@ -83,13 +106,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( @@ -109,25 +132,47 @@ 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([ [USER_INPUT, input], [ERROR_MESSAGE, INVALID_ARGUMENT_ERROR], ]), - { + reusable({ class: "provider_error", message: "Request contains an invalid argument.", code: "INVALID_ARGUMENT", retryable: false, - }, + }), + ); + // A saved thinking block without a valid signature fails every later prompt. + assert.deepEqual( + yield* turn([ + [USER_INPUT, input], + [ERROR_MESSAGE, MISSING_SIGNATURE_ERROR], + ]), + POISONED_HISTORY, + ); + assert.deepEqual( + yield* turn([ + [USER_INPUT, input], + [ERROR_MESSAGE, INVALID_SIGNATURE_ERROR], + ]), + POISONED_HISTORY, + ); + // Reply text is never evidence, even when it quotes the error. + assert.isUndefined( + yield* turn([ + [USER_INPUT, input], + [PLANNER_RESPONSE, QUOTED_SIGNATURE_ERROR], + ]), ); // A completed tool keeps its usage, and the turn still fails. assert.equal( @@ -142,7 +187,7 @@ describe("watchAntigravityPromptOutcome", () => { stopReason: "end_turn", usage: { inputTokens: 40, outputTokens: 5, totalTokens: 45 }, }, - ))?.class, + ))?.failure.class, "usage_limit", ); assert.isUndefined( @@ -152,12 +197,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), ); @@ -193,19 +241,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 00de1e836296e..3474a07a781fe 100644 --- a/apps/server/src/provider/acp/AntigravityTurnOutcome.ts +++ b/apps/server/src/provider/acp/AntigravityTurnOutcome.ts @@ -19,6 +19,10 @@ 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/; +// Claude rejects a saved thinking block without a valid signature. It sits in the native +// history, so every later prompt to that session fails the same way. +const INVALID_THINKING_SIGNATURE = + /\bmessages\.\d+\.content\.\d+(?:\.thinking\.signature: Field required|: Invalid `signature` in `thinking` block)/; const RESET_TIMESTAMP = /\bquotaResetTimeStamp:([^\s\]]+)/; const RESET_DELAY = /\bquotaResetDelay:((?:\d+(?:\.\d+)?(?:ns|us|µs|μs|ms|h|m|s))+)(?=[\s\]])/; // google.rpc.RetryInfo, which plain rate limits carry instead of a quota reset. @@ -91,7 +95,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()) @@ -100,30 +109,50 @@ 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 (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 (INVALID_THINKING_SIGNATURE.test(text)) { + return { + failure: makeProviderFailure({ + class: "provider_error", + message: + "Antigravity's saved conversation can no longer be sent to the model. The next message starts a fresh Antigravity session with this thread's context.", + code: "INVALID_THINKING_SIGNATURE", + retryable: true, + }), + threadDisposition: "broken", + }; } 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, - resetAt: resetAtFrom(text, now) ?? null, - }); + return reusable( + makeProviderFailure({ + class: rateLimited ? "usage_limit" : "provider_error", + message: message!.trim(), + code: status ?? null, + retryable: rateLimited || Number(httpStatus) >= 500, + resetAt: resetAtFrom(text, now) ?? null, + }), + ); } const conversationSegments = ["antigravity-acp", "conversations"]; diff --git a/packages/provider-acp/src/server/adapter.ts b/packages/provider-acp/src/server/adapter.ts index ad0b716f37ce2..d6e98a9efa132 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; /** @@ -7476,10 +7480,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); @@ -7558,7 +7567,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, @@ -7602,7 +7611,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( retry: context.providerRetry.retry, retryStartedAt: context.providerRetry.startedAt, }), - threadDisposition: "reusable", + threadDisposition, } : { type: "turn.terminal", @@ -8196,7 +8205,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" @@ -8219,6 +8229,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; } @@ -8483,6 +8496,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 16956f96dec5a04f14e732e678efc14718df9036 Mon Sep 17 00:00:00 2001 From: PR Batch Tester Date: Sun, 1 Nov 2026 21:06:44 +0000 Subject: [PATCH 05/16] fix(server): keep broken history invalid after sign-out --- .../ProviderTurnStartService.test.ts | 30 +++++++++++++++++-- .../ProviderTurnStartService.ts | 7 ++++- 2 files changed, 33 insertions(+), 4 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts index cdbd259440ab2..5a8acfadd9708 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts @@ -1785,11 +1785,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; @@ -1815,6 +1836,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 e0797e784ba00..c0b56ba28e984 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, }, }); From 77851ead1dd4abee8a61685cbcabc2e5f616b519 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Tue, 3 Nov 2026 23:09:46 +0000 Subject: [PATCH 06/16] fix(antigravity): preserve broken history across dispatch paths --- .../src/orchestration-v2/Orchestrator.ts | 4 +- .../ProviderSwitch.integration.test.ts | 79 +++++++++++++++---- 2 files changed, 66 insertions(+), 17 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 08ba4ab5cc0aa..149266517d3dc 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -2021,7 +2021,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const providerThread: OrchestrationV2ProviderThread = { ...deliveryProviderThread, providerSessionId, - status: "not_loaded", + status: deliveryProviderThread.status === "error" ? "error" : "not_loaded", firstRunOrdinal: queuedProviderThread.firstRunOrdinal ?? ordinal, lastRunOrdinal: ordinal, handoffIds: appendContextHandoffId( @@ -7030,7 +7030,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: [ diff --git a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts index 2570c28e36b77..88b050ea0fa17 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts @@ -113,6 +113,7 @@ function makeTestAdapter(input: { >; readonly failure?: ReturnType; readonly failedRunOrdinals?: ReadonlySet; + readonly brokenRunOrdinals?: ReadonlySet; readonly interruptedRunOrdinals?: ReadonlySet; readonly holdRunOrdinal?: number; readonly pendingBackgroundWork?: Ref.Ref; @@ -212,6 +213,7 @@ function makeTestAdapter(input: { resumeThread: ({ providerThread }) => Effect.gen(function* () { if ( + (input.brokenRunOrdinals !== undefined && providerThread.status === "error") || input.failResume || (input.failResumeOnce !== undefined && (yield* Ref.getAndSet(input.failResumeOnce, false))) @@ -367,7 +369,9 @@ function makeTestAdapter(input: { }), } : { status: terminalStatus, failure: null }), - threadDisposition: "reusable", + threadDisposition: input.brokenRunOrdinals?.has(turnInput.runOrdinal) + ? "broken" + : "reusable", }, ]; const reportedUsage = input.tokenUsageByRunOrdinal?.[turnInput.runOrdinal]; @@ -383,6 +387,17 @@ function makeTestAdapter(input: { }, }); } + if (input.brokenRunOrdinals?.has(turnInput.runOrdinal)) { + yield* PubSub.publish(events, { + type: "provider_thread.updated", + driver: input.driver, + providerThread: { + ...turnInput.providerThread, + status: "error", + updatedAt: eventTime, + }, + }); + } for (const event of providerEvents) { yield* PubSub.publish(events, event); } @@ -1583,19 +1598,20 @@ describe("orchestration v2 provider switching", () => { it.live.each( ( [ - ["failed", false, false, false], + ["failed", false, false, false, false], + ["failed", false, false, false, true], ["failed", true, true, true], ["interrupted", false, true, false], ["interrupted", true, false, true], ] as const ).map( - ([status, queued, returning, native]) => + ([status, queued, returning, native, broken]) => [ - `${status} context ${queued ? "through the queue" : "immediately"} to ${returning ? "a returning" : "a new"} provider via ${native ? "native history" : "text"}`, - { status, queued, returning, native }, + `${status}${broken ? " broken" : ""} context ${queued ? "through the queue" : "immediately"} to ${returning ? "a returning" : "a new"} provider via ${native ? "native history" : "text"}`, + { status, queued, returning, native, broken }, ] as const, ), - )("hands off %s", ([, { status, queued, returning, native }]) => + )("hands off %s", ([, { status, queued, returning, native, broken }]) => Effect.scoped( Effect.gen(function* () { const cwd = yield* checkpointWorkspace(`handoff-${status}-${queued}-${returning}`); @@ -1606,6 +1622,7 @@ describe("orchestration v2 provider switching", () => { const sourceOrdinal = returning ? 2 : 1; const sourceSelection = returning ? CLAUDE_MODEL_SELECTION : CODEX_MODEL_SELECTION; const targetSelection = returning ? CODEX_MODEL_SELECTION : CLAUDE_MODEL_SELECTION; + const generation = yield* Ref.make(0); const originalPrompt = "Keep the release marker violet and preserve the existing API."; const partialResponse = "I checked the API and found the release configuration."; const layerRegistry = ProviderAdapterRegistry.layerFromAdapters( @@ -1628,6 +1645,12 @@ describe("orchestration v2 provider switching", () => { ...(modelSelection.instanceId === sourceSelection.instanceId ? { failedRunOrdinals: new Set(status === "failed" ? [sourceOrdinal] : []), + ...(broken + ? { + brokenRunOrdinals: new Set([sourceOrdinal]), + nativeThreadGeneration: generation, + } + : {}), interruptedRunOrdinals: new Set( status === "interrupted" ? [sourceOrdinal] : [], ), @@ -1742,8 +1765,17 @@ describe("orchestration v2 provider switching", () => { yield* dispatch("back", "Finish the remaining work", sourceSelection); yield* waitForRun(sourceOrdinal + 2, "completed"); const back = (yield* Ref.get(capturedTurns)).at(-1)!; - assert.notInclude(back.text, originalPrompt); - assert.notInclude(back.text, partialResponse); + if (broken) { + const source = (yield* Ref.get(capturedTurns)).find( + (turn) => turn.modelSelection.instanceId === sourceSelection.instanceId, + )!; + assert.notEqual(back.nativeThreadId, source.nativeThreadId); + assert.include(back.text, originalPrompt); + assert.include(back.text, partialResponse); + } else { + assert.notInclude(back.text, originalPrompt); + assert.notInclude(back.text, partialResponse); + } }).pipe( Effect.provide( ProviderReplayHarness.layerWithRegistry( @@ -2666,6 +2698,7 @@ describe("orchestration v2 provider switching", () => { it.live.each([ ["automatic", "Claude"], ["queue.resume", "compatible Codex"], + ["queue.resume", "broken current provider"], ] as const)( "delivers a queued PR update on the current provider through %s (%s)", ([delivery, target]) => @@ -2676,10 +2709,17 @@ describe("orchestration v2 provider switching", () => { const started = yield* Deferred.make(); const release = yield* Deferred.make(); const compatible = target === "compatible Codex"; + const broken = target === "broken current provider"; + const generation = yield* Ref.make(0); const selection = { - ...(compatible - ? { ...CODEX_MODEL_SELECTION, instanceId: ProviderInstanceId.make("codex-alternate") } - : CLAUDE_MODEL_SELECTION), + ...(broken + ? CODEX_MODEL_SELECTION + : compatible + ? { + ...CODEX_MODEL_SELECTION, + instanceId: ProviderInstanceId.make("codex-alternate"), + } + : CLAUDE_MODEL_SELECTION), options: [{ id: "reasoningEffort", value: "low" }], }; const targetCapabilities = compatible @@ -2702,10 +2742,13 @@ describe("orchestration v2 provider switching", () => { holdFirstTurn: started, releaseFirstTurn: release, failedRunOrdinals: new Set([1]), + ...(broken + ? { brokenRunOrdinals: new Set([1]), nativeThreadGeneration: generation } + : {}), failure: makeProviderFailure({ - message: "Usage limit reached", - class: "usage_limit", - code: "usage_limit", + message: broken ? "Saved history is unusable" : "Usage limit reached", + class: broken ? "provider_error" : "usage_limit", + code: broken ? "INVALID_THINKING_SIGNATURE" : "usage_limit", }), }), makeTestAdapter({ @@ -2717,6 +2760,7 @@ describe("orchestration v2 provider switching", () => { capturedTurns, }), ]; + if (broken) adapters.pop(); const layerRegistry = compatible ? Layer.succeed( ProviderAdapterRegistry.ProviderAdapterRegistryV2, @@ -2840,7 +2884,7 @@ describe("orchestration v2 provider switching", () => { const turns = yield* Ref.get(capturedTurns); assert.lengthOf(turns, delivery === "automatic" ? 3 : 2); const executed = turns.at(-1)!; - assert.equal(executed.driver, compatible ? CODEX_DRIVER : CLAUDE_DRIVER); + assert.equal(executed.driver, compatible || broken ? CODEX_DRIVER : CLAUDE_DRIVER); assert.include(executed.text, "Review the new PR comments"); assert.deepEqual(executed.modelSelection, selection); assert.deepEqual(after.thread.modelSelection, selection); @@ -2854,6 +2898,11 @@ describe("orchestration v2 provider switching", () => { assert.equal(executed.providerThreadId, turns[1]!.providerThreadId); assert.equal(executed.nativeThreadId, turns[1]!.nativeThreadId); } + if (broken) { + assert.notEqual(executed.nativeThreadId, turns[0]!.nativeThreadId); + assert.isFalse(executed.nativeThreadHasTurns); + assert.include(executed.text, "Original work"); + } if (compatible) { assert.equal(executed.nativeThreadId, turns[0]!.nativeThreadId); assert.isEmpty(after.contextHandoffs); From 8e52820bc169c7b6d65b0e55881a4a9b25c8e0fd Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Tue, 3 Nov 2026 23:18:15 +0000 Subject: [PATCH 07/16] fix(antigravity): retain broken history when stop races outcome --- .../Adapters/AcpAdapterV2.test.ts | 290 ++++++++++-------- packages/provider-acp/src/server/adapter.ts | 5 +- 2 files changed, 164 insertions(+), 131 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts index 1da87fea517ec..3686d71531572 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts @@ -733,139 +733,173 @@ describe("AcpAdapterV2", () => { assert.equal(promptOrdinal, 2, "continuation turn must not dispatch session/prompt"); }).pipe(Effect.provide(layerTest), Effect.scoped), ); - it.live("fails an end_turn the outcome check rejects and restarts before the next turn", () => - Effect.gen(function* () { - const path = yield* Path.Path; - const instanceId = ProviderInstanceId.make("acp-test-prompt-outcome"); - const threadId = ThreadId.make("thread-acp-prompt-outcome"); - const quota = { - class: "usage_limit", - message: "Quota reached.", - code: "RESOURCE_EXHAUSTED", - retryable: true, - resetAt: "2030-01-01T00:00:00.000Z", - } as const; - const poisoned = { - class: "provider_error", - message: "Saved history is unusable.", - code: "INVALID_THINKING_SIGNATURE", - retryable: true, - } as const; - // The runtime ordinal of each prompt, and the session each check watched. - const promptRuntimes: Array = []; - const watchedSessions: Array = []; - const adapter = yield* makeAcpAdapterV2({ - instanceId, - selfInvocation: yield* resolveSelfInvocation(), - flavor: { - driver: ACP_TEST_DRIVER, - capabilities: AcpProviderCapabilitiesV2, - watchPromptOutcome: (sessionId) => - Effect.sync(() => { - watchedSessions.push(sessionId); - const outcome = - watchedSessions.length === 1 - ? { failure: quota, threadDisposition: "reusable" as const } - : watchedSessions.length === 3 - ? { failure: poisoned, threadDisposition: "broken" as const } - : undefined; - return () => Effect.succeed(outcome); - }), - makeRuntime: makeMockRuntime({ - childProcessSpawner: yield* ChildProcessSpawner.ChildProcessSpawner, - mockAgentPath: yield* path.fromFileUrl( - new URL("../../../scripts/acp-mock-agent.ts", import.meta.url), - ), - wrapRuntime: (runtime, runtimeOrdinal) => ({ - ...runtime, - prompt: () => - Effect.sync(() => { - promptRuntimes.push(runtimeOrdinal); - return { 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"), - modelSelection, - runtimePolicy, - }); - const providerThread = yield* runtime.ensureThread({ - threadId, - modelSelection, - runtimePolicy, - }); - const runTurn = (ordinal: number) => - Effect.gen(function* () { - yield* runtime.startTurn( - makeTurnInput({ - threadId, - providerThread, - instanceId, - runtimePolicy, - now: yield* DateTime.now, - ordinal, + it.live.each([false, true])( + "fails a rejected outcome and preserves broken history with stop=%s", + (interrupt) => + Effect.gen(function* () { + const path = yield* Path.Path; + const checking = yield* Deferred.make(); + const releaseCheck = yield* Deferred.make(); + const cancelled = yield* Deferred.make(); + const idAllocator = yield* IdAllocator.IdAllocatorV2; + const instanceId = ProviderInstanceId.make("acp-test-prompt-outcome"); + const threadId = ThreadId.make("thread-acp-prompt-outcome"); + const quota = { + class: "usage_limit", + message: "Quota reached.", + code: "RESOURCE_EXHAUSTED", + retryable: true, + resetAt: "2030-01-01T00:00:00.000Z", + } as const; + const poisoned = { + class: "provider_error", + message: "Saved history is unusable.", + code: "INVALID_THINKING_SIGNATURE", + retryable: true, + } as const; + // The runtime ordinal of each prompt, and the session each check watched. + const promptRuntimes: Array = []; + const watchedSessions: Array = []; + const adapter = yield* makeAcpAdapterV2({ + instanceId, + selfInvocation: yield* resolveSelfInvocation(), + flavor: { + driver: ACP_TEST_DRIVER, + capabilities: AcpProviderCapabilitiesV2, + watchPromptOutcome: (sessionId) => + Effect.sync(() => { + watchedSessions.push(sessionId); + const outcome = + watchedSessions.length === 1 + ? { failure: quota, threadDisposition: "reusable" as const } + : watchedSessions.length === 3 + ? { failure: poisoned, threadDisposition: "broken" as const } + : undefined; + return () => + interrupt && outcome?.threadDisposition === "broken" + ? Deferred.succeed(checking, undefined).pipe( + Effect.andThen(Deferred.await(releaseCheck)), + Effect.as(outcome), + ) + : Effect.succeed(outcome); + }), + makeRuntime: makeMockRuntime({ + childProcessSpawner: yield* ChildProcessSpawner.ChildProcessSpawner, + mockAgentPath: yield* path.fromFileUrl( + new URL("../../../scripts/acp-mock-agent.ts", import.meta.url), + ), + wrapRuntime: (runtime, runtimeOrdinal) => ({ + ...runtime, + cancel: Deferred.succeed(cancelled, undefined).pipe(Effect.asVoid), + prompt: () => + Effect.sync(() => { + promptRuntimes.push(runtimeOrdinal); + return { stopReason: "end_turn" as const }; + }), + }), }), - ); - return Array.from( - yield* runtime.events.pipe( - Stream.takeUntil((event) => event.type === "turn.terminal"), - Stream.runCollect, - ), - ); + }, }); - - const failed = (yield* runTurn(1)).at(-1); - if (failed?.type !== "turn.terminal" || failed.status !== "failed") { - return assert.fail("expected a failed terminal"); - } - assert.deepEqual(failed.failure, quota); - assert.equal(failed.threadDisposition, "reusable"); - const completed = (yield* runTurn(2)).at(-1); - assert.equal(completed?.type === "turn.terminal" ? completed.status : null, "completed"); - assert.deepEqual(promptRuntimes, [1, 2], "the next turn runs on a fresh runtime"); - assert.equal(watchedSessions.length, 2); - assert.equal(watchedSessions[0], watchedSessions[1]); - - // Poisoned native history marks the thread broken; its persisted error row is never resumed. - const poisonedEvents = yield* runTurn(3); - const broken = poisonedEvents.at(-1); - if (broken?.type !== "turn.terminal" || broken.status !== "failed") { - return assert.fail("expected a failed terminal"); - } - assert.deepEqual(broken.failure, poisoned); - assert.equal(broken.threadDisposition, "broken"); - const brokenThread = poisonedEvents.findLast( - (event) => event.type === "provider_thread.updated", - ); - assert.equal( - brokenThread?.type === "provider_thread.updated" - ? brokenThread.providerThread.status - : null, - "error", - ); - const resumed = yield* Effect.result( - runtime.resumeThread({ + 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, - providerThread: { ...providerThread, status: "error" }, + providerSessionId: ProviderSessionId.make("provider-session-acp-prompt-outcome"), modelSelection, runtimePolicy, - }), - ); - assert.equal( - resumed._tag === "Failure" ? resumed.failure._tag : null, - "ProviderAdapterResumeThreadError", - ); - }).pipe(Effect.provide(layerTest), Effect.scoped), + }); + const providerThread = yield* runtime.ensureThread({ + threadId, + modelSelection, + runtimePolicy, + }); + const runTurn = (ordinal: number) => + Effect.gen(function* () { + yield* runtime.startTurn( + makeTurnInput({ + threadId, + providerThread, + instanceId, + runtimePolicy, + now: yield* DateTime.now, + ordinal, + }), + ); + return Array.from( + yield* runtime.events.pipe( + Stream.takeUntil((event) => event.type === "turn.terminal"), + Stream.runCollect, + ), + ); + }); + + const failed = (yield* runTurn(1)).at(-1); + if (failed?.type !== "turn.terminal" || failed.status !== "failed") { + return assert.fail("expected a failed terminal"); + } + assert.deepEqual(failed.failure, quota); + assert.equal(failed.threadDisposition, "reusable"); + const completed = (yield* runTurn(2)).at(-1); + assert.equal(completed?.type === "turn.terminal" ? completed.status : null, "completed"); + assert.deepEqual(promptRuntimes, [1, 2], "the next turn runs on a fresh runtime"); + assert.equal(watchedSessions.length, 2); + assert.equal(watchedSessions[0], watchedSessions[1]); + + // Poisoned native history marks the thread broken; its persisted error row is never resumed. + const poisonedEvents = yield* Effect.gen(function* () { + if (!interrupt) return yield* runTurn(3); + const turn = yield* Effect.forkChild(runTurn(3)); + yield* Deferred.await(checking); + const stop = yield* Effect.forkChild( + runtime.interruptTurn({ + providerThread, + providerTurnId: idAllocator.derive.providerTurn({ + driver: ACP_TEST_DRIVER, + nativeTurnId: acpScopedNativeId(instanceId, `${watchedSessions[2]}:turn:3`), + }), + }), + ); + yield* Deferred.await(cancelled); + yield* Deferred.succeed(releaseCheck, undefined); + const events = yield* Fiber.join(turn); + yield* Fiber.join(stop); + return events; + }); + const broken = poisonedEvents.at(-1); + if ( + broken?.type !== "turn.terminal" || + broken.status !== (interrupt ? "interrupted" : "failed") + ) { + return assert.fail("expected a failed terminal"); + } + assert.deepEqual(broken.failure, interrupt ? null : poisoned); + assert.equal(broken.threadDisposition, "broken"); + const brokenThread = poisonedEvents.findLast( + (event) => event.type === "provider_thread.updated", + ); + assert.equal( + brokenThread?.type === "provider_thread.updated" + ? brokenThread.providerThread.status + : null, + "error", + ); + 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* () { diff --git a/packages/provider-acp/src/server/adapter.ts b/packages/provider-acp/src/server/adapter.ts index 18a821fe17d24..ba1b73f24c7bd 100644 --- a/packages/provider-acp/src/server/adapter.ts +++ b/packages/provider-acp/src/server/adapter.ts @@ -7481,8 +7481,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( ): Effect.fn.Return { if (context.finalized) return; const settledStatus = context.interrupted ? "interrupted" : status; - const threadDisposition = - settledStatus === "failed" ? (options?.threadDisposition ?? "reusable") : "reusable"; + const threadDisposition = options?.threadDisposition ?? "reusable"; context.finalizedStatus = settledStatus; context.finalized = true; const directStopQuarantine = yield* Ref.get(stoppedRunQuarantine); @@ -7615,7 +7614,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( runOrdinal: context.input.runOrdinal, status: settledStatus, failure: null, - threadDisposition: "reusable", + threadDisposition, }, ); const subagentsRequiringCarryover = [...context.subagents.values()].filter( From 57868a9380ddbc172593c8e198f5d424961c0248 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Tue, 3 Nov 2026 23:21:12 +0000 Subject: [PATCH 08/16] test(antigravity): keep stop coverage in the existing outcome case --- .../Adapters/AcpAdapterV2.test.ts | 332 +++++++++--------- 1 file changed, 172 insertions(+), 160 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts index aae5d9ea922b4..ab7bf518372b6 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts @@ -733,173 +733,185 @@ describe("AcpAdapterV2", () => { assert.equal(promptOrdinal, 2, "continuation turn must not dispatch session/prompt"); }).pipe(Effect.provide(layerTest), Effect.scoped), ); - it.live.each([false, true])( - "fails a rejected outcome and preserves broken history with stop=%s", - (interrupt) => - Effect.gen(function* () { - const path = yield* Path.Path; - const checking = yield* Deferred.make(); - const releaseCheck = yield* Deferred.make(); - const cancelled = yield* Deferred.make(); - const idAllocator = yield* IdAllocator.IdAllocatorV2; - const instanceId = ProviderInstanceId.make("acp-test-prompt-outcome"); - const threadId = ThreadId.make("thread-acp-prompt-outcome"); - const quota = { - class: "usage_limit", - message: "Quota reached.", - code: "RESOURCE_EXHAUSTED", - retryable: true, - resetAt: "2030-01-01T00:00:00.000Z", - } as const; - const poisoned = { - class: "provider_error", - message: "Saved history is unusable.", - code: "INVALID_THINKING_SIGNATURE", - retryable: true, - } as const; - // The runtime ordinal of each prompt, and the session each check watched. - const promptRuntimes: Array = []; - const watchedSessions: Array = []; - const adapter = yield* makeAcpAdapterV2({ - instanceId, - selfInvocation: yield* resolveSelfInvocation(), - flavor: { - driver: ACP_TEST_DRIVER, - capabilities: AcpProviderCapabilitiesV2, - watchPromptOutcome: (sessionId) => - Effect.sync(() => { - watchedSessions.push(sessionId); - const outcome = - watchedSessions.length === 1 - ? { failure: quota, threadDisposition: "reusable" as const } - : watchedSessions.length === 3 - ? { failure: poisoned, threadDisposition: "broken" as const } - : undefined; - return () => - interrupt && outcome?.threadDisposition === "broken" - ? Deferred.succeed(checking, undefined).pipe( - Effect.andThen(Deferred.await(releaseCheck)), - Effect.as(outcome), - ) - : Effect.succeed(outcome); - }), - makeRuntime: makeMockRuntime({ - childProcessSpawner: yield* ChildProcessSpawner.ChildProcessSpawner, - mockAgentPath: yield* path.fromFileUrl( - new URL("../../../scripts/acp-mock-agent.ts", import.meta.url), - ), - wrapRuntime: (runtime, runtimeOrdinal) => ({ - ...runtime, - cancel: Deferred.succeed(cancelled, undefined).pipe(Effect.asVoid), - prompt: () => - Effect.sync(() => { - promptRuntimes.push(runtimeOrdinal); - return { stopReason: "end_turn" as const }; - }), - }), + it.live("fails an end_turn the outcome check rejects and restarts before the next turn", () => + Effect.gen(function* () { + const path = yield* Path.Path; + const checking = yield* Deferred.make(); + const releaseCheck = yield* Deferred.make(); + const cancelled = yield* Deferred.make(); + const idAllocator = yield* IdAllocator.IdAllocatorV2; + const instanceId = ProviderInstanceId.make("acp-test-prompt-outcome"); + const threadId = ThreadId.make("thread-acp-prompt-outcome"); + const quota = { + class: "usage_limit", + message: "Quota reached.", + code: "RESOURCE_EXHAUSTED", + retryable: true, + resetAt: "2030-01-01T00:00:00.000Z", + } as const; + const poisoned = { + class: "provider_error", + message: "Saved history is unusable.", + code: "INVALID_THINKING_SIGNATURE", + retryable: true, + } as const; + // The runtime ordinal of each prompt, and the session each check watched. + const promptRuntimes: Array = []; + const watchedSessions: Array = []; + const adapter = yield* makeAcpAdapterV2({ + instanceId, + selfInvocation: yield* resolveSelfInvocation(), + flavor: { + driver: ACP_TEST_DRIVER, + capabilities: AcpProviderCapabilitiesV2, + watchPromptOutcome: (sessionId) => + Effect.sync(() => { + watchedSessions.push(sessionId); + const outcome = + watchedSessions.length === 1 + ? { failure: quota, threadDisposition: "reusable" as const } + : watchedSessions.length >= 3 + ? { failure: poisoned, threadDisposition: "broken" as const } + : undefined; + const interrupted = watchedSessions.length === 4; + return () => + interrupted + ? Deferred.succeed(checking, undefined).pipe( + Effect.andThen(Deferred.await(releaseCheck)), + Effect.as(outcome), + ) + : Effect.succeed(outcome); }), - }, - }); - 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"), - modelSelection, - runtimePolicy, + makeRuntime: makeMockRuntime({ + childProcessSpawner: yield* ChildProcessSpawner.ChildProcessSpawner, + mockAgentPath: yield* path.fromFileUrl( + new URL("../../../scripts/acp-mock-agent.ts", import.meta.url), + ), + wrapRuntime: (runtime, runtimeOrdinal) => ({ + ...runtime, + cancel: Deferred.succeed(cancelled, undefined).pipe(Effect.asVoid), + prompt: () => + Effect.sync(() => { + promptRuntimes.push(runtimeOrdinal); + return { 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"), + modelSelection, + runtimePolicy, + }); + const providerThread = yield* runtime.ensureThread({ + threadId, + modelSelection, + runtimePolicy, + }); + const runTurn = (ordinal: number) => + Effect.gen(function* () { + yield* runtime.startTurn( + makeTurnInput({ + threadId, + providerThread, + instanceId, + runtimePolicy, + now: yield* DateTime.now, + ordinal, + }), + ); + return Array.from( + yield* runtime.events.pipe( + Stream.takeUntil((event) => event.type === "turn.terminal"), + Stream.runCollect, + ), + ); }); - const providerThread = yield* runtime.ensureThread({ + + const failed = (yield* runTurn(1)).at(-1); + if (failed?.type !== "turn.terminal" || failed.status !== "failed") { + return assert.fail("expected a failed terminal"); + } + assert.deepEqual(failed.failure, quota); + assert.equal(failed.threadDisposition, "reusable"); + const completed = (yield* runTurn(2)).at(-1); + assert.equal(completed?.type === "turn.terminal" ? completed.status : null, "completed"); + assert.deepEqual(promptRuntimes, [1, 2], "the next turn runs on a fresh runtime"); + assert.equal(watchedSessions.length, 2); + assert.equal(watchedSessions[0], watchedSessions[1]); + + // Poisoned native history marks the thread broken; its persisted error row is never resumed. + const poisonedEvents = yield* runTurn(3); + const broken = poisonedEvents.at(-1); + if (broken?.type !== "turn.terminal" || broken.status !== "failed") { + return assert.fail("expected a failed terminal"); + } + assert.deepEqual(broken.failure, poisoned); + assert.equal(broken.threadDisposition, "broken"); + const brokenThread = poisonedEvents.findLast( + (event) => event.type === "provider_thread.updated", + ); + assert.equal( + brokenThread?.type === "provider_thread.updated" + ? brokenThread.providerThread.status + : null, + "error", + ); + const resumed = yield* Effect.result( + runtime.resumeThread({ threadId, + providerThread: { ...providerThread, status: "error" }, modelSelection, runtimePolicy, - }); - const runTurn = (ordinal: number) => - Effect.gen(function* () { - yield* runtime.startTurn( - makeTurnInput({ - threadId, - providerThread, - instanceId, - runtimePolicy, - now: yield* DateTime.now, - ordinal, - }), - ); - return Array.from( - yield* runtime.events.pipe( - Stream.takeUntil((event) => event.type === "turn.terminal"), - Stream.runCollect, - ), - ); - }); + }), + ); + assert.equal( + resumed._tag === "Failure" ? resumed.failure._tag : null, + "ProviderAdapterResumeThreadError", + ); - const failed = (yield* runTurn(1)).at(-1); - if (failed?.type !== "turn.terminal" || failed.status !== "failed") { - return assert.fail("expected a failed terminal"); - } - assert.deepEqual(failed.failure, quota); - assert.equal(failed.threadDisposition, "reusable"); - const completed = (yield* runTurn(2)).at(-1); - assert.equal(completed?.type === "turn.terminal" ? completed.status : null, "completed"); - assert.deepEqual(promptRuntimes, [1, 2], "the next turn runs on a fresh runtime"); - assert.equal(watchedSessions.length, 2); - assert.equal(watchedSessions[0], watchedSessions[1]); - - // Poisoned native history marks the thread broken; its persisted error row is never resumed. - const poisonedEvents = yield* Effect.gen(function* () { - if (!interrupt) return yield* runTurn(3); - const turn = yield* Effect.forkChild(runTurn(3)); - yield* Deferred.await(checking); - const stop = yield* Effect.forkChild( - runtime.interruptTurn({ - providerThread, - providerTurnId: idAllocator.derive.providerTurn({ - driver: ACP_TEST_DRIVER, - nativeTurnId: acpScopedNativeId(instanceId, `${watchedSessions[2]}:turn:3`), - }), - }), - ); - yield* Deferred.await(cancelled); - yield* Deferred.succeed(releaseCheck, undefined); - const events = yield* Fiber.join(turn); - yield* Fiber.join(stop); - return events; - }); - const broken = poisonedEvents.at(-1); - if ( - broken?.type !== "turn.terminal" || - broken.status !== (interrupt ? "interrupted" : "failed") - ) { - return assert.fail("expected a failed terminal"); - } - assert.deepEqual(broken.failure, interrupt ? null : poisoned); - assert.equal(broken.threadDisposition, "broken"); - const brokenThread = poisonedEvents.findLast( - (event) => event.type === "provider_thread.updated", - ); - assert.equal( - brokenThread?.type === "provider_thread.updated" - ? brokenThread.providerThread.status - : null, - "error", - ); - const resumed = yield* Effect.result( - runtime.resumeThread({ - threadId, - providerThread: { ...providerThread, status: "error" }, - modelSelection, - runtimePolicy, + const turn = yield* Effect.forkChild(runTurn(4)); + yield* Deferred.await(checking); + const stop = yield* Effect.forkChild( + runtime.interruptTurn({ + providerThread, + providerTurnId: idAllocator.derive.providerTurn({ + driver: ACP_TEST_DRIVER, + nativeTurnId: acpScopedNativeId(instanceId, `${watchedSessions[3]}:turn:4`), }), - ); - assert.equal( - resumed._tag === "Failure" ? resumed.failure._tag : null, - "ProviderAdapterResumeThreadError", - ); - }).pipe(Effect.provide(layerTest), Effect.scoped), + }), + ); + yield* Deferred.await(cancelled); + yield* Deferred.succeed(releaseCheck, undefined); + const interruptedEvents = yield* Fiber.join(turn); + yield* Fiber.join(stop); + const interrupted = interruptedEvents.at(-1); + assert.equal( + interrupted?.type === "turn.terminal" ? interrupted.status : null, + "interrupted", + ); + assert.equal( + interrupted?.type === "turn.terminal" ? interrupted.threadDisposition : null, + "broken", + ); + const interruptedThread = interruptedEvents.findLast( + (event) => event.type === "provider_thread.updated", + ); + assert.equal( + interruptedThread?.type === "provider_thread.updated" + ? interruptedThread.providerThread.status + : null, + "error", + ); + }).pipe(Effect.provide(layerTest), Effect.scoped), ); it.live("retains returned usage when Stop finalizes before the prompt callback", () => Effect.gen(function* () { From 891a00e9157ce7951cde038c7a38787a61180ae0 Mon Sep 17 00:00:00 2001 From: shivam <91240327+shivamhwp@users.noreply.github.com> Date: Wed, 4 Nov 2026 02:25:00 +0000 Subject: [PATCH 09/16] test(antigravity): drop the quoted-reply row that passes on main Co-Authored-By: Claude Opus 5.5 --- .../src/provider/acp/AntigravityTurnOutcome.test.ts | 10 ---------- 1 file changed, 10 deletions(-) diff --git a/apps/server/src/provider/acp/AntigravityTurnOutcome.test.ts b/apps/server/src/provider/acp/AntigravityTurnOutcome.test.ts index 6665f6de1ff53..475f7bebfd713 100644 --- a/apps/server/src/provider/acp/AntigravityTurnOutcome.test.ts +++ b/apps/server/src/provider/acp/AntigravityTurnOutcome.test.ts @@ -44,9 +44,6 @@ const INVALID_SIGNATURE_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 QUOTED_SIGNATURE_ERROR = Buffer.from( - "The API said: messages.1.content.0.thinking.signature: Field required", -); const reusable = (failure: OrchestrationV2ProviderFailure) => ({ failure, threadDisposition: "reusable" }) as const; const POISONED_HISTORY = { @@ -167,13 +164,6 @@ describe("watchAntigravityPromptOutcome", () => { ]), POISONED_HISTORY, ); - // Reply text is never evidence, even when it quotes the error. - assert.isUndefined( - yield* turn([ - [USER_INPUT, input], - [PLANNER_RESPONSE, QUOTED_SIGNATURE_ERROR], - ]), - ); // A completed tool keeps its usage, and the turn still fails. assert.equal( (yield* turn( From 387b9e986db33ab970d6765572fde0ad99e2efa9 Mon Sep 17 00:00:00 2001 From: shivam <91240327+shivamhwp@users.noreply.github.com> Date: Wed, 4 Nov 2026 02:31:11 +0000 Subject: [PATCH 10/16] fix(acp): tag broken native history errors --- packages/provider-acp/src/server/adapter.ts | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/packages/provider-acp/src/server/adapter.ts b/packages/provider-acp/src/server/adapter.ts index c272dd54e3c88..fc60eeab1d0b2 100644 --- a/packages/provider-acp/src/server/adapter.ts +++ b/packages/provider-acp/src/server/adapter.ts @@ -8514,7 +8514,10 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( }) { 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* new ProviderAdapter.ProviderAdapterProtocolError({ + driver, + detail: "The native session history is unusable.", + }); } return yield* runtimeTransitionPermit.withPermit( Effect.gen(function* () { From f1762d83b1e81cb8cfe2c21c81cccb8fde88c099 Mon Sep 17 00:00:00 2001 From: shivam <91240327+shivamhwp@users.noreply.github.com> Date: Thu, 5 Nov 2026 06:14:42 +0000 Subject: [PATCH 11/16] docs(acp): clarify recovery after broken native history --- packages/provider-acp/src/server/adapter.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/provider-acp/src/server/adapter.ts b/packages/provider-acp/src/server/adapter.ts index 09809f8ee3e79..edcc6c2890218 100644 --- a/packages/provider-acp/src/server/adapter.ts +++ b/packages/provider-acp/src/server/adapter.ts @@ -314,7 +314,7 @@ 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. A `broken` thread's native - * history is unusable, so it is never resumed and the next turn starts fresh. + * history is unusable, so the next turn falls back to a fresh session. */ readonly watchPromptOutcome?: ( sessionId: string, From 7a1af726f4fb3572caef175fe18d938504365369 Mon Sep 17 00:00:00 2001 From: shivam <91240327+shivamhwp@users.noreply.github.com> Date: Thu, 5 Nov 2026 19:57:51 +0000 Subject: [PATCH 12/16] fix(server): resume-fallback recovery no longer reintroduces context from before /clear A fresh-context run sets a history boundary, but the resume-fallback recovery summary still covered every earlier run. A post-/clear broken native session could therefore hand pre-clear history to its replacement. The recovery handoff now starts at the latest started fresh-context boundary, using the same contextBoundaryOrdinal helper the orchestrator uses. Co-Authored-By: Claude Opus 5.5 --- .../orchestration-v2/ContextHandoffService.ts | 24 ++++++++++ .../src/orchestration-v2/Orchestrator.ts | 24 +--------- .../ProviderTurnStartService.ts | 19 +++++--- .../ProviderSwitch.integration.test.ts | 44 +++++++++++++++++-- 4 files changed, 78 insertions(+), 33 deletions(-) diff --git a/apps/server/src/orchestration-v2/ContextHandoffService.ts b/apps/server/src/orchestration-v2/ContextHandoffService.ts index 10f2701446268..37a37ea78e13d 100644 --- a/apps/server/src/orchestration-v2/ContextHandoffService.ts +++ b/apps/server/src/orchestration-v2/ContextHandoffService.ts @@ -26,6 +26,7 @@ import { import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; import { createOrchestrationV2TurnItemVisibility } from "@t3tools/shared/orchestrationV2Timeline"; import type { ProjectionStoreV2Shape } from "./ProjectionStore.ts"; +import { cancelledBeforeStart } from "./RestartBackgroundNote.ts"; export class ContextHandoffPrepareError extends Schema.TaggedError()( "ContextHandoffPrepareError", @@ -114,6 +115,29 @@ function compactText(text: string, maxLength = 240): string { return `${compacted.slice(0, last >= 0xd800 && last <= 0xdbff ? end - 1 : end)}...`; } +/** Ordinal of the latest started fresh-context run. Runs before it never reach a provider again. */ +export function contextBoundaryOrdinal( + runs: ReadonlyArray< + Pick< + OrchestrationV2Run, + "ordinal" | "freshContext" | "status" | "startedAt" | "restartContinuationOfRunId" + > + >, +): number { + let boundary = 0; + for (const run of runs) { + if ( + run.freshContext === true && + run.status !== "queued" && + !cancelledBeforeStart(run) && + run.ordinal > boundary + ) { + boundary = run.ordinal; + } + } + return boundary; +} + export function visibleDeltaRunOrdinals( projection: { readonly runs: ReadonlyArray> }, items: ReadonlyArray>, diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index edf87f9b25d0b..6b3779f2f32e4 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -113,6 +113,7 @@ import { import { CommandReceiptStoreV2 } from "./CommandReceiptStore.ts"; import { ContextHandoffServiceV2, + contextBoundaryOrdinal, readForkHistoryItems, readInheritedHandoffItems, visibleConversationItems, @@ -999,29 +1000,6 @@ function appendContextHandoffId( return handoffId === null ? handoffIds : Array.from(new Set([...handoffIds, handoffId])); } -/** Ordinal of the latest started fresh-context run. Runs before it never reach a provider again. */ -function contextBoundaryOrdinal( - runs: ReadonlyArray< - Pick< - OrchestrationV2Run, - "ordinal" | "freshContext" | "status" | "startedAt" | "restartContinuationOfRunId" - > - >, -): number { - let boundary = 0; - for (const run of runs) { - if ( - run.freshContext === true && - run.status !== "queued" && - !cancelledBeforeStart(run) && - run.ordinal > boundary - ) { - boundary = run.ordinal; - } - } - return boundary; -} - /** Threads with no started run (imported, or reserved by another queued run) never cross it. */ function providerThreadPredatesBoundary( providerThread: Pick, diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index b703029946855..ea3dca6f52dca 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -1079,6 +1079,8 @@ export const layer: Layer.Layer< type: "provider_resume_fallback", }); const createdAt = yield* DateTime.now; + const boundary = ContextHandoffService.contextBoundaryOrdinal(projection.runs); + const from = Math.max(1, boundary); const handoff = yield* contextHandoffService.prepareProviderHandoff({ threadId: projection.thread.id, targetRunId: run.id, @@ -1087,15 +1089,18 @@ export const layer: Layer.Layer< toProviderThreadId: providerThread.id, fromProviderInstanceId: providerThread.providerInstanceId, toProviderInstanceId: run.providerInstanceId, - coveredRunOrdinals: { from: 1, to: Math.max(1, run.ordinal - 1) }, + coveredRunOrdinals: { from, to: Math.max(from, run.ordinal - 1) }, strategy: "full_thread_summary", runs: projection.runs, - items: (yield* projectionStore.getTurnStartHistory(input.threadId)).filter( - (item) => - item.runId === null || - projection.runs.some( - (source) => source.id === item.runId && source.ordinal < run.ordinal, - ), + items: (yield* projectionStore.getTurnStartHistory(input.threadId)).filter((item) => + item.runId === null + ? boundary === 0 + : projection.runs.some( + (source) => + source.id === item.runId && + source.ordinal >= from && + source.ordinal < run.ordinal, + ), ), createdAt, }); diff --git a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts index 8446cbdae84b0..bf1edfe58bac1 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts @@ -2799,13 +2799,18 @@ describe("orchestration v2 provider switching", () => { ), ); - it.live.each(["immediately", "behind a running turn"] as const)( + it.live.each([ + "immediately", + "behind a running turn", + "and recovers it after it breaks", + ] as const)( "starts a fresh provider session %s and never hands earlier runs to a later provider", (variant) => Effect.scoped( Effect.gen(function* () { - const queued = variant !== "immediately"; - const key = queued ? "queued" : "immediate"; + const queued = variant === "behind a running turn"; + const broken = variant === "and recovers it after it breaks"; + const key = queued ? "queued" : broken ? "broken" : "immediate"; const cwd = yield* checkpointWorkspace(`fresh-context-${key}`); const capturedTurns = yield* Ref.make>([]); const closedSessions = yield* Ref.make>([]); @@ -2827,6 +2832,17 @@ describe("orchestration v2 provider switching", () => { ...(queued ? { holdFirstTurn: started, releaseFirstTurn: release, holdRunOrdinal: 2 } : {}), + ...(broken + ? { + failedRunOrdinals: new Set([3]), + brokenRunOrdinals: new Set([3]), + failure: makeProviderFailure({ + message: "Invalid thinking signature", + class: "provider_error", + code: "INVALID_THINKING_SIGNATURE", + }), + } + : {}), }), makeTestAdapter({ instanceId: CODEX_MODEL_SELECTION.instanceId, @@ -2920,6 +2936,11 @@ describe("orchestration v2 provider switching", () => { // Claude hosts one native thread per session, so an idle old one is closed. assert.deepEqual(yield* Ref.get(closedSessions), queued ? [] : [firstSession.id]); yield* Ref.set(backgroundWork, false); + if (broken) { + yield* send("epsilon", CLAUDE_MODEL_SELECTION); + yield* worker.drain(); + yield* waitForIdle(freshThreadId); + } yield* send("delta", CODEX_MODEL_SELECTION); yield* worker.drain(); yield* waitForIdle(freshThreadId); @@ -2948,6 +2969,23 @@ describe("orchestration v2 provider switching", () => { if (queued) { assert.equal(turnFor("gamma").providerThreadId, turnFor("beta").providerThreadId); } + if (broken) { + const recovered = turnFor("epsilon"); + assert.equal(recovered.providerThreadId, turnFor("beta").providerThreadId); + assert.notEqual(recovered.nativeThreadId, turnFor("beta").nativeThreadId); + assert.include(recovered.text, "Prompt beta"); + assert.notInclude(recovered.text, "Prompt alpha"); + assert.notInclude(recovered.text, "Prompt zero"); + const recoveredRun = projection.runs.find( + (run) => run.userMessageId === `message:fresh-context:${key}:epsilon`, + )!; + assert.equal( + projection.contextHandoffs.find( + (handoff) => handoff.targetRunId === recoveredRun.id, + )!.coveredRunOrdinals.from, + 3, + ); + } const codexTurn = turnFor("delta"); assert.notEqual(codexTurn.providerThreadId, turnFor("zero").providerThreadId); assert.include(codexTurn.text, "Prompt beta"); From 623c24ea38ad1395397fbaef8c7e1ddb67bfdeb8 Mon Sep 17 00:00:00 2001 From: shivam <91240327+shivamhwp@users.noreply.github.com> Date: Sat, 7 Nov 2026 14:25:12 +0000 Subject: [PATCH 13/16] fix(server): keep a broken Antigravity session marked across instance handover Co-Authored-By: Claude Opus 5.5 --- .../src/orchestration-v2/Orchestrator.ts | 1 + .../acp/AntigravityTurnOutcome.test.ts | 30 +++++-------------- 2 files changed, 8 insertions(+), 23 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index ac2d25a0c15fc..205a35f152504 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -1976,6 +1976,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const deliveryProviderThread = canResumeAcrossInstances ? { ...queuedProviderThread, + status: activeProviderThread.status, nativeThreadRef: activeProviderThread.nativeThreadRef, nativeConversationHeadRef: activeProviderThread.nativeConversationHeadRef, nativeMetadata: activeProviderThread.nativeMetadata, diff --git a/apps/server/src/provider/acp/AntigravityTurnOutcome.test.ts b/apps/server/src/provider/acp/AntigravityTurnOutcome.test.ts index 475f7bebfd713..b51d750312286 100644 --- a/apps/server/src/provider/acp/AntigravityTurnOutcome.test.ts +++ b/apps/server/src/provider/acp/AntigravityTurnOutcome.test.ts @@ -46,16 +46,6 @@ const QUOTED_NOTICE = Buffer.from( ); const reusable = (failure: OrchestrationV2ProviderFailure) => ({ failure, threadDisposition: "reusable" }) as const; -const POISONED_HISTORY = { - failure: { - class: "provider_error", - message: - "Antigravity's saved conversation can no longer be sent to the model. The next message starts a fresh Antigravity session with this thread's context.", - code: "INVALID_THINKING_SIGNATURE", - retryable: true, - }, - threadDisposition: "broken", -} as const; describe("watchAntigravityPromptOutcome", () => { it.effect("fails only turns whose own steps end in an upstream error", () => @@ -150,20 +140,14 @@ describe("watchAntigravityPromptOutcome", () => { }), ); // A saved thinking block without a valid signature fails every later prompt. - assert.deepEqual( - yield* turn([ + for (const error of [MISSING_SIGNATURE_ERROR, INVALID_SIGNATURE_ERROR]) { + const outcome = yield* turn([ [USER_INPUT, input], - [ERROR_MESSAGE, MISSING_SIGNATURE_ERROR], - ]), - POISONED_HISTORY, - ); - assert.deepEqual( - yield* turn([ - [USER_INPUT, input], - [ERROR_MESSAGE, INVALID_SIGNATURE_ERROR], - ]), - POISONED_HISTORY, - ); + [ERROR_MESSAGE, error], + ]); + assert.equal(outcome?.failure.code, "INVALID_THINKING_SIGNATURE"); + assert.equal(outcome?.threadDisposition, "broken"); + } // A completed tool keeps its usage, and the turn still fails. assert.equal( (yield* turn( From 154f5cdeb65db91e70a000fc095012b1dc3fd936 Mon Sep 17 00:00:00 2001 From: shivam <91240327+shivamhwp@users.noreply.github.com> Date: Sat, 7 Nov 2026 14:49:19 +0000 Subject: [PATCH 14/16] test(server): cover broken history across queued instance handover --- .../ProviderSwitch.integration.test.ts | 29 +++++++++++-------- 1 file changed, 17 insertions(+), 12 deletions(-) diff --git a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts index bf1edfe58bac1..9076f1056cc59 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts @@ -3132,6 +3132,7 @@ describe("orchestration v2 provider switching", () => { ["automatic", "Claude"], ["queue.resume", "compatible Codex"], ["queue.resume", "broken current provider"], + ["queue.resume", "broken compatible Codex"], ] as const)( "delivers a queued PR update on the current provider through %s (%s)", ([delivery, target]) => @@ -3141,17 +3142,18 @@ describe("orchestration v2 provider switching", () => { const capturedTurns = yield* Ref.make>([]); const started = yield* Deferred.make(); const release = yield* Deferred.make(); - const compatible = target === "compatible Codex"; - const broken = target === "broken current provider"; + const compatible = target === "compatible Codex" || target === "broken compatible Codex"; + const broken = + target === "broken current provider" || target === "broken compatible Codex"; const generation = yield* Ref.make(0); const selection = { - ...(broken - ? CODEX_MODEL_SELECTION - : compatible - ? { - ...CODEX_MODEL_SELECTION, - instanceId: ProviderInstanceId.make("codex-alternate"), - } + ...(compatible + ? { + ...CODEX_MODEL_SELECTION, + instanceId: ProviderInstanceId.make("codex-alternate"), + } + : broken + ? CODEX_MODEL_SELECTION : CLAUDE_MODEL_SELECTION), options: [{ id: "reasoningEffort", value: "low" }], }; @@ -3160,7 +3162,7 @@ describe("orchestration v2 provider switching", () => { ...CodexProviderCapabilitiesV2, context: { ...CodexProviderCapabilitiesV2.context, - canConsumeHandoffSummaries: false, + canConsumeHandoffSummaries: broken, }, } : ClaudeProviderCapabilitiesV2; @@ -3191,9 +3193,12 @@ describe("orchestration v2 provider switching", () => { modelSelection: selection, responseByRunOrdinal: {}, capturedTurns, + ...(broken && compatible + ? { brokenRunOrdinals: new Set(), nativeThreadGeneration: generation } + : {}), }), ]; - if (broken) adapters.pop(); + if (broken && !compatible) adapters.pop(); const layerRegistry = compatible ? Layer.succeed( ProviderAdapterRegistry.ProviderAdapterRegistryV2, @@ -3336,7 +3341,7 @@ describe("orchestration v2 provider switching", () => { assert.isFalse(executed.nativeThreadHasTurns); assert.include(executed.text, "Original work"); } - if (compatible) { + if (compatible && !broken) { assert.equal(executed.nativeThreadId, turns[0]!.nativeThreadId); assert.isEmpty(after.contextHandoffs); } From 22e0c58560f97aa0d40b6178e415e143b041c0d8 Mon Sep 17 00:00:00 2001 From: shivam <91240327+shivamhwp@users.noreply.github.com> Date: Tue, 10 Nov 2026 14:26:40 +0000 Subject: [PATCH 15/16] test(antigravity): drop the lost-signature tests another row already covers The queued PR update row for a broken current provider exercised only the handover status line that the broken compatible Codex row also needs, so it goes. The remaining rows lose setup they did not need. Co-Authored-By: Claude Fable 5.2 --- .../Adapters/AcpAdapterV2.test.ts | 10 ------ .../ProviderTurnStartService.test.ts | 22 +++---------- .../ProviderSwitch.integration.test.ts | 31 ++++++------------- 3 files changed, 14 insertions(+), 49 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts index 6920fee110e97..b2bafd6b8486b 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts @@ -1085,7 +1085,6 @@ describe("AcpAdapterV2", () => { return assert.fail("expected a failed terminal"); } assert.deepEqual(failed.failure, quota); - assert.equal(failed.threadDisposition, "reusable"); const completed = (yield* runTurn(2)).at(-1); assert.equal(completed?.type === "turn.terminal" ? completed.status : null, "completed"); assert.deepEqual(promptRuntimes, [1, 2], "the next turn runs on a fresh runtime"); @@ -1146,15 +1145,6 @@ describe("AcpAdapterV2", () => { interrupted?.type === "turn.terminal" ? interrupted.threadDisposition : null, "broken", ); - const interruptedThread = interruptedEvents.findLast( - (event) => event.type === "provider_thread.updated", - ); - assert.equal( - interruptedThread?.type === "provider_thread.updated" - ? interruptedThread.providerThread.status - : null, - "error", - ); }).pipe(Effect.provide(layerTest), Effect.scoped), ); it.live("retains returned usage when Stop finalizes before the prompt callback", () => diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts index c823a0af4953b..d222c898b89a0 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts @@ -1797,23 +1797,11 @@ effectIt.effect.each(["not_loaded", "error"] as const)( 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.id === "new-provider-thread" + ? { ...thread, status: providerThreadStatus } : thread, ), }); @@ -1842,9 +1830,9 @@ effectIt.effect.each(["not_loaded", "error"] as const)( ]); expect(projection.providerTurns).toEqual([]); expect(projection.checkpoints).toEqual([]); - expect(projection.providerThreads.find((thread) => thread.id === target.id)?.status).toBe( - providerThreadStatus, - ); + expect( + projection.providerThreads.find((thread) => thread.id === "new-provider-thread")?.status, + ).toBe(providerThreadStatus); }), ); diff --git a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts index 9076f1056cc59..c0fa679abaac3 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts @@ -2836,11 +2836,6 @@ describe("orchestration v2 provider switching", () => { ? { failedRunOrdinals: new Set([3]), brokenRunOrdinals: new Set([3]), - failure: makeProviderFailure({ - message: "Invalid thinking signature", - class: "provider_error", - code: "INVALID_THINKING_SIGNATURE", - }), } : {}), }), @@ -3131,7 +3126,6 @@ describe("orchestration v2 provider switching", () => { it.live.each([ ["automatic", "Claude"], ["queue.resume", "compatible Codex"], - ["queue.resume", "broken current provider"], ["queue.resume", "broken compatible Codex"], ] as const)( "delivers a queued PR update on the current provider through %s (%s)", @@ -3142,19 +3136,13 @@ describe("orchestration v2 provider switching", () => { const capturedTurns = yield* Ref.make>([]); const started = yield* Deferred.make(); const release = yield* Deferred.make(); - const compatible = target === "compatible Codex" || target === "broken compatible Codex"; - const broken = - target === "broken current provider" || target === "broken compatible Codex"; + const compatible = target !== "Claude"; + const broken = target === "broken compatible Codex"; const generation = yield* Ref.make(0); const selection = { ...(compatible - ? { - ...CODEX_MODEL_SELECTION, - instanceId: ProviderInstanceId.make("codex-alternate"), - } - : broken - ? CODEX_MODEL_SELECTION - : CLAUDE_MODEL_SELECTION), + ? { ...CODEX_MODEL_SELECTION, instanceId: ProviderInstanceId.make("codex-alternate") } + : CLAUDE_MODEL_SELECTION), options: [{ id: "reasoningEffort", value: "low" }], }; const targetCapabilities = compatible @@ -3181,9 +3169,9 @@ describe("orchestration v2 provider switching", () => { ? { brokenRunOrdinals: new Set([1]), nativeThreadGeneration: generation } : {}), failure: makeProviderFailure({ - message: broken ? "Saved history is unusable" : "Usage limit reached", - class: broken ? "provider_error" : "usage_limit", - code: broken ? "INVALID_THINKING_SIGNATURE" : "usage_limit", + message: "Usage limit reached", + class: "usage_limit", + code: "usage_limit", }), }), makeTestAdapter({ @@ -3193,12 +3181,11 @@ describe("orchestration v2 provider switching", () => { modelSelection: selection, responseByRunOrdinal: {}, capturedTurns, - ...(broken && compatible + ...(broken ? { brokenRunOrdinals: new Set(), nativeThreadGeneration: generation } : {}), }), ]; - if (broken && !compatible) adapters.pop(); const layerRegistry = compatible ? Layer.succeed( ProviderAdapterRegistry.ProviderAdapterRegistryV2, @@ -3322,7 +3309,7 @@ describe("orchestration v2 provider switching", () => { const turns = yield* Ref.get(capturedTurns); assert.lengthOf(turns, delivery === "automatic" ? 3 : 2); const executed = turns.at(-1)!; - assert.equal(executed.driver, compatible || broken ? CODEX_DRIVER : CLAUDE_DRIVER); + assert.equal(executed.driver, compatible ? CODEX_DRIVER : CLAUDE_DRIVER); assert.include(executed.text, "Review the new PR comments"); assert.deepEqual(executed.modelSelection, selection); assert.deepEqual(after.thread.modelSelection, selection); From 5ca0848d7c91de2211d02de7dffefcb99e2daf92 Mon Sep 17 00:00:00 2001 From: shivam <91240327+shivamhwp@users.noreply.github.com> Date: Sun, 15 Nov 2026 14:22:50 +0000 Subject: [PATCH 16/16] fix(antigravity): a restarted server no longer resumes an unusable session before replacing it After a server restart, the first message to a thread whose native history was marked unusable opened its session with the broken native id, resumed it, and only then started the replacement. The session now opens without that id and recovery uses it, so the runtime starts once. Co-Authored-By: Claude Opus 5.5 Co-Authored-By: Claude Fable 5.2 --- .../ProviderTurnStartService.ts | 13 +++++-- .../ProviderSwitch.integration.test.ts | 36 +++++++++++++++++-- 2 files changed, 45 insertions(+), 4 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index b1eb2657680fb..fa5409adfa21c 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -785,6 +785,12 @@ export const layer: Layer.Layer< run, checkpointScope, }); + const brokenNativeThread = providerThread.status === "error"; + // Opened without the broken native id, a session that was not live starts fresh. + const opensFreshSession = + brokenNativeThread && + providerThread.nativeThreadRef !== null && + !(yield* providerSessions.isLive(providerSessionId)); const sessionResult = yield* Effect.result( providerSessions.open({ threadId: projection.thread.id, @@ -794,7 +800,7 @@ export const layer: Layer.Layer< ...(existingSessionProjection === undefined ? {} : { resumeFromSession: existingSessionProjection }), - ...(providerThread.nativeThreadRef?.nativeId == null + ...(providerThread.nativeThreadRef?.nativeId == null || brokenNativeThread ? {} : { initialNativeThreadId: providerThread.nativeThreadRef.nativeId }), ...(providerThread.nativeMetadata?.itemIdentityVersion === undefined @@ -1432,7 +1438,10 @@ export const layer: Layer.Layer< providerSessionId, // Clear the native ref to force a fresh session while preserving row identity. existingProviderThread: { ...providerThread, nativeThreadRef: null }, - forceNewNativeThread: true, + // A fresh session that never loaded the broken history needs no replacement. + ...(opensFreshSession && resumed._tag === "Failure" + ? {} + : { forceNewNativeThread: true }), }), ); if (replacement === undefined) return undefined; diff --git a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts index e5887dc6bf75a..911aec5a48780 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts @@ -58,6 +58,7 @@ import * as EffectOutbox from "../EffectOutbox.ts"; import * as ProjectionReplay from "./ProjectionReplay.ts"; import * as ProjectionStore from "../ProjectionStore.ts"; import * as ProviderAdapterRegistry from "../ProviderAdapterRegistry.ts"; +import * as ProviderSessionManager from "../ProviderSessionManager.ts"; import { makeProviderFailure } from "@t3tools/provider-core/server/failure"; import { CLAUDE_MODEL_SELECTION, @@ -132,6 +133,8 @@ function makeTestAdapter(input: { readonly backgroundTasksEndWithProcess?: true; readonly backgroundWork?: Ref.Ref; readonly closedSessions?: Ref.Ref>; + /** Like ACP: opening resumes the initial native id, and forcing a new thread restarts the runtime. */ + readonly runtimeStarts?: Ref.Ref>; }): ProviderAdapter.ProviderAdapterV2["Service"] { return { instanceId: input.instanceId, @@ -141,6 +144,12 @@ function makeTestAdapter(input: { openSession: (sessionInput) => Effect.gen(function* () { const events = yield* PubSub.unbounded(); + if (input.runtimeStarts !== undefined) { + yield* Ref.update(input.runtimeStarts, (starts) => [ + ...starts, + sessionInput.initialNativeThreadId ?? "new", + ]); + } if (input.closedSessions !== undefined) { const closedSessions = input.closedSessions; yield* Effect.addFinalizer(() => @@ -194,6 +203,9 @@ function makeTestAdapter(input: { }), ensureThread: (threadInput) => Effect.gen(function* () { + if (threadInput.forceNewNativeThread && input.runtimeStarts !== undefined) { + yield* Ref.update(input.runtimeStarts, (starts) => [...starts, "replacement"]); + } const createdAt = yield* DateTime.now; const generation = input.nativeThreadGeneration === undefined @@ -3324,16 +3336,19 @@ describe("orchestration v2 provider switching", () => { "immediately", "behind a running turn", "and recovers it after it breaks", + "and recovers it after it breaks and the server restarts", ] as const)( "starts a fresh provider session %s and never hands earlier runs to a later provider", (variant) => Effect.gen(function* () { const queued = variant === "behind a running turn"; - const broken = variant === "and recovers it after it breaks"; - const key = queued ? "queued" : broken ? "broken" : "immediate"; + const restarted = variant === "and recovers it after it breaks and the server restarts"; + const broken = restarted || variant === "and recovers it after it breaks"; + const key = queued ? "queued" : restarted ? "restarted" : broken ? "broken" : "immediate"; const cwd = yield* checkpointWorkspace(`fresh-context-${key}`); const capturedTurns = yield* Ref.make>([]); const closedSessions = yield* Ref.make>([]); + const runtimeStarts = yield* Ref.make>([]); // A queued start has nobody to refuse, so a busy session must stay up. const backgroundWork = yield* Ref.make(queued); const started = yield* Deferred.make(); @@ -3347,6 +3362,7 @@ describe("orchestration v2 provider switching", () => { responseByRunOrdinal: {}, capturedTurns, closedSessions, + runtimeStarts, backgroundWork, nativeThreadGeneration: yield* Ref.make(0), ...(queued @@ -3451,7 +3467,18 @@ describe("orchestration v2 provider switching", () => { // Claude hosts one native thread per session, so an idle old one is closed. assert.deepEqual(yield* Ref.get(closedSessions), queued ? [] : [firstSession.id]); yield* Ref.set(backgroundWork, false); + let startsBeforeRecovery = 0; if (broken) { + if (restarted) { + const brokenProjection = yield* orchestrator.getThreadProjection(freshThreadId); + const brokenThread = brokenProjection.providerThreads.find( + (thread) => thread.id === brokenProjection.thread.activeProviderThreadId, + )!; + assert.equal(brokenThread.status, "error"); + const sessions = yield* ProviderSessionManager.ProviderSessionManagerV2; + yield* sessions.close(brokenThread.providerSessionId!); + } + startsBeforeRecovery = (yield* Ref.get(runtimeStarts)).length; yield* send("epsilon", CLAUDE_MODEL_SELECTION); yield* worker.drain(); yield* waitForIdle(freshThreadId); @@ -3491,6 +3518,11 @@ describe("orchestration v2 provider switching", () => { assert.include(recovered.text, "Prompt beta"); assert.notInclude(recovered.text, "Prompt alpha"); assert.notInclude(recovered.text, "Prompt zero"); + // The broken native id is never reopened; a reopened session starts fresh only once. + assert.deepEqual( + (yield* Ref.get(runtimeStarts)).slice(startsBeforeRecovery), + restarted ? ["new"] : ["replacement"], + ); const recoveredRun = projection.runs.find( (run) => run.userMessageId === `message:fresh-context:${key}:epsilon`, )!;