feat(updates): continue active threads across server restarts

apps/desktop/src/settings/DesktopClientSettings.test.ts

@@ -23,6 +23,7 @@ const clientSettings: ClientSettings = {
   confirmThreadArchive: true,
   confirmThreadDelete: false,
   confirmThreadUnpin: false,
+  continueThreadsAfterServerUpdate: true,
   dismissedProviderUpdateNotificationKeys: [],
   diffIgnoreWhitespace: true,
   environmentIdentificationMode: "artwork",

apps/server/src/cloud/selfUpdate.test.ts

@@ -1,5 +1,6 @@
 import * as NodeServices from "@effect/platform-node/NodeServices";
 import { expect, it } from "@effect/vitest";
+import { ServerSelfUpdateError, ThreadId } from "@t3tools/contracts";
 import { HostProcessExecutablePath } from "@t3tools/shared/hostProcess";
 import * as Deferred from "effect/Deferred";
 import * as Effect from "effect/Effect";
@@ -104,6 +105,75 @@ const makeHarness = Effect.fn("test.make_self_update_harness")(function* (
 });
 
 it.layer(NodeServices.layer)("server self update", (it) => {
+  it.effect("marks running threads at the boot-service handoff", () =>
+    Effect.gen(function* () {
+      const events: string[] = [];
+      const selfUpdate = ServerSelfUpdate.withRunningThreadContinuation({
+        mode: "web",
+        selfUpdate: {
+          update: (_input, reportProgress = () => Effect.void) =>
+            reportProgress("downloading").pipe(
+              Effect.andThen(reportProgress("installing")),
+              Effect.as({
+                targetVersion: "1.1.0",
+                method: "boot-service" as const,
+                updateId: "update-id",
+              }),
+            ),
+          commitDesktopUpdate: () => Effect.never,
+        },
+        prepare: Effect.sync(() => {
+          events.push("prepare");
+          return [ThreadId.make("thread-running")];
+        }),
+        clear: () => Effect.sync(() => void events.push("clear")),
+      });
+
+      yield* selfUpdate.update({ targetVersion: "1.1.0", continueRunningThreads: true }, (stage) =>
+        Effect.sync(() => void events.push(stage)),
+      );
+
+      expect(events).toEqual(["downloading", "prepare", "installing"]);
+    }),
+  );
+
+  it.effect("marks desktop threads only when the prepared update commits", () =>
+    Effect.gen(function* () {
+      const threadId = ThreadId.make("thread-running-desktop");
+      const events: string[] = [];
+      const commitError = new ServerSelfUpdateError({ reason: "install failed" });
+      const selfUpdate = ServerSelfUpdate.withRunningThreadContinuation({
+        mode: "desktop",
+        selfUpdate: {
+          update: (_input, reportProgress = () => Effect.void) =>
+            reportProgress("installing").pipe(
+              Effect.as({
+                targetVersion: "1.2.0",
+                method: "desktop-app" as const,
+                desktopUpdateToken: "desktop-token",
+              }),
+            ),
+          commitDesktopUpdate: () =>
+            Effect.sync(() => events.push("commit")).pipe(Effect.andThen(Effect.fail(commitError))),
+        },
+        prepare: Effect.sync(() => {
+          events.push("prepare");
+          return [threadId];
+        }),
+        clear: (threadIds) => Effect.sync(() => void events.push(`clear:${threadIds.join(",")}`)),
+      });
+
+      yield* selfUpdate.update({ targetVersion: "1.2.0", continueRunningThreads: true }, (stage) =>
+        Effect.sync(() => void events.push(stage)),
+      );
+      expect(events).toEqual(["installing"]);
+      expect(yield* selfUpdate.commitDesktopUpdate("desktop-token").pipe(Effect.flip)).toBe(
+        commitError,
+      );
+      expect(events).toEqual(["installing", "prepare", "commit", `clear:${threadId}`]);
+    }),
+  );
+
   it.effect("stages and preflights before asking the launcher for an update ID", () =>
     Effect.gen(function* () {
       const { selfUpdate, order } = yield* makeHarness();

apps/server/src/cloud/selfUpdate.ts

@@ -4,6 +4,7 @@ import {
   type ServerSelfUpdateInput,
   type ServerSelfUpdateProgressStage,
   type ServerSelfUpdateResult,
+  type ThreadId,
 } from "@t3tools/contracts";
 import { HostProcessExecutablePath } from "@t3tools/shared/hostProcess";
 import * as Context from "effect/Context";
@@ -41,14 +42,77 @@ export class ServerSelfUpdate extends Context.Service<
   {
     readonly update: (
       input: ServerSelfUpdateInput,
-      reportProgress?: (stage: ServerSelfUpdateProgressStage) => Effect.Effect<void>,
+      reportProgress?: (
+        stage: ServerSelfUpdateProgressStage,
+      ) => Effect.Effect<void, ServerSelfUpdateError>,
     ) => Effect.Effect<ServerSelfUpdateResult, ServerSelfUpdateError>;
     readonly commitDesktopUpdate: (
       requestId: string,
     ) => Effect.Effect<never, ServerSelfUpdateError>;
   }
 >()("t3/cloud/selfUpdate/ServerSelfUpdate") {}
 
+export function withRunningThreadContinuation(input: {
+  readonly mode: ServerConfig.RuntimeMode;
+  readonly selfUpdate: ServerSelfUpdate["Service"];
+  readonly prepare: Effect.Effect<ReadonlyArray<ThreadId>, ServerSelfUpdateError>;
+  readonly clear: (threadIds: ReadonlyArray<ThreadId>) => Effect.Effect<void>;
+}): ServerSelfUpdate["Service"] {
+  const desktopContinuationTokens = new Set<string>();
+
+  const update: ServerSelfUpdate["Service"]["update"] = (
+    request,
+    reportProgress = () => Effect.void,
+  ) => {
+    let prepared = false;
+    let continuationThreadIds: ReadonlyArray<ThreadId> = [];
+    return input.selfUpdate
+      .update(request, (stage) =>
+        (request.continueRunningThreads === true &&
+        input.mode !== "desktop" &&
+        stage === "installing" &&
+        !prepared
+          ? input.prepare.pipe(
+              Effect.tap((threadIds) =>
+                Effect.sync(() => {
+                  prepared = true;
+                  continuationThreadIds = threadIds;
+                }),
+              ),
+              Effect.asVoid,
+            )
+          : Effect.void
+        ).pipe(Effect.andThen(reportProgress(stage))),
+      )
+      .pipe(
+        Effect.tap((result) =>
+          Effect.sync(() => {
+            if (
+              result.method === "desktop-app" &&
+              result.desktopUpdateToken !== undefined &&
+              request.continueRunningThreads === true
+            ) {
+              desktopContinuationTokens.add(result.desktopUpdateToken);
+            }
+          }),
+        ),
+        Effect.onError(() => input.clear(continuationThreadIds)),
+      );
+  };
+
+  return ServerSelfUpdate.of({
+    update,
+    commitDesktopUpdate: (requestId) =>
+      Effect.gen(function* () {
+        const shouldContinue = desktopContinuationTokens.delete(requestId);
+        const continuationThreadIds = shouldContinue ? yield* input.prepare : [];
+        return yield* input.selfUpdate
+          .commitDesktopUpdate(requestId)
+          .pipe(Effect.onError(() => input.clear(continuationThreadIds)));
+      }),
+  });
+}
+
 export const make = Effect.fn("cloud.server_self_update.make")(function* () {
   const serverConfig = yield* ServerConfig.ServerConfig;
   const desktopAppUpdate = yield* DesktopAppUpdate.DesktopAppUpdate;

apps/server/src/desktopUpdate/DesktopAppUpdate.ts

@@ -49,7 +49,9 @@ export class DesktopAppUpdate extends Context.Service<
     /** Checks and downloads through the desktop app, then returns a token
         while this server is still connected. `commit` starts installation. */
     readonly run: (
-      reportProgress: (stage: ServerSelfUpdateProgressStage) => Effect.Effect<void>,
+      reportProgress: (
+        stage: ServerSelfUpdateProgressStage,
+      ) => Effect.Effect<void, ServerSelfUpdateError>,
     ) => Effect.Effect<ServerSelfUpdateResult, ServerSelfUpdateError>;
     /** Starts the prepared install. Success stops this server, so this effect
         returns only when installation fails or times out. */
@@ -72,11 +74,15 @@ export const make = Effect.fn("desktopUpdate.desktopAppUpdate.make")(function* (
   const consumeReports = (
     requestId: string,
     changes: Stream.Stream<DesktopUpdateStatusReport>,
-    reportProgress: (stage: ServerSelfUpdateProgressStage) => Effect.Effect<void>,
+    reportProgress: (
+      stage: ServerSelfUpdateProgressStage,
+    ) => Effect.Effect<void, ServerSelfUpdateError>,
   ) =>
     Effect.gen(function* () {
       const lastStage = yield* Ref.make<ServerSelfUpdateProgressStage | null>(null);
-      const emitStage = (stage: ServerSelfUpdateProgressStage | null): Effect.Effect<void> =>
+      const emitStage = (
+        stage: ServerSelfUpdateProgressStage | null,
+      ): Effect.Effect<void, ServerSelfUpdateError> =>
         stage === null
           ? Effect.void
           : Ref.get(lastStage).pipe(
@@ -90,7 +96,9 @@ export const make = Effect.fn("desktopUpdate.desktopAppUpdate.make")(function* (
       const terminal = yield* changes.pipe(
         Stream.filter((report) => report.requestId === requestId),
         Stream.mapEffect(
-          (report): Effect.Effect<Option.Option<DesktopUpdateStatusReport>> =>
+          (
+            report,
+          ): Effect.Effect<Option.Option<DesktopUpdateStatusReport>, ServerSelfUpdateError> =>
             report.outcome === undefined
               ? emitStage(desktopUpdateProgressStage(report.state)).pipe(
                   Effect.as(Option.none<DesktopUpdateStatusReport>()),

apps/server/src/environment/ServerEnvironment.test.ts

@@ -244,11 +244,13 @@ it.layer(NodeServices.layer)("ServerEnvironmentLive", (it) => {
       expect(withFd.capabilities.serverSelfUpdate).toBe("desktop-managed");
       expect(withFd.capabilities.desktopAppUpdate).toBe(true);
       expect(withFd.capabilities.serverSelfUpdateProgress).toBe(true);
+      expect(withFd.capabilities.serverUpdateThreadContinuation).toBe(true);
 
       const withoutFd = yield* describeWith({ mode: "desktop" });
       expect(withoutFd.capabilities.serverSelfUpdate).toBe("desktop-managed");
       expect(withoutFd.capabilities.desktopAppUpdate).toBeUndefined();
       expect(withoutFd.capabilities.serverSelfUpdateProgress).toBeUndefined();
+      expect(withoutFd.capabilities.serverUpdateThreadContinuation).toBeUndefined();
 
       const web = yield* describeWith({ mode: "web", desktopTelemetryControlFd: 5 });
       expect(web.capabilities.desktopAppUpdate).toBeUndefined();

apps/server/src/environment/ServerEnvironment.ts

@@ -224,7 +224,10 @@ export const make = Effect.gen(function* () {
       threadPullRequestLinking: true,
       ...(serverSelfUpdate === null ? {} : { serverSelfUpdate }),
       ...(serverSelfUpdate === "boot-service" || desktopAppUpdate
-        ? { serverSelfUpdateProgress: true }
+        ? {
+            serverSelfUpdateProgress: true,
+            serverUpdateThreadContinuation: true,
+          }
         : {}),
       ...(desktopAppUpdate ? { desktopAppUpdate: true } : {}),
     },

apps/server/src/provider/Layers/CodexAdapter.ts

@@ -2010,6 +2010,7 @@ export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* (
     provider: PROVIDER,
     capabilities: {
       sessionModelSwitch: "in-session",
+      promptlessTurnContinuation: true,
     },
     startSession,
     sendTurn,

apps/server/src/provider/Layers/ProviderService.test.ts

@@ -246,6 +246,7 @@ function makeFakeCodexAdapter(provider: ProviderDriverKind = CODEX_DRIVER) {
     provider,
     capabilities: {
       sessionModelSwitch: "in-session",
+      ...(provider === CODEX_DRIVER ? { promptlessTurnContinuation: true } : {}),
     },
     startSession,
     sendTurn,
@@ -960,6 +961,48 @@ it.effect(
 );
 
 routing.layer("ProviderServiceLive routing", (it) => {
+  it.effect("allows promptless continuation only for capable providers", () =>
+    Effect.gen(function* () {
+      const provider = yield* ProviderService.ProviderService;
+      const codexThreadId = asThreadId("thread-promptless-continuation");
+      yield* provider.startSession(codexThreadId, {
+        provider: CODEX_DRIVER,
+        providerInstanceId: codexInstanceId,
+        threadId: codexThreadId,
+        runtimeMode: "full-access",
+      });
+
+      yield* provider.sendTurn({ threadId: codexThreadId, continuation: true });
+      assert.deepEqual(routing.codex.sendTurn.mock.calls.at(-1)?.[0], {
+        threadId: codexThreadId,
+        continuation: true,
+      });
+
+      const claudeThreadId = asThreadId("thread-promptless-continuation-unsupported");
+      yield* provider.startSession(claudeThreadId, {
+        provider: CLAUDE_AGENT_DRIVER,
+        providerInstanceId: claudeAgentInstanceId,
+        threadId: claudeThreadId,
+        runtimeMode: "full-access",
+      });
+      const failure = yield* Effect.flip(
+        provider.sendTurn({ threadId: claudeThreadId, continuation: true }),
+      );
+      assert.instanceOf(failure, ProviderValidationError);
+      assert.include(failure.issue, "requires an explicit continuation prompt");
+      assert.equal(routing.claude.sendTurn.mock.calls.length, 0);
+
+      yield* provider.stopSession({ threadId: codexThreadId });
+      yield* provider.stopSession({ threadId: claudeThreadId });
+      routing.codex.startSession.mockClear();
+      routing.codex.sendTurn.mockClear();
+      routing.codex.stopSession.mockClear();
+      routing.claude.startSession.mockClear();
+      routing.claude.sendTurn.mockClear();
+      routing.claude.stopSession.mockClear();
+    }),
+  );
+
   it.effect("routes provider operations and rollback conversation", () =>
     Effect.gen(function* () {
       const provider = yield* ProviderService.ProviderService;

apps/server/src/provider/Layers/ProviderService.ts

@@ -723,7 +723,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* (
     });
 
     const attachments = parsed.attachments ?? [];
-    if (!parsed.input && attachments.length === 0) {
+    if (!parsed.input && attachments.length === 0 && parsed.continuation !== true) {
       return yield* toValidationError(
         "ProviderService.sendTurn",
         "Either input text or at least one attachment is required",
@@ -782,6 +782,17 @@ const makeProviderService = Effect.fn("makeProviderService")(function* (
         operation: "ProviderService.sendTurn",
         allowRecovery: true,
       });
+      if (
+        input.continuation === true &&
+        !input.input &&
+        attachments.length === 0 &&
+        routed.adapter.capabilities.promptlessTurnContinuation !== true
+      ) {
+        return yield* toValidationError(
+          "ProviderService.sendTurn",
+          `Provider '${routed.adapter.provider}' requires an explicit continuation prompt`,
+        );
+      }
       metricProvider = routed.adapter.provider;
       metricModel = input.modelSelection?.model;
       yield* Effect.annotateCurrentSpan({

apps/server/src/provider/Services/ProviderAdapter.ts

@@ -32,6 +32,9 @@ export interface ProviderAdapterCapabilities {
    * Declares whether changing the model on an existing session is supported.
    */
   readonly sessionModelSwitch: ProviderSessionModelSwitchMode;
+  /** Starts a resumed turn with no synthetic user prompt. Omitted means the
+      adapter needs an explicit continuation instruction. */
+  readonly promptlessTurnContinuation?: boolean;
 }
 
 export interface ProviderThreadTurnSnapshot {

apps/server/src/server.test.ts

@@ -936,6 +936,8 @@ const buildAppUnderTest = (options?: {
         Layer.mock(ServerRuntimeStartup.ServerRuntimeStartup)({
           awaitCommandReady: Effect.void,
           markHttpListening: Effect.void,
+          markRunningProviderSessionsForContinuation: Effect.succeed([]),
+          clearProviderSessionContinuationMarkers: () => Effect.void,
           enqueueCommand: (effect) => effect,
           ...options?.layers?.serverRuntimeStartup,
         }),