diff --git a/apps/server/integration/orchestrationEngine.integration.test.ts b/apps/server/integration/orchestrationEngine.integration.test.ts index a577fc59ebc2..92b7ed4c11c4 100644 --- a/apps/server/integration/orchestrationEngine.integration.test.ts +++ b/apps/server/integration/orchestrationEngine.integration.test.ts @@ -888,7 +888,7 @@ it.live("reverts to an earlier checkpoint and trims checkpoint projections + git ); it.live( - "appends checkpoint.revert.failed activity when revert is requested without an active session", + "appends checkpoint.revert.failed activity when revert has no persisted provider binding", () => withHarness((harness) => Effect.gen(function* () { @@ -917,7 +917,7 @@ it.live( assert.equal( String( (failureActivity?.payload as { readonly detail?: string } | undefined)?.detail, - ).includes("No active provider session"), + ).includes("no persisted provider binding"), true, ); }), diff --git a/apps/server/integration/orphanedProviderSessionStartup.integration.test.ts b/apps/server/integration/orphanedProviderSessionStartup.integration.test.ts index e07681ae153b..cc2319bcdeb6 100644 --- a/apps/server/integration/orphanedProviderSessionStartup.integration.test.ts +++ b/apps/server/integration/orphanedProviderSessionStartup.integration.test.ts @@ -120,7 +120,7 @@ const startupDependencies = Layer.mergeAll( stopSession: () => Effect.die("unused"), listSessions: () => Effect.succeed([]), getCapabilities: () => Effect.die("unused"), - assertConversationRollbackSupported: () => Effect.die("unused"), + prepareConversationRollback: () => Effect.die("unused"), getInstanceInfo: () => Effect.die("unused"), rollbackConversation: () => Effect.die("unused"), uploadFeedback: () => Effect.die("unused"), diff --git a/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts b/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts index b88f7f012d46..f4e44b450a7c 100644 --- a/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts +++ b/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts @@ -94,9 +94,9 @@ function createProviderServiceHarness( const rollbackConversation = vi.fn( (_input: { readonly threadId: ThreadId; readonly numTurns: number }) => Effect.void, ); - const assertConversationRollbackSupported = vi.fn< - ProviderServiceShape["assertConversationRollbackSupported"] - >(() => Effect.void); + const prepareConversationRollback = vi.fn( + () => Effect.void, + ); const unsupported = () => Effect.die(new Error("Unsupported provider call in test")) as Effect.Effect; @@ -124,7 +124,7 @@ function createProviderServiceHarness( stopSession: () => unsupported(), listSessions, getCapabilities: () => Effect.succeed({ sessionModelSwitch: "in-session" }), - assertConversationRollbackSupported, + prepareConversationRollback, getInstanceInfo: (instanceId) => Effect.succeed({ instanceId, @@ -149,7 +149,10 @@ function createProviderServiceHarness( return { service, - assertConversationRollbackSupported, + setSessionActive: (active: boolean) => { + hasSession = active; + }, + prepareConversationRollback, rollbackConversation, emit, }; @@ -1608,12 +1611,12 @@ describe("CheckpointReactor", () => { const threadId = ThreadId.make("thread-1"); const createdAt = "2026-01-01T00:00:00.000Z"; const checked = yield* Deferred.make(); - harness.provider.assertConversationRollbackSupported.mockImplementation(() => + harness.provider.prepareConversationRollback.mockImplementation(() => Deferred.succeed(checked, undefined).pipe( Effect.andThen( Effect.fail( new ProviderValidationError({ - operation: "ProviderService.assertConversationRollbackSupported", + operation: "ProviderService.prepareConversationRollback", issue: "Provider 'antigravity' does not support conversation rewind.", }), ), @@ -1683,6 +1686,111 @@ describe("CheckpointReactor", () => { }), ); + effectIt.effect.each([ + { provider: "codex", turnCount: 0, recover: true }, + { provider: "codex", turnCount: 1, recover: true }, + { provider: "claudeAgent", turnCount: 0, recover: true }, + { provider: "claudeAgent", turnCount: 1, recover: true }, + { provider: "claudeAgent", turnCount: 1, recover: false }, + ])( + "prepares an inactive $provider session before reverting to $turnCount (recover=$recover)", + ({ provider, turnCount, recover }) => + Effect.gen(function* () { + const harness = yield* Effect.promise(() => + createHarness({ + hasSession: false, + providerName: ProviderDriverKind.make(provider), + }), + ); + const threadId = ThreadId.make("thread-1"); + const createdAt = "2026-01-01T00:00:00.000Z"; + yield* harness.engine.dispatch({ + type: "thread.session.set", + commandId: CommandId.make("cmd-recovery-ready-session"), + threadId, + session: { + threadId, + status: "ready", + providerName: provider, + runtimeMode: "approval-required", + activeTurnId: null, + lastError: null, + updatedAt: createdAt, + }, + createdAt, + }); + for (const checkpointTurnCount of [1, 2]) { + yield* harness.engine.dispatch({ + type: "thread.turn.diff.complete", + commandId: CommandId.make(`cmd-recovery-diff-${checkpointTurnCount}`), + threadId, + turnId: asTurnId(`turn-${checkpointTurnCount}`), + completedAt: createdAt, + checkpointRef: checkpointRefForThreadTurn(threadId, checkpointTurnCount), + status: "ready", + files: [], + checkpointTurnCount, + createdAt, + }); + } + const before = (yield* Effect.promise(() => harness.readModel())).threads[0]; + const prepared = yield* Deferred.make(); + harness.provider.prepareConversationRollback.mockImplementation(() => + Deferred.succeed(prepared, undefined).pipe( + Effect.andThen( + recover + ? Effect.sync(() => harness.provider.setSessionActive(true)) + : Effect.fail( + new ProviderValidationError({ + operation: "ProviderService.prepareConversationRollback", + issue: "Persisted session could not be resumed.", + }), + ), + ), + ), + ); + yield* harness.engine.dispatch({ + type: "thread.checkpoint.revert", + commandId: CommandId.make("cmd-recovery-revert"), + threadId, + turnCount, + createdAt, + }); + yield* Deferred.await(prepared); + yield* Effect.promise(() => harness.drain()); + const after = (yield* Effect.promise(() => harness.readModel())).threads[0]; + if (recover) { + expect(after?.checkpoints).toHaveLength(turnCount); + expect( + after?.activities.some((activity) => activity.kind === "checkpoint.revert.failed"), + ).toBe(false); + expect(harness.provider.rollbackConversation).toHaveBeenCalledWith({ + threadId, + numTurns: 2 - turnCount, + }); + expect(NodeFS.readFileSync(NodePath.join(harness.cwd, "README.md"), "utf8")).toBe( + turnCount === 0 ? "v1\n" : "v2\n", + ); + expect(gitRefExists(harness.cwd, checkpointRefForThreadTurn(threadId, 2))).toBe(false); + } else { + expect(after?.checkpoints).toEqual(before?.checkpoints); + expect(after?.messages).toEqual(before?.messages); + expect(after?.latestTurn).toEqual(before?.latestTurn); + expect(after?.activities).toContainEqual( + expect.objectContaining({ + kind: "checkpoint.revert.failed", + payload: expect.objectContaining({ + detail: expect.stringContaining("Persisted session could not be resumed."), + }), + }), + ); + expect(harness.provider.rollbackConversation).not.toHaveBeenCalled(); + expect(NodeFS.readFileSync(NodePath.join(harness.cwd, "README.md"), "utf8")).toBe("v3\n"); + expect(gitRefExists(harness.cwd, checkpointRefForThreadTurn(threadId, 2))).toBe(true); + } + }), + ); + it("executes provider revert and emits thread.reverted for checkpoint revert requests", async () => { const harness = await createHarness(); const createdAt = "2026-01-01T00:00:00.000Z"; @@ -1916,7 +2024,7 @@ describe("CheckpointReactor", () => { }); }); - it("appends an error activity when revert is requested without an active session", async () => { + it("appends an error activity when rollback preparation leaves no session cwd", async () => { const harness = await createHarness({ hasSession: false }); const createdAt = "2026-01-01T00:00:00.000Z"; @@ -1925,7 +2033,7 @@ describe("CheckpointReactor", () => { type: "thread.checkpoint.revert", commandId: CommandId.make("cmd-revert-no-session"), threadId: ThreadId.make("thread-1"), - turnCount: 1, + turnCount: 0, createdAt, }), ); diff --git a/apps/server/src/orchestration/Layers/CheckpointReactor.ts b/apps/server/src/orchestration/Layers/CheckpointReactor.ts index d4d6b9409808..08f95fc8c00e 100644 --- a/apps/server/src/orchestration/Layers/CheckpointReactor.ts +++ b/apps/server/src/orchestration/Layers/CheckpointReactor.ts @@ -699,26 +699,6 @@ const make = Effect.gen(function* () { return; } - const sessionRuntime = yield* resolveSessionRuntimeForThread(event.payload.threadId); - if (Option.isNone(sessionRuntime)) { - yield* appendRevertFailureActivity({ - threadId: event.payload.threadId, - turnCount: event.payload.turnCount, - detail: "No active provider session with workspace cwd is bound to this thread.", - createdAt: now, - }).pipe(Effect.catch(() => Effect.void)); - return; - } - if (!(yield* checkpointStore.isGitRepository(sessionRuntime.value.cwd))) { - yield* appendRevertFailureActivity({ - threadId: event.payload.threadId, - turnCount: event.payload.turnCount, - detail: "Checkpoints are unavailable because this project is not a git repository.", - createdAt: now, - }).pipe(Effect.catch(() => Effect.void)); - return; - } - const currentTurnCount = thread.checkpoints.reduce( (maxTurnCount, checkpoint) => Math.max(maxTurnCount, checkpoint.checkpointTurnCount), 0, @@ -751,7 +731,27 @@ const make = Effect.gen(function* () { return; } - yield* providerService.assertConversationRollbackSupported(event.payload.threadId); + yield* providerService.prepareConversationRollback(event.payload.threadId); + + const sessionRuntime = yield* resolveSessionRuntimeForThread(event.payload.threadId); + if (Option.isNone(sessionRuntime)) { + yield* appendRevertFailureActivity({ + threadId: event.payload.threadId, + turnCount: event.payload.turnCount, + detail: "No active provider session with workspace cwd is bound to this thread.", + createdAt: now, + }).pipe(Effect.catch(() => Effect.void)); + return; + } + if (!(yield* checkpointStore.isGitRepository(sessionRuntime.value.cwd))) { + yield* appendRevertFailureActivity({ + threadId: event.payload.threadId, + turnCount: event.payload.turnCount, + detail: "Checkpoints are unavailable because this project is not a git repository.", + createdAt: now, + }).pipe(Effect.catch(() => Effect.void)); + return; + } const restored = yield* checkpointStore.restoreCheckpoint({ cwd: sessionRuntime.value.cwd, diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts index d93fec5a3cf6..c3d687f4f52b 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts @@ -363,7 +363,7 @@ describe("ProviderCommandReactor", () => { Effect.succeed({ sessionModelSwitch: input?.sessionModelSwitch ?? "in-session", }), - assertConversationRollbackSupported: () => unsupported(), + prepareConversationRollback: () => unsupported(), getInstanceInfo: (instanceId) => { const raw = String(instanceId); const driverKind = ProviderDriverKind.make( diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts index 1094ab48b7ac..4b4a994a58be 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts @@ -126,7 +126,7 @@ function createProviderServiceHarness() { stopSession: () => unsupported(), listSessions: () => Effect.succeed([...runtimeSessions]), getCapabilities: () => Effect.succeed({ sessionModelSwitch: "in-session" }), - assertConversationRollbackSupported: () => unsupported(), + prepareConversationRollback: () => unsupported(), getInstanceInfo: (instanceId) => { const driverKind = ProviderDriverKind.make(String(instanceId)); return Effect.succeed({ diff --git a/apps/server/src/provider/Layers/ProviderService.test.ts b/apps/server/src/provider/Layers/ProviderService.test.ts index 1ce1a396796f..98e360cac94d 100644 --- a/apps/server/src/provider/Layers/ProviderService.test.ts +++ b/apps/server/src/provider/Layers/ProviderService.test.ts @@ -58,13 +58,15 @@ import { ProviderUnsupportedError, ProviderValidationError, ProviderWorkspaceMissingError, + ProviderSessionDirectoryPersistenceError, type ProviderAdapterError, } from "../Errors.ts"; import type { ProviderAdapterShape } from "../Services/ProviderAdapter.ts"; import * as ProviderAdapterRegistry from "../Services/ProviderAdapterRegistry.ts"; import * as ProviderService from "../Services/ProviderService.ts"; import * as ProviderSessionDirectory from "../Services/ProviderSessionDirectory.ts"; -import { makeProviderServiceLive } from "./ProviderService.ts"; +import { makeProviderServiceLive, type ProviderServiceLiveOptions } from "./ProviderService.ts"; +import * as McpProviderSession from "../../mcp/McpProviderSession.ts"; import * as ProviderEventLoggers from "./ProviderEventLoggers.ts"; import { ProviderSessionDirectoryLive } from "./ProviderSessionDirectory.ts"; import * as NodeServices from "@effect/platform-node/NodeServices"; @@ -416,6 +418,7 @@ const hasMetricSnapshot = ( function makeProviderServiceLayer( input: { readonly directory?: ProviderSessionDirectory.ProviderSessionDirectory["Service"]; + readonly liveOptions?: ProviderServiceLiveOptions; readonly supportsConversationRollback?: boolean; readonly analyticsLayer?: Layer.Layer; readonly registry?: ProviderAdapterRegistry.ProviderAdapterRegistry["Service"]; @@ -446,7 +449,7 @@ function makeProviderServiceLayer( const layer = it.layer( Layer.mergeAll( - makeProviderServiceLive().pipe( + makeProviderServiceLive(input.liveOptions).pipe( Layer.provide(NodeServices.layer), Layer.provide(providerAdapterLayer), Layer.provide(directoryLayer), @@ -1180,6 +1183,103 @@ antigravityInstanceRouting.layer("ProviderServiceLive instance-owned conversatio ); }); +for (const failure of ["persistence", "provider mismatch", "stop"] as const) { + const threadId = asThreadId(`thread-recovery-${failure}`); + const revoked: ThreadId[] = []; + const issued: ThreadId[] = []; + const recovery = makeProviderServiceLayer({ + directory: { + getBinding: () => + Effect.succeed( + Option.some({ + threadId, + provider: CODEX_DRIVER, + providerInstanceId: codexInstanceId, + resumeCursor: { threadId }, + runtimePayload: { cwd: fixtureCwd("recovery-cleanup") }, + }), + ), + upsert: () => + failure !== "provider mismatch" + ? Effect.fail( + new ProviderSessionDirectoryPersistenceError({ + operation: "upsert", + detail: "Cannot save recovered binding.", + }), + ) + : Effect.void, + getProvider: () => Effect.succeed(CODEX_DRIVER), + listThreadIds: () => Effect.succeed([]), + listBindings: () => Effect.succeed([]), + recordImportedTranscript: () => Effect.die("unused"), + }, + liveOptions: { + issueMcpCredential: ({ threadId, providerInstanceId, preview }) => + Effect.sync(() => { + issued.push(threadId); + return { + config: { + environmentId: EnvironmentId.make("test"), + threadId, + providerInstanceId, + providerSessionId: "test-session", + endpoint: "http://localhost/mcp", + authorizationHeader: "Bearer test-credential", + preview, + }, + }; + }), + revokeMcpCredential: (threadId) => + Effect.sync(() => { + revoked.push(threadId); + }), + }, + }); + recovery.layer(`ProviderServiceLive recovery cleanup after ${failure}`, (it) => { + it.effect("stops the failed recovery and revokes its MCP credential", () => + Effect.gen(function* () { + const provider = yield* ProviderService.ProviderService; + if (failure === "provider mismatch") { + const startSession = recovery.codex.startSession.getMockImplementation()!; + recovery.codex.startSession.mockImplementationOnce((input) => + startSession(input).pipe( + Effect.map((session) => ({ ...session, provider: CLAUDE_AGENT_DRIVER })), + ), + ); + } + if (failure === "stop") { + const stopSession = recovery.codex.stopSession.getMockImplementation()!; + recovery.codex.stopSession.mockImplementationOnce((threadId) => + stopSession(threadId).pipe( + Effect.andThen( + Effect.fail( + new ProviderAdapterRequestError({ + provider: CODEX_DRIVER, + method: "stopSession", + detail: "Stop acknowledgement failed.", + }), + ), + ), + ), + ); + } + const error = yield* Effect.flip(provider.prepareConversationRollback(threadId)); + assert.include( + error.message, + failure === "provider mismatch" + ? "Adapter/provider mismatch" + : "Cannot save recovered binding", + ); + assert.deepEqual(issued, [threadId]); + assert.equal(yield* recovery.codex.hasSession(threadId), false); + assert.deepEqual(revoked, [threadId]); + assert.equal(McpProviderSession.readMcpProviderSession(threadId), undefined); + assert.equal(recovery.codex.rollbackThread.mock.calls.length, 0); + }), + ); + }); +} + const unsupportedRollback = makeProviderServiceLayer({ supportsConversationRollback: false }); unsupportedRollback.layer("ProviderServiceLive unsupported rewind", (it) => { it.effect("rejects rewind without starting or changing the provider conversation", () => @@ -1202,9 +1302,7 @@ unsupportedRollback.layer("ProviderServiceLive unsupported rewind", (it) => { unsupportedRollback.codex.startSession.mockClear(); unsupportedRollback.codex.rollbackThread.mockClear(); - const preflightError = yield* Effect.flip( - provider.assertConversationRollbackSupported(threadId), - ); + const preflightError = yield* Effect.flip(provider.prepareConversationRollback(threadId)); const rollbackError = yield* Effect.flip( provider.rollbackConversation({ threadId, numTurns: 1 }), ); @@ -1515,6 +1613,13 @@ it.effect( yield* Effect.gen(function* () { const provider = yield* ProviderService.ProviderService; + yield* provider.prepareConversationRollback(startedSession.threadId); + const sessions = yield* provider.listSessions(); + assert.equal( + sessions.find((session) => session.threadId === startedSession.threadId)?.cwd, + fixtureCwd("project"), + ); + assert.equal(secondCodex.rollbackThread.mock.calls.length, 0); yield* provider.rollbackConversation({ threadId: startedSession.threadId, numTurns: 1, @@ -2514,8 +2619,14 @@ routing.layer("ProviderServiceLive routing", (it) => { routing.codex.startSession.mockClear(); routing.codex.rollbackThread.mockClear(); - yield* provider.assertConversationRollbackSupported(initial.threadId); - assert.equal(routing.codex.startSession.mock.calls.length, 0); + yield* provider.prepareConversationRollback(initial.threadId); + assert.equal(routing.codex.startSession.mock.calls.length, 1); + assert.equal(routing.codex.rollbackThread.mock.calls.length, 0); + const preparedSessions = yield* provider.listSessions(); + assert.equal( + preparedSessions.find((session) => session.threadId === initial.threadId)?.cwd, + fixtureCwd("project"), + ); yield* provider.rollbackConversation({ threadId: initial.threadId, diff --git a/apps/server/src/provider/Layers/ProviderService.ts b/apps/server/src/provider/Layers/ProviderService.ts index 5b9059bfa643..12741a868876 100644 --- a/apps/server/src/provider/Layers/ProviderService.ts +++ b/apps/server/src/provider/Layers/ProviderService.ts @@ -249,6 +249,8 @@ export interface ProviderServiceLiveOptions { * test see whether a credential was requested at all. */ readonly issueMcpCredential?: typeof McpSessionRegistry.issueActiveMcpCredential; + /** Same seam as `issueMcpCredential`, for observing credential revocation. */ + readonly revokeMcpCredential?: typeof McpSessionRegistry.revokeActiveMcpThread; } interface TurnAnalyticsMetadata { @@ -477,6 +479,8 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( ); const issueMcpCredential = options?.issueMcpCredential ?? McpSessionRegistry.issueActiveMcpCredential; + const revokeMcpCredential = + options?.revokeMcpCredential ?? McpSessionRegistry.revokeActiveMcpThread; const fileSystem = yield* FileSystem.FileSystem; const runtimeEventPubSub = yield* PubSub.unbounded(); const pendingCompactions = new Map(); @@ -899,7 +903,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( return credential; }); const clearMcpSession = (threadId: ThreadId) => - McpSessionRegistry.revokeActiveMcpThread(threadId).pipe( + revokeMcpCredential(threadId).pipe( Effect.tap(() => Effect.sync(() => McpProviderSession.clearMcpProviderSession(threadId))), ); @@ -1182,8 +1186,8 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( const persistedModelSelection = readPersistedModelSelection(input.binding.runtimePayload); yield* prepareMcpSession(input.binding.threadId, bindingInstanceId); - const resumed = yield* adapter - .startSession({ + return yield* Effect.gen(function* () { + const resumed = yield* adapter.startSession({ threadId: input.binding.threadId, provider: input.binding.provider, providerInstanceId: bindingInstanceId, @@ -1191,26 +1195,37 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( ...(persistedModelSelection ? { modelSelection: persistedModelSelection } : {}), ...(hasResumeCursor ? { resumeCursor: input.binding.resumeCursor } : {}), runtimeMode: input.binding.runtimeMode ?? "full-access", - }) - .pipe(Effect.onError(() => clearMcpSession(input.binding.threadId))); - if (resumed.provider !== adapter.provider) { - yield* clearMcpSession(input.binding.threadId); - return yield* toValidationError( - input.operation, - `Adapter/provider mismatch while recovering thread '${input.binding.threadId}'. Expected '${adapter.provider}', received '${resumed.provider}'.`, - ); - } + }); + if (resumed.provider !== adapter.provider) { + return yield* toValidationError( + input.operation, + `Adapter/provider mismatch while recovering thread '${input.binding.threadId}'. Expected '${adapter.provider}', received '${resumed.provider}'.`, + ); + } - yield* upsertSessionBinding( - { ...resumed, providerInstanceId: bindingInstanceId }, - input.binding.threadId, + yield* upsertSessionBinding( + { ...resumed, providerInstanceId: bindingInstanceId }, + input.binding.threadId, + ); + yield* analytics.record("provider.session.recovered", { + provider: resumed.provider, + strategy: "resume-thread", + hasResumeCursor: resumed.resumeCursor !== undefined, + }); + return { adapter, session: resumed } as const; + }).pipe( + Effect.onError(() => + adapter.stopSession(input.binding.threadId).pipe( + Effect.catchCause((cause) => + Effect.logWarning("Failed to stop a session after recovery failed", { + threadId: input.binding.threadId, + errorTag: causeErrorTag(cause), + }), + ), + Effect.ensuring(clearMcpSession(input.binding.threadId)), + ), + ), ); - yield* analytics.record("provider.session.recovered", { - provider: resumed.provider, - strategy: "resume-thread", - hasResumeCursor: resumed.resumeCursor !== undefined, - }); - return { adapter, session: resumed } as const; }).pipe( withMetrics({ counter: providerSessionsTotal, @@ -2075,19 +2090,26 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( const getInstanceInfo: ProviderServiceMethod<"getInstanceInfo"> = (instanceId) => registry.getInstanceInfo(instanceId); - const assertConversationRollbackSupported: ProviderServiceMethod<"assertConversationRollbackSupported"> = - Effect.fn("assertConversationRollbackSupported")(function* (threadId) { + const prepareConversationRollback: ProviderServiceMethod<"prepareConversationRollback"> = + Effect.fn("prepareConversationRollback")(function* (threadId) { const routed = yield* resolveRoutableSession({ threadId, - operation: "ProviderService.assertConversationRollbackSupported", + operation: "ProviderService.prepareConversationRollback", allowRecovery: false, }); if (routed.adapter.capabilities.supportsConversationRollback === false) { return yield* toValidationError( - "ProviderService.assertConversationRollbackSupported", + "ProviderService.prepareConversationRollback", `Provider '${routed.adapter.provider}' does not support conversation rewind.`, ); } + if (!routed.isActive) { + yield* resolveRoutableSession({ + threadId, + operation: "ProviderService.prepareConversationRollback", + allowRecovery: true, + }); + } }); const rollbackConversation: ProviderServiceMethod<"rollbackConversation"> = Effect.fn( @@ -2103,7 +2125,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( } let metricProvider = "unknown"; return yield* Effect.gen(function* () { - yield* assertConversationRollbackSupported(input.threadId); + yield* prepareConversationRollback(input.threadId); const routed = yield* resolveRoutableSession({ threadId: input.threadId, operation: "ProviderService.rollbackConversation", @@ -2263,7 +2285,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( listSessions, getCapabilities, getInstanceInfo, - assertConversationRollbackSupported, + prepareConversationRollback, rollbackConversation, uploadFeedback, // Each access creates a fresh PubSub subscription so that multiple diff --git a/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts b/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts index 2538e05e37dc..1e662e75d96a 100644 --- a/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts +++ b/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts @@ -199,7 +199,7 @@ describe("ProviderSessionReaper", () => { stopSession, listSessions: () => Effect.succeed([]), getCapabilities: () => Effect.succeed({ sessionModelSwitch: "in-session" }), - assertConversationRollbackSupported: () => unsupported(), + prepareConversationRollback: () => unsupported(), getInstanceInfo: (instanceId) => { const driverKind = ProviderDriverKind.make(String(instanceId)); return Effect.succeed({ diff --git a/apps/server/src/provider/Services/ProviderService.ts b/apps/server/src/provider/Services/ProviderService.ts index c189e2916ff1..73bc1e8ddc40 100644 --- a/apps/server/src/provider/Services/ProviderService.ts +++ b/apps/server/src/provider/Services/ProviderService.ts @@ -107,9 +107,9 @@ export interface ProviderServiceShape { ) => Effect.Effect; /** - * Reject unsupported rewind before files change, without resuming the session. + * Reject unsupported rewind, then recover its persisted session before files change. */ - readonly assertConversationRollbackSupported: ( + readonly prepareConversationRollback: ( threadId: ThreadId, ) => Effect.Effect; diff --git a/apps/server/src/serverRuntimeStartup.reconcile.test.ts b/apps/server/src/serverRuntimeStartup.reconcile.test.ts index 37fd210ee6da..f035ea0a6277 100644 --- a/apps/server/src/serverRuntimeStartup.reconcile.test.ts +++ b/apps/server/src/serverRuntimeStartup.reconcile.test.ts @@ -65,7 +65,7 @@ const makeProviderService = (liveThreadIds: ReadonlyArray = []) => stopSession: () => Effect.die("unused"), listSessions: () => Effect.succeed(liveThreadIds.map((threadId) => ({ threadId }) as never)), getCapabilities: () => Effect.die("unused"), - assertConversationRollbackSupported: () => Effect.die("unused"), + prepareConversationRollback: () => Effect.die("unused"), getInstanceInfo: () => Effect.die("unused"), rollbackConversation: () => Effect.die("unused"), uploadFeedback: () => Effect.die("unused"),