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 { SshCommandError } from "@t3tools/ssh/errors";
10import * as NetService from "@t3tools/shared/Net";
11import { waitForHttpReady } from "@t3tools/shared/httpReadiness";
12import * as Digest from "effect/crypto/Digest";
13import * as Exit from "effect/Exit";
14import * as Effect from "effect/Effect";
15import * as FileSystem from "effect/FileSystem";
16import * as Path from "effect/Path";
17import * as Scope from "effect/Scope";
18import * as Semaphore from "effect/Semaphore";
19import * as Schema from "effect/Schema";
20import * as Stream from "effect/Stream";
21import * as Hex from "effect/encoding/Hex";
22import * as HttpClient from "effect/http/HttpClient";
23import * as ChildProcess from "effect/process/ChildProcess";
24import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner";
25import * as ServerConfig from "../config.ts";
26import { releaseAgentDeviceSession } from "./AgentDeviceTarget.ts";
27import * as DeviceHost from "./DeviceHost.ts";
28import { makeDeviceHubOperations } from "./DeviceHubOperations.ts";
29import { quoteRemoteArg, remoteDeviceEnvironment, remoteDeviceScript } from "./sshDeviceScript.ts";
30
31const UNHEALTHY_PROBE_LIMIT = 3;
32
33/** The host bootstrap failed on a host identified as Windows. */
34export class SshWindowsHostError extends Schema.TaggedError<SshWindowsHostError>()(
35 "SshWindowsHostError",
36 { cause: Schema.Defect() },
37) {
38 override get message(): string {
39 return "Connected over SSH, but this host runs Windows, and SSH device hosts currently need macOS or Linux. To use this machine's devices, run T3 Code on it and connect to that environment.";
40 }
41}
42export const isSshWindowsHostError = Schema.is(SshWindowsHostError);
43
44const Probe = Schema.Struct({
45 nodePath: Schema.String,
46 tools: Schema.optional(DeviceToolVersions),
47 platforms: Schema.Array(DevicePlatformAvailability),
48});
49const Started = Schema.Struct({
50 ...Probe.fields,
51 hubPort: Schema.Int,
52 daemonPort: Schema.optionalKey(Schema.Int),
53 token: Schema.optionalKey(Schema.String),
54 entryPath: Schema.optionalKey(Schema.String),
55 helpers: Schema.Struct({
56 serveSimAxSettings: Schema.NullOr(Schema.String),
57 serveSimCli: Schema.NullOr(Schema.String),
58 }),
59});
60const decodeProbe = Schema.decodeUnknownEffect(Schema.fromJsonString(Probe));
61const decodeStarted = Schema.decodeUnknownEffect(Schema.fromJsonString(Started));
62const targetFor = (config: SshDeviceHostConfig) => ({
63 alias: config.target,
64 hostname: config.target,
65 username: null,
66 port: config.port ?? null,
67});
68const identityArgs = (config: SshDeviceHostConfig) =>
69 config.identityFile ? ["-i", config.identityFile] : [];
70const commandArgs = (script: string) => [
71 "sh",
72 "-c",
73 quoteRemoteArg(remoteDeviceEnvironment + script),
74];
75const bootstrap = (
76 config: SshDeviceHostConfig,
77 owner: string,
78 mode: "probe" | "start" | "agent-start" | "stop-agent" | "stop",
79) =>
80 runSshCommand(targetFor(config), {
81 preHostArgs: identityArgs(config),
82 remoteCommandArgs: commandArgs(
83 'command -v node >/dev/null 2>&1 || { echo "Node is missing from the non-interactive SSH PATH" >&2; exit 1; }; exec node',
84 ),
85 stdin: remoteDeviceScript(owner, mode),
86 timeoutMs: mode === "start" || mode === "agent-start" ? 1_300_000 : 45_000,
87 }).pipe(
88 Effect.mapError(
89 (cause) => new DeviceHost.DeviceHostError({ hostId: config.id, step: mode, cause }),
90 ),
91 );
92
93const isSshCommandError = Schema.is(SshCommandError);
94
95// cmd expands only %OS% and PowerShell only $env:OS; POSIX shells expand neither to Windows_NT.
96const explainWindowsHost = Effect.fn("SshDeviceHost.explainWindowsHost")(function* (
97 config: SshDeviceHostConfig,
98 error: DeviceHost.DeviceHostError,
99) {
100 const cause = error.cause;
101 // Windows ssh.exe preserves full remote exit statuses such as 9009.
102 if (
103 !isSshCommandError(cause) ||
104 cause.exitCode === null ||
105 cause.exitCode < 1 ||
106 (cause.exitCode > 254 && cause.exitCode !== 9009)
107 )
108 return yield* error;
109 let windows = yield* runSshCommand(targetFor(config), {
110 preHostArgs: identityArgs(config),
111 remoteCommandArgs: ["echo %OS% $env:OS"],
112 timeoutMs: 15_000,
113 }).pipe(
114 Effect.map(({ stdout }) => /(?:^|\s)Windows_NT(?:\s|$)/u.test(stdout)),
115 Effect.orElseSucceed(() => false),
116 );
117 if (!windows) {
118 // Git Bash does not expand Windows shell variables, but its Node still reports win32.
119 windows = yield* runSshCommand(targetFor(config), {
120 preHostArgs: identityArgs(config),
121 remoteCommandArgs: ["node", "-p", "process.platform"],
122 timeoutMs: 15_000,
123 }).pipe(
124 Effect.map(({ stdout }) => stdout.trim() === "win32"),
125 Effect.orElseSucceed(() => false),
126 );
127 }
128 if (!windows) return yield* error;
129 return yield* new DeviceHost.DeviceHostError({
130 hostId: config.id,
131 step: error.step,
132 cause: new SshWindowsHostError({ cause }),
133 });
134});
135
136const ownerFor = Effect.fn("SshDeviceHost.ownerFor")(function* (hostId: string) {
137 const fs = yield* FileSystem.FileSystem;
138 const server = yield* ServerConfig.ServerConfig;
139 const environmentId = yield* fs
140 .readFileString(server.environmentIdPath)
141 .pipe(Effect.orElseSucceed(() => server.stateDir));
142 const digest = yield* Digest.Digest;
143 const owner = yield* digest
144 .digest(
145 Digest.sha256,
146 new TextEncoder().encode(`${environmentId}\0${server.stateDir}\0${hostId}`),
147 )
148 .pipe(Effect.map(Hex.encode), Effect.orDie);
149 return owner.slice(0, 24);
150});
151
152export const probe = Effect.fn("SshDeviceHost.probe")(function* (
153 config: SshDeviceHostConfig,
154 owner?: string,
155) {
156 const result = yield* bootstrap(config, owner ?? (yield* ownerFor(config.id)), "probe").pipe(
157 Effect.catchTags({ DeviceHostError: (error) => explainWindowsHost(config, error) }),
158 );
159 const value = yield* decodeProbe(result.stdout.trim()).pipe(
160 Effect.mapError(
161 (cause) =>
162 new DeviceHost.DeviceHostError({ hostId: config.id, step: "reading probe result", cause }),
163 ),
164 );
165 return {
166 id: config.id,
167 label: config.label,
168 kind: "ssh",
169 tools: value.tools,
170 hubInstalled:
171 value.tools?.hub.installedVersions.includes(value.tools.hub.requiredVersion) ?? false,
172 agentDeviceInstalled:
173 value.tools?.agent.installedVersions.includes(value.tools.agent.requiredVersion) ?? false,
174 platforms: value.platforms,
175 } satisfies DeviceHostSummary;
176});
177
178export const make = Effect.fn("SshDeviceHost.make")(function* (
179 config: SshDeviceHostConfig,
180 onReady: (
181 ready: DeviceHost.DeviceHostAgentReady,
182 ) => Effect.Effect<void, DeviceHost.DeviceHostError> = () => Effect.void,
183 onStatus: (
184 status: "starting" | "ready" | "failed",
185 detail?: string,
186 ) => Effect.Effect<void> = () => Effect.void,
187) {
188 const fs = yield* FileSystem.FileSystem;
189 const path = yield* Path.Path;
190 const server = yield* ServerConfig.ServerConfig;
191 const net = yield* NetService.NetService;
192 const http = yield* HttpClient.HttpClient;
193 const spawner = yield* ChildProcessSpawner.ChildProcessSpawner;
194 const digest = yield* Digest.Digest;
195 const parentScope = yield* Scope.Scope;
196 const ssh = yield* resolveSshCommand;
197 const owner = yield* ownerFor(config.id);
198 const provide = <A, E>(
199 effect: Effect.Effect<
200 A,
201 E,
202 | FileSystem.FileSystem
203 | Path.Path
204 | ChildProcessSpawner.ChildProcessSpawner
205 | Digest.Digest
206 | ServerConfig.ServerConfig
207 >,
208 ) =>
209 effect.pipe(
210 Effect.provideService(FileSystem.FileSystem, fs),
211 Effect.provideService(Path.Path, path),
212 Effect.provideService(ServerConfig.ServerConfig, server),
213 Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner),
214 Effect.provideService(Digest.Digest, digest),
215 );
216 const lock = yield* Semaphore.make(1);
217 let stopped = false;
218 let activated = false;
219 let wantsAgent = false;
220 let ready:
221 | (DeviceHost.DeviceHostReady & {
222 agentDevice?: DeviceHost.DeviceHostAgentReady["agentDevice"];
223 })
224 | null = null;
225 let connectionScope: Scope.Closeable | null = null;
226 let summary: DeviceHostSummary = {
227 id: config.id,
228 label: config.label,
229 kind: "ssh",
230 hubInstalled: false,
231 agentDeviceInstalled: false,
232 platforms: [],
233 };
234
235 const run: DeviceHost.DeviceHostReady["run"] = (command, args, options) =>
236 provide(
237 runSshCommand(targetFor(config), {
238 preHostArgs: identityArgs(config),
239 // Match the hub cwd so relative executables and PATH entries resolve identically.
240 remoteCommandArgs: commandArgs(
241 `cd "$HOME/.t3/device/hosts/${owner}" && exec ${[command, ...args].map(quoteRemoteArg).join(" ")}`,
242 ),
243 keepErrorOutputTail: true,
244 ...(options?.stdin === undefined ? {} : { stdin: options.stdin }),
245 ...(options?.timeoutMs === undefined ? {} : { timeoutMs: options.timeoutMs }),
246 }),
247 ).pipe(
248 Effect.map((result) => ({ ...result, code: 0 })),
249 Effect.catch((error) =>
250 Effect.succeed({
251 stdout: "stdout" in error ? (error.stdout ?? "") : "",
252 stderr: error.message,
253 code: "exitCode" in error ? (error.exitCode ?? 127) : 127,
254 }),
255 ),
256 );
257
258 const platformAvailability: DeviceHost.DeviceHost["Service"]["platformAvailability"] = (
259 platform,
260 ) =>
261 provide(probe(config, owner)).pipe(
262 Effect.map((value) => {
263 summary = { ...summary, platforms: value.platforms };
264 return value.platforms.find((p) => p.platform === platform)!;
265 }),
266 Effect.orElseSucceed(() => ({
267 platform,
268 available: false,
269 reason: "Cannot reach device host. Test its SSH connection in Settings.",
270 })),
271 );
272
273 const connectOnce = Effect.fn("SshDeviceHost.connectOnce")(function* (): Effect.fn.Return<
274 DeviceHost.DeviceHostReady & { agentDevice?: DeviceHost.DeviceHostAgentReady["agentDevice"] },
275 DeviceHost.DeviceHostError
276 > {
277 activated = true;
278 const result = yield* provide(bootstrap(config, owner, wantsAgent ? "agent-start" : "start"));
279 yield* onStatus("starting");
280 const remote = yield* decodeStarted(result.stdout.trim()).pipe(
281 Effect.mapError(
282 (cause) =>
283 new DeviceHost.DeviceHostError({
284 hostId: config.id,
285 step: "reading host endpoints",
286 cause,
287 }),
288 ),
289 );
290 summary = {
291 ...summary,
292 platforms: remote.platforms,
293 tools: remote.tools,
294 hubInstalled: true,
295 agentDeviceInstalled: wantsAgent || summary.agentDeviceInstalled,
296 };
297 const hubPort = yield* net.reserveLoopbackPort("127.0.0.1").pipe(
298 Effect.mapError(
299 (cause) =>
300 new DeviceHost.DeviceHostError({
301 hostId: config.id,
302 step: "reserving hub port",
303 cause,
304 }),
305 ),
306 );
307 const daemonPort = yield* net.reserveLoopbackPort("127.0.0.1").pipe(
308 Effect.mapError(
309 (cause) =>
310 new DeviceHost.DeviceHostError({
311 hostId: config.id,
312 step: "reserving daemon port",
313 cause,
314 }),
315 ),
316 );
317 const scope = yield* Scope.make();
318 connectionScope = scope;
319 const child = yield* spawner
320 .spawn(
321 ChildProcess.make(
322 ssh,
323 [
324 ...baseSshArgs(targetFor(config), { batchMode: "yes" }),
325 ...identityArgs(config),
326 "-o",
327 "ExitOnForwardFailure=yes",
328 // Multiplexed clients exit after handing forwards to the master; own this connection.
329 "-o",
330 "ControlMaster=no",
331 "-o",
332 "ControlPath=none",
333 "-o",
334 "ControlPersist=no",
335 "-o",
336 "ServerAliveInterval=10",
337 "-o",
338 "ServerAliveCountMax=3",
339 "-T",
340 "-L",
341 `127.0.0.1:${hubPort}:127.0.0.1:${remote.hubPort}`,
342 ...(remote.daemonPort === undefined
343 ? []
344 : ["-L", `127.0.0.1:${daemonPort}:127.0.0.1:${remote.daemonPort}`]),
345 config.target,
346 // The server holds the only write end of this stdin, so when it exits for any
347 // reason the remote `cat` sees EOF and ssh exits once its forwarded connections
348 // close, instead of being orphaned.
349 "sh",
350 "-c",
351 quoteRemoteArg("exec cat >/dev/null"),
352 ],
353 { stdin: "pipe", stdout: "ignore", stderr: "pipe" },
354 ),
355 )
356 .pipe(
357 Effect.provideService(Scope.Scope, scope),
358 Effect.mapError(
359 (cause) =>
360 new DeviceHost.DeviceHostError({ hostId: config.id, step: "forwarding ports", cause }),
361 ),
362 );
363 let stderr = "";
364 yield* child.stderr.pipe(
365 Stream.decodeText(),
366 Stream.runForEach((chunk) =>
367 Effect.sync(() => {
368 stderr = (stderr + chunk).slice(-2000);
369 }),
370 ),
371 Effect.forkIn(scope),
372 );
373 const hubOrigin = `http://127.0.0.1:${hubPort}`;
374 const next = {
375 nodePath: remote.nodePath,
376 hub: { origin: hubOrigin },
377 devices: makeDeviceHubOperations({
378 hostId: config.id,
379 origin: hubOrigin,
380 http,
381 run,
382 // Keep probing negative reports so tooling installed after connecting is still found.
383 platformAvailable: (platform) =>
384 Effect.suspend(() =>
385 summary.platforms.some((p) => p.platform === platform && p.available)
386 ? Effect.succeed(true)
387 : platformAvailability(platform).pipe(
388 Effect.map((availability) => availability.available),
389 ),
390 ),
391 }),
392 ...(remote.daemonPort !== undefined &&
393 remote.token !== undefined &&
394 remote.entryPath !== undefined
395 ? {
396 agentDevice: {
397 baseUrl: `http://127.0.0.1:${daemonPort}`,
398 token: remote.token,
399 entryPath: remote.entryPath,
400 },
401 }
402 : {}),
403 helpers: remote.helpers,
404 run,
405 };
406 for (const [baseUrl, route] of [
407 [next.hub.origin, "/readyz"],
408 ...(next.agentDevice ? [[next.agentDevice.baseUrl, "/health"]] : []),
409 ]) {
410 yield* waitForHttpReady({
411 baseUrl: baseUrl!,
412 path: route!,
413 timeoutMs: 15000,
414 makeError: () =>
415 new DeviceHost.DeviceHostError({
416 hostId: config.id,
417 step: "waiting for SSH forward",
418 cause: new Error(stderr || "Forwarded endpoint did not answer."),
419 }),
420 }).pipe(Effect.provideService(HttpClient.HttpClient, http));
421 }
422 if (next.agentDevice) yield* onReady({ ...next, agentDevice: next.agentDevice });
423 ready = next.agentDevice
424 ? { ...next, releaseAgentSession: releaseAgentDeviceSession(http, next.agentDevice) }
425 : next;
426 yield* onStatus("ready");
427 // Reconnect also repairs helpers that died while SSH itself stayed connected.
428 // A loaded host can miss a probe or two while still working, and reconnecting
429 // then drops the forward in use, so only consecutive misses count.
430 const unhealthy = Effect.gen(function* () {
431 let misses = 0;
432 while (true) {
433 yield* Effect.sleep("10 seconds");
434 const alive = yield* http.get(`${next.hub.origin}/readyz`).pipe(
435 Effect.timeout("5 seconds"),
436 Effect.map((r) => r.status === 200),
437 Effect.orElseSucceed(() => false),
438 );
439 const daemonAlive = next.agentDevice
440 ? yield* http.get(`${next.agentDevice!.baseUrl}/health`).pipe(
441 Effect.timeout("5 seconds"),
442 Effect.map((r) => r.status === 200),
443 Effect.orElseSucceed(() => false),
444 )
445 : true;
446 misses = alive && daemonAlive ? 0 : misses + 1;
447 if (misses >= UNHEALTHY_PROBE_LIMIT) return;
448 }
449 });
450 yield* Effect.gen(function* () {
451 yield* Effect.raceFirst(child.exitCode.pipe(Effect.ignore), unhealthy);
452 if (stopped || connectionScope !== scope) return;
453 ready = null;
454 yield* onStatus("starting", "Reconnecting to device host…");
455 yield* Scope.close(scope, Exit.void);
456 let delay = 1000;
457 while (true) {
458 if (stopped || connectionScope !== scope) return;
459 yield* Effect.sleep(delay);
460 const result = yield* lock
461 .withPermit(
462 Effect.suspend(() => (stopped || ready ? Effect.void : connect().pipe(Effect.asVoid))),
463 )
464 .pipe(Effect.result);
465 if (result._tag === "Success") return;
466 yield* onStatus("failed", result.failure.message);
467 if (connectionScope && connectionScope !== scope)
468 yield* Scope.close(connectionScope, Exit.void);
469 connectionScope = scope;
470 delay = Math.min(delay * 2, 30000);
471 }
472 }).pipe(Effect.forkIn(parentScope));
473 return next;
474 });
475
476 const closeConnection = Effect.suspend(() => {
477 const failedScope = connectionScope;
478 connectionScope = null;
479 ready = null;
480 return failedScope ? Scope.close(failedScope, Exit.void) : Effect.void;
481 });
482
483 const connect = Effect.fn("SshDeviceHost.connect")(function* () {
484 for (let attempt = 0; ; attempt++) {
485 const result = yield* connectOnce().pipe(
486 Effect.onInterrupt(() => closeConnection),
487 Effect.result,
488 );
489 if (result._tag === "Success") return result.success;
490 yield* closeConnection;
491 // SSH binds after the reservation is released, so a competing bind needs fresh ports.
492 if (
493 attempt >= 2 ||
494 !["forwarding ports", "waiting for SSH forward"].includes(result.failure.step)
495 )
496 return yield* result.failure;
497 }
498 });
499
500 const ensureReady: DeviceHost.DeviceHost["Service"]["ensureReady"] = (onPhase) =>
501 lock.withPermit(
502 Effect.gen(function* () {
503 stopped = false;
504 if (ready) return ready;
505 summary = yield* provide(probe(config, owner));
506 yield* onPhase(
507 summary.hubInstalled ? "starting" : "installing",
508 summary.hubInstalled
509 ? undefined
510 : deviceToolInstallMessage("device hub", summary.tools?.hub),
511 );
512 return yield* connect().pipe(
513 Effect.tapError(() =>
514 connectionScope ? Scope.close(connectionScope, Exit.void) : Effect.void,
515 ),
516 );
517 }),
518 );
519 const stop = lock.withPermit(
520 Effect.gen(function* () {
521 stopped = true;
522 ready = null;
523 if (connectionScope) yield* Scope.close(connectionScope, Exit.void);
524 connectionScope = null;
525 if (activated) yield* provide(bootstrap(config, owner, "stop")).pipe(Effect.ignore);
526 activated = false;
527 wantsAgent = false;
528 }),
529 );
530 const changeAgent = (enabled: boolean) =>
531 lock.withPermit(
532 Effect.gen(function* () {
533 wantsAgent = enabled;
534 if (enabled && ready?.agentDevice) return { ...ready, agentDevice: ready.agentDevice };
535 if (!enabled && !ready?.agentDevice) return null;
536 ready = null;
537 const previousScope = connectionScope;
538 connectionScope = null;
539 if (previousScope) yield* Scope.close(previousScope, Exit.void);
540 if (!enabled) yield* provide(bootstrap(config, owner, "stop-agent"));
541 return yield* connect().pipe(
542 Effect.onError(() =>
543 Effect.gen(function* () {
544 const failedScope = connectionScope;
545 connectionScope = null;
546 if (failedScope) yield* Scope.close(failedScope, Exit.void);
547 if (enabled)
548 yield* provide(bootstrap(config, owner, "stop-agent")).pipe(Effect.ignore);
549 }),
550 ),
551 );
552 }),
553 );
554 yield* Effect.addFinalizer(() => stop);
555 return {
556 id: config.id,
557 label: config.label,
558 summary: Effect.sync(() => summary),
559 inspect: provide(probe(config, owner)).pipe(
560 Effect.tap((value) =>
561 Effect.sync(() => {
562 summary = value;
563 }),
564 ),
565 ),
566 current: Effect.sync(() => ready),
567 ensureReady,
568 ensureAgentReady: (onPhase) =>
569 ensureReady(onPhase).pipe(
570 Effect.flatMap(() =>
571 onPhase(
572 summary.agentDeviceInstalled ? "starting" : "installing",
573 summary.agentDeviceInstalled
574 ? undefined
575 : deviceToolInstallMessage("agent tools", summary.tools?.agent),
576 ),
577 ),
578 Effect.flatMap(() => changeAgent(true)),
579 Effect.flatMap((value) =>
580 value?.agentDevice
581 ? Effect.succeed({ ...value, agentDevice: value.agentDevice })
582 : Effect.fail(
583 new DeviceHost.DeviceHostError({
584 hostId: config.id,
585 step: "starting agent tools",
586 cause: new Error("Daemon endpoint missing"),
587 }),
588 ),
589 ),
590 ),
591 stopAgent: changeAgent(false).pipe(Effect.asVoid, Effect.ignore),
592 stop,
593 platformAvailability,
594 } satisfies DeviceHost.DeviceHost["Service"];
595});