1import * as Duration from "effect/Duration";
2import * as Effect from "effect/Effect";
3import * as Layer from "effect/Layer";
4import * as ChildProcess from "effect/process/ChildProcess";
5import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner";
6import * as Schedule from "effect/Schedule";
7import * as Schema from "effect/Schema";
8import * as Stream from "effect/Stream";
9import * as SqlClient from "effect/sql/SqlClient";
10
11import * as HostProcess from "@t3tools/shared/HostProcess";
12
13import * as OrchestrationEventStore from "../persistence/OrchestrationEventStore.ts";
14import { forkParked } from "../serverActivation.ts";
15import * as Settings from "../serverSettings.ts";
16
17// Bridges gaps between back-to-back runs, so a long-idle machine does not sleep mid-queue.
18const RELEASE_GRACE = Duration.seconds(60);
19
20/** Each command holds its inhibitor until it exits, and exits on its own if the server dies. */
21function inhibitorCommand(
22 platform: NodeJS.Platform,
23 serverPid: number,
24): { readonly command: string; readonly args: ReadonlyArray<string> } | undefined {
25 switch (platform) {
26 case "darwin":
27 return { command: "caffeinate", args: ["-i", "-w", String(serverPid)] };
28 case "linux":
29 return {
30 command: "systemd-inhibit",
31 args: [
32 "--what=idle:sleep",
33 "--who=T3 Code",
34 "--why=Agents are working",
35 "--mode=block",
36 "tail",
37 `--pid=${serverPid}`,
38 "-f",
39 "/dev/null",
40 ],
41 };
42 case "win32":
43 return {
44 command: "powershell.exe",
45 args: [
46 "-NoProfile",
47 "-NonInteractive",
48 "-Command",
49 // ES_CONTINUOUS | ES_SYSTEM_REQUIRED, held by this thread until Wait-Process returns.
50 `$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}`,
51 ],
52 };
53 default:
54 return undefined;
55 }
56}
57
58/**
59 * Runs `hold` while `wanted` is true and for `grace` after it turns false; a true inside the
60 * grace keeps the same hold. `hold` runs until interrupted.
61 */
62export const holdWhileWanted = <E, R>(
63 wanted: Stream.Stream<boolean, E, R>,
64 hold: Effect.Effect<void>,
65 grace: Duration.Input = RELEASE_GRACE,
66) =>
67 wanted.pipe(
68 Stream.changes,
69 Stream.switchMap((on) =>
70 on ? Stream.succeed(true) : Stream.fromEffect(Effect.as(Effect.sleep(grace), false)),
71 ),
72 Stream.changes,
73 Stream.switchMap((on) => (on ? Stream.fromEffect(hold) : Stream.empty)),
74 Stream.runDrain,
75 );
76
77// Reads, the event subscription and the inhibitor retry until they succeed; the cap keeps a
78// host without an inhibitor (e.g. no logind) from retrying and warning constantly.
79const backoff = Schedule.exponential("5 seconds").pipe(
80 Schedule.modifyDelay(({ duration }) =>
81 Effect.succeed(Duration.min(duration, Duration.minutes(10))),
82 ),
83);
84const retryWithBackoff =
85 (message: string) =>
86 <A, E, R>(effect: Effect.Effect<A, E, R>) =>
87 effect.pipe(
88 Effect.tapCause((cause) => Effect.logWarning(message, { cause })),
89 Effect.retry(backoff),
90 );
91
92class KeepAwakeInhibitorExited extends Schema.TaggedError<KeepAwakeInhibitorExited>()(
93 "KeepAwakeInhibitorExited",
94 { exitCode: Schema.Number },
95) {}
96
97export const layer = Layer.effectDiscard(
98 Effect.gen(function* () {
99 const command = inhibitorCommand(yield* HostProcess.Platform, process.pid);
100 if (command === undefined) return;
101 const settings = yield* Settings.ServerSettingsService;
102 const events = yield* OrchestrationEventStore.OrchestrationEventStore;
103 const sql = yield* SqlClient.SqlClient;
104 const spawner = yield* ChildProcessSpawner.ChildProcessSpawner;
105
106 const hasActiveRun = sql<{ readonly active: number }>`
107 SELECT EXISTS (
108 SELECT 1 FROM orchestration_v2_projection_runs
109 WHERE status IN ('preparing', 'starting', 'running')
110 ) AS active
111 `.pipe(
112 Effect.map((rows) => rows[0]?.active === 1),
113 retryWithBackoff("keep-awake could not read active runs"),
114 );
115 // Read the cursor before the snapshot so a run changing in between is replayed, not missed.
116 const active = Stream.unwrap(
117 events.latestApplicationSequence.pipe(
118 retryWithBackoff("keep-awake could not read the event cursor"),
119 Effect.map((afterSequence) =>
120 Stream.concat(
121 Stream.succeed(undefined),
122 events
123 .streamApplicationEvents({ afterSequence })
124 .pipe(
125 Stream.filter(
126 (stored) =>
127 "event" in stored &&
128 (stored.event.type === "run.created" || stored.event.type === "run.updated"),
129 ),
130 ),
131 ),
132 ),
133 ),
134 ).pipe(
135 Stream.mapEffect(() => hasActiveRun),
136 // A failed subscription resubscribes, which re-reads the cursor and the snapshot.
137 Stream.tapCause((cause) =>
138 Effect.logWarning("keep-awake run event stream failed", { cause }),
139 ),
140 Stream.retry(backoff),
141 );
142 const enabled = Stream.unwrap(
143 Effect.gen(function* () {
144 const changes = yield* settings.subscribeChanges;
145 const current = yield* settings.getSettings.pipe(
146 Effect.map((value) => value.keepAwakeWhileAgentsWork),
147 Effect.orElseSucceed(() => true),
148 );
149 return Stream.concat(
150 Stream.succeed(current),
151 Stream.map(changes, (value) => value.keepAwakeWhileAgentsWork),
152 );
153 }),
154 );
155
156 const holdOnce = spawner
157 .spawn(
158 ChildProcess.make(command.command, command.args, {
159 detached: false,
160 shell: false,
161 stdin: "ignore",
162 stdout: "ignore",
163 stderr: "ignore",
164 }),
165 )
166 .pipe(
167 Effect.tap((child) =>
168 Effect.logInfo("keep-awake holding sleep inhibitor", { pid: Number(child.pid) }),
169 ),
170 Effect.tap(() =>
171 Effect.addFinalizer(() => Effect.logInfo("keep-awake released sleep inhibitor")),
172 ),
173 Effect.flatMap((child) => child.exitCode),
174 Effect.flatMap((exitCode) =>
175 Effect.fail(new KeepAwakeInhibitorExited({ exitCode: Number(exitCode) })),
176 ),
177 Effect.scoped,
178 );
179 const hold = holdOnce.pipe(
180 retryWithBackoff("keep-awake sleep inhibitor stopped"),
181 Effect.ignore,
182 );
183
184 yield* forkParked(
185 holdWhileWanted(
186 Stream.zipLatestWith(enabled, active, (on, busy) => on && busy),
187 hold,
188 ).pipe(Effect.catchCause((cause) => Effect.logWarning("keep-awake stopped", { cause }))),
189 );
190 }),
191);