feat(desktop): restart to update when threads finish

apps/desktop/src/ipc/methods/updates.ts

@@ -43,11 +43,13 @@ export const downloadUpdate = DesktopIpc.makeIpcMethod({
 
 export const installUpdate = DesktopIpc.makeIpcMethod({
   channel: IpcChannels.UPDATE_INSTALL_CHANNEL,
-  payload: Schema.Void,
+  payload: Schema.Union([Schema.Void, Schema.Struct({ expectedVersion: Schema.String })]),
   result: DesktopUpdateActionResultSchema,
-  handler: Effect.fn("desktop.ipc.updates.install")(function* () {
+  handler: Effect.fn("desktop.ipc.updates.install")(function* (input) {
     const updates = yield* DesktopUpdates.DesktopUpdates;
-    return yield* updates.install;
+    return yield* input === undefined
+      ? updates.install
+      : updates.installPrepared(input.expectedVersion, { waitForCheck: false });
   }),
 });
 

apps/desktop/src/preload.ts

@@ -230,7 +230,7 @@ contextBridge.exposeInMainWorld("desktopBridge", {
     ipcRenderer.invoke(IpcChannels.UPDATE_SET_CHANNEL_CHANNEL, channel),
   checkForUpdate: () => ipcRenderer.invoke(IpcChannels.UPDATE_CHECK_CHANNEL),
   downloadUpdate: () => ipcRenderer.invoke(IpcChannels.UPDATE_DOWNLOAD_CHANNEL),
-  installUpdate: () => ipcRenderer.invoke(IpcChannels.UPDATE_INSTALL_CHANNEL),
+  installUpdate: (input) => ipcRenderer.invoke(IpcChannels.UPDATE_INSTALL_CHANNEL, input),
   onUpdateState: (listener) => {
     const wrappedListener = (_event: Electron.IpcRendererEvent, state: unknown) => {
       if (typeof state !== "object" || state === null) return;

apps/desktop/src/updates/DesktopUpdates.test.ts

@@ -656,6 +656,34 @@ describe("DesktopUpdates", () => {
     ).pipe(Effect.provide(Layer.merge(TestClock.layer(), harness.layer)));
   });
 
+  it.effect("refuses an automatic install immediately while a check is active", () =>
+    Effect.gen(function* () {
+      const checkStarted = yield* Deferred.make<void>();
+      const releaseCheck = yield* Deferred.make<void>();
+      const harness = makeHarness({
+        checkForUpdates: Deferred.succeed(checkStarted, undefined).pipe(
+          Effect.andThen(Deferred.await(releaseCheck)),
+        ),
+      });
+      yield* Effect.scoped(
+        Effect.gen(function* () {
+          const updates = yield* DesktopUpdates.DesktopUpdates;
+          yield* updates.configure;
+          harness.emit("update-downloaded", { version: "1.2.4" });
+          yield* flushCallbacks;
+          const checking = yield* updates.check("test").pipe(Effect.forkChild);
+          yield* Deferred.await(checkStarted);
+          const result = yield* updates.installPrepared("1.2.4", { waitForCheck: false });
+          assert.isFalse(result.accepted);
+          assert.equal(result.state.status, "checking");
+          assert.equal(harness.quitAndInstalls(), 0);
+          yield* Deferred.succeed(releaseCheck, undefined);
+          yield* Fiber.join(checking);
+        }),
+      ).pipe(Effect.provide(Layer.merge(TestClock.layer(), harness.layer)));
+    }),
+  );
+
   it.effect("rejects a prepared install when the downloaded version changed", () => {
     const harness = makeHarness();
 

apps/desktop/src/updates/DesktopUpdates.ts

@@ -185,6 +185,7 @@ export class DesktopUpdates extends Context.Service<
     readonly install: Effect.Effect<DesktopUpdateActionResult>;
     readonly installPrepared: (
       expectedVersion: string,
+      options?: { readonly waitForCheck: boolean },
     ) => Effect.Effect<DesktopPreparedUpdateInstallResult>;
   }
 >()("@t3tools/desktop/updates/DesktopUpdates") {}
@@ -535,7 +536,7 @@ export const make = Effect.gen(function* () {
     );
   });
 
-  const installDownloadedUpdate = (expectedVersion?: string) =>
+  const installDownloadedUpdate = (expectedVersion?: string, waitForCheck = true) =>
     Effect.scoped(
       Effect.gen(function* () {
         const actionCompletions = yield* PubSub.subscribe(finishedUpdateActions);
@@ -556,7 +557,9 @@ export const make = Effect.gen(function* () {
                 return "refused" as const;
               }
               if (Option.isSome(activeAction)) {
-                return activeAction.value === "check" && expectedVersion !== undefined
+                return waitForCheck &&
+                  activeAction.value === "check" &&
+                  expectedVersion !== undefined
                   ? ("wait-for-check" as const)
                   : ("refused" as const);
               }
@@ -638,7 +641,7 @@ export const make = Effect.gen(function* () {
       }),
     ).pipe(Effect.withSpan("desktop.updates.installDownloadedUpdate"));
 
-  const installWithExpectedVersion = (expectedVersion?: string) =>
+  const installWithExpectedVersion = (expectedVersion?: string, waitForCheck = true) =>
     Effect.gen(function* () {
       if (yield* Ref.get(desktopState.quitting)) {
         return {
@@ -648,7 +651,7 @@ export const make = Effect.gen(function* () {
           state: yield* Ref.get(updateStateRef),
         };
       }
-      const result = yield* installDownloadedUpdate(expectedVersion);
+      const result = yield* installDownloadedUpdate(expectedVersion, waitForCheck);
       return {
         accepted: result.accepted,
         completed: result.completed,
@@ -989,7 +992,8 @@ export const make = Effect.gen(function* () {
     install: installWithExpectedVersion().pipe(
       Effect.map(({ accepted, completed, state }) => ({ accepted, completed, state })),
     ),
-    installPrepared: (expectedVersion) => installWithExpectedVersion(expectedVersion),
+    installPrepared: (expectedVersion, options) =>
+      installWithExpectedVersion(expectedVersion, options?.waitForCheck ?? true),
   });
 });
 

apps/server/src/auth/RpcAuthorization.ts

@@ -28,6 +28,7 @@ export const RPC_REQUIRED_SCOPES = {
   [ORCHESTRATION_WS_METHODS.searchThreads]: AuthOrchestrationReadScope,
   [ORCHESTRATION_WS_METHODS.subscribeShell]: AuthOrchestrationReadScope,
   [ORCHESTRATION_WS_METHODS.getArchivedShellSnapshot]: AuthOrchestrationReadScope,
+  [ORCHESTRATION_WS_METHODS.awaitRestartReady]: AuthOrchestrationReadScope,
   [ORCHESTRATION_WS_METHODS.subscribeThread]: AuthOrchestrationReadScope,
   [WS_METHODS.serverProbe]: AuthOrchestrationReadScope,
   [WS_METHODS.serverGetConfig]: AuthOrchestrationReadScope,

apps/server/src/orchestration/Services/ProviderCommandReactor.ts

@@ -28,7 +28,7 @@ export interface ProviderCommandReactorShape {
 
   /**
    * Resolves when the internal processing queue is empty and idle.
-   * Intended for test use to replace timing-sensitive sleeps.
+   * Used by restart readiness and tests to wait for processing to finish.
    */
   readonly drain: Effect.Effect<void>;
 }

apps/server/src/orchestration/Services/ProviderRuntimeIngestion.ts

@@ -27,7 +27,7 @@ export interface ProviderRuntimeIngestionShape {
 
   /**
    * Resolves when the internal processing queue is empty and idle.
-   * Intended for test use to replace timing-sensitive sleeps.
+   * Used by restart readiness and tests to wait for processing to finish.
    */
   readonly drain: Effect.Effect<void>;
 }

apps/server/src/orchestration/restartReadiness.test.ts

@@ -0,0 +1,93 @@
+import { describe, it, expect } from "@effect/vitest";
+import {
+  ProjectId,
+  ProviderInstanceId,
+  ThreadId,
+  type OrchestrationThreadShell,
+} from "@t3tools/contracts";
+import * as Deferred from "effect/Deferred";
+import * as Effect from "effect/Effect";
+import * as Fiber from "effect/Fiber";
+import { awaitRestartReady } from "./restartReadiness.ts";
+
+const ready: OrchestrationThreadShell = {
+  id: ThreadId.make("thread"),
+  projectId: ProjectId.make("project"),
+  title: "Thread",
+  modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5" },
+  runtimeMode: "full-access",
+  interactionMode: "default",
+  pullRequests: [],
+  branch: null,
+  worktreePath: null,
+  latestTurn: null,
+  createdAt: "2026-09-14T00:00:00Z",
+  updatedAt: "2026-09-14T00:00:00Z",
+  archivedAt: null,
+  settledOverride: null,
+  settledAt: null,
+  session: null,
+  latestUserMessageAt: null,
+  hasPendingApprovals: false,
+  hasPendingUserInput: false,
+  hasActionableProposedPlan: false,
+};
+
+describe("restart readiness", () => {
+  it.effect("waits for message finalization and checkpoint capture despite an idle summary", () =>
+    Effect.gen(function* () {
+      const ingestionEntered = yield* Deferred.make<void>();
+      const messagesPersisted = yield* Deferred.make<void>();
+      const checkpointEntered = yield* Deferred.make<void>();
+      const checkpointCaptured = yield* Deferred.make<void>();
+      const returned = yield* Deferred.make<void>();
+      const check = yield* awaitRestartReady({
+        readThreads: Effect.succeed([ready]),
+        drainCommands: Effect.void,
+        drainIngestion: Deferred.succeed(ingestionEntered, undefined).pipe(
+          Effect.andThen(Deferred.await(messagesPersisted)),
+        ),
+        drainCheckpoints: Deferred.succeed(checkpointEntered, undefined).pipe(
+          Effect.andThen(Deferred.await(checkpointCaptured)),
+        ),
+      }).pipe(
+        Effect.tap(() => Deferred.succeed(returned, undefined)),
+        Effect.forkChild,
+      );
+      yield* Deferred.await(ingestionEntered);
+      expect(yield* Deferred.isDone(returned)).toBe(false);
+      expect(yield* Deferred.isDone(checkpointEntered)).toBe(false);
+      yield* Deferred.succeed(messagesPersisted, undefined);
+      yield* Deferred.await(checkpointEntered);
+      expect(yield* Deferred.isDone(returned)).toBe(false);
+      yield* Deferred.succeed(checkpointCaptured, undefined);
+      expect(yield* Fiber.join(check)).toBe(true);
+    }),
+  );
+  it.effect("rechecks authoritative state after draining", () =>
+    Effect.gen(function* () {
+      let reads = 0;
+      const result = yield* awaitRestartReady({
+        readThreads: Effect.sync(() => [
+          reads++ === 0 ? ready : { ...ready, backgroundLiveness: "working" as const },
+        ]),
+        drainCommands: Effect.void,
+        drainIngestion: Effect.void,
+        drainCheckpoints: Effect.void,
+      });
+      expect(result).toBe(false);
+      expect(reads).toBe(2);
+    }),
+  );
+  it.effect("does not wait on workers while a thread still needs input", () =>
+    Effect.gen(function* () {
+      const result = yield* awaitRestartReady({
+        readThreads: Effect.succeed([{ ...ready, hasPendingUserInput: true }]),
+        drainCommands: Effect.die("must not drain"),
+        drainIngestion: Effect.die("must not drain"),
+        drainCheckpoints: Effect.die("must not drain"),
+      });
+      expect(result).toBe(false);
+    }),
+  );
+});

apps/server/src/orchestration/restartReadiness.ts

@@ -0,0 +1,16 @@
+import { isThreadBlockingRestart, type OrchestrationThreadShell } from "@t3tools/contracts";
+import * as Effect from "effect/Effect";
+
+/** Waits for persisted turn finalization, then checks authoritative thread state again. */
+export const awaitRestartReady = Effect.fn("orchestration.awaitRestartReady")(function* <E>(input: {
+  readonly readThreads: Effect.Effect<ReadonlyArray<OrchestrationThreadShell>, E>;
+  readonly drainCommands: Effect.Effect<void>;
+  readonly drainIngestion: Effect.Effect<void>;
+  readonly drainCheckpoints: Effect.Effect<void>;
+}) {
+  if ((yield* input.readThreads).some(isThreadBlockingRestart)) return false;
+  yield* input.drainCommands;
+  yield* input.drainIngestion;
+  yield* input.drainCheckpoints;
+  return !(yield* input.readThreads).some(isThreadBlockingRestart);
+});

apps/server/src/server.test.ts

@@ -1,3 +1,6 @@
+import { ProviderCommandReactor } from "./orchestration/Services/ProviderCommandReactor.ts";
+import { ProviderRuntimeIngestionService } from "./orchestration/Services/ProviderRuntimeIngestion.ts";
+import { CheckpointReactor } from "./orchestration/Services/CheckpointReactor.ts";
 import * as NodeHttpServer from "@effect/platform-node/NodeHttpServer";
 import * as NodeSocket from "@effect/platform-node/NodeSocket";
 import * as NodeServices from "@effect/platform-node/NodeServices";
@@ -972,6 +975,12 @@ const buildAppUnderTest = (options?: {
             latestSequence: Effect.succeed(0),
             ...options?.layers?.orchestrationEngine,
           }),
+          Layer.mock(ProviderCommandReactor)({ start: () => Effect.void, drain: Effect.void }),
+          Layer.mock(ProviderRuntimeIngestionService)({
+            start: () => Effect.void,
+            drain: Effect.void,
+          }),
+          Layer.mock(CheckpointReactor)({ start: () => Effect.void, drain: Effect.void }),
           Layer.mock(ThreadDeletionReactor)({
             start: () => Effect.void,
             drainThrough: () => Effect.void,
@@ -8139,6 +8148,17 @@ it.layer(NodeServices.layer)("server router seam", (it) => {
       }).pipe(Effect.provide(NodeHttpServer.layerTest)),
   );
 
+  it.effect("routes restart readiness through the authenticated websocket", () =>
+    Effect.gen(function* () {
+      yield* buildAppUnderTest();
+      const wsUrl = yield* getWsServerUrl("/ws");
+      const ready = yield* Effect.scoped(
+        withWsRpcClient(wsUrl, (client) => client[ORCHESTRATION_WS_METHODS.awaitRestartReady]({})),
+      );
+      assert.isTrue(ready);
+    }).pipe(Effect.provide(NodeHttpServer.layerTest)),
+  );
+
   it.effect("routes websocket rpc orchestration methods", () =>
     Effect.gen(function* () {
       const now = "2026-01-01T00:00:00.000Z";

apps/server/src/ws.ts

@@ -1,3 +1,7 @@
+import { awaitRestartReady } from "./orchestration/restartReadiness.ts";
+import { ProviderCommandReactor } from "./orchestration/Services/ProviderCommandReactor.ts";
+import { ProviderRuntimeIngestionService } from "./orchestration/Services/ProviderRuntimeIngestion.ts";
+import { CheckpointReactor } from "./orchestration/Services/CheckpointReactor.ts";
 import {
   sameUsageLimitCommandCoverage,
   withUsageLimitsCommands,
@@ -1897,6 +1901,35 @@ const makeWsRpcLayer = (
             }),
             { "rpc.aggregate": "orchestration" },
           ),
+        [ORCHESTRATION_WS_METHODS.awaitRestartReady]: () =>
+          observeRpcEffect(
+            ORCHESTRATION_WS_METHODS.awaitRestartReady,
+            Effect.gen(function* () {
+              const commands = yield* ProviderCommandReactor;
+              const ingestion = yield* ProviderRuntimeIngestionService;
+              const checkpoints = yield* CheckpointReactor;
+              return yield* awaitRestartReady({
+                readThreads: Effect.all([
+                  projectionSnapshotQuery.getShellSnapshot(),
+                  projectionSnapshotQuery.getArchivedShellSnapshot(),
+                ]).pipe(
+                  Effect.map(([active, archived]) => [...active.threads, ...archived.threads]),
+                ),
+                drainCommands: commands.drain,
+                drainIngestion: ingestion.drain,
+                drainCheckpoints: checkpoints.drain,
+              });
+            }).pipe(
+              Effect.mapError(
+                (cause) =>
+                  new OrchestrationGetSnapshotError({
+                    message: "Failed to establish restart readiness",
+                    cause,
+                  }),
+              ),
+            ),
+            { "rpc.aggregate": "orchestration" },
+          ),
         [ORCHESTRATION_WS_METHODS.getArchivedShellSnapshot]: (_input) =>
           observeRpcEffect(
             ORCHESTRATION_WS_METHODS.getArchivedShellSnapshot,