authorgravatar for maria@kuuro.netmaria <maria@kuuro.net> 2026-09-02 04:28:16-04:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-02 04:28:16-04:00
log5b7d72aad14ed37e8e5e4c02a6d49814bfe528ac
tree9b11a577e538ebb99793512667fd67e208b1363a
parent14f15cfed48bc7519e8e5eeac03f483f4b11ef3a
signature Signed by PGP key B5690EEEBB952194

feat(updates): continue active threads across server restarts (#9167)

Co-authored-by: maria-rcks <254055478+maria-rcks@users.noreply.github.com>

26 files changed, 1292 insertions(+), 71 deletions(-)

apps/desktop/src/settings/DesktopClientSettings.test.ts+1
...@@ -23,6 +23,7 @@ const clientSettings: ClientSettings = {...@@ -23,6 +23,7 @@ const clientSettings: ClientSettings = {
23 confirmThreadArchive: true,23 confirmThreadArchive: true,
24 confirmThreadDelete: false,24 confirmThreadDelete: false,
25 confirmThreadUnpin: false,25 confirmThreadUnpin: false,
26 continueThreadsAfterServerUpdate: true,
26 contextWindowMeterEnabled: false,27 contextWindowMeterEnabled: false,
27 dismissedProviderUpdateNotificationKeys: [],28 dismissedProviderUpdateNotificationKeys: [],
28 diffIgnoreWhitespace: true,29 diffIgnoreWhitespace: true,
apps/server/src/cloud/selfUpdate.test.ts+217
...@@ -1,6 +1,8 @@...@@ -1,6 +1,8 @@
1import * as NodeServices from "@effect/platform-node/NodeServices";1import * as NodeServices from "@effect/platform-node/NodeServices";
2import { expect, it } from "@effect/vitest";2import { expect, it } from "@effect/vitest";
3import { ServerSelfUpdateError, ThreadId } from "@t3tools/contracts";
3import { HostProcessExecutablePath } from "@t3tools/shared/hostProcess";4import { HostProcessExecutablePath } from "@t3tools/shared/hostProcess";
5import * as Cause from "effect/Cause";
4import * as Deferred from "effect/Deferred";6import * as Deferred from "effect/Deferred";
5import * as Effect from "effect/Effect";7import * as Effect from "effect/Effect";
6import * as FileSystem from "effect/FileSystem";8import * as FileSystem from "effect/FileSystem";
...@@ -104,6 +106,221 @@ const makeHarness = Effect.fn("test.make_self_update_harness")(function* (...@@ -104,6 +106,221 @@ const makeHarness = Effect.fn("test.make_self_update_harness")(function* (
104});106});
105107
106it.layer(NodeServices.layer)("server self update", (it) => {108it.layer(NodeServices.layer)("server self update", (it) => {
109 it.effect("marks running threads at the boot-service handoff", () =>
110 Effect.gen(function* () {
111 const events: string[] = [];
112 const selfUpdate = yield* ServerSelfUpdate.withRunningThreadContinuation({
113 mode: "web",
114 selfUpdate: {
115 update: (_input, reportProgress = () => Effect.void) =>
116 reportProgress("downloading").pipe(
117 Effect.andThen(reportProgress("installing")),
118 Effect.as({
119 targetVersion: "1.1.0",
120 method: "boot-service" as const,
121 updateId: "update-id",
122 }),
123 ),
124 commitDesktopUpdate: () => Effect.never,
125 },
126 prepare: Effect.sync(() => {
127 events.push("prepare");
128 return [ThreadId.make("thread-running")];
129 }),
130 clear: () => Effect.sync(() => void events.push("clear")),
131 });
132
133 yield* selfUpdate.update({ targetVersion: "1.1.0", continueRunningThreads: true }, (stage) =>
134 Effect.sync(() => void events.push(stage)),
135 );
136
137 expect(events).toEqual(["downloading", "prepare", "installing"]);
138 }),
139 );
140
141 it.effect("marks desktop threads only when the prepared update commits", () =>
142 Effect.gen(function* () {
143 const threadId = ThreadId.make("thread-running-desktop");
144 const events: string[] = [];
145 const commitError = new ServerSelfUpdateError({ reason: "install failed" });
146 const selfUpdate = yield* ServerSelfUpdate.withRunningThreadContinuation({
147 mode: "desktop",
148 selfUpdate: {
149 update: (_input, reportProgress = () => Effect.void) =>
150 reportProgress("installing").pipe(
151 Effect.as({
152 targetVersion: "1.2.0",
153 method: "desktop-app" as const,
154 desktopUpdateToken: "desktop-token",
155 }),
156 ),
157 commitDesktopUpdate: () =>
158 Effect.sync(() => events.push("commit")).pipe(Effect.andThen(Effect.fail(commitError))),
159 },
160 prepare: Effect.sync(() => {
161 events.push("prepare");
162 return [threadId];
163 }),
164 clear: (threadIds) => Effect.sync(() => void events.push(`clear:${threadIds.join(",")}`)),
165 });
166
167 yield* selfUpdate.update({ targetVersion: "1.2.0", continueRunningThreads: true }, (stage) =>
168 Effect.sync(() => void events.push(stage)),
169 );
170 expect(events).toEqual(["installing"]);
171 expect(yield* selfUpdate.commitDesktopUpdate("desktop-token").pipe(Effect.flip)).toBe(
172 commitError,
173 );
174 expect(events).toEqual(["installing", "prepare", "commit", `clear:${threadId}`]);
175 expect(yield* selfUpdate.commitDesktopUpdate("desktop-token").pipe(Effect.flip)).toBe(
176 commitError,
177 );
178 expect(events).toEqual([
179 "installing",
180 "prepare",
181 "commit",
182 `clear:${threadId}`,
183 "prepare",
184 "commit",
185 `clear:${threadId}`,
186 ]);
187 }),
188 );
189
190 it.effect("reports a failed continuation-marker cleanup", () =>
191 Effect.gen(function* () {
192 const updateError = new ServerSelfUpdateError({ reason: "update failed" });
193 const clearError = new ServerSelfUpdateError({ reason: "marker cleanup failed" });
194 const selfUpdate = yield* ServerSelfUpdate.withRunningThreadContinuation({
195 mode: "web",
196 selfUpdate: {
197 update: (_input, reportProgress = () => Effect.void) =>
198 reportProgress("installing").pipe(Effect.andThen(Effect.fail(updateError))),
199 commitDesktopUpdate: () => Effect.never,
200 },
201 prepare: Effect.succeed([ThreadId.make("thread-cleanup-failure")]),
202 clear: () => Effect.fail(clearError),
203 });
204
205 expect(
206 yield* selfUpdate
207 .update({ targetVersion: "1.1.0", continueRunningThreads: true })
208 .pipe(Effect.flip),
209 ).toBe(clearError);
210 }),
211 );
212
213 it.effect("keeps continuation markers after the boot-service handoff is accepted", () =>
214 Effect.gen(function* () {
215 const events: string[] = [];
216 const selfUpdate = yield* ServerSelfUpdate.withRunningThreadContinuation({
217 mode: "web",
218 selfUpdate: {
219 update: (
220 _input,
221 reportProgress = () => Effect.void,
222 onHandoffAccepted = () => Effect.void,
223 ) =>
224 reportProgress("installing").pipe(
225 Effect.andThen(onHandoffAccepted()),
226 Effect.andThen(Effect.interrupt),
227 ),
228 commitDesktopUpdate: () => Effect.never,
229 },
230 prepare: Effect.sync(() => {
231 events.push("prepare");
232 return [ThreadId.make("thread-accepted-boot-handoff")];
233 }),
234 clear: () => Effect.sync(() => void events.push("clear")),
235 });
236
237 const exit = yield* selfUpdate
238 .update({ targetVersion: "1.1.0", continueRunningThreads: true })
239 .pipe(Effect.exit);
240
241 expect(exit._tag).toBe("Failure");
242 expect(events).toEqual(["prepare"]);
243 }),
244 );
245
246 it.effect("keeps continuation markers after the desktop handoff is accepted", () =>
247 Effect.gen(function* () {
248 const events: string[] = [];
249 const selfUpdate = yield* ServerSelfUpdate.withRunningThreadContinuation({
250 mode: "desktop",
251 selfUpdate: {
252 update: () =>
253 Effect.succeed({
254 targetVersion: "1.2.0",
255 method: "desktop-app" as const,
256 desktopUpdateToken: "accepted-desktop-token",
257 }),
258 commitDesktopUpdate: (_requestId, onHandoffAccepted = () => Effect.void) =>
259 onHandoffAccepted().pipe(Effect.andThen(Effect.interrupt)),
260 },
261 prepare: Effect.sync(() => {
262 events.push("prepare");
263 return [ThreadId.make("thread-accepted-desktop-handoff")];
264 }),
265 clear: () => Effect.sync(() => void events.push("clear")),
266 });
267
268 yield* selfUpdate.update({
269 targetVersion: "1.2.0",
270 continueRunningThreads: true,
271 });
272 const exit = yield* selfUpdate
273 .commitDesktopUpdate("accepted-desktop-token")
274 .pipe(Effect.exit);
275
276 expect(exit._tag).toBe("Failure");
277 expect(events).toEqual(["prepare"]);
278 }),
279 );
280
281 it.effect("clears continuation markers for mixed failure and interrupt causes", () =>
282 Effect.gen(function* () {
283 const events: string[] = [];
284 const commitError = new ServerSelfUpdateError({ reason: "install failed" });
285 const selfUpdate = yield* ServerSelfUpdate.withRunningThreadContinuation({
286 mode: "desktop",
287 selfUpdate: {
288 update: () =>
289 Effect.succeed({
290 targetVersion: "1.2.0",
291 method: "desktop-app" as const,
292 desktopUpdateToken: "failed-desktop-token",
293 }),
294 commitDesktopUpdate: (_requestId, onHandoffAccepted = () => Effect.void) =>
295 onHandoffAccepted().pipe(
296 Effect.andThen(
297 Effect.failCause(
298 Cause.fromReasons([
299 Cause.makeFailReason(commitError),
300 Cause.makeInterruptReason(),
301 ]),
302 ),
303 ),
304 ),
305 },
306 prepare: Effect.sync(() => [ThreadId.make("thread-failed-desktop-install")]),
307 clear: () => Effect.sync(() => void events.push("clear")),
308 });
309
310 yield* selfUpdate.update({
311 targetVersion: "1.2.0",
312 continueRunningThreads: true,
313 });
314 const exit = yield* selfUpdate.commitDesktopUpdate("failed-desktop-token").pipe(Effect.exit);
315 expect(exit._tag).toBe("Failure");
316 if (exit._tag === "Failure") {
317 expect(Cause.hasInterrupts(exit.cause)).toBe(true);
318 expect(Cause.hasInterruptsOnly(exit.cause)).toBe(false);
319 }
320 expect(events).toEqual(["clear"]);
321 }),
322 );
323
107 it.effect("stages and preflights before asking the launcher for an update ID", () =>324 it.effect("stages and preflights before asking the launcher for an update ID", () =>
108 Effect.gen(function* () {325 Effect.gen(function* () {
109 const { selfUpdate, order } = yield* makeHarness();326 const { selfUpdate, order } = yield* makeHarness();
apps/server/src/cloud/selfUpdate.ts+124-8
...@@ -4,15 +4,18 @@ import {...@@ -4,15 +4,18 @@ import {
4 type ServerSelfUpdateInput,4 type ServerSelfUpdateInput,
5 type ServerSelfUpdateProgressStage,5 type ServerSelfUpdateProgressStage,
6 type ServerSelfUpdateResult,6 type ServerSelfUpdateResult,
7 type ThreadId,
7} from "@t3tools/contracts";8} from "@t3tools/contracts";
8import { HostProcessExecutablePath } from "@t3tools/shared/hostProcess";9import { HostProcessExecutablePath } from "@t3tools/shared/hostProcess";
10import * as Cause from "effect/Cause";
9import * as Context from "effect/Context";11import * as Context from "effect/Context";
10import * as Duration from "effect/Duration";12import * as Duration from "effect/Duration";
11import * as Effect from "effect/Effect";13import * as Effect from "effect/Effect";
14import * as HashSet from "effect/HashSet";
15import * as Ref from "effect/Ref";
12import * as FileSystem from "effect/FileSystem";16import * as FileSystem from "effect/FileSystem";
13import * as Layer from "effect/Layer";17import * as Layer from "effect/Layer";
14import * as Path from "effect/Path";18import * as Path from "effect/Path";
15import * as Ref from "effect/Ref";
1619
17import * as ServerConfig from "../config.ts";20import * as ServerConfig from "../config.ts";
18import * as DesktopAppUpdate from "../desktopUpdate/DesktopAppUpdate.ts";21import * as DesktopAppUpdate from "../desktopUpdate/DesktopAppUpdate.ts";
...@@ -41,14 +44,125 @@ export class ServerSelfUpdate extends Context.Service<...@@ -41,14 +44,125 @@ export class ServerSelfUpdate extends Context.Service<
41 {44 {
42 readonly update: (45 readonly update: (
43 input: ServerSelfUpdateInput,46 input: ServerSelfUpdateInput,
44 reportProgress?: (stage: ServerSelfUpdateProgressStage) => Effect.Effect<void>,47 reportProgress?: (
48 stage: ServerSelfUpdateProgressStage,
49 ) => Effect.Effect<void, ServerSelfUpdateError>,
50 onHandoffAccepted?: () => Effect.Effect<void>,
45 ) => Effect.Effect<ServerSelfUpdateResult, ServerSelfUpdateError>;51 ) => Effect.Effect<ServerSelfUpdateResult, ServerSelfUpdateError>;
46 readonly commitDesktopUpdate: (52 readonly commitDesktopUpdate: (
47 requestId: string,53 requestId: string,
54 onHandoffAccepted?: () => Effect.Effect<void>,
48 ) => Effect.Effect<never, ServerSelfUpdateError>;55 ) => Effect.Effect<never, ServerSelfUpdateError>;
49 }56 }
50>()("t3/cloud/selfUpdate/ServerSelfUpdate") {}57>()("t3/cloud/selfUpdate/ServerSelfUpdate") {}
5158
59export const withRunningThreadContinuation = Effect.fn(
60 "cloud.server_self_update.withRunningThreadContinuation",
61)(function* (input: {
62 readonly mode: ServerConfig.RuntimeMode;
63 readonly selfUpdate: ServerSelfUpdate["Service"];
64 readonly prepare: Effect.Effect<ReadonlyArray<ThreadId>, ServerSelfUpdateError>;
65 readonly clear: (
66 threadIds: ReadonlyArray<ThreadId>,
67 ) => Effect.Effect<void, ServerSelfUpdateError>;
68}) {
69 const desktopContinuationTokens = yield* Ref.make(HashSet.empty<string>());
70 const clearOnError = <A>(
71 effect: Effect.Effect<A, ServerSelfUpdateError>,
72 threadIds: () => ReadonlyArray<ThreadId>,
73 handoffAccepted: () => boolean,
74 ): Effect.Effect<A, ServerSelfUpdateError> =>
75 effect.pipe(
76 Effect.catchCause((cause) =>
77 (handoffAccepted() && Cause.hasInterruptsOnly(cause)
78 ? Effect.void
79 : input.clear(threadIds())
80 ).pipe(Effect.andThen(Effect.failCause(cause))),
81 ),
82 );
83
84 const update: ServerSelfUpdate["Service"]["update"] = (
85 request,
86 reportProgress = () => Effect.void,
87 ) => {
88 let prepared = false;
89 let handoffAccepted = false;
90 let continuationThreadIds: ReadonlyArray<ThreadId> = [];
91 return clearOnError(
92 input.selfUpdate
93 .update(
94 request,
95 (stage) =>
96 (request.continueRunningThreads === true &&
97 input.mode !== "desktop" &&
98 stage === "installing" &&
99 !prepared
100 ? input.prepare.pipe(
101 Effect.tap((threadIds) =>
102 Effect.sync(() => {
103 prepared = true;
104 continuationThreadIds = threadIds;
105 }),
106 ),
107 Effect.asVoid,
108 )
109 : Effect.void
110 ).pipe(Effect.andThen(reportProgress(stage))),
111 () =>
112 Effect.sync(() => {
113 handoffAccepted = true;
114 }),
115 )
116 .pipe(
117 Effect.tap((result) => {
118 if (
119 result.method === "desktop-app" &&
120 result.desktopUpdateToken !== undefined &&
121 request.continueRunningThreads === true
122 ) {
123 return Ref.update(desktopContinuationTokens, HashSet.add(result.desktopUpdateToken));
124 }
125 return Effect.void;
126 }),
127 ),
128 () => continuationThreadIds,
129 () => handoffAccepted,
130 );
131 };
132
133 return ServerSelfUpdate.of({
134 update,
135 commitDesktopUpdate: (requestId) =>
136 Effect.gen(function* () {
137 const shouldContinue = yield* Ref.modify(desktopContinuationTokens, (tokens) => [
138 HashSet.has(tokens, requestId),
139 HashSet.remove(tokens, requestId),
140 ]);
141 let handoffAccepted = false;
142 let continuationThreadIds: ReadonlyArray<ThreadId> = [];
143 return yield* clearOnError(
144 Effect.gen(function* () {
145 continuationThreadIds = shouldContinue ? yield* input.prepare : [];
146 return yield* input.selfUpdate.commitDesktopUpdate(requestId, () =>
147 Effect.sync(() => {
148 handoffAccepted = true;
149 }),
150 );
151 }),
152 () => continuationThreadIds,
153 () => handoffAccepted,
154 ).pipe(
155 Effect.catchCause((cause) =>
156 (shouldContinue && !handoffAccepted
157 ? Ref.update(desktopContinuationTokens, HashSet.add(requestId))
158 : Effect.void
159 ).pipe(Effect.andThen(Effect.failCause(cause))),
160 ),
161 );
162 }),
163 });
164});
165
52export const make = Effect.fn("cloud.server_self_update.make")(function* () {166export const make = Effect.fn("cloud.server_self_update.make")(function* () {
53 const serverConfig = yield* ServerConfig.ServerConfig;167 const serverConfig = yield* ServerConfig.ServerConfig;
54 const desktopAppUpdate = yield* DesktopAppUpdate.DesktopAppUpdate;168 const desktopAppUpdate = yield* DesktopAppUpdate.DesktopAppUpdate;
...@@ -68,7 +182,7 @@ export const make = Effect.fn("cloud.server_self_update.make")(function* () {...@@ -68,7 +182,7 @@ export const make = Effect.fn("cloud.server_self_update.make")(function* () {
68182
69 const update: ServerSelfUpdate["Service"]["update"] = Effect.fn(183 const update: ServerSelfUpdate["Service"]["update"] = Effect.fn(
70 "cloud.server_self_update.update",184 "cloud.server_self_update.update",
71 )(function* (input, reportProgress = () => Effect.void) {185 )(function* (input, reportProgress = () => Effect.void, onHandoffAccepted = () => Effect.void) {
72 if (capability === "desktop-managed") {186 if (capability === "desktop-managed") {
73 // input.targetVersion is meaningless here: the desktop app's own187 // input.targetVersion is meaningless here: the desktop app's own
74 // update feed decides what it downloads, and the result carries what188 // update feed decides what it downloads, and the result carries what
...@@ -180,9 +294,8 @@ export const make = Effect.fn("cloud.server_self_update.make")(function* () {...@@ -180,9 +294,8 @@ export const make = Effect.fn("cloud.server_self_update.make")(function* () {
180 );294 );
181295
182 yield* reportProgress("installing");296 yield* reportProgress("installing");
183 const updateId = yield* launcher297 const updateId = yield* Effect.uninterruptible(
184 .requestUpdate({ targetVersion, dbPath: serverConfig.dbPath })298 launcher.requestUpdate({ targetVersion, dbPath: serverConfig.dbPath }).pipe(
185 .pipe(
186 Effect.mapError((error) =>299 Effect.mapError((error) =>
187 failWith(300 failWith(
188 error._tag === "ServiceLauncherRejectedError"301 error._tag === "ServiceLauncherRejectedError"
...@@ -191,7 +304,9 @@ export const make = Effect.fn("cloud.server_self_update.make")(function* () {...@@ -191,7 +304,9 @@ export const make = Effect.fn("cloud.server_self_update.make")(function* () {
191 error,304 error,
192 ),305 ),
193 ),306 ),
194 );307 Effect.tap(() => onHandoffAccepted()),
308 ),
309 );
195310
196 yield* Effect.logInfo("Server update prepared; handing off to the service launcher.", {311 yield* Effect.logInfo("Server update prepared; handing off to the service launcher.", {
197 updateId,312 updateId,
...@@ -204,7 +319,8 @@ export const make = Effect.fn("cloud.server_self_update.make")(function* () {...@@ -204,7 +319,8 @@ export const make = Effect.fn("cloud.server_self_update.make")(function* () {
204319
205 return ServerSelfUpdate.of({320 return ServerSelfUpdate.of({
206 update,321 update,
207 commitDesktopUpdate: (requestId) => desktopAppUpdate.commit(requestId),322 commitDesktopUpdate: (requestId, onHandoffAccepted) =>
323 desktopAppUpdate.commit(requestId, onHandoffAccepted),
208 });324 });
209});325});
210326
apps/server/src/desktopUpdate/DesktopAppUpdate.ts+22-10
...@@ -49,11 +49,16 @@ export class DesktopAppUpdate extends Context.Service<...@@ -49,11 +49,16 @@ export class DesktopAppUpdate extends Context.Service<
49 /** Checks and downloads through the desktop app, then returns a token49 /** Checks and downloads through the desktop app, then returns a token
50 while this server is still connected. `commit` starts installation. */50 while this server is still connected. `commit` starts installation. */
51 readonly run: (51 readonly run: (
52 reportProgress: (stage: ServerSelfUpdateProgressStage) => Effect.Effect<void>,52 reportProgress: (
53 stage: ServerSelfUpdateProgressStage,
54 ) => Effect.Effect<void, ServerSelfUpdateError>,
53 ) => Effect.Effect<ServerSelfUpdateResult, ServerSelfUpdateError>;55 ) => Effect.Effect<ServerSelfUpdateResult, ServerSelfUpdateError>;
54 /** Starts the prepared install. Success stops this server, so this effect56 /** Starts the prepared install. Success stops this server, so this effect
55 returns only when installation fails or times out. */57 returns only when installation fails or times out. */
56 readonly commit: (requestId: string) => Effect.Effect<never, ServerSelfUpdateError>;58 readonly commit: (
59 requestId: string,
60 onHandoffAccepted?: () => Effect.Effect<void>,
61 ) => Effect.Effect<never, ServerSelfUpdateError>;
57 }62 }
58>()("t3/desktopUpdate/DesktopAppUpdate") {}63>()("t3/desktopUpdate/DesktopAppUpdate") {}
5964
...@@ -72,11 +77,15 @@ export const make = Effect.fn("desktopUpdate.desktopAppUpdate.make")(function* (...@@ -72,11 +77,15 @@ export const make = Effect.fn("desktopUpdate.desktopAppUpdate.make")(function* (
72 const consumeReports = (77 const consumeReports = (
73 requestId: string,78 requestId: string,
74 changes: Stream.Stream<DesktopUpdateStatusReport>,79 changes: Stream.Stream<DesktopUpdateStatusReport>,
75 reportProgress: (stage: ServerSelfUpdateProgressStage) => Effect.Effect<void>,80 reportProgress: (
81 stage: ServerSelfUpdateProgressStage,
82 ) => Effect.Effect<void, ServerSelfUpdateError>,
76 ) =>83 ) =>
77 Effect.gen(function* () {84 Effect.gen(function* () {
78 const lastStage = yield* Ref.make<ServerSelfUpdateProgressStage | null>(null);85 const lastStage = yield* Ref.make<ServerSelfUpdateProgressStage | null>(null);
79 const emitStage = (stage: ServerSelfUpdateProgressStage | null): Effect.Effect<void> =>86 const emitStage = (
87 stage: ServerSelfUpdateProgressStage | null,
88 ): Effect.Effect<void, ServerSelfUpdateError> =>
80 stage === null89 stage === null
81 ? Effect.void90 ? Effect.void
82 : Ref.get(lastStage).pipe(91 : Ref.get(lastStage).pipe(
...@@ -90,7 +99,9 @@ export const make = Effect.fn("desktopUpdate.desktopAppUpdate.make")(function* (...@@ -90,7 +99,9 @@ export const make = Effect.fn("desktopUpdate.desktopAppUpdate.make")(function* (
90 const terminal = yield* changes.pipe(99 const terminal = yield* changes.pipe(
91 Stream.filter((report) => report.requestId === requestId),100 Stream.filter((report) => report.requestId === requestId),
92 Stream.mapEffect(101 Stream.mapEffect(
93 (report): Effect.Effect<Option.Option<DesktopUpdateStatusReport>> =>102 (
103 report,
104 ): Effect.Effect<Option.Option<DesktopUpdateStatusReport>, ServerSelfUpdateError> =>
94 report.outcome === undefined105 report.outcome === undefined
95 ? emitStage(desktopUpdateProgressStage(report.state)).pipe(106 ? emitStage(desktopUpdateProgressStage(report.state)).pipe(
96 Effect.as(Option.none<DesktopUpdateStatusReport>()),107 Effect.as(Option.none<DesktopUpdateStatusReport>()),
...@@ -176,7 +187,7 @@ export const make = Effect.fn("desktopUpdate.desktopAppUpdate.make")(function* (...@@ -176,7 +187,7 @@ export const make = Effect.fn("desktopUpdate.desktopAppUpdate.make")(function* (
176187
177 const commit: DesktopAppUpdate["Service"]["commit"] = Effect.fn(188 const commit: DesktopAppUpdate["Service"]["commit"] = Effect.fn(
178 "desktopUpdate.desktopAppUpdate.commit",189 "desktopUpdate.desktopAppUpdate.commit",
179 )(function* (requestId) {190 )(function* (requestId, onHandoffAccepted = () => Effect.void) {
180 if (!available) {191 if (!available) {
181 return yield* failWith("This server cannot commit a desktop app update.");192 return yield* failWith("This server cannot commit a desktop app update.");
182 }193 }
...@@ -187,11 +198,12 @@ export const make = Effect.fn("desktopUpdate.desktopAppUpdate.make")(function* (...@@ -187,11 +198,12 @@ export const make = Effect.fn("desktopUpdate.desktopAppUpdate.make")(function* (
187 onNone: () => changes,198 onNone: () => changes,
188 onSome: (report) => Stream.concat(Stream.make(report), changes),199 onSome: (report) => Stream.concat(Stream.make(report), changes),
189 });200 });
190 yield* receiver201 yield* Effect.uninterruptible(
191 .commitDesktopUpdate(requestId)202 receiver.commitDesktopUpdate(requestId).pipe(
192 .pipe(
193 Effect.mapError((error) => failWith("Could not reach the T3 Code desktop app.", error)),203 Effect.mapError((error) => failWith("Could not reach the T3 Code desktop app.", error)),
194 );204 Effect.tap(() => onHandoffAccepted()),
205 ),
206 );
195 return yield* reports.pipe(207 return yield* reports.pipe(
196 Stream.filter((report) => report.requestId === requestId && report.outcome === "failed"),208 Stream.filter((report) => report.requestId === requestId && report.outcome === "failed"),
197 Stream.runHead,209 Stream.runHead,
apps/server/src/environment/ServerEnvironment.test.ts+2
...@@ -244,11 +244,13 @@ it.layer(NodeServices.layer)("ServerEnvironmentLive", (it) => {...@@ -244,11 +244,13 @@ it.layer(NodeServices.layer)("ServerEnvironmentLive", (it) => {
244 expect(withFd.capabilities.serverSelfUpdate).toBe("desktop-managed");244 expect(withFd.capabilities.serverSelfUpdate).toBe("desktop-managed");
245 expect(withFd.capabilities.desktopAppUpdate).toBe(true);245 expect(withFd.capabilities.desktopAppUpdate).toBe(true);
246 expect(withFd.capabilities.serverSelfUpdateProgress).toBe(true);246 expect(withFd.capabilities.serverSelfUpdateProgress).toBe(true);
247 expect(withFd.capabilities.serverUpdateThreadContinuation).toBe(true);
247248
248 const withoutFd = yield* describeWith({ mode: "desktop" });249 const withoutFd = yield* describeWith({ mode: "desktop" });
249 expect(withoutFd.capabilities.serverSelfUpdate).toBe("desktop-managed");250 expect(withoutFd.capabilities.serverSelfUpdate).toBe("desktop-managed");
250 expect(withoutFd.capabilities.desktopAppUpdate).toBeUndefined();251 expect(withoutFd.capabilities.desktopAppUpdate).toBeUndefined();
251 expect(withoutFd.capabilities.serverSelfUpdateProgress).toBeUndefined();252 expect(withoutFd.capabilities.serverSelfUpdateProgress).toBeUndefined();
253 expect(withoutFd.capabilities.serverUpdateThreadContinuation).toBeUndefined();
252254
253 const web = yield* describeWith({ mode: "web", desktopTelemetryControlFd: 5 });255 const web = yield* describeWith({ mode: "web", desktopTelemetryControlFd: 5 });
254 expect(web.capabilities.desktopAppUpdate).toBeUndefined();256 expect(web.capabilities.desktopAppUpdate).toBeUndefined();
apps/server/src/environment/ServerEnvironment.ts+4-1
...@@ -224,7 +224,10 @@ export const make = Effect.gen(function* () {...@@ -224,7 +224,10 @@ export const make = Effect.gen(function* () {
224 threadPullRequestLinking: true,224 threadPullRequestLinking: true,
225 ...(serverSelfUpdate === null ? {} : { serverSelfUpdate }),225 ...(serverSelfUpdate === null ? {} : { serverSelfUpdate }),
226 ...(serverSelfUpdate === "boot-service" || desktopAppUpdate226 ...(serverSelfUpdate === "boot-service" || desktopAppUpdate
227 ? { serverSelfUpdateProgress: true }227 ? {
228 serverSelfUpdateProgress: true,
229 serverUpdateThreadContinuation: true,
230 }
228 : {}),231 : {}),
229 ...(desktopAppUpdate ? { desktopAppUpdate: true } : {}),232 ...(desktopAppUpdate ? { desktopAppUpdate: true } : {}),
230 },233 },
apps/server/src/provider/Layers/CodexAdapter.ts+1
...@@ -2010,6 +2010,7 @@ export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* (...@@ -2010,6 +2010,7 @@ export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* (
2010 provider: PROVIDER,2010 provider: PROVIDER,
2011 capabilities: {2011 capabilities: {
2012 sessionModelSwitch: "in-session",2012 sessionModelSwitch: "in-session",
2013 promptlessTurnContinuation: true,
2013 },2014 },
2014 startSession,2015 startSession,
2015 sendTurn,2016 sendTurn,
apps/server/src/provider/Layers/ProviderService.test.ts+51
...@@ -246,6 +246,7 @@ function makeFakeCodexAdapter(provider: ProviderDriverKind = CODEX_DRIVER) {...@@ -246,6 +246,7 @@ function makeFakeCodexAdapter(provider: ProviderDriverKind = CODEX_DRIVER) {
246 provider,246 provider,
247 capabilities: {247 capabilities: {
248 sessionModelSwitch: "in-session",248 sessionModelSwitch: "in-session",
249 ...(provider === CODEX_DRIVER ? { promptlessTurnContinuation: true } : {}),
249 },250 },
250 startSession,251 startSession,
251 sendTurn,252 sendTurn,
...@@ -960,6 +961,56 @@ it.effect(...@@ -960,6 +961,56 @@ it.effect(
960);961);
961962
962routing.layer("ProviderServiceLive routing", (it) => {963routing.layer("ProviderServiceLive routing", (it) => {
964 it.effect("allows promptless continuation only for capable providers", () =>
965 Effect.gen(function* () {
966 const provider = yield* ProviderService.ProviderService;
967 const codexThreadId = asThreadId("thread-promptless-continuation");
968 yield* provider.startSession(codexThreadId, {
969 provider: CODEX_DRIVER,
970 providerInstanceId: codexInstanceId,
971 threadId: codexThreadId,
972 runtimeMode: "full-access",
973 });
974
975 yield* provider.sendTurn({ threadId: codexThreadId, continuation: true });
976 assert.deepEqual(routing.codex.sendTurn.mock.calls.at(-1)?.[0], {
977 threadId: codexThreadId,
978 continuation: true,
979 });
980
981 const claudeThreadId = asThreadId("thread-promptless-continuation-unsupported");
982 yield* provider.startSession(claudeThreadId, {
983 provider: CLAUDE_AGENT_DRIVER,
984 providerInstanceId: claudeAgentInstanceId,
985 threadId: claudeThreadId,
986 runtimeMode: "full-access",
987 });
988 const failure = yield* Effect.flip(
989 provider.sendTurn({ threadId: claudeThreadId, continuation: true }),
990 );
991 assert.instanceOf(failure, ProviderValidationError);
992 assert.include(failure.issue, "requires an explicit continuation prompt");
993 assert.equal(routing.claude.sendTurn.mock.calls.length, 0);
994
995 yield* provider.stopSession({ threadId: claudeThreadId });
996 routing.claude.startSession.mockClear();
997 const stoppedFailure = yield* Effect.flip(
998 provider.sendTurn({ threadId: claudeThreadId, continuation: true }),
999 );
1000 assert.instanceOf(stoppedFailure, ProviderValidationError);
1001 assert.include(stoppedFailure.issue, "requires an explicit continuation prompt");
1002 assert.equal(routing.claude.startSession.mock.calls.length, 0);
1003
1004 yield* provider.stopSession({ threadId: codexThreadId });
1005 routing.codex.startSession.mockClear();
1006 routing.codex.sendTurn.mockClear();
1007 routing.codex.stopSession.mockClear();
1008 routing.claude.startSession.mockClear();
1009 routing.claude.sendTurn.mockClear();
1010 routing.claude.stopSession.mockClear();
1011 }),
1012 );
1013
963 it.effect("routes provider operations and rollback conversation", () =>1014 it.effect("routes provider operations and rollback conversation", () =>
964 Effect.gen(function* () {1015 Effect.gen(function* () {
965 const provider = yield* ProviderService.ProviderService;1016 const provider = yield* ProviderService.ProviderService;
apps/server/src/provider/Layers/ProviderService.ts+21-3
...@@ -723,7 +723,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* (...@@ -723,7 +723,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* (
723 });723 });
724724
725 const attachments = parsed.attachments ?? [];725 const attachments = parsed.attachments ?? [];
726 if (!parsed.input && attachments.length === 0) {726 if (!parsed.input && attachments.length === 0 && parsed.continuation !== true) {
727 return yield* toValidationError(727 return yield* toValidationError(
728 "ProviderService.sendTurn",728 "ProviderService.sendTurn",
729 "Either input text or at least one attachment is required",729 "Either input text or at least one attachment is required",
...@@ -777,11 +777,29 @@ const makeProviderService = Effect.fn("makeProviderService")(function* (...@@ -777,11 +777,29 @@ const makeProviderService = Effect.fn("makeProviderService")(function* (
777 let metricProvider = "unknown";777 let metricProvider = "unknown";
778 let metricModel = input.modelSelection?.model;778 let metricModel = input.modelSelection?.model;
779 return yield* Effect.gen(function* () {779 return yield* Effect.gen(function* () {
780 const routed = yield* resolveRoutableSession({780 let routed = yield* resolveRoutableSession({
781 threadId: input.threadId,781 threadId: input.threadId,
782 operation: "ProviderService.sendTurn",782 operation: "ProviderService.sendTurn",
783 allowRecovery: true,783 allowRecovery: false,
784 });784 });
785 if (
786 input.continuation === true &&
787 !input.input &&
788 attachments.length === 0 &&
789 routed.adapter.capabilities.promptlessTurnContinuation !== true
790 ) {
791 return yield* toValidationError(
792 "ProviderService.sendTurn",
793 `Provider '${routed.adapter.provider}' requires an explicit continuation prompt`,
794 );
795 }
796 if (!routed.isActive) {
797 routed = yield* resolveRoutableSession({
798 threadId: input.threadId,
799 operation: "ProviderService.sendTurn",
800 allowRecovery: true,
801 });
802 }
785 metricProvider = routed.adapter.provider;803 metricProvider = routed.adapter.provider;
786 metricModel = input.modelSelection?.model;804 metricModel = input.modelSelection?.model;
787 yield* Effect.annotateCurrentSpan({805 yield* Effect.annotateCurrentSpan({
apps/server/src/provider/Services/ProviderAdapter.ts+3
...@@ -32,6 +32,9 @@ export interface ProviderAdapterCapabilities {...@@ -32,6 +32,9 @@ export interface ProviderAdapterCapabilities {
32 * Declares whether changing the model on an existing session is supported.32 * Declares whether changing the model on an existing session is supported.
33 */33 */
34 readonly sessionModelSwitch: ProviderSessionModelSwitchMode;34 readonly sessionModelSwitch: ProviderSessionModelSwitchMode;
35 /** Starts a resumed turn with no synthetic user prompt. Omitted means the
36 adapter needs an explicit continuation instruction. */
37 readonly promptlessTurnContinuation?: boolean;
35}38}
3639
37export interface ProviderThreadTurnSnapshot {40export interface ProviderThreadTurnSnapshot {
apps/server/src/server.test.ts+2
...@@ -937,6 +937,8 @@ const buildAppUnderTest = (options?: {...@@ -937,6 +937,8 @@ const buildAppUnderTest = (options?: {
937 Layer.mock(ServerRuntimeStartup.ServerRuntimeStartup)({937 Layer.mock(ServerRuntimeStartup.ServerRuntimeStartup)({
938 awaitCommandReady: Effect.void,938 awaitCommandReady: Effect.void,
939 markHttpListening: Effect.void,939 markHttpListening: Effect.void,
940 markRunningProviderSessionsForContinuation: Effect.succeed([]),
941 clearProviderSessionContinuationMarkers: () => Effect.void,
940 enqueueCommand: (effect) => effect,942 enqueueCommand: (effect) => effect,
941 ...options?.layers?.serverRuntimeStartup,943 ...options?.layers?.serverRuntimeStartup,
942 }),944 }),
apps/server/src/serverRuntimeStartup.reconcile.test.ts+365-4
...@@ -3,10 +3,12 @@ import {...@@ -3,10 +3,12 @@ import {
3 type OrchestrationCommand,3 type OrchestrationCommand,
4 ProviderDriverKind,4 ProviderDriverKind,
5 ProviderInstanceId,5 ProviderInstanceId,
6 type ProviderSendTurnInput,
6 ThreadId,7 ThreadId,
7 TurnId,8 TurnId,
8} from "@t3tools/contracts";9} from "@t3tools/contracts";
9import { assert, it } from "@effect/vitest";10import { assert, it } from "@effect/vitest";
11import * as Deferred from "effect/Deferred";
10import * as Effect from "effect/Effect";12import * as Effect from "effect/Effect";
11import * as Option from "effect/Option";13import * as Option from "effect/Option";
12import * as Stream from "effect/Stream";14import * as Stream from "effect/Stream";
...@@ -27,10 +29,12 @@ const makeThread = (...@@ -27,10 +29,12 @@ const makeThread = (
27 status: "starting" | "running" | "ready" | "stopped" | "error",29 status: "starting" | "running" | "ready" | "stopped" | "error",
28 activeTurnId: TurnId | null = null,30 activeTurnId: TurnId | null = null,
29 archivedAt: string | null = null,31 archivedAt: string | null = null,
32 deletedAt: string | null = null,
30) => ({33) => ({
31 id: ThreadId.make(id),34 id: ThreadId.make(id),
32 archivedAt,35 archivedAt,
33 deletedAt: null,36 deletedAt,
37 interactionMode: "default" as const,
34 session: {38 session: {
35 threadId: ThreadId.make(id),39 threadId: ThreadId.make(id),
36 status,40 status,
...@@ -67,6 +71,7 @@ const queryWithThreads = (threads: ReadonlyArray<ReturnType<typeof makeThread>>)...@@ -67,6 +71,7 @@ const queryWithThreads = (threads: ReadonlyArray<ReturnType<typeof makeThread>>)
67const runReconciliation = (input: {71const runReconciliation = (input: {
68 readonly threads: ReadonlyArray<ReturnType<typeof makeThread>>;72 readonly threads: ReadonlyArray<ReturnType<typeof makeThread>>;
69 readonly liveThreadIds?: ReadonlyArray<ThreadId>;73 readonly liveThreadIds?: ReadonlyArray<ThreadId>;
74 readonly providerService?: ProviderService.ProviderService["Service"];
70 readonly directory: ProviderSessionDirectory.ProviderSessionDirectory["Service"];75 readonly directory: ProviderSessionDirectory.ProviderSessionDirectory["Service"];
71 readonly dispatch: OrchestrationEngine.OrchestrationEngineService["Service"]["dispatch"];76 readonly dispatch: OrchestrationEngine.OrchestrationEngineService["Service"]["dispatch"];
72}) =>77}) =>
...@@ -77,7 +82,7 @@ const runReconciliation = (input: {...@@ -77,7 +82,7 @@ const runReconciliation = (input: {
77 ),82 ),
78 Effect.provideService(83 Effect.provideService(
79 ProviderService.ProviderService,84 ProviderService.ProviderService,
80 makeProviderService(input.liveThreadIds),85 input.providerService ?? makeProviderService(input.liveThreadIds),
81 ),86 ),
82 Effect.provideService(ProviderSessionDirectory.ProviderSessionDirectory, input.directory),87 Effect.provideService(ProviderSessionDirectory.ProviderSessionDirectory, input.directory),
83 Effect.provideService(OrchestrationEngine.OrchestrationEngineService, {88 Effect.provideService(OrchestrationEngine.OrchestrationEngineService, {
...@@ -90,6 +95,347 @@ const runReconciliation = (input: {...@@ -90,6 +95,347 @@ const runReconciliation = (input: {
90 Effect.provide(NodeServices.layer),95 Effect.provide(NodeServices.layer),
91 );96 );
9297
98it.effect("marks active running sessions that have persisted resume state", () => {
99 const active = makeThread("thread-mark-active", "running", TurnId.make("turn-mark-active"));
100 const archived = makeThread(
101 "thread-mark-archived",
102 "running",
103 TurnId.make("turn-mark-archived"),
104 updatedAt,
105 );
106 const ready = makeThread("thread-mark-ready", "ready");
107 const missingResumeState = makeThread(
108 "thread-mark-missing-resume-state",
109 "running",
110 TurnId.make("turn-mark-missing-resume-state"),
111 );
112 const bindingReads: ThreadId[] = [];
113 const upserts: ProviderSessionDirectory.ProviderRuntimeBinding[] = [];
114
115 return ServerRuntimeStartup.markRunningProviderSessionsForContinuation.pipe(
116 Effect.provideService(
117 ProjectionSnapshotQuery.ProjectionSnapshotQuery,
118 queryWithThreads([active, archived, ready, missingResumeState]),
119 ),
120 Effect.provideService(ProviderSessionDirectory.ProviderSessionDirectory, {
121 getBinding: (threadId) =>
122 Effect.sync(() => bindingReads.push(threadId)).pipe(
123 Effect.as(
124 Option.some({
125 threadId,
126 provider: ProviderDriverKind.make("codex"),
127 providerInstanceId,
128 ...(threadId === active.id ? { resumeCursor: { threadId } } : {}),
129 runtimePayload: { activeTurnId: "turn-mark-active" },
130 }),
131 ),
132 ),
133 upsert: (binding) => Effect.sync(() => upserts.push(binding)),
134 getProvider: () => Effect.die("unused"),
135 listThreadIds: () => Effect.die("unused"),
136 listBindings: () => Effect.die("unused"),
137 }),
138 Effect.tap((marked) =>
139 Effect.sync(() => {
140 assert.deepStrictEqual(bindingReads, [active.id, missingResumeState.id]);
141 assert.deepStrictEqual(marked, [active.id]);
142 assert.deepStrictEqual(upserts[0]?.runtimePayload, {
143 activeTurnId: "turn-mark-active",
144 continueAfterServerUpdate: active.session.activeTurnId,
145 });
146 }),
147 ),
148 );
149});
150
151it.effect("continues marked sessions after activation with provider-specific input", () =>
152 Effect.gen(function* () {
153 const codex = makeThread(
154 "thread-continue-codex",
155 "running",
156 TurnId.make("turn-continue-codex"),
157 );
158 const fallback = makeThread("thread-continue-fallback", "starting");
159 const fallbackContinuationTurnId = TurnId.make("turn-continue-fallback");
160 const fallbackProviderInstanceId = ProviderInstanceId.make("claudeAgent");
161 const continuationSent = yield* Deferred.make<void>();
162 const continuationCleared = yield* Deferred.make<void>();
163 const sends: ProviderSendTurnInput[] = [];
164 const dispatched: OrchestrationCommand[] = [];
165 const upserts: ProviderSessionDirectory.ProviderRuntimeBinding[] = [];
166 const bindings = new Map<ThreadId, ProviderSessionDirectory.ProviderRuntimeBinding>(
167 [codex, fallback].map((thread) => [
168 thread.id,
169 {
170 threadId: thread.id,
171 provider:
172 thread.id === codex.id
173 ? ProviderDriverKind.make("codex")
174 : ProviderDriverKind.make("claudeAgent"),
175 providerInstanceId:
176 thread.id === codex.id ? providerInstanceId : fallbackProviderInstanceId,
177 status: "running" as const,
178 runtimePayload: {
179 continueAfterServerUpdate:
180 thread.id === codex.id ? codex.session.activeTurnId : fallbackContinuationTurnId,
181 },
182 },
183 ]),
184 );
185 const providerService: ProviderService.ProviderService["Service"] = {
186 ...makeProviderService(),
187 getCapabilities: (instanceId) =>
188 Effect.succeed({
189 sessionModelSwitch: "in-session",
190 ...(instanceId === providerInstanceId ? { promptlessTurnContinuation: true } : {}),
191 }),
192 sendTurn: (input) =>
193 Effect.gen(function* () {
194 sends.push(input);
195 if (sends.length === 2) {
196 yield* Deferred.succeed(continuationSent, undefined);
197 }
198 return {
199 threadId: input.threadId,
200 turnId: TurnId.make(`continued-${String(input.threadId)}`),
201 };
202 }),
203 };
204
205 yield* runReconciliation({
206 threads: [codex, fallback],
207 providerService,
208 directory: {
209 getBinding: (threadId) =>
210 Effect.sync(() => {
211 const binding = bindings.get(threadId);
212 return binding === undefined ? Option.none() : Option.some(binding);
213 }),
214 upsert: (binding) =>
215 Effect.sync(() => {
216 bindings.set(binding.threadId, binding);
217 upserts.push(binding);
218 const clearedCount = upserts.filter((candidate) => {
219 const payload = candidate.runtimePayload;
220 return (
221 payload !== null &&
222 typeof payload === "object" &&
223 !Array.isArray(payload) &&
224 "continueAfterServerUpdate" in payload &&
225 payload.continueAfterServerUpdate === null
226 );
227 }).length;
228 return clearedCount === 1;
229 }).pipe(
230 Effect.flatMap((firstMarkerCleared) =>
231 firstMarkerCleared ? Deferred.succeed(continuationCleared, undefined) : Effect.void,
232 ),
233 ),
234 getProvider: () => Effect.die("unused"),
235 listThreadIds: () => Effect.die("unused"),
236 listBindings: () => Effect.die("unused"),
237 },
238 dispatch: (command) =>
239 Effect.sync(() => dispatched.push(command)).pipe(
240 Effect.as({ sequence: dispatched.length }),
241 ),
242 });
243 yield* Deferred.await(continuationSent);
244 yield* Deferred.await(continuationCleared);
245
246 assert.deepStrictEqual(
247 sends.toSorted((left, right) => String(left.threadId).localeCompare(String(right.threadId))),
248 [
249 { threadId: codex.id, continuation: true, interactionMode: "default" },
250 {
251 threadId: fallback.id,
252 input: "Continue where you left off.",
253 interactionMode: "default",
254 },
255 ],
256 );
257 assert.deepStrictEqual(
258 dispatched.map((command) =>
259 command.type === "thread.session.set"
260 ? {
261 threadId: command.threadId,
262 status: command.session.status,
263 activeTurnId: command.session.activeTurnId,
264 }
265 : null,
266 ),
267 [
268 {
269 threadId: codex.id,
270 status: "starting",
271 activeTurnId: null,
272 },
273 {
274 threadId: fallback.id,
275 status: "starting",
276 activeTurnId: fallback.session.activeTurnId,
277 },
278 ],
279 );
280 for (const [thread, continuationTurnId] of [
281 [codex, codex.session.activeTurnId],
282 [fallback, fallbackContinuationTurnId],
283 ] as const) {
284 assert.deepStrictEqual(
285 upserts
286 .filter((binding) => binding.threadId === thread.id)
287 .map((binding) => binding.runtimePayload)[0],
288 {
289 continueAfterServerUpdate: continuationTurnId,
290 activeTurnId: null,
291 },
292 );
293 }
294 assert.equal(
295 upserts.some((binding) => {
296 const payload = binding.runtimePayload;
297 return (
298 payload !== null &&
299 typeof payload === "object" &&
300 !Array.isArray(payload) &&
301 "continueAfterServerUpdate" in payload &&
302 payload.continueAfterServerUpdate === null
303 );
304 }),
305 true,
306 );
307 }),
308);
309
310it.effect("does not continue archived or deleted marked sessions", () => {
311 const archived = makeThread(
312 "thread-continue-archived",
313 "running",
314 TurnId.make("turn-continue-archived"),
315 updatedAt,
316 );
317 const deleted = makeThread(
318 "thread-continue-deleted",
319 "running",
320 TurnId.make("turn-continue-deleted"),
321 null,
322 updatedAt,
323 );
324 const sends: ProviderSendTurnInput[] = [];
325 const dispatched: OrchestrationCommand[] = [];
326
327 return runReconciliation({
328 threads: [archived, deleted],
329 providerService: {
330 ...makeProviderService(),
331 sendTurn: (input) =>
332 Effect.sync(() => {
333 sends.push(input);
334 return {
335 threadId: input.threadId,
336 turnId: TurnId.make("unexpected-archived-turn"),
337 };
338 }),
339 },
340 directory: {
341 getBinding: (threadId) => {
342 const thread = threadId === archived.id ? archived : deleted;
343 return Effect.succeed(
344 Option.some({
345 threadId,
346 provider: ProviderDriverKind.make("codex"),
347 providerInstanceId,
348 status: "running" as const,
349 resumeCursor: { cursor: threadId },
350 runtimePayload: {
351 continueAfterServerUpdate: thread.session.activeTurnId,
352 },
353 }),
354 );
355 },
356 upsert: () => Effect.void,
357 getProvider: () => Effect.die("unused"),
358 listThreadIds: () => Effect.die("unused"),
359 listBindings: () => Effect.die("unused"),
360 },
361 dispatch: (command) =>
362 Effect.sync(() => dispatched.push(command)).pipe(Effect.as({ sequence: dispatched.length })),
363 }).pipe(
364 Effect.tap(() =>
365 Effect.sync(() => {
366 assert.deepStrictEqual(sends, []);
367 assert.deepStrictEqual(
368 dispatched.map((command) =>
369 command.type === "thread.session.set"
370 ? { threadId: command.threadId, status: command.session.status }
371 : null,
372 ),
373 [
374 { threadId: archived.id, status: "error" },
375 { threadId: deleted.id, status: "error" },
376 ],
377 );
378 }),
379 ),
380 );
381});
382
383it.effect("retries continuation preparation before settling a persistent failure", () => {
384 const thread = makeThread(
385 "thread-continuation-preparation-failure",
386 "running",
387 TurnId.make("turn-continuation-preparation-failure"),
388 );
389 const dispatched: OrchestrationCommand[] = [];
390 const failure = new OrchestrationCommandInvariantError({
391 commandType: "thread.session.set",
392 detail: "simulated continuation preparation failure",
393 });
394
395 return runReconciliation({
396 threads: [thread],
397 directory: {
398 getBinding: () =>
399 Effect.succeed(
400 Option.some({
401 threadId: thread.id,
402 provider: ProviderDriverKind.make("codex"),
403 providerInstanceId,
404 status: "running" as const,
405 resumeCursor: { cursor: thread.id },
406 runtimePayload: {
407 continueAfterServerUpdate: thread.session.activeTurnId,
408 },
409 }),
410 ),
411 upsert: () => Effect.void,
412 getProvider: () => Effect.die("unused"),
413 listThreadIds: () => Effect.die("unused"),
414 listBindings: () => Effect.die("unused"),
415 },
416 dispatch: (command) => {
417 if (command.type !== "thread.session.set") {
418 return Effect.die("unexpected command");
419 }
420 dispatched.push(command);
421 return command.session.status === "starting"
422 ? Effect.fail(failure)
423 : Effect.succeed({ sequence: dispatched.length });
424 },
425 }).pipe(
426 Effect.tap(() =>
427 Effect.sync(() =>
428 assert.deepStrictEqual(
429 dispatched.map(
430 (command) => command.type === "thread.session.set" && command.session.status,
431 ),
432 ["starting", "starting", "error"],
433 ),
434 ),
435 ),
436 );
437});
438
93it.effect("reconciles multiple active and archived orphans but skips live sessions", () => {439it.effect("reconciles multiple active and archived orphans but skips live sessions", () => {
94 const starting = makeThread("thread-starting", "starting");440 const starting = makeThread("thread-starting", "starting");
95 const running = makeThread("thread-running", "running", TurnId.make("turn-running"));441 const running = makeThread("thread-running", "running", TurnId.make("turn-running"));
...@@ -123,7 +469,13 @@ it.effect("reconciles multiple active and archived orphans but skips live sessio...@@ -123,7 +469,13 @@ it.effect("reconciles multiple active and archived orphans but skips live sessio
123 providerInstanceId,469 providerInstanceId,
124 status: "running" as const,470 status: "running" as const,
125 resumeCursor: { cursor: candidate },471 resumeCursor: { cursor: candidate },
126 runtimePayload: { activeTurnId: "stale", unrelated: candidate },472 runtimePayload: {
473 activeTurnId: "stale",
474 unrelated: candidate,
475 ...(candidate === staleActiveTurn.id
476 ? { continueAfterServerUpdate: "turn-from-an-earlier-update" }
477 : {}),
478 },
127 }),479 }),
128 ),480 ),
129 ),481 ),
...@@ -157,7 +509,16 @@ it.effect("reconciles multiple active and archived orphans but skips live sessio...@@ -157,7 +509,16 @@ it.effect("reconciles multiple active and archived orphans but skips live sessio
157 assert.equal(upserts.length, orphanIds.length);509 assert.equal(upserts.length, orphanIds.length);
158 for (const binding of upserts) {510 for (const binding of upserts) {
159 assert.equal(binding.status, "stopped");511 assert.equal(binding.status, "stopped");
160 assert.deepStrictEqual(binding.runtimePayload, { activeTurnId: null });512 assert.deepStrictEqual(
513 binding.runtimePayload,
514 binding.threadId === staleActiveTurn.id
515 ? {
516 activeTurnId: null,
517 unrelated: binding.threadId,
518 continueAfterServerUpdate: null,
519 }
520 : { activeTurnId: null, unrelated: binding.threadId },
521 );
161 assert.deepStrictEqual(binding.resumeCursor, { cursor: binding.threadId });522 assert.deepStrictEqual(binding.resumeCursor, { cursor: binding.threadId });
162 }523 }
163 }),524 }),
apps/server/src/serverRuntimeStartup.ts+308-36
...@@ -6,6 +6,7 @@ import {...@@ -6,6 +6,7 @@ import {
6 ProjectId,6 ProjectId,
7 ProviderInstanceId,7 ProviderInstanceId,
8 ThreadId,8 ThreadId,
9 TurnId,
9} from "@t3tools/contracts";10} from "@t3tools/contracts";
10import * as Cause from "effect/Cause";11import * as Cause from "effect/Cause";
11import * as Console from "effect/Console";12import * as Console from "effect/Console";
...@@ -65,6 +66,13 @@ export class ServerRuntimeStartup extends Context.Service<...@@ -65,6 +66,13 @@ export class ServerRuntimeStartup extends Context.Service<
65 {66 {
66 readonly awaitCommandReady: Effect.Effect<void, ServerRuntimeStartupError>;67 readonly awaitCommandReady: Effect.Effect<void, ServerRuntimeStartupError>;
67 readonly markHttpListening: Effect.Effect<void>;68 readonly markHttpListening: Effect.Effect<void>;
69 readonly markRunningProviderSessionsForContinuation: Effect.Effect<
70 ReadonlyArray<ThreadId>,
71 ServerUpdateThreadContinuationError
72 >;
73 readonly clearProviderSessionContinuationMarkers: (
74 threadIds: ReadonlyArray<ThreadId>,
75 ) => Effect.Effect<void, ServerUpdateThreadContinuationError>;
68 readonly enqueueCommand: <A, E>(76 readonly enqueueCommand: <A, E>(
69 effect: Effect.Effect<A, E>,77 effect: Effect.Effect<A, E>,
70 ) => Effect.Effect<A, E | ServerRuntimeStartupError>;78 ) => Effect.Effect<A, E | ServerRuntimeStartupError>;
...@@ -294,6 +302,138 @@ const runStartupPhase = <A, E, R>(phase: string, effect: Effect.Effect<A, E, R>)...@@ -294,6 +302,138 @@ const runStartupPhase = <A, E, R>(phase: string, effect: Effect.Effect<A, E, R>)
294302
295const ORPHANED_PROVIDER_SESSION_ERROR =303const ORPHANED_PROVIDER_SESSION_ERROR =
296 "Provider session did not survive a server restart. Send a new message to continue.";304 "Provider session did not survive a server restart. Send a new message to continue.";
305const SERVER_UPDATE_CONTINUATION_KEY = "continueAfterServerUpdate";
306const SERVER_UPDATE_CONTINUATION_PROMPT = "Continue where you left off.";
307
308class ProviderSessionContinuationError extends Schema.TaggedErrorClass<ProviderSessionContinuationError>()(
309 "ProviderSessionContinuationError",
310 {
311 threadId: ThreadId,
312 },
313) {
314 override get message(): string {
315 return `Could not continue thread '${this.threadId}': the provider instance is missing.`;
316 }
317}
318
319export class ServerUpdateThreadContinuationError extends Schema.TaggedErrorClass<ServerUpdateThreadContinuationError>()(
320 "ServerUpdateThreadContinuationError",
321 {
322 cause: Schema.Defect(),
323 },
324) {
325 override get message(): string {
326 return "Could not prepare running threads to continue after the update.";
327 }
328}
329
330function hasServerUpdateContinuationMarker(
331 runtimePayload: unknown,
332): runtimePayload is Record<string, unknown> {
333 return (
334 runtimePayload !== null &&
335 typeof runtimePayload === "object" &&
336 !Array.isArray(runtimePayload) &&
337 SERVER_UPDATE_CONTINUATION_KEY in runtimePayload
338 );
339}
340
341function readRuntimePayload(runtimePayload: unknown): Record<string, unknown> {
342 return runtimePayload !== null &&
343 typeof runtimePayload === "object" &&
344 !Array.isArray(runtimePayload)
345 ? (runtimePayload as Record<string, unknown>)
346 : {};
347}
348
349const isServerUpdateThreadContinuationError = Schema.is(ServerUpdateThreadContinuationError);
350
351function readServerUpdateContinuationTurnId(runtimePayload: unknown): TurnId | null {
352 if (!hasServerUpdateContinuationMarker(runtimePayload)) {
353 return null;
354 }
355 const value = runtimePayload[SERVER_UPDATE_CONTINUATION_KEY];
356 return typeof value === "string" && value.length > 0 ? TurnId.make(value) : null;
357}
358
359const toServerUpdateThreadContinuationError = (cause: unknown) =>
360 isServerUpdateThreadContinuationError(cause)
361 ? cause
362 : new ServerUpdateThreadContinuationError({ cause });
363
364export const markRunningProviderSessionsForContinuation = Effect.gen(function* () {
365 const directory = yield* ProviderSessionDirectory.ProviderSessionDirectory;
366 const query = yield* ProjectionSnapshotQuery.ProjectionSnapshotQuery;
367 const { threads } = yield* query.getCommandReadModel();
368 const running = threads.filter(
369 (thread) =>
370 thread.archivedAt === null &&
371 thread.deletedAt === null &&
372 thread.session?.status === "running" &&
373 thread.session.activeTurnId !== null,
374 );
375
376 const marked: ThreadId[] = [];
377 return yield* Effect.gen(function* () {
378 for (const thread of running) {
379 const activeTurnId = thread.session?.activeTurnId;
380 if (activeTurnId === null || activeTurnId === undefined) {
381 continue;
382 }
383 const binding = yield* directory.getBinding(thread.id);
384 if (Option.isNone(binding)) {
385 continue;
386 }
387 if (binding.value.resumeCursor === null || binding.value.resumeCursor === undefined) {
388 continue;
389 }
390 yield* directory.upsert({
391 ...binding.value,
392 runtimePayload: {
393 ...readRuntimePayload(binding.value.runtimePayload),
394 [SERVER_UPDATE_CONTINUATION_KEY]: activeTurnId,
395 },
396 });
397 marked.push(thread.id);
398 }
399 return marked;
400 }).pipe(
401 Effect.catchCause((cause) =>
402 clearProviderSessionContinuationMarkers(marked).pipe(Effect.andThen(Effect.failCause(cause))),
403 ),
404 );
405}).pipe(Effect.mapError(toServerUpdateThreadContinuationError));
406
407const clearContinuationMarkers = (
408 directory: ProviderSessionDirectory.ProviderSessionDirectory["Service"],
409 threadIds: ReadonlyArray<ThreadId>,
410) =>
411 Effect.forEach(
412 threadIds,
413 (threadId) =>
414 directory.getBinding(threadId).pipe(
415 Effect.flatMap(
416 Option.match({
417 onNone: () => Effect.void,
418 onSome: (binding) =>
419 directory.upsert({
420 ...binding,
421 runtimePayload: {
422 ...readRuntimePayload(binding.runtimePayload),
423 [SERVER_UPDATE_CONTINUATION_KEY]: null,
424 },
425 }),
426 }),
427 ),
428 ),
429 { concurrency: "unbounded", discard: true },
430 );
431
432export const clearProviderSessionContinuationMarkers = (threadIds: ReadonlyArray<ThreadId>) =>
433 Effect.gen(function* () {
434 const directory = yield* ProviderSessionDirectory.ProviderSessionDirectory;
435 yield* clearContinuationMarkers(directory, threadIds);
436 }).pipe(Effect.mapError(toServerUpdateThreadContinuationError));
297437
298export const reconcileProviderSessions = Effect.gen(function* () {438export const reconcileProviderSessions = Effect.gen(function* () {
299 const crypto = yield* Crypto.Crypto;439 const crypto = yield* Crypto.Crypto;
...@@ -320,52 +460,165 @@ export const reconcileProviderSessions = Effect.gen(function* () {...@@ -320,52 +460,165 @@ export const reconcileProviderSessions = Effect.gen(function* () {
320 if (session === null) {460 if (session === null) {
321 continue;461 continue;
322 }462 }
323 yield* Effect.gen(function* () {463 const binding = yield* directory.getBinding(thread.id).pipe(
324 const binding = yield* directory.getBinding(thread.id);
325 if (Option.isSome(binding)) {
326 yield* directory.upsert({
327 ...binding.value,
328 status: "stopped",
329 runtimePayload: { activeTurnId: null },
330 });
331 }
332 }).pipe(
333 Effect.catchCause((cause) =>464 Effect.catchCause((cause) =>
334 Cause.hasInterrupts(cause)465 Cause.hasInterrupts(cause)
335 ? Effect.failCause(cause)466 ? Effect.failCause(cause)
336 : Effect.logWarning("failed to reconcile orphaned provider session directory binding", {467 : Effect.logWarning("failed to read orphaned provider session directory binding", {
337 threadId: thread.id,468 threadId: thread.id,
338 cause,469 cause,
339 }),470 }).pipe(Effect.as(Option.none())),
340 ),471 ),
341 );472 );
473 const continuationMarkerPresent =
474 Option.isSome(binding) && hasServerUpdateContinuationMarker(binding.value.runtimePayload);
475 const continuationTurnId = Option.isSome(binding)
476 ? readServerUpdateContinuationTurnId(binding.value.runtimePayload)
477 : null;
478 const continuationMarked =
479 continuationTurnId !== null &&
480 (session.activeTurnId === null || continuationTurnId === session.activeTurnId);
481 const settleAsError = (lastError: string) =>
482 Effect.gen(function* () {
483 yield* Effect.gen(function* () {
484 if (Option.isSome(binding)) {
485 yield* directory.upsert({
486 ...binding.value,
487 status: "stopped",
488 runtimePayload: {
489 ...readRuntimePayload(binding.value.runtimePayload),
490 activeTurnId: null,
491 ...(continuationMarkerPresent ? { [SERVER_UPDATE_CONTINUATION_KEY]: null } : {}),
492 },
493 });
494 }
495 }).pipe(
496 Effect.catchCause((cause) =>
497 Cause.hasInterrupts(cause)
498 ? Effect.failCause(cause)
499 : Effect.logWarning(
500 "failed to reconcile orphaned provider session directory binding",
501 { threadId: thread.id, cause },
502 ),
503 ),
504 );
342505
343 yield* Effect.gen(function* () {506 yield* Effect.gen(function* () {
344 const reconciledAt = DateTime.formatIso(yield* DateTime.now);507 const reconciledAt = DateTime.formatIso(yield* DateTime.now);
345 yield* orchestrationEngine.dispatch({508 yield* orchestrationEngine.dispatch({
346 type: "thread.session.set",509 type: "thread.session.set",
347 commandId: CommandId.make(yield* crypto.randomUUIDv4),510 commandId: CommandId.make(yield* crypto.randomUUIDv4),
348 threadId: thread.id,511 threadId: thread.id,
349 session: {512 session: {
350 ...session,513 ...session,
351 status: "error",514 status: "error",
352 activeTurnId: null,515 activeTurnId: null,
353 lastError: ORPHANED_PROVIDER_SESSION_ERROR,516 lastError,
354 updatedAt: reconciledAt,517 updatedAt: reconciledAt,
355 },518 },
356 createdAt: reconciledAt,519 createdAt: reconciledAt,
520 });
521 }).pipe(
522 Effect.retry({ times: 1 }),
523 Effect.catchCause((cause) =>
524 Cause.hasInterrupts(cause)
525 ? Effect.failCause(cause)
526 : Effect.logWarning("failed to settle orphaned provider session projection", {
527 threadId: thread.id,
528 cause,
529 }),
530 ),
531 );
357 });532 });
358 }).pipe(533
359 Effect.retry({ times: 1 }),534 if (
360 Effect.catchCause((cause) =>535 Option.isSome(binding) &&
361 Cause.hasInterrupts(cause)536 continuationMarked &&
362 ? Effect.failCause(cause)537 thread.archivedAt === null &&
363 : Effect.logWarning("failed to settle orphaned provider session projection", {538 thread.deletedAt === null
539 ) {
540 const prepared = yield* Effect.gen(function* () {
541 yield* directory.upsert({
542 ...binding.value,
543 status: "starting",
544 runtimePayload: {
545 ...readRuntimePayload(binding.value.runtimePayload),
546 activeTurnId: null,
547 },
548 });
549 const resumedAt = DateTime.formatIso(yield* DateTime.now);
550 yield* orchestrationEngine.dispatch({
551 type: "thread.session.set",
552 commandId: CommandId.make(yield* crypto.randomUUIDv4),
553 threadId: thread.id,
554 session: {
555 ...session,
556 status: "starting",
557 activeTurnId: null,
558 lastError: null,
559 updatedAt: resumedAt,
560 },
561 createdAt: resumedAt,
562 });
563 }).pipe(Effect.retry({ times: 1 }), Effect.exit);
564 if (Exit.isFailure(prepared)) {
565 if (Cause.hasInterrupts(prepared.cause)) {
566 return yield* Effect.failCause(prepared.cause);
567 }
568 yield* Effect.logWarning("failed to prepare provider session continuation", {
569 threadId: thread.id,
570 cause: prepared.cause,
571 });
572 yield* settleAsError(ORPHANED_PROVIDER_SESSION_ERROR);
573 continue;
574 }
575
576 yield* forkParked(
577 Effect.gen(function* () {
578 const continuation = Effect.gen(function* () {
579 const providerInstanceId = binding.value.providerInstanceId;
580 if (providerInstanceId === undefined) {
581 return yield* new ProviderSessionContinuationError({
582 threadId: thread.id,
583 });
584 }
585 const capabilities = yield* providerService.getCapabilities(providerInstanceId);
586 yield* providerService.sendTurn({
364 threadId: thread.id,587 threadId: thread.id,
365 cause,588 ...(capabilities.promptlessTurnContinuation === true
366 }),589 ? { continuation: true }
367 ),590 : { input: SERVER_UPDATE_CONTINUATION_PROMPT }),
368 );591 interactionMode: thread.interactionMode,
592 });
593 });
594 const continuationExit = yield* Effect.exit(continuation);
595 if (Exit.isSuccess(continuationExit) || Cause.hasInterrupts(continuationExit.cause)) {
596 if (Exit.isSuccess(continuationExit)) {
597 yield* clearContinuationMarkers(directory, [thread.id]).pipe(
598 Effect.uninterruptible,
599 Effect.catchCause((cause) =>
600 Effect.logWarning("failed to clear completed provider session continuation", {
601 threadId: thread.id,
602 cause,
603 }),
604 ),
605 );
606 }
607 return;
608 }
609 yield* Effect.logWarning("failed to continue provider session after server update", {
610 threadId: thread.id,
611 cause: continuationExit.cause,
612 });
613 yield* settleAsError(
614 "Could not continue this thread after the server update. Send a new message to continue.",
615 ).pipe(Effect.ignoreCause);
616 }),
617 );
618 continue;
619 }
620
621 yield* settleAsError(ORPHANED_PROVIDER_SESSION_ERROR);
369 }622 }
370}).pipe(623}).pipe(
371 Effect.catchCause((cause) =>624 Effect.catchCause((cause) =>
...@@ -390,6 +643,8 @@ export const make = (options?: StartupOptions) =>...@@ -390,6 +643,8 @@ export const make = (options?: StartupOptions) =>
390 const lifecycleEvents = yield* ServerLifecycleEvents.ServerLifecycleEvents;643 const lifecycleEvents = yield* ServerLifecycleEvents.ServerLifecycleEvents;
391 const serverSettings = yield* ServerSettings.ServerSettingsService;644 const serverSettings = yield* ServerSettings.ServerSettingsService;
392 const serverEnvironment = yield* ServerEnvironment.ServerEnvironment;645 const serverEnvironment = yield* ServerEnvironment.ServerEnvironment;
646 const projectionSnapshotQuery = yield* ProjectionSnapshotQuery.ProjectionSnapshotQuery;
647 const providerSessionDirectory = yield* ProviderSessionDirectory.ProviderSessionDirectory;
393 const crypto = yield* Crypto.Crypto;648 const crypto = yield* Crypto.Crypto;
394 const launcher = yield* ServiceLauncherClient.ServiceLauncherClient;649 const launcher = yield* ServiceLauncherClient.ServiceLauncherClient;
395650
...@@ -577,6 +832,23 @@ export const make = (options?: StartupOptions) =>...@@ -577,6 +832,23 @@ export const make = (options?: StartupOptions) =>
577 return {832 return {
578 awaitCommandReady: commandGate.awaitCommandReady,833 awaitCommandReady: commandGate.awaitCommandReady,
579 markHttpListening: Deferred.succeed(httpListening, undefined),834 markHttpListening: Deferred.succeed(httpListening, undefined),
835 markRunningProviderSessionsForContinuation: markRunningProviderSessionsForContinuation.pipe(
836 Effect.provideService(
837 ProjectionSnapshotQuery.ProjectionSnapshotQuery,
838 projectionSnapshotQuery,
839 ),
840 Effect.provideService(
841 ProviderSessionDirectory.ProviderSessionDirectory,
842 providerSessionDirectory,
843 ),
844 ),
845 clearProviderSessionContinuationMarkers: (threadIds) =>
846 clearProviderSessionContinuationMarkers(threadIds).pipe(
847 Effect.provideService(
848 ProviderSessionDirectory.ProviderSessionDirectory,
849 providerSessionDirectory,
850 ),
851 ),
580 enqueueCommand: commandGate.enqueueCommand,852 enqueueCommand: commandGate.enqueueCommand,
581 } satisfies ServerRuntimeStartup["Service"];853 } satisfies ServerRuntimeStartup["Service"];
582 });854 });
apps/server/src/ws.ts+31-6
...@@ -50,7 +50,7 @@ import {...@@ -50,7 +50,7 @@ import {
50 ProviderUploadFeedbackError,50 ProviderUploadFeedbackError,
51 RelayClientInstallFailedError,51 RelayClientInstallFailedError,
52 type RelayClientInstallProgressEvent,52 type RelayClientInstallProgressEvent,
53 type ServerSelfUpdateError,53 ServerSelfUpdateError,
54 type ServerSelfUpdateProgressEvent,54 type ServerSelfUpdateProgressEvent,
55 type FilesystemBrowseFailure,55 type FilesystemBrowseFailure,
56 FilesystemBrowseError,56 FilesystemBrowseError,
...@@ -509,7 +509,7 @@ const makeWsRpcLayer = (...@@ -509,7 +509,7 @@ const makeWsRpcLayer = (
509 const providerRegistry = yield* ProviderRegistry.ProviderRegistry;509 const providerRegistry = yield* ProviderRegistry.ProviderRegistry;
510 const providerService = yield* ProviderService.ProviderService;510 const providerService = yield* ProviderService.ProviderService;
511 const providerMaintenanceRunner = yield* ProviderMaintenanceRunner.ProviderMaintenanceRunner;511 const providerMaintenanceRunner = yield* ProviderMaintenanceRunner.ProviderMaintenanceRunner;
512 const serverSelfUpdate = yield* ServerSelfUpdate.ServerSelfUpdate;512 const serverUpdate = yield* ServerSelfUpdate.ServerSelfUpdate;
513 const config = yield* ServerConfig.ServerConfig;513 const config = yield* ServerConfig.ServerConfig;
514 const lifecycleEvents = yield* ServerLifecycleEvents.ServerLifecycleEvents;514 const lifecycleEvents = yield* ServerLifecycleEvents.ServerLifecycleEvents;
515 const serverSettings = yield* ServerSettings.ServerSettingsService;515 const serverSettings = yield* ServerSettings.ServerSettingsService;
...@@ -1700,14 +1700,14 @@ const makeWsRpcLayer = (...@@ -1700,14 +1700,14 @@ const makeWsRpcLayer = (
1700 },1700 },
1701 ),1701 ),
1702 [WS_METHODS.serverUpdateServer]: (input) =>1702 [WS_METHODS.serverUpdateServer]: (input) =>
1703 observeRpcEffect(WS_METHODS.serverUpdateServer, serverSelfUpdate.update(input), {1703 observeRpcEffect(WS_METHODS.serverUpdateServer, serverUpdate.update(input), {
1704 "rpc.aggregate": "server",1704 "rpc.aggregate": "server",
1705 }),1705 }),
1706 [WS_METHODS.serverUpdateServerWithProgress]: (input) =>1706 [WS_METHODS.serverUpdateServerWithProgress]: (input) =>
1707 observeRpcStream(1707 observeRpcStream(
1708 WS_METHODS.serverUpdateServerWithProgress,1708 WS_METHODS.serverUpdateServerWithProgress,
1709 Stream.callback<ServerSelfUpdateProgressEvent, ServerSelfUpdateError>((queue) =>1709 Stream.callback<ServerSelfUpdateProgressEvent, ServerSelfUpdateError>((queue) =>
1710 serverSelfUpdate1710 serverUpdate
1711 .update(input, (stage) =>1711 .update(input, (stage) =>
1712 Queue.offer(queue, {1712 Queue.offer(queue, {
1713 type: "progress",1713 type: "progress",
...@@ -1733,7 +1733,7 @@ const makeWsRpcLayer = (...@@ -1733,7 +1733,7 @@ const makeWsRpcLayer = (
1733 [WS_METHODS.serverCommitDesktopUpdate]: (input) =>1733 [WS_METHODS.serverCommitDesktopUpdate]: (input) =>
1734 observeRpcEffect(1734 observeRpcEffect(
1735 WS_METHODS.serverCommitDesktopUpdate,1735 WS_METHODS.serverCommitDesktopUpdate,
1736 serverSelfUpdate.commitDesktopUpdate(input.requestId),1736 serverUpdate.commitDesktopUpdate(input.requestId),
1737 { "rpc.aggregate": "server" },1737 { "rpc.aggregate": "server" },
1738 ),1738 ),
1739 [WS_METHODS.serverUpsertKeybinding]: (rule) =>1739 [WS_METHODS.serverUpsertKeybinding]: (rule) =>
...@@ -2578,7 +2578,32 @@ const makeWsRpcLayer = (...@@ -2578,7 +2578,32 @@ const makeWsRpcLayer = (
2578export const websocketRpcRouteLayer = Layer.unwrap(2578export const websocketRpcRouteLayer = Layer.unwrap(
2579 Effect.gen(function* () {2579 Effect.gen(function* () {
2580 const previewAutomationBroker = yield* PreviewAutomationBroker.PreviewAutomationBroker;2580 const previewAutomationBroker = yield* PreviewAutomationBroker.PreviewAutomationBroker;
2581 const serverSelfUpdate = yield* ServerSelfUpdate.ServerSelfUpdate;2581 const baseServerSelfUpdate = yield* ServerSelfUpdate.ServerSelfUpdate;
2582 const config = yield* ServerConfig.ServerConfig;
2583 const startup = yield* ServerRuntimeStartup.ServerRuntimeStartup;
2584 const serverSelfUpdate = yield* ServerSelfUpdate.withRunningThreadContinuation({
2585 mode: config.mode,
2586 selfUpdate: baseServerSelfUpdate,
2587 prepare: startup.markRunningProviderSessionsForContinuation.pipe(
2588 Effect.mapError(
2589 (cause) =>
2590 new ServerSelfUpdateError({
2591 reason: "Could not prepare running threads to continue after the update.",
2592 cause,
2593 }),
2594 ),
2595 ),
2596 clear: (threadIds) =>
2597 startup.clearProviderSessionContinuationMarkers(threadIds).pipe(
2598 Effect.mapError(
2599 (cause) =>
2600 new ServerSelfUpdateError({
2601 reason: "Could not clear thread continuation markers after the update failed.",
2602 cause,
2603 }),
2604 ),
2605 ),
2606 });
2582 const pullRequests = yield* PullRequestService.PullRequestService;2607 const pullRequests = yield* PullRequestService.PullRequestService;
2583 return HttpRouter.add(2608 return HttpRouter.add(
2584 "GET",2609 "GET",
apps/web/src/components/ChatView.tsx+4
...@@ -428,6 +428,7 @@ import {...@@ -428,6 +428,7 @@ import {
428 resolveServerSelfUpdateCapability,428 resolveServerSelfUpdateCapability,
429 serverUpdateGuidance,429 serverUpdateGuidance,
430 supportsDesktopAppUpdate,430 supportsDesktopAppUpdate,
431 supportsServerUpdateThreadContinuation,
431} from "../versionSkew";432} from "../versionSkew";
432import { useAssetUrls } from "../assets/assetUrls";433import { useAssetUrls } from "../assets/assetUrls";
433434
...@@ -2221,6 +2222,7 @@ function ChatViewContent(props: ChatViewProps) {...@@ -2221,6 +2222,7 @@ function ChatViewContent(props: ChatViewProps) {
2221 const serverUpdateEnvironmentId = activeThread?.environmentId ?? null;2222 const serverUpdateEnvironmentId = activeThread?.environmentId ?? null;
2222 const versionMismatchSelfUpdate = resolveServerSelfUpdateCapability(serverConfig);2223 const versionMismatchSelfUpdate = resolveServerSelfUpdateCapability(serverConfig);
2223 const versionMismatchDesktopAppUpdate = supportsDesktopAppUpdate(serverConfig);2224 const versionMismatchDesktopAppUpdate = supportsDesktopAppUpdate(serverConfig);
2225 const versionMismatchThreadContinuation = supportsServerUpdateThreadContinuation(serverConfig);
2224 const serverUpdateState = useAtomValue(2226 const serverUpdateState = useAtomValue(
2225 serverEnvironment.updateStateAtom(serverUpdateEnvironmentId),2227 serverEnvironment.updateStateAtom(serverUpdateEnvironmentId),
2226 );2228 );
...@@ -2356,6 +2358,7 @@ function ChatViewContent(props: ChatViewProps) {...@@ -2356,6 +2358,7 @@ function ChatViewContent(props: ChatViewProps) {
2356 serverLabel={versionMismatchServerLabel}2358 serverLabel={versionMismatchServerLabel}
2357 selfUpdate={versionMismatchSelfUpdate}2359 selfUpdate={versionMismatchSelfUpdate}
2358 desktopAppUpdate={versionMismatchDesktopAppUpdate}2360 desktopAppUpdate={versionMismatchDesktopAppUpdate}
2361 threadContinuation={versionMismatchThreadContinuation}
2359 targetVersion={versionMismatch.clientVersion}2362 targetVersion={versionMismatch.clientVersion}
2360 label={updateFailed ? "Retry" : "Update"}2363 label={updateFailed ? "Retry" : "Update"}
2361 variant="ghost"2364 variant="ghost"
...@@ -2391,6 +2394,7 @@ function ChatViewContent(props: ChatViewProps) {...@@ -2391,6 +2394,7 @@ function ChatViewContent(props: ChatViewProps) {
2391 serverUpdateEnvironmentId,2394 serverUpdateEnvironmentId,
2392 versionMismatchSelfUpdate,2395 versionMismatchSelfUpdate,
2393 versionMismatchDesktopAppUpdate,2396 versionMismatchDesktopAppUpdate,
2397 versionMismatchThreadContinuation,
2394 versionMismatchServerLabel,2398 versionMismatchServerLabel,
2395 ]);2399 ]);
2396 const providerStatuses = serverConfig?.providers ?? EMPTY_PROVIDERS;2400 const providerStatuses = serverConfig?.providers ?? EMPTY_PROVIDERS;
apps/web/src/components/ServerUpdateAction.test.tsx+50
...@@ -8,11 +8,17 @@ import { beforeEach, describe, expect, it, vi } from "vite-plus/test";...@@ -8,11 +8,17 @@ import { beforeEach, describe, expect, it, vi } from "vite-plus/test";
8const testState = vi.hoisted(() => ({8const testState = vi.hoisted(() => ({
9 updateServer: vi.fn(),9 updateServer: vi.fn(),
10 toast: vi.fn(),10 toast: vi.fn(),
11 continueThreadsAfterServerUpdate: false,
11}));12}));
1213
13vi.mock("~/hooks/useCopyToClipboard", () => ({14vi.mock("~/hooks/useCopyToClipboard", () => ({
14 useCopyToClipboard: () => ({ copyToClipboard: vi.fn() }),15 useCopyToClipboard: () => ({ copyToClipboard: vi.fn() }),
15}));16}));
17vi.mock("~/hooks/useSettings", () => ({
18 useClientSettings: (
19 selector: (settings: { continueThreadsAfterServerUpdate: boolean }) => unknown,
20 ) => selector({ continueThreadsAfterServerUpdate: testState.continueThreadsAfterServerUpdate }),
21}));
16vi.mock("~/state/server", () => ({22vi.mock("~/state/server", () => ({
17 serverEnvironment: { updateServer: Symbol("updateServer") },23 serverEnvironment: { updateServer: Symbol("updateServer") },
18}));24}));
...@@ -47,6 +53,7 @@ describe("ServerUpdateAction", () => {...@@ -47,6 +53,7 @@ describe("ServerUpdateAction", () => {
47 beforeEach(() => {53 beforeEach(() => {
48 testState.updateServer.mockReset();54 testState.updateServer.mockReset();
49 testState.toast.mockReset();55 testState.toast.mockReset();
56 testState.continueThreadsAfterServerUpdate = false;
50 });57 });
5158
52 it("reports success only after the shared update flow reconnects", async () => {59 it("reports success only after the shared update flow reconnects", async () => {
...@@ -141,6 +148,49 @@ describe("ServerUpdateAction", () => {...@@ -141,6 +148,49 @@ describe("ServerUpdateAction", () => {
141 description: "Desktop app relaunched on 0.0.34.",148 description: "Desktop app relaunched on 0.0.34.",
142 });149 });
143 });150 });
151
152 it("leaves thread continuation off by default", async () => {
153 testState.updateServer.mockResolvedValue(
154 AsyncResult.success({ targetVersion: "0.0.31", method: "boot-service" as const }),
155 );
156 const action = ServerUpdateAction({
157 environmentId: "env-test" as EnvironmentId,
158 serverLabel: "Test server",
159 selfUpdate: "boot-service",
160 threadContinuation: true,
161 targetVersion: "0.0.31",
162 }) as ActionElement;
163
164 action.props.onClick?.();
165 await flushPromises();
166
167 expect(testState.updateServer).toHaveBeenCalledWith({
168 environmentId: "env-test",
169 input: { targetVersion: "0.0.31" },
170 });
171 });
172
173 it("applies the saved thread continuation preference automatically", async () => {
174 testState.updateServer.mockResolvedValue(
175 AsyncResult.success({ targetVersion: "0.0.31", method: "boot-service" as const }),
176 );
177 testState.continueThreadsAfterServerUpdate = true;
178 const action = ServerUpdateAction({
179 environmentId: "env-test" as EnvironmentId,
180 serverLabel: "Test server",
181 selfUpdate: "boot-service",
182 threadContinuation: true,
183 targetVersion: "0.0.31",
184 }) as ActionElement;
185
186 action.props.onClick?.();
187 await flushPromises();
188
189 expect(testState.updateServer).toHaveBeenCalledWith({
190 environmentId: "env-test",
191 input: { targetVersion: "0.0.31", continueRunningThreads: true },
192 });
193 });
144});194});
145195
146describe("ServerUpdateProgress", () => {196describe("ServerUpdateProgress", () => {
apps/web/src/components/ServerUpdateAction.tsx+13-1
...@@ -8,6 +8,7 @@ import type { ComponentProps } from "react";...@@ -8,6 +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 { serverEnvironment } from "~/state/server";12import { serverEnvironment } from "~/state/server";
12import { useAtomCommand } from "~/state/use-atom-command";13import { useAtomCommand } from "~/state/use-atom-command";
13import { manualServerUpdateCommand } from "~/versionSkew";14import { manualServerUpdateCommand } from "~/versionSkew";
...@@ -78,6 +79,7 @@ export function ServerUpdateAction({...@@ -78,6 +79,7 @@ export function ServerUpdateAction({
78 serverLabel,79 serverLabel,
79 selfUpdate,80 selfUpdate,
80 desktopAppUpdate = false,81 desktopAppUpdate = false,
82 threadContinuation = false,
81 targetVersion,83 targetVersion,
82 label = "Update",84 label = "Update",
83 variant = "outline",85 variant = "outline",
...@@ -88,11 +90,16 @@ export function ServerUpdateAction({...@@ -88,11 +90,16 @@ export function ServerUpdateAction({
88 /** The desktop app supervising this server accepts remote update90 /** The desktop app supervising this server accepts remote update
89 requests (capabilities.desktopAppUpdate). */91 requests (capabilities.desktopAppUpdate). */
90 readonly desktopAppUpdate?: boolean;92 readonly desktopAppUpdate?: boolean;
93 /** The server can durably continue running provider turns after updating. */
94 readonly threadContinuation?: boolean;
91 readonly targetVersion: string;95 readonly targetVersion: string;
92 readonly label?: string;96 readonly label?: string;
93 readonly variant?: ComponentProps<typeof Button>["variant"];97 readonly variant?: ComponentProps<typeof Button>["variant"];
94}) {98}) {
95 const isDesktopAppUpdate = selfUpdate === "desktop-managed";99 const isDesktopAppUpdate = selfUpdate === "desktop-managed";
100 const continueThreadsAfterServerUpdate = useClientSettings(
101 (settings) => settings.continueThreadsAfterServerUpdate,
102 );
96 const updateServer = useAtomCommand(serverEnvironment.updateServer, {103 const updateServer = useAtomCommand(serverEnvironment.updateServer, {
97 reportFailure: false,104 reportFailure: false,
98 });105 });
...@@ -137,7 +144,12 @@ export function ServerUpdateAction({...@@ -137,7 +144,12 @@ export function ServerUpdateAction({
137 try {144 try {
138 const result = await updateServer({145 const result = await updateServer({
139 environmentId,146 environmentId,
140 input: { targetVersion },147 input: {
148 targetVersion,
149 ...(threadContinuation && continueThreadsAfterServerUpdate
150 ? { continueRunningThreads: true }
151 : {}),
152 },
141 });153 });
142 if (result._tag === "Failure") {154 if (result._tag === "Failure") {
143 if (isAtomCommandInterrupted(result)) {155 if (isAtomCommandInterrupted(result)) {
apps/web/src/components/settings/ConnectionsSettings.tsx+5
...@@ -105,6 +105,7 @@ import {...@@ -105,6 +105,7 @@ import {
105 resolveServerConfigVersionMismatch,105 resolveServerConfigVersionMismatch,
106 resolveServerSelfUpdateCapability,106 resolveServerSelfUpdateCapability,
107 supportsDesktopAppUpdate,107 supportsDesktopAppUpdate,
108 supportsServerUpdateThreadContinuation,
108} from "~/versionSkew";109} from "~/versionSkew";
109import { hasCloudPublicConfig } from "~/cloud/publicConfig";110import { hasCloudPublicConfig } from "~/cloud/publicConfig";
110import { useCloudLinkController } from "~/cloud/useCloudLinkController";111import { useCloudLinkController } from "~/cloud/useCloudLinkController";
...@@ -1487,6 +1488,7 @@ function SavedBackendListRow({...@@ -1487,6 +1488,7 @@ function SavedBackendListRow({
1487 serverLabel={`${environment.label} server`}1488 serverLabel={`${environment.label} server`}
1488 selfUpdate={resolveServerSelfUpdateCapability(environment.serverConfig)}1489 selfUpdate={resolveServerSelfUpdateCapability(environment.serverConfig)}
1489 desktopAppUpdate={supportsDesktopAppUpdate(environment.serverConfig)}1490 desktopAppUpdate={supportsDesktopAppUpdate(environment.serverConfig)}
1491 threadContinuation={supportsServerUpdateThreadContinuation(environment.serverConfig)}
1490 targetVersion={versionMismatch.clientVersion}1492 targetVersion={versionMismatch.clientVersion}
1491 label={serverUpdateState.status === "failed" ? "Retry" : "Update"}1493 label={serverUpdateState.status === "failed" ? "Retry" : "Update"}
1492 />1494 />
...@@ -3064,6 +3066,9 @@ export function ConnectionsSettings() {...@@ -3064,6 +3066,9 @@ export function ConnectionsSettings() {
3064 }3066 }
3065 selfUpdate={resolveServerSelfUpdateCapability(primaryServerConfig)}3067 selfUpdate={resolveServerSelfUpdateCapability(primaryServerConfig)}
3066 desktopAppUpdate={supportsDesktopAppUpdate(primaryServerConfig)}3068 desktopAppUpdate={supportsDesktopAppUpdate(primaryServerConfig)}
3069 threadContinuation={supportsServerUpdateThreadContinuation(
3070 primaryServerConfig,
3071 )}
3067 targetVersion={primaryVersionMismatch.clientVersion}3072 targetVersion={primaryVersionMismatch.clientVersion}
3068 label={primaryServerUpdateState.status === "failed" ? "Retry" : "Update"}3073 label={primaryServerUpdateState.status === "failed" ? "Retry" : "Update"}
3069 />3074 />
apps/web/src/components/settings/SettingsPanels.tsx+34
...@@ -533,6 +533,10 @@ export function useSettingsRestore(onRestored?: () => void) {...@@ -533,6 +533,10 @@ export function useSettingsRestore(onRestored?: () => void) {
533 DEFAULT_UNIFIED_SETTINGS.enableProviderUpdateChecks533 DEFAULT_UNIFIED_SETTINGS.enableProviderUpdateChecks
534 ? ["Provider update checks"]534 ? ["Provider update checks"]
535 : []),535 : []),
536 ...(settings.continueThreadsAfterServerUpdate !==
537 DEFAULT_UNIFIED_SETTINGS.continueThreadsAfterServerUpdate
538 ? ["Continue threads after server updates"]
539 : []),
536 ...(isBackgroundActivityDirty ? ["Background activity"] : []),540 ...(isBackgroundActivityDirty ? ["Background activity"] : []),
537 ...(settings.defaultThreadEnvMode !== DEFAULT_UNIFIED_SETTINGS.defaultThreadEnvMode541 ...(settings.defaultThreadEnvMode !== DEFAULT_UNIFIED_SETTINGS.defaultThreadEnvMode
538 ? ["New thread mode"]542 ? ["New thread mode"]
...@@ -591,6 +595,7 @@ export function useSettingsRestore(onRestored?: () => void) {...@@ -591,6 +595,7 @@ export function useSettingsRestore(onRestored?: () => void) {
591 settings.glassOpacity,595 settings.glassOpacity,
592 settings.enableLegacyTokenStreaming,596 settings.enableLegacyTokenStreaming,
593 settings.enableProviderUpdateChecks,597 settings.enableProviderUpdateChecks,
598 settings.continueThreadsAfterServerUpdate,
594 settings.sidebarAutoSettleAfterDays,599 settings.sidebarAutoSettleAfterDays,
595 settings.sidebarAutoSettleOnMerge,600 settings.sidebarAutoSettleOnMerge,
596 settings.sidebarProjectGroupingMode,601 settings.sidebarProjectGroupingMode,
...@@ -681,6 +686,7 @@ export function useSettingsRestore(onRestored?: () => void) {...@@ -681,6 +686,7 @@ export function useSettingsRestore(onRestored?: () => void) {
681 sidebarAutoSettleOnMerge: DEFAULT_UNIFIED_SETTINGS.sidebarAutoSettleOnMerge,686 sidebarAutoSettleOnMerge: DEFAULT_UNIFIED_SETTINGS.sidebarAutoSettleOnMerge,
682 enableLegacyTokenStreaming: DEFAULT_UNIFIED_SETTINGS.enableLegacyTokenStreaming,687 enableLegacyTokenStreaming: DEFAULT_UNIFIED_SETTINGS.enableLegacyTokenStreaming,
683 enableProviderUpdateChecks: DEFAULT_UNIFIED_SETTINGS.enableProviderUpdateChecks,688 enableProviderUpdateChecks: DEFAULT_UNIFIED_SETTINGS.enableProviderUpdateChecks,
689 continueThreadsAfterServerUpdate: DEFAULT_UNIFIED_SETTINGS.continueThreadsAfterServerUpdate,
684 backgroundActivity: DEFAULT_UNIFIED_SETTINGS.backgroundActivity,690 backgroundActivity: DEFAULT_UNIFIED_SETTINGS.backgroundActivity,
685 backgroundActivityProfile: DEFAULT_UNIFIED_SETTINGS.backgroundActivityProfile,691 backgroundActivityProfile: DEFAULT_UNIFIED_SETTINGS.backgroundActivityProfile,
686 automaticGitFetchInterval: DEFAULT_UNIFIED_SETTINGS.automaticGitFetchInterval,692 automaticGitFetchInterval: DEFAULT_UNIFIED_SETTINGS.automaticGitFetchInterval,
...@@ -2185,6 +2191,34 @@ export function GeneralSettingsPanel() {...@@ -2185,6 +2191,34 @@ export function GeneralSettingsPanel() {
2185 }2191 }
2186 />2192 />
21872193
2194 <SettingsRow
2195 {...searchableSetting("continue-threads-after-server-update")}
2196 description="Automatically resume active threads after a server update restarts the environment."
2197 resetAction={
2198 settings.continueThreadsAfterServerUpdate !==
2199 DEFAULT_UNIFIED_SETTINGS.continueThreadsAfterServerUpdate ? (
2200 <SettingResetButton
2201 label="continue threads after server updates"
2202 onClick={() =>
2203 updateSettings({
2204 continueThreadsAfterServerUpdate:
2205 DEFAULT_UNIFIED_SETTINGS.continueThreadsAfterServerUpdate,
2206 })
2207 }
2208 />
2209 ) : null
2210 }
2211 control={
2212 <Switch
2213 checked={settings.continueThreadsAfterServerUpdate}
2214 onCheckedChange={(checked) =>
2215 updateSettings({ continueThreadsAfterServerUpdate: Boolean(checked) })
2216 }
2217 aria-label="Continue threads after server updates"
2218 />
2219 }
2220 />
2221
2188 <SettingsRow2222 <SettingsRow
2189 serverScoped2223 serverScoped
2190 id={searchableSetting("background-activity").id}2224 id={searchableSetting("background-activity").id}
apps/web/src/components/settings/settingsSearch.ts+6
...@@ -191,6 +191,12 @@ export const SETTINGS_SEARCH_ITEMS = [...@@ -191,6 +191,12 @@ export const SETTINGS_SEARCH_ITEMS = [
191 to: "/settings/general",191 to: "/settings/general",
192 searchTerms: ["installed cli versions newer available codex claude cursor grok opencode"],192 searchTerms: ["installed cli versions newer available codex claude cursor grok opencode"],
193 },193 },
194 {
195 id: "continue-threads-after-server-update",
196 title: "Continue threads after server updates",
197 to: "/settings/general",
198 searchTerms: ["resume running active work restart desktop update automatically"],
199 },
194 {200 {
195 id: "background-activity",201 id: "background-activity",
196 title: "Background activity",202 title: "Background activity",
apps/web/src/versionSkew.ts+8
...@@ -106,6 +106,14 @@ export function supportsDesktopAppUpdate(...@@ -106,6 +106,14 @@ export function supportsDesktopAppUpdate(
106 return serverConfig?.environment.capabilities.desktopAppUpdate === true;106 return serverConfig?.environment.capabilities.desktopAppUpdate === true;
107}107}
108108
109/** True when the connected server can recover opted-in running turns after
110 its self-update restart. */
111export function supportsServerUpdateThreadContinuation(
112 serverConfig: Pick<ServerConfig, "environment"> | null | undefined,
113): boolean {
114 return serverConfig?.environment.capabilities.serverUpdateThreadContinuation === true;
115}
116
109/** The command to hand users whose server cannot update itself. */117/** The command to hand users whose server cannot update itself. */
110export function manualServerUpdateCommand(targetVersion: string): string {118export function manualServerUpdateCommand(targetVersion: string): string {
111 return `npx t3@${targetVersion}`;119 return `npx t3@${targetVersion}`;
docs/user/updating.md+6-2
...@@ -15,8 +15,12 @@ update the server, and the version difference remains visible in Connections....@@ -15,8 +15,12 @@ update the server, and the version difference remains visible in Connections.
1515
16## Before You Update16## Before You Update
1717
18Let active agent work and terminal commands finish first. Updating restarts the server, so the18Updating restarts the server, so the connection will disappear briefly. **Settings** → **General**
19connection will disappear briefly and work that is still running may be interrupted.19has a **Continue threads after server updates** preference. It is off by default. When enabled, the
20update buttons automatically resume supported provider threads after the replacement server is
21ready. Providers with native promptless continuation use it; other providers receive a short
22instruction to continue where they left off. Terminal commands and other running work may still be
23interrupted during the update.
2024
21The update does not remove saved threads, settings, or project files.25The update does not remove saved threads, settings, or project files.
2226
packages/contracts/src/environment.ts+3
...@@ -93,6 +93,9 @@ export const ExecutionEnvironmentCapabilities = Schema.Struct({...@@ -93,6 +93,9 @@ export const ExecutionEnvironmentCapabilities = Schema.Struct({
93 /** Server can stream self-update progress before acknowledging the93 /** Server can stream self-update progress before acknowledging the
94 restart. Clients fall back to server.updateServer when absent. */94 restart. Clients fall back to server.updateServer when absent. */
95 serverSelfUpdateProgress: Schema.optionalKey(Schema.Boolean),95 serverSelfUpdateProgress: Schema.optionalKey(Schema.Boolean),
96 /** Server can durably mark running provider turns before a self-update and
97 continue them after the replacement process starts. */
98 serverUpdateThreadContinuation: Schema.optionalKey(Schema.Boolean),
96 /** Agent-activity publishes (push notifications and Live Activities)99 /** Agent-activity publishes (push notifications and Live Activities)
97 currently leave this environment: the publish opt-in is enabled and the100 currently leave this environment: the publish opt-in is enabled and the
98 relay link credentials exist. Clients skip seeding a Live Activity when101 relay link credentials exist. Clients skip seeding a Live Activity when
packages/contracts/src/provider.ts+3
...@@ -67,6 +67,9 @@ export type ProviderSessionStartInput = typeof ProviderSessionStartInput.Type;...@@ -67,6 +67,9 @@ export type ProviderSessionStartInput = typeof ProviderSessionStartInput.Type;
6767
68export const ProviderSendTurnInput = Schema.Struct({68export const ProviderSendTurnInput = Schema.Struct({
69 threadId: ThreadId,69 threadId: ThreadId,
70 /** Internal recovery signal. Allows an empty turn only for adapters that
71 explicitly support promptless continuation. */
72 continuation: Schema.optional(Schema.Boolean),
70 input: Schema.optional(73 input: Schema.optional(
71 TrimmedNonEmptyString.check(Schema.isMaxLength(PROVIDER_SEND_TURN_MAX_INPUT_CHARS)),74 TrimmedNonEmptyString.check(Schema.isMaxLength(PROVIDER_SEND_TURN_MAX_INPUT_CHARS)),
72 ),75 ),
packages/contracts/src/server.ts+4
...@@ -754,6 +754,10 @@ export const ServerSelfUpdateInput = Schema.Struct({...@@ -754,6 +754,10 @@ export const ServerSelfUpdateInput = Schema.Struct({
754 /** Exact npm version of the `t3` package to install (never a dist-tag, so754 /** Exact npm version of the `t3` package to install (never a dist-tag, so
755 the server and the acknowledging client agree on what was requested). */755 the server and the acknowledging client agree on what was requested). */
756 targetVersion: TrimmedNonEmptyString,756 targetVersion: TrimmedNonEmptyString,
757 /** Opt-in recovery for provider turns that are running when the server
758 hands off to its replacement. Missing and false keep restart behavior
759 conservative under version skew. */
760 continueRunningThreads: Schema.optionalKey(Schema.Boolean),
757});761});
758export type ServerSelfUpdateInput = typeof ServerSelfUpdateInput.Type;762export type ServerSelfUpdateInput = typeof ServerSelfUpdateInput.Type;
759763
packages/contracts/src/settings.ts+4
...@@ -207,6 +207,9 @@ export const ClientSettingsSchema = Schema.Struct({...@@ -207,6 +207,9 @@ export const ClientSettingsSchema = Schema.Struct({
207 confirmThreadArchive: Schema.Boolean.pipe(Schema.withDecodingDefault(Effect.succeed(false))),207 confirmThreadArchive: Schema.Boolean.pipe(Schema.withDecodingDefault(Effect.succeed(false))),
208 confirmThreadDelete: Schema.Boolean.pipe(Schema.withDecodingDefault(Effect.succeed(true))),208 confirmThreadDelete: Schema.Boolean.pipe(Schema.withDecodingDefault(Effect.succeed(true))),
209 confirmThreadUnpin: Schema.Boolean.pipe(Schema.withDecodingDefault(Effect.succeed(false))),209 confirmThreadUnpin: Schema.Boolean.pipe(Schema.withDecodingDefault(Effect.succeed(false))),
210 continueThreadsAfterServerUpdate: Schema.Boolean.pipe(
211 Schema.withDecodingDefault(Effect.succeed(false)),
212 ),
210 dismissedProviderUpdateNotificationKeys: Schema.Array(TrimmedNonEmptyString).pipe(213 dismissedProviderUpdateNotificationKeys: Schema.Array(TrimmedNonEmptyString).pipe(
211 Schema.withDecodingDefault(Effect.succeed([])),214 Schema.withDecodingDefault(Effect.succeed([])),
212 ),215 ),
...@@ -959,6 +962,7 @@ export const ClientSettingsPatch = Schema.Struct({...@@ -959,6 +962,7 @@ export const ClientSettingsPatch = Schema.Struct({
959 confirmThreadArchive: Schema.optionalKey(Schema.Boolean),962 confirmThreadArchive: Schema.optionalKey(Schema.Boolean),
960 confirmThreadDelete: Schema.optionalKey(Schema.Boolean),963 confirmThreadDelete: Schema.optionalKey(Schema.Boolean),
961 confirmThreadUnpin: Schema.optionalKey(Schema.Boolean),964 confirmThreadUnpin: Schema.optionalKey(Schema.Boolean),
965 continueThreadsAfterServerUpdate: Schema.optionalKey(Schema.Boolean),
962 diffIgnoreWhitespace: Schema.optionalKey(Schema.Boolean),966 diffIgnoreWhitespace: Schema.optionalKey(Schema.Boolean),
963 environmentIdentificationMode: Schema.optionalKey(EnvironmentIdentificationMode),967 environmentIdentificationMode: Schema.optionalKey(EnvironmentIdentificationMode),
964 glassOpacity: Schema.optionalKey(GlassOpacity),968 glassOpacity: Schema.optionalKey(GlassOpacity),