authorgravatar for maria@kuuro.netmaria <maria@kuuro.net> 2026-09-06 21:36:08-03:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-06 21:36:08-03:00
log1abc717f0dbd08027b5be4ec45ae6c5bccf96418
treecf44d63e9e682e393f40f490e074b2ef66cad6b3
parenta12589dc0bec60b498e72bdd2190871ee6b5c80d
signature Signed by PGP key B5690EEEBB952194

fix(server): keep interrupted threads resumable after restarts (#10421)

5 files changed, 304 insertions(+), 49 deletions(-)

apps/server/integration/orphanedProviderSessionStartup.integration.test.ts+123
......@@ -7,10 +7,13 @@ import {
77 ProjectId,
88 ProviderDriverKind,
99 ProviderInstanceId,
10 type ProviderSendTurnInput,
1011 ThreadId,
12 TurnId,
1113} from "@t3tools/contracts";
1214import { assert, it } from "@effect/vitest";
1315import * as DateTime from "effect/DateTime";
16import * as Deferred from "effect/Deferred";
1417import * as Effect from "effect/Effect";
1518import * as Exit from "effect/Exit";
1619import * as Layer from "effect/Layer";
......@@ -361,3 +364,123 @@ it.effect(
361364 ),
362365 ),
363366);
367
368it.effect.each(["opt-in desktop restart", "marked remote update"] as const)(
369 "continues a newer persisted turn after %s",
370 (restart) =>
371 Effect.gen(function* () {
372 const config = yield* ServerConfig.ServerConfig;
373 const createdAt = DateTime.formatIso(yield* DateTime.now);
374 const activeTurnId = TurnId.make("turn-started-after-original-send");
375 const originalTurnId = TurnId.make("turn-from-original-send");
376 const sent = yield* Deferred.make<ProviderSendTurnInput>();
377
378 yield* Effect.gen(function* () {
379 const engine = yield* OrchestrationEngine.OrchestrationEngineService;
380 const directory = yield* ProviderSessionDirectory.ProviderSessionDirectory;
381 yield* engine.dispatch({
382 type: "project.create",
383 commandId: CommandId.make("create-restart-project"),
384 projectId,
385 title: "Restart continuation",
386 workspaceRoot: "/tmp/startup-orphan-project",
387 defaultModelSelection: { instanceId: providerInstanceId, model: "gpt-5" },
388 createdAt,
389 });
390 yield* engine.dispatch({
391 type: "thread.create",
392 commandId: CommandId.make("create-restart-thread"),
393 threadId,
394 projectId,
395 title: "Newer running turn",
396 modelSelection: { instanceId: providerInstanceId, model: "gpt-5" },
397 interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
398 runtimeMode: "full-access",
399 branch: null,
400 worktreePath: null,
401 createdAt,
402 });
403 yield* engine.dispatch({
404 type: "thread.session.set",
405 commandId: CommandId.make("persist-newer-running-turn"),
406 threadId,
407 session: {
408 threadId,
409 status: "running",
410 providerName: "codex",
411 providerInstanceId,
412 runtimeMode: "full-access",
413 activeTurnId,
414 lastError: null,
415 updatedAt: createdAt,
416 },
417 createdAt,
418 });
419 yield* directory.upsert({
420 threadId,
421 provider: ProviderDriverKind.make("codex"),
422 providerInstanceId,
423 status: "running",
424 resumeCursor,
425 runtimePayload: { activeTurnId: originalTurnId },
426 });
427 if (restart === "marked remote update") {
428 assert.deepStrictEqual(
429 yield* ServerRuntimeStartup.markRunningProviderSessionsForContinuation,
430 [threadId],
431 );
432 }
433 }).pipe(Effect.provide(makePersistedRuntimeLayer(config.dbPath)));
434
435 yield* Effect.gen(function* () {
436 const query = yield* ProjectionSnapshotQuery.ProjectionSnapshotQuery;
437 const directory = yield* ProviderSessionDirectory.ProviderSessionDirectory;
438 const provider = yield* ProviderService.ProviderService;
439 const before = Option.getOrThrow(yield* query.getThreadDetailById(threadId));
440 assert.equal(before.session?.activeTurnId, activeTurnId);
441 assert.propertyVal(
442 Option.getOrThrow(yield* directory.getBinding(threadId)).runtimePayload,
443 "activeTurnId",
444 originalTurnId,
445 );
446 yield* ServerRuntimeStartup.reconcileProviderSessions.pipe(
447 Effect.provideService(ProviderService.ProviderService, {
448 ...provider,
449 getCapabilities: () =>
450 Effect.succeed({
451 sessionModelSwitch: "in-session",
452 promptlessTurnContinuation: true,
453 }),
454 sendTurn: (input) =>
455 Deferred.succeed(sent, input).pipe(
456 Effect.as({ threadId, turnId: TurnId.make("continued-turn") }),
457 ),
458 }),
459 Effect.provide(
460 ServerSettings.layerTest({
461 continueThreadsAfterServerUpdate: restart === "opt-in desktop restart",
462 }),
463 ),
464 );
465 const after = Option.getOrThrow(yield* query.getThreadDetailById(threadId));
466 assert.equal(after.session?.status, "starting");
467 assert.equal(after.session?.activeTurnId, null);
468 assert.equal(after.session?.lastError, null);
469 assert.deepStrictEqual(yield* Deferred.await(sent), {
470 threadId,
471 continuation: true,
472 interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
473 });
474 }).pipe(
475 Effect.provide(
476 Layer.mergeAll(makePersistedRuntimeLayer(config.dbPath), startupDependencies),
477 ),
478 );
479 }).pipe(
480 Effect.provide(
481 ServerConfig.layerTest(process.cwd(), { prefix: "t3-restart-newer-turn-" }).pipe(
482 Layer.provideMerge(NodeServices.layer),
483 ),
484 ),
485 ),
486);
apps/server/src/provider/Layers/CodexSessionRuntime.test.ts+111-19
......@@ -781,25 +781,123 @@ describe("isRecoverableThreadResumeError", () => {
781781});
782782
783783describe("openCodexThread", () => {
784 it.effect("resumes metadata when historical turns contain unknown error values", () =>
785 Effect.gen(function* () {
786 const response = makeThreadOpenResponse("saved-thread");
787 const calls: unknown[] = [];
788 const opened = yield* openCodexThread({
789 client: {
790 request: () => Effect.die("A valid resumed thread must not start fresh"),
791 raw: {
792 request: (method, payload) => {
793 calls.push({ method, payload });
794 return Effect.succeed({
795 ...response,
796 thread: {
797 ...response.thread,
798 turns: [
799 {
800 id: "old-turn",
801 status: "failed",
802 items: [],
803 error: {
804 message: "Historical provider error",
805 codexErrorInfo: "misalignment_policy_violation",
806 },
807 },
808 ],
809 },
810 });
811 },
812 },
813 },
814 threadId: ThreadId.make("thread-1"),
815 runtimeMode: "auto",
816 cwd: "/tmp/project",
817 requestedModel: "gpt-5.3-codex",
818 serviceTier: "fast",
819 resumeThreadId: "saved-thread",
820 });
821
822 NodeAssert.deepStrictEqual(opened, {
823 cwd: response.cwd,
824 model: response.model,
825 thread: { id: "saved-thread" },
826 });
827 NodeAssert.deepStrictEqual(calls, [
828 {
829 method: "thread/resume",
830 payload: {
831 threadId: "saved-thread",
832 cwd: "/tmp/project",
833 model: "gpt-5.3-codex",
834 serviceTier: "fast",
835 approvalPolicy: "on-request",
836 sandbox: "workspace-write",
837 approvalsReviewer: "auto_review",
838 excludeTurns: true,
839 },
840 },
841 ]);
842 }),
843 );
844
845 it.effect("rejects malformed required resume metadata without starting a fresh thread", () =>
846 Effect.gen(function* () {
847 for (const invalidMetadata of [
848 { cwd: null },
849 { model: 42 },
850 { thread: { id: null } },
851 { thread: {} },
852 ]) {
853 const error = yield* openCodexThread({
854 client: {
855 request: () => Effect.die("Invalid resume metadata must not start a fresh thread"),
856 raw: {
857 request: () =>
858 Effect.succeed({ ...makeThreadOpenResponse("saved-thread"), ...invalidMetadata }),
859 },
860 },
861 threadId: ThreadId.make("thread-1"),
862 runtimeMode: "full-access",
863 cwd: "/tmp/project",
864 requestedModel: "gpt-5.3-codex",
865 serviceTier: undefined,
866 resumeThreadId: "saved-thread",
867 }).pipe(Effect.flip);
868
869 NodeAssert.ok(isCodexAppServerRequestError(error));
870 NodeAssert.equal(error.operation, "decode-payload");
871 NodeAssert.equal(error.method, "thread/resume");
872 }
873 }),
874 );
875
784876 it.effect("falls back to thread/start when resume fails recoverably", () =>
785877 Effect.gen(function* () {
786878 const calls: Array<{ method: "thread/start" | "thread/resume"; payload: unknown }> = [];
787879 const started = makeThreadOpenResponse("fresh-thread");
788880 const client = {
789 request: <M extends "thread/start" | "thread/resume">(
790 method: M,
791 payload: CodexRpc.ClientRequestParamsByMethod[M],
792 ) => {
793 calls.push({ method, payload });
794 if (method === "thread/resume") {
881 raw: {
882 request: (
883 method: "thread/resume",
884 payload: CodexRpc.ClientRequestParamsByMethod["thread/resume"],
885 ) => {
886 calls.push({ method, payload });
795887 return Effect.fail(
796888 new CodexErrors.CodexAppServerRequestError({
797889 code: -32603,
798890 errorMessage: "thread not found",
799891 }),
800892 );
801 }
802 return Effect.succeed(started as CodexRpc.ClientRequestResponsesByMethod[M]);
893 },
894 },
895 request: (
896 method: "thread/start",
897 payload: CodexRpc.ClientRequestParamsByMethod["thread/start"],
898 ) => {
899 calls.push({ method, payload });
900 return Effect.succeed(started);
803901 },
804902 };
805903
......@@ -824,21 +922,15 @@ describe("openCodexThread", () => {
824922 it.effect("propagates non-recoverable resume failures", () =>
825923 Effect.gen(function* () {
826924 const client = {
827 request: <M extends "thread/start" | "thread/resume">(
828 method: M,
829 _payload: CodexRpc.ClientRequestParamsByMethod[M],
830 ) => {
831 if (method === "thread/resume") {
832 return Effect.fail(
925 request: () => Effect.die("Non-recoverable resume failures must not start a fresh thread"),
926 raw: {
927 request: () =>
928 Effect.fail(
833929 new CodexErrors.CodexAppServerRequestError({
834930 code: -32603,
835931 errorMessage: "timed out waiting for server",
836932 }),
837 );
838 }
839 return Effect.succeed(
840 makeThreadOpenResponse("fresh-thread") as CodexRpc.ClientRequestResponsesByMethod[M],
841 );
933 ),
842934 },
843935 };
844936
apps/server/src/provider/Layers/CodexSessionRuntime.ts+38-11
......@@ -676,17 +676,29 @@ export function isRecoverableThreadResumeError(error: unknown): boolean {
676676 return RECOVERABLE_THREAD_RESUME_ERROR_SNIPPETS.some((snippet) => message.includes(snippet));
677677}
678678
679type CodexThreadOpenResponse =
680 | CodexRpc.ClientRequestResponsesByMethod["thread/start"]
681 | CodexRpc.ClientRequestResponsesByMethod["thread/resume"];
682
683type CodexThreadOpenMethod = "thread/start" | "thread/resume";
679const CodexThreadResumeMetadata = Schema.Struct({
680 cwd: Schema.String,
681 model: Schema.String,
682 thread: Schema.Struct({ id: Schema.String }),
683});
684const decodeCodexThreadResumeMetadata = Schema.decodeUnknownEffect(CodexThreadResumeMetadata);
684685
685686interface CodexThreadOpenClient {
686 readonly request: <M extends CodexThreadOpenMethod>(
687 method: M,
688 payload: CodexRpc.ClientRequestParamsByMethod[M],
689 ) => Effect.Effect<CodexRpc.ClientRequestResponsesByMethod[M], CodexErrors.CodexAppServerError>;
687 readonly raw: {
688 readonly request: (
689 method: "thread/resume",
690 payload: CodexRpc.ClientRequestParamsByMethod["thread/resume"] & {
691 readonly excludeTurns?: boolean;
692 },
693 ) => Effect.Effect<unknown, CodexErrors.CodexAppServerError>;
694 };
695 readonly request: (
696 method: "thread/start",
697 payload: CodexRpc.ClientRequestParamsByMethod["thread/start"],
698 ) => Effect.Effect<
699 CodexRpc.ClientRequestResponsesByMethod["thread/start"],
700 CodexErrors.CodexAppServerError
701 >;
690702}
691703
692704export const openCodexThread = (input: {
......@@ -697,7 +709,7 @@ export const openCodexThread = (input: {
697709 readonly requestedModel: string | undefined;
698710 readonly serviceTier: CodexServiceTier | undefined;
699711 readonly resumeThreadId: string | undefined;
700}): Effect.Effect<CodexThreadOpenResponse, CodexErrors.CodexAppServerError> => {
712}): Effect.Effect<typeof CodexThreadResumeMetadata.Type, CodexErrors.CodexAppServerError> => {
701713 const resumeThreadId = input.resumeThreadId;
702714 const startParams = buildThreadStartParams({
703715 cwd: input.cwd,
......@@ -710,12 +722,27 @@ export const openCodexThread = (input: {
710722 return input.client.request("thread/start", startParams);
711723 }
712724
713 return input.client
725 // Older providers may still return history despite excludeTurns. Only the
726 // session metadata is needed here, so unrelated historical items cannot
727 // prevent resuming a valid provider thread.
728 return input.client.raw
714729 .request("thread/resume", {
715730 threadId: resumeThreadId,
716731 ...startParams,
732 excludeTurns: true,
717733 })
718734 .pipe(
735 Effect.flatMap((response) =>
736 decodeCodexThreadResumeMetadata(response).pipe(
737 Effect.mapError((error) =>
738 CodexErrors.CodexAppServerRequestError.invalidPayload(
739 "thread/resume",
740 "decode-payload",
741 error,
742 ),
743 ),
744 ),
745 ),
719746 Effect.catchIf(isRecoverableThreadResumeError, (error) =>
720747 Effect.logWarning("codex app-server thread resume fell back to fresh start", {
721748 threadId: input.threadId,
apps/server/src/serverRuntimeStartup.reconcile.test.ts+27-14
......@@ -170,9 +170,16 @@ it.effect("marks active running sessions that have persisted resume state", () =
170170 );
171171});
172172
173it.effect.each(["marked update", "opt-in restart"] as const)(
174 "continues %s sessions after activation with provider-specific input",
175 (recovery) =>
173it.effect.each(
174 (["marked update", "opt-in restart"] as const).flatMap((recovery) =>
175 (["current", "previous", "missing"] as const).map((persistedTurn) => ({
176 recovery,
177 persistedTurn,
178 })),
179 ),
180)(
181 "continues $recovery sessions with a $persistedTurn directory turn",
182 ({ recovery, persistedTurn }) =>
176183 Effect.gen(function* () {
177184 const codex = makeThread(
178185 "thread-continue-codex",
......@@ -204,15 +211,22 @@ it.effect.each(["marked update", "opt-in restart"] as const)(
204211 thread.id === codex.id ? providerInstanceId : fallbackProviderInstanceId,
205212 status: "running" as const,
206213 resumeCursor: { threadId: thread.id },
207 runtimePayload:
208 recovery === "marked update"
214 runtimePayload: {
215 activeTurnId:
216 thread.id === codex.id && persistedTurn !== "current"
217 ? persistedTurn === "previous"
218 ? "previous-provider-turn"
219 : null
220 : thread.session.activeTurnId,
221 ...(recovery === "marked update"
209222 ? {
210223 continueAfterServerUpdate:
211224 thread.id === codex.id
212225 ? codex.session.activeTurnId
213226 : fallbackContinuationTurnId,
214227 }
215 : { activeTurnId: thread.session.activeTurnId },
228 : {}),
229 },
216230 },
217231 ]),
218232 );
......@@ -276,6 +290,12 @@ it.effect.each(["marked update", "opt-in restart"] as const)(
276290 Effect.as({ sequence: dispatched.length }),
277291 ),
278292 });
293 assert.isTrue(
294 dispatched.every(
295 (command) =>
296 command.type === "thread.session.set" && command.session.status === "starting",
297 ),
298 );
279299 yield* Deferred.await(continuationSent);
280300 yield* Deferred.await(continuationCleared);
281301
......@@ -713,9 +733,7 @@ for (const scenario of [
713733 "stopped projection",
714734 "finished projection",
715735 "stopped binding",
716 "finished binding",
717736 "missing cursor",
718 "mismatched turn",
719737 "marked without cursor",
720738 "marked stopped projection",
721739 "marked superseded turn",
......@@ -748,12 +766,7 @@ for (const scenario of [
748766 status: scenario === "stopped binding" ? "stopped" : "running",
749767 ...(scenario.includes("cursor") ? {} : { resumeCursor: { threadId: thread.id } }),
750768 runtimePayload: {
751 activeTurnId:
752 scenario === "finished binding"
753 ? null
754 : scenario === "mismatched turn" || scenario === "marked superseded turn"
755 ? "another-turn"
756 : turnId,
769 activeTurnId: scenario === "marked superseded turn" ? "another-turn" : turnId,
757770 ...(scenario.startsWith("marked") ? { continueAfterServerUpdate: turnId } : {}),
758771 },
759772 }),
apps/server/src/serverRuntimeStartup.ts+5-5
......@@ -555,7 +555,8 @@ export const reconcileProviderSessions = Effect.gen(function* () {
555555 continuationTurnId !== null &&
556556 (session.activeTurnId === null || continuationTurnId === session.activeTurnId) &&
557557 Option.isSome(binding) &&
558 (readRuntimePayload(binding.value.runtimePayload).activeTurnId == null ||
558 (session.activeTurnId !== null ||
559 readRuntimePayload(binding.value.runtimePayload).activeTurnId == null ||
559560 readRuntimePayload(binding.value.runtimePayload).activeTurnId === continuationTurnId);
560561 const preparedWhileReady =
561562 session.status === "ready" &&
......@@ -564,16 +565,15 @@ export const reconcileProviderSessions = Effect.gen(function* () {
564565 Option.isSome(binding) &&
565566 readRuntimePayload(binding.value.runtimePayload).activeTurnId === null &&
566567 readRuntimePayload(binding.value.runtimePayload).continueAfterServerUpdatePrepared === true;
567 // Abrupt shutdowns cannot write an update marker. Require both durable
568 // records to agree on an unfinished turn before recovering one implicitly.
568 // Runtime events advance the projection's turn, but not the directory's
569 // last admitted turn. Use the projection to identify interrupted work.
569570 const interruptedByRestart =
570571 continueAfterRestart &&
571572 session.status === "running" &&
572573 session.activeTurnId !== null &&
573574 Option.isSome(binding) &&
574575 binding.value.status === "running" &&
575 binding.value.resumeCursor != null &&
576 readRuntimePayload(binding.value.runtimePayload).activeTurnId === session.activeTurnId;
576 binding.value.resumeCursor != null;
577577 const settleAsError = (lastError: string) =>
578578 Effect.gen(function* () {
579579 yield* Effect.gen(function* () {