From eeaac49a9d20d94970b26fe7775fc8ddb4672f4f Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Wed, 14 Oct 2026 00:38:17 -0700 Subject: [PATCH] fix: update all no longer runs agents twice or loses background work (#19291) Co-authored-by: Claude Opus 5.5 --- .../src/backend/DesktopBackendManager.test.ts | 2 +- .../src/backend/DesktopBackendManager.ts | 4 +- .../src/updates/DesktopUpdates.test.ts | 63 +++++ apps/desktop/src/updates/DesktopUpdates.ts | 72 ++++- .../src/updates/remoteUpdateFlow.test.ts | 18 ++ apps/desktop/src/updates/remoteUpdateFlow.ts | 13 + .../desktop/src/updates/updateMachine.test.ts | 14 + apps/desktop/src/updates/updateMachine.ts | 20 ++ .../desktop/src/updates/updatesTestHarness.ts | 32 ++- apps/server/src/mcp/OrchestratorMcpService.ts | 7 +- .../DelegatedCompletionDelivery.test.ts | 72 ++++- .../src/orchestration-v2/EffectOutbox.ts | 29 ++ .../FoundationPersistence.test.ts | 31 +++ .../src/orchestration-v2/Orchestrator.ts | 26 +- .../ProjectionRecovery.test.ts | 52 ++++ .../src/orchestration-v2/ProjectionStore.ts | 52 +++- .../ProviderRuntimeRecoveryService.ts | 26 ++ .../ProviderSessionManager.test.ts | 36 +++ .../ProviderSessionManager.ts | 6 +- .../orchestration-v2/RestartBackgroundNote.ts | 38 +++ .../RestartContinuation.test.ts | 231 +++++++++++++--- .../orchestration-v2/RestartContinuation.ts | 68 ++++- .../ThreadManagementService.test.ts | 123 +++++++++ .../ThreadManagementService.ts | 111 +++++--- .../src/orchestration-v2/runtimeLayer.test.ts | 41 ++- apps/server/src/process/agentScope.test.ts | 65 ++++- apps/server/src/process/agentScope.ts | 87 +++++- apps/server/src/terminal/Manager.ts | 1 + ...iderUpdateLaunchNotification.logic.test.ts | 3 +- .../ProviderUpdateLaunchNotification.logic.ts | 50 +++- .../src/components/ProviderUpdatesAction.tsx | 9 +- .../components/ServerUpdateAction.test.tsx | 5 +- .../web/src/components/ServerUpdateAction.tsx | 63 +++-- .../components/desktopUpdate.logic.test.ts | 20 ++ .../web/src/components/desktopUpdate.logic.ts | 7 + .../components/sidebar/SidebarUpdatePill.tsx | 202 +++++++++----- .../sidebar/UpdateEverythingDialog.tsx | 11 +- .../sidebar/updateEverything.logic.test.ts | 138 +++++++++- .../sidebar/updateEverything.logic.ts | 260 +++++++++++++++--- .../sidebar/updateEverythingRun.tsx | 142 ++++++++++ .../components/sidebar/useUpdateEverything.ts | 136 ++++++++- apps/web/src/routes/__root.tsx | 2 + docs/internals/server-updates.md | 11 +- docs/user/updating.md | 13 +- packages/contracts/src/ipc.ts | 7 + packages/shared/src/AgentScope.ts | 6 + 46 files changed, 2059 insertions(+), 366 deletions(-) create mode 100644 apps/web/src/components/sidebar/updateEverythingRun.tsx diff --git a/apps/desktop/src/backend/DesktopBackendManager.test.ts b/apps/desktop/src/backend/DesktopBackendManager.test.ts index 24303aa66997c4fddeb45f4f8c3c1cca21835d6a..e3e9ba62f317c461d22cbd1d8db74e885979afd6 100644 --- a/apps/desktop/src/backend/DesktopBackendManager.test.ts +++ b/apps/desktop/src/backend/DesktopBackendManager.test.ts @@ -264,7 +264,7 @@ describe("DesktopBackendManager", () => { assert.equal(spawnedCommand.options.additionalFds?.fd5?.type, "output"); assert.equal( Duration.toMillis(Duration.fromInputUnsafe(spawnedCommand.options.forceKillAfter)), - 2_000, + 5_000, ); assert.deepEqual(yield* decodeBootstrap(bootstrapJson), configWithObservability); diff --git a/apps/desktop/src/backend/DesktopBackendManager.ts b/apps/desktop/src/backend/DesktopBackendManager.ts index 6b38dadcbdf85b7b3c3c3fe7150ace3cf68d7686..313e8ba5ef37b698ef72c3cb6bde5c8b6caad488 100644 --- a/apps/desktop/src/backend/DesktopBackendManager.ts +++ b/apps/desktop/src/backend/DesktopBackendManager.ts @@ -65,7 +65,9 @@ const MAX_PREFLIGHT_FAILURE_ATTEMPTS = 5; const DEFAULT_BACKEND_READINESS_TIMEOUT = Duration.minutes(1); const DEFAULT_BACKEND_READINESS_INTERVAL = Duration.millis(100); const DEFAULT_BACKEND_READINESS_REQUEST_TIMEOUT = Duration.seconds(1); -const DEFAULT_BACKEND_TERMINATE_GRACE = Duration.seconds(2); +// Same as the service launcher. A killed backend leaves its agents running, +// and a busy one needs a few seconds to stop every provider session. +const DEFAULT_BACKEND_TERMINATE_GRACE = Duration.seconds(5); const DEFAULT_BACKEND_OUTPUT_DRAIN_TIMEOUT = Duration.seconds(5); const BACKEND_READINESS_PATH = "/.well-known/t3/environment"; const { logWarning: logBackendProcessWarning } = diff --git a/apps/desktop/src/updates/DesktopUpdates.test.ts b/apps/desktop/src/updates/DesktopUpdates.test.ts index debb3f080080b41c88e9773e9af768a8ceea892b..26ad0f1c4e705d82e65d2b02ab1a4eb7921a4700 100644 --- a/apps/desktop/src/updates/DesktopUpdates.test.ts +++ b/apps/desktop/src/updates/DesktopUpdates.test.ts @@ -652,6 +652,67 @@ describe("DesktopUpdates", () => { ).pipe(Effect.provide(Layer.merge(TestClock.layer(), harness.layer))); }); + it.effect("records the version an install hands to the installer", () => { + let recordsAtStop: ReadonlyArray = []; + const harness = makeHarness({ + stopBackend: Effect.sync(() => { + recordsAtStop = [...harness.installRecords.values()]; + }), + }); + + return Effect.scoped( + Effect.gen(function* () { + const updates = yield* DesktopUpdates.DesktopUpdates; + yield* updates.configure; + harness.emit("update-downloaded", { version: "1.2.4" }); + yield* flushCallbacks; + + assert.isTrue((yield* updates.install).accepted); + assert.deepEqual( + recordsAtStop.map((record) => JSON.parse(record)), + [{ fromVersion: "1.2.3", targetVersion: "1.2.4" }], + ); + }), + ).pipe(Effect.provide(Layer.merge(TestClock.layer(), harness.layer))); + }); + + it.effect("flags an install that did not apply and keeps the flag through checks", () => { + const harness = makeHarness({ + installRecord: JSON.stringify({ fromVersion: "1.2.3", targetVersion: "1.2.4" }), + }); + + return Effect.scoped( + Effect.gen(function* () { + const updates = yield* DesktopUpdates.DesktopUpdates; + yield* updates.configure; + assert.equal((yield* updates.getState).notAppliedVersion, "1.2.4"); + // Kept, so the next start on 1.2.3 still knows 1.2.4 did not apply. + assert.isDefined(harness.previousInstallRecord()); + + harness.emit("update-available", { version: "1.2.4" }); + yield* flushCallbacks; + const state = yield* updates.getState; + assert.equal(state.status, "available"); + assert.equal(state.notAppliedVersion, "1.2.4"); + }), + ).pipe(Effect.provide(Layer.merge(TestClock.layer(), harness.layer))); + }); + + it.effect("does not flag an install that applied", () => { + const harness = makeHarness({ + installRecord: JSON.stringify({ fromVersion: "1.2.2", targetVersion: "1.2.3" }), + }); + + return Effect.scoped( + Effect.gen(function* () { + const updates = yield* DesktopUpdates.DesktopUpdates; + yield* updates.configure; + assert.isUndefined((yield* updates.getState).notAppliedVersion); + assert.isUndefined(harness.previousInstallRecord()); + }), + ).pipe(Effect.provide(Layer.merge(TestClock.layer(), harness.layer))); + }); + it.effect("drops the update restart marker when an install is interrupted", () => Effect.gen(function* () { const stopping = yield* Deferred.make(); @@ -703,6 +764,8 @@ describe("DesktopUpdates", () => { assert.deepEqual(harness.installSteps, ["quitAndInstall", "startBackend"]); // The restarted old backend must release its tunnel on a later quit. assert.equal(harness.updateRestartMarkers.size, 0); + // The app never quit, so the next start must not report an unapplied install. + assert.equal(harness.installRecords.size, 0); }), ).pipe(Effect.provide(Layer.merge(TestClock.layer(), harness.layer))); }); diff --git a/apps/desktop/src/updates/DesktopUpdates.ts b/apps/desktop/src/updates/DesktopUpdates.ts index f0f483b97bff903c7039c15591a171cab9e81206..77069d3ba3fe76a704ef2409b9c545d1207a16ea 100644 --- a/apps/desktop/src/updates/DesktopUpdates.ts +++ b/apps/desktop/src/updates/DesktopUpdates.ts @@ -46,6 +46,7 @@ import { reduceDesktopUpdateStateOnInstallFailure, reduceDesktopUpdateStateOnNoUpdate, reduceDesktopUpdateStateOnUpdateAvailable, + resolveNotAppliedInstallVersion, } from "./updateMachine.ts"; const AUTO_UPDATE_STARTUP_DELAY = "15 seconds"; @@ -72,6 +73,14 @@ const UpdateInfo = Schema.Struct({ const DownloadProgressInfo = Schema.Struct({ percent: Schema.Number, }); +/** Saved in the runtime folder before an install quits the app, read on the next start. */ +export const DESKTOP_UPDATE_INSTALL_RECORD_FILE = "desktop-update-install.json"; +const DesktopUpdateInstallRecord = Schema.fromJsonString( + Schema.Struct({ fromVersion: Schema.String, targetVersion: Schema.String }), +); +const decodeInstallRecord = Schema.decodeUnknownEffect(DesktopUpdateInstallRecord); +const encodeInstallRecord = Schema.encodeEffect(DesktopUpdateInstallRecord); + const decodeAppUpdateYmlConfig = Schema.decodeUnknownEffect(AppUpdateYmlConfig); const decodeUpdateInfo = Schema.decodeUnknownEffect(UpdateInfo); const decodeDownloadProgressInfo = Schema.decodeUnknownEffect(DownloadProgressInfo); @@ -347,6 +356,7 @@ export const make = Effect.gen(function* () { const finishedUpdateActions = yield* PubSub.unbounded(); const updaterConfiguredRef = yield* Ref.make(false); const lastLoggedDownloadMilestoneRef = yield* Ref.make(-1); + const notAppliedVersionRef = yield* Ref.make(Option.none()); const updateStateRef = yield* Ref.make( createInitialDesktopUpdateState( environment.appVersion, @@ -600,11 +610,57 @@ export const make = Effect.gen(function* () { .remove(updateRestartMarkerPath, { force: true }) .pipe(Effect.ignore); + // Records the version an install hands to the OS installer, so the next + // start can tell whether it applied. Best effort, like the restart marker. + const installRecordPath = environment.path.join( + updateRestartMarkerDir, + DESKTOP_UPDATE_INSTALL_RECORD_FILE, + ); + const writeInstallRecord = (targetVersion: string) => + encodeInstallRecord({ fromVersion: environment.appVersion, targetVersion }).pipe( + Effect.flatMap((text) => fileSystem.writeFileString(installRecordPath, text)), + Effect.catch((error) => + logUpdaterWarning("Could not record the update install.", { errorTag: error._tag }), + ), + ); + const removeInstallRecord = fileSystem + .remove(installRecordPath, { force: true }) + .pipe(Effect.ignore); + + // Reads the record of the last install. Returns the version it did not + // apply, when the app came back on the version it installed from. That + // record stays through later starts, until the version changes or a new + // install replaces it, so no restart forgets the failed version. + const readNotAppliedInstallVersion = fileSystem.readFileString(installRecordPath).pipe( + Effect.flatMap(decodeInstallRecord), + Effect.option, + Effect.map((record) => + Option.flatMap(record, (value) => + Option.fromNullishOr(resolveNotAppliedInstallVersion(value, environment.appVersion)), + ), + ), + Effect.tap((notAppliedVersion) => + Option.isNone(notAppliedVersion) ? removeInstallRecord : Effect.void, + ), + ); + + const baseUpdateState = (channel: DesktopUpdateChannel, enabled: boolean) => + Ref.get(notAppliedVersionRef).pipe( + Effect.map((notAppliedVersion) => ({ + ...createBaseUpdateState(channel, enabled, environment), + ...Option.match(notAppliedVersion, { + onNone: () => ({}), + onSome: (version) => ({ notAppliedVersion: version }), + }), + })), + ); + const resetInstallAction = Effect.all( [ finishUpdateAction("install"), Ref.set(desktopState.quitting, false), removeUpdateRestartMarker, + removeInstallRecord, ], { discard: true }, ); @@ -621,6 +677,7 @@ export const make = Effect.gen(function* () { yield* Ref.set(desktopState.quitting, false); yield* removeUpdateRestartMarker; + yield* removeInstallRecord; yield* Effect.gen(function* () { const instances = yield* pool.list; const restartExit = yield* Effect.forEach(instances, (instance) => instance.start, { @@ -691,6 +748,8 @@ export const make = Effect.gen(function* () { return yield* Effect.gen(function* () { yield* writeUpdateRestartMarker; + const { downloadedVersion } = yield* Ref.get(updateStateRef); + if (downloadedVersion !== null) yield* writeInstallRecord(downloadedVersion); // Stop every backend in the pool, not just the primary. With // parallel WSL + Windows backends, leaving the WSL instance up // means quitAndInstall's app.quit() exits before the pool's @@ -985,6 +1044,15 @@ export const make = Effect.gen(function* () { const appUpdateYmlConfig = yield* readAppUpdateYml; yield* Ref.set(appUpdateYmlConfigRef, appUpdateYmlConfig); + const notAppliedVersion = yield* readNotAppliedInstallVersion; + yield* Ref.set(notAppliedVersionRef, notAppliedVersion); + if (Option.isSome(notAppliedVersion)) { + yield* logUpdaterWarning("the last update install did not apply", { + version: environment.appVersion, + targetVersion: notAppliedVersion.value, + }); + } + if (config.mockUpdates) { yield* electronUpdater.setFeedURL({ provider: "generic", @@ -994,7 +1062,7 @@ export const make = Effect.gen(function* () { const settings = yield* desktopSettings.get; const enabled = yield* shouldEnableAutoUpdates; - yield* setState(createBaseUpdateState(settings.updateChannel, enabled, environment)); + yield* setState(yield* baseUpdateState(settings.updateChannel, enabled)); if (!enabled) { return; } @@ -1065,7 +1133,7 @@ export const make = Effect.gen(function* () { ); const enabled = yield* shouldEnableAutoUpdates; - yield* setState(createBaseUpdateState(nextChannel, enabled, environment)); + yield* setState(yield* baseUpdateState(nextChannel, enabled)); if (!enabled || !(yield* Ref.get(updaterConfiguredRef))) { return yield* Ref.get(updateStateRef); diff --git a/apps/desktop/src/updates/remoteUpdateFlow.test.ts b/apps/desktop/src/updates/remoteUpdateFlow.test.ts index 7fd50a82f49985594af95e180b771d2f0837174d..acc8479013fcd1eaafdffbe6b9cdbd41abbb1d78 100644 --- a/apps/desktop/src/updates/remoteUpdateFlow.test.ts +++ b/apps/desktop/src/updates/remoteUpdateFlow.test.ts @@ -34,6 +34,24 @@ function makeState(overrides: Partial = {}): DesktopUpdateSt } describe("nextRemoteDesktopUpdateStep", () => { + it("does not install again a version that did not apply on this machine", () => { + const step = nextRemoteDesktopUpdateStep( + makeState({ status: "available", availableVersion: "1.2.4", notAppliedVersion: "1.2.4" }), + NO_ATTEMPTS, + null, + ); + assert.equal(step.action, "done"); + assert.include(step.action === "done" ? step.reason : "", "1.2.4 did not apply"); + assert.deepEqual( + nextRemoteDesktopUpdateStep( + makeState({ status: "available", availableVersion: "1.2.5", notAppliedVersion: "1.2.4" }), + NO_ATTEMPTS, + null, + ), + { action: "download" }, + ); + }); + it("fails immediately when updates are disabled, preferring the known reason", () => { const state = makeState({ enabled: false, status: "disabled" }); assert.deepEqual(nextRemoteDesktopUpdateStep(state, NO_ATTEMPTS, "dev build"), { diff --git a/apps/desktop/src/updates/remoteUpdateFlow.ts b/apps/desktop/src/updates/remoteUpdateFlow.ts index 20043ba22cca30c0263521db5da597215b1c0ebe..a1523baee58cee1e1d3f7ad13a643151e847cc77 100644 --- a/apps/desktop/src/updates/remoteUpdateFlow.ts +++ b/apps/desktop/src/updates/remoteUpdateFlow.ts @@ -52,6 +52,19 @@ export function nextRemoteDesktopUpdateStep( reason: disabledReason ?? "Automatic updates are disabled on this machine.", }; } + // An install of this version already failed to apply here. Installing it + // again from afar would stop the app's threads for nothing, so only the user + // on this machine retries it. + if ( + state.notAppliedVersion !== undefined && + (state.downloadedVersion ?? state.availableVersion) === state.notAppliedVersion + ) { + return { + action: "done", + outcome: "failed", + reason: `Update ${state.notAppliedVersion} did not apply on this machine. Retry it from the desktop app there.`, + }; + } // Mirror installDownloadedUpdate's own admission rule exactly, so the run // never reports an install that the updater then refuses. A download // survives an unrelated background updater error (errorContext null) and diff --git a/apps/desktop/src/updates/updateMachine.test.ts b/apps/desktop/src/updates/updateMachine.test.ts index b3450c8717506521d454faf08d42409a9000eef3..de57e0a8e2c705bd01aab48192fafdac566bc7b6 100644 --- a/apps/desktop/src/updates/updateMachine.test.ts +++ b/apps/desktop/src/updates/updateMachine.test.ts @@ -11,6 +11,7 @@ import { reduceDesktopUpdateStateOnInstallFailure, reduceDesktopUpdateStateOnNoUpdate, reduceDesktopUpdateStateOnUpdateAvailable, + resolveNotAppliedInstallVersion, } from "./updateMachine.ts"; const runtimeInfo = { @@ -255,3 +256,16 @@ describe("updateMachine", () => { expect(state.omittedReleaseCount).toBe(0); }); }); + +describe("resolveNotAppliedInstallVersion", () => { + const record = { fromVersion: "1.2.3", targetVersion: "1.2.4" }; + + it("flags the target when the app came back on the version it installed from", () => { + expect(resolveNotAppliedInstallVersion(record, "1.2.3")).toBe("1.2.4"); + }); + + it("stays quiet when the install applied or the app runs another version", () => { + expect(resolveNotAppliedInstallVersion(record, "1.2.4")).toBeNull(); + expect(resolveNotAppliedInstallVersion(record, "1.2.5")).toBeNull(); + }); +}); diff --git a/apps/desktop/src/updates/updateMachine.ts b/apps/desktop/src/updates/updateMachine.ts index 85d8782ea66887a5748e4dfed9b40a497557ecd6..48fcc07acda9bc0e028b8f3da8846851dfcd4e9e 100644 --- a/apps/desktop/src/updates/updateMachine.ts +++ b/apps/desktop/src/updates/updateMachine.ts @@ -211,3 +211,23 @@ export function reduceDesktopUpdateStateOnInstallFailure( canRetry: true, }; } + +/** The install this app handed to the OS installer, saved right before it quits. */ +export interface DesktopUpdateInstallRecord { + readonly fromVersion: string; + readonly targetVersion: string; +} + +/** + * Read on the next start: the version an install did not apply, because the + * app came back on the version it installed from. Null when the install + * applied, or when the app now runs some other version. + */ +export function resolveNotAppliedInstallVersion( + record: DesktopUpdateInstallRecord, + appVersion: string, +): string | null { + return record.fromVersion === appVersion && record.targetVersion !== appVersion + ? record.targetVersion + : null; +} diff --git a/apps/desktop/src/updates/updatesTestHarness.ts b/apps/desktop/src/updates/updatesTestHarness.ts index 9d4556af42b47a55af67110a141bbf1a6083de4b..c3701280ad55a89bd7aa514c4ac26c83a2c6e976 100644 --- a/apps/desktop/src/updates/updatesTestHarness.ts +++ b/apps/desktop/src/updates/updatesTestHarness.ts @@ -37,6 +37,8 @@ export interface UpdatesHarnessOptions { readonly platform?: NodeJS.Platform; /** Contents of the resources/package-type marker a Linux package ships. */ readonly packageType?: string | undefined; + /** Contents of the install record a previous run left behind. */ + readonly installRecord?: string | undefined; } export function makeHarness(options: UpdatesHarnessOptions = {}) { @@ -208,12 +210,21 @@ export function makeHarness(options: UpdatesHarnessOptions = {}) { } satisfies DesktopAppSettings.DesktopAppSettings["Service"]) : DesktopAppSettings.layer; - // Tracks the restart markers installs leave, so installs stay free of real - // disk I/O that would outrun the tests' settle loops. + // Tracks the restart markers and install records installs leave, so + // installs stay free of real disk I/O that would outrun the tests' settle loops. const updateRestartMarkers = new Set(); + // Contents of the install record, by path once an install writes one. + const installRecords = new Map(); + let previousInstallRecord = options.installRecord; + const isInstallRecord = (path: string) => + path.endsWith(DesktopUpdates.DESKTOP_UPDATE_INSTALL_RECORD_FILE); const layerFileSystem = FileSystem.layerNoop({ - readFileString: (path) => - path === "/missing/resources/package-type" && options.packageType !== undefined + readFileString: (path) => { + const installRecord = isInstallRecord(path) + ? (installRecords.get(path) ?? previousInstallRecord) + : undefined; + if (installRecord !== undefined) return Effect.succeed(installRecord); + return path === "/missing/resources/package-type" && options.packageType !== undefined ? Effect.succeed(options.packageType) : Effect.fail( PlatformError.systemError({ @@ -222,15 +233,19 @@ export function makeHarness(options: UpdatesHarnessOptions = {}) { _tag: "NotFound", pathOrDescriptor: path, }), - ), + ); + }, makeDirectory: () => Effect.void, - writeFileString: (path) => + writeFileString: (path, data) => Effect.sync(() => { - updateRestartMarkers.add(path); + if (isInstallRecord(path)) installRecords.set(path, data); + else updateRestartMarkers.add(path); }), remove: (path) => Effect.sync(() => { updateRestartMarkers.delete(path); + installRecords.delete(path); + if (isInstallRecord(path)) previousInstallRecord = undefined; }), }); @@ -259,6 +274,9 @@ export function makeHarness(options: UpdatesHarnessOptions = {}) { quitAndInstalls: () => quitAndInstallCount, installSteps, updateRestartMarkers, + installRecords, + /** The record the test started with, until the app removes it. */ + previousInstallRecord: () => previousInstallRecord, downloadCount: () => downloadCount, feedUrls: (): ElectronUpdater.ElectronUpdaterFeedUrl[] => feedUrls, fullChangelog: () => fullChangelog, diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index 20607c8292aa5d4540391d1c4aac149c62c3a929..4055544a394731d34eac33ff7cdb9dee0a1353c0 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -81,6 +81,7 @@ import { DispatchModeLimit, type DispatchModeRefusal, } from "../orchestration-v2/DispatchModeLimit.ts"; +import { runOutcome } from "../orchestration-v2/RestartBackgroundNote.ts"; import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts"; import * as ThreadLaunchService from "../orchestration-v2/ThreadLaunchService.ts"; import * as GitVcsDriver from "../vcs/GitVcsDriver.ts"; @@ -330,9 +331,9 @@ function invalidOptionSelections( } function taskStatusForRun( - run: Pick | undefined, + run: Pick | undefined, ): OrchestratorMcpDelegateTaskResult["status"] { - switch (run?.status) { + switch (run === undefined ? undefined : runOutcome(run)) { case "queued": return "queued"; case "waiting": @@ -2642,7 +2643,7 @@ const make = Effect.gen(function* () { return { threadId: input.threadId, runId: result.run?.id ?? null, - status: result.run?.status ?? "idle", + status: result.run === null ? "idle" : runOutcome(result.run), timedOut: result.timedOut, } satisfies OrchestratorMcpThreadWaitResult; }), diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index ad0c6703c1d9ba1f765e079a1e89bb578ae94fa1..003844efa7206f55289db15ee5fc89e36097d32e 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -803,6 +803,8 @@ const seedRestartCancelledChild = (input: { readonly continuationPending: boolean; /** "completed" seeds a settled turn whose background work the restart cancelled. */ readonly runStatus?: "cancelled" | "completed"; + /** Background work the restart cancelled on the child's run. */ + readonly lostWork?: OrchestrationV2Run["restartCancelledBackgroundWork"]; readonly now: DateTime.Utc; }) => Effect.gen(function* () { @@ -880,16 +882,22 @@ const seedRestartCancelledChild = (input: { }, ], }); + const childRun = runEvent({ + threadId: childThreadId, + runId: childRunId, + ordinal: 1, + status: input.runStatus ?? "cancelled", + now: input.now, + }); yield* eventSink.writeWithEffects({ commandId: reconcileCommandId(input.name), events: [ - runEvent({ - threadId: childThreadId, - runId: childRunId, - ordinal: 1, - status: input.runStatus ?? "cancelled", - now: input.now, - }), + input.lostWork === undefined + ? childRun + : { + ...childRun, + payload: { ...childRun.payload, restartCancelledBackgroundWork: input.lostWork }, + }, ], effects: input.continuationPending ? [ @@ -1427,6 +1435,56 @@ it.layer(layerTest)("delegated tasks across a server restart", (it) => { }), ); + it.effect("reports a child whose monitor the restart cut as interrupted, not completed", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const now = yield* DateTime.now; + const threadId = ThreadId.make("thread:restart-monitor-parent"); + const projectId = ProjectId.make("project:restart-monitor-parent"); + const runId = RunId.make("run:restart-monitor-parent"); + const rootNodeId = NodeId.make("node:restart-monitor-parent-root"); + yield* seedParentWithTerminalTask({ + threadId, + projectId, + runId, + rootNodeId, + taskId: NodeId.make("node:restart-monitor-parent-settled"), + deliveryState: "delivered", + now, + }); + // The child ended its turn to wait on a gate monitor; the restart cut it. + const child = yield* seedRestartCancelledChild({ + parentThreadId: threadId, + projectId, + parentRunId: runId, + rootNodeId, + name: "restart-monitor-child", + completionWake: "always", + continuationPending: true, + runStatus: "completed", + lostWork: [{ kind: "monitor", label: "Gate attempt 9" }], + now, + }); + yield* orchestrator.recoverDelegatedTasks; + const held = yield* orchestrator.getThreadProjection(threadId); + assert.equal(held.subagents.find((row) => row.id === child.taskId)?.status, "running"); + + // Continuation is off, so the cut run is the outcome. It still goes to the parent. + yield* continueRestartedRun({ + threadId: child.childThreadId, + sourceRunId: child.childRunId, + }).pipe( + Effect.provide(ServerSettings.layerTest({ continueThreadsAfterServerUpdate: false })), + ); + + const settled = yield* orchestrator.getThreadProjection(threadId); + const task = settled.subagents.find((row) => row.id === child.taskId); + assert.equal(task?.status, "interrupted"); + assert.equal(task?.result, "Child task ended with status interrupted."); + assert.equal(task?.completionDelivery?.state, "claimed"); + }), + ); + // A running provider turn does not prove the provider consumed the delivery // (Claude marks the turn running before it reads the prompt), so a cut // delivery is offered again even when a continuation resumes that turn. diff --git a/apps/server/src/orchestration-v2/EffectOutbox.ts b/apps/server/src/orchestration-v2/EffectOutbox.ts index 0cb3de2e0012f1bfaf3368bcf13929acab555938..c50429f3ed4d2b1be4697e78eb9b1bb5a4626e62 100644 --- a/apps/server/src/orchestration-v2/EffectOutbox.ts +++ b/apps/server/src/orchestration-v2/EffectOutbox.ts @@ -224,6 +224,8 @@ export interface EffectOutboxV2Shape { readonly signalCancellations: (effectIds: ReadonlyArray) => Effect.Effect; readonly awaitCancellation: (effectId: string) => Effect.Effect; readonly clearCancellation: (effectId: string) => Effect.Effect; + /** Resolves once the effect is neither pending nor running, or right away when it does not exist. */ + readonly awaitSettled: (effectId: string) => Effect.Effect; readonly reconcileAfterProcessLoss: Effect.Effect< { readonly requeued: number; readonly cancelled: number }, EffectOutboxError @@ -307,6 +309,18 @@ export const layer: Layer.Layer = La // for distinct threads without allowing notifications to grow unbounded. const available = yield* Queue.dropping(64); const cancellationSignals = new Map>(); + const settledSignals = new Map>(); + const signalSettled = (effectIds: ReadonlyArray) => + Effect.forEach( + effectIds, + (effectId) => { + const signal = settledSignals.get(effectId); + if (signal === undefined) return Effect.void; + settledSignals.delete(effectId); + return Deferred.succeed(signal, undefined); + }, + { discard: true }, + ); const notifyAvailable = (count = 1) => Queue.offerAll( available, @@ -477,6 +491,7 @@ export const layer: Layer.Layer = La }, { discard: true }, ); + yield* signalSettled(effectIds); // Cancellation can unblock other pending work on the same thread. // This method is deliberately post-commit, so it is also the safe // place to wake claimers after the durable status change. @@ -487,6 +502,18 @@ export const layer: Layer.Layer = La Effect.sync(() => { cancellationSignals.delete(effectId); }), + awaitSettled: (effectId) => + Effect.gen(function* () { + // Registered before the read, so a settle right after it still resolves. + const signal = settledSignals.get(effectId) ?? Deferred.makeUnsafe(); + settledSignals.set(effectId, signal); + const effect = yield* service.get(effectId); + if (Option.exists(effect, ({ status }) => status === "pending" || status === "running")) { + return yield* Deferred.await(signal); + } + // Settled already: release every waiter that shares this signal. + yield* signalSettled([effectId]); + }), reconcileAfterProcessLoss: Effect.gen(function* () { const now = DateTime.formatIso(yield* DateTime.now); const cancelledRows = yield* sql<{ readonly effect_id: string }>` @@ -599,6 +626,7 @@ export const layer: Layer.Layer = La `; if (rows.length === 1) { cancellationSignals.delete(effectId); + yield* signalSettled([effectId]); yield* notifyAvailable(); } return rows.length === 1; @@ -657,6 +685,7 @@ export const layer: Layer.Layer = La `; if (rows.length === 1) { cancellationSignals.delete(effectId); + yield* signalSettled([effectId]); yield* notifyAvailable(); } return rows.length === 1; diff --git a/apps/server/src/orchestration-v2/FoundationPersistence.test.ts b/apps/server/src/orchestration-v2/FoundationPersistence.test.ts index d6bd2b1b823f31ef924bb7f6f4369259819a4173..023b1dbf22563ccb80051b62073ee67be7698304 100644 --- a/apps/server/src/orchestration-v2/FoundationPersistence.test.ts +++ b/apps/server/src/orchestration-v2/FoundationPersistence.test.ts @@ -2084,6 +2084,37 @@ it.layer(layerTest)("orchestration V2 foundation persistence", (it) => { }), ); + it.effect("wakes a settle waiter when its effect finishes", () => + Effect.gen(function* () { + const outbox = yield* EffectOutbox.EffectOutboxV2; + const effectId = "effect:foundation-await-settled"; + yield* outbox.enqueue([ + { + id: effectId, + commandId: CommandId.make("command:foundation-await-settled"), + threadId: ThreadId.make("thread:foundation-await-settled"), + request: { + type: "provider-runtime.continue", + sourceRunId: RunId.make("run:foundation-await-settled"), + }, + }, + ]); + const waiter = yield* outbox.awaitSettled(effectId).pipe(Effect.forkChild); + const claimed = yield* outbox.claimNext({ + workerId: "settle-worker", + leaseDurationMs: 30_000, + }); + assert.isTrue(Option.isSome(claimed)); + yield* Effect.yieldNow; + assert.isUndefined(waiter.pollUnsafe()); + + yield* outbox.succeed({ effectId, workerId: "settle-worker" }); + yield* Fiber.join(waiter); + // A settled effect does not wait. + yield* outbox.awaitSettled(effectId); + }), + ); + it.effect("runs title generation beside critical work while serializing each lane", () => Effect.gen(function* () { const outbox = yield* EffectOutbox.EffectOutboxV2; diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 6853c6457d6d021335ca655157b53fa9f23c0d38..35a2ecf02c69185dea806950c30678eef5c34688 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -88,7 +88,7 @@ import { CommandPolicyV2, resolveMessageDispatchIntent } from "./CommandPolicy.t import { CommandReceiptStoreV2 } from "./CommandReceiptStore.ts"; import { ContextHandoffServiceV2 } from "./ContextHandoffService.ts"; import { notificationTurnItem } from "@t3tools/provider-core/server/notification"; -import { isRestartNoteSource } from "./RestartBackgroundNote.ts"; +import { isRestartContinuationSource, runOutcome } from "./RestartBackgroundNote.ts"; import { isBufferedProviderNotification, isUndeliveredMailboxSteer, @@ -293,6 +293,16 @@ export interface OrchestratorV2Shape { childThreadId: ThreadId, taskId?: NodeId, ) => Effect.Effect; + /** + * Whether a restart continuation of this cut run has not started yet, so the + * run's status is not its outcome. A failed read counts as not pending. + */ + readonly restartContinuationPending: (run: OrchestrationV2Run) => Effect.Effect; + /** + * Resolves once that continuation has run or was declined. A declined one + * records no thread event, so a waiter needs this to wake. + */ + readonly awaitRestartContinuation: (run: OrchestrationV2Run) => Effect.Effect; readonly dispatch: ( command: OrchestrationV2ServerCommand, ) => Effect.Effect; @@ -4624,8 +4634,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const source = projection.runs.find((run) => run.id === command.restartContinuationOfRunId); if ( !source || - source.status !== "cancelled" || - isRestartNoteSource(source, projection.providerTurns) || + !isRestartContinuationSource(source, projection.providerTurns) || projection.thread.archivedAt !== null || projection.thread.deletedAt !== null || projection.thread.providerInstanceId !== source.providerInstanceId || @@ -10147,7 +10156,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio : undefined : taskRuns.toSorted((a, b) => b.ordinal - a.ordinal)[0]; if (childRun === undefined) return; - const terminalStatus = delegatedTaskTerminalStatus(childRun.status); + const outcome = runOutcome(childRun); + const terminalStatus = delegatedTaskTerminalStatus(outcome); if (terminalStatus === null) { return; } @@ -10213,7 +10223,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio } const now = yield* DateTime.now; - const result = subagentResultForRun(childProjection, childRun); + const result = subagentResultForRun(childProjection, { ...childRun, status: outcome }); const parentRun = task.runId === null ? undefined @@ -11459,6 +11469,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio recoverDelegatedTasks, recoverDelegatedTask, delegatedTaskResultPending, + restartContinuationPending: (run) => + awaitsRestartContinuation(run).pipe(Effect.orElseSucceed(() => false)), + awaitRestartContinuation: (run) => + effectOutbox.awaitSettled(`effect:restart-continuation:${run.id}`).pipe(Effect.ignore), dispatch: dispatchWithReceipt, searchThreadStream: (input) => projectionStore @@ -11607,6 +11621,8 @@ const layerUnavailable: Layer.Layer = Layer.succeed( recoverDelegatedTasks: Effect.void, recoverDelegatedTask: () => Effect.void, delegatedTaskResultPending: () => Effect.succeed(false), + restartContinuationPending: () => Effect.succeed(false), + awaitRestartContinuation: () => Effect.void, dispatch: (command) => Effect.fail( new OrchestratorDispatchError({ diff --git a/apps/server/src/orchestration-v2/ProjectionRecovery.test.ts b/apps/server/src/orchestration-v2/ProjectionRecovery.test.ts index e616b1d4f3b5ffcbb6eafbdc9ebb35b60750ee24..fee72ee72fae21bca4e262fea46e5cd521670400 100644 --- a/apps/server/src/orchestration-v2/ProjectionRecovery.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionRecovery.test.ts @@ -209,6 +209,58 @@ it.effect.each(["startup", "shutdown"] as const)( }).pipe(Effect.provide(layerRecoveryTest)), ); +it.effect("stops a session its thread detached, which kept its live status", () => + Effect.gen(function* () { + const projections = yield* ProjectionStore.ProjectionStoreV2; + const recovery = yield* ProviderRuntimeRecovery.ProviderRuntimeRecoveryService; + const now = yield* DateTime.now; + const threadId = yield* createThread("detached-session"); + const providerSessionId = ProviderSessionId.make("session:recovery:detached"); + yield* projections.apply({ + id: EventId.make("event:recovery:detached-session:attached"), + type: "provider-session.attached", + threadId, + driver, + providerInstanceId, + occurredAt: now, + payload: { + id: providerSessionId, + driver, + providerInstanceId, + status: "ready", + cwd: "/workspace", + model: modelSelection.model, + capabilities: CodexProviderCapabilitiesV2, + createdAt: now, + updatedAt: now, + lastError: null, + }, + }); + // The server died before the release that should follow the detach. + yield* projections.apply({ + id: EventId.make("event:recovery:detached-session:detached"), + type: "provider-session.detached", + threadId, + driver, + providerInstanceId, + occurredAt: now, + payload: { providerSessionId, detachedAt: now, reason: "Thread settled." }, + }); + assert.deepEqual(yield* projections.getRecoveryThreadIds("runtime"), [threadId]); + + const summary = yield* recovery.reconcile("startup"); + + assert.equal(summary.stoppedSessions, 1); + const sql = yield* SqlClient.SqlClient; + const rows = yield* sql<{ readonly status: string }>` + SELECT status FROM orchestration_v2_projection_provider_sessions + WHERE provider_session_id = ${providerSessionId} + `; + assert.deepEqual(rows, [{ status: "stopped" }]); + assert.deepEqual(yield* projections.getRecoveryThreadIds("runtime"), []); + }).pipe(Effect.provide(layerRecoveryTest)), +); + it.effect("selects unfinished recovery work without reading settled thread histories", () => Effect.gen(function* () { const projections = yield* ProjectionStore.ProjectionStoreV2; diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 605d1b01b6f5e6ec41140122d46d0af38d6587e5..7c156ad71361df6579a78347bd4661f3a9c57768 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -3666,6 +3666,16 @@ export const layer: Layer.Layer = ON bindings.provider_session_id = sessions.provider_session_id WHERE sessions.status NOT IN ('stopped', 'error') UNION + -- A detach drops the binding but keeps the session's status. + SELECT sessions.thread_id + FROM orchestration_v2_projection_provider_sessions AS sessions + WHERE sessions.status NOT IN ('stopped', 'error') + AND sessions.thread_id IS NOT NULL + AND NOT EXISTS ( + SELECT 1 FROM orchestration_v2_projection_provider_session_bindings AS bindings + WHERE bindings.provider_session_id = sessions.provider_session_id + ) + UNION SELECT thread_id FROM pending_provider_threads UNION SELECT nodes.thread_id FROM pending_provider_threads @@ -4123,24 +4133,38 @@ export const layer: Layer.Layer = ORDER BY COALESCE(subagent.started_at, ''), subagent.subagent_id ASC `, sql` - SELECT DISTINCT session.payload_json + SELECT session.payload_json FROM orchestration_v2_projection_provider_sessions AS session - JOIN orchestration_v2_projection_provider_session_bindings AS binding - ON binding.provider_session_id = session.provider_session_id - WHERE binding.thread_id = ${threadId} - AND ( - session.status NOT IN ('stopped', 'error') - OR session.provider_session_id IN ( - SELECT provider_thread.provider_session_id - FROM orchestration_v2_projection_provider_threads AS provider_thread - WHERE provider_thread.provider_thread_id IN ( - SELECT run.provider_thread_id - FROM orchestration_v2_projection_runs AS run - WHERE run.thread_id = ${threadId} - AND run.status IN ('queued', 'preparing', 'starting', 'running', 'waiting') + WHERE ( + session.provider_session_id IN ( + SELECT binding.provider_session_id + FROM orchestration_v2_projection_provider_session_bindings AS binding + WHERE binding.thread_id = ${threadId} + ) + AND ( + session.status NOT IN ('stopped', 'error') + OR session.provider_session_id IN ( + SELECT provider_thread.provider_session_id + FROM orchestration_v2_projection_provider_threads AS provider_thread + WHERE provider_thread.provider_thread_id IN ( + SELECT run.provider_thread_id + FROM orchestration_v2_projection_runs AS run + WHERE run.thread_id = ${threadId} + AND run.status IN ('queued', 'preparing', 'starting', 'running', 'waiting') + ) ) ) ) + -- This thread detached it, which keeps its status. Its process + -- did not outlive the server either, so recovery stops it. + OR ( + session.thread_id = ${threadId} + AND session.status NOT IN ('stopped', 'error') + AND NOT EXISTS ( + SELECT 1 FROM orchestration_v2_projection_provider_session_bindings AS binding + WHERE binding.provider_session_id = session.provider_session_id + ) + ) ORDER BY session.updated_at ASC, session.provider_session_id ASC `, sql` diff --git a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts index 27cacf50da9416e1bbbc37426eb019850d05c551..d40e517f7effe45c4bcd872f312444047009b144 100644 --- a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts +++ b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts @@ -7,6 +7,7 @@ import { type OrchestrationV2RestartCancelledBackgroundWork, type OrchestrationV2Subagent, type OrchestrationV2ThreadProjection, + isOrchestrationV2WorkActive, ThreadId, } from "@t3tools/contracts"; import * as Cause from "effect/Cause"; @@ -445,6 +446,31 @@ export const make = Effect.gen(function* () { }); } } + // A blocking delegate_task wait dies with the turn that made it, and its + // continuation never got the tool result. Let the child's result wake + // the parent like an async task, so it reaches the continued turn. + const cutRunIds = new Set(runs.map((run) => run.id)); + for (const task of projection.subagents ?? []) { + if ( + task.origin !== "app_owned" || + task.runId === null || + !cutRunIds.has(task.runId) || + task.completionWake !== "settled_only" || + !isOrchestrationV2WorkActive(task.status) + ) + continue; + events.push({ + id: yield* allocateEventId(), + type: "subagent.updated", + threadId: projection.thread.id, + runId: task.runId, + nodeId: task.id, + driver: task.driver, + providerInstanceId: task.providerInstanceId, + occurredAt: now, + payload: { ...task, completionWake: "always", updatedAt: now }, + }); + } // Process loss also orphans background-capable turn items on already- // settled runs (e.g. post-settle Waiting work). Skip items already // cancelled above for recovered nonterminal runs to avoid duplicate diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts index ceb230b00894095db7889b718d0d60e3f8646d8b..39bf2e2a8766a96837ba9ef726a1fda03f614fb4 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -1681,6 +1681,42 @@ it.effect("ProviderSessionManagerV2 releases live sessions when its layer shuts }), ); +it.effect("ProviderSessionManagerV2 releases sessions in parallel on server shutdown", () => + Effect.gen(function* () { + const state = yield* Ref.make(emptyState); + yield* Effect.gen(function* () { + const eventSink = yield* EventSink.EventSinkV2; + const idAllocator = yield* IdAllocator.IdAllocatorV2; + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; + const now = yield* DateTime.now; + for (const name of ["a", "b", "c"]) { + const threadId = ThreadId.make(`thread-provider-session-manager-parallel-shutdown-${name}`); + yield* eventSink.write({ + events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })], + }); + yield* manager.open({ + threadId, + providerSessionId: yield* idAllocator.allocate.providerSession({ + providerInstanceId: modelSelection.instanceId, + threadId, + }), + modelSelection, + runtimePolicy, + }); + } + + // Every close hangs until its time box ends. One time box must cover all + // three sessions, or a busy server outlives the launcher's kill timer. + const stopping = yield* manager.shutdown.pipe(Effect.forkChild); + yield* TestClock.adjust("30 seconds"); + yield* Effect.yieldNow; + assert.isDefined(stopping.pollUnsafe()); + }).pipe( + Effect.provide(layerTest({ state, idleTimeoutMs: 60_000, hangSessionScopeClose: true })), + ); + }), +); + it.effect("ProviderSessionManagerV2 closes event subscriptions normally on server shutdown", () => Effect.gen(function* () { const state = yield* Ref.make(emptyState); diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.ts index f0ec5282ef5b58fdd28b546ae825f46502c93d91..d7d1e12c10e8cc3e001c662cc6d3c706fa6d68aa 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -1999,8 +1999,8 @@ export const layerWithOptions = ( // Parent of every session scope. On layer close, shutdown releases the // live sessions first, then closes any session whose open is still in // flight, time-boxed so a stuck adapter cannot hold up server shutdown. - // Parallel, so one session whose close hangs does not stop the rest from - // closing within the time box. + // Both steps run in parallel. The service launcher kills the server + // 5 s after it asks it to stop, and a busy server can hold 15 sessions. const sessionScopes = yield* Scope.make("parallel"); const shutdown = Effect.gen(function* () { const activeSessions = [...(yield* Ref.get(sessions)).values()]; @@ -2018,7 +2018,7 @@ export const layerWithOptions = ( }), ), ), - { discard: true }, + { concurrency: "unbounded", discard: true }, ); }); yield* Effect.addFinalizer(() => diff --git a/apps/server/src/orchestration-v2/RestartBackgroundNote.ts b/apps/server/src/orchestration-v2/RestartBackgroundNote.ts index 11b2cfe37c0877aa3ea2a6fef7cf71a0e849da53..0ce213da85fe3b240d548aa3da83e8472526197c 100644 --- a/apps/server/src/orchestration-v2/RestartBackgroundNote.ts +++ b/apps/server/src/orchestration-v2/RestartBackgroundNote.ts @@ -123,6 +123,44 @@ export function isRestartNoteSource( ); } +/** + * Work an agent ends its turn to wait for: a monitor, a subagent, or a + * background task. A restart that cancels it wakes the settled turn, because + * nothing else will. A shell does not wake it: most are dev servers that + * nobody waits on, so their note waits for the next user message. + */ +export function isAwaitedWork(work: Pick): boolean { + return work.kind !== "shell"; +} + +/** + * A run's status as its waiter should read it: a delegating parent, + * `task_status`, or `t3_thread_wait`. A turn that ended to wait on work the + * restart then cancelled did not finish, so it is "interrupted", not + * "completed", and its last message is not a final result. + */ +export function runOutcome( + run: Pick, +): OrchestrationV2Run["status"] { + return (run.status === "completed" || run.status === "cancelled") && + (run.restartCancelledBackgroundWork ?? []).some(isAwaitedWork) + ? "interrupted" + : run.status; +} + +/** + * Whether a restart continuation may prompt from `source`: a turn the restart + * cut mid-way, or a settled turn that lost awaited work in the restart. + */ +export function isRestartContinuationSource( + source: Pick, + providerTurns: ReadonlyArray>, +): boolean { + return isRestartNoteSource(source, providerTurns) + ? (source.restartCancelledBackgroundWork ?? []).some(isAwaitedWork) + : source.status === "cancelled"; +} + /** * The note a restart continuation of `source` carries, and whether the turn it * continues had settled (the note is then the whole prompt). A continuation cut diff --git a/apps/server/src/orchestration-v2/RestartContinuation.test.ts b/apps/server/src/orchestration-v2/RestartContinuation.test.ts index eb03389adc62646b23f044f53875072294a8e419..9a9408d3a55b9e3e5dfe9c666ed80ef63b8bc6f5 100644 --- a/apps/server/src/orchestration-v2/RestartContinuation.test.ts +++ b/apps/server/src/orchestration-v2/RestartContinuation.test.ts @@ -223,6 +223,48 @@ it.effect.each(["completed", "waiting", "cancelled"] as const)( }), ); +it.effect.each(["completed", "waiting", "cancelled"] as const)( + "wakes a %s run whose monitor the restart cancelled, with the note as its prompt", + (status) => + Effect.gen(function* () { + const base = makeProjection(); + const work = [ + { kind: "monitor" as const, label: "Watch CI run 42" }, + { kind: "shell" as const, label: "vp run dev" }, + ]; + const projection = { + ...base, + runs: [{ ...base.runs[0]!, status, restartCancelledBackgroundWork: work }], + providerTurns: [{ ...base.providerTurns[0]!, status: "completed" }], + } as unknown as OrchestrationV2ThreadProjection; + const commands: Parameters< + ThreadManagementService.ThreadManagementService["Service"]["dispatch"] + >[0][] = []; + yield* continueRestartedRun({ threadId, sourceRunId: runId }).pipe( + Effect.provide( + Layer.merge( + Layer.mock(ThreadManagementService.ThreadManagementService)({ + getThreadRecords: () => Effect.succeed(projection), + recoverDelegatedTask: () => Effect.void, + dispatch: (command) => { + commands.push(command); + return Effect.succeed({} as never); + }, + }), + ServerSettings.layerTest({ continueThreadsAfterServerUpdate: true }), + ), + ), + ); + assert.lengthOf(commands, 1); + const command = commands[0]!; + assert.equal(command.type, "message.dispatch"); + if (command.type !== "message.dispatch") return; + assert.equal(command.restartContinuationOfRunId, runId); + assert.include(command.text, "- monitor: Watch CI run 42"); + assert.include(command.text, "- shell: vp run dev"); + }), +); + it.effect("does not continue a failed run that lost background work", () => Effect.gen(function* () { const base = makeProjection(); @@ -319,6 +361,62 @@ it.effect.each([ }), ); +it.effect("lets a cut blocking delegate_task wake its parent like an async task", () => + Effect.gen(function* () { + const base = makeProjection(); + const task = (id: string, completionWake: "settled_only" | "always", status: string) => ({ + id: NodeId.make(id), + threadId, + runId, + origin: "app_owned", + childThreadId: ThreadId.make(`thread:${id}`), + driver, + providerInstanceId: instanceId, + completionWake, + status, + }); + const projection = { + ...base, + subagents: [ + task("task:waited", "settled_only", "running"), + task("task:async", "always", "running"), + task("task:finished", "settled_only", "completed"), + ], + } as unknown as OrchestrationV2ThreadProjection; + let committed: Parameters[0] | undefined; + const recovery = yield* ProviderRuntimeRecovery.make.pipe( + Effect.provide( + Layer.mergeAll( + ServerSettings.layerTest({ continueThreadsAfterServerUpdate: true }), + Layer.mock(ProjectionStore.ProjectionStoreV2)({ + getRecoveryThreadIds: () => Effect.succeed([threadId]), + getRuntimeRecoveryProjection: () => Effect.succeed(projection), + }), + Layer.mock(EventSink.EventSinkV2)({ + commitCommand: (input) => { + committed = input; + return Effect.succeed({ committed: true, cancelledEffectCount: 0 } as never); + }, + }), + IdAllocator.layer, + Layer.mock(EffectWorker.OrchestrationEffectWorkerV2)({}), + Layer.mock(EffectOutbox.EffectOutboxV2)({ + reconcileAfterProcessLoss: Effect.succeed({ requeued: 0, cancelled: 0 }), + }), + ), + ), + ); + yield* recovery.reconcile("startup"); + const upgraded = (committed?.events ?? []).flatMap((event) => + event.type === "subagent.updated" ? [event.payload] : [], + ); + assert.deepEqual( + upgraded.map((row) => ({ id: row.id, completionWake: row.completionWake })), + [{ id: NodeId.make("task:waited"), completionWake: "always" }], + ); + }), +); + it.effect("does not duplicate delivery and yields to newer user work or opt-out", () => Effect.gen(function* () { let projection = makeProjection(); @@ -424,7 +522,7 @@ it.effect("prepares no continuation for background work another provider thread ); yield* recovery.prepareForShutdown; assert.lengthOf(writes, 0); - // Work on the settled run's own provider thread must not wake it either. + // The same work on the settled run's own provider thread does wake it. const ownWork = { ...projection, turnItems: [{ ...projection.turnItems[0]!, providerThreadId }], @@ -451,18 +549,54 @@ it.effect("prepares no continuation for background work another provider thread ), ); yield* ownRecovery.prepareForShutdown; - assert.lengthOf(writes, 0); + assert.lengthOf(writes, 1); + assert.deepEqual(writes[0]!.effects[0]?.request, { + type: "provider-runtime.continue", + sourceRunId: runId, + }); }), ); +const backgroundSubagent = { + turnItem: { + id: "turn-item:background-subagent", + runId, + nodeId: null, + providerThreadId, + providerInstanceId: instanceId, + type: "subagent", + subagentId: "subagent:background", + title: "Background reviewer", + status: "running", + }, + work: { + kind: "subagent" as const, + label: "Background reviewer", + id: "turn-item:background-subagent", + }, +}; +const devServer = { + turnItem: { + id: "turn-item:dev-server", + runId, + nodeId: null, + providerThreadId, + providerInstanceId: instanceId, + type: "command_execution", + input: "vp run dev", + status: "running", + }, + work: { kind: "shell" as const, label: "vp run dev", id: "turn-item:dev-server" }, +}; + it.effect.each([ - ["completed", scratchProjectId], - ["completed", ProjectId.make("restart-project")], - ["waiting", scratchProjectId], - ["waiting", ProjectId.make("restart-project")], + ["completed", scratchProjectId, backgroundSubagent, true], + ["waiting", ProjectId.make("restart-project"), backgroundSubagent, true], + ["completed", ProjectId.make("restart-project"), devServer, false], + ["waiting", scratchProjectId, devServer, false], ] as const)( - "cleans up background work without waking a %s run in project %s", - ([status, projectId]) => + "cleans up background work of a %s run in project %s and wakes it only for awaited work", + ([status, projectId, background, wakes]) => Effect.gen(function* () { const base = makeProjection(); const projection = { @@ -470,29 +604,20 @@ it.effect.each([ thread: { ...base.thread, projectId }, runs: [{ ...base.runs[0]!, status }], providerTurns: [{ ...base.providerTurns[0]!, status: "completed" }], - turnItems: [ - { - id: "turn-item:background-subagent", - runId, - nodeId: null, - providerThreadId, - providerInstanceId: instanceId, - type: "subagent", - subagentId: "subagent:background", - title: "Background reviewer", - status: "running", - }, - ], - subagents: [ - { - id: "subagent:background", - runId, - origin: "provider_native", - driver, - providerInstanceId: instanceId, - status: "running", - }, - ], + turnItems: [background.turnItem], + subagents: + background === backgroundSubagent + ? [ + { + id: "subagent:background", + runId, + origin: "provider_native", + driver, + providerInstanceId: instanceId, + status: "running", + }, + ] + : [], } as unknown as OrchestrationV2ThreadProjection; for (const trigger of ["startup", "shutdown"] as const) { const commits: Parameters[0][] = []; @@ -530,25 +655,51 @@ it.effect.each([ ); if (trigger === "shutdown") yield* recovery.prepareForShutdown; yield* recovery.reconcile(trigger); - assert.lengthOf(writes, 0); + // Shutdown records the wake before it cancels the work; a startup + // after a kill records it with the cancellation. + const recorded = [...writes.flatMap((write) => write.effects), ...commits[0]!.effects]; + assert.deepEqual( + recorded.map((effect) => effect.request), + wakes ? [{ type: "provider-runtime.continue", sourceRunId: runId }] : [], + ); assert.lengthOf(commits, 1); - assert.lengthOf(commits[0]!.effects, 0); const events = commits[0]!.events; - for (const type of ["turn-item.updated", "subagent.updated"] as const) - assert.isTrue( - events.some((event) => event.type === type && event.payload.status === "cancelled"), - ); + assert.isTrue( + events.some( + (event) => event.type === "turn-item.updated" && event.payload.status === "cancelled", + ), + ); const note = events.find((event) => event.type === "run.background-work-cancelled"); assert.equal(note?.runId, runId); - assert.deepEqual(note?.payload.restartCancelledBackgroundWork, [ - { kind: "subagent", label: "Background reviewer", id: "turn-item:background-subagent" }, - ]); + assert.deepEqual(note?.payload.restartCancelledBackgroundWork, [background.work]); if (status === "completed") assert.isFalse(events.some((event) => event.type === "run.updated")); } }), ); +it("does not wake a settled run for work left by a session that already stopped", () => { + const base = makeProjection(); + const projection = { + ...base, + runs: [{ ...base.runs[0]!, status: "completed" }], + providerThreads: [{ ...base.providerThreads[0]!, status: "idle" }], + turnItems: [backgroundSubagent.turnItem], + } as unknown as OrchestrationV2ThreadProjection; + assert.equal(restartContinuationRun(projection, now)?.id, runId); + for (const status of ["stopped", "error"]) + assert.isUndefined( + restartContinuationRun( + { + ...projection, + providerSessions: [{ ...projection.providerSessions[0]!, status }], + } as OrchestrationV2ThreadProjection, + now, + ), + status, + ); +}); + it.effect("does not cancel or resume a run that completes while shutdown intent commits", () => Effect.gen(function* () { let projection = makeProjection(); diff --git a/apps/server/src/orchestration-v2/RestartContinuation.ts b/apps/server/src/orchestration-v2/RestartContinuation.ts index cb435d10435086f9f34e5e180144c7c84f1513f9..99b2124718a65f560f7e621f5d0b782f7f106c5a 100644 --- a/apps/server/src/orchestration-v2/RestartContinuation.ts +++ b/apps/server/src/orchestration-v2/RestartContinuation.ts @@ -16,6 +16,10 @@ import * as ServerSettings from "../serverSettings.ts"; import { isNativeMaintenanceCommand } from "./Orchestrator.ts"; import * as ThreadManagementService from "./ThreadManagementService.ts"; import { + cancelledRosterTaskWork, + cancelledTurnItemWork, + isAwaitedWork, + isRestartContinuationSource, isRestartNoteSource, restartCancelledBackgroundWorkNote, restartContinuationNote, @@ -29,7 +33,9 @@ const MAX_CONTINUATION_CHAIN = 3; /** * Whether a restart must leave this run for the user: the thread is put away, * the provider turn already ended, or the user owes an answer. Checked when - * intent is recorded and again when the continuation is delivered. + * intent is recorded and again when the continuation is delivered. A + * `settled` run is woken for the awaited work it lost, so its ended provider + * turn does not block it. */ function restartContinuationBlocked( projection: Pick< @@ -38,6 +44,7 @@ function restartContinuationBlocked( >, source: OrchestrationV2Run, now: DateTime.Utc, + settled: boolean, ): boolean { const { thread } = projection; if ( @@ -49,6 +56,7 @@ function restartContinuationBlocked( return true; const turn = latestProviderTurnForAttempt(projection.providerTurns, source.activeAttemptId); if ( + !settled && turn !== undefined && (turn.status === "completed" || turn.status === "failed" || turn.status === "interrupted") ) @@ -82,8 +90,38 @@ function restartContinuationBlocked( } /** - * Resume only an unfinished root turn. Leftover background work is cleaned up - * separately and reported on the next user turn; it must not wake a settled run. + * Open awaited work (see `isAwaitedWork`) on a provider thread. App-owned + * delegations are not counted: their own threads continue and report back. + */ +function hasOpenAwaitedWork( + projection: Pick, + providerThread: ProjectionRuntimeRecoveryState["providerThreads"][number], +): boolean { + const openItem = projection.turnItems.some((item) => { + if (item.status !== "pending" && item.status !== "running" && item.status !== "waiting") + return false; + if (item.type === "subagent" && item.origin === "app_owned" && item.childThreadId !== null) + return false; + const work = cancelledTurnItemWork(item); + const providerThreadId = + item.providerThreadId ?? + projection.runs.find((run) => run.id === item.runId)?.providerThreadId; + return work !== undefined && isAwaitedWork(work) && providerThreadId === providerThread.id; + }); + return ( + openItem || + (providerThread.ownerNodeId === null && + (providerThread.pendingBackgroundTasks ?? []).some((task) => + isAwaitedWork(cancelledRosterTaskWork(task)), + )) + ); +} + +/** + * The run a restart continues: an unfinished root turn, or a settled one + * waiting on awaited work that the restart cancels. A settled run needs a + * live session, so stale records from an older crash never wake it. Other + * leftover work is cleaned up and reported on the next user turn. */ export function restartContinuationRun( projection: Pick< @@ -109,9 +147,10 @@ export function restartContinuationRun( if (!run) return; const preparedContinuation = run.status === "starting" && run.restartContinuationOfRunId !== undefined; - if (run.status !== "running" && !preparedContinuation) return; - if (restartContinuationBlocked(projection, run, now)) return; - const liveTurnRequired = !preparedContinuation; + const settled = run.status === "completed" || run.status === "waiting"; + if (run.status !== "running" && !preparedContinuation && !settled) return; + if (restartContinuationBlocked(projection, run, now, settled)) return; + const liveTurnRequired = !preparedContinuation && !settled; if (projection.thread.providerInstanceId !== run.providerInstanceId) return; const providerThread = projection.providerThreads.find( (thread) => thread.id === run.providerThreadId, @@ -138,9 +177,10 @@ export function restartContinuationRun( session === undefined || session.providerInstanceId !== run.providerInstanceId || session.driver !== providerThread.driver || - (liveTurnRequired && (session.status === "stopped" || session.status === "error")) + (!preparedContinuation && (session.status === "stopped" || session.status === "error")) ) return; + if (settled) return hasOpenAwaitedWork(projection, providerThread) ? run : undefined; if ( liveTurnRequired && !projection.providerTurns.some( @@ -173,13 +213,17 @@ export const continueRestartedRun = Effect.fn("RestartContinuation.continueResta return; if (projection.messages.some((message) => message.id === messageId)) return; const source = projection.runs.find((run) => run.id === input.sourceRunId); - // Pending effects from older versions may target settled background work, - // including waiting runs that reconciliation subsequently cancelled. + // Reconciliation cancels a waiting run it cannot checkpoint, so a settled + // source can be cancelled too. It continues only for awaited work. if ( !source || - source.status !== "cancelled" || - isRestartNoteSource(source, projection.providerTurns) || - restartContinuationBlocked(projection, source, yield* DateTime.now) + !isRestartContinuationSource(source, projection.providerTurns) || + restartContinuationBlocked( + projection, + source, + yield* DateTime.now, + isRestartNoteSource(source, projection.providerTurns), + ) ) return; // A user submission after reconciliation takes precedence over an automatic diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.test.ts b/apps/server/src/orchestration-v2/ThreadManagementService.test.ts index fb6e9af45424cfb0e118176bf78f9918d0413320..836d67380d431780745077259a7a3dc324b0765a 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.test.ts @@ -597,6 +597,7 @@ it.effect.each([ Layer.provide( Layer.mock(Orchestrator.OrchestratorV2)({ getThreadEventSequence: () => Effect.succeed(0), + restartContinuationPending: () => Effect.succeed(false), // No run update arrives, so the timeout path runs while a final // projection read can still observe a terminal run. streamStoredEventsFrom: () => @@ -666,6 +667,7 @@ it.effect("waitForThread reads the run again only when the run updates", () => Layer.provide( Layer.mock(Orchestrator.OrchestratorV2)({ getThreadEventSequence: () => Effect.succeed(0), + restartContinuationPending: () => Effect.succeed(false), // Only the run.updated stream carries events in this test. streamStoredEventsFrom: (input) => input?.eventType === "run.updated" @@ -705,6 +707,126 @@ it.effect("waitForThread reads the run again only when the run updates", () => }), ); +it.effect("waitForThread follows a restart continuation instead of the cut run", () => + Effect.gen(function* () { + const projectId = ProjectId.make("project:thread-management:wait-restart"); + const threadId = ThreadId.make("thread:thread-management:wait-restart"); + const runId = RunId.make("run:thread-management:wait-restart"); + const continuationId = RunId.make("run:thread-management:wait-restart:2"); + const subscribed = yield* Deferred.make(); + const events = yield* Queue.unbounded(); + // The restart cut the run; its continuation effect has not started yet. + let runs: Array<{ + id: RunId; + status: OrchestrationV2Run["status"]; + restartContinuationOfRunId?: RunId; + }> = [{ id: runId, status: "cancelled" }]; + let pending = true; + const stored = (sequence: number, event: object) => + ({ sequence, event: { threadId, ...event } }) as unknown as OrchestrationV2StoredEvent; + const layerTest = ThreadManagementService.layer.pipe( + Layer.provide(Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({})), + Layer.provide( + Layer.mock(Orchestrator.OrchestratorV2)({ + getThreadEventSequence: () => Effect.succeed(0), + restartContinuationPending: (run) => Effect.succeed(pending && run.id === runId), + awaitRestartContinuation: () => Effect.never, + streamStoredEventsFrom: (input) => + input?.eventType === "run.updated" + ? Stream.fromEffect(Deferred.succeed(subscribed, undefined)).pipe( + Stream.drain, + Stream.concat(Stream.fromQueue(events)), + ) + : Stream.never, + getThreadRecords: () => + Effect.sync( + () => + ({ + thread: { id: threadId, projectId, deletedAt: null }, + runs, + }) as unknown as OrchestrationV2ThreadProjection, + ), + }), + ), + Layer.provide(ServerSettings.layerTest()), + Layer.provide(layerReceiptsTest), + ); + const service = yield* ThreadManagementService.ThreadManagementService.pipe( + Effect.provide(layerTest), + ); + const fiber = yield* service + .waitForThread({ projectId, threadId, runId, timeoutMs: 60 * 60 * 1_000 }) + .pipe(Effect.forkChild); + yield* Deferred.await(subscribed); + const continuation = { id: continuationId, restartContinuationOfRunId: runId }; + runs = [...runs, { ...continuation, status: "running" }]; + pending = false; + yield* Queue.offer( + events, + stored(1, { type: "run.updated", payload: { ...continuation, status: "running" } }), + ); + runs = [runs[0]!, { ...continuation, status: "completed" }]; + yield* Queue.offer( + events, + stored(2, { type: "run.updated", payload: { ...continuation, status: "completed" } }), + ); + const result = yield* Fiber.join(fiber); + + expect(result).toMatchObject({ + timedOut: false, + run: { id: continuationId, status: "completed" }, + }); + }), +); + +it.effect("waitForThread returns the cut run once its continuation is declined", () => + Effect.gen(function* () { + const projectId = ProjectId.make("project:thread-management:wait-declined"); + const threadId = ThreadId.make("thread:thread-management:wait-declined"); + const runId = RunId.make("run:thread-management:wait-declined"); + const subscribed = yield* Deferred.make(); + const declined = yield* Deferred.make(); + let pending = true; + const layerTest = ThreadManagementService.layer.pipe( + Layer.provide(Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({})), + Layer.provide( + Layer.mock(Orchestrator.OrchestratorV2)({ + getThreadEventSequence: () => Effect.succeed(0), + restartContinuationPending: () => Effect.succeed(pending), + awaitRestartContinuation: () => Deferred.await(declined), + // Declining records no thread event. + streamStoredEventsFrom: (input) => + input?.eventType === "run.updated" + ? Stream.fromEffect(Deferred.succeed(subscribed, undefined)).pipe( + Stream.drain, + Stream.concat(Stream.never), + ) + : Stream.never, + getThreadRecords: () => + Effect.succeed({ + thread: { id: threadId, projectId, deletedAt: null }, + runs: [{ id: runId, status: "cancelled" }], + } as unknown as OrchestrationV2ThreadProjection), + }), + ), + Layer.provide(ServerSettings.layerTest()), + Layer.provide(layerReceiptsTest), + ); + const service = yield* ThreadManagementService.ThreadManagementService.pipe( + Effect.provide(layerTest), + ); + const fiber = yield* service + .waitForThread({ projectId, threadId, runId, timeoutMs: 60 * 60 * 1_000 }) + .pipe(Effect.forkChild); + yield* Deferred.await(subscribed); + pending = false; + yield* Deferred.succeed(declined, undefined); + const result = yield* Fiber.join(fiber); + + expect(result).toMatchObject({ timedOut: false, run: { id: runId, status: "cancelled" } }); + }), +); + it.effect.each([ "running", "missing", @@ -938,6 +1060,7 @@ it.effect.each([ Layer.provide( Layer.mock(Orchestrator.OrchestratorV2)({ getThreadEventSequence: () => Effect.succeed(0), + restartContinuationPending: () => Effect.succeed(false), getThreadRecords: () => Effect.succeed({ thread: { id: threadId, projectId, deletedAt: null }, diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.ts b/apps/server/src/orchestration-v2/ThreadManagementService.ts index a3d6f7b8526097d18540fd92f88c648fcd8c7c2a..5ef8d2366b71235d930dc2cadf54541c9b56ebf8 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.ts @@ -486,6 +486,20 @@ export function latestRun( return projection.runs.toSorted((left, right) => right.ordinal - left.ordinal)[0]; } +/** The newest run in `run`'s restart continuation chain, or `run` itself. */ +function latestRestartContinuation( + runs: ReadonlyArray, + run: OrchestrationV2Run, +): OrchestrationV2Run { + let current = run; + for (let hops = 0; hops < runs.length; hops++) { + const next = runs.find((candidate) => candidate.restartContinuationOfRunId === current.id); + if (next === undefined) break; + current = next; + } + return current; +} + export function latestActiveRun( projection: Pick, ): OrchestrationV2Run | undefined { @@ -844,65 +858,76 @@ const make = Effect.gen(function* () { if (selectedRun === undefined) { return { threadId: input.threadId, run: null, timedOut: false }; } - if (isTerminalRunStatus(selectedRun.status)) { - return { threadId: input.threadId, run: selectedRun, timedOut: false }; + // A restart cuts a run and continues it as a new run. Follow that chain, + // so a cut run is not the answer while its continuation is pending. + const outcome = (runs: ReadonlyArray) => + Effect.gen(function* () { + const current = runs.find((candidate) => candidate.id === selectedRun.id); + if (current === undefined) { + return yield* new ThreadManagementRunNotFoundError({ + threadId: input.threadId, + runId: selectedRun.id, + }); + } + const run = latestRestartContinuation(runs, current); + const pending = + isTerminalRunStatus(run.status) && + (yield* orchestrator.restartContinuationPending(run)); + return { run, pending, settled: isTerminalRunStatus(run.status) && !pending }; + }); + const initial = yield* outcome(target.runs); + if (initial.settled) { + return { threadId: input.threadId, run: initial.run, timedOut: false }; } - // Re-read the run only when the thread records a run update or its - // deletion, instead of polling the projection for up to an hour. Each - // stream keeps one event type, so transcript events never fill its buffer. + // Re-read the runs only when the thread records an update to the waited + // run or a continuation, or its deletion, instead of polling the + // projection for up to an hour. Each stream keeps one event type, so + // transcript events never fill its buffer. A declined continuation + // records no event, so its effect settling wakes the wait too. const wait = Stream.merge( - orchestrator.streamStoredEventsFrom({ - threadId: input.threadId, - afterSequence, - eventType: "run.updated", - }), - orchestrator.streamStoredEventsFrom({ - threadId: input.threadId, - afterSequence, - eventType: "thread.deleted", - }), + Stream.merge( + orchestrator.streamStoredEventsFrom({ + threadId: input.threadId, + afterSequence, + eventType: "run.updated", + }), + orchestrator.streamStoredEventsFrom({ + threadId: input.threadId, + afterSequence, + eventType: "thread.deleted", + }), + ).pipe( + Stream.mapError(loadError), + Stream.filter( + (stored) => + stored.event.type !== "run.updated" || + stored.event.payload.id === selectedRun.id || + stored.event.payload.restartContinuationOfRunId !== undefined, + ), + ), + initial.pending + ? Stream.fromEffect(orchestrator.awaitRestartContinuation(initial.run)) + : Stream.empty, ).pipe( - Stream.mapError(loadError), - Stream.filter( - (stored) => - stored.event.type !== "run.updated" || stored.event.payload.id === selectedRun.id, - ), Stream.mapEffect(() => - getProjectThreadRecords(input, ["runs"], { runIds: [selectedRun.id] }).pipe( - Effect.flatMap((current) => { - const run = current.runs.find((candidate) => candidate.id === selectedRun.id); - return run === undefined - ? Effect.fail( - new ThreadManagementRunNotFoundError({ - threadId: input.threadId, - runId: selectedRun.id, - }), - ) - : Effect.succeed(run); - }), + getProjectThreadRecords(input, ["runs"]).pipe( + Effect.flatMap((current) => outcome(current.runs)), ), ), - Stream.filter((run) => isTerminalRunStatus(run.status)), + Stream.filter((current) => current.settled), Stream.runHead, Effect.timeoutOption(Duration.millis(Math.max(1, input.timeoutMs))), ); const waited = Option.flatten(yield* wait); if (Option.isSome(waited)) { - return { threadId: input.threadId, run: waited.value, timedOut: false }; - } - const current = yield* getProjectThreadRecords(input, ["runs"], { runIds: [selectedRun.id] }); - const run = current.runs.find((candidate) => candidate.id === selectedRun.id); - if (run === undefined) { - return yield* new ThreadManagementRunNotFoundError({ - threadId: input.threadId, - runId: selectedRun.id, - }); + return { threadId: input.threadId, run: waited.value.run, timedOut: false }; } // The run may have reached a terminal status while the timeout was // winning the race; the final projection read decides what actually // happened, so only a still-active run counts as timed out. - return { threadId: input.threadId, run, timedOut: !isTerminalRunStatus(run.status) }; + const final = yield* outcome((yield* getProjectThreadRecords(input, ["runs"])).runs); + return { threadId: input.threadId, run: final.run, timedOut: !final.settled }; }); const settleAfterRun: ThreadManagementServiceShape["settleAfterRun"] = Effect.fn( diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index 9bd7f38ddb8bb5e7e9d23e371f1157b06e74a59b..d86a042e173dcc3a0a26fe7a97c88c1a31d50e8f 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -3251,18 +3251,28 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { }), ); - it.effect.each(["failed", "completed", "waiting", "cancelled"] as const)( - "does not admit a restart continuation of a %s run with only lost background work", - (status) => + it.effect.each([ + ["failed", "shell", false], + ["completed", "shell", false], + ["waiting", "shell", false], + ["cancelled", "shell", false], + // A settled run lost work it was waiting on, so nothing else wakes it. + ["failed", "monitor", false], + ["completed", "monitor", true], + ["waiting", "monitor", true], + ["cancelled", "monitor", true], + ] as const)( + "admits a restart continuation of a %s run that lost a %s: %s", + ([status, kind, admitted]) => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; const eventSink = yield* EventSink.EventSinkV2; - const threadId = ThreadId.make(`runtime-layer-restart-${status}-source`); + const threadId = ThreadId.make(`runtime-layer-restart-${status}-${kind}-source`); yield* orchestrator.dispatch({ type: "thread.create", createdBy: "user", creationSource: "web", - commandId: CommandId.make(`restart-${status}-create`), + commandId: CommandId.make(`restart-${status}-${kind}-create`), threadId, projectId: ProjectId.make("restart-project"), title: "Restart", @@ -3276,9 +3286,9 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { type: "message.dispatch", createdBy: "user", creationSource: "web", - commandId: CommandId.make(`restart-${status}-user-message`), + commandId: CommandId.make(`restart-${status}-${kind}-user-message`), threadId, - messageId: MessageId.make(`restart-${status}-user-message`), + messageId: MessageId.make(`restart-${status}-${kind}-user-message`), text: "Original work", attachments: [], modelSelection, @@ -3287,13 +3297,13 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { const original = (yield* orchestrator.getThreadProjection(threadId)).runs[0]!; const now = yield* DateTime.now; yield* eventSink.commitCommand({ - commandId: CommandId.make(`restart-${status}-reconcile`), + commandId: CommandId.make(`restart-${status}-${kind}-reconcile`), threadId, commandType: "provider-runtime.reconcile", acceptedAt: now, events: [ { - id: EventId.make(`restart-${status}-run`), + id: EventId.make(`restart-${status}-${kind}-run`), type: "run.updated", threadId, runId: original.id, @@ -3301,14 +3311,14 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { payload: { ...original, status, completedAt: now }, }, { - id: EventId.make(`restart-${status}-work`), + id: EventId.make(`restart-${status}-${kind}-work`), type: "run.background-work-cancelled", threadId, runId: original.id, occurredAt: now, payload: { runId: original.id, - restartCancelledBackgroundWork: [{ kind: "shell", label: "sleep 25" }], + restartCancelledBackgroundWork: [{ kind, label: "sleep 25" }], }, }, ], @@ -3318,9 +3328,9 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { type: "message.dispatch", createdBy: "agent", creationSource: "server", - commandId: CommandId.make(`restart-${status}-continuation`), + commandId: CommandId.make(`restart-${status}-${kind}-continuation`), threadId, - messageId: MessageId.make(`restart-${status}-continuation`), + messageId: MessageId.make(`restart-${status}-${kind}-continuation`), text: "Note: the T3 server restarted.", attachments: [], modelSelection, @@ -3328,8 +3338,11 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { restartContinuationOfRunId: original.id, }); const projection = yield* orchestrator.getThreadProjection(threadId); - assert.lengthOf(projection.runs, 1); assert.equal(projection.runs[0]?.status, status); + assert.deepEqual( + projection.runs.slice(1).map((run) => run.restartContinuationOfRunId), + admitted ? [original.id] : [], + ); }), ); diff --git a/apps/server/src/process/agentScope.test.ts b/apps/server/src/process/agentScope.test.ts index 537bcc646748be12fac5b38edef72ffa896e295a..f29cd867244af2e49915fea7f7e01e444b874521 100644 --- a/apps/server/src/process/agentScope.test.ts +++ b/apps/server/src/process/agentScope.test.ts @@ -7,10 +7,18 @@ import * as NodeServices from "@effect/platform-node/NodeServices"; import { describe, expect, it } from "@effect/vitest"; import * as HostProcess from "@t3tools/shared/HostProcess"; import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; import { ChildProcessSpawner } from "effect/process"; +import * as ServerConfig from "../config.ts"; import * as ProcessRunner from "../processRunner.ts"; -import { agentSliceMemoryLimits, classifyScope, make, type ScopeState } from "./agentScope.ts"; +import { + agentSliceMemoryLimits, + classifyScope, + make, + ownedAgentSlice, + type ScopeState, +} from "./agentScope.ts"; const GiB = 1024 ** 3; @@ -55,11 +63,14 @@ describe("agentSliceMemoryLimits", () => { }); // Runs the real service against a fake user manager. `units` holds the -// ActiveState and Result that `systemctl show` reports for each scope. +// ActiveState and Result that `systemctl show` reports for each scope or +// slice, and `calls` records every command in order. const withFakeSystemd = ( body: (input: { readonly scope: Effect.Success; readonly units: Map; + readonly calls: ReadonlyArray>; + readonly stateDir: string; }) => Effect.Effect, ) => Effect.gen(function* () { @@ -68,6 +79,7 @@ const withFakeSystemd = ( NodeFS.writeFileSync(NodePath.join(bin, name), "", { mode: 0o755 }); } const units = new Map(); + const calls: Array> = []; const output = (stdout: string): ProcessRunner.ProcessRunOutput => ({ stdout, stderr: "", @@ -81,6 +93,7 @@ const withFakeSystemd = ( const runner = ProcessRunner.ProcessRunner.of({ run: ({ args }) => Effect.sync(() => { + calls.push(args); const [, verb, unit = ""] = args; if (verb === "reset-failed") units.delete(unit); if (verb !== "show") return output(""); @@ -97,14 +110,25 @@ const withFakeSystemd = ( Effect.provideService(HostProcess.Platform, "linux"), Effect.provideService(HostProcess.Environment, { PATH: bin, XDG_RUNTIME_DIR: "/run/user/1" }), ); - return yield* body({ scope, units }).pipe( + const { stateDir } = yield* ServerConfig.ServerConfig; + return yield* body({ scope, units, calls, stateDir }).pipe( Effect.ensuring(Effect.sync(() => NodeFS.rmSync(bin, { recursive: true, force: true }))), ); - }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)); + }).pipe( + Effect.scoped, + Effect.provide( + ServerConfig.layerTest(process.cwd(), { prefix: "t3-agent-scope-" }).pipe( + Layer.provideMerge(NodeServices.layer), + ), + ), + ); const unitOf = (launch: { readonly args: ReadonlyArray }) => launch.args.find((arg) => arg.startsWith("--unit="))?.slice("--unit=".length) ?? ""; +const sliceOf = (launch: { readonly args: ReadonlyArray }) => + launch.args.find((arg) => arg.startsWith("--slice="))?.slice("--slice=".length) ?? ""; + describe("AgentScope service", () => { it.effect("labels a thread from its newest scope until its next session clears it", () => withFakeSystemd(({ scope, units }) => @@ -142,4 +166,37 @@ describe("AgentScope service", () => { }), ), ); + + it.effect("stops agents a killed server left before the first launch", () => + withFakeSystemd(({ scope, units, calls, stateDir }) => + Effect.gen(function* () { + units.set(ownedAgentSlice(stateDir), { activeState: "active", result: "success" }); + const agent = yield* scope.wrap({ command: "agent", args: [], name: "t", threadId: "a" }); + const stops = calls.filter(([, verb]) => verb === "stop"); + expect(stops).toEqual([["--user", "stop", ownedAgentSlice(stateDir)]]); + // Agents run in this server's own slice; terminals stay outside it. + expect(sliceOf(agent)).toBe(ownedAgentSlice(stateDir)); + // An agent that ignores SIGTERM is killed well inside the stop's wait. + expect(agent.args).toContain("--property=TimeoutStopSec=10s"); + const terminal = yield* scope.wrap({ + command: "agent", + args: [], + name: "terminal", + survivesRestart: true, + }); + expect(sliceOf(terminal)).toBe("app-t3code-agents.slice"); + // Only the first launch looks for leftovers. + expect(calls.filter(([, verb]) => verb === "stop")).toHaveLength(1); + }), + ), + ); + + it.effect("leaves a quiet slice alone", () => + withFakeSystemd(({ scope, calls }) => + Effect.gen(function* () { + yield* scope.wrap({ command: "agent", args: [], name: "t", threadId: "a" }); + expect(calls.filter(([, verb]) => verb === "stop")).toEqual([]); + }), + ), + ); }); diff --git a/apps/server/src/process/agentScope.ts b/apps/server/src/process/agentScope.ts index c1c470b3be540a654684e8bbb930f0bec5435c8b..bfe013827301f73a4864a78bf29a40398865a5ea 100644 --- a/apps/server/src/process/agentScope.ts +++ b/apps/server/src/process/agentScope.ts @@ -15,6 +15,7 @@ import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Path from "effect/Path"; +import * as ServerConfig from "../config.ts"; import * as ProcessRunner from "../processRunner.ts"; /** @@ -24,6 +25,18 @@ import * as ProcessRunner from "../processRunner.ts"; */ const AGENT_SLICE = "app-t3code-agents.slice"; +/** + * The child slice for one server's agents, named after its state directory. + * Only one server uses a state directory at a time, so an agent in this slice + * when the server starts was left by an earlier server on the same state, + * for example one the launcher killed during an update. Other servers on the + * machine, such as dev servers, have their own slice. + */ +export function ownedAgentSlice(stateDir: string): string { + const hash = NodeCrypto.createHash("sha256").update(stateDir).digest("hex").slice(0, 12); + return `app-t3code-agents-${hash}.slice`; +} + const GiB = 1024 ** 3; /** @@ -84,9 +97,12 @@ function parseCgroupCounter(text: string, key: string): number | undefined { return match === null ? undefined : Number(match[1]); } +// A stop sends SIGTERM, then SIGKILL after this long. systemd's 90 s default +// would outlast the wait for a previous server's agents below. +const STOP_TIMEOUT_OPTION = "--property=TimeoutStopSec=10s"; const STOPPING_POLL_INTERVAL = "100 millis"; const STOPPING_POLL_ATTEMPTS = 20; -// A stopping scope can wait out TimeoutStopSec (90 s by default) for a stuck process. +// A stopping scope can wait out its TimeoutStopSec for a stuck process. const CLEAR_FAILED_INTERVAL = "1 second"; const CLEAR_FAILED_ATTEMPTS = 120; const MAX_TRACKED_THREADS = 500; @@ -99,6 +115,8 @@ export const make = Effect.gen(function* () { const fileSystem = yield* FileSystem.FileSystem; const path = yield* Path.Path; const hostEnvironment = yield* HostProcess.Environment; + const { stateDir } = yield* ServerConfig.ServerConfig; + const ownedSlice = ownedAgentSlice(stateDir); const layerScope = yield* Effect.scope; const resolve = (command: string, env: NodeJS.ProcessEnv) => @@ -111,8 +129,45 @@ export const make = Effect.gen(function* () { const run = (command: string, args: ReadonlyArray) => runner.run({ command, args, timeout: "5 seconds" }).pipe(Effect.option); - // Runs once, on the first agent launch. It proves this process can reach a - // systemd user manager, creates the agents slice, and sets its limits. + const readCgroupFile = (controlGroup: string, file: string) => + fileSystem + .readFileString(path.join("/sys/fs/cgroup", controlGroup, file)) + .pipe(Effect.orElseSucceed(() => "")); + + // Old agents keep their scopes when the server is killed, and the restarted + // server resumes their sessions. Stop them first, so two copies of one + // session never work in the same worktree. + const stopLeftoverAgents = (systemctl: string) => + Effect.gen(function* () { + const shown = yield* run(systemctl, [ + "--user", + "show", + ownedSlice, + "--property=ActiveState,ControlGroup", + ]); + if (shown._tag === "None") return; + const values = parseKeyValues(shown.value.stdout); + if (values.get("ActiveState") !== "active") return; + // An active slice can be empty once its last agent exited. + const controlGroup = values.get("ControlGroup") ?? ""; + const events = controlGroup ? yield* readCgroupFile(controlGroup, "cgroup.events") : ""; + if (parseCgroupCounter(events, "populated") === 0) return; + yield* Effect.logWarning("Stopping agents left by a previous server", { slice: ownedSlice }); + // Agents get SIGTERM, then SIGKILL after their stop timeout. + const stopped = yield* runner + .run({ command: systemctl, args: ["--user", "stop", ownedSlice], timeout: "30 seconds" }) + .pipe(Effect.option); + if (stopped._tag === "None" || stopped.value.code !== 0) { + yield* Effect.logWarning("Could not stop agents left by a previous server", { + slice: ownedSlice, + stderr: stopped._tag === "Some" ? stopped.value.stderr.trim() : undefined, + }); + } + }); + + // Runs once, on the first agent or terminal launch. It proves this process + // can reach a systemd user manager, stops agents a previous server left, + // creates the agents slice, and sets its limits. const systemd = yield* Effect.cached( Effect.gen(function* () { const runtimeDir = hostEnvironment.XDG_RUNTIME_DIR; @@ -120,6 +175,7 @@ export const make = Effect.gen(function* () { const systemdRun = yield* resolve("systemd-run", hostEnvironment); const systemctl = yield* resolve("systemctl", hostEnvironment); if (systemdRun === null || systemctl === null) return undefined; + yield* stopLeftoverAgents(systemctl); const probe = (options: ReadonlyArray) => run(systemdRun, [ "--user", @@ -133,12 +189,13 @@ export const make = Effect.gen(function* () { ]).pipe(Effect.map((result) => result._tag === "Some" && result.value.code === 0)); // Stop the whole scope when the kernel kills one of its processes, and // record the result as oom-kill. Scopes take OOMPolicy from systemd 253; - // older user managers still get plain scopes, without the OOM label. - const oomOptions = ["--property=OOMPolicy=stop"]; + // older user managers still get scopes, without the OOM label. + const stopOptions = [STOP_TIMEOUT_OPTION]; + const oomOptions = [...stopOptions, "--property=OOMPolicy=stop"]; const scopeOptions = (yield* probe(oomOptions)) ? oomOptions - : (yield* probe([])) - ? [] + : (yield* probe(stopOptions)) + ? stopOptions : undefined; if (scopeOptions === undefined) { yield* Effect.logInfo("Agent scopes are off: no systemd user manager"); @@ -176,11 +233,6 @@ export const make = Effect.gen(function* () { // at a time, so only its latest scope can explain a failure. const scopeByThread = new Map(); - const readCgroupFile = (controlGroup: string, file: string) => - fileSystem - .readFileString(path.join("/sys/fs/cgroup", controlGroup, file)) - .pipe(Effect.orElseSucceed(() => "")); - const readScope = (systemctl: string, unit: string) => Effect.gen(function* () { const shown = yield* run(systemctl, [ @@ -290,7 +342,14 @@ export const make = Effect.gen(function* () { }); return { - wrap: Effect.fn("AgentScope.wrap")(function* ({ command, args, name, threadId, env }) { + wrap: Effect.fn("AgentScope.wrap")(function* ({ + command, + args, + name, + threadId, + env, + survivesRestart, + }) { const resolved = yield* resolve(command, env ?? hostEnvironment); // Leave unresolved commands alone so the spawn reports the missing binary. if (resolved === null) return { command, args }; @@ -301,7 +360,7 @@ export const make = Effect.gen(function* () { const systemdScope: AgentSystemdScope = { systemdRun: scope.systemdRun, runtimeDir: scope.runtimeDir, - slice: AGENT_SLICE, + slice: survivesRestart === true ? AGENT_SLICE : ownedSlice, unit, // Without a thread nothing reads the result, so let systemd unload // the unit even when it fails. diff --git a/apps/server/src/terminal/Manager.ts b/apps/server/src/terminal/Manager.ts index 684a305d0e9def868640dde7c525527f5c072f0f..4f79d07ff40f9fc832c90ad0f18bce1ca004aa9e 100644 --- a/apps/server/src/terminal/Manager.ts +++ b/apps/server/src/terminal/Manager.ts @@ -2196,6 +2196,7 @@ export const makeWithOptions = Effect.fn("TerminalManager.makeWithOptions")(func args: candidate.args ?? [], name: "terminal", env: spawnEnv, + survivesRestart: true, }); const attempt = yield* Effect.result( options.ptyAdapter.spawn({ diff --git a/apps/web/src/components/ProviderUpdateLaunchNotification.logic.test.ts b/apps/web/src/components/ProviderUpdateLaunchNotification.logic.test.ts index d388f9f036cb96f339e9b3fc6569e21bbe9b4379..f192241a24774336ff444aa146c454c6836f553a 100644 --- a/apps/web/src/components/ProviderUpdateLaunchNotification.logic.test.ts +++ b/apps/web/src/components/ProviderUpdateLaunchNotification.logic.test.ts @@ -1,6 +1,6 @@ import { describe, expect, it } from "vite-plus/test"; import { - type EnvironmentId, + EnvironmentId, ProviderDriverKind, ProviderInstanceId, type ServerProvider, @@ -1076,6 +1076,7 @@ describe("getProviderUpdateRunToastView", () => { providerDriver: string, result: Parameters[0][number]["result"], ) => ({ + environmentId: EnvironmentId.make(machineLabel), machineLabel, driver: driver(providerDriver), instanceId: instanceId(providerDriver), diff --git a/apps/web/src/components/ProviderUpdateLaunchNotification.logic.ts b/apps/web/src/components/ProviderUpdateLaunchNotification.logic.ts index e245fcb7cf30a4d99346c2b3bc6c0263921d5679..f4868d5333ec715fda5e0f322d40aa0c4809ddec 100644 --- a/apps/web/src/components/ProviderUpdateLaunchNotification.logic.ts +++ b/apps/web/src/components/ProviderUpdateLaunchNotification.logic.ts @@ -333,6 +333,7 @@ export function getProviderUpdateProgressToastView(input: { /** One provider update sent by the cross-machine "Update all", with its result. */ export interface ProviderUpdateRun { + readonly environmentId: EnvironmentId; readonly machineLabel: string; readonly driver: ProviderDriverKind; readonly instanceId: ProviderInstanceId; @@ -342,6 +343,35 @@ export interface ProviderUpdateRun { >; } +/** + * What one provider update in a cross-machine run did. A failure carries a + * line that names the machine and provider, such as "alvin · Codex: npm exited". + */ +export function getProviderUpdateRunOutcome( + run: ProviderUpdateRun, +): + | { readonly status: "interrupted" | "updated" } + | { readonly status: "failed"; readonly line: string } { + if (isAtomCommandInterrupted(run.result)) return { status: "interrupted" }; + const label = `${run.machineLabel} · ${PROVIDER_DISPLAY_NAMES[run.driver] ?? run.driver}`; + if (run.result._tag === "Failure") { + const error = squashAtomCommandFailure(run.result); + return { + status: "failed", + line: `${label}: ${error instanceof Error ? error.message : "Provider update failed."}`, + }; + } + const updateState = run.result.value.providers.find( + (provider) => provider.instanceId === run.instanceId, + )?.updateState; + return updateState?.status === "succeeded" + ? { status: "updated" } + : { + status: "failed", + line: `${label}: ${updateState?.message ?? "Provider update did not finish."}`, + }; +} + /** * Summarize a cross-machine "Update all" as one toast, or null when every * request was interrupted. Each update that did not succeed gets its own line, @@ -350,23 +380,15 @@ export interface ProviderUpdateRun { export function getProviderUpdateRunToastView( runs: ReadonlyArray, ): Pick | null { - const settled = runs.filter((run) => !isAtomCommandInterrupted(run.result)); + const settled = runs + .map(getProviderUpdateRunOutcome) + .filter((outcome) => outcome.status !== "interrupted"); if (settled.length === 0) { return null; } - const failureLines = settled.flatMap((run) => { - const label = `${run.machineLabel} · ${PROVIDER_DISPLAY_NAMES[run.driver] ?? run.driver}`; - if (run.result._tag === "Failure") { - const error = squashAtomCommandFailure(run.result); - return [`${label}: ${error instanceof Error ? error.message : "Provider update failed."}`]; - } - const updateState = run.result.value.providers.find( - (provider) => provider.instanceId === run.instanceId, - )?.updateState; - return updateState?.status === "succeeded" - ? [] - : [`${label}: ${updateState?.message ?? "Provider update did not finish."}`]; - }); + const failureLines = settled.flatMap((outcome) => + outcome.status === "failed" ? [outcome.line] : [], + ); if (failureLines.length === 0) { return { type: "success", diff --git a/apps/web/src/components/ProviderUpdatesAction.tsx b/apps/web/src/components/ProviderUpdatesAction.tsx index 8d01d2e7578e2a5b227cca4ffae0119c6b3891ad..483ac520ad76d65bde5e0c0b685ccf578fa20b58 100644 --- a/apps/web/src/components/ProviderUpdatesAction.tsx +++ b/apps/web/src/components/ProviderUpdatesAction.tsx @@ -20,7 +20,8 @@ import { Tooltip, TooltipPopup, TooltipTrigger } from "./ui/tooltip"; * Each server queues updates that share an installer, so sending them all * together is safe. The server still checks permissions: a session that * cannot operate a machine gets a failure line for it. The returned promise - * settles when every update has finished. + * settles with every result when every update has finished. `quiet` skips the + * toast, for a caller that reports the results itself. */ export function useProviderUpdates() { const { environments } = useEnvironments(); @@ -44,10 +45,11 @@ export function useProviderUpdates() { [environments], ); - const updateAll = async () => { + const updateAll = async ({ quiet = false }: { readonly quiet?: boolean } = {}) => { const runs = await Promise.all( machines.flatMap(({ environmentId, label, candidates }) => candidates.map(async (candidate): Promise => ({ + environmentId, machineLabel: label, driver: candidate.driver, instanceId: candidate.instanceId, @@ -58,7 +60,7 @@ export function useProviderUpdates() { })), ), ); - const view = getProviderUpdateRunToastView(runs); + const view = quiet ? null : getProviderUpdateRunToastView(runs); if (view) { toastManager.add( stackedThreadToast({ @@ -67,6 +69,7 @@ export function useProviderUpdates() { }), ); } + return runs; }; return { machines, updateAll }; diff --git a/apps/web/src/components/ServerUpdateAction.test.tsx b/apps/web/src/components/ServerUpdateAction.test.tsx index 385e12af7dfc0917b27b5ee65842d3d38e48fb49..8c951be4075a306a1e3c9ee5daf345ccc05a139c 100644 --- a/apps/web/src/components/ServerUpdateAction.test.tsx +++ b/apps/web/src/components/ServerUpdateAction.test.tsx @@ -498,13 +498,12 @@ describe("ServerUpdateProgress", () => { expect(markup).toContain("Restarting…"); // The wait state is monochrome and calm: no versions, no step rail, no - // success/warning colors, one duty-cycled pulse on the dot. + // success/warning colors, and a static dot that never repaints. expect(markup).not.toContain("0.0.30"); expect(markup).not.toContain("Resum"); expect(markup).not.toContain("text-success"); expect(markup).not.toContain("text-primary"); - expect(markup).toContain("animate-status-pulse"); - expect(markup).not.toContain("animate-spin"); + expect(markup).not.toContain("animate-"); }); it("folds the sub-second installing handoff into the download phase", () => { diff --git a/apps/web/src/components/ServerUpdateAction.tsx b/apps/web/src/components/ServerUpdateAction.tsx index 0f103235ce64805a3826b4833aeecedccc9d1d93..012911f920174c0d51739a7a5a0590b148a86d71 100644 --- a/apps/web/src/components/ServerUpdateAction.tsx +++ b/apps/web/src/components/ServerUpdateAction.tsx @@ -65,17 +65,26 @@ type UpdateButtonProps = Pick, "variant" | "size" readonly appearance?: "button" | "icon"; }; +/** What {@link useServerUpdate} did. "skipped" covers a missing permission, an + update already in flight, and an interrupted request. */ +export type ServerUpdateOutcome = + | { readonly status: "updated" | "skipped" } + | { readonly status: "failed"; readonly message: string }; + /** Updates one server and waits until it is back on the target version. - Reports its own success or failure toast; never rejects. */ + Toasts its success or failure unless `quiet`; never rejects. */ export function useServerUpdate() { const updateServer = useAtomCommand(serverEnvironment.updateServer, { reportFailure: false }); - return async (target: ServerUpdateTarget, failureTitle = "Server update failed") => { + return async ( + target: ServerUpdateTarget, + { quiet = false }: { readonly quiet?: boolean } = {}, + ): Promise => { const { environmentId, serverLabel, selfUpdate, targetVersion } = target; if ( !canUpdateServer(appAtomRegistry.get(environmentSession.sessionStateAtom(environmentId))) || pendingUpdateEnvironmentIds.has(environmentId) ) - return; + return { status: "skipped" }; pendingUpdateEnvironmentIds.add(environmentId); try { const result = await updateServer({ @@ -88,23 +97,30 @@ export function useServerUpdate() { }, }); if (result._tag === "Failure") { - if (isAtomCommandInterrupted(result)) return; + if (isAtomCommandInterrupted(result)) return { status: "skipped" }; throw squashAtomCommandFailure(result); } - toastManager.add({ - type: "success", - title: `${serverLabel} updated`, - description: - selfUpdate === "desktop-managed" - ? `Desktop app relaunched on ${result.value.targetVersion}.` - : `Reconnected on t3@${result.value.targetVersion}.`, - }); + if (!quiet) { + toastManager.add({ + type: "success", + title: `${serverLabel} updated`, + description: + selfUpdate === "desktop-managed" + ? `Desktop app relaunched on ${result.value.targetVersion}.` + : `Reconnected on t3@${result.value.targetVersion}.`, + }); + } + return { status: "updated" }; } catch (error) { - toastManager.add({ - type: "error", - title: failureTitle, - description: updateFailureMessage(error), - }); + const message = updateFailureMessage(error); + if (!quiet) { + toastManager.add({ + type: "error", + title: `${serverLabel} update failed`, + description: message, + }); + } + return { status: "failed", message }; } finally { pendingUpdateEnvironmentIds.delete(environmentId); } @@ -145,9 +161,7 @@ export function ServerUpdatesAction({ )) ?? true; if (!confirmed) return; } - await Promise.all( - available.map((target) => update(target, `${target.serverLabel} update failed`)), - ); + await Promise.all(available.map((target) => update(target))); } finally { pending.current = false; setIsPending(false); @@ -175,7 +189,7 @@ export function canUpdateServer( /** * One-row status for an in-flight server update: "Downloading…" then - * "Restarting…". The update is a wait, not a warning: a single pulsing dot + * "Restarting…". The update is a wait, not a warning: a single static dot * and label, no step rail, no versions. Failure turns the row red with the * rollback reason. */ @@ -197,10 +211,7 @@ export function ServerUpdateProgress({ } return (
-
); @@ -381,7 +392,7 @@ export function OutdatedServerUpdateAction({ } catch (error) { toastManager.add({ type: "error", - title: "Server update failed", + title: `${serverLabel} update failed`, description: updateFailureMessage(error), }); } finally { diff --git a/apps/web/src/components/desktopUpdate.logic.test.ts b/apps/web/src/components/desktopUpdate.logic.test.ts index fcc97825b681c277f8e36006f9e86b0be5c6d32a..04b1bac77fa9249580d74580691a5edbb3424c3d 100644 --- a/apps/web/src/components/desktopUpdate.logic.test.ts +++ b/apps/web/src/components/desktopUpdate.logic.test.ts @@ -57,6 +57,26 @@ describe("desktop update button state", () => { expect(getDesktopUpdateButtonTooltip(state)).toContain("Click to retry"); }); + it("says when the same version did not apply before", () => { + const state: DesktopUpdateState = { + ...baseState, + status: "downloaded", + availableVersion: "1.1.0", + downloadedVersion: "1.1.0", + notAppliedVersion: "1.1.0", + }; + expect(getDesktopUpdateButtonTooltip(state)).toBe( + "Update 1.1.0 did not apply. Click to retry.", + ); + expect( + getDesktopUpdateButtonTooltip({ + ...state, + availableVersion: "1.2.0", + downloadedVersion: "1.2.0", + }), + ).toContain("1.2.0 downloaded"); + }); + it("keeps install action available after an install error", () => { const state: DesktopUpdateState = { ...baseState, diff --git a/apps/web/src/components/desktopUpdate.logic.ts b/apps/web/src/components/desktopUpdate.logic.ts index 4a169cb3ef409ac8216b4497ef46b1efffaf5fbb..3eb2726bfd350b3b17bef82b14847e633beb2b1d 100644 --- a/apps/web/src/components/desktopUpdate.logic.ts +++ b/apps/web/src/components/desktopUpdate.logic.ts @@ -71,6 +71,13 @@ export function getArm64IntelBuildWarningDescription(state: DesktopUpdateState): } export function getDesktopUpdateButtonTooltip(state: DesktopUpdateState): string { + if ( + state.notAppliedVersion !== undefined && + (state.status === "available" || state.status === "downloaded") && + getDesktopUpdateDownloadedVersion(state) === state.notAppliedVersion + ) { + return `Update ${state.notAppliedVersion} did not apply. Click to retry.`; + } if (state.status === "available") { return `Update ${state.availableVersion ?? "available"} ready to download`; } diff --git a/apps/web/src/components/sidebar/SidebarUpdatePill.tsx b/apps/web/src/components/sidebar/SidebarUpdatePill.tsx index 1d0ea0332430c9d656649d236dcd2f253b712d4e..5f756069f98eb8694057cccfce5c3ba7160e62e1 100644 --- a/apps/web/src/components/sidebar/SidebarUpdatePill.tsx +++ b/apps/web/src/components/sidebar/SidebarUpdatePill.tsx @@ -1,12 +1,23 @@ +import { useAtomValue } from "@effect/atom-react"; import type { DesktopBridge, DesktopUpdateState } from "@t3tools/contracts"; +import { Atom } from "effect/reactivity"; import { TriangleAlertIcon } from "lucide-react"; -import { type ComponentProps, useCallback, useEffect, useId, useRef, useState } from "react"; +import { + type ComponentProps, + useCallback, + useEffect, + useId, + useMemo, + useRef, + useState, +} from "react"; import { flushSync } from "react-dom"; import { isElectron } from "../../env"; import { useMediaQuery } from "../../hooks/useMediaQuery"; import { cn } from "../../lib/utils"; import { ensureLocalApi } from "../../localApi"; import { useDesktopUpdateState } from "../../state/desktopUpdate"; +import { serverEnvironment } from "../../state/server"; import { stackedThreadToast, toastManager } from "../ui/toast"; import { canCheckForUpdate, @@ -33,7 +44,13 @@ import { } from "./DesktopUpdateStatusIcon"; import { SidebarUpdateReleaseNotes } from "./SidebarUpdateReleaseNotes"; import { UpdateEverythingDialog } from "./UpdateEverythingDialog"; -import { useUpdateEverything } from "./useUpdateEverything"; +import { + describeUpdateRun, + resolveUpdateRunMachineStatus, + type UpdateEverythingRun, +} from "./updateEverything.logic"; +import { useUpdateEverythingRun } from "./updateEverythingRun"; +import { type ReportLocalUpdateError, useUpdateEverything } from "./useUpdateEverything"; type SidebarUpdatePopoverChangeDetails = Parameters< NonNullable["onOpenChange"]> @@ -68,27 +85,34 @@ export function openSidebarUpdateReleaseNotesPopoverOnForwardTab( function resolveSidebarUpdatePresentation({ action, showsEverything, + run, isDownloading, showCheckIcon, }: { readonly action: ReturnType; - /** Remote work is waiting or running, and this app is not downloading. */ + /** Remote work is waiting, and this app is not downloading. */ readonly showsEverything: boolean; + /** The running "update everything" pass. Its state sets the icon. */ + readonly run: UpdateEverythingRun | null; readonly isDownloading: boolean; readonly showCheckIcon: boolean; }) { - const showUpdateDetails = action !== "none" || showsEverything || isDownloading; + const showUpdateDetails = action !== "none" || showsEverything || run !== null || isDownloading; const iconStatus: DesktopUpdateStatusIconState = showCheckIcon ? "checking" : isDownloading ? "downloading" - : showsEverything - ? "everything" - : action === "install" + : run !== null + ? run.localVersion !== null && action === "install" ? "downloaded" - : action === "download" - ? "available" - : "idle"; + : "everything" + : showsEverything + ? "everything" + : action === "install" + ? "downloaded" + : action === "download" + ? "available" + : "idle"; return { iconStatus, @@ -117,11 +141,18 @@ function SidebarUpdateArchitectureWarningContent() { ); } +const toastUpdateError: ReportLocalUpdateError = (title, description) => { + toastManager.add(stackedThreadToast({ type: "error", title, description })); +}; + /** Downloads this app's update. Resolves true when it is ready to install; - failures toast and resolve false. */ + failures go to `reportError` (a toast by default) and resolve false. */ async function downloadLocalUpdate( bridge: DesktopBridge, - { notifyDownloaded }: { readonly notifyDownloaded: boolean }, + { + notifyDownloaded, + reportError = toastUpdateError, + }: { readonly notifyDownloaded: boolean; readonly reportError?: ReportLocalUpdateError }, ): Promise { try { const result = await bridge.downloadUpdate(); @@ -131,60 +162,83 @@ async function downloadLocalUpdate( const actionError = shouldToastDesktopUpdateActionResult(result) ? getDesktopUpdateActionError(result) : null; - if (actionError) { - toastManager.add( - stackedThreadToast({ - type: "error", - title: "Could not download update", - description: actionError, - }), - ); - } + if (actionError) reportError("Could not download update", actionError); return result.completed; } catch (error) { - toastManager.add( - stackedThreadToast({ - type: "error", - title: "Could not start update download", - description: error instanceof Error ? error.message : "An unexpected error occurred.", - }), + reportError( + "Could not start update download", + error instanceof Error ? error.message : "An unexpected error occurred.", ); return false; } } -/** Installs this app's downloaded update, which relaunches the app. - Failures toast. A refused install (the download is not marked ready yet) +/** Installs this app's downloaded update, which relaunches the app. Resolves + true when the relaunch is under way. Failures go to `reportError` (a toast + by default). A refused install (the download is not marked ready yet) leaves the restart to the user. */ -async function installLocalUpdate(bridge: DesktopBridge): Promise { +async function installLocalUpdate( + bridge: DesktopBridge, + reportError: ReportLocalUpdateError = toastUpdateError, +): Promise { try { const result = await bridge.installUpdate(); if (!result.accepted) { showDesktopUpdateDownloadedToast(bridge, result.state); - return; + return false; } const actionError = shouldToastDesktopUpdateActionResult(result) ? getDesktopUpdateActionError(result) : null; - if (!actionError) return; - toastManager.add( - stackedThreadToast({ - type: "error", - title: "Could not install update", - description: actionError, - }), - ); + if (!actionError) return true; + reportError("Could not install update", actionError); + return false; } catch (error) { - toastManager.add( - stackedThreadToast({ - type: "error", - title: "Could not install update", - description: error instanceof Error ? error.message : "An unexpected error occurred.", - }), + reportError( + "Could not install update", + error instanceof Error ? error.message : "An unexpected error occurred.", ); + return false; } } +/** Tooltip lines for a running pass: this app first, then one per machine. */ +function useUpdateRunTooltip( + run: UpdateEverythingRun | null, + state: DesktopUpdateState | null, + localReady: boolean, +): string | null { + const statusesAtom = useMemo( + () => + Atom.make((get) => + run === null + ? [] + : run.machines.map((machine) => ({ + label: machine.label, + status: resolveUpdateRunMachineStatus( + machine, + run.phase, + get(serverEnvironment.updateStateAtom(machine.environmentId)), + ), + })), + ), + [run], + ); + const statuses = useAtomValue(statusesAtom); + if (run === null) return null; + const local = + run.localVersion === null + ? null + : state?.status === "downloading" + ? `downloading${state.downloadPercent === null ? "" : ` ${Math.floor(state.downloadPercent)}%`}` + : run.phase === "installing" + ? "restarting" + : localReady + ? "restarts last" + : "waiting"; + return describeUpdateRun(statuses, local).join("\n"); +} + /** * The sidebar update button. With only a local app update it downloads, then * installs. When other machines' servers or any providers are behind, one @@ -198,7 +252,6 @@ export function SidebarUpdatePill() { function SidebarUpdateControl() { const state = useDesktopUpdateState(); const [isActionPending, setIsActionPending] = useState(false); - const [isUpdatingEverything, setIsUpdatingEverything] = useState(false); const [isConfirmingEverything, setIsConfirmingEverything] = useState(false); const [checkAnimationKey, setCheckAnimationKey] = useState(0); const [isCheckAnimationLatched, setIsCheckAnimationLatched] = useState(false); @@ -224,7 +277,9 @@ function SidebarUpdateControl() { const everything = useUpdateEverything(localUpdateVersion); const runsEverything = everything.hasUpdates && !isDownloading; // Providers leave the list once queued, so the run keeps its own state. - const isRunningEverything = isUpdatingEverything && !isDownloading; + const run = useUpdateEverythingRun(); + const isRunningEverything = run !== null; + const runTooltip = useUpdateRunTooltip(run, state, action === "install"); const showCheckIcon = shouldShowDesktopUpdateCheckIcon({ isAnimationLatched: isCheckAnimationLatched, isChecking: state?.status === "checking", @@ -232,21 +287,23 @@ function SidebarUpdateControl() { }); const { iconStatus, showUpdateDetails, showUpdateIconState } = resolveSidebarUpdatePresentation({ action, - showsEverything: runsEverything || isRunningEverything, + showsEverything: runsEverything, + run, isDownloading, showCheckIcon, }); - const tooltip = isRunningEverything - ? "Updating…" - : runsEverything - ? everything.summary - : showUpdateDetails - ? state - ? getDesktopUpdateButtonTooltip(state) - : "Update available" - : showCheckIcon - ? "Checking for updates…" - : "Check for updates"; + const tooltip = + runTooltip !== null + ? runTooltip + : runsEverything + ? everything.summary + : showUpdateDetails + ? state + ? getDesktopUpdateButtonTooltip(state) + : "Update available" + : showCheckIcon + ? "Checking for updates…" + : "Check for updates"; const disabled = showCheckIcon ? true : runsEverything @@ -254,11 +311,10 @@ function SidebarUpdateControl() { : showUpdateDetails ? isDesktopUpdateButtonDisabled(state) : !canCheckForUpdate(state); - const isInteractionDisabled = disabled || isActionPending; - const showReleaseNotesPopover = shouldUseSidebarUpdateReleaseNotesPopover( - showUpdateDetails, - state, - ); + const isInteractionDisabled = disabled || isActionPending || isRunningEverything; + // During a run the tooltip shows its progress instead of release notes. + const showReleaseNotesPopover = + !isRunningEverything && shouldUseSidebarUpdateReleaseNotesPopover(showUpdateDetails, state); useEffect(() => { if (!showReleaseNotesPopover) { @@ -349,13 +405,17 @@ function SidebarUpdateControl() { const bridge = window.desktopBridge; setIsConfirmingEverything(false); setIsActionPending(true); - setIsUpdatingEverything(true); try { await everything.run({ ...(bridge && action === "download" - ? { downloadLocal: () => downloadLocalUpdate(bridge, { notifyDownloaded: false }) } + ? { + downloadLocal: (reportError) => + downloadLocalUpdate(bridge, { notifyDownloaded: false, reportError }), + } + : {}), + ...(bridge && action !== "none" + ? { installLocal: (reportError) => installLocalUpdate(bridge, reportError) } : {}), - ...(bridge && action !== "none" ? { installLocal: () => installLocalUpdate(bridge) } : {}), }); } catch (error) { toastManager.add( @@ -366,7 +426,6 @@ function SidebarUpdateControl() { }), ); } finally { - setIsUpdatingEverything(false); setIsActionPending(false); } }, [action, everything]); @@ -381,7 +440,7 @@ function SidebarUpdateControl() { }, [prefersReducedMotion, state?.status]); // Web has no local app to check, so it only shows the button for remote work. - if (!isElectron && !runsEverything && !isUpdatingEverything) return null; + if (!isElectron && !runsEverything && !isRunningEverything) return null; const updateButton = (