From 49f8a95e2e5a20e97eb0e5c19161c3f2f0c7fb36 Mon Sep 17 00:00:00 2001 From: Melih <115730239+EinMelih@users.noreply.github.com> Date: Sun, 18 Oct 2026 00:16:58 +0200 Subject: [PATCH] fix(orchestration): support Antigravity and non-Git conversation rollback and ACP connect indicator --- .../CheckpointRollbackService.ts | 92 +++++++---- .../src/orchestration-v2/CheckpointService.ts | 16 +- .../ProviderTurnStartService.ts | 143 +++++++++++++++++- .../src/provider/AntigravityProvider.test.ts | 1 - .../src/provider/AntigravityProvider.ts | 1 - packages/provider-acp/src/server/adapter.ts | 25 +-- 6 files changed, 222 insertions(+), 56 deletions(-) diff --git a/apps/server/src/orchestration-v2/CheckpointRollbackService.ts b/apps/server/src/orchestration-v2/CheckpointRollbackService.ts index 89726eee19f5b..57a9d729fe740 100644 --- a/apps/server/src/orchestration-v2/CheckpointRollbackService.ts +++ b/apps/server/src/orchestration-v2/CheckpointRollbackService.ts @@ -225,30 +225,6 @@ export const layer: Layer.Layer< }); } - const modelSelection = projection.thread.modelSelection; - const resolvedRuntimePolicy = yield* runtimePolicy.resolve({ - thread: projection.thread, - modelSelection, - }); - const existingSession = projection.providerSessions.find( - (candidate) => candidate.id === providerThread.providerSessionId, - ); - const session = yield* sessions.open({ - threadId: input.threadId, - providerSessionId: providerThread.providerSessionId, - modelSelection, - runtimePolicy: resolvedRuntimePolicy, - ...(existingSession === undefined ? {} : { resumeFromSession: existingSession }), - ...(providerThread.nativeThreadRef?.nativeId == null - ? {} - : { initialNativeThreadId: providerThread.nativeThreadRef.nativeId }), - ...(providerThread.nativeMetadata?.itemIdentityVersion === undefined - ? {} - : { - initialProviderItemIdentityVersion: providerThread.nativeMetadata.itemIdentityVersion, - }), - }); - const runsToRollback = projection.runs.filter( (run) => run.ordinal > targetOrdinal && @@ -273,14 +249,66 @@ export const layer: Layer.Layer< (turn.runAttemptId === null || !rolledBackAttemptIds.has(turn.runAttemptId)), ); - const snapshot = - runsToRollback.length === 0 - ? { providerThread } - : yield* session.rollbackThread({ - providerThread, - target: rollbackTarget, - providerThreadTurns, - }); + let snapshot: { readonly providerThread: typeof providerThread }; + if (providerThread.driver === "antigravity") { + if (runsToRollback.length === 0) { + snapshot = { providerThread }; + } else { + if ( + yield* Effect.try(() => sessions.isLive(providerThread.providerSessionId)).pipe( + Effect.flatten, + Effect.orElseSucceed(() => false), + ) + ) { + yield* sessions.close(providerThread.providerSessionId); + } + snapshot = { + providerThread: { + ...providerThread, + nativeThreadRef: null, + nativeConversationHeadRef: null, + nativeMetadata: { + ...providerThread.nativeMetadata, + itemIdentityVersion: 2, + }, + status: "idle", + }, + }; + } + } else { + const modelSelection = projection.thread.modelSelection; + const resolvedRuntimePolicy = yield* runtimePolicy.resolve({ + thread: projection.thread, + modelSelection, + }); + const existingSession = projection.providerSessions.find( + (candidate) => candidate.id === providerThread.providerSessionId, + ); + const session = yield* sessions.open({ + threadId: input.threadId, + providerSessionId: providerThread.providerSessionId, + modelSelection, + runtimePolicy: resolvedRuntimePolicy, + ...(existingSession === undefined ? {} : { resumeFromSession: existingSession }), + ...(providerThread.nativeThreadRef?.nativeId == null + ? {} + : { initialNativeThreadId: providerThread.nativeThreadRef.nativeId }), + ...(providerThread.nativeMetadata?.itemIdentityVersion === undefined + ? {} + : { + initialProviderItemIdentityVersion: + providerThread.nativeMetadata.itemIdentityVersion, + }), + }); + snapshot = + runsToRollback.length === 0 + ? { providerThread } + : yield* session.rollbackThread({ + providerThread, + target: rollbackTarget, + providerThreadTurns, + }); + } if (input.restoreFiles !== false) yield* checkpoints.restore({ scope, checkpoint }); const staleCheckpoints = projection.checkpoints.filter( (candidate) => diff --git a/apps/server/src/orchestration-v2/CheckpointService.ts b/apps/server/src/orchestration-v2/CheckpointService.ts index b49da7740b82c..59b90836b1860 100644 --- a/apps/server/src/orchestration-v2/CheckpointService.ts +++ b/apps/server/src/orchestration-v2/CheckpointService.ts @@ -329,7 +329,7 @@ export const layer: Layer.Layer< }).pipe(Effect.as(false)), ), ) - : false; + : true; return makeCheckpoint({ id: checkpointId, scope: input.scope, @@ -388,7 +388,7 @@ export const layer: Layer.Layer< ordinalWithinScope: input.ordinalWithinScope, appRunOrdinal: input.appRunOrdinal, ref: checkpointRef, - status: "missing", + status: "ready", files: [], capturedAt: input.capturedAt, }); @@ -519,6 +519,9 @@ export const layer: Layer.Layer< cause: `Checkpoint status is ${input.checkpoint.status}.`, }); } + if (!(yield* isGitCheckpointable(input.scope.cwd))) { + return; + } const restored = yield* checkpointStore.restoreCheckpoint({ cwd: input.scope.cwd, @@ -548,9 +551,12 @@ export const layer: Layer.Layer< const deleteStaleRefs: CheckpointServiceV2Shape["deleteStaleRefs"] = (input) => withWorkspaceLock( input.scope.cwd, - checkpointStore.deleteCheckpointRefs({ - cwd: input.scope.cwd, - checkpointRefs: input.checkpoints.map((checkpoint) => checkpoint.ref), + Effect.gen(function* () { + if (!(yield* isGitCheckpointable(input.scope.cwd))) return; + yield* checkpointStore.deleteCheckpointRefs({ + cwd: input.scope.cwd, + checkpointRefs: input.checkpoints.map((checkpoint) => checkpoint.ref), + }); }), ).pipe( Effect.mapError( diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index 36354f8c1f4ef..60dc3e9a02c7f 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -511,6 +511,98 @@ export const layer: Layer.Layer< const existingSessionProjection = projection.providerSessions.find( (candidate) => candidate.id === providerSessionId, ); + const isColdOrRestartingAcpConnect = + providerThread.driver === "antigravity" + ? !(yield* Effect.try(() => providerSessions.isLive(providerSessionId)).pipe( + Effect.flatten, + Effect.orElseSucceed(() => true), + )) || + providerThread.nativeThreadRef === null || + projection.runs.findLast( + (candidate) => + candidate.ordinal < run.ordinal && + candidate.providerThreadId === providerThread.id, + )?.status === "interrupted" + : false; + const connectStartedAt = isColdOrRestartingAcpConnect ? yield* DateTime.now : null; + const connectTurnItem: Extract< + OrchestrationV2TurnItem, + { readonly type: "dynamic_tool" } + > | null = + connectStartedAt === null + ? null + : { + id: idAllocator.derive.runSignalTurnItem({ + runId: run.id, + signal: "provider-connect", + }), + threadId: projection.thread.id, + runId: run.id, + nodeId: rootNode.id, + providerThreadId: providerThread.id, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: + Math.max( + 0, + ...projection.turnItems + .filter((item) => item.runId === run.id) + .map((item) => item.ordinal), + ) + 1, + status: "running", + title: "Connecting to ACP…", + startedAt: connectStartedAt, + completedAt: null, + updatedAt: connectStartedAt, + type: "dynamic_tool", + toolName: "acp_connect", + input: {}, + }; + if (connectTurnItem !== null) { + yield* eventSink.writeIfRunCurrent({ + threadId: projection.thread.id, + runId: run.id, + activeAttemptId: attempt.id, + expectedStatus: "starting", + events: [ + { + id: yield* idAllocator.allocate.event({ threadId: projection.thread.id }), + type: "turn-item.updated", + threadId: projection.thread.id, + runId: run.id, + nodeId: rootNode.id, + providerInstanceId: run.providerInstanceId, + occurredAt: connectStartedAt!, + payload: connectTurnItem, + }, + ], + }); + } + const dismissConnectTurnItem = (completedAt: DateTime.Utc) => + connectTurnItem === null + ? Effect.void + : Effect.gen(function* () { + yield* eventSink.write({ + events: [ + { + id: yield* idAllocator.allocate.event({ threadId: projection.thread.id }), + type: "turn-item.updated", + threadId: projection.thread.id, + runId: run.id, + nodeId: rootNode.id, + providerInstanceId: run.providerInstanceId, + occurredAt: completedAt, + payload: { + ...connectTurnItem, + status: "idle", + completedAt, + updatedAt: completedAt, + }, + }, + ], + }); + }); const sessionResult = yield* Effect.result( providerSessions.open({ threadId: projection.thread.id, @@ -537,11 +629,13 @@ export const layer: Layer.Layer< readonly error: Error; }) => Effect.gen(function* () { + const now = yield* DateTime.now; + yield* dismissConnectTurnItem(now); const nestedCause = "cause" in failed.error ? failed.error.cause : undefined; yield* settleRunBeforeStart({ signal: failed.signal, status: "failed", - now: yield* DateTime.now, + now, providerInstanceId: run.providerInstanceId, itemProviderThreadId: providerThread.id, item: { @@ -561,7 +655,10 @@ export const layer: Layer.Layer< }); }); if (sessionResult._tag === "Failure") { - if (input.willRetry === true) return yield* sessionResult.failure; + if (input.willRetry === true) { + yield* dismissConnectTurnItem(yield* DateTime.now); + return yield* sessionResult.failure; + } yield* settleStartFailure({ signal: "provider-session-open-failure", title: "Provider session failed to open", @@ -576,7 +673,10 @@ export const layer: Layer.Layer< Effect.gen(function* () { const loaded = yield* Effect.result(load); if (loaded._tag === "Success") return loaded.success; - if (input.willRetry === true) return yield* loaded.failure; + if (input.willRetry === true) { + yield* dismissConnectTurnItem(yield* DateTime.now); + return yield* loaded.failure; + } yield* settleStartFailure({ signal: "provider-thread-load-failure", title: "Provider turn failed to start", @@ -705,12 +805,15 @@ export const layer: Layer.Layer< toProviderInstanceId: run.providerInstanceId, coveredRunOrdinals: { from: 1, to: Math.max(1, run.ordinal - 1) }, strategy: "full_thread_summary", - runs: projection.runs, + runs: projection.runs.filter((source) => source.status !== "rolled_back"), items: (yield* projectionStore.getTurnStartHistory(input.threadId)).filter( (item) => item.runId === null || projection.runs.some( - (source) => source.id === item.runId && source.ordinal < run.ordinal, + (source) => + source.id === item.runId && + source.ordinal < run.ordinal && + source.status !== "rolled_back", ), ), createdAt, @@ -759,6 +862,7 @@ export const layer: Layer.Layer< }); if (loadedProviderThread === undefined) return; if (!(yield* isCurrentAttemptInStatus("starting"))) { + yield* dismissConnectTurnItem(yield* DateTime.now); return; } const now = yield* DateTime.now; @@ -831,6 +935,25 @@ export const layer: Layer.Layer< startedAt: now, }; const events: Array = [ + ...(connectTurnItem === null + ? [] + : [ + { + id: yield* idAllocator.allocate.event({ threadId: projection.thread.id }), + type: "turn-item.updated" as const, + threadId: projection.thread.id, + runId: run.id, + nodeId: rootNode.id, + providerInstanceId: run.providerInstanceId, + occurredAt: now, + payload: { + ...connectTurnItem, + status: "idle" as const, + completedAt: now, + updatedAt: now, + }, + }, + ]), { id: yield* idAllocator.allocate.event({ threadId: projection.thread.id, @@ -917,6 +1040,7 @@ export const layer: Layer.Layer< events, }); if (!runningWrite.committed) { + yield* dismissConnectTurnItem(now); return; } const routableSubagents = projection.subagents.filter((subagent) => @@ -1048,8 +1172,13 @@ export const layer: Layer.Layer< (source) => source.ordinal < run.ordinal && source.providerThreadId === providerThread.id && - (source.status === "failed" || source.status === "interrupted") && - !deliveredAttemptIds.has(source.activeAttemptId), + source.status !== "rolled_back" && + (((source.status === "failed" || source.status === "interrupted") && + !deliveredAttemptIds.has(source.activeAttemptId)) || + (runningProviderThread.nativeThreadRef?.nativeId != null && + !nativeInputRunIds.has(source.id) && + !legacyInputRunIds.has(source.id) && + !legacyRecoveredRunIds.has(source.id))), ); const missedRunIds = new Set(missedRuns.map((source) => source.id)); const missedItems = diff --git a/apps/server/src/provider/AntigravityProvider.test.ts b/apps/server/src/provider/AntigravityProvider.test.ts index 407c1b428754b..fd2d98904711f 100644 --- a/apps/server/src/provider/AntigravityProvider.test.ts +++ b/apps/server/src/provider/AntigravityProvider.test.ts @@ -248,7 +248,6 @@ it.layer(layerTest)("Antigravity provider snapshots", (it) => { models: [], setup: { canAuthenticate: true, canInstall: true }, showInteractionModeToggle: false, - supportsConversationRollback: false, supportsTextGeneration: false, }); expect(yield* Ref.get(harness.probeCalls)).toBe(0); diff --git a/apps/server/src/provider/AntigravityProvider.ts b/apps/server/src/provider/AntigravityProvider.ts index a136d861b0be4..10377a993c7a7 100644 --- a/apps/server/src/provider/AntigravityProvider.ts +++ b/apps/server/src/provider/AntigravityProvider.ts @@ -152,7 +152,6 @@ export const makeAntigravityProvider = Effect.fn("makeAntigravityProvider")(func }, }), setup: { canAuthenticate: true, canInstall: true }, - supportsConversationRollback: false, supportsTextGeneration: false, workspaceSnapshots: [], } satisfies ServerProviderDraft; diff --git a/packages/provider-acp/src/server/adapter.ts b/packages/provider-acp/src/server/adapter.ts index 6bb8555bf99f3..94544e82b9896 100644 --- a/packages/provider-acp/src/server/adapter.ts +++ b/packages/provider-acp/src/server/adapter.ts @@ -6842,16 +6842,21 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( detail: "ACP runtime did not produce a session id", }); } - const providerThread = makeProviderThread({ - driver, - providerInstanceId: options.instanceId, - idAllocator, - appThreadId: threadInput.threadId, - providerSessionId: input.providerSessionId, - nativeThreadId: sessionId, - ...(itemIdentityVersion === undefined ? {} : { itemIdentityVersion }), - now, - }); + const providerThread = { + ...makeProviderThread({ + driver, + providerInstanceId: options.instanceId, + idAllocator, + appThreadId: threadInput.threadId, + providerSessionId: input.providerSessionId, + nativeThreadId: sessionId, + ...(itemIdentityVersion === undefined ? {} : { itemIdentityVersion }), + now, + }), + ...(threadInput.existingProviderThread === undefined + ? {} + : { id: threadInput.existingProviderThread.id }), + }; yield* Ref.update(providerThreadByNativeSessionId, (current) => new Map(current).set(sessionId, providerThread), );