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..8984102729ce --- /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.andThen(Effect.never), Effect.scoped); + 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..5c9c3efc502e --- /dev/null +++ b/apps/server/src/background/KeepAwake.ts @@ -0,0 +1,191 @@ +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 * 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 OrchestrationEventStore from "../persistence/OrchestrationEventStore.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; + } +} + +/** + * 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, + 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.fromEffect(hold) : Stream.empty)), + Stream.runDrain, + ); + +// 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(backoff), + ); + +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 events = yield* OrchestrationEventStore.OrchestrationEventStore; + 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), + 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), + events + .streamApplicationEvents({ afterSequence }) + .pipe( + Stream.filter( + (stored) => + "event" in stored && + (stored.event.type === "run.created" || stored.event.type === "run.updated"), + ), + ), + ), + ), + ), + ).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; + 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 holdOnce = 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.flatMap((child) => child.exitCode), + Effect.flatMap((exitCode) => + Effect.fail(new KeepAwakeInhibitorExited({ exitCode: Number(exitCode) })), + ), + Effect.scoped, + ); + const hold = holdOnce.pipe( + retryWithBackoff("keep-awake sleep inhibitor stopped"), + Effect.ignore, + ); + + 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({