1import {
2 type DeviceHostSummary,
3 DevicePlatformAvailability,
4 DeviceToolVersions,
5 deviceToolInstallMessage,
6 type SshDeviceHostConfig,
7} from "@t3tools/contracts";
8import { runSshCommand, baseSshArgs, resolveSshCommand } from "@t3tools/ssh/command";
9import * as NetService from "@t3tools/shared/Net";
10import { waitForHttpReady } from "@t3tools/shared/httpReadiness";
11import * as Crypto from "effect/Crypto";
12import * as Exit from "effect/Exit";
13import * as Effect from "effect/Effect";
14import * as FileSystem from "effect/FileSystem";
15import * as Path from "effect/Path";
16import * as Scope from "effect/Scope";
17import * as Semaphore from "effect/Semaphore";
18import * as Schema from "effect/Schema";
19import * as Stream from "effect/Stream";
20import * as Hex from "effect/encoding/Hex";
21import * as HttpClient from "effect/http/HttpClient";
22import * as ChildProcess from "effect/process/ChildProcess";
23import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner";
24import * as ServerConfig from "../config.ts";
25import * as DeviceHost from "./DeviceHost.ts";
26import { quoteRemoteArg, remoteDeviceEnvironment, remoteDeviceScript } from "./sshDeviceScript.ts";
27
28const Probe = Schema.Struct({
29 nodePath: Schema.String,
30 tools: Schema.optional(DeviceToolVersions),
31 platforms: Schema.Array(DevicePlatformAvailability),
32});
33const Started = Schema.Struct({
34 ...Probe.fields,
35 hubPort: Schema.Int,
36 daemonPort: Schema.optionalKey(Schema.Int),
37 token: Schema.optionalKey(Schema.String),
38 entryPath: Schema.optionalKey(Schema.String),
39 helpers: Schema.Struct({
40 serveSimAxSettings: Schema.NullOr(Schema.String),
41 serveSimCli: Schema.NullOr(Schema.String),
42 }),
43});
44const decodeProbe = Schema.decodeUnknownEffect(Schema.fromJsonString(Probe));
45const decodeStarted = Schema.decodeUnknownEffect(Schema.fromJsonString(Started));
46const targetFor = (config: SshDeviceHostConfig) => ({
47 alias: config.target,
48 hostname: config.target,
49 username: null,
50 port: config.port ?? null,
51});
52const identityArgs = (config: SshDeviceHostConfig) =>
53 config.identityFile ? ["-i", config.identityFile] : [];
54const commandArgs = (script: string) => [
55 "sh",
56 "-c",
57 quoteRemoteArg(remoteDeviceEnvironment + script),
58];
59const bootstrap = (
60 config: SshDeviceHostConfig,
61 owner: string,
62 mode: "probe" | "start" | "agent-start" | "stop-agent" | "stop",
63) =>
64 runSshCommand(targetFor(config), {
65 preHostArgs: identityArgs(config),
66 remoteCommandArgs: commandArgs(
67 'command -v node >/dev/null 2>&1 || { echo "Node is missing from the non-interactive SSH PATH" >&2; exit 1; }; exec node',
68 ),
69 stdin: remoteDeviceScript(owner, mode),
70 timeoutMs: mode === "start" || mode === "agent-start" ? 1_300_000 : 45_000,
71 }).pipe(
72 Effect.mapError(
73 (cause) => new DeviceHost.DeviceHostError({ hostId: config.id, step: mode, cause }),
74 ),
75 );
76
77const ownerFor = Effect.fn("SshDeviceHost.ownerFor")(function* (hostId: string) {
78 const fs = yield* FileSystem.FileSystem;
79 const server = yield* ServerConfig.ServerConfig;
80 const environmentId = yield* fs
81 .readFileString(server.environmentIdPath)
82 .pipe(Effect.orElseSucceed(() => server.stateDir));
83 const crypto = yield* Crypto.Crypto;
84 const owner = yield* crypto
85 .digest("SHA-256", new TextEncoder().encode(`${environmentId}\0${server.stateDir}\0${hostId}`))
86 .pipe(Effect.map(Hex.encode), Effect.orDie);
87 return owner.slice(0, 24);
88});
89
90export const probe = Effect.fn("SshDeviceHost.probe")(function* (
91 config: SshDeviceHostConfig,
92 owner?: string,
93) {
94 const result = yield* bootstrap(config, owner ?? (yield* ownerFor(config.id)), "probe");
95 const value = yield* decodeProbe(result.stdout.trim()).pipe(
96 Effect.mapError(
97 (cause) =>
98 new DeviceHost.DeviceHostError({ hostId: config.id, step: "reading probe result", cause }),
99 ),
100 );
101 return {
102 id: config.id,
103 label: config.label,
104 kind: "ssh",
105 tools: value.tools,
106 hubInstalled:
107 value.tools?.hub.installedVersions.includes(value.tools.hub.requiredVersion) ?? false,
108 agentDeviceInstalled:
109 value.tools?.agent.installedVersions.includes(value.tools.agent.requiredVersion) ?? false,
110 platforms: value.platforms,
111 } satisfies DeviceHostSummary;
112});
113
114export const make = Effect.fn("SshDeviceHost.make")(function* (
115 config: SshDeviceHostConfig,
116 onReady: (
117 ready: DeviceHost.DeviceHostAgentReady,
118 ) => Effect.Effect<void, DeviceHost.DeviceHostError> = () => Effect.void,
119 onStatus: (
120 status: "starting" | "ready" | "failed",
121 detail?: string,
122 ) => Effect.Effect<void> = () => Effect.void,
123) {
124 const fs = yield* FileSystem.FileSystem;
125 const path = yield* Path.Path;
126 const server = yield* ServerConfig.ServerConfig;
127 const net = yield* NetService.NetService;
128 const http = yield* HttpClient.HttpClient;
129 const spawner = yield* ChildProcessSpawner.ChildProcessSpawner;
130 const crypto = yield* Crypto.Crypto;
131 const parentScope = yield* Scope.Scope;
132 const ssh = yield* resolveSshCommand;
133 const owner = yield* ownerFor(config.id);
134 const provide = <A, E>(
135 effect: Effect.Effect<
136 A,
137 E,
138 | FileSystem.FileSystem
139 | Path.Path
140 | ChildProcessSpawner.ChildProcessSpawner
141 | Crypto.Crypto
142 | ServerConfig.ServerConfig
143 >,
144 ) =>
145 effect.pipe(
146 Effect.provideService(FileSystem.FileSystem, fs),
147 Effect.provideService(Path.Path, path),
148 Effect.provideService(ServerConfig.ServerConfig, server),
149 Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner),
150 Effect.provideService(Crypto.Crypto, crypto),
151 );
152 const lock = yield* Semaphore.make(1);
153 let stopped = false;
154 let activated = false;
155 let wantsAgent = false;
156 let ready:
157 | (DeviceHost.DeviceHostReady & {
158 agentDevice?: DeviceHost.DeviceHostAgentReady["agentDevice"];
159 })
160 | null = null;
161 let connectionScope: Scope.Closeable | null = null;
162 let summary: DeviceHostSummary = {
163 id: config.id,
164 label: config.label,
165 kind: "ssh",
166 hubInstalled: false,
167 agentDeviceInstalled: false,
168 platforms: [],
169 };
170
171 const run: DeviceHost.DeviceHostReady["run"] = (command, args, options) =>
172 provide(
173 runSshCommand(targetFor(config), {
174 preHostArgs: identityArgs(config),
175 remoteCommandArgs: commandArgs(`exec ${[command, ...args].map(quoteRemoteArg).join(" ")}`),
176 ...(options?.stdin === undefined ? {} : { stdin: options.stdin }),
177 ...(options?.timeoutMs === undefined ? {} : { timeoutMs: options.timeoutMs }),
178 }),
179 ).pipe(
180 Effect.map((result) => ({ ...result, code: 0 })),
181 Effect.catch((error) =>
182 Effect.succeed({
183 stdout: "stdout" in error ? (error.stdout ?? "") : "",
184 stderr: error.message,
185 code: "exitCode" in error ? (error.exitCode ?? 127) : 127,
186 }),
187 ),
188 );
189
190 const connectOnce = Effect.fn("SshDeviceHost.connectOnce")(function* (): Effect.fn.Return<
191 DeviceHost.DeviceHostReady & { agentDevice?: DeviceHost.DeviceHostAgentReady["agentDevice"] },
192 DeviceHost.DeviceHostError
193 > {
194 activated = true;
195 const result = yield* provide(bootstrap(config, owner, wantsAgent ? "agent-start" : "start"));
196 yield* onStatus("starting");
197 const remote = yield* decodeStarted(result.stdout.trim()).pipe(
198 Effect.mapError(
199 (cause) =>
200 new DeviceHost.DeviceHostError({
201 hostId: config.id,
202 step: "reading host endpoints",
203 cause,
204 }),
205 ),
206 );
207 summary = {
208 ...summary,
209 platforms: remote.platforms,
210 tools: remote.tools,
211 hubInstalled: true,
212 agentDeviceInstalled: wantsAgent || summary.agentDeviceInstalled,
213 };
214 const hubPort = yield* net.reserveLoopbackPort("127.0.0.1").pipe(
215 Effect.mapError(
216 (cause) =>
217 new DeviceHost.DeviceHostError({
218 hostId: config.id,
219 step: "reserving hub port",
220 cause,
221 }),
222 ),
223 );
224 const daemonPort = yield* net.reserveLoopbackPort("127.0.0.1").pipe(
225 Effect.mapError(
226 (cause) =>
227 new DeviceHost.DeviceHostError({
228 hostId: config.id,
229 step: "reserving daemon port",
230 cause,
231 }),
232 ),
233 );
234 const scope = yield* Scope.make();
235 connectionScope = scope;
236 const child = yield* spawner
237 .spawn(
238 ChildProcess.make(
239 ssh,
240 [
241 ...baseSshArgs(targetFor(config), { batchMode: "yes" }),
242 ...identityArgs(config),
243 "-o",
244 "ExitOnForwardFailure=yes",
245 "-o",
246 "ServerAliveInterval=10",
247 "-o",
248 "ServerAliveCountMax=3",
249 "-N",
250 "-L",
251 `127.0.0.1:${hubPort}:127.0.0.1:${remote.hubPort}`,
252 ...(remote.daemonPort === undefined
253 ? []
254 : ["-L", `127.0.0.1:${daemonPort}:127.0.0.1:${remote.daemonPort}`]),
255 config.target,
256 ],
257 { stdin: "ignore", stdout: "ignore", stderr: "pipe" },
258 ),
259 )
260 .pipe(
261 Effect.provideService(Scope.Scope, scope),
262 Effect.mapError(
263 (cause) =>
264 new DeviceHost.DeviceHostError({ hostId: config.id, step: "forwarding ports", cause }),
265 ),
266 );
267 let stderr = "";
268 yield* child.stderr.pipe(
269 Stream.decodeText(),
270 Stream.runForEach((chunk) =>
271 Effect.sync(() => {
272 stderr = (stderr + chunk).slice(-2000);
273 }),
274 ),
275 Effect.forkIn(scope),
276 );
277 const next = {
278 nodePath: remote.nodePath,
279 hub: { origin: `http://127.0.0.1:${hubPort}` },
280 ...(remote.daemonPort !== undefined &&
281 remote.token !== undefined &&
282 remote.entryPath !== undefined
283 ? {
284 agentDevice: {
285 baseUrl: `http://127.0.0.1:${daemonPort}`,
286 token: remote.token,
287 entryPath: remote.entryPath,
288 },
289 }
290 : {}),
291 helpers: remote.helpers,
292 run,
293 };
294 for (const [baseUrl, route] of [
295 [next.hub.origin, "/readyz"],
296 ...(next.agentDevice ? [[next.agentDevice.baseUrl, "/health"]] : []),
297 ]) {
298 yield* waitForHttpReady({
299 baseUrl: baseUrl!,
300 path: route!,
301 timeoutMs: 15000,
302 makeError: () =>
303 new DeviceHost.DeviceHostError({
304 hostId: config.id,
305 step: "waiting for SSH forward",
306 cause: new Error(stderr || "Forwarded endpoint did not answer."),
307 }),
308 }).pipe(Effect.provideService(HttpClient.HttpClient, http));
309 }
310 if (next.agentDevice) yield* onReady({ ...next, agentDevice: next.agentDevice });
311 ready = next;
312 yield* onStatus("ready");
313 // Reconnect also repairs helpers that died while SSH itself stayed connected.
314 const unhealthy = Effect.gen(function* () {
315 while (true) {
316 yield* Effect.sleep("10 seconds");
317 const alive = yield* http.get(`${next.hub.origin}/readyz`).pipe(
318 Effect.timeout("5 seconds"),
319 Effect.map((r) => r.status === 200),
320 Effect.orElseSucceed(() => false),
321 );
322 const daemonAlive = next.agentDevice
323 ? yield* http.get(`${next.agentDevice!.baseUrl}/health`).pipe(
324 Effect.timeout("5 seconds"),
325 Effect.map((r) => r.status === 200),
326 Effect.orElseSucceed(() => false),
327 )
328 : true;
329 if (!alive || !daemonAlive) return;
330 }
331 });
332 yield* Effect.gen(function* () {
333 yield* Effect.raceFirst(child.exitCode.pipe(Effect.ignore), unhealthy);
334 if (stopped || connectionScope !== scope) return;
335 ready = null;
336 yield* onStatus("starting", "Reconnecting to device host…");
337 yield* Scope.close(scope, Exit.void);
338 let delay = 1000;
339 while (true) {
340 if (stopped || connectionScope !== scope) return;
341 yield* Effect.sleep(delay);
342 const result = yield* lock
343 .withPermit(
344 Effect.suspend(() => (stopped || ready ? Effect.void : connect().pipe(Effect.asVoid))),
345 )
346 .pipe(Effect.result);
347 if (result._tag === "Success") return;
348 yield* onStatus("failed", result.failure.message);
349 if (connectionScope && connectionScope !== scope)
350 yield* Scope.close(connectionScope, Exit.void);
351 connectionScope = scope;
352 delay = Math.min(delay * 2, 30000);
353 }
354 }).pipe(Effect.forkIn(parentScope));
355 return next;
356 });
357
358 const connect = Effect.fn("SshDeviceHost.connect")(function* () {
359 for (let attempt = 0; ; attempt++) {
360 const result = yield* connectOnce().pipe(Effect.result);
361 if (result._tag === "Success") return result.success;
362 const failedScope = connectionScope;
363 connectionScope = null;
364 if (failedScope) yield* Scope.close(failedScope, Exit.void);
365 // SSH binds after the reservation is released, so a competing bind needs fresh ports.
366 if (
367 attempt >= 2 ||
368 !["forwarding ports", "waiting for SSH forward"].includes(result.failure.step)
369 )
370 return yield* result.failure;
371 }
372 });
373
374 const ensureReady: DeviceHost.DeviceHost["Service"]["ensureReady"] = (onPhase) =>
375 lock.withPermit(
376 Effect.gen(function* () {
377 stopped = false;
378 if (ready) return ready;
379 summary = yield* provide(probe(config, owner));
380 yield* onPhase(
381 summary.hubInstalled ? "starting" : "installing",
382 summary.hubInstalled
383 ? undefined
384 : deviceToolInstallMessage("device hub", summary.tools?.hub),
385 );
386 return yield* connect().pipe(
387 Effect.tapError(() =>
388 connectionScope ? Scope.close(connectionScope, Exit.void) : Effect.void,
389 ),
390 );
391 }),
392 );
393 const stop = lock.withPermit(
394 Effect.gen(function* () {
395 stopped = true;
396 ready = null;
397 if (connectionScope) yield* Scope.close(connectionScope, Exit.void);
398 connectionScope = null;
399 if (activated) yield* provide(bootstrap(config, owner, "stop")).pipe(Effect.ignore);
400 activated = false;
401 wantsAgent = false;
402 }),
403 );
404 const changeAgent = (enabled: boolean) =>
405 lock.withPermit(
406 Effect.gen(function* () {
407 wantsAgent = enabled;
408 if (enabled && ready?.agentDevice) return { ...ready, agentDevice: ready.agentDevice };
409 if (!enabled && !ready?.agentDevice) return null;
410 ready = null;
411 const previousScope = connectionScope;
412 connectionScope = null;
413 if (previousScope) yield* Scope.close(previousScope, Exit.void);
414 if (!enabled) yield* provide(bootstrap(config, owner, "stop-agent"));
415 return yield* connect().pipe(
416 Effect.onError(() =>
417 Effect.gen(function* () {
418 const failedScope = connectionScope;
419 connectionScope = null;
420 if (failedScope) yield* Scope.close(failedScope, Exit.void);
421 if (enabled)
422 yield* provide(bootstrap(config, owner, "stop-agent")).pipe(Effect.ignore);
423 }),
424 ),
425 );
426 }),
427 );
428 yield* Effect.addFinalizer(() => stop);
429 return {
430 id: config.id,
431 summary: Effect.sync(() => summary),
432 inspect: provide(probe(config, owner)).pipe(
433 Effect.tap((value) =>
434 Effect.sync(() => {
435 summary = value;
436 }),
437 ),
438 ),
439 current: Effect.sync(() => ready),
440 ensureReady,
441 ensureAgentReady: (onPhase) =>
442 ensureReady(onPhase).pipe(
443 Effect.flatMap(() =>
444 onPhase(
445 summary.agentDeviceInstalled ? "starting" : "installing",
446 summary.agentDeviceInstalled
447 ? undefined
448 : deviceToolInstallMessage("agent tools", summary.tools?.agent),
449 ),
450 ),
451 Effect.flatMap(() => changeAgent(true)),
452 Effect.flatMap((value) =>
453 value?.agentDevice
454 ? Effect.succeed({ ...value, agentDevice: value.agentDevice })
455 : Effect.fail(
456 new DeviceHost.DeviceHostError({
457 hostId: config.id,
458 step: "starting agent tools",
459 cause: new Error("Daemon endpoint missing"),
460 }),
461 ),
462 ),
463 ),
464 stopAgent: changeAgent(false).pipe(Effect.asVoid, Effect.ignore),
465 stop,
466 platformAvailability: (platform) =>
467 provide(probe(config, owner)).pipe(
468 Effect.map((value) => {
469 summary = { ...summary, platforms: value.platforms };
470 return value.platforms.find((p) => p.platform === platform)!;
471 }),
472 Effect.orElseSucceed(() => ({
473 platform,
474 available: false,
475 reason: "Cannot reach device host. Test its SSH connection in Settings.",
476 })),
477 ),
478 } satisfies DeviceHost.DeviceHost["Service"];
479});