authorgravatar for maria@kuuro.netmaria <maria@kuuro.net> 2026-09-04 18:35:42-04:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-04 18:35:42-04:00
logb906ce2d73d025e801877f42a35b7a2f5629806f
tree17c954deddab43d1179fb18df23979bad108e955
parent5a2f3ebf6ee6eab6124ad68272c2fb5d74bc0f5d
signature Signed by PGP key B5690EEEBB952194

fix(server): recover opted-in threads after machine restarts (#9803)

11 files changed, 708 insertions(+), 183 deletions(-)

apps/desktop/src/settings/DesktopClientSettings.test.ts-1
...@@ -26,7 +26,6 @@ const clientSettings: ClientSettings = {...@@ -26,7 +26,6 @@ const clientSettings: ClientSettings = {
26 confirmThreadArchive: true,26 confirmThreadArchive: true,
27 confirmThreadDelete: false,27 confirmThreadDelete: false,
28 confirmThreadUnpin: false,28 confirmThreadUnpin: false,
29 continueThreadsAfterServerUpdate: true,
30 contextWindowMeterEnabled: false,29 contextWindowMeterEnabled: false,
31 composerCollapseOnBlur: false,30 composerCollapseOnBlur: false,
32 composerCollapseOnScroll: true,31 composerCollapseOnScroll: true,
apps/server/src/provider/Layers/ProviderService.test.ts+119
...@@ -452,6 +452,125 @@ function makeProviderServiceLayer(...@@ -452,6 +452,125 @@ function makeProviderServiceLayer(
452 };452 };
453}453}
454454
455for (const [enabled, completed] of [
456 [false, false],
457 [true, false],
458 [true, true],
459] as const) {
460 it.effect(
461 `persists shutdown recovery before stopping providers when enabled=${enabled}, completed=${completed}`,
462 () =>
463 Effect.gen(function* () {
464 const codex = makeFakeCodexAdapter();
465 const persistence = yield* Layer.build(
466 ProviderSessionDirectoryLive.pipe(
467 Layer.provide(
468 ProviderSessionRuntime.layer.pipe(Layer.provide(SqlitePersistenceMemory)),
469 ),
470 ),
471 );
472 const directory = yield* ProviderSessionDirectory.ProviderSessionDirectory.pipe(
473 Effect.provide(persistence),
474 );
475 const threadId = asThreadId("shutdown-recovery");
476 const turnId = asTurnId("shutdown-recovery-turn");
477 const scope = yield* Scope.make();
478 const services = yield* Layer.build(
479 makeProviderServiceLive().pipe(
480 Layer.provide(
481 Layer.succeed(ProviderSessionDirectory.ProviderSessionDirectory, directory),
482 ),
483 Layer.provide(
484 Layer.succeed(
485 ProviderAdapterRegistry.ProviderAdapterRegistry,
486 makeStaticInstanceRegistry([[codexInstanceId, codex.adapter]]),
487 ),
488 ),
489 Layer.provide(ServerSettings.layerTest({ continueThreadsAfterServerUpdate: enabled })),
490 Layer.provide(serverConfigTestLayer),
491 Layer.provide(AnalyticsService.layerTest),
492 Layer.provide(
493 Layer.succeed(
494 ProviderEventLoggers.ProviderEventLoggers,
495 ProviderEventLoggers.NoOpProviderEventLoggers,
496 ),
497 ),
498 ),
499 ).pipe(Scope.provide(scope));
500 const provider = yield* ProviderService.ProviderService.pipe(Effect.provide(services));
501 const session = yield* provider.startSession(threadId, {
502 provider: CODEX_DRIVER,
503 providerInstanceId: codexInstanceId,
504 threadId,
505 runtimeMode: "full-access",
506 });
507 codex.listSessions.mockReturnValue(
508 Effect.succeed([
509 {
510 ...session,
511 status: completed ? "ready" : "running",
512 activeTurnId: completed ? undefined : turnId,
513 },
514 ]),
515 );
516 const pending = yield* directory.getBinding(threadId);
517 assert(Option.isSome(pending));
518 yield* directory.upsert({
519 ...pending.value,
520 runtimePayload: { activeTurnId: null, continueAfterServerUpdate: turnId },
521 });
522 const accepted = yield* provider.sendTurn({ threadId, continuation: true });
523 const admitted = yield* directory.getBinding(threadId);
524 assert(Option.isSome(admitted));
525 assert.propertyVal(admitted.value.runtimePayload, "activeTurnId", accepted.turnId);
526 assert.propertyVal(admitted.value.runtimePayload, "continueAfterServerUpdate", null);
527 if (completed) {
528 // Updates can mark an already-admitted turn immediately before it finishes.
529 yield* directory.upsert({
530 ...admitted.value,
531 runtimePayload: {
532 continueAfterServerUpdate: accepted.turnId,
533 continueAfterServerUpdatePrepared: null,
534 },
535 });
536 }
537 const markers: unknown[] = [];
538 codex.stopAll.mockImplementation(() =>
539 Effect.gen(function* () {
540 const binding = yield* directory.getBinding(threadId);
541 assert(Option.isSome(binding));
542 markers.push(binding.value.runtimePayload);
543 }).pipe(Effect.orDie),
544 );
545 yield* Scope.close(scope, Exit.void);
546 const binding = yield* directory.getBinding(threadId);
547 assert(Option.isSome(binding));
548 assert.equal(codex.stopAll.mock.calls.length, 1);
549 assert.deepStrictEqual(binding.value.resumeCursor, session.resumeCursor);
550 assert.equal(binding.value.status, "stopped");
551 assert.propertyVal(markers[0], "activeTurnId", completed ? null : turnId);
552 if (enabled && !completed) {
553 assert.propertyVal(markers[0], "continueAfterServerUpdate", turnId);
554 assert.propertyVal(binding.value.runtimePayload, "continueAfterServerUpdate", turnId);
555 } else if (completed) {
556 assert.propertyVal(
557 binding.value.runtimePayload,
558 "continueAfterServerUpdate",
559 accepted.turnId,
560 );
561 assert.propertyVal(
562 binding.value.runtimePayload,
563 "continueAfterServerUpdatePrepared",
564 null,
565 );
566 } else {
567 assert.propertyVal(markers[0], "continueAfterServerUpdate", null);
568 assert.propertyVal(binding.value.runtimePayload, "continueAfterServerUpdate", null);
569 }
570 }).pipe(Effect.provide(NodeServices.layer)),
571 );
572}
573
455it.effect("ProviderServiceLive catches stopAll failures during shutdown", () =>574it.effect("ProviderServiceLive catches stopAll failures during shutdown", () =>
456 Effect.gen(function* () {575 Effect.gen(function* () {
457 const codex = makeFakeCodexAdapter();576 const codex = makeFakeCodexAdapter();
apps/server/src/provider/Layers/ProviderService.ts+17
...@@ -223,6 +223,7 @@ function toRuntimePayloadFromSession(...@@ -223,6 +223,7 @@ function toRuntimePayloadFromSession(
223 session: ProviderSession,223 session: ProviderSession,
224 extra?: {224 extra?: {
225 readonly modelSelection?: unknown;225 readonly modelSelection?: unknown;
226 readonly continueAfterServerUpdate?: TurnId;
226 readonly lastRuntimeEvent?: string;227 readonly lastRuntimeEvent?: string;
227 readonly lastRuntimeEventAt?: string;228 readonly lastRuntimeEventAt?: string;
228 },229 },
...@@ -232,6 +233,9 @@ function toRuntimePayloadFromSession(...@@ -232,6 +233,9 @@ function toRuntimePayloadFromSession(
232 model: session.model ?? null,233 model: session.model ?? null,
233 activeTurnId: session.activeTurnId ?? null,234 activeTurnId: session.activeTurnId ?? null,
234 lastError: session.lastError ?? null,235 lastError: session.lastError ?? null,
236 ...(extra?.continueAfterServerUpdate !== undefined
237 ? { continueAfterServerUpdate: extra.continueAfterServerUpdate }
238 : {}),
235 ...(extra?.modelSelection !== undefined ? { modelSelection: extra.modelSelection } : {}),239 ...(extra?.modelSelection !== undefined ? { modelSelection: extra.modelSelection } : {}),
236 ...(extra?.lastRuntimeEvent !== undefined ? { lastRuntimeEvent: extra.lastRuntimeEvent } : {}),240 ...(extra?.lastRuntimeEvent !== undefined ? { lastRuntimeEvent: extra.lastRuntimeEvent } : {}),
237 ...(extra?.lastRuntimeEventAt !== undefined241 ...(extra?.lastRuntimeEventAt !== undefined
...@@ -831,6 +835,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* (...@@ -831,6 +835,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* (
831 threadId: ThreadId,835 threadId: ThreadId,
832 extra?: {836 extra?: {
833 readonly modelSelection?: unknown;837 readonly modelSelection?: unknown;
838 readonly continueAfterServerUpdate?: TurnId;
834 readonly lastRuntimeEvent?: string;839 readonly lastRuntimeEvent?: string;
835 readonly lastRuntimeEventAt?: string;840 readonly lastRuntimeEventAt?: string;
836 },841 },
...@@ -1433,6 +1438,9 @@ const makeProviderService = Effect.fn("makeProviderService")(function* (...@@ -1433,6 +1438,9 @@ const makeProviderService = Effect.fn("makeProviderService")(function* (
1433 runtimePayload: {1438 runtimePayload: {
1434 ...(input.modelSelection !== undefined ? { modelSelection: input.modelSelection } : {}),1439 ...(input.modelSelection !== undefined ? { modelSelection: input.modelSelection } : {}),
1435 activeTurnId: turn.turnId,1440 activeTurnId: turn.turnId,
1441 // Admission and marker consumption must survive the same restart.
1442 continueAfterServerUpdate: null,
1443 continueAfterServerUpdatePrepared: null,
1436 lastRuntimeEvent: "provider.sendTurn",1444 lastRuntimeEvent: "provider.sendTurn",
1437 lastRuntimeEventAt: yield* nowIso,1445 lastRuntimeEventAt: yield* nowIso,
1438 },1446 },
...@@ -1733,6 +1741,8 @@ const makeProviderService = Effect.fn("makeProviderService")(function* (...@@ -1733,6 +1741,8 @@ const makeProviderService = Effect.fn("makeProviderService")(function* (
1733 status: "stopped",1741 status: "stopped",
1734 runtimePayload: {1742 runtimePayload: {
1735 activeTurnId: null,1743 activeTurnId: null,
1744 continueAfterServerUpdate: null,
1745 continueAfterServerUpdatePrepared: null,
1736 },1746 },
1737 });1747 });
1738 yield* analytics.record("provider.session.stopped", {1748 yield* analytics.record("provider.session.stopped", {
...@@ -1942,6 +1952,10 @@ const makeProviderService = Effect.fn("makeProviderService")(function* (...@@ -1942,6 +1952,10 @@ const makeProviderService = Effect.fn("makeProviderService")(function* (
1942 );1952 );
19431953
1944 const runStopAll = Effect.fn("runStopAll")(function* () {1954 const runStopAll = Effect.fn("runStopAll")(function* () {
1955 const continueAfterRestart = yield* serverSettings.getSettings.pipe(
1956 Effect.map((settings) => settings.continueThreadsAfterServerUpdate),
1957 Effect.orElseSucceed(() => false),
1958 );
1945 const properties = yield* Ref.modify(turnAnalytics, (state) => {1959 const properties = yield* Ref.modify(turnAnalytics, (state) => {
1946 const completed: Array<Readonly<Record<string, unknown>>> = [];1960 const completed: Array<Readonly<Record<string, unknown>>> = [];
1947 for (const [sessionKey, session] of state.sessions) {1961 for (const [sessionKey, session] of state.sessions) {
...@@ -1969,6 +1983,9 @@ const makeProviderService = Effect.fn("makeProviderService")(function* (...@@ -1969,6 +1983,9 @@ const makeProviderService = Effect.fn("makeProviderService")(function* (
1969 yield* Effect.forEach(activeSessions, (session) =>1983 yield* Effect.forEach(activeSessions, (session) =>
1970 Effect.flatMap(nowIso, (lastRuntimeEventAt) =>1984 Effect.flatMap(nowIso, (lastRuntimeEventAt) =>
1971 upsertSessionBinding(session, session.threadId, {1985 upsertSessionBinding(session, session.threadId, {
1986 ...(continueAfterRestart && session.status === "running" && session.activeTurnId
1987 ? { continueAfterServerUpdate: session.activeTurnId }
1988 : {}),
1972 lastRuntimeEvent: "provider.stopAll",1989 lastRuntimeEvent: "provider.stopAll",
1973 lastRuntimeEventAt,1990 lastRuntimeEventAt,
1974 }),1991 }),
apps/server/src/serverRuntimeStartup.reconcile.test.ts+469-160
...@@ -1,6 +1,7 @@...@@ -1,6 +1,7 @@
1import * as NodeServices from "@effect/platform-node/NodeServices";1import * as NodeServices from "@effect/platform-node/NodeServices";
2import {2import {
3 type OrchestrationCommand,3 type OrchestrationCommand,
4 type OrchestrationSessionStatus,
4 ProviderDriverKind,5 ProviderDriverKind,
5 ProviderInstanceId,6 ProviderInstanceId,
6 type ProviderSendTurnInput,7 type ProviderSendTurnInput,
...@@ -10,15 +11,21 @@ import {...@@ -10,15 +11,21 @@ import {
10import { assert, it } from "@effect/vitest";11import { assert, it } from "@effect/vitest";
11import * as Deferred from "effect/Deferred";12import * as Deferred from "effect/Deferred";
12import * as Effect from "effect/Effect";13import * as Effect from "effect/Effect";
14import * as Layer from "effect/Layer";
13import * as Option from "effect/Option";15import * as Option from "effect/Option";
14import * as Stream from "effect/Stream";16import * as Stream from "effect/Stream";
1517
16import { OrchestrationCommandInvariantError } from "./orchestration/Errors.ts";18import { OrchestrationCommandInvariantError } from "./orchestration/Errors.ts";
17import * as OrchestrationEngine from "./orchestration/Services/OrchestrationEngine.ts";19import * as OrchestrationEngine from "./orchestration/Services/OrchestrationEngine.ts";
18import * as ProjectionSnapshotQuery from "./orchestration/Services/ProjectionSnapshotQuery.ts";20import * as ProjectionSnapshotQuery from "./orchestration/Services/ProjectionSnapshotQuery.ts";
19import { ProviderSessionDirectoryPersistenceError } from "./provider/Errors.ts";21import {
22 ProviderSessionDirectoryPersistenceError,
23 ProviderSessionNotFoundError,
24} from "./provider/Errors.ts";
20import * as ProviderService from "./provider/Services/ProviderService.ts";25import * as ProviderService from "./provider/Services/ProviderService.ts";
21import * as ProviderSessionDirectory from "./provider/Services/ProviderSessionDirectory.ts";26import * as ProviderSessionDirectory from "./provider/Services/ProviderSessionDirectory.ts";
27import { ServerActivation } from "./serverActivation.ts";
28import * as ServerSettings from "./serverSettings.ts";
22import * as ServerRuntimeStartup from "./serverRuntimeStartup.ts";29import * as ServerRuntimeStartup from "./serverRuntimeStartup.ts";
2330
24const providerInstanceId = ProviderInstanceId.make("codex");31const providerInstanceId = ProviderInstanceId.make("codex");
...@@ -26,7 +33,7 @@ const updatedAt = "2026-08-20T12:00:00.000Z";...@@ -26,7 +33,7 @@ const updatedAt = "2026-08-20T12:00:00.000Z";
2633
27const makeThread = (34const makeThread = (
28 id: string,35 id: string,
29 status: "starting" | "running" | "ready" | "stopped" | "error",36 status: OrchestrationSessionStatus,
30 activeTurnId: TurnId | null = null,37 activeTurnId: TurnId | null = null,
31 archivedAt: string | null = null,38 archivedAt: string | null = null,
32 deletedAt: string | null = null,39 deletedAt: string | null = null,
...@@ -73,6 +80,7 @@ const queryWithThreads = (threads: ReadonlyArray<ReturnType<typeof makeThread>>)...@@ -73,6 +80,7 @@ const queryWithThreads = (threads: ReadonlyArray<ReturnType<typeof makeThread>>)
7380
74const runReconciliation = (input: {81const runReconciliation = (input: {
75 readonly threads: ReadonlyArray<ReturnType<typeof makeThread>>;82 readonly threads: ReadonlyArray<ReturnType<typeof makeThread>>;
83 readonly continueAfterRestart?: boolean;
76 readonly liveThreadIds?: ReadonlyArray<ThreadId>;84 readonly liveThreadIds?: ReadonlyArray<ThreadId>;
77 readonly providerService?: ProviderService.ProviderService["Service"];85 readonly providerService?: ProviderService.ProviderService["Service"];
78 readonly directory: ProviderSessionDirectory.ProviderSessionDirectory["Service"];86 readonly directory: ProviderSessionDirectory.ProviderSessionDirectory["Service"];
...@@ -97,7 +105,14 @@ const runReconciliation = (input: {...@@ -97,7 +105,14 @@ const runReconciliation = (input: {
97 subscribeDomainEvents: Effect.succeed(Stream.empty),105 subscribeDomainEvents: Effect.succeed(Stream.empty),
98 latestSequence: Effect.succeed(0),106 latestSequence: Effect.succeed(0),
99 }),107 }),
100 Effect.provide(NodeServices.layer),108 Effect.provide(
109 Layer.mergeAll(
110 ServerSettings.layerTest({
111 continueThreadsAfterServerUpdate: input.continueAfterRestart ?? false,
112 }),
113 NodeServices.layer,
114 ),
115 ),
101 );116 );
102117
103it.effect("marks active running sessions that have persisted resume state", () => {118it.effect("marks active running sessions that have persisted resume state", () => {
...@@ -138,7 +153,7 @@ it.effect("marks active running sessions that have persisted resume state", () =...@@ -138,7 +153,7 @@ it.effect("marks active running sessions that have persisted resume state", () =
138 upsert: (binding) => Effect.sync(() => upserts.push(binding)),153 upsert: (binding) => Effect.sync(() => upserts.push(binding)),
139 getProvider: () => Effect.die("unused"),154 getProvider: () => Effect.die("unused"),
140 listThreadIds: () => Effect.die("unused"),155 listThreadIds: () => Effect.die("unused"),
141 listBindings: () => Effect.die("unused"),156 listBindings: () => Effect.succeed([]),
142 }),157 }),
143 Effect.tap((marked) =>158 Effect.tap((marked) =>
144 Effect.sync(() => {159 Effect.sync(() => {
...@@ -147,169 +162,186 @@ it.effect("marks active running sessions that have persisted resume state", () =...@@ -147,169 +162,186 @@ it.effect("marks active running sessions that have persisted resume state", () =
147 assert.deepStrictEqual(upserts[0]?.runtimePayload, {162 assert.deepStrictEqual(upserts[0]?.runtimePayload, {
148 activeTurnId: "turn-mark-active",163 activeTurnId: "turn-mark-active",
149 continueAfterServerUpdate: active.session.activeTurnId,164 continueAfterServerUpdate: active.session.activeTurnId,
165 continueAfterServerUpdatePrepared: null,
150 });166 });
151 }),167 }),
152 ),168 ),
153 );169 );
154});170});
155171
156it.effect("continues marked sessions after activation with provider-specific input", () =>172it.effect.each(["marked update", "opt-in restart"] as const)(
157 Effect.gen(function* () {173 "continues %s sessions after activation with provider-specific input",
158 const codex = makeThread(174 (recovery) =>
159 "thread-continue-codex",175 Effect.gen(function* () {
160 "running",176 const codex = makeThread(
161 TurnId.make("turn-continue-codex"),177 "thread-continue-codex",
162 );178 "running",
163 const fallback = makeThread("thread-continue-fallback", "starting");179 TurnId.make("turn-continue-codex"),
164 const fallbackContinuationTurnId = TurnId.make("turn-continue-fallback");180 );
165 const fallbackProviderInstanceId = ProviderInstanceId.make("claudeAgent");181 const fallbackContinuationTurnId = TurnId.make("turn-continue-fallback");
166 const continuationSent = yield* Deferred.make<void>();182 const fallback = makeThread(
167 const continuationCleared = yield* Deferred.make<void>();183 "thread-continue-fallback",
168 const sends: ProviderSendTurnInput[] = [];184 recovery === "marked update" ? "starting" : "running",
169 const dispatched: OrchestrationCommand[] = [];185 recovery === "marked update" ? null : fallbackContinuationTurnId,
170 const upserts: ProviderSessionDirectory.ProviderRuntimeBinding[] = [];186 );
171 const bindings = new Map<ThreadId, ProviderSessionDirectory.ProviderRuntimeBinding>(187 const fallbackProviderInstanceId = ProviderInstanceId.make("claudeAgent");
172 [codex, fallback].map((thread) => [188 const continuationSent = yield* Deferred.make<void>();
173 thread.id,189 const continuationCleared = yield* Deferred.make<void>();
174 {190 const sends: ProviderSendTurnInput[] = [];
175 threadId: thread.id,191 const dispatched: OrchestrationCommand[] = [];
176 provider:192 const upserts: ProviderSessionDirectory.ProviderRuntimeBinding[] = [];
177 thread.id === codex.id193 const bindings = new Map<ThreadId, ProviderSessionDirectory.ProviderRuntimeBinding>(
178 ? ProviderDriverKind.make("codex")194 [codex, fallback].map((thread) => [
179 : ProviderDriverKind.make("claudeAgent"),195 thread.id,
180 providerInstanceId:196 {
181 thread.id === codex.id ? providerInstanceId : fallbackProviderInstanceId,197 threadId: thread.id,
182 status: "running" as const,198 provider:
183 runtimePayload: {199 thread.id === codex.id
184 continueAfterServerUpdate:200 ? ProviderDriverKind.make("codex")
185 thread.id === codex.id ? codex.session.activeTurnId : fallbackContinuationTurnId,201 : ProviderDriverKind.make("claudeAgent"),
202 providerInstanceId:
203 thread.id === codex.id ? providerInstanceId : fallbackProviderInstanceId,
204 status: "running" as const,
205 resumeCursor: { threadId: thread.id },
206 runtimePayload:
207 recovery === "marked update"
208 ? {
209 continueAfterServerUpdate:
210 thread.id === codex.id
211 ? codex.session.activeTurnId
212 : fallbackContinuationTurnId,
213 }
214 : { activeTurnId: thread.session.activeTurnId },
186 },215 },
187 },216 ]),
188 ]),217 );
189 );218 const providerService: ProviderService.ProviderService["Service"] = {
190 const providerService: ProviderService.ProviderService["Service"] = {219 ...makeProviderService(),
191 ...makeProviderService(),220 getCapabilities: (instanceId) =>
192 getCapabilities: (instanceId) =>221 Effect.succeed({
193 Effect.succeed({222 sessionModelSwitch: "in-session",
194 sessionModelSwitch: "in-session",223 ...(instanceId === providerInstanceId ? { promptlessTurnContinuation: true } : {}),
195 ...(instanceId === providerInstanceId ? { promptlessTurnContinuation: true } : {}),
196 }),
197 sendTurn: (input) =>
198 Effect.gen(function* () {
199 sends.push(input);
200 if (sends.length === 2) {
201 yield* Deferred.succeed(continuationSent, undefined);
202 }
203 return {
204 threadId: input.threadId,
205 turnId: TurnId.make(`continued-${String(input.threadId)}`),
206 };
207 }),
208 };
209
210 yield* runReconciliation({
211 threads: [codex, fallback],
212 providerService,
213 directory: {
214 getBinding: (threadId) =>
215 Effect.sync(() => {
216 const binding = bindings.get(threadId);
217 return binding === undefined ? Option.none() : Option.some(binding);
218 }),224 }),
219 upsert: (binding) =>225 sendTurn: (input) =>
220 Effect.sync(() => {226 Effect.gen(function* () {
221 bindings.set(binding.threadId, binding);227 sends.push(input);
222 upserts.push(binding);228 if (sends.length === 2) {
223 const clearedCount = upserts.filter((candidate) => {229 yield* Deferred.succeed(continuationSent, undefined);
224 const payload = candidate.runtimePayload;230 }
225 return (231 return {
226 payload !== null &&232 threadId: input.threadId,
227 typeof payload === "object" &&233 turnId: TurnId.make(`continued-${String(input.threadId)}`),
228 !Array.isArray(payload) &&234 };
229 "continueAfterServerUpdate" in payload &&235 }),
230 payload.continueAfterServerUpdate === null236 };
231 );237
232 }).length;238 yield* runReconciliation({
233 return clearedCount === 1;239 threads: [codex, fallback],
234 }).pipe(240 continueAfterRestart: recovery === "opt-in restart",
235 Effect.flatMap((firstMarkerCleared) =>241 providerService,
236 firstMarkerCleared ? Deferred.succeed(continuationCleared, undefined) : Effect.void,242 directory: {
243 getBinding: (threadId) =>
244 Effect.sync(() => {
245 const binding = bindings.get(threadId);
246 return binding === undefined ? Option.none() : Option.some(binding);
247 }),
248 upsert: (binding) =>
249 Effect.sync(() => {
250 bindings.set(binding.threadId, binding);
251 upserts.push(binding);
252 const clearedCount = upserts.filter((candidate) => {
253 const payload = candidate.runtimePayload;
254 return (
255 payload !== null &&
256 typeof payload === "object" &&
257 !Array.isArray(payload) &&
258 "continueAfterServerUpdate" in payload &&
259 payload.continueAfterServerUpdate === null
260 );
261 }).length;
262 return clearedCount === 1;
263 }).pipe(
264 Effect.flatMap((firstMarkerCleared) =>
265 firstMarkerCleared ? Deferred.succeed(continuationCleared, undefined) : Effect.void,
266 ),
237 ),267 ),
268 getProvider: () => Effect.die("unused"),
269 listThreadIds: () => Effect.die("unused"),
270 listBindings: () => Effect.succeed([]),
271 },
272 dispatch: (command) =>
273 Effect.sync(() => dispatched.push(command)).pipe(
274 Effect.as({ sequence: dispatched.length }),
238 ),275 ),
239 getProvider: () => Effect.die("unused"),276 });
240 listThreadIds: () => Effect.die("unused"),277 yield* Deferred.await(continuationSent);
241 listBindings: () => Effect.die("unused"),278 yield* Deferred.await(continuationCleared);
242 },
243 dispatch: (command) =>
244 Effect.sync(() => dispatched.push(command)).pipe(
245 Effect.as({ sequence: dispatched.length }),
246 ),
247 });
248 yield* Deferred.await(continuationSent);
249 yield* Deferred.await(continuationCleared);
250279
251 assert.deepStrictEqual(
252 sends.toSorted((left, right) => String(left.threadId).localeCompare(String(right.threadId))),
253 [
254 { threadId: codex.id, continuation: true, interactionMode: "default" },
255 {
256 threadId: fallback.id,
257 input: "Continue where you left off.",
258 interactionMode: "default",
259 },
260 ],
261 );
262 assert.deepStrictEqual(
263 dispatched.map((command) =>
264 command.type === "thread.session.set"
265 ? {
266 threadId: command.threadId,
267 status: command.session.status,
268 activeTurnId: command.session.activeTurnId,
269 }
270 : null,
271 ),
272 [
273 {
274 threadId: codex.id,
275 status: "starting",
276 activeTurnId: null,
277 },
278 {
279 threadId: fallback.id,
280 status: "starting",
281 activeTurnId: fallback.session.activeTurnId,
282 },
283 ],
284 );
285 for (const [thread, continuationTurnId] of [
286 [codex, codex.session.activeTurnId],
287 [fallback, fallbackContinuationTurnId],
288 ] as const) {
289 assert.deepStrictEqual(280 assert.deepStrictEqual(
290 upserts281 sends.toSorted((left, right) =>
291 .filter((binding) => binding.threadId === thread.id)282 String(left.threadId).localeCompare(String(right.threadId)),
292 .map((binding) => binding.runtimePayload)[0],283 ),
293 {284 [
294 continueAfterServerUpdate: continuationTurnId,285 { threadId: codex.id, continuation: true, interactionMode: "default" },
295 activeTurnId: null,286 {
296 },287 threadId: fallback.id,
288 input: "Continue where you left off.",
289 interactionMode: "default",
290 },
291 ],
292 );
293 assert.deepStrictEqual(
294 dispatched.map((command) =>
295 command.type === "thread.session.set"
296 ? {
297 threadId: command.threadId,
298 status: command.session.status,
299 activeTurnId: command.session.activeTurnId,
300 }
301 : null,
302 ),
303 [
304 {
305 threadId: codex.id,
306 status: "starting",
307 activeTurnId: null,
308 },
309 {
310 threadId: fallback.id,
311 status: "starting",
312 activeTurnId: null,
313 },
314 ],
297 );315 );
298 }316 for (const [thread, continuationTurnId] of [
299 assert.equal(317 [codex, codex.session.activeTurnId],
300 upserts.some((binding) => {318 [fallback, fallbackContinuationTurnId],
301 const payload = binding.runtimePayload;319 ] as const) {
302 return (320 assert.deepStrictEqual(
303 payload !== null &&321 upserts
304 typeof payload === "object" &&322 .filter((binding) => binding.threadId === thread.id)
305 !Array.isArray(payload) &&323 .map((binding) => binding.runtimePayload)[0],
306 "continueAfterServerUpdate" in payload &&324 {
307 payload.continueAfterServerUpdate === null325 continueAfterServerUpdate: continuationTurnId,
326 continueAfterServerUpdatePrepared: true,
327 activeTurnId: null,
328 },
308 );329 );
309 }),330 }
310 true,331 assert.equal(
311 );332 upserts.some((binding) => {
312 }),333 const payload = binding.runtimePayload;
334 return (
335 payload !== null &&
336 typeof payload === "object" &&
337 !Array.isArray(payload) &&
338 "continueAfterServerUpdate" in payload &&
339 payload.continueAfterServerUpdate === null
340 );
341 }),
342 true,
343 );
344 }),
313);345);
314346
315it.effect("does not continue archived or deleted marked sessions", () => {347it.effect("does not continue archived or deleted marked sessions", () => {
...@@ -361,7 +393,7 @@ it.effect("does not continue archived or deleted marked sessions", () => {...@@ -361,7 +393,7 @@ it.effect("does not continue archived or deleted marked sessions", () => {
361 upsert: () => Effect.void,393 upsert: () => Effect.void,
362 getProvider: () => Effect.die("unused"),394 getProvider: () => Effect.die("unused"),
363 listThreadIds: () => Effect.die("unused"),395 listThreadIds: () => Effect.die("unused"),
364 listBindings: () => Effect.die("unused"),396 listBindings: () => Effect.succeed([]),
365 },397 },
366 dispatch: (command) =>398 dispatch: (command) =>
367 Effect.sync(() => dispatched.push(command)).pipe(Effect.as({ sequence: dispatched.length })),399 Effect.sync(() => dispatched.push(command)).pipe(Effect.as({ sequence: dispatched.length })),
...@@ -416,7 +448,7 @@ it.effect("retries continuation preparation before settling a persistent failure...@@ -416,7 +448,7 @@ it.effect("retries continuation preparation before settling a persistent failure
416 upsert: () => Effect.void,448 upsert: () => Effect.void,
417 getProvider: () => Effect.die("unused"),449 getProvider: () => Effect.die("unused"),
418 listThreadIds: () => Effect.die("unused"),450 listThreadIds: () => Effect.die("unused"),
419 listBindings: () => Effect.die("unused"),451 listBindings: () => Effect.succeed([]),
420 },452 },
421 dispatch: (command) => {453 dispatch: (command) => {
422 if (command.type !== "thread.session.set") {454 if (command.type !== "thread.session.set") {
...@@ -487,7 +519,7 @@ it.effect("reconciles multiple active and archived orphans but skips live sessio...@@ -487,7 +519,7 @@ it.effect("reconciles multiple active and archived orphans but skips live sessio
487 upsert: (binding) => Effect.sync(() => upserts.push(binding)),519 upsert: (binding) => Effect.sync(() => upserts.push(binding)),
488 getProvider: () => Effect.die("unused"),520 getProvider: () => Effect.die("unused"),
489 listThreadIds: () => Effect.die("unused"),521 listThreadIds: () => Effect.die("unused"),
490 listBindings: () => Effect.die("unused"),522 listBindings: () => Effect.succeed([]),
491 },523 },
492 dispatch: (command) =>524 dispatch: (command) =>
493 Effect.sync(() => dispatched.push(command)).pipe(Effect.as({ sequence: dispatched.length })),525 Effect.sync(() => dispatched.push(command)).pipe(Effect.as({ sequence: dispatched.length })),
...@@ -521,6 +553,7 @@ it.effect("reconciles multiple active and archived orphans but skips live sessio...@@ -521,6 +553,7 @@ it.effect("reconciles multiple active and archived orphans but skips live sessio
521 activeTurnId: null,553 activeTurnId: null,
522 unrelated: binding.threadId,554 unrelated: binding.threadId,
523 continueAfterServerUpdate: null,555 continueAfterServerUpdate: null,
556 continueAfterServerUpdatePrepared: null,
524 }557 }
525 : { activeTurnId: null, unrelated: binding.threadId },558 : { activeTurnId: null, unrelated: binding.threadId },
526 );559 );
...@@ -565,7 +598,7 @@ it.effect(...@@ -565,7 +598,7 @@ it.effect(
565 upsert: () => Effect.fail(writeFailure),598 upsert: () => Effect.fail(writeFailure),
566 getProvider: () => Effect.die("unused"),599 getProvider: () => Effect.die("unused"),
567 listThreadIds: () => Effect.die("unused"),600 listThreadIds: () => Effect.die("unused"),
568 listBindings: () => Effect.die("unused"),601 listBindings: () => Effect.succeed([]),
569 },602 },
570 dispatch: (command) =>603 dispatch: (command) =>
571 Effect.sync(() => dispatched.push(command)).pipe(604 Effect.sync(() => dispatched.push(command)).pipe(
...@@ -602,7 +635,7 @@ it.effect("retries failed projections and continues after a persistent failure",...@@ -602,7 +635,7 @@ it.effect("retries failed projections and continues after a persistent failure",
602 upsert: () => Effect.void,635 upsert: () => Effect.void,
603 getProvider: () => Effect.die("unused"),636 getProvider: () => Effect.die("unused"),
604 listThreadIds: () => Effect.die("unused"),637 listThreadIds: () => Effect.die("unused"),
605 listBindings: () => Effect.die("unused"),638 listBindings: () => Effect.succeed([]),
606 },639 },
607 dispatch: (command) => {640 dispatch: (command) => {
608 if (command.type !== "thread.session.set") {641 if (command.type !== "thread.session.set") {
...@@ -651,7 +684,7 @@ it.effect("does not fail startup when the live provider session inventory cannot...@@ -651,7 +684,7 @@ it.effect("does not fail startup when the live provider session inventory cannot
651 upsert: () => Effect.die("unused"),684 upsert: () => Effect.die("unused"),
652 getProvider: () => Effect.die("unused"),685 getProvider: () => Effect.die("unused"),
653 listThreadIds: () => Effect.die("unused"),686 listThreadIds: () => Effect.die("unused"),
654 listBindings: () => Effect.die("unused"),687 listBindings: () => Effect.succeed([]),
655 }),688 }),
656 Effect.provideService(OrchestrationEngine.OrchestrationEngineService, {689 Effect.provideService(OrchestrationEngine.OrchestrationEngineService, {
657 readEvents: () => Stream.empty,690 readEvents: () => Stream.empty,
...@@ -662,7 +695,283 @@ it.effect("does not fail startup when the live provider session inventory cannot...@@ -662,7 +695,283 @@ it.effect("does not fail startup when the live provider session inventory cannot
662 subscribeDomainEvents: Effect.succeed(Stream.empty),695 subscribeDomainEvents: Effect.succeed(Stream.empty),
663 latestSequence: Effect.succeed(0),696 latestSequence: Effect.succeed(0),
664 }),697 }),
665 Effect.provide(NodeServices.layer),698 Effect.provide(Layer.mergeAll(NodeServices.layer, ServerSettings.layerTest())),
666 Effect.tap(() => Effect.sync(() => assert.equal(queried, false))),699 Effect.tap(() => Effect.sync(() => assert.equal(queried, false))),
667 );700 );
668});701});
702
703for (const scenario of [
704 "disabled",
705 "stopped projection",
706 "finished projection",
707 "stopped binding",
708 "finished binding",
709 "missing cursor",
710 "mismatched turn",
711 "marked without cursor",
712 "marked stopped projection",
713 "marked superseded turn",
714] as const) {
715 it.effect(`does not recover an interrupted session with ${scenario}`, () => {
716 const turnId = TurnId.make("turn-excluded-recovery");
717 const thread = makeThread(
718 "thread-excluded-recovery",
719 scenario.includes("stopped projection")
720 ? "stopped"
721 : scenario === "finished projection"
722 ? "ready"
723 : scenario === "marked superseded turn"
724 ? "starting"
725 : "running",
726 scenario === "marked superseded turn" ? null : turnId,
727 );
728 const dispatched: OrchestrationCommand[] = [];
729 const upserts: ProviderSessionDirectory.ProviderRuntimeBinding[] = [];
730 return runReconciliation({
731 threads: [thread],
732 continueAfterRestart: scenario !== "disabled",
733 directory: {
734 getBinding: () =>
735 Effect.succeed(
736 Option.some({
737 threadId: thread.id,
738 provider: ProviderDriverKind.make("codex"),
739 providerInstanceId,
740 status: scenario === "stopped binding" ? "stopped" : "running",
741 ...(scenario.includes("cursor") ? {} : { resumeCursor: { threadId: thread.id } }),
742 runtimePayload: {
743 activeTurnId:
744 scenario === "finished binding"
745 ? null
746 : scenario === "mismatched turn" || scenario === "marked superseded turn"
747 ? "another-turn"
748 : turnId,
749 ...(scenario.startsWith("marked") ? { continueAfterServerUpdate: turnId } : {}),
750 },
751 }),
752 ),
753 upsert: (binding) =>
754 Effect.sync(() => {
755 upserts.push(binding);
756 }),
757 getProvider: () => Effect.die("unused"),
758 listThreadIds: () => Effect.die("unused"),
759 listBindings: () => Effect.succeed([]),
760 },
761 dispatch: (command) =>
762 Effect.sync(() => {
763 dispatched.push(command);
764 return { sequence: dispatched.length };
765 }),
766 }).pipe(
767 Effect.tap(() =>
768 Effect.sync(() => {
769 assert.deepStrictEqual(
770 dispatched.map(
771 (command) => command.type === "thread.session.set" && command.session.status,
772 ),
773 ["error"],
774 );
775 assert.deepStrictEqual(
776 upserts.map((binding) => binding.status),
777 ["stopped"],
778 );
779 }),
780 ),
781 );
782 });
783}
784
785for (const preparedStatus of [
786 "starting",
787 "ready",
788 "ready with failed scan",
789 "completed after update marking",
790] as const) {
791 it.effect(`recovers again if startup exits with a prepared ${preparedStatus} session`, () =>
792 Effect.gen(function* () {
793 const turnId = TurnId.make("turn-interrupted-startup");
794 const thread = makeThread("thread-interrupted-startup", "running", turnId);
795 const activation = yield* Deferred.make<void>();
796 const cleared = yield* Deferred.make<void>();
797 const sends: ProviderSendTurnInput[] = [];
798 let binding: ProviderSessionDirectory.ProviderRuntimeBinding = {
799 threadId: thread.id,
800 provider: ProviderDriverKind.make("codex"),
801 providerInstanceId,
802 status: "running",
803 resumeCursor: { threadId: thread.id },
804 runtimePayload: { activeTurnId: turnId },
805 };
806 const input = {
807 threads: [thread],
808 continueAfterRestart: true,
809 providerService: {
810 ...makeProviderService(),
811 getCapabilities: () =>
812 Effect.succeed({
813 sessionModelSwitch: "in-session" as const,
814 promptlessTurnContinuation: true,
815 }),
816 sendTurn: (input: ProviderSendTurnInput) =>
817 Effect.sync(() => {
818 sends.push(input);
819 return { threadId: input.threadId, turnId: TurnId.make("turn-recovered") };
820 }),
821 },
822 directory: {
823 getBinding: () => Effect.sync(() => Option.some(binding)),
824 upsert: (next: ProviderSessionDirectory.ProviderRuntimeBinding) =>
825 Effect.gen(function* () {
826 binding = next;
827 if (binding.status !== "starting" || sends.length === 0) return;
828 yield* Deferred.succeed(cleared, undefined);
829 }),
830 getProvider: () => Effect.die("unused"),
831 listThreadIds: () => Effect.die("unused"),
832 listBindings: () =>
833 preparedStatus === "ready with failed scan"
834 ? Effect.fail(
835 new ProviderSessionDirectoryPersistenceError({
836 operation: "listBindings",
837 detail: "unreadable unrelated binding",
838 }),
839 )
840 : Effect.sync(() => [{ ...binding, lastSeenAt: "2026-01-01T00:00:00.000Z" }]),
841 },
842 dispatch: (command: OrchestrationCommand) =>
843 Effect.sync(() => {
844 if (command.type === "thread.session.set") {
845 thread.session.status = command.session.status;
846 thread.session.activeTurnId = command.session.activeTurnId;
847 }
848 return { sequence: 1 };
849 }),
850 };
851
852 yield* runReconciliation(input).pipe(
853 Effect.provideService(ServerActivation, Deferred.await(activation)),
854 Effect.scoped,
855 );
856 assert.deepStrictEqual(sends, []);
857 assert.equal(thread.session.status, "starting");
858 assert.equal(thread.session.activeTurnId, null);
859 assert.deepStrictEqual(binding.runtimePayload, {
860 activeTurnId: null,
861 continueAfterServerUpdate: turnId,
862 continueAfterServerUpdatePrepared: true,
863 });
864
865 if (preparedStatus === "completed after update marking") {
866 thread.session.status = "ready";
867 binding = {
868 ...binding,
869 status: "stopped",
870 runtimePayload: {
871 activeTurnId: null,
872 continueAfterServerUpdate: turnId,
873 continueAfterServerUpdatePrepared: null,
874 },
875 };
876 yield* runReconciliation(input);
877 assert.deepStrictEqual(sends, []);
878 assert.equal(thread.session.status, "ready");
879 return;
880 }
881 thread.session.status =
882 preparedStatus === "ready with failed scan" ? "ready" : preparedStatus;
883 yield* runReconciliation(input);
884 yield* Deferred.await(cleared);
885 assert.deepStrictEqual(sends, [
886 { threadId: thread.id, continuation: true, interactionMode: "default" },
887 ]);
888 assert.deepStrictEqual(binding.runtimePayload, {
889 activeTurnId: null,
890 continueAfterServerUpdate: null,
891 continueAfterServerUpdatePrepared: null,
892 });
893 }),
894 );
895}
896
897it.effect("settles failed opt-in recovery without retrying the provider turn", () =>
898 Effect.gen(function* () {
899 const turnId = TurnId.make("turn-failed-recovery");
900 const thread = makeThread("thread-failed-recovery", "running", turnId);
901 const settled = yield* Deferred.make<void>();
902 const sends: ProviderSendTurnInput[] = [];
903 const dispatched: OrchestrationCommand[] = [];
904 const preparedPayloads: unknown[] = [];
905 let binding: ProviderSessionDirectory.ProviderRuntimeBinding = {
906 threadId: thread.id,
907 provider: ProviderDriverKind.make("codex"),
908 providerInstanceId,
909 status: "running",
910 resumeCursor: { threadId: thread.id },
911 runtimePayload: { activeTurnId: turnId },
912 };
913 yield* runReconciliation({
914 threads: [thread],
915 continueAfterRestart: true,
916 providerService: {
917 ...makeProviderService(),
918 getCapabilities: () =>
919 Effect.succeed({ sessionModelSwitch: "in-session", promptlessTurnContinuation: true }),
920 sendTurn: (input) =>
921 Effect.gen(function* () {
922 sends.push(input);
923 preparedPayloads.push(binding.runtimePayload);
924 return yield* Effect.fail(
925 new ProviderSessionNotFoundError({ threadId: input.threadId }),
926 );
927 }),
928 },
929 directory: {
930 getBinding: () => Effect.sync(() => Option.some(binding)),
931 upsert: (next) =>
932 Effect.sync(() => {
933 binding = next;
934 }),
935 getProvider: () => Effect.die("unused"),
936 listThreadIds: () => Effect.die("unused"),
937 listBindings: () => Effect.succeed([]),
938 },
939 dispatch: (command) =>
940 Effect.gen(function* () {
941 dispatched.push(command);
942 if (command.type === "thread.session.set" && command.session.status === "error") {
943 yield* Deferred.succeed(settled, undefined);
944 }
945 return { sequence: dispatched.length };
946 }),
947 });
948 yield* Deferred.await(settled);
949 assert.equal(sends.length, 1);
950 assert.deepStrictEqual(preparedPayloads, [
951 {
952 activeTurnId: null,
953 continueAfterServerUpdate: turnId,
954 continueAfterServerUpdatePrepared: true,
955 },
956 ]);
957 assert.deepStrictEqual(
958 dispatched.map(
959 (command) =>
960 command.type === "thread.session.set" && {
961 status: command.session.status,
962 activeTurnId: command.session.activeTurnId,
963 },
964 ),
965 [
966 { status: "starting", activeTurnId: null },
967 { status: "error", activeTurnId: null },
968 ],
969 );
970 assert.equal(binding.status, "stopped");
971 assert.deepStrictEqual(binding.runtimePayload, {
972 activeTurnId: null,
973 continueAfterServerUpdate: null,
974 continueAfterServerUpdatePrepared: null,
975 });
976 }),
977);
apps/server/src/serverRuntimeStartup.ts+77-6
...@@ -394,6 +394,7 @@ export const markRunningProviderSessionsForContinuation = Effect.gen(function* (...@@ -394,6 +394,7 @@ export const markRunningProviderSessionsForContinuation = Effect.gen(function* (
394 runtimePayload: {394 runtimePayload: {
395 ...readRuntimePayload(binding.value.runtimePayload),395 ...readRuntimePayload(binding.value.runtimePayload),
396 [SERVER_UPDATE_CONTINUATION_KEY]: activeTurnId,396 [SERVER_UPDATE_CONTINUATION_KEY]: activeTurnId,
397 continueAfterServerUpdatePrepared: null,
397 },398 },
398 });399 });
399 marked.push(thread.id);400 marked.push(thread.id);
...@@ -423,6 +424,7 @@ const clearContinuationMarkers = (...@@ -423,6 +424,7 @@ const clearContinuationMarkers = (
423 runtimePayload: {424 runtimePayload: {
424 ...readRuntimePayload(binding.runtimePayload),425 ...readRuntimePayload(binding.runtimePayload),
425 [SERVER_UPDATE_CONTINUATION_KEY]: null,426 [SERVER_UPDATE_CONTINUATION_KEY]: null,
427 continueAfterServerUpdatePrepared: null,
426 },428 },
427 }),429 }),
428 }),430 }),
...@@ -443,17 +445,56 @@ export const reconcileProviderSessions = Effect.gen(function* () {...@@ -443,17 +445,56 @@ export const reconcileProviderSessions = Effect.gen(function* () {
443 const orchestrationEngine = yield* OrchestrationEngine.OrchestrationEngineService;445 const orchestrationEngine = yield* OrchestrationEngine.OrchestrationEngineService;
444 const providerService = yield* ProviderService.ProviderService;446 const providerService = yield* ProviderService.ProviderService;
445 const query = yield* ProjectionSnapshotQuery.ProjectionSnapshotQuery;447 const query = yield* ProjectionSnapshotQuery.ProjectionSnapshotQuery;
448 const settings = yield* ServerSettings.ServerSettingsService;
449 const continueAfterRestart = yield* settings.getSettings.pipe(
450 Effect.map((value) => value.continueThreadsAfterServerUpdate),
451 Effect.catch((cause) =>
452 Effect.logWarning("could not read restart continuation preference", { cause }).pipe(
453 Effect.as(false),
454 ),
455 ),
456 );
446457
447 const liveThreadIds = new Set(458 const liveThreadIds = new Set(
448 (yield* providerService.listSessions()).map((session) => session.threadId),459 (yield* providerService.listSessions()).map((session) => session.threadId),
449 );460 );
450 const { threads } = yield* query.getCommandReadModel();461 const { threads } = yield* query.getCommandReadModel();
462 // Provider startup can report ready before the continuation is submitted.
463 // Find those markers in one read rather than querying every idle thread.
464 const preparedThreadIds = new Set(
465 (yield* directory.listBindings().pipe(
466 Effect.catch((cause) =>
467 Effect.logWarning("failed to read prepared provider continuations", { cause }).pipe(
468 Effect.andThen(
469 Effect.forEach(
470 threads.filter(
471 (thread) => thread.session?.status === "ready" && !liveThreadIds.has(thread.id),
472 ),
473 (thread) =>
474 directory.getBinding(thread.id).pipe(Effect.orElseSucceed(() => Option.none())),
475 ),
476 ),
477 Effect.map((bindings) =>
478 bindings.flatMap((binding) => (Option.isSome(binding) ? [binding.value] : [])),
479 ),
480 ),
481 ),
482 ))
483 .filter(
484 (binding) =>
485 readServerUpdateContinuationTurnId(binding.runtimePayload) !== null &&
486 readRuntimePayload(binding.runtimePayload).activeTurnId === null &&
487 readRuntimePayload(binding.runtimePayload).continueAfterServerUpdatePrepared === true,
488 )
489 .map((binding) => binding.threadId),
490 );
451 const orphanedThreads = threads.filter(491 const orphanedThreads = threads.filter(
452 (thread) =>492 (thread) =>
453 thread.session !== null &&493 thread.session !== null &&
454 (thread.session.status === "starting" ||494 (thread.session.status === "starting" ||
455 thread.session.status === "running" ||495 thread.session.status === "running" ||
456 thread.session.activeTurnId !== null) &&496 thread.session.activeTurnId !== null ||
497 (thread.session.status === "ready" && preparedThreadIds.has(thread.id))) &&
457 !liveThreadIds.has(thread.id),498 !liveThreadIds.has(thread.id),
458 );499 );
459500
...@@ -479,7 +520,27 @@ export const reconcileProviderSessions = Effect.gen(function* () {...@@ -479,7 +520,27 @@ export const reconcileProviderSessions = Effect.gen(function* () {
479 : null;520 : null;
480 const continuationMarked =521 const continuationMarked =
481 continuationTurnId !== null &&522 continuationTurnId !== null &&
482 (session.activeTurnId === null || continuationTurnId === session.activeTurnId);523 (session.activeTurnId === null || continuationTurnId === session.activeTurnId) &&
524 Option.isSome(binding) &&
525 (readRuntimePayload(binding.value.runtimePayload).activeTurnId == null ||
526 readRuntimePayload(binding.value.runtimePayload).activeTurnId === continuationTurnId);
527 const preparedWhileReady =
528 session.status === "ready" &&
529 session.activeTurnId === null &&
530 continuationMarked &&
531 Option.isSome(binding) &&
532 readRuntimePayload(binding.value.runtimePayload).activeTurnId === null &&
533 readRuntimePayload(binding.value.runtimePayload).continueAfterServerUpdatePrepared === true;
534 // Abrupt shutdowns cannot write an update marker. Require both durable
535 // records to agree on an unfinished turn before recovering one implicitly.
536 const interruptedByRestart =
537 continueAfterRestart &&
538 session.status === "running" &&
539 session.activeTurnId !== null &&
540 Option.isSome(binding) &&
541 binding.value.status === "running" &&
542 binding.value.resumeCursor != null &&
543 readRuntimePayload(binding.value.runtimePayload).activeTurnId === session.activeTurnId;
483 const settleAsError = (lastError: string) =>544 const settleAsError = (lastError: string) =>
484 Effect.gen(function* () {545 Effect.gen(function* () {
485 yield* Effect.gen(function* () {546 yield* Effect.gen(function* () {
...@@ -490,7 +551,12 @@ export const reconcileProviderSessions = Effect.gen(function* () {...@@ -490,7 +551,12 @@ export const reconcileProviderSessions = Effect.gen(function* () {
490 runtimePayload: {551 runtimePayload: {
491 ...readRuntimePayload(binding.value.runtimePayload),552 ...readRuntimePayload(binding.value.runtimePayload),
492 activeTurnId: null,553 activeTurnId: null,
493 ...(continuationMarkerPresent ? { [SERVER_UPDATE_CONTINUATION_KEY]: null } : {}),554 ...(continuationMarkerPresent || interruptedByRestart
555 ? {
556 [SERVER_UPDATE_CONTINUATION_KEY]: null,
557 continueAfterServerUpdatePrepared: null,
558 }
559 : {}),
494 },560 },
495 });561 });
496 }562 }
...@@ -535,7 +601,9 @@ export const reconcileProviderSessions = Effect.gen(function* () {...@@ -535,7 +601,9 @@ export const reconcileProviderSessions = Effect.gen(function* () {
535601
536 if (602 if (
537 Option.isSome(binding) &&603 Option.isSome(binding) &&
538 continuationMarked &&604 (continuationMarked || interruptedByRestart) &&
605 (session.status === "running" || session.status === "starting" || preparedWhileReady) &&
606 binding.value.resumeCursor != null &&
539 thread.archivedAt === null &&607 thread.archivedAt === null &&
540 thread.deletedAt === null608 thread.deletedAt === null
541 ) {609 ) {
...@@ -545,6 +613,9 @@ export const reconcileProviderSessions = Effect.gen(function* () {...@@ -545,6 +613,9 @@ export const reconcileProviderSessions = Effect.gen(function* () {
545 status: "starting",613 status: "starting",
546 runtimePayload: {614 runtimePayload: {
547 ...readRuntimePayload(binding.value.runtimePayload),615 ...readRuntimePayload(binding.value.runtimePayload),
616 // Keep recovery durable if this process also exits before sending.
617 [SERVER_UPDATE_CONTINUATION_KEY]: session.activeTurnId ?? continuationTurnId,
618 continueAfterServerUpdatePrepared: true,
548 activeTurnId: null,619 activeTurnId: null,
549 },620 },
550 });621 });
...@@ -608,12 +679,12 @@ export const reconcileProviderSessions = Effect.gen(function* () {...@@ -608,12 +679,12 @@ export const reconcileProviderSessions = Effect.gen(function* () {
608 }679 }
609 return;680 return;
610 }681 }
611 yield* Effect.logWarning("failed to continue provider session after server update", {682 yield* Effect.logWarning("failed to continue provider session after server restart", {
612 threadId: thread.id,683 threadId: thread.id,
613 cause: continuationExit.cause,684 cause: continuationExit.cause,
614 });685 });
615 yield* settleAsError(686 yield* settleAsError(
616 "Could not continue this thread after the server update. Send a new message to continue.",687 "Could not continue this thread after the server restart. Send a new message to continue.",
617 ).pipe(Effect.ignoreCause);688 ).pipe(Effect.ignoreCause);
618 }),689 }),
619 );690 );
apps/web/src/components/ServerUpdateAction.test.tsx+2-1
...@@ -15,7 +15,8 @@ vi.mock("~/hooks/useCopyToClipboard", () => ({...@@ -15,7 +15,8 @@ vi.mock("~/hooks/useCopyToClipboard", () => ({
15 useCopyToClipboard: () => ({ copyToClipboard: vi.fn() }),15 useCopyToClipboard: () => ({ copyToClipboard: vi.fn() }),
16}));16}));
17vi.mock("~/hooks/useSettings", () => ({17vi.mock("~/hooks/useSettings", () => ({
18 useClientSettings: (18 useEnvironmentSettings: (
19 _environmentId: EnvironmentId,
19 selector: (settings: { continueThreadsAfterServerUpdate: boolean }) => unknown,20 selector: (settings: { continueThreadsAfterServerUpdate: boolean }) => unknown,
20 ) => selector({ continueThreadsAfterServerUpdate: testState.continueThreadsAfterServerUpdate }),21 ) => selector({ continueThreadsAfterServerUpdate: testState.continueThreadsAfterServerUpdate }),
21}));22}));
apps/web/src/components/ServerUpdateAction.tsx+3-2
...@@ -8,7 +8,7 @@ import type { ComponentProps } from "react";...@@ -8,7 +8,7 @@ import type { ComponentProps } from "react";
88
9import { requestConfirmDialog } from "~/confirmDialog";9import { requestConfirmDialog } from "~/confirmDialog";
10import { useCopyToClipboard } from "~/hooks/useCopyToClipboard";10import { useCopyToClipboard } from "~/hooks/useCopyToClipboard";
11import { useClientSettings } from "~/hooks/useSettings";11import { useEnvironmentSettings } from "~/hooks/useSettings";
12import { serverEnvironment } from "~/state/server";12import { serverEnvironment } from "~/state/server";
13import { useAtomCommand } from "~/state/use-atom-command";13import { useAtomCommand } from "~/state/use-atom-command";
14import { manualServerUpdateCommand } from "~/versionSkew";14import { manualServerUpdateCommand } from "~/versionSkew";
...@@ -99,7 +99,8 @@ export function ServerUpdateAction({...@@ -99,7 +99,8 @@ export function ServerUpdateAction({
99 readonly size?: ComponentProps<typeof Button>["size"];99 readonly size?: ComponentProps<typeof Button>["size"];
100}) {100}) {
101 const isDesktopAppUpdate = selfUpdate === "desktop-managed";101 const isDesktopAppUpdate = selfUpdate === "desktop-managed";
102 const continueThreadsAfterServerUpdate = useClientSettings(102 const continueThreadsAfterServerUpdate = useEnvironmentSettings(
103 environmentId,
103 (settings) => settings.continueThreadsAfterServerUpdate,104 (settings) => settings.continueThreadsAfterServerUpdate,
104 );105 );
105 const updateServer = useAtomCommand(serverEnvironment.updateServer, {106 const updateServer = useAtomCommand(serverEnvironment.updateServer, {
apps/web/src/components/settings/SettingsPanels.tsx+5-4
...@@ -564,7 +564,7 @@ export function useSettingsRestore(onRestored?: () => void) {...@@ -564,7 +564,7 @@ export function useSettingsRestore(onRestored?: () => void) {
564 : []),564 : []),
565 ...(settings.continueThreadsAfterServerUpdate !==565 ...(settings.continueThreadsAfterServerUpdate !==
566 DEFAULT_UNIFIED_SETTINGS.continueThreadsAfterServerUpdate566 DEFAULT_UNIFIED_SETTINGS.continueThreadsAfterServerUpdate
567 ? ["Continue threads after server updates"]567 ? ["Continue threads after restarts"]
568 : []),568 : []),
569 ...(isBackgroundActivityDirty ? ["Background activity"] : []),569 ...(isBackgroundActivityDirty ? ["Background activity"] : []),
570 ...(settings.defaultThreadEnvMode !== DEFAULT_UNIFIED_SETTINGS.defaultThreadEnvMode570 ...(settings.defaultThreadEnvMode !== DEFAULT_UNIFIED_SETTINGS.defaultThreadEnvMode
...@@ -2438,12 +2438,13 @@ export function GeneralSettingsPanel() {...@@ -2438,12 +2438,13 @@ export function GeneralSettingsPanel() {
24382438
2439 <SettingsRow2439 <SettingsRow
2440 {...searchableSetting("continue-threads-after-server-update")}2440 {...searchableSetting("continue-threads-after-server-update")}
2441 description="Automatically resume active threads after a server update restarts the environment."2441 serverScoped
2442 description="Automatically resume interrupted threads when this environment starts again after an update, crash, or machine restart."
2442 resetAction={2443 resetAction={
2443 settings.continueThreadsAfterServerUpdate !==2444 settings.continueThreadsAfterServerUpdate !==
2444 DEFAULT_UNIFIED_SETTINGS.continueThreadsAfterServerUpdate ? (2445 DEFAULT_UNIFIED_SETTINGS.continueThreadsAfterServerUpdate ? (
2445 <SettingResetButton2446 <SettingResetButton
2446 label="continue threads after server updates"2447 label="continue threads after restarts"
2447 onClick={() =>2448 onClick={() =>
2448 updateSettings({2449 updateSettings({
2449 continueThreadsAfterServerUpdate:2450 continueThreadsAfterServerUpdate:
...@@ -2459,7 +2460,7 @@ export function GeneralSettingsPanel() {...@@ -2459,7 +2460,7 @@ export function GeneralSettingsPanel() {
2459 onCheckedChange={(checked) =>2460 onCheckedChange={(checked) =>
2460 updateSettings({ continueThreadsAfterServerUpdate: Boolean(checked) })2461 updateSettings({ continueThreadsAfterServerUpdate: Boolean(checked) })
2461 }2462 }
2462 aria-label="Continue threads after server updates"2463 aria-label="Continue threads after restarts"
2463 />2464 />
2464 }2465 }
2465 />2466 />
apps/web/src/components/settings/settingsSearch.ts+4-2
...@@ -218,9 +218,11 @@ export const SETTINGS_SEARCH_ITEMS = [...@@ -218,9 +218,11 @@ export const SETTINGS_SEARCH_ITEMS = [
218 },218 },
219 {219 {
220 id: "continue-threads-after-server-update",220 id: "continue-threads-after-server-update",
221 title: "Continue threads after server updates",221 title: "Continue threads after restarts",
222 to: "/settings/general",222 to: "/settings/general",
223 searchTerms: ["resume running active work restart desktop update automatically"],223 searchTerms: [
224 "resume running active interrupted work restart reboot machine crash desktop update automatically",
225 ],
224 },226 },
225 {227 {
226 id: "background-activity",228 id: "background-activity",
docs/user/updating.md+7-3
...@@ -10,9 +10,13 @@ notice....@@ -10,9 +10,13 @@ notice.
10Server updates restart the connection and can interrupt active agents and10Server updates restart the connection and can interrupt active agents and
11terminal commands. Saved threads, settings, and project files remain.11terminal commands. Saved threads, settings, and project files remain.
1212
13**Settings → General → Continue threads after server updates** is off by default.13**Settings → General → Continue threads after restarts** is off by default.
14Enable it to resume supported active threads once the replacement server is14Enable it for each environment to resume supported active threads after an
15ready. Terminal commands may still be interrupted.15update, crash, or machine restart. T3 Code must start again on that machine;
16the setting does not enable automatic startup. Terminal commands may still be
17interrupted, and threads without saved provider resume state need a new message.
18If you previously enabled continuation for updates, enable this environment
19setting once to allow recovery without a connected client.
1620
17## Update a connected server21## Update a connected server
1822
packages/contracts/src/settings.ts+5-4
...@@ -255,9 +255,6 @@ export const ClientSettingsSchema = Schema.Struct({...@@ -255,9 +255,6 @@ export const ClientSettingsSchema = Schema.Struct({
255 confirmThreadArchive: Schema.Boolean.pipe(Schema.withDecodingDefault(Effect.succeed(false))),255 confirmThreadArchive: Schema.Boolean.pipe(Schema.withDecodingDefault(Effect.succeed(false))),
256 confirmThreadDelete: Schema.Boolean.pipe(Schema.withDecodingDefault(Effect.succeed(true))),256 confirmThreadDelete: Schema.Boolean.pipe(Schema.withDecodingDefault(Effect.succeed(true))),
257 confirmThreadUnpin: Schema.Boolean.pipe(Schema.withDecodingDefault(Effect.succeed(false))),257 confirmThreadUnpin: Schema.Boolean.pipe(Schema.withDecodingDefault(Effect.succeed(false))),
258 continueThreadsAfterServerUpdate: Schema.Boolean.pipe(
259 Schema.withDecodingDefault(Effect.succeed(false)),
260 ),
261 dismissedProviderUpdateNotificationKeys: Schema.Array(TrimmedNonEmptyString).pipe(258 dismissedProviderUpdateNotificationKeys: Schema.Array(TrimmedNonEmptyString).pipe(
262 Schema.withDecodingDefault(Effect.succeed([])),259 Schema.withDecodingDefault(Effect.succeed([])),
263 ),260 ),
...@@ -835,6 +832,10 @@ export const ServerSettings = Schema.Struct({...@@ -835,6 +832,10 @@ export const ServerSettings = Schema.Struct({
835 Schema.withDecodingDefault(Effect.succeed(false)),832 Schema.withDecodingDefault(Effect.succeed(false)),
836 ),833 ),
837 enableProviderUpdateChecks: Schema.Boolean.pipe(Schema.withDecodingDefault(Effect.succeed(true))),834 enableProviderUpdateChecks: Schema.Boolean.pipe(Schema.withDecodingDefault(Effect.succeed(true))),
835 // Retain the update-era key; recovery now needs an environment-owned opt-in.
836 continueThreadsAfterServerUpdate: Schema.Boolean.pipe(
837 Schema.withDecodingDefault(Effect.succeed(false)),
838 ),
838 /**839 /**
839 * Whether agents may drive the in-app preview browser. Turning this off840 * Whether agents may drive the in-app preview browser. Turning this off
840 * withholds the MCP credential, so the `t3-code` server (and with it every841 * withholds the MCP credential, so the `t3-code` server (and with it every
...@@ -1105,6 +1106,7 @@ export const ServerSettingsPatch = Schema.Struct({...@@ -1105,6 +1106,7 @@ export const ServerSettingsPatch = Schema.Struct({
1105 // Server settings1106 // Server settings
1106 enableLegacyTokenStreaming: Schema.optionalKey(Schema.Boolean),1107 enableLegacyTokenStreaming: Schema.optionalKey(Schema.Boolean),
1107 enableProviderUpdateChecks: Schema.optionalKey(Schema.Boolean),1108 enableProviderUpdateChecks: Schema.optionalKey(Schema.Boolean),
1109 continueThreadsAfterServerUpdate: Schema.optionalKey(Schema.Boolean),
1108 enableAgentBrowserAccess: Schema.optionalKey(Schema.Boolean),1110 enableAgentBrowserAccess: Schema.optionalKey(Schema.Boolean),
1109 sidebarAutoSettleAfterDays: Schema.optionalKey(Schema.NullOr(SidebarAutoSettleAfterDays)),1111 sidebarAutoSettleAfterDays: Schema.optionalKey(Schema.NullOr(SidebarAutoSettleAfterDays)),
1110 sidebarAutoSettleOnMerge: Schema.optionalKey(Schema.Boolean),1112 sidebarAutoSettleOnMerge: Schema.optionalKey(Schema.Boolean),
...@@ -1181,7 +1183,6 @@ export const ClientSettingsPatch = Schema.Struct({...@@ -1181,7 +1183,6 @@ export const ClientSettingsPatch = Schema.Struct({
1181 confirmThreadArchive: Schema.optionalKey(Schema.Boolean),1183 confirmThreadArchive: Schema.optionalKey(Schema.Boolean),
1182 confirmThreadDelete: Schema.optionalKey(Schema.Boolean),1184 confirmThreadDelete: Schema.optionalKey(Schema.Boolean),
1183 confirmThreadUnpin: Schema.optionalKey(Schema.Boolean),1185 confirmThreadUnpin: Schema.optionalKey(Schema.Boolean),
1184 continueThreadsAfterServerUpdate: Schema.optionalKey(Schema.Boolean),
1185 diffIgnoreWhitespace: Schema.optionalKey(Schema.Boolean),1186 diffIgnoreWhitespace: Schema.optionalKey(Schema.Boolean),
1186 diffLayout: Schema.optionalKey(DiffLayout),1187 diffLayout: Schema.optionalKey(DiffLayout),
1187 environmentIdentificationMode: Schema.optionalKey(EnvironmentIdentificationMode),1188 environmentIdentificationMode: Schema.optionalKey(EnvironmentIdentificationMode),