// The link: this machine dials out to the LLeMbas instance it is logged in to, and the web // UI drives sessions here over ACP. Nothing listens on this machine — the connection is outbound, // one per device, re-dialled with backoff when it drops — and what the instance may ask is bounded // by this machine's own global `remote:` block (agent.ts: Limits), never by the server. // // One WebSocket message is one JSON-RPC message. The token goes in the Authorization header and // never in the address. A token the instance refuses (revoked from its Settings → Security, or the // account lost the permission) ends the link for good: retrying it would only knock on a closed door. import { existsSync, readFileSync } from "node:fs" import { hostname } from "node:os" import { resolve as resolvePath } from "node:path" import { expandHome } from "../config/paths.ts" import { remoteBlock } from "../config/remote.ts" import { refusedLocally } from "../project/local.ts" import type { Mode } from "../config/schema.ts" import { HARNESS_VERSION } from "../harness.ts" import { instances, keyFile, rememberProtocols, type Instance } from "../lembas/login.ts" import { discover, linkProtocol, ProtocolError, spoken } from "../lembas/client.ts" import { VERSION } from "../version.ts" import { AcpAgent, type Limits } from "./agent.ts" import type { Hub } from "./hub.ts" import { Peer, type Transport } from "./rpc.ts" export class LinkError extends Error {} /** The device's limits, read from the global `remote:` block each time one is looked at — so a * running service follows `lembas remote …` (or an edit of config.yaml) without a restart. Every * property is a getter: never spread this object, or it stops following. * * With remote work off the roots are none, whatever `remote.roots` lists: the link is up, the * instance is told `roots: []`, and every session, directory and terminal is refused. * * `sharedOnly`: the link a terminal holds for /remote. While remote work is off it reaches only * what that terminal shares; once it is on, the roots count like the service's. */ export function limitsFromConfig(o: { sharedOnly?: boolean } = {}): Limits & { enabled: boolean } { const r = () => remoteBlock() return { get enabled() { return r().enabled === true }, get sharedOnly() { return o.sharedOnly === true && r().enabled !== true }, get roots() { return r().enabled === true ? (r().roots ?? []) : [] }, get maxMode() { return (r().max_mode ?? "edit") as Mode }, get approvalTimeoutMs() { return (r().approval_timeout ?? 600) * 1000 }, get requireTrust() { return r().require_trust !== false }, get terminal() { return r().terminal === true }, get terminalIntegration() { return r().terminal_integration !== false }, } } /** What the instance is told when the link opens (the hello, initialize's limits and * capabilities): when it changes, the link is dialled again so the instance hears the new one. */ function announced(L: Limits & { enabled: boolean }): string { return JSON.stringify([L.enabled, L.sharedOnly, L.roots, L.maxMode, L.requireTrust, L.terminal]) } export function pickInstance(name?: string): Instance { const all = instances() if (name) { const one = all[name] if (!one) throw new LinkError(`not logged in to ${name}${Object.keys(all).length ? ` — logged in: ${Object.keys(all).join(", ")}` : ""}`) return one } const list = Object.values(all) if (!list.length) throw new LinkError("not logged in to any LLeMbas instance — lembas login
") if (list.length > 1) throw new LinkError(`logged in to several instances; name one: ${Object.keys(all).join(", ")}`) return list[0]! } export function linkUrl(instance: Instance): string { const u = new URL(instance.base_url) u.protocol = u.protocol === "https:" ? "wss:" : "ws:" u.pathname = "/api/devices/link" return u.toString() } function wsTransport(ws: WebSocket): Transport { let onMsg: (t: string) => void = () => {} let onEnd: () => void = () => {} ws.addEventListener("message", (e) => onMsg(typeof e.data === "string" ? e.data : new TextDecoder().decode(e.data as ArrayBuffer))) ws.addEventListener("close", () => onEnd()) return { send: (t) => ws.send(t), onMessage: (fn) => (onMsg = fn), onClose: (fn) => (onEnd = fn), close: () => ws.close() } } export interface LinkStatus { state: "connecting" | "linked" | "waiting" | "stopped" instance: string since: string detail?: string attempts: number } export interface LinkOptions { instance?: string /** Each change of state, for the service's status file and log. */ onStatus?: (s: LinkStatus) => void /** Stop for good when this aborts. */ signal?: AbortSignal /** Only for tests: the backoff's first wait and its ceiling. */ backoffMs?: [number, number] /** The hub this link serves: the terminals' sessions reach the instance through it. */ hub?: Hub /** A terminal's link for /remote: it runs whatever `remote.enabled` says; with remote * work off, only the sessions the terminal shares can be reached. */ sharedOnly?: boolean /** Only for tests: how often the config is looked at while linked (default 2 s). */ pollMs?: number } /** Dial, serve, and dial again, until stopped or refused. Resolves when it stops for good. * * It links whether remote work is on or not, and with no roots at all: the instance then hears * `roots: []` and can tell a device that allows no folder yet from one that is not running, and a * `lembas remote add` or `enable` makes it serve, without a restart. */ export async function runLink(o: LinkOptions = {}): Promise { const local = refusedLocally("the link to a LLeMbas instance (lembas service, /remote)") if (local) throw new LinkError(local) const limits = limitsFromConfig({ sharedOnly: o.sharedOnly }) const poll = o.pollMs ?? 2000 const instance = pickInstance(o.instance) const token = readFileSync(keyFile(instance.connection), "utf8").trim() // A service runs for weeks: the webui connections' models are read again every ten minutes, so a // session the web UI opens here starts with the instance's models as they are then. const freshen = setInterval(() => void import("../lembas/webui.ts").then((w) => w.refreshWebui()), 10 * 60_000) freshen.unref?.() const [first, ceiling] = o.backoffMs ?? [1000, 60_000] let wait = first let attempts = 0 const status = (state: LinkStatus["state"], detail?: string): LinkStatus => { const s = { state, instance: instance.base_url, since: new Date().toISOString(), detail, attempts } o.onStatus?.(s) return s } // The CA the login was given, as a full path (an older login kept what was typed: relative, or // with ~, which a service started elsewhere does not find). One that is gone is said, not skipped. const caPath = instance.ca ? resolvePath(expandHome(instance.ca)) : undefined const caGone = caPath && !existsSync(caPath) ? caPath : undefined const loginCa = caPath && !caGone ? readFileSync(caPath, "utf8") : undefined while (!o.signal?.aborted) { attempts++ status("connecting") // What the instance speaks now: asked at every dial, since it may have been updated // since the login; what was said last when it cannot be asked. let protocols = instances()[instance.connection]?.protocols ?? instance.protocols // What the socket trusts is what discovery just got through with: the login's CA, or the // system's bundle when the instance's certificate needs it (Bun's own roots do not hold a // private CA). Why discovery failed is why the dial will, said in the status. let ca = loginCa let why: string | undefined let older: string | undefined try { const found = await discover(instance.base_url, loginCa ? { ca: caPath! } : undefined) protocols = spoken(found.discovery) rememberProtocols(instance.connection, protocols) if (found.tls?.ca && found.tls.ca !== caPath) ca = readFileSync(found.tls.ca, "utf8") } catch (e) { why = (e as Error).message // A server that answers and speaks another protocol: no dial would get further. if (e instanceof ProtocolError) older = why } if (older) { clearInterval(freshen) return status("stopped", older) } const unreachable = [caGone && `the CA file given at login, ${caGone}, is not there`, why ?? "could not reach the instance"].filter(Boolean).join("; ") const protocol = linkProtocol(protocols) const outcome = await new Promise<{ refused?: string; dropped?: string; final?: boolean; changed?: boolean }>((resolve) => { let opened = false let changed = false let watch: ReturnType | undefined const ws = new WebSocket(linkUrl(instance), { headers: { authorization: `Bearer ${token}`, "user-agent": `lembas-cli/${VERSION}` }, ...(ca ? { tls: { ca } } : {}), } as unknown as string[]) const stop = () => ws.close() o.signal?.addEventListener("abort", stop, { once: true }) ws.addEventListener("open", () => { opened = true wait = first const peer = new Peer(wsTransport(ws)) new AcpAgent(peer, limits, instance.connection, o.hub, protocol) peer.notify("_lembas/hello", { protocol, version: VERSION, harness_spec: HARNESS_VERSION, device: { host: hostname(), platform: `${process.platform}-${process.arch}` }, limits: { roots: limits.roots, max_mode: limits.maxMode, require_trust: limits.requireTrust }, }) status("linked") // The limits apply at once, each time they are looked at. What the instance was told at // the start is told again by dialling again, once nothing is working here: the sessions // are the hub's, so they outlive the link either way. const said = announced(limits) watch = setInterval(() => { if (changed || announced(limits) === said || o.hub?.busy()) return changed = true ws.close() }, poll) watch.unref?.() }) ws.addEventListener("close", (e) => { o.signal?.removeEventListener("abort", stop) clearInterval(watch) if (changed) return resolve({ changed: true }) // 4401/4403: the instance's word that this token is done; 1008 policy, likewise. if ([4401, 4403, 1008].includes(e.code)) return resolve({ refused: e.reason || "the instance refused this device's token" }) // 4400: the hello's protocol is not one it speaks. Dialling again would say the same. if (e.code === 4400) return resolve({ refused: `the instance does not speak link protocol ${protocol}, which this LLeMbas CLI ${VERSION} speaks: it needs a LLeMbas 2.0 server — update the server, or use LLeMbas CLI 1.x with it${e.reason ? ` (${e.reason})` : ""}`, final: true }) resolve({ dropped: opened ? `the link dropped (${e.code}${e.reason ? `: ${e.reason}` : ""})` : unreachable }) }) ws.addEventListener("error", () => {}) }) if (o.signal?.aborted) break if (outcome.changed) { status("waiting", "the remote settings changed; linking again") wait = first continue } if (outcome.refused) { clearInterval(freshen) return status("stopped", outcome.final ? outcome.refused : `${outcome.refused} — sign in again: lembas login`) } status("waiting", `${outcome.dropped}; trying again in ${Math.round(wait / 1000)}s`) await sleep(wait, o.signal) wait = Math.min(ceiling, wait * 2) } clearInterval(freshen) return status("stopped", "stopped") } /** Wait, or stop waiting when `signal` aborts. */ function sleep(ms: number, signal?: AbortSignal): Promise { return new Promise((r) => { const t = setTimeout(done, ms) function done() { clearTimeout(t) signal?.removeEventListener("abort", done) r() } signal?.addEventListener("abort", done, { once: true }) }) }