diff --git a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts index dff370bd61101..6d213d12285ee 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts @@ -885,6 +885,10 @@ describe("AcpAdapterV2", () => { 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 = { @@ -894,6 +898,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 = []; @@ -906,8 +916,20 @@ 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; + const interrupted = watchedSessions.length === 4; + return () => + interrupted + ? Deferred.succeed(checking, undefined).pipe( + Effect.andThen(Deferred.await(releaseCheck)), + Effect.as(outcome), + ) + : Effect.succeed(outcome); }), makeRuntime: makeMockRuntime({ childProcessSpawner: yield* ChildProcessSpawner.ChildProcessSpawner, @@ -916,6 +938,7 @@ describe("AcpAdapterV2", () => { ), wrapRuntime: (runtime, runtimeOrdinal) => ({ ...runtime, + cancel: Deferred.succeed(cancelled, undefined).pipe(Effect.asVoid), prompt: () => Effect.sync(() => { promptRuntimes.push(runtimeOrdinal); @@ -954,23 +977,89 @@ 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", + ); + + 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`), + }), + }), + ); + 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", () => 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 d40ca624a11d2..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, @@ -2185,7 +2163,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( @@ -7264,7 +7242,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/ProviderTurnStartService.test.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts index d1b5168258874..c823a0af4953b 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts @@ -1791,11 +1791,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; @@ -1821,6 +1842,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 54b16d7aa2aef..ea3dca6f52dca 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -626,7 +626,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, }, }); @@ -1074,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, @@ -1082,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 e9b2636998386..bf1edfe58bac1 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( @@ -2767,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>([]); @@ -2795,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, @@ -2888,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); @@ -2916,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"); @@ -3061,6 +3131,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]) => @@ -3071,10 +3142,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 @@ -3097,10 +3175,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({ @@ -3112,6 +3193,7 @@ describe("orchestration v2 provider switching", () => { capturedTurns, }), ]; + if (broken) adapters.pop(); const layerRegistry = compatible ? Layer.succeed( ProviderAdapterRegistry.ProviderAdapterRegistryV2, @@ -3235,7 +3317,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); @@ -3249,6 +3331,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); diff --git a/apps/server/src/provider/acp/AntigravityTurnOutcome.test.ts b/apps/server/src/provider/acp/AntigravityTurnOutcome.test.ts index 09b8730bd6001..475f7bebfd713 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,29 @@ 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 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 +103,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 +129,40 @@ 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, ); // A completed tool keeps its usage, and the turn still fails. assert.equal( @@ -142,7 +177,7 @@ describe("watchAntigravityPromptOutcome", () => { stopReason: "end_turn", usage: { inputTokens: 40, outputTokens: 5, totalTokens: 45 }, }, - ))?.class, + ))?.failure.class, "usage_limit", ); assert.isUndefined( @@ -152,12 +187,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 +231,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 41e25f2e417e1..2cbe4d62b3a4b 100644 --- a/apps/server/src/provider/acp/AntigravityTurnOutcome.ts +++ b/apps/server/src/provider/acp/AntigravityTurnOutcome.ts @@ -20,6 +20,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. @@ -106,7 +110,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()) @@ -115,30 +124,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 678cd7edf2296..edcc6c2890218 100644 --- a/packages/provider-acp/src/server/adapter.ts +++ b/packages/provider-acp/src/server/adapter.ts @@ -294,6 +294,11 @@ export function acpPerTurnPromptUsage( }; } +interface AcpPromptOutcomeFailure { + readonly failure: OrchestrationV2ProviderFailure; + readonly threadDisposition: "reusable" | "broken"; +} + export interface AcpAdapterV2Flavor { readonly buildPrompt?: (input: { readonly text: string; @@ -308,14 +313,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 the next turn falls back to a fresh session. */ readonly watchPromptOutcome?: ( sessionId: string, ) => Effect.Effect< - ( - result: EffectAcpSchema.PromptResponse, - ) => Effect.Effect + (result: EffectAcpSchema.PromptResponse) => Effect.Effect >; readonly requiresSignIn?: (cause: unknown) => boolean; /** @@ -7571,10 +7575,14 @@ 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 = options?.threadDisposition ?? "reusable"; context.finalizedStatus = settledStatus; context.finalized = true; const watchdog = context.inactivityWatchdog; @@ -7658,7 +7666,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, @@ -7702,7 +7710,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( retry: context.providerRetry.retry, retryStartedAt: context.providerRetry.startedAt, }), - threadDisposition: "reusable", + threadDisposition, } : { type: "turn.terminal", @@ -7712,7 +7720,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( @@ -8396,7 +8404,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" @@ -8419,6 +8428,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; } @@ -8683,6 +8695,13 @@ 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* new ProviderAdapter.ProviderAdapterProtocolError({ + driver, + detail: "The native session history is unusable.", + }); + } return yield* runtimeTransitionPermit.withPermit( Effect.gen(function* () { yield* awaitRuntimeTeardown();