diff --git a/apps/server/integration/orphanedProviderSessionStartup.integration.test.ts b/apps/server/integration/orphanedProviderSessionStartup.integration.test.ts index 86a00323c049..e07681ae153b 100644 --- a/apps/server/integration/orphanedProviderSessionStartup.integration.test.ts +++ b/apps/server/integration/orphanedProviderSessionStartup.integration.test.ts @@ -7,10 +7,13 @@ import { ProjectId, ProviderDriverKind, ProviderInstanceId, + type ProviderSendTurnInput, ThreadId, + TurnId, } from "@t3tools/contracts"; import { assert, it } from "@effect/vitest"; import * as DateTime from "effect/DateTime"; +import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; import * as Layer from "effect/Layer"; @@ -361,3 +364,123 @@ it.effect( ), ), ); + +it.effect.each(["opt-in desktop restart", "marked remote update"] as const)( + "continues a newer persisted turn after %s", + (restart) => + Effect.gen(function* () { + const config = yield* ServerConfig.ServerConfig; + const createdAt = DateTime.formatIso(yield* DateTime.now); + const activeTurnId = TurnId.make("turn-started-after-original-send"); + const originalTurnId = TurnId.make("turn-from-original-send"); + const sent = yield* Deferred.make(); + + yield* Effect.gen(function* () { + const engine = yield* OrchestrationEngine.OrchestrationEngineService; + const directory = yield* ProviderSessionDirectory.ProviderSessionDirectory; + yield* engine.dispatch({ + type: "project.create", + commandId: CommandId.make("create-restart-project"), + projectId, + title: "Restart continuation", + workspaceRoot: "/tmp/startup-orphan-project", + defaultModelSelection: { instanceId: providerInstanceId, model: "gpt-5" }, + createdAt, + }); + yield* engine.dispatch({ + type: "thread.create", + commandId: CommandId.make("create-restart-thread"), + threadId, + projectId, + title: "Newer running turn", + modelSelection: { instanceId: providerInstanceId, model: "gpt-5" }, + interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, + runtimeMode: "full-access", + branch: null, + worktreePath: null, + createdAt, + }); + yield* engine.dispatch({ + type: "thread.session.set", + commandId: CommandId.make("persist-newer-running-turn"), + threadId, + session: { + threadId, + status: "running", + providerName: "codex", + providerInstanceId, + runtimeMode: "full-access", + activeTurnId, + lastError: null, + updatedAt: createdAt, + }, + createdAt, + }); + yield* directory.upsert({ + threadId, + provider: ProviderDriverKind.make("codex"), + providerInstanceId, + status: "running", + resumeCursor, + runtimePayload: { activeTurnId: originalTurnId }, + }); + if (restart === "marked remote update") { + assert.deepStrictEqual( + yield* ServerRuntimeStartup.markRunningProviderSessionsForContinuation, + [threadId], + ); + } + }).pipe(Effect.provide(makePersistedRuntimeLayer(config.dbPath))); + + yield* Effect.gen(function* () { + const query = yield* ProjectionSnapshotQuery.ProjectionSnapshotQuery; + const directory = yield* ProviderSessionDirectory.ProviderSessionDirectory; + const provider = yield* ProviderService.ProviderService; + const before = Option.getOrThrow(yield* query.getThreadDetailById(threadId)); + assert.equal(before.session?.activeTurnId, activeTurnId); + assert.propertyVal( + Option.getOrThrow(yield* directory.getBinding(threadId)).runtimePayload, + "activeTurnId", + originalTurnId, + ); + yield* ServerRuntimeStartup.reconcileProviderSessions.pipe( + Effect.provideService(ProviderService.ProviderService, { + ...provider, + getCapabilities: () => + Effect.succeed({ + sessionModelSwitch: "in-session", + promptlessTurnContinuation: true, + }), + sendTurn: (input) => + Deferred.succeed(sent, input).pipe( + Effect.as({ threadId, turnId: TurnId.make("continued-turn") }), + ), + }), + Effect.provide( + ServerSettings.layerTest({ + continueThreadsAfterServerUpdate: restart === "opt-in desktop restart", + }), + ), + ); + const after = Option.getOrThrow(yield* query.getThreadDetailById(threadId)); + assert.equal(after.session?.status, "starting"); + assert.equal(after.session?.activeTurnId, null); + assert.equal(after.session?.lastError, null); + assert.deepStrictEqual(yield* Deferred.await(sent), { + threadId, + continuation: true, + interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, + }); + }).pipe( + Effect.provide( + Layer.mergeAll(makePersistedRuntimeLayer(config.dbPath), startupDependencies), + ), + ); + }).pipe( + Effect.provide( + ServerConfig.layerTest(process.cwd(), { prefix: "t3-restart-newer-turn-" }).pipe( + Layer.provideMerge(NodeServices.layer), + ), + ), + ), +); diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts index 0136c3fbf170..3385137a2dae 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts @@ -781,25 +781,123 @@ describe("isRecoverableThreadResumeError", () => { }); describe("openCodexThread", () => { + it.effect("resumes metadata when historical turns contain unknown error values", () => + Effect.gen(function* () { + const response = makeThreadOpenResponse("saved-thread"); + const calls: unknown[] = []; + const opened = yield* openCodexThread({ + client: { + request: () => Effect.die("A valid resumed thread must not start fresh"), + raw: { + request: (method, payload) => { + calls.push({ method, payload }); + return Effect.succeed({ + ...response, + thread: { + ...response.thread, + turns: [ + { + id: "old-turn", + status: "failed", + items: [], + error: { + message: "Historical provider error", + codexErrorInfo: "misalignment_policy_violation", + }, + }, + ], + }, + }); + }, + }, + }, + threadId: ThreadId.make("thread-1"), + runtimeMode: "auto", + cwd: "/tmp/project", + requestedModel: "gpt-5.3-codex", + serviceTier: "fast", + resumeThreadId: "saved-thread", + }); + + NodeAssert.deepStrictEqual(opened, { + cwd: response.cwd, + model: response.model, + thread: { id: "saved-thread" }, + }); + NodeAssert.deepStrictEqual(calls, [ + { + method: "thread/resume", + payload: { + threadId: "saved-thread", + cwd: "/tmp/project", + model: "gpt-5.3-codex", + serviceTier: "fast", + approvalPolicy: "on-request", + sandbox: "workspace-write", + approvalsReviewer: "auto_review", + excludeTurns: true, + }, + }, + ]); + }), + ); + + it.effect("rejects malformed required resume metadata without starting a fresh thread", () => + Effect.gen(function* () { + for (const invalidMetadata of [ + { cwd: null }, + { model: 42 }, + { thread: { id: null } }, + { thread: {} }, + ]) { + const error = yield* openCodexThread({ + client: { + request: () => Effect.die("Invalid resume metadata must not start a fresh thread"), + raw: { + request: () => + Effect.succeed({ ...makeThreadOpenResponse("saved-thread"), ...invalidMetadata }), + }, + }, + threadId: ThreadId.make("thread-1"), + runtimeMode: "full-access", + cwd: "/tmp/project", + requestedModel: "gpt-5.3-codex", + serviceTier: undefined, + resumeThreadId: "saved-thread", + }).pipe(Effect.flip); + + NodeAssert.ok(isCodexAppServerRequestError(error)); + NodeAssert.equal(error.operation, "decode-payload"); + NodeAssert.equal(error.method, "thread/resume"); + } + }), + ); + it.effect("falls back to thread/start when resume fails recoverably", () => Effect.gen(function* () { const calls: Array<{ method: "thread/start" | "thread/resume"; payload: unknown }> = []; const started = makeThreadOpenResponse("fresh-thread"); const client = { - request: ( - method: M, - payload: CodexRpc.ClientRequestParamsByMethod[M], - ) => { - calls.push({ method, payload }); - if (method === "thread/resume") { + raw: { + request: ( + method: "thread/resume", + payload: CodexRpc.ClientRequestParamsByMethod["thread/resume"], + ) => { + calls.push({ method, payload }); return Effect.fail( new CodexErrors.CodexAppServerRequestError({ code: -32603, errorMessage: "thread not found", }), ); - } - return Effect.succeed(started as CodexRpc.ClientRequestResponsesByMethod[M]); + }, + }, + request: ( + method: "thread/start", + payload: CodexRpc.ClientRequestParamsByMethod["thread/start"], + ) => { + calls.push({ method, payload }); + return Effect.succeed(started); }, }; @@ -824,21 +922,15 @@ describe("openCodexThread", () => { it.effect("propagates non-recoverable resume failures", () => Effect.gen(function* () { const client = { - request: ( - method: M, - _payload: CodexRpc.ClientRequestParamsByMethod[M], - ) => { - if (method === "thread/resume") { - return Effect.fail( + request: () => Effect.die("Non-recoverable resume failures must not start a fresh thread"), + raw: { + request: () => + Effect.fail( new CodexErrors.CodexAppServerRequestError({ code: -32603, errorMessage: "timed out waiting for server", }), - ); - } - return Effect.succeed( - makeThreadOpenResponse("fresh-thread") as CodexRpc.ClientRequestResponsesByMethod[M], - ); + ), }, }; diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.ts index 4b88b7ce01c0..4ee5845e3a77 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.ts @@ -676,17 +676,29 @@ export function isRecoverableThreadResumeError(error: unknown): boolean { return RECOVERABLE_THREAD_RESUME_ERROR_SNIPPETS.some((snippet) => message.includes(snippet)); } -type CodexThreadOpenResponse = - | CodexRpc.ClientRequestResponsesByMethod["thread/start"] - | CodexRpc.ClientRequestResponsesByMethod["thread/resume"]; - -type CodexThreadOpenMethod = "thread/start" | "thread/resume"; +const CodexThreadResumeMetadata = Schema.Struct({ + cwd: Schema.String, + model: Schema.String, + thread: Schema.Struct({ id: Schema.String }), +}); +const decodeCodexThreadResumeMetadata = Schema.decodeUnknownEffect(CodexThreadResumeMetadata); interface CodexThreadOpenClient { - readonly request: ( - method: M, - payload: CodexRpc.ClientRequestParamsByMethod[M], - ) => Effect.Effect; + readonly raw: { + readonly request: ( + method: "thread/resume", + payload: CodexRpc.ClientRequestParamsByMethod["thread/resume"] & { + readonly excludeTurns?: boolean; + }, + ) => Effect.Effect; + }; + readonly request: ( + method: "thread/start", + payload: CodexRpc.ClientRequestParamsByMethod["thread/start"], + ) => Effect.Effect< + CodexRpc.ClientRequestResponsesByMethod["thread/start"], + CodexErrors.CodexAppServerError + >; } export const openCodexThread = (input: { @@ -697,7 +709,7 @@ export const openCodexThread = (input: { readonly requestedModel: string | undefined; readonly serviceTier: CodexServiceTier | undefined; readonly resumeThreadId: string | undefined; -}): Effect.Effect => { +}): Effect.Effect => { const resumeThreadId = input.resumeThreadId; const startParams = buildThreadStartParams({ cwd: input.cwd, @@ -710,12 +722,27 @@ export const openCodexThread = (input: { return input.client.request("thread/start", startParams); } - return input.client + // Older providers may still return history despite excludeTurns. Only the + // session metadata is needed here, so unrelated historical items cannot + // prevent resuming a valid provider thread. + return input.client.raw .request("thread/resume", { threadId: resumeThreadId, ...startParams, + excludeTurns: true, }) .pipe( + Effect.flatMap((response) => + decodeCodexThreadResumeMetadata(response).pipe( + Effect.mapError((error) => + CodexErrors.CodexAppServerRequestError.invalidPayload( + "thread/resume", + "decode-payload", + error, + ), + ), + ), + ), Effect.catchIf(isRecoverableThreadResumeError, (error) => Effect.logWarning("codex app-server thread resume fell back to fresh start", { threadId: input.threadId, diff --git a/apps/server/src/serverRuntimeStartup.reconcile.test.ts b/apps/server/src/serverRuntimeStartup.reconcile.test.ts index e46c2a8ca010..37fd210ee6da 100644 --- a/apps/server/src/serverRuntimeStartup.reconcile.test.ts +++ b/apps/server/src/serverRuntimeStartup.reconcile.test.ts @@ -170,9 +170,16 @@ it.effect("marks active running sessions that have persisted resume state", () = ); }); -it.effect.each(["marked update", "opt-in restart"] as const)( - "continues %s sessions after activation with provider-specific input", - (recovery) => +it.effect.each( + (["marked update", "opt-in restart"] as const).flatMap((recovery) => + (["current", "previous", "missing"] as const).map((persistedTurn) => ({ + recovery, + persistedTurn, + })), + ), +)( + "continues $recovery sessions with a $persistedTurn directory turn", + ({ recovery, persistedTurn }) => Effect.gen(function* () { const codex = makeThread( "thread-continue-codex", @@ -204,15 +211,22 @@ it.effect.each(["marked update", "opt-in restart"] as const)( thread.id === codex.id ? providerInstanceId : fallbackProviderInstanceId, status: "running" as const, resumeCursor: { threadId: thread.id }, - runtimePayload: - recovery === "marked update" + runtimePayload: { + activeTurnId: + thread.id === codex.id && persistedTurn !== "current" + ? persistedTurn === "previous" + ? "previous-provider-turn" + : null + : thread.session.activeTurnId, + ...(recovery === "marked update" ? { continueAfterServerUpdate: thread.id === codex.id ? codex.session.activeTurnId : fallbackContinuationTurnId, } - : { activeTurnId: thread.session.activeTurnId }, + : {}), + }, }, ]), ); @@ -276,6 +290,12 @@ it.effect.each(["marked update", "opt-in restart"] as const)( Effect.as({ sequence: dispatched.length }), ), }); + assert.isTrue( + dispatched.every( + (command) => + command.type === "thread.session.set" && command.session.status === "starting", + ), + ); yield* Deferred.await(continuationSent); yield* Deferred.await(continuationCleared); @@ -713,9 +733,7 @@ for (const scenario of [ "stopped projection", "finished projection", "stopped binding", - "finished binding", "missing cursor", - "mismatched turn", "marked without cursor", "marked stopped projection", "marked superseded turn", @@ -748,12 +766,7 @@ for (const scenario of [ status: scenario === "stopped binding" ? "stopped" : "running", ...(scenario.includes("cursor") ? {} : { resumeCursor: { threadId: thread.id } }), runtimePayload: { - activeTurnId: - scenario === "finished binding" - ? null - : scenario === "mismatched turn" || scenario === "marked superseded turn" - ? "another-turn" - : turnId, + activeTurnId: scenario === "marked superseded turn" ? "another-turn" : turnId, ...(scenario.startsWith("marked") ? { continueAfterServerUpdate: turnId } : {}), }, }), diff --git a/apps/server/src/serverRuntimeStartup.ts b/apps/server/src/serverRuntimeStartup.ts index 6f8bbc053b0c..0905439257ff 100644 --- a/apps/server/src/serverRuntimeStartup.ts +++ b/apps/server/src/serverRuntimeStartup.ts @@ -555,7 +555,8 @@ export const reconcileProviderSessions = Effect.gen(function* () { continuationTurnId !== null && (session.activeTurnId === null || continuationTurnId === session.activeTurnId) && Option.isSome(binding) && - (readRuntimePayload(binding.value.runtimePayload).activeTurnId == null || + (session.activeTurnId !== null || + readRuntimePayload(binding.value.runtimePayload).activeTurnId == null || readRuntimePayload(binding.value.runtimePayload).activeTurnId === continuationTurnId); const preparedWhileReady = session.status === "ready" && @@ -564,16 +565,15 @@ export const reconcileProviderSessions = Effect.gen(function* () { Option.isSome(binding) && readRuntimePayload(binding.value.runtimePayload).activeTurnId === null && readRuntimePayload(binding.value.runtimePayload).continueAfterServerUpdatePrepared === true; - // Abrupt shutdowns cannot write an update marker. Require both durable - // records to agree on an unfinished turn before recovering one implicitly. + // Runtime events advance the projection's turn, but not the directory's + // last admitted turn. Use the projection to identify interrupted work. const interruptedByRestart = continueAfterRestart && session.status === "running" && session.activeTurnId !== null && Option.isSome(binding) && binding.value.status === "running" && - binding.value.resumeCursor != null && - readRuntimePayload(binding.value.runtimePayload).activeTurnId === session.activeTurnId; + binding.value.resumeCursor != null; const settleAsError = (lastError: string) => Effect.gen(function* () { yield* Effect.gen(function* () {