From bade0f1086c4a2b91b9ab14053b02f8b8490afe4 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 10 Nov 2026 18:01:32 -0800 Subject: [PATCH 1/4] feat(server): keep the computer awake while agents work MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Holds an OS sleep inhibitor (caffeinate / systemd-inhibit / SetThreadExecutionState) while any run is preparing, starting or running, and releases it 60s after the last one ends. On by default, with a toggle in Settings → Maintenance on web and mobile. Co-Authored-By: Claude Opus 5.5 --- .../SettingsServerControlsRouteScreen.tsx | 15 ++ apps/server/src/background/KeepAwake.test.ts | 36 +++++ apps/server/src/background/KeepAwake.ts | 149 ++++++++++++++++++ apps/server/src/server.ts | 2 + .../components/settings/SettingsPanels.tsx | 33 ++++ .../src/components/settings/settingsSearch.ts | 7 + packages/contracts/src/settings.ts | 3 + 7 files changed, 245 insertions(+) create mode 100644 apps/server/src/background/KeepAwake.test.ts create mode 100644 apps/server/src/background/KeepAwake.ts diff --git a/apps/mobile/src/features/settings/SettingsServerControlsRouteScreen.tsx b/apps/mobile/src/features/settings/SettingsServerControlsRouteScreen.tsx index 793e3ae75412..cab43a0e4363 100644 --- a/apps/mobile/src/features/settings/SettingsServerControlsRouteScreen.tsx +++ b/apps/mobile/src/features/settings/SettingsServerControlsRouteScreen.tsx @@ -555,6 +555,21 @@ function ServerSettingsDetail(props: { readonly page: SettingsPage }) { onValueChange={(value) => write({ continueThreadsAfterServerUpdate: value })} /> + + write({ keepAwakeWhileAgentsWork: value })} + /> + ) : null} diff --git a/apps/server/src/background/KeepAwake.test.ts b/apps/server/src/background/KeepAwake.test.ts new file mode 100644 index 000000000000..1ced5247466f --- /dev/null +++ b/apps/server/src/background/KeepAwake.test.ts @@ -0,0 +1,36 @@ +import { describe, expect, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as Queue from "effect/Queue"; +import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; + +import * as KeepAwake from "./KeepAwake.ts"; + +describe("KeepAwake.holdWhileWanted", () => { + it.effect("keeps one hold across a gap shorter than the grace and releases after it", () => + Effect.gen(function* () { + const wanted = yield* Queue.unbounded(); + const transitions = yield* Queue.unbounded<"held" | "released">(); + const hold = Effect.acquireRelease(Queue.offer(transitions, "held"), () => + Queue.offer(transitions, "released"), + ).pipe(Effect.asVoid); + yield* KeepAwake.holdWhileWanted(Stream.fromQueue(wanted), hold, "60 seconds").pipe( + Effect.forkScoped, + ); + + yield* Queue.offer(wanted, true); + expect(yield* Queue.take(transitions)).toBe("held"); + + yield* Queue.offer(wanted, false); + yield* TestClock.adjust("30 seconds"); + yield* Queue.offer(wanted, true); + yield* TestClock.adjust("120 seconds"); + expect(yield* Queue.size(transitions)).toBe(0); + + yield* Queue.offer(wanted, false); + yield* TestClock.adjust("120 seconds"); + expect(yield* Queue.take(transitions)).toBe("released"); + expect(yield* Queue.size(transitions)).toBe(0); + }), + ); +}); diff --git a/apps/server/src/background/KeepAwake.ts b/apps/server/src/background/KeepAwake.ts new file mode 100644 index 000000000000..1a82c5b23f26 --- /dev/null +++ b/apps/server/src/background/KeepAwake.ts @@ -0,0 +1,149 @@ +import * as Duration from "effect/Duration"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as ChildProcess from "effect/process/ChildProcess"; +import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner"; +import type * as Scope from "effect/Scope"; +import * as Stream from "effect/Stream"; +import * as SqlClient from "effect/sql/SqlClient"; + +import * as HostProcess from "@t3tools/shared/HostProcess"; + +import * as Orchestrator from "../orchestration-v2/Orchestrator.ts"; +import { forkParked } from "../serverActivation.ts"; +import * as Settings from "../serverSettings.ts"; + +// Bridges gaps between back-to-back runs, so a long-idle machine does not sleep mid-queue. +const RELEASE_GRACE = Duration.seconds(60); + +/** Each command holds its inhibitor until it exits, and exits on its own if the server dies. */ +function inhibitorCommand( + platform: NodeJS.Platform, + serverPid: number, +): { readonly command: string; readonly args: ReadonlyArray } | undefined { + switch (platform) { + case "darwin": + return { command: "caffeinate", args: ["-i", "-w", String(serverPid)] }; + case "linux": + return { + command: "systemd-inhibit", + args: [ + "--what=idle:sleep", + "--who=T3 Code", + "--why=Agents are working", + "--mode=block", + "tail", + `--pid=${serverPid}`, + "-f", + "/dev/null", + ], + }; + case "win32": + return { + command: "powershell.exe", + args: [ + "-NoProfile", + "-NonInteractive", + "-Command", + // ES_CONTINUOUS | ES_SYSTEM_REQUIRED, held by this thread until Wait-Process returns. + `$t = Add-Type -Name Power -Namespace T3 -PassThru -MemberDefinition '[DllImport("kernel32.dll")] public static extern uint SetThreadExecutionState(uint f);'; [void]$t::SetThreadExecutionState([uint32]2147483649); Wait-Process -Id ${serverPid}`, + ], + }; + default: + return undefined; + } +} + +/** Holds while `wanted` is true and for `grace` after it turns false; a true inside the grace keeps the same hold. */ +export const holdWhileWanted = ( + wanted: Stream.Stream, + hold: Effect.Effect, + grace: Duration.Input = RELEASE_GRACE, +) => + wanted.pipe( + Stream.changes, + Stream.switchMap((on) => + on ? Stream.succeed(true) : Stream.fromEffect(Effect.as(Effect.sleep(grace), false)), + ), + Stream.changes, + Stream.switchMap((on) => (on ? Stream.unwrap(Effect.as(hold, Stream.never)) : Stream.empty)), + Stream.runDrain, + ); + +export const layer = Layer.effectDiscard( + Effect.gen(function* () { + const command = inhibitorCommand(yield* HostProcess.Platform, process.pid); + if (command === undefined) return; + const settings = yield* Settings.ServerSettingsService; + const engine = yield* Orchestrator.OrchestratorV2; + const sql = yield* SqlClient.SqlClient; + const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; + + const hasActiveRun = sql<{ readonly active: number }>` + SELECT EXISTS ( + SELECT 1 FROM orchestration_v2_projection_runs + WHERE status IN ('preparing', 'starting', 'running') + ) AS active + `.pipe( + Effect.map((rows) => rows[0]?.active === 1), + Effect.catchCause((cause) => + Effect.logWarning("keep-awake could not read active runs", { cause }).pipe( + Effect.as(false), + ), + ), + ); + const active = Stream.concat( + Stream.fromEffect(hasActiveRun), + engine.streamDomainEvents.pipe( + Stream.filter((event) => event.type === "run.created" || event.type === "run.updated"), + Stream.mapEffect(() => hasActiveRun), + ), + ); + const enabled = Stream.unwrap( + Effect.gen(function* () { + const changes = yield* settings.subscribeChanges; + const current = yield* settings.getSettings.pipe( + Effect.map((value) => value.keepAwakeWhileAgentsWork), + Effect.orElseSucceed(() => true), + ); + return Stream.concat( + Stream.succeed(current), + Stream.map(changes, (value) => value.keepAwakeWhileAgentsWork), + ); + }), + ); + + const hold = spawner + .spawn( + ChildProcess.make(command.command, command.args, { + detached: false, + shell: false, + stdin: "ignore", + stdout: "ignore", + stderr: "ignore", + }), + ) + .pipe( + Effect.tap((child) => + Effect.logInfo("keep-awake holding sleep inhibitor", { pid: Number(child.pid) }), + ), + Effect.tap(() => + Effect.addFinalizer(() => Effect.logInfo("keep-awake released sleep inhibitor")), + ), + Effect.asVoid, + Effect.catchCause((cause) => + Effect.logWarning("keep-awake could not start the sleep inhibitor", { + command: command.command, + cause, + }), + ), + ); + + yield* forkParked( + holdWhileWanted( + Stream.zipLatestWith(enabled, active, (on, busy) => on && busy), + hold, + ).pipe(Effect.catchCause((cause) => Effect.logWarning("keep-awake stopped", { cause }))), + ); + }), +); diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 47190f60dcf5..1af620b71676 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -28,6 +28,7 @@ import * as HttpApiBuilder from "effect/http-api/HttpApiBuilder"; import * as BackgroundPolicy from "./background/BackgroundPolicy.ts"; import * as HostPowerMonitor from "./background/HostPowerMonitor.ts"; +import * as KeepAwake from "./background/KeepAwake.ts"; import * as ServerConfig from "./config.ts"; import { withUntracedRequests } from "./http.ts"; import * as ServerHttp from "./http.ts"; @@ -652,6 +653,7 @@ const layerRuntimeCoreDependenciesBase = Layer.mergeAll( Layer.provide(layerServerRestart), ), layerThreadPullRequestWorker, + KeepAwake.layer, DelegatedTaskAnalytics.layer, Layer.effectDiscard( Effect.gen(function* () { diff --git a/apps/web/src/components/settings/SettingsPanels.tsx b/apps/web/src/components/settings/SettingsPanels.tsx index a37bf1581990..73cb0fdc194a 100644 --- a/apps/web/src/components/settings/SettingsPanels.tsx +++ b/apps/web/src/components/settings/SettingsPanels.tsx @@ -550,6 +550,9 @@ export function useSettingsRestore(onRestored?: () => void) { ? ["Continue threads after restarts"] : []), ...(isBackgroundActivityDirty ? ["Background activity"] : []), + ...(settings.keepAwakeWhileAgentsWork !== DEFAULT_UNIFIED_SETTINGS.keepAwakeWhileAgentsWork + ? ["Keep awake while agents work"] + : []), ...(settings.defaultThreadEnvMode !== DEFAULT_UNIFIED_SETTINGS.defaultThreadEnvMode ? ["New thread mode"] : []), @@ -670,6 +673,7 @@ export function useSettingsRestore(onRestored?: () => void) { enableProviderUpdateChecks: DEFAULT_UNIFIED_SETTINGS.enableProviderUpdateChecks, continueThreadsAfterServerUpdate: DEFAULT_UNIFIED_SETTINGS.continueThreadsAfterServerUpdate, backgroundActivity: DEFAULT_UNIFIED_SETTINGS.backgroundActivity, + keepAwakeWhileAgentsWork: DEFAULT_UNIFIED_SETTINGS.keepAwakeWhileAgentsWork, backgroundActivityProfile: DEFAULT_UNIFIED_SETTINGS.backgroundActivityProfile, automaticGitFetchInterval: DEFAULT_UNIFIED_SETTINGS.automaticGitFetchInterval, providerHealthRefreshInterval: DEFAULT_UNIFIED_SETTINGS.providerHealthRefreshInterval, @@ -2026,6 +2030,35 @@ export function MaintenanceSettingsPanel() { + + updateSettings({ + keepAwakeWhileAgentsWork: DEFAULT_UNIFIED_SETTINGS.keepAwakeWhileAgentsWork, + }) + } + /> + ) : null + } + control={ + + updateSettings({ keepAwakeWhileAgentsWork: Boolean(checked) }) + } + aria-label="Keep awake while agents work" + /> + } + /> diff --git a/apps/web/src/components/settings/settingsSearch.ts b/apps/web/src/components/settings/settingsSearch.ts index f7492035874d..e11cf905c798 100644 --- a/apps/web/src/components/settings/settingsSearch.ts +++ b/apps/web/src/components/settings/settingsSearch.ts @@ -749,6 +749,13 @@ export const SETTINGS_SEARCH_ITEMS = [ "resume running active interrupted work restart reboot machine crash desktop update automatically", ], }, + { + id: "keep-awake-while-agents-work", + title: "Keep awake while agents work", + to: "/settings/maintenance", + scope: "environment-defaults", + searchTerms: ["caffeinate sleep prevent stay awake power laptop lid inhibit"], + }, { id: "background-activity", title: "Background activity", diff --git a/packages/contracts/src/settings.ts b/packages/contracts/src/settings.ts index cd112d84bf9f..95e39b975087 100644 --- a/packages/contracts/src/settings.ts +++ b/packages/contracts/src/settings.ts @@ -1355,6 +1355,8 @@ export const ServerSettings = Schema.Struct({ continueThreadsAfterServerUpdate: Schema.Boolean.pipe( Schema.withDecodingDefault(Effect.succeed(true)), ), + /** Holds an OS sleep inhibitor while any run is active; closing a laptop lid still sleeps it. */ + keepAwakeWhileAgentsWork: Schema.Boolean.pipe(Schema.withDecodingDefault(Effect.succeed(true))), enableAgentBrowserAccess: Schema.Boolean.pipe(Schema.withDecodingDefault(Effect.succeed(true))), cloudSpeechCredentials: CloudSpeechCredentials.pipe( @@ -1802,6 +1804,7 @@ export const ServerSettingsPatch = Schema.Struct({ responseStreamingMode: Schema.optionalKey(ResponseStreamingMode), enableProviderUpdateChecks: Schema.optionalKey(Schema.Boolean), continueThreadsAfterServerUpdate: Schema.optionalKey(Schema.Boolean), + keepAwakeWhileAgentsWork: Schema.optionalKey(Schema.Boolean), enableAgentBrowserAccess: Schema.optionalKey(Schema.Boolean), cloudSpeechCredentials: Schema.optionalKey( Schema.Struct({ From 3120eb54af91cad958a1ada146c8a48fea4a3623 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 12 Nov 2026 15:55:37 -0800 Subject: [PATCH 2/4] fix(server): keep-awake restarts a dead inhibitor and cannot miss run changes Read the event cursor before the active-run snapshot so a run that changes in between is replayed, keep the last known state when the read fails, and restart the inhibitor with capped backoff if it exits while still wanted. Co-Authored-By: Claude Opus 5.5 --- apps/server/src/background/KeepAwake.test.ts | 2 +- apps/server/src/background/KeepAwake.ts | 83 ++++++++++++++------ 2 files changed, 62 insertions(+), 23 deletions(-) diff --git a/apps/server/src/background/KeepAwake.test.ts b/apps/server/src/background/KeepAwake.test.ts index 1ced5247466f..8984102729ce 100644 --- a/apps/server/src/background/KeepAwake.test.ts +++ b/apps/server/src/background/KeepAwake.test.ts @@ -13,7 +13,7 @@ describe("KeepAwake.holdWhileWanted", () => { const transitions = yield* Queue.unbounded<"held" | "released">(); const hold = Effect.acquireRelease(Queue.offer(transitions, "held"), () => Queue.offer(transitions, "released"), - ).pipe(Effect.asVoid); + ).pipe(Effect.andThen(Effect.never), Effect.scoped); yield* KeepAwake.holdWhileWanted(Stream.fromQueue(wanted), hold, "60 seconds").pipe( Effect.forkScoped, ); diff --git a/apps/server/src/background/KeepAwake.ts b/apps/server/src/background/KeepAwake.ts index 1a82c5b23f26..c0972f0f51a0 100644 --- a/apps/server/src/background/KeepAwake.ts +++ b/apps/server/src/background/KeepAwake.ts @@ -3,13 +3,14 @@ import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as ChildProcess from "effect/process/ChildProcess"; import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner"; -import type * as Scope from "effect/Scope"; +import * as Schedule from "effect/Schedule"; +import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; import * as SqlClient from "effect/sql/SqlClient"; import * as HostProcess from "@t3tools/shared/HostProcess"; -import * as Orchestrator from "../orchestration-v2/Orchestrator.ts"; +import * as OrchestrationEventStore from "../persistence/OrchestrationEventStore.ts"; import { forkParked } from "../serverActivation.ts"; import * as Settings from "../serverSettings.ts"; @@ -54,10 +55,13 @@ function inhibitorCommand( } } -/** Holds while `wanted` is true and for `grace` after it turns false; a true inside the grace keeps the same hold. */ +/** + * Runs `hold` while `wanted` is true and for `grace` after it turns false; a true inside the + * grace keeps the same hold. `hold` runs until interrupted. + */ export const holdWhileWanted = ( wanted: Stream.Stream, - hold: Effect.Effect, + hold: Effect.Effect, grace: Duration.Input = RELEASE_GRACE, ) => wanted.pipe( @@ -66,16 +70,21 @@ export const holdWhileWanted = ( on ? Stream.succeed(true) : Stream.fromEffect(Effect.as(Effect.sleep(grace), false)), ), Stream.changes, - Stream.switchMap((on) => (on ? Stream.unwrap(Effect.as(hold, Stream.never)) : Stream.empty)), + Stream.switchMap((on) => (on ? Stream.fromEffect(hold) : Stream.empty)), Stream.runDrain, ); +class KeepAwakeInhibitorExited extends Schema.TaggedError()( + "KeepAwakeInhibitorExited", + { exitCode: Schema.Number }, +) {} + export const layer = Layer.effectDiscard( Effect.gen(function* () { const command = inhibitorCommand(yield* HostProcess.Platform, process.pid); if (command === undefined) return; const settings = yield* Settings.ServerSettingsService; - const engine = yield* Orchestrator.OrchestratorV2; + const events = yield* OrchestrationEventStore.OrchestrationEventStore; const sql = yield* SqlClient.SqlClient; const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; @@ -85,19 +94,33 @@ export const layer = Layer.effectDiscard( WHERE status IN ('preparing', 'starting', 'running') ) AS active `.pipe( - Effect.map((rows) => rows[0]?.active === 1), + Effect.map((rows) => [rows[0]?.active === 1]), + // A failed read keeps the last known state rather than guessing. Effect.catchCause((cause) => - Effect.logWarning("keep-awake could not read active runs", { cause }).pipe( - Effect.as(false), - ), + Effect.logWarning("keep-awake could not read active runs", { cause }).pipe(Effect.as([])), ), ); - const active = Stream.concat( - Stream.fromEffect(hasActiveRun), - engine.streamDomainEvents.pipe( - Stream.filter((event) => event.type === "run.created" || event.type === "run.updated"), - Stream.mapEffect(() => hasActiveRun), + // Read the cursor before the snapshot so a run changing in between is replayed, not missed. + const active = Stream.unwrap( + events.latestApplicationSequence.pipe( + Effect.map((afterSequence) => + Stream.concat( + Stream.succeed(undefined), + events + .streamApplicationEvents({ afterSequence }) + .pipe( + Stream.filter( + (stored) => + "event" in stored && + (stored.event.type === "run.created" || stored.event.type === "run.updated"), + ), + ), + ), + ), ), + ).pipe( + Stream.mapEffect(() => hasActiveRun), + Stream.flattenIterable, ); const enabled = Stream.unwrap( Effect.gen(function* () { @@ -113,7 +136,7 @@ export const layer = Layer.effectDiscard( }), ); - const hold = spawner + const holdOnce = spawner .spawn( ChildProcess.make(command.command, command.args, { detached: false, @@ -130,14 +153,30 @@ export const layer = Layer.effectDiscard( Effect.tap(() => Effect.addFinalizer(() => Effect.logInfo("keep-awake released sleep inhibitor")), ), - Effect.asVoid, - Effect.catchCause((cause) => - Effect.logWarning("keep-awake could not start the sleep inhibitor", { - command: command.command, - cause, - }), + Effect.flatMap((child) => child.exitCode), + Effect.flatMap((exitCode) => + Effect.fail(new KeepAwakeInhibitorExited({ exitCode: Number(exitCode) })), ), + Effect.scoped, ); + // An inhibitor that dies while agents still work is restarted; on hosts without one + // (e.g. no logind) the backoff keeps retries and warnings rare. + const hold = holdOnce.pipe( + Effect.tapCause((cause) => + Effect.logWarning("keep-awake sleep inhibitor stopped", { + command: command.command, + cause, + }), + ), + Effect.retry( + Schedule.exponential("5 seconds").pipe( + Schedule.modifyDelay(({ duration }) => + Effect.succeed(Duration.min(duration, Duration.minutes(10))), + ), + ), + ), + Effect.ignore, + ); yield* forkParked( holdWhileWanted( From 5664796bf632e6eb1bd982dd12a986c899e30b0f Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 12 Nov 2026 16:07:59 -0800 Subject: [PATCH 3/4] fix(server): keep-awake retries failed reads instead of dropping them Co-Authored-By: Claude Opus 5.5 --- apps/server/src/background/KeepAwake.ts | 45 ++++++++++++------------- 1 file changed, 21 insertions(+), 24 deletions(-) diff --git a/apps/server/src/background/KeepAwake.ts b/apps/server/src/background/KeepAwake.ts index c0972f0f51a0..24268e53186b 100644 --- a/apps/server/src/background/KeepAwake.ts +++ b/apps/server/src/background/KeepAwake.ts @@ -74,6 +74,22 @@ export const holdWhileWanted = ( Stream.runDrain, ); +// Reads and the inhibitor retry until they succeed; the cap keeps a host without an +// inhibitor (e.g. no logind) from retrying and warning constantly. +const retryWithBackoff = + (message: string) => + (effect: Effect.Effect) => + effect.pipe( + Effect.tapCause((cause) => Effect.logWarning(message, { cause })), + Effect.retry( + Schedule.exponential("5 seconds").pipe( + Schedule.modifyDelay(({ duration }) => + Effect.succeed(Duration.min(duration, Duration.minutes(10))), + ), + ), + ), + ); + class KeepAwakeInhibitorExited extends Schema.TaggedError()( "KeepAwakeInhibitorExited", { exitCode: Schema.Number }, @@ -94,15 +110,13 @@ export const layer = Layer.effectDiscard( WHERE status IN ('preparing', 'starting', 'running') ) AS active `.pipe( - Effect.map((rows) => [rows[0]?.active === 1]), - // A failed read keeps the last known state rather than guessing. - Effect.catchCause((cause) => - Effect.logWarning("keep-awake could not read active runs", { cause }).pipe(Effect.as([])), - ), + Effect.map((rows) => rows[0]?.active === 1), + retryWithBackoff("keep-awake could not read active runs"), ); // Read the cursor before the snapshot so a run changing in between is replayed, not missed. const active = Stream.unwrap( events.latestApplicationSequence.pipe( + retryWithBackoff("keep-awake could not read the event cursor"), Effect.map((afterSequence) => Stream.concat( Stream.succeed(undefined), @@ -118,10 +132,7 @@ export const layer = Layer.effectDiscard( ), ), ), - ).pipe( - Stream.mapEffect(() => hasActiveRun), - Stream.flattenIterable, - ); + ).pipe(Stream.mapEffect(() => hasActiveRun)); const enabled = Stream.unwrap( Effect.gen(function* () { const changes = yield* settings.subscribeChanges; @@ -159,22 +170,8 @@ export const layer = Layer.effectDiscard( ), Effect.scoped, ); - // An inhibitor that dies while agents still work is restarted; on hosts without one - // (e.g. no logind) the backoff keeps retries and warnings rare. const hold = holdOnce.pipe( - Effect.tapCause((cause) => - Effect.logWarning("keep-awake sleep inhibitor stopped", { - command: command.command, - cause, - }), - ), - Effect.retry( - Schedule.exponential("5 seconds").pipe( - Schedule.modifyDelay(({ duration }) => - Effect.succeed(Duration.min(duration, Duration.minutes(10))), - ), - ), - ), + retryWithBackoff("keep-awake sleep inhibitor stopped"), Effect.ignore, ); From 7b21f61a7cf756fbfc2b8be136eb3cb836d28e44 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 12 Nov 2026 16:13:49 -0800 Subject: [PATCH 4/4] fix(server): keep-awake resubscribes when the run event stream fails Co-Authored-By: Claude Opus 5.5 --- apps/server/src/background/KeepAwake.ts | 26 +++++++++++++++---------- 1 file changed, 16 insertions(+), 10 deletions(-) diff --git a/apps/server/src/background/KeepAwake.ts b/apps/server/src/background/KeepAwake.ts index 24268e53186b..5c9c3efc502e 100644 --- a/apps/server/src/background/KeepAwake.ts +++ b/apps/server/src/background/KeepAwake.ts @@ -74,20 +74,19 @@ export const holdWhileWanted = ( Stream.runDrain, ); -// Reads and the inhibitor retry until they succeed; the cap keeps a host without an -// inhibitor (e.g. no logind) from retrying and warning constantly. +// Reads, the event subscription and the inhibitor retry until they succeed; the cap keeps a +// host without an inhibitor (e.g. no logind) from retrying and warning constantly. +const backoff = Schedule.exponential("5 seconds").pipe( + Schedule.modifyDelay(({ duration }) => + Effect.succeed(Duration.min(duration, Duration.minutes(10))), + ), +); const retryWithBackoff = (message: string) => (effect: Effect.Effect) => effect.pipe( Effect.tapCause((cause) => Effect.logWarning(message, { cause })), - Effect.retry( - Schedule.exponential("5 seconds").pipe( - Schedule.modifyDelay(({ duration }) => - Effect.succeed(Duration.min(duration, Duration.minutes(10))), - ), - ), - ), + Effect.retry(backoff), ); class KeepAwakeInhibitorExited extends Schema.TaggedError()( @@ -132,7 +131,14 @@ export const layer = Layer.effectDiscard( ), ), ), - ).pipe(Stream.mapEffect(() => hasActiveRun)); + ).pipe( + Stream.mapEffect(() => hasActiveRun), + // A failed subscription resubscribes, which re-reads the cursor and the snapshot. + Stream.tapCause((cause) => + Effect.logWarning("keep-awake run event stream failed", { cause }), + ), + Stream.retry(backoff), + ); const enabled = Stream.unwrap( Effect.gen(function* () { const changes = yield* settings.subscribeChanges;