diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.test.ts b/apps/server/src/provider/Layers/ClaudeAdapter.test.ts index 759a1641093..8f579473176 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.test.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.test.ts @@ -4247,6 +4247,52 @@ describe("ClaudeAdapterLive", () => { ); }); + it.effect("rejects an echoed Claude resume cursor when the SDK starts a fresh session", () => { + const harness = makeHarness(); + return Effect.gen(function* () { + const adapter = yield* ClaudeAdapter; + const persistedResumeCursor = { + threadId: RESUME_THREAD_ID, + resume: "550e8400-e29b-41d4-a716-446655440000", + turnCount: 3, + }; + const session = yield* adapter.startSession({ + threadId: RESUME_THREAD_ID, + provider: ProviderDriverKind.make("claudeAgent"), + resumeCursor: persistedResumeCursor, + runtimeMode: "full-access", + }); + assert.notEqual(adapter.isSameResumeCursor, undefined); + if (adapter.isSameResumeCursor === undefined) return; + + const verification = yield* adapter + .isSameResumeCursor(RESUME_THREAD_ID, persistedResumeCursor, session.resumeCursor) + .pipe(Effect.forkChild); + harness.query.emit({ + type: "system", + subtype: "init", + apiKeySource: "none", + claude_code_version: "test", + cwd: "/tmp/claude-adapter-test", + tools: [], + mcp_servers: [], + model: SYNTHETIC_CLAUDE_STANDARD_MODEL, + permissionMode: "bypassPermissions", + slash_commands: [], + output_style: "default", + skills: [], + plugins: [], + session_id: "7368d0c7-40a3-4d8a-bcc1-ac80c49f2719", + uuid: "fresh-init", + } as unknown as SDKMessage); + + assert.equal(yield* Fiber.join(verification), false); + }).pipe( + Effect.provideService(Random.Random, makeDeterministicRandomService()), + Effect.provide(harness.layer), + ); + }); + it.effect("preserves durable resume ids across Claude resume hooks", () => { const harness = makeHarness(); return Effect.gen(function* () { @@ -4259,7 +4305,7 @@ describe("ClaudeAdapterLive", () => { Effect.forkChild, ); - yield* adapter.startSession({ + const session = yield* adapter.startSession({ threadId: RESUME_THREAD_ID, provider: ProviderDriverKind.make("claudeAgent"), resumeCursor: { @@ -4331,6 +4377,17 @@ describe("ClaudeAdapterLive", () => { } | undefined; assert.equal(resumeCursor?.resume, durableSessionId); + assert.notEqual(adapter.isSameResumeCursor, undefined); + if (adapter.isSameResumeCursor !== undefined) { + assert.equal( + yield* adapter.isSameResumeCursor( + RESUME_THREAD_ID, + session.resumeCursor, + activeSessions[0]?.resumeCursor, + ), + true, + ); + } }).pipe( Effect.provideService(Random.Random, makeDeterministicRandomService()), Effect.provide(harness.layer), diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.ts b/apps/server/src/provider/Layers/ClaudeAdapter.ts index 18b5e996395..4d37e2ad1d2 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.ts @@ -69,6 +69,7 @@ import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; import * as FileSystem from "effect/FileSystem"; import * as Fiber from "effect/Fiber"; +import * as Option from "effect/Option"; import * as Path from "effect/Path"; import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; @@ -291,6 +292,7 @@ interface ClaudeSessionContext { * effort override inherit this. */ currentEffort: string | undefined; resumeSessionId: string | undefined; + readonly confirmedResumeSessionId: Deferred.Deferred; readonly pendingApprovals: Map; readonly pendingUserInputs: Map; readonly turns: Array<{ @@ -2047,6 +2049,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( const nextThreadId = message.session_id; context.resumeSessionId = message.session_id; yield* updateResumeCursor(context); + yield* Deferred.succeed(context.confirmedResumeSessionId, message.session_id); if (context.lastThreadStartedId !== nextThreadId) { context.lastThreadStartedId = nextThreadId; @@ -3886,6 +3889,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( const runPromise = Effect.runPromiseWith(runtimeContext); const promptQueue = yield* Queue.unbounded(); + const confirmedResumeSessionId = yield* Deferred.make(); const prompt = Stream.fromQueue(promptQueue).pipe( Stream.filter((item) => item.type === "message"), Stream.map((item) => item.message), @@ -4461,6 +4465,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( currentApiModelId: apiModelId, currentEffort: effectiveEffort ?? undefined, resumeSessionId: sessionId, + confirmedResumeSessionId, pendingApprovals, pendingUserInputs, turns: [], @@ -4804,6 +4809,24 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( sessionModelSwitch: "in-session", }, startSession, + isSameResumeCursor: (threadId, persisted, recovered) => + Effect.gen(function* () { + const persistedResume = readClaudeResumeState(persisted)?.resume; + const recoveredResume = readClaudeResumeState(recovered)?.resume; + const context = sessions.get(threadId); + if ( + persistedResume === undefined || + recoveredResume === undefined || + persistedResume !== recoveredResume || + context === undefined + ) { + return false; + } + const confirmed = yield* Deferred.await(context.confirmedResumeSessionId).pipe( + Effect.timeoutOption("10 seconds"), + ); + return Option.isSome(confirmed) && confirmed.value === persistedResume; + }), sendTurn, interruptTurn, readThread, diff --git a/apps/server/src/provider/Layers/CodexAdapter.ts b/apps/server/src/provider/Layers/CodexAdapter.ts index fa8511ee09a..58a1f352dfe 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.ts @@ -2324,6 +2324,12 @@ export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* ( promptlessTurnContinuation: true, }, startSession, + isSameResumeCursor: (_threadId, persisted, recovered) => + Effect.succeed( + isCodexResumeCursorSchema(persisted) && + isCodexResumeCursorSchema(recovered) && + persisted.threadId === recovered.threadId, + ), sendTurn, interruptTurn, readThread, diff --git a/apps/server/src/provider/Layers/CursorAdapter.ts b/apps/server/src/provider/Layers/CursorAdapter.ts index 50a3131ff95..057af2a44e4 100644 --- a/apps/server/src/provider/Layers/CursorAdapter.ts +++ b/apps/server/src/provider/Layers/CursorAdapter.ts @@ -1205,6 +1205,15 @@ export function makeCursorAdapter( provider: PROVIDER, capabilities: { sessionModelSwitch: "in-session" }, startSession, + isSameResumeCursor: (_threadId, persisted, recovered) => { + const persistedSession = parseCursorResume(persisted); + const recoveredSession = parseCursorResume(recovered); + return Effect.succeed( + persistedSession !== undefined && + recoveredSession !== undefined && + persistedSession.sessionId === recoveredSession.sessionId, + ); + }, sendTurn, interruptTurn, readThread, diff --git a/apps/server/src/provider/Layers/GrokAdapter.ts b/apps/server/src/provider/Layers/GrokAdapter.ts index da9bdf6030f..ee7de2fa078 100644 --- a/apps/server/src/provider/Layers/GrokAdapter.ts +++ b/apps/server/src/provider/Layers/GrokAdapter.ts @@ -2114,6 +2114,15 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte provider: PROVIDER, capabilities: { sessionModelSwitch: "in-session" }, startSession, + isSameResumeCursor: (_threadId, persisted, recovered) => { + const persistedSession = parseGrokResume(persisted); + const recoveredSession = parseGrokResume(recovered); + return Effect.succeed( + persistedSession !== undefined && + recoveredSession !== undefined && + persistedSession.sessionId === recoveredSession.sessionId, + ); + }, sendTurn, interruptTurn, readThread, diff --git a/apps/server/src/provider/Layers/OpenCodeAdapter.ts b/apps/server/src/provider/Layers/OpenCodeAdapter.ts index d0b4f0de78c..3e9bf8b9cfe 100644 --- a/apps/server/src/provider/Layers/OpenCodeAdapter.ts +++ b/apps/server/src/provider/Layers/OpenCodeAdapter.ts @@ -3267,6 +3267,15 @@ export function makeOpenCodeAdapter( sessionModelSwitch: "in-session", }, startSession, + isSameResumeCursor: (_threadId, persisted, recovered) => { + const persistedSession = parseOpenCodeResume(persisted); + const recoveredSession = parseOpenCodeResume(recovered); + return Effect.succeed( + persistedSession !== undefined && + recoveredSession !== undefined && + persistedSession.sessionId === recoveredSession.sessionId, + ); + }, sendTurn, interruptTurn, respondToRequest, diff --git a/apps/server/src/provider/Layers/ProviderService.test.ts b/apps/server/src/provider/Layers/ProviderService.test.ts index a81528e1814..17755d1dd03 100644 --- a/apps/server/src/provider/Layers/ProviderService.test.ts +++ b/apps/server/src/provider/Layers/ProviderService.test.ts @@ -246,6 +246,19 @@ function makeFakeCodexAdapter(provider: ProviderDriverKind = CODEX_DRIVER) { ...(provider === CODEX_DRIVER ? { promptlessTurnContinuation: true } : {}), }, startSession, + isSameResumeCursor: (_threadId, persisted, recovered) => { + const readOpaque = (value: unknown) => + typeof value === "object" && + value !== null && + !Array.isArray(value) && + typeof (value as { opaque?: unknown }).opaque === "string" + ? (value as { opaque: string }).opaque + : undefined; + const persistedOpaque = readOpaque(persisted); + return Effect.succeed( + persistedOpaque !== undefined && persistedOpaque === readOpaque(recovered), + ); + }, sendTurn, interruptTurn, respondToRequest, @@ -1464,11 +1477,14 @@ routing.layer("ProviderServiceLive routing", (it) => { routing.codex.startSession.mockClear(); routing.codex.sendTurn.mockClear(); - yield* provider.sendTurn({ - threadId: initial.threadId, - input: "resume", - attachments: [], - }); + yield* provider.sendTurn( + { + threadId: initial.threadId, + input: "resume", + attachments: [], + }, + { requireResumeCursor: initial.resumeCursor }, + ); assert.equal(routing.codex.startSession.mock.calls.length, 1); const resumedStartInput = routing.codex.startSession.mock.calls[0]?.[0]; @@ -1489,6 +1505,47 @@ routing.layer("ProviderServiceLive routing", (it) => { }), ); + it.effect("rejects automatic continuation when recovery starts a fresh conversation", () => + Effect.gen(function* () { + const provider = yield* ProviderService.ProviderService; + const threadId = asThreadId("thread-strict-resume-mismatch"); + const initial = yield* provider.startSession(threadId, { + provider: CODEX_DRIVER, + providerInstanceId: codexInstanceId, + threadId, + cwd: "/tmp/project-strict-resume", + runtimeMode: "full-access", + }); + + yield* routing.codex.stopAll(); + routing.codex.startSession.mockClear(); + routing.codex.sendTurn.mockClear(); + routing.codex.stopSession.mockClear(); + routing.codex.startSession.mockImplementationOnce(() => + Effect.succeed({ + ...initial, + resumeCursor: { opaque: "fresh-provider-conversation" }, + }), + ); + + const failure = yield* provider + .sendTurn( + { + threadId, + input: "Continue where you left off.", + attachments: [], + }, + { requireResumeCursor: initial.resumeCursor }, + ) + .pipe(Effect.flip); + + assert.instanceOf(failure, ProviderValidationError); + assert.match(failure.message, /did not resume the persisted conversation/u); + assert.equal(routing.codex.sendTurn.mock.calls.length, 0); + assert.deepStrictEqual(routing.codex.stopSession.mock.calls, [[threadId]]); + }), + ); + it.effect("recovers stale claudeAgent sessions for sendTurn using persisted cwd", () => Effect.gen(function* () { const provider = yield* ProviderService.ProviderService; diff --git a/apps/server/src/provider/Layers/ProviderService.ts b/apps/server/src/provider/Layers/ProviderService.ts index a75f2977d4d..b7b227d8cc8 100644 --- a/apps/server/src/provider/Layers/ProviderService.ts +++ b/apps/server/src/provider/Layers/ProviderService.ts @@ -409,9 +409,17 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( () => reconcileInstanceSubscriptions, ).pipe(Effect.forkScoped); + const isSameResumeCursor = ( + adapter: ProviderAdapterShape, + threadId: ThreadId, + persisted: unknown, + recovered: unknown, + ) => adapter.isSameResumeCursor?.(threadId, persisted, recovered) ?? Effect.succeed(false); + const recoverSessionForThread = Effect.fn("recoverSessionForThread")(function* (input: { readonly binding: ProviderSessionDirectory.ProviderRuntimeBinding; readonly operation: string; + readonly requireResumeCursor?: unknown; }) { const bindingInstanceId = yield* requireBindingInstanceId(input.operation, input.binding); yield* Effect.annotateCurrentSpan({ @@ -431,6 +439,20 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( (session) => session.threadId === input.binding.threadId, ); if (existing) { + const resumeCursorMatches = + input.requireResumeCursor === undefined || + (yield* isSameResumeCursor( + adapter, + input.binding.threadId, + input.requireResumeCursor, + existing.resumeCursor, + )); + if (input.requireResumeCursor !== undefined && !resumeCursorMatches) { + return yield* toValidationError( + input.operation, + `Cannot automatically continue thread '${input.binding.threadId}' because its active provider session does not match the persisted conversation.`, + ); + } yield* upsertSessionBinding( { ...existing, providerInstanceId: bindingInstanceId }, input.binding.threadId, @@ -474,6 +496,23 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( ); } + const resumeCursorMatches = + input.requireResumeCursor === undefined || + (yield* isSameResumeCursor( + adapter, + input.binding.threadId, + input.requireResumeCursor, + resumed.resumeCursor, + )); + if (input.requireResumeCursor !== undefined && !resumeCursorMatches) { + yield* adapter.stopSession(input.binding.threadId).pipe(Effect.ignore); + yield* clearMcpSession(input.binding.threadId); + return yield* toValidationError( + input.operation, + `Cannot automatically continue thread '${input.binding.threadId}' because the provider did not resume the persisted conversation.`, + ); + } + yield* upsertSessionBinding( { ...resumed, providerInstanceId: bindingInstanceId }, input.binding.threadId, @@ -498,6 +537,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( readonly threadId: ThreadId; readonly operation: string; readonly allowRecovery: boolean; + readonly requireResumeCursor?: unknown; }) { const bindingOption = yield* directory.getBinding(input.threadId); const binding = Option.getOrUndefined(bindingOption); @@ -512,6 +552,24 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( const hasRequestedSession = yield* adapter.hasSession(input.threadId); if (hasRequestedSession) { + if (input.requireResumeCursor !== undefined) { + const activeSessions = yield* adapter.listSessions(); + const activeSession = activeSessions.find((session) => session.threadId === input.threadId); + const resumeCursorMatches = + activeSession !== undefined && + (yield* isSameResumeCursor( + adapter, + input.threadId, + input.requireResumeCursor, + activeSession.resumeCursor, + )); + if (activeSession === undefined || !resumeCursorMatches) { + return yield* toValidationError( + input.operation, + `Cannot automatically continue thread '${input.threadId}' because its active provider session does not match the persisted conversation.`, + ); + } + } return { adapter, instanceId, @@ -534,6 +592,9 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( const recovered = yield* recoverSessionForThread({ binding, operation: input.operation, + ...(input.requireResumeCursor !== undefined + ? { requireResumeCursor: input.requireResumeCursor } + : {}), }); return { adapter: recovered.adapter, @@ -715,7 +776,10 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( }, ); - const sendTurn: ProviderServiceMethod<"sendTurn"> = Effect.fn("sendTurn")(function* (rawInput) { + const sendTurnImpl = function* ( + rawInput: Parameters>[0], + options: Parameters>[1], + ) { const parsed = yield* decodeInputOrValidationError({ operation: "ProviderService.sendTurn", schema: ProviderSendTurnInput, @@ -781,6 +845,9 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( threadId: input.threadId, operation: "ProviderService.sendTurn", allowRecovery: false, + ...(options?.requireResumeCursor !== undefined + ? { requireResumeCursor: options.requireResumeCursor } + : {}), }); if ( input.continuation === true && @@ -798,6 +865,9 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( threadId: input.threadId, operation: "ProviderService.sendTurn", allowRecovery: true, + ...(options?.requireResumeCursor !== undefined + ? { requireResumeCursor: options.requireResumeCursor } + : {}), }); } metricProvider = routed.adapter.provider; @@ -852,7 +922,8 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( }), }), ); - }); + }; + const sendTurn: ProviderServiceMethod<"sendTurn"> = Effect.fn("sendTurn")(sendTurnImpl); const interruptTurn: ProviderServiceMethod<"interruptTurn"> = Effect.fn("interruptTurn")( function* (rawInput) { diff --git a/apps/server/src/provider/Services/ProviderAdapter.ts b/apps/server/src/provider/Services/ProviderAdapter.ts index dcf8eff4a27..89413d2f625 100644 --- a/apps/server/src/provider/Services/ProviderAdapter.ts +++ b/apps/server/src/provider/Services/ProviderAdapter.ts @@ -61,6 +61,17 @@ export interface ProviderAdapterShape { input: ProviderSessionStartInput, ) => Effect.Effect; + /** + * Confirms that a recovered session still represents the persisted provider + * conversation. Used by automatic restart recovery to avoid sending a + * continuation turn into a newly-created empty session. + */ + readonly isSameResumeCursor?: ( + threadId: ThreadId, + persisted: unknown, + recovered: unknown, + ) => Effect.Effect; + /** * Send a turn to an active provider session. */ diff --git a/apps/server/src/provider/Services/ProviderService.ts b/apps/server/src/provider/Services/ProviderService.ts index 545641d2e86..f1106fce207 100644 --- a/apps/server/src/provider/Services/ProviderService.ts +++ b/apps/server/src/provider/Services/ProviderService.ts @@ -51,6 +51,10 @@ export interface ProviderServiceShape { */ readonly sendTurn: ( input: ProviderSendTurnInput, + options?: { + /** Fail recovery unless the adapter resumed this exact provider conversation. */ + readonly requireResumeCursor?: unknown; + }, ) => Effect.Effect; /** diff --git a/apps/server/src/serverRuntimeStartup.reconcile.test.ts b/apps/server/src/serverRuntimeStartup.reconcile.test.ts index aa1b1a7f978..cc9a1ae7838 100644 --- a/apps/server/src/serverRuntimeStartup.reconcile.test.ts +++ b/apps/server/src/serverRuntimeStartup.reconcile.test.ts @@ -307,6 +307,129 @@ it.effect("continues marked sessions after activation with provider-specific inp }), ); +const verifyOrdinaryRestartContinuation = (markerCleared: boolean) => + Effect.gen(function* () { + const thread = makeThread( + "thread-continue-after-restart", + "running", + TurnId.make("turn-continue-after-restart"), + ); + const continuationSent = yield* Deferred.make(); + const continuationCleared = yield* Deferred.make(); + const sends: ProviderSendTurnInput[] = []; + const sendOptions: Array< + Parameters[1] + > = []; + const dispatched: OrchestrationCommand[] = []; + const upserts: ProviderSessionDirectory.ProviderRuntimeBinding[] = []; + const providerService: ProviderService.ProviderService["Service"] = { + ...makeProviderService(), + getCapabilities: () => + Effect.succeed({ + sessionModelSwitch: "in-session", + promptlessTurnContinuation: true, + }), + sendTurn: (input, options) => + Effect.sync(() => { + sends.push(input); + sendOptions.push(options); + return { + threadId: input.threadId, + turnId: TurnId.make("continued-after-restart"), + }; + }).pipe(Effect.tap(() => Deferred.succeed(continuationSent, undefined))), + }; + + yield* runReconciliation({ + threads: [thread], + providerService, + directory: { + getBinding: () => + Effect.succeed( + Option.some({ + threadId: thread.id, + provider: ProviderDriverKind.make("codex"), + providerInstanceId, + status: "running" as const, + resumeCursor: { threadId: "provider-thread-resume" }, + runtimePayload: { + activeTurnId: thread.session.activeTurnId, + lastRuntimeEvent: "provider.sendTurn", + ...(markerCleared ? { continueAfterServerUpdate: null } : {}), + }, + }), + ), + upsert: (binding) => + Effect.sync(() => { + upserts.push(binding); + const payload = binding.runtimePayload; + return ( + payload !== null && + typeof payload === "object" && + !Array.isArray(payload) && + "continueAfterServerUpdate" in payload && + payload.continueAfterServerUpdate === null + ); + }).pipe( + Effect.flatMap((cleared) => + cleared ? Deferred.succeed(continuationCleared, undefined) : Effect.void, + ), + ), + getProvider: () => Effect.die("unused"), + listThreadIds: () => Effect.die("unused"), + listBindings: () => Effect.die("unused"), + }, + dispatch: (command) => + Effect.sync(() => dispatched.push(command)).pipe( + Effect.as({ sequence: dispatched.length }), + ), + }); + yield* Deferred.await(continuationSent); + yield* Deferred.await(continuationCleared); + + assert.deepStrictEqual(sends, [ + { threadId: thread.id, continuation: true, interactionMode: "default" }, + ]); + assert.deepStrictEqual(sendOptions, [ + { requireResumeCursor: { threadId: "provider-thread-resume" } }, + ]); + assert.deepStrictEqual( + dispatched.map((command) => + command.type === "thread.session.set" + ? { + status: command.session.status, + activeTurnId: command.session.activeTurnId, + lastError: command.session.lastError, + } + : null, + ), + [{ status: "starting", activeTurnId: null, lastError: null }], + ); + assert.deepStrictEqual( + upserts.map((binding) => binding.runtimePayload), + [ + { + activeTurnId: null, + lastRuntimeEvent: "provider.sendTurn", + continueAfterServerUpdate: thread.session.activeTurnId, + }, + { + activeTurnId: thread.session.activeTurnId, + lastRuntimeEvent: "provider.sendTurn", + continueAfterServerUpdate: null, + }, + ], + ); + }); + +it.effect.each([ + { markerState: "absent", markerCleared: false }, + { markerState: "cleared", markerCleared: true }, +] as const)( + "continues an active session after reopening T3 Code when the update marker is $markerState", + ({ markerCleared }) => verifyOrdinaryRestartContinuation(markerCleared), +); + it.effect("does not continue archived or deleted marked sessions", () => { const archived = makeThread( "thread-continue-archived", @@ -438,7 +561,7 @@ it.effect("retries continuation preparation before settling a persistent failure it.effect("reconciles multiple active and archived orphans but skips live sessions", () => { const starting = makeThread("thread-starting", "starting"); - const running = makeThread("thread-running", "running", TurnId.make("turn-running")); + const running = makeThread("thread-running", "running"); const staleActiveTurn = makeThread( "thread-stale-active-turn", "ready", diff --git a/apps/server/src/serverRuntimeStartup.ts b/apps/server/src/serverRuntimeStartup.ts index 064796b2810..d48bf06b241 100644 --- a/apps/server/src/serverRuntimeStartup.ts +++ b/apps/server/src/serverRuntimeStartup.ts @@ -472,14 +472,28 @@ export const reconcileProviderSessions = Effect.gen(function* () { }).pipe(Effect.as(Option.none())), ), ); - const continuationMarkerPresent = - Option.isSome(binding) && hasServerUpdateContinuationMarker(binding.value.runtimePayload); const continuationTurnId = Option.isSome(binding) ? readServerUpdateContinuationTurnId(binding.value.runtimePayload) : null; + const continuationMarkerPresent = continuationTurnId !== null; const continuationMarked = continuationTurnId !== null && (session.activeTurnId === null || continuationTurnId === session.activeTurnId); + // Quitting the T3 Code desktop app also stops its embedded backend, but + // unlike a managed self-update it does not write a continuation marker. + // On the next app launch, the projected active turn and persisted provider + // cursor make the interrupted session a continuation candidate. Before a + // turn is sent, ProviderService verifies that recovery retained the same + // provider conversation. The same startup path also covers standalone + // backend restarts. A stale explicit update marker remains authoritative + // and must not resume a different turn. + const restartContinuationEligible = + continuationMarked || + (!continuationMarkerPresent && + session.activeTurnId !== null && + Option.isSome(binding) && + binding.value.resumeCursor !== null && + binding.value.resumeCursor !== undefined); const settleAsError = (lastError: string) => Effect.gen(function* () { yield* Effect.gen(function* () { @@ -535,10 +549,15 @@ export const reconcileProviderSessions = Effect.gen(function* () { if ( Option.isSome(binding) && - continuationMarked && + restartContinuationEligible && thread.archivedAt === null && thread.deletedAt === null ) { + const recoveryTurnId = continuationTurnId ?? session.activeTurnId; + if (recoveryTurnId === null) { + yield* settleAsError(ORPHANED_PROVIDER_SESSION_ERROR); + continue; + } const prepared = yield* Effect.gen(function* () { yield* directory.upsert({ ...binding.value, @@ -546,6 +565,7 @@ export const reconcileProviderSessions = Effect.gen(function* () { runtimePayload: { ...readRuntimePayload(binding.value.runtimePayload), activeTurnId: null, + [SERVER_UPDATE_CONTINUATION_KEY]: recoveryTurnId, }, }); const resumedAt = DateTime.formatIso(yield* DateTime.now); @@ -585,13 +605,16 @@ export const reconcileProviderSessions = Effect.gen(function* () { }); } const capabilities = yield* providerService.getCapabilities(providerInstanceId); - yield* providerService.sendTurn({ - threadId: thread.id, - ...(capabilities.promptlessTurnContinuation === true - ? { continuation: true } - : { input: SERVER_UPDATE_CONTINUATION_PROMPT }), - interactionMode: thread.interactionMode, - }); + yield* providerService.sendTurn( + { + threadId: thread.id, + ...(capabilities.promptlessTurnContinuation === true + ? { continuation: true } + : { input: SERVER_UPDATE_CONTINUATION_PROMPT }), + interactionMode: thread.interactionMode, + }, + { requireResumeCursor: binding.value.resumeCursor }, + ); }); const continuationExit = yield* Effect.exit(continuation); if (Exit.isSuccess(continuationExit) || Cause.hasInterrupts(continuationExit.cause)) { @@ -608,12 +631,12 @@ export const reconcileProviderSessions = Effect.gen(function* () { } return; } - yield* Effect.logWarning("failed to continue provider session after server update", { + yield* Effect.logWarning("failed to continue provider session after T3 Code restarted", { threadId: thread.id, cause: continuationExit.cause, }); yield* settleAsError( - "Could not continue this thread after the server update. Send a new message to continue.", + "Could not continue this thread after T3 Code restarted. Send a new message to continue.", ).pipe(Effect.ignoreCause); }), );