feat(desktop/fleet): run a command/script locally (a 'this machine' fleet target)

apps/desktop/electron/main/services/fleet.ts

@@ -9,11 +9,51 @@ import {
   type SudoMode,
 } from '@diskpush/fleet-core'
 import type { Connection, FleetCommand, FleetHostResult, FleetList, HostUpdateReport } from '@diskpush/schemas'
-import { sshConfigConnections } from '@diskpush/ssh-core'
+import { sshConfigConnections, type SshSession } from '@diskpush/ssh-core'
+import { userInfo } from 'node:os'
 import { IPC, type FleetRequest } from '../../shared/contract.js'
 import { dropSession, sessionFor } from './sessions.js'
+import { LOCAL_CONNECTION_ID, isLocalConnectionId, localSession } from './local-session.js'
 import { store } from './store.js'
 
+/**
+ * The synthetic "this machine" target.
+ *
+ * It appears in the fleet server list like any other server, so the user can
+ * tick it and run a command/script LOCALLY — including one that fans out to the
+ * fleet itself (an ssh loop, an rsync push). Everything else about a fleet run
+ * (interpreter, hazard check, streaming, history) is identical; only the
+ * transport differs, which is decided in connectFleet().
+ */
+function localhostConnection(): Connection {
+  const now = new Date().toISOString()
+  return {
+    id: LOCAL_CONNECTION_ID,
+    name: 'localhost (this machine)',
+    host: 'localhost',
+    port: 22,
+    username: userInfo().username,
+    authType: 'agent',
+    keyPath: null,
+    defaultLocalPath: null,
+    defaultRemotePath: null,
+    jumpHost: null,
+    rsyncPath: null,
+    connectTimeoutSeconds: 15,
+    keepaliveSeconds: 30,
+    forwardAgent: false,
+    tags: ['local'],
+    notes: 'Runs on this machine, not over SSH.',
+    createdAt: now,
+    updatedAt: now,
+  }
+}
+
+/** Local host runs in-process; every other host is a pooled SSH session. */
+function connectFleet(connection: Connection): Promise<SshSession> {
+  return isLocalConnectionId(connection.id) ? Promise.resolve(localSession()) : sessionFor(connection)
+}
+
 /**
  * Fleet operations for the desktop.
  *
@@ -30,7 +70,12 @@ const running = new Map<string, RunningFleet>()
 export async function fleetServers(): Promise<Connection[]> {
   const saved = await (await store()).listConnections()
   const savedNames = new Set(saved.map((connection) => connection.name))
-  return [...saved, ...sshConfigConnections().filter((host) => !savedNames.has(host.name))]
+  // localhost first, then saved connections, then ~/.ssh/config hosts.
+  return [
+    localhostConnection(),
+    ...saved,
+    ...sshConfigConnections().filter((host) => !savedNames.has(host.name)),
+  ]
 }
 
 export async function fleetCommands(): Promise<FleetCommand[]> {
@@ -133,7 +178,7 @@ export async function startFleet(request: FleetRequest, sender: WebContents): Pr
         onFailure: request.onFailure,
         runId,
         signal: controller.signal,
-        connect: (connection) => sessionFor(connection),
+        connect: (connection) => connectFleet(connection),
         // No release: the desktop pools sessions across browsing and
         // transfers, so closing one here would shut a file pane's connection
         // out from under it.
@@ -163,7 +208,11 @@ export async function startFleet(request: FleetRequest, sender: WebContents): Pr
       // A command that changed sshd, the login shell or a key would leave a
       // pooled session pointing at a server that no longer works the way the
       // session assumes. Cheaper to reconnect than to debug that later.
-      if (request.sudo) for (const connection of connections) dropSession(connection.id)
+      if (request.sudo) {
+        for (const connection of connections) {
+          if (!isLocalConnectionId(connection.id)) dropSession(connection.id)
+        }
+      }
     }
   })()
 
@@ -258,7 +307,7 @@ export async function checkFleetServers(input: {
     connections,
     concurrency: input.concurrency,
     timeoutSeconds: input.timeoutSeconds,
-    connect: (connection) => sessionFor(connection),
+    connect: (connection) => connectFleet(connection),
   })
   return reports
 }

apps/desktop/electron/main/services/local-session.test.ts

@@ -0,0 +1,55 @@
+import { describe, it, expect } from 'vitest'
+import { localExecStream, isLocalConnectionId, LOCAL_CONNECTION_ID } from './local-session.js'
+
+describe('isLocalConnectionId', () => {
+  it('recognizes the reserved local id and nothing else', () => {
+    expect(isLocalConnectionId(LOCAL_CONNECTION_ID)).toBe(true)
+    expect(isLocalConnectionId('local')).toBe(true)
+    expect(isLocalConnectionId('ssh-config:dev1')).toBe(false)
+    expect(isLocalConnectionId('some-uuid')).toBe(false)
+  })
+})
+
+describe('localExecStream', () => {
+  it('captures stdout and a zero exit, and streams whole lines', async () => {
+    const lines: string[] = []
+    const handle = localExecStream("printf 'one\\ntwo\\n'", { onStdout: (l) => lines.push(l) })
+    const res = await handle.finished
+    expect(res.code).toBe(0)
+    expect(res.timedOut).toBe(false)
+    expect(res.stdout).toBe('one\ntwo\n')
+    expect(lines).toEqual(['one', 'two'])
+  })
+
+  it('captures a non-zero exit code', async () => {
+    const res = await localExecStream('exit 3').finished
+    expect(res.code).toBe(3)
+  })
+
+  it('separates stderr and streams it', async () => {
+    const errs: string[] = []
+    const res = await localExecStream('echo oops 1>&2', { onStderr: (l) => errs.push(l) }).finished
+    expect(res.stderr).toBe('oops\n')
+    expect(errs).toEqual(['oops'])
+    expect(res.code).toBe(0)
+  })
+
+  it('feeds stdin to the command (as sudo -S would use)', async () => {
+    const res = await localExecStream('cat', { stdin: 'secret-line\n' }).finished
+    expect(res.stdout).toBe('secret-line\n')
+  })
+
+  it('enforces a timeout and marks the result timedOut', async () => {
+    const res = await localExecStream('sleep 30', { timeoutSeconds: 1 }).finished
+    expect(res.timedOut).toBe(true)
+    expect(res.code).not.toBe(0)
+  }, 15000)
+
+  it('can be cancelled', async () => {
+    const handle = localExecStream('sleep 30')
+    setTimeout(() => handle.cancel(), 50)
+    const res = await handle.finished
+    expect(res.code).not.toBe(0)
+    expect(res.timedOut).toBe(false)
+  }, 15000)
+})

apps/desktop/electron/main/services/local-session.ts

@@ -0,0 +1,118 @@
+import { spawn } from 'node:child_process'
+import type { ExecHandle, ExecResult, ExecStreamOptions, SshSession } from '@diskpush/ssh-core'
+
+/**
+ * A "session" that runs a fleet command on THIS machine instead of over SSH.
+ *
+ * It implements the exact slice of SshSession that the fleet runner uses —
+ * `execStream(command, options): ExecHandle` — so `runFleet` can target the
+ * local host with no changes: the desktop's connect() hands back a LocalSession
+ * for the synthetic `localhost` connection and a pooled SshSession for the rest.
+ *
+ * The command is the same interpreter invocation the runner builds for a remote
+ * host (e.g. `sh -c '<script>'`, optionally wrapped in sudo), run through the
+ * local shell — so a script that fans out to the fleet itself (an ssh loop, an
+ * rsync push) runs exactly as it would if you pasted it into a terminal here.
+ */
+export function localExecStream(command: string, options: ExecStreamOptions = {}): ExecHandle {
+  // detached so the child is its own process group leader: a script that spawns
+  // its own children (an ssh loop, an rsync) can then be stopped as a group,
+  // rather than leaving the real work running after the shell is signalled.
+  const child = spawn('/bin/sh', ['-c', command], { stdio: ['pipe', 'pipe', 'pipe'], detached: true })
+
+  // Signal the whole group (negative pid); fall back to the bare process.
+  const killTree = (signal: NodeJS.Signals) => {
+    try {
+      if (child.pid) process.kill(-child.pid, signal)
+      else child.kill(signal)
+    } catch {
+      try { child.kill(signal) } catch { /* already gone */ }
+    }
+  }
+
+  let stdout = ''
+  let stderr = ''
+  let timedOut = false
+  let settled = false
+
+  // Emit whole lines as they arrive; keep the trailing partial line buffered.
+  const linePump = (emit?: (line: string) => void) => {
+    let buf = ''
+    return {
+      push(chunk: string) {
+        buf += chunk
+        let nl = buf.indexOf('\n')
+        while (nl !== -1) {
+          emit?.(buf.slice(0, nl))
+          buf = buf.slice(nl + 1)
+          nl = buf.indexOf('\n')
+        }
+      },
+      flush() {
+        if (buf.length > 0) emit?.(buf)
+        buf = ''
+      },
+    }
+  }
+  const outPump = linePump(options.onStdout)
+  const errPump = linePump(options.onStderr)
+
+  child.stdout.setEncoding('utf8')
+  child.stderr.setEncoding('utf8')
+  child.stdout.on('data', (chunk: string) => { stdout += chunk; outPump.push(chunk) })
+  child.stderr.on('data', (chunk: string) => { stderr += chunk; errPump.push(chunk) })
+
+  // Feed sudo -S its password (and anything else the built command reads), then
+  // close stdin so a command waiting on input does not hang the whole run.
+  if (options.stdin !== undefined) child.stdin.write(options.stdin)
+  child.stdin.end()
+
+  let timer: NodeJS.Timeout | undefined
+  if (options.timeoutSeconds && options.timeoutSeconds > 0) {
+    timer = setTimeout(() => {
+      timedOut = true
+      killTree('SIGKILL')
+    }, options.timeoutSeconds * 1000)
+  }
+
+  const finished = new Promise<ExecResult>((resolve) => {
+    const done = (code: number) => {
+      if (settled) return
+      settled = true
+      if (timer) clearTimeout(timer)
+      outPump.flush()
+      errPump.flush()
+      resolve({ stdout, stderr, code, timedOut })
+    }
+    child.on('error', (err) => {
+      // Could not even start the shell: report it as a non-zero exit with the
+      // reason on stderr, the same shape a failed remote exec produces.
+      stderr += (stderr ? '\n' : '') + (err instanceof Error ? err.message : String(err))
+      done(127)
+    })
+    child.on('close', (code, signal) => done(code ?? (signal ? 1 : 0)))
+  })
+
+  return {
+    finished,
+    cancel: () => {
+      if (settled) return
+      killTree('SIGINT')
+      // Give it a moment to leave cleanly, then insist.
+      setTimeout(() => { if (!settled) killTree('SIGKILL') }, 2000)
+    },
+  }
+}
+
+/** A LocalSession shaped as the fleet runner's SshSession dependency. */
+export function localSession(): SshSession {
+  return { execStream: localExecStream } as unknown as SshSession
+}
+
+/** The reserved id of the synthetic "this machine" fleet target. */
+export const LOCAL_CONNECTION_ID = 'local'
+
+/** True when a connection id refers to the local host rather than an SSH server. */
+export function isLocalConnectionId(id: string): boolean {
+  return id === LOCAL_CONNECTION_ID
+}