The first public release of LLeMbas CLI: a terminal coding agent and project manager for any LLM API, with permission modes, git snapshots, memory and skills, knowledge bases, MCP, voice, and a link to a LLeMbas instance whose web UI can work its sessions too. Signed Linux binaries for x64 and arm64. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
355 files changed
+47028
No files matched your search
+1167
File diff suppressed because it is too large.
Load diff
+340
@@ -0,0 +1,340 @@
|
||||
// The hub: one session, seen and used from the terminal and the web UI alike.
|
||||
//
|
||||
// A session runs in exactly one process — whichever has it open — and every other side reaches it
|
||||
// through that one. The process holding the LLeMbas link (the service, or a terminal that shared a
|
||||
// session with /remote while no service ran) is the hub: it listens on a Unix socket in the state
|
||||
// directory, and every terminal on this machine connects to it.
|
||||
//
|
||||
// - A session the web UI started runs in the hub. Opening it in a terminal asks the hub to let go
|
||||
// of it first (refused while it is working), and the terminal carries it on from the store —
|
||||
// the same sessions.db, so nothing is lost and nothing is forked.
|
||||
// - A session a terminal runs is *shared*: the hub tells the instance about it, passes on every
|
||||
// event it makes, and hands it what the web UI sends (a prompt, a message mid-task, stop, mode,
|
||||
// model). An approval goes to both sides and the first answer counts.
|
||||
// - With the service running, every terminal session is shared by itself, within the device's
|
||||
// own `remote:` limits. Without it, /remote shares the one session, and the terminal holds the
|
||||
// link itself — so that session lasts on the web exactly as long as the terminal does.
|
||||
//
|
||||
// The socket is the user's own (0600, in ~/.local/state/lembas). It answers JSON-RPC, one
|
||||
// message a line, the same framing as `serve --stdio`.
|
||||
import { chmodSync, existsSync, mkdirSync, rmSync } from "node:fs"
|
||||
import { connect, createServer, type Server, type Socket } from "node:net"
|
||||
import { join } from "node:path"
|
||||
import { paths } from "../config/paths.ts"
|
||||
import type { Mode } from "../config/schema.ts"
|
||||
import type { CommandInfo } from "../session/commands.ts"
|
||||
import { clampMode, refusalFor, type AcpAgent, type Limits } from "./agent.ts"
|
||||
import { addPendingDeleted } from "./pending.ts"
|
||||
import { INVALID_PARAMS, Peer, RpcError, type Transport } from "./rpc.ts"
|
||||
|
||||
export const hubSocket = () => join(paths.state, "hub.sock")
|
||||
|
||||
/** One message a line, over a Unix socket. */
|
||||
export function socketTransport(sock: Socket): Transport {
|
||||
let onMsg: (t: string) => void = () => {}
|
||||
let onEnd: () => void = () => {}
|
||||
let buf = ""
|
||||
let ended = false
|
||||
sock.setEncoding("utf8")
|
||||
sock.on("data", (chunk: string) => {
|
||||
buf += chunk
|
||||
let nl: number
|
||||
while ((nl = buf.indexOf("\n")) >= 0) {
|
||||
const line = buf.slice(0, nl)
|
||||
buf = buf.slice(nl + 1)
|
||||
if (line.trim()) onMsg(line)
|
||||
}
|
||||
})
|
||||
const end = () => {
|
||||
if (ended) return
|
||||
ended = true
|
||||
onEnd()
|
||||
}
|
||||
sock.on("close", end)
|
||||
sock.on("error", () => {})
|
||||
return {
|
||||
send: (t) => void (sock.writable && sock.write(t + "\n")),
|
||||
onMessage: (fn) => (onMsg = fn),
|
||||
onClose: (fn) => (onEnd = fn),
|
||||
close: () => sock.destroy(),
|
||||
}
|
||||
}
|
||||
|
||||
/** Whether something answers on the socket (a hub that is alive, not a file left behind). */
|
||||
export function answers(path = hubSocket(), timeoutMs = 1000): Promise<boolean> {
|
||||
return new Promise((resolve) => {
|
||||
const sock = connect(path)
|
||||
const done = (v: boolean) => {
|
||||
clearTimeout(timer)
|
||||
sock.destroy()
|
||||
resolve(v)
|
||||
}
|
||||
const timer = setTimeout(() => done(false), timeoutMs)
|
||||
sock.once("connect", () => done(true))
|
||||
sock.once("error", () => done(false))
|
||||
})
|
||||
}
|
||||
|
||||
/** Connect to the hub, if one runs. */
|
||||
export function connectHub(path = hubSocket()): Promise<Peer | undefined> {
|
||||
if (!existsSync(path)) return Promise.resolve(undefined)
|
||||
return new Promise((resolve) => {
|
||||
const sock = connect(path)
|
||||
const timer = setTimeout(() => (sock.destroy(), resolve(undefined)), 2000)
|
||||
sock.once("connect", () => {
|
||||
clearTimeout(timer)
|
||||
resolve(new Peer(socketTransport(sock)))
|
||||
})
|
||||
sock.once("error", () => {
|
||||
clearTimeout(timer)
|
||||
resolve(undefined)
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
/** A session a terminal runs and the hub passes on. */
|
||||
export interface Shared {
|
||||
sessionId: string
|
||||
/** The project's root: what the limits judge, and what the instance is told. */
|
||||
cwd: string
|
||||
/** The directory it works in, which @path mentions resolve against. */
|
||||
workdir: string
|
||||
title: string
|
||||
model: string
|
||||
/** The connection its model is spoken to through. */
|
||||
modelConnection?: string
|
||||
mode: Mode
|
||||
effort: string
|
||||
/** /remote: the person shared it by name, so the device's remote limits do not decide. */
|
||||
explicit: boolean
|
||||
busy: boolean
|
||||
client: Peer
|
||||
pid: number
|
||||
/** Its turns as its events said them: the one running, the last one done. */
|
||||
turnId?: string
|
||||
lastTurnId?: string
|
||||
/** Its command set, as the terminal sent it. */
|
||||
commands?: CommandInfo[]
|
||||
}
|
||||
|
||||
export interface HubOptions {
|
||||
/** The service: every terminal session is shared by itself (within the limits). A terminal that
|
||||
* became the hub for /remote shares only what is shared by name. */
|
||||
service: boolean
|
||||
limits: Limits & { enabled: boolean }
|
||||
}
|
||||
|
||||
export class Hub {
|
||||
readonly shared = new Map<string, Shared>()
|
||||
/** The sessions the link runs here: kept across links, so one that drops and comes
|
||||
* back finds them still working. */
|
||||
readonly held = new Map<string, import("./agent.ts").Session>()
|
||||
/** The ACP agent on the link while it is up. */
|
||||
agent?: AcpAgent
|
||||
private server?: Server
|
||||
private clients = new Set<Peer>()
|
||||
|
||||
constructor(readonly o: HubOptions) {}
|
||||
|
||||
/** Take the socket. False: another hub is alive on it. */
|
||||
async listen(path = hubSocket()): Promise<boolean> {
|
||||
mkdirSync(paths.state, { recursive: true })
|
||||
if (existsSync(path)) {
|
||||
if (await answers(path)) return false
|
||||
rmSync(path, { force: true })
|
||||
}
|
||||
return new Promise((resolve) => {
|
||||
const server = createServer((sock) => this.accept(new Peer(socketTransport(sock))))
|
||||
server.once("error", () => resolve(false))
|
||||
server.listen(path, () => {
|
||||
try {
|
||||
chmodSync(path, 0o600)
|
||||
} catch {}
|
||||
this.server = server
|
||||
this.path = path
|
||||
resolve(true)
|
||||
})
|
||||
})
|
||||
}
|
||||
private path?: string
|
||||
|
||||
close() {
|
||||
for (const c of this.clients) c.close()
|
||||
this.server?.close()
|
||||
if (this.path) rmSync(this.path, { force: true })
|
||||
this.server = undefined
|
||||
}
|
||||
|
||||
/** The link came up: the instance hears about every shared session. */
|
||||
attach(agent: AcpAgent) {
|
||||
this.agent = agent
|
||||
for (const s of this.shared.values()) this.announce(s)
|
||||
// Their command sets go once the client has initialized the link (agent.ts, initialize).
|
||||
}
|
||||
|
||||
detach(agent: AcpAgent) {
|
||||
if (this.agent === agent) this.agent = undefined
|
||||
}
|
||||
|
||||
/** Whether anything works here now: a session the link runs, or one a terminal shares mid-turn. */
|
||||
busy(): boolean {
|
||||
return [...this.held.values()].some((s) => s.prompting || s.queue) || [...this.shared.values()].some((s) => s.busy)
|
||||
}
|
||||
|
||||
private announce(s: Shared) {
|
||||
this.agent?.notifyInstance("_lembas/session/announce", {
|
||||
sessionId: s.sessionId,
|
||||
cwd: s.cwd,
|
||||
title: s.title,
|
||||
model: s.model,
|
||||
// The session's model is this link's instance's.
|
||||
instance: this.agent?.isInstance(s.modelConnection) ?? false,
|
||||
mode: s.mode,
|
||||
effort: s.effort,
|
||||
busy: s.busy,
|
||||
origin: "terminal",
|
||||
})
|
||||
}
|
||||
|
||||
/** What the hub would say to a terminal asking to open this session: undefined when it may. */
|
||||
private releaseFor(peer: Peer, id: string): void {
|
||||
const other = this.shared.get(id)
|
||||
if (other && other.client !== peer && !other.client.isClosed) throw new RpcError(INVALID_PARAMS, `this session is open in another terminal (pid ${other.pid})`)
|
||||
const held = this.agent?.holding(id)
|
||||
if (held === "busy") throw new RpcError(INVALID_PARAMS, "this session is working on a prompt from the web UI — wait for it to finish, then open it")
|
||||
if (held) this.agent!.drop(id)
|
||||
}
|
||||
|
||||
private accept(peer: Peer) {
|
||||
this.clients.add(peer)
|
||||
let pid = 0
|
||||
peer.handle("hub/hello", (p) => {
|
||||
pid = Number(p?.pid) || 0
|
||||
return { service: this.o.service, linked: Boolean(this.agent), pid: process.pid }
|
||||
})
|
||||
// Before a terminal resumes a session: the hub lets go of its own copy, or says why not.
|
||||
peer.handle("hub/release", (p) => {
|
||||
this.releaseFor(peer, String(p?.sessionId ?? ""))
|
||||
return {}
|
||||
})
|
||||
peer.handle("hub/share", (p) => {
|
||||
const id = String(p?.sessionId ?? "")
|
||||
if (!id) throw new RpcError(INVALID_PARAMS, "no session")
|
||||
this.releaseFor(peer, id)
|
||||
const explicit = p?.explicit === true
|
||||
const cwd = String(p?.cwd ?? "")
|
||||
if (!explicit) {
|
||||
if (!this.o.service) return { shared: false, reason: "no service runs here; /remote shares this session" }
|
||||
const why = refusalFor(this.o.limits, cwd)
|
||||
if (why) return { shared: false, reason: why }
|
||||
}
|
||||
const s: Shared = {
|
||||
sessionId: id,
|
||||
cwd,
|
||||
workdir: typeof p?.workdir === "string" && p.workdir ? p.workdir : cwd,
|
||||
title: String(p?.title ?? ""),
|
||||
model: String(p?.model ?? ""),
|
||||
...(typeof p?.modelConnection === "string" ? { modelConnection: p.modelConnection } : {}),
|
||||
mode: (p?.mode ?? "manual") as Mode,
|
||||
effort: String(p?.effort ?? ""),
|
||||
explicit,
|
||||
busy: p?.busy === true,
|
||||
client: peer,
|
||||
pid,
|
||||
}
|
||||
this.shared.set(id, s)
|
||||
this.announce(s)
|
||||
return { shared: true, linked: Boolean(this.agent) }
|
||||
})
|
||||
peer.handle("hub/unshare", (p) => {
|
||||
this.unshare(peer, String(p?.sessionId ?? ""))
|
||||
return {}
|
||||
})
|
||||
peer.on("hub/event", (p) => {
|
||||
const s = this.shared.get(String(p?.sessionId ?? ""))
|
||||
const e = p?.event
|
||||
if (!s || s.client !== peer || !e || typeof e !== "object") return
|
||||
if (e.type === "task") {
|
||||
s.busy = e.state === "start"
|
||||
if (typeof e.turnId === "string") {
|
||||
if (e.state === "start") s.turnId = e.turnId
|
||||
else {
|
||||
s.lastTurnId = e.turnId
|
||||
if (s.turnId === e.turnId) s.turnId = undefined
|
||||
}
|
||||
}
|
||||
}
|
||||
if (e.type === "mode") s.mode = e.mode
|
||||
if (e.type === "model") {
|
||||
s.model = String(e.ref ?? s.model)
|
||||
if (typeof e.connection === "string") s.modelConnection = e.connection
|
||||
s.effort = String(e.effort ?? s.effort)
|
||||
}
|
||||
if (e.type === "title") s.title = String(e.title ?? "")
|
||||
this.agent?.emitShared(s.sessionId, e)
|
||||
})
|
||||
peer.handle("hub/ask", async (p) => {
|
||||
const s = this.shared.get(String(p?.sessionId ?? ""))
|
||||
if (!s || s.client !== peer) throw new RpcError(INVALID_PARAMS, "this session is not shared")
|
||||
if (!this.agent) throw new RpcError(INVALID_PARAMS, "not linked")
|
||||
return this.agent.askInstance(p.params)
|
||||
})
|
||||
// A terminal deleted a session: already let go of here (hub/release first), so the
|
||||
// instance is told, and deletes the chat that is that session.
|
||||
peer.on("hub/deleted", (p) => {
|
||||
const id = String(p?.sessionId ?? "")
|
||||
if (!id) return
|
||||
const s = this.shared.get(id)
|
||||
if (s && s.client === peer) this.shared.delete(id)
|
||||
this.deleted(id)
|
||||
})
|
||||
peer.on("hub/settled", (p) => this.agent?.notifyInstance("_lembas/permission/settled", { sessionId: p?.sessionId, toolCallId: p?.toolCallId }))
|
||||
// ask_user and plan_submit in a terminal's session: the same card in the web UI, the
|
||||
// first answer wins — exactly as an approval.
|
||||
for (const [method, card] of [
|
||||
["hub/question", "_lembas/question"],
|
||||
["hub/plan", "_lembas/plan"],
|
||||
] as const)
|
||||
peer.handle(method, async (p) => {
|
||||
const s = this.shared.get(String(p?.sessionId ?? ""))
|
||||
if (!s || s.client !== peer) throw new RpcError(INVALID_PARAMS, "this session is not shared")
|
||||
if (!this.agent) throw new RpcError(INVALID_PARAMS, "not linked")
|
||||
const answer: any = await this.agent.askInstanceCard(card, p.params)
|
||||
// A plan approved from the web never puts a session shared by the device's limits (not by
|
||||
// name) in a mode beyond remote.max_mode; where even manual is beyond it, nothing is decided.
|
||||
if (card === "_lembas/plan" && answer?.outcome === "approve" && !s.explicit) {
|
||||
const mode = clampMode(answer.mode === "manual" ? "manual" : "edit", this.o.limits.maxMode)
|
||||
return mode === "edit" || mode === "manual" ? { ...answer, mode } : { outcome: "dismissed" }
|
||||
}
|
||||
return answer
|
||||
})
|
||||
// Its command set, at sharing and whenever it changes.
|
||||
peer.on("hub/commands", (p) => {
|
||||
const s = this.shared.get(String(p?.sessionId ?? ""))
|
||||
if (!s || s.client !== peer || !Array.isArray(p?.commands)) return
|
||||
s.commands = p.commands
|
||||
this.agent?.commandsShared(s.sessionId, s.commands!)
|
||||
})
|
||||
peer.onClose(() => {
|
||||
this.clients.delete(peer)
|
||||
for (const s of [...this.shared.values()]) if (s.client === peer) this.unshare(peer, s.sessionId)
|
||||
})
|
||||
}
|
||||
|
||||
/** A session deleted on this machine, to the instance — or, with no link up now, kept for the
|
||||
* next one (pending.ts): a delete made while the link is down is not lost, and its chat does
|
||||
* not stay in the web UI for good. */
|
||||
deleted(id: string) {
|
||||
if (this.agent) this.agent.notifyDeleted(id)
|
||||
else addPendingDeleted([id])
|
||||
}
|
||||
|
||||
private unshare(peer: Peer, id: string) {
|
||||
const s = this.shared.get(id)
|
||||
if (!s || s.client !== peer) return
|
||||
this.shared.delete(id)
|
||||
// With the service, the session is still reachable: the next prompt from the web UI opens it
|
||||
// here from the store. Without it, it is gone from the web UI's reach with the terminal.
|
||||
this.agent?.notifyInstance("_lembas/session/left", { sessionId: id, available: this.o.service })
|
||||
}
|
||||
}
|
||||
+172
@@ -0,0 +1,172 @@
|
||||
// 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 { loadConfig } from "../config/load.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, 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 {}
|
||||
|
||||
export function limitsFromConfig(): Limits & { enabled: boolean } {
|
||||
const r = loadConfig().config.remote ?? {}
|
||||
return {
|
||||
enabled: r.enabled === true,
|
||||
roots: r.roots ?? [],
|
||||
maxMode: (r.max_mode ?? "edit") as Mode,
|
||||
approvalTimeoutMs: (r.approval_timeout ?? 600) * 1000,
|
||||
requireTrust: r.require_trust !== false,
|
||||
terminal: r.terminal === true,
|
||||
terminalIntegration: r.terminal_integration !== false,
|
||||
}
|
||||
}
|
||||
|
||||
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 <address>")
|
||||
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
|
||||
}
|
||||
|
||||
/** Dial, serve, and dial again, until stopped or refused. Resolves when it stops for good. */
|
||||
export async function runLink(o: LinkOptions = {}): Promise<LinkStatus> {
|
||||
const configured = limitsFromConfig()
|
||||
const limits = o.sharedOnly && !configured.enabled ? { ...configured, roots: [], sharedOnly: true } : configured
|
||||
if (!o.sharedOnly && !limits.enabled) throw new LinkError("remote work is off on this machine — set remote.enabled: true (and remote.roots) in ~/.config/lembas/config.yaml")
|
||||
if (!o.sharedOnly && !limits.roots.length) throw new LinkError("remote.roots is empty, so no directory could be worked in — list the directories a remote session may use")
|
||||
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
|
||||
}
|
||||
const ca = instance.ca && existsSync(instance.ca) ? readFileSync(instance.ca, "utf8") : undefined
|
||||
/** A 4400 to a hello that said 2: the protocols known were stale (or discovery said more than the
|
||||
* link takes) — the next dial, at once, says 1; only a 4400 to that stops the link. */
|
||||
let downgraded = false
|
||||
|
||||
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
|
||||
try {
|
||||
const { discovery } = await discover(instance.base_url, ca ? { ca: instance.ca } : undefined)
|
||||
protocols = spoken(discovery)
|
||||
rememberProtocols(instance.connection, protocols)
|
||||
} catch {}
|
||||
const protocol = downgraded ? 1 : linkProtocol(protocols)
|
||||
const outcome = await new Promise<{ refused?: string; dropped?: string; final?: boolean }>((resolve) => {
|
||||
let opened = false
|
||||
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")
|
||||
})
|
||||
ws.addEventListener("close", (e) => {
|
||||
o.signal?.removeEventListener("abort", stop)
|
||||
// 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 and this CLI speak different link protocols — update one of them", final: true })
|
||||
resolve({ dropped: opened ? `the link dropped (${e.code}${e.reason ? `: ${e.reason}` : ""})` : "could not reach the instance" })
|
||||
})
|
||||
ws.addEventListener("error", () => {})
|
||||
})
|
||||
if (o.signal?.aborted) break
|
||||
if (outcome.final && protocol === 2 && !downgraded) {
|
||||
downgraded = true
|
||||
status("waiting", "the instance does not take link protocol 2; trying again with 1")
|
||||
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 new Promise((r) => setTimeout(r, wait))
|
||||
wait = Math.min(ceiling, wait * 2)
|
||||
}
|
||||
clearInterval(freshen)
|
||||
return status("stopped", "stopped")
|
||||
}
|
||||
@@ -0,0 +1,61 @@
|
||||
// Deletes the instance has not heard about yet.
|
||||
//
|
||||
// A session deleted on this machine is deleted in the web UI too: the terminal tells the hub
|
||||
// (`hub/deleted`), and the hub tells the instance (`_lembas/session/deleted`) over the link. Any of
|
||||
// those steps may find the next one missing — a terminal with no hub (no service, no /remote), a
|
||||
// hub with no link up (the instance down, the network out), `lembas sessions delete` with neither
|
||||
// — and the chat would stay in the web UI, pointing at a session that no longer exists. So whoever
|
||||
// cannot pass the word on writes it here, and the service sends everything here once the next
|
||||
// link has initialized, then clears it.
|
||||
//
|
||||
// A plain JSON list of session ids in the state directory, beside the hub's socket. Written whole
|
||||
// through a rename, so a reader never sees half of it; deduplicated; the newest DELETED_MAX kept —
|
||||
// a machine whose link never comes back must not grow it for ever, and a chat that old is long
|
||||
// out of anybody's mind.
|
||||
import { existsSync, mkdirSync, readFileSync, renameSync, rmSync, writeFileSync } from "node:fs"
|
||||
import { join } from "node:path"
|
||||
import { paths } from "../config/paths.ts"
|
||||
|
||||
export const DELETED_MAX = 500
|
||||
|
||||
export const deletedFile = () => join(paths.state, "deleted.json")
|
||||
|
||||
/** The session ids waiting to be told to the instance, oldest first. */
|
||||
export function pendingDeleted(): string[] {
|
||||
try {
|
||||
const v = JSON.parse(readFileSync(deletedFile(), "utf8"))
|
||||
return Array.isArray(v) ? v.filter((x): x is string => typeof x === "string" && x.length > 0) : []
|
||||
} catch {
|
||||
return []
|
||||
}
|
||||
}
|
||||
|
||||
function write(ids: string[]) {
|
||||
const file = deletedFile()
|
||||
if (!ids.length) return void rmSync(file, { force: true })
|
||||
mkdirSync(paths.state, { recursive: true })
|
||||
const tmp = `${file}.${process.pid}.tmp`
|
||||
writeFileSync(tmp, JSON.stringify(ids), { mode: 0o600 })
|
||||
renameSync(tmp, file)
|
||||
}
|
||||
|
||||
/** Remember deletes to tell the instance later. Never in the way of the delete itself: a state
|
||||
* directory that cannot be written to only loses the word. */
|
||||
export function addPendingDeleted(ids: string[]) {
|
||||
const add = ids.filter(Boolean)
|
||||
if (!add.length) return
|
||||
try {
|
||||
// An id already waiting moves to the end: it is the newest word about it.
|
||||
const list = [...pendingDeleted().filter((x) => !add.includes(x)), ...new Set(add)]
|
||||
write(list.slice(-DELETED_MAX))
|
||||
} catch {}
|
||||
}
|
||||
|
||||
/** Forget the ones that were told. Re-read first, so an id another process added meanwhile (a
|
||||
* terminal, `lembas sessions delete`) is kept for the next link. */
|
||||
export function clearPendingDeleted(sent: string[]) {
|
||||
if (!sent.length || !existsSync(deletedFile())) return
|
||||
try {
|
||||
write(pendingDeleted().filter((x) => !sent.includes(x)))
|
||||
} catch {}
|
||||
}
|
||||
@@ -0,0 +1,187 @@
|
||||
// A prompt that came over ACP, made into what the TUI would have sent for the same thing typed
|
||||
//: `/name` expanded by the shared command set (session/commands.ts), `@path` mentions
|
||||
// attached by the same expander as the TUI's (project/attach.ts, same limits), and the blocks a web
|
||||
// composer adds — images and files — stored the way the TUI stores a pasted image or a dropped
|
||||
// file: written under the session's attachment directory and attached by path, so a vision model
|
||||
// sees the image, any other model is told it is there, and the model can read a file again later.
|
||||
import { mkdirSync, rmSync, statSync, writeFileSync } from "node:fs"
|
||||
import { basename, extname, join, resolve } from "node:path"
|
||||
import type { App } from "../app.ts"
|
||||
import type { AttachmentInfo } from "../bus/index.ts"
|
||||
import { paths } from "../config/paths.ts"
|
||||
import type { Mode } from "../config/schema.ts"
|
||||
import { attachmentsFor } from "../project/attach.ts"
|
||||
import { realPath } from "../permission/evaluate.ts"
|
||||
import { protectedPath } from "../permission/hardline.ts"
|
||||
import { expandHome } from "../config/paths.ts"
|
||||
import { IMAGE_TYPES } from "../project/image.ts"
|
||||
import type { ImagePart } from "../provider/types.ts"
|
||||
import { expandSlash } from "../session/commands.ts"
|
||||
import { INVALID_PARAMS, RpcError } from "./rpc.ts"
|
||||
|
||||
/** Per block, and for the whole prompt, in bytes as decoded. */
|
||||
export const BLOCK_LIMIT = 10 * 1024 * 1024
|
||||
export const PROMPT_LIMIT = 25 * 1024 * 1024
|
||||
const IMAGE_MIMES = ["image/png", "image/jpeg", "image/webp", "image/gif"]
|
||||
|
||||
/** Where a session's attachments are kept. */
|
||||
export const attachmentDir = (sessionId: string) => join(paths.state, "attachments", sessionId.replace(/[^\w.-]/g, "_"))
|
||||
|
||||
export function dropAttachments(sessionId: string) {
|
||||
rmSync(attachmentDir(sessionId), { recursive: true, force: true })
|
||||
}
|
||||
|
||||
/** A prompt's blocks, read: the text typed, and the files that came with it (not written yet). */
|
||||
export interface Blocks {
|
||||
text: string
|
||||
files: { name: string; mimeType: string; data: Buffer }[]
|
||||
}
|
||||
|
||||
function decoded(b64: unknown, what: string): Buffer {
|
||||
if (typeof b64 !== "string") throw new RpcError(INVALID_PARAMS, `${what} has no data`)
|
||||
// The size is known before decoding: base64 is 4 characters for 3 bytes.
|
||||
if (b64.length * 0.75 > BLOCK_LIMIT + 3) throw new RpcError(INVALID_PARAMS, `${what} is larger than ${BLOCK_LIMIT / 1024 / 1024} MB; send a smaller one`)
|
||||
return Buffer.from(b64, "base64")
|
||||
}
|
||||
|
||||
const EXT: Record<string, string> = Object.fromEntries(Object.entries(IMAGE_TYPES).map(([ext, mime]) => [mime, ext]))
|
||||
|
||||
function safeName(name: string, fallback: string): string {
|
||||
const n = basename(name || "").replace(/[^\w.+-]/g, "_").replace(/^\.+/, "").slice(0, 80)
|
||||
return n || fallback
|
||||
}
|
||||
|
||||
function nameOfUri(uri: unknown): string {
|
||||
if (typeof uri !== "string" || !uri) return ""
|
||||
try {
|
||||
return decodeURIComponent(new URL(uri, "file:///").pathname.split("/").pop() ?? "")
|
||||
} catch {
|
||||
return uri.split("/").pop() ?? ""
|
||||
}
|
||||
}
|
||||
|
||||
/** The blocks of `session/prompt`: text, image, resource (text or blob), resource_link. */
|
||||
export function readBlocks(blocks: unknown): Blocks {
|
||||
if (!Array.isArray(blocks)) throw new RpcError(INVALID_PARAMS, "prompt must be an array of content blocks")
|
||||
const parts: string[] = []
|
||||
const files: Blocks["files"] = []
|
||||
let total = 0
|
||||
const add = (f: Blocks["files"][number]) => {
|
||||
if (f.data.length > BLOCK_LIMIT) throw new RpcError(INVALID_PARAMS, `${f.name} is larger than ${BLOCK_LIMIT / 1024 / 1024} MB; send a smaller one`)
|
||||
total += f.data.length
|
||||
if (total > PROMPT_LIMIT) throw new RpcError(INVALID_PARAMS, `the attachments come to more than ${PROMPT_LIMIT / 1024 / 1024} MB together; send fewer at a time`)
|
||||
files.push(f)
|
||||
}
|
||||
for (const b of blocks as any[]) {
|
||||
if (b?.type === "text" && typeof b.text === "string") parts.push(b.text)
|
||||
else if (b?.type === "image") {
|
||||
const mime = String(b.mimeType ?? "")
|
||||
if (!IMAGE_MIMES.includes(mime)) throw new RpcError(INVALID_PARAMS, `an image must be PNG, JPEG, WebP or GIF, not ${mime || "unnamed"}`)
|
||||
const n = files.length + 1
|
||||
const base = safeName(nameOfUri(b.uri), `image-${n}`)
|
||||
add({ name: extname(base) ? base : `${base}${EXT[mime] ?? ".png"}`, mimeType: mime, data: decoded(b.data, "an image") })
|
||||
} else if (b?.type === "resource" && b.resource && typeof b.resource === "object") {
|
||||
const r = b.resource
|
||||
const name = safeName(nameOfUri(r.uri), `file-${files.length + 1}${typeof r.text === "string" ? ".txt" : ""}`)
|
||||
if (typeof r.text === "string") add({ name, mimeType: String(r.mimeType ?? "text/plain"), data: Buffer.from(r.text, "utf8") })
|
||||
else if (r.blob !== undefined) add({ name, mimeType: String(r.mimeType ?? "application/octet-stream"), data: decoded(r.blob, name) })
|
||||
} else if (b?.type === "resource_link" && typeof b.uri === "string") parts.push(`(see ${b.uri})`)
|
||||
}
|
||||
return { text: parts.join("\n\n").trim(), files }
|
||||
}
|
||||
|
||||
/** What a prompt becomes. */
|
||||
export type Prepared =
|
||||
| { kind: "run"; prompt: string; shown: string; extra: (string | ImagePart)[]; attachments: AttachmentInfo[]; atPaths: string[]; model?: string; mode?: Mode; command?: string }
|
||||
| { kind: "action"; name: "compact" | "undo" | "plan"; arg: string }
|
||||
| { kind: "message"; name: string; text: string }
|
||||
|
||||
/** Whether an @path in a prompt from the web may be attached: where it really is — links
|
||||
* followed — inside the session's project and inside the device's `remote.roots` (when given),
|
||||
* and not a protected path. Anything else stays text: an @ is never a way out of the project. */
|
||||
export function mayAttach(abs: string, root: string, roots?: string[]): boolean {
|
||||
const real = realPath(abs)
|
||||
const within = (p: string, dir: string) => {
|
||||
const d = realPath(dir)
|
||||
return p === d || p.startsWith(d.endsWith("/") ? d : d + "/")
|
||||
}
|
||||
if (!within(real, root)) return false
|
||||
if (roots && !roots.some((r) => within(real, expandHome(r)))) return false
|
||||
return !protectedPath(real) && !protectedPath(abs)
|
||||
}
|
||||
|
||||
function mimeOf(path: string): string {
|
||||
return IMAGE_TYPES[extname(path).toLowerCase()] ?? (/\.(md|txt|ts|tsx|js|py|json|ya?ml|toml|sh|css|html|rs|go|c|h|java)$/i.test(path) ? "text/plain" : "application/octet-stream")
|
||||
}
|
||||
|
||||
/** The prompt, for `app`'s session `sessionId`: files written, the command expanded, @files read.
|
||||
* Throws INVALID_PARAMS with a sentence for a prompt that cannot be sent. */
|
||||
export function preparePrompt(app: App, sessionId: string, blocks: unknown, o: { roots?: string[] } = {}): Prepared {
|
||||
const { text, files } = readBlocks(blocks)
|
||||
if (!text && !files.length) throw new RpcError(INVALID_PARAMS, "the prompt has no text")
|
||||
let cmd = expandSlash(app, text)
|
||||
// A command that sends no prompt has nowhere to put the files: then the text is only text.
|
||||
if (files.length && cmd && cmd.kind !== "prompt") cmd = undefined
|
||||
if (cmd?.kind === "action" || cmd?.kind === "message") return cmd
|
||||
const ctx = app.engine.o.toolCtx
|
||||
const vision = app.engine.model.spec.vision === true
|
||||
// The files written first, under names that stay apart from the ones before.
|
||||
const saved: { abs: string; info: AttachmentInfo }[] = []
|
||||
if (files.length) {
|
||||
const dir = attachmentDir(sessionId)
|
||||
mkdirSync(dir, { recursive: true, mode: 0o700 })
|
||||
const stamp = new Date().toISOString().replace(/[-:]/g, "").replace(/\..*$/, "")
|
||||
for (const f of files) {
|
||||
// A part of a uuid after the stamp: two uploads of one name (in one prompt, or one second) stay two.
|
||||
const abs = resolve(dir, `${stamp}-${crypto.randomUUID().slice(0, 8)}-${f.name}`)
|
||||
writeFileSync(abs, f.data, { mode: 0o600 })
|
||||
saved.push({ abs, info: { name: f.name, mimeType: f.mimeType, size: f.data.length } })
|
||||
}
|
||||
}
|
||||
const prompt = cmd?.kind === "prompt" ? cmd.prompt : text || `(attached: ${saved.map((s) => s.info.name).join(", ")})`
|
||||
const shown = cmd?.kind === "prompt" ? cmd.shown : prompt
|
||||
// @files only from what the person typed — never from a command's own text (Working-notes #18).
|
||||
// And only what may be read from here: the web is a remote keyboard, not one at this machine.
|
||||
const mentioned = attachmentsFor(cmd?.kind === "prompt" ? cmd.attachFrom : text, ctx, vision, (abs) => mayAttach(abs, app.project.root, o.roots))
|
||||
const uploaded = saved.flatMap((s) => attachmentsFor(`@${s.abs}`, ctx, vision))
|
||||
const extra = [...mentioned, ...uploaded].flatMap((a) => (a.image ? [a.text, a.image] : [a.text]))
|
||||
const attachments: AttachmentInfo[] = [...mentioned.map((a) => mentionInfo(a.path, ctx.cwd)), ...saved.map((s) => s.info)]
|
||||
return {
|
||||
kind: "run",
|
||||
prompt,
|
||||
shown,
|
||||
extra,
|
||||
attachments,
|
||||
atPaths: mentioned.map((a) => a.path),
|
||||
...(cmd?.kind === "prompt" ? { command: cmd.name, ...(cmd.model ? { model: cmd.model } : {}), ...(cmd.mode ? { mode: cmd.mode } : {}) } : {}),
|
||||
}
|
||||
}
|
||||
|
||||
/** An @file, @directory or @image of a prompt, as a chip. */
|
||||
export function mentionInfo(path: string, cwd: string): AttachmentInfo {
|
||||
return { name: path, mimeType: path.endsWith("/") ? "inode/directory" : mimeOf(path), size: sizeOf(resolve(cwd, path)) }
|
||||
}
|
||||
|
||||
function sizeOf(p: string): number {
|
||||
try {
|
||||
return statSync(p).size
|
||||
} catch {
|
||||
return 0
|
||||
}
|
||||
}
|
||||
|
||||
/** The attachments of a stored user message, from the parts the expander wrote (`<file path=…>`,
|
||||
* `<image path=…>`, `<directory path=…>`), for `_lembas/session/history`. */
|
||||
export function storedAttachments(texts: string[], cwd: string): AttachmentInfo[] {
|
||||
const out: AttachmentInfo[] = []
|
||||
for (const t of texts) {
|
||||
const m = /^<(file|image|directory) path="([^"]*)"/.exec(t)
|
||||
if (!m) continue
|
||||
const path = m[2]!
|
||||
const abs = resolve(cwd, path)
|
||||
// An upload is stored under a name with its time in front; the chip shows the name it came with.
|
||||
const name = abs.startsWith(join(paths.state, "attachments") + "/") ? basename(path).replace(/^\d{8}T\d{6}-(?:[0-9a-f]{8}-)?/, "") : path
|
||||
out.push({ name, mimeType: m[1] === "directory" ? "inode/directory" : mimeOf(path), size: sizeOf(abs) })
|
||||
}
|
||||
return out
|
||||
}
|
||||
+162
@@ -0,0 +1,162 @@
|
||||
// JSON-RPC 2.0 over lines: the framing ACP uses on stdio, and what the LLeMbas link carries one
|
||||
// message per WebSocket frame. Both ends may call and both may be called; a request waits for the
|
||||
// answer with the same id. Written here rather than taken from the ACP SDK so the same few rules
|
||||
// are what LLeMbas implements in Python, and nothing about them is hidden in a dependency.
|
||||
|
||||
export interface Transport {
|
||||
send(text: string): void
|
||||
onMessage(fn: (text: string) => void): void
|
||||
onClose(fn: () => void): void
|
||||
close(): void
|
||||
}
|
||||
|
||||
export class RpcError extends Error {
|
||||
constructor(
|
||||
public code: number,
|
||||
message: string,
|
||||
public data?: unknown,
|
||||
) {
|
||||
super(message)
|
||||
}
|
||||
}
|
||||
|
||||
export const PARSE_ERROR = -32700
|
||||
export const INVALID_REQUEST = -32600
|
||||
export const METHOD_NOT_FOUND = -32601
|
||||
export const INVALID_PARAMS = -32602
|
||||
export const INTERNAL_ERROR = -32603
|
||||
|
||||
type Handler = (params: any) => unknown | Promise<unknown>
|
||||
|
||||
export class Peer {
|
||||
private handlers = new Map<string, Handler>()
|
||||
private notices = new Map<string, Handler>()
|
||||
private waiting = new Map<number | string, { resolve: (v: any) => void; reject: (e: Error) => void }>()
|
||||
private seq = 0
|
||||
private closed = false
|
||||
private closers: (() => void)[] = []
|
||||
|
||||
constructor(private t: Transport) {
|
||||
t.onMessage((text) => void this.receive(text))
|
||||
t.onClose(() => {
|
||||
this.closed = true
|
||||
for (const w of this.waiting.values()) w.reject(new RpcError(INTERNAL_ERROR, "the connection closed"))
|
||||
this.waiting.clear()
|
||||
for (const fn of this.closers) fn()
|
||||
})
|
||||
}
|
||||
|
||||
/** Answer `method` requests. */
|
||||
handle(method: string, fn: Handler) {
|
||||
this.handlers.set(method, fn)
|
||||
}
|
||||
|
||||
/** React to `method` notifications. */
|
||||
on(method: string, fn: Handler) {
|
||||
this.notices.set(method, fn)
|
||||
}
|
||||
|
||||
onClose(fn: () => void) {
|
||||
this.closers.push(fn)
|
||||
}
|
||||
|
||||
get isClosed() {
|
||||
return this.closed
|
||||
}
|
||||
|
||||
request<T = unknown>(method: string, params?: unknown, timeoutMs?: number): Promise<T> {
|
||||
if (this.closed) return Promise.reject(new RpcError(INTERNAL_ERROR, "the connection is closed"))
|
||||
const id = ++this.seq
|
||||
return new Promise<T>((resolve, reject) => {
|
||||
let timer: ReturnType<typeof setTimeout> | undefined
|
||||
if (timeoutMs)
|
||||
timer = setTimeout(() => {
|
||||
this.waiting.delete(id)
|
||||
reject(new RpcError(INTERNAL_ERROR, `${method}: no answer in ${Math.round(timeoutMs / 1000)}s`))
|
||||
}, timeoutMs)
|
||||
this.waiting.set(id, {
|
||||
resolve: (v) => (clearTimeout(timer), resolve(v)),
|
||||
reject: (e) => (clearTimeout(timer), reject(e)),
|
||||
})
|
||||
this.write({ jsonrpc: "2.0", id, method, ...(params === undefined ? {} : { params }) })
|
||||
})
|
||||
}
|
||||
|
||||
notify(method: string, params?: unknown) {
|
||||
if (this.closed) return
|
||||
this.write({ jsonrpc: "2.0", method, ...(params === undefined ? {} : { params }) })
|
||||
}
|
||||
|
||||
close() {
|
||||
this.t.close()
|
||||
}
|
||||
|
||||
private write(message: unknown) {
|
||||
try {
|
||||
this.t.send(JSON.stringify(message))
|
||||
} catch {}
|
||||
}
|
||||
|
||||
private async receive(text: string) {
|
||||
let msg: any
|
||||
try {
|
||||
msg = JSON.parse(text)
|
||||
} catch {
|
||||
return this.write({ jsonrpc: "2.0", id: null, error: { code: PARSE_ERROR, message: "Parse error" } })
|
||||
}
|
||||
if (!msg || msg.jsonrpc !== "2.0") return
|
||||
// An answer to one of ours.
|
||||
if (msg.method === undefined && msg.id !== undefined) {
|
||||
const w = this.waiting.get(msg.id)
|
||||
if (!w) return
|
||||
this.waiting.delete(msg.id)
|
||||
if (msg.error) w.reject(new RpcError(msg.error.code ?? INTERNAL_ERROR, msg.error.message ?? "error", msg.error.data))
|
||||
else w.resolve(msg.result)
|
||||
return
|
||||
}
|
||||
const method = String(msg.method ?? "")
|
||||
// A notification: nothing is answered, even an unknown one.
|
||||
if (msg.id === undefined || msg.id === null) {
|
||||
const fn = this.notices.get(method)
|
||||
if (fn) {
|
||||
try {
|
||||
await fn(msg.params)
|
||||
} catch {}
|
||||
}
|
||||
return
|
||||
}
|
||||
const fn = this.handlers.get(method)
|
||||
if (!fn) return this.write({ jsonrpc: "2.0", id: msg.id, error: { code: METHOD_NOT_FOUND, message: `Method not found: ${method}` } })
|
||||
try {
|
||||
const result = await fn(msg.params ?? {})
|
||||
this.write({ jsonrpc: "2.0", id: msg.id, result: result ?? null })
|
||||
} catch (e) {
|
||||
const err = e instanceof RpcError ? e : new RpcError(INTERNAL_ERROR, (e as Error).message)
|
||||
this.write({ jsonrpc: "2.0", id: msg.id, error: { code: err.code, message: err.message, ...(err.data === undefined ? {} : { data: err.data }) } })
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** Newline-delimited JSON on a pair of streams: ACP over stdio. */
|
||||
export function stdioTransport(input: NodeJS.ReadableStream = process.stdin, output: NodeJS.WritableStream = process.stdout): Transport {
|
||||
let onMsg: (t: string) => void = () => {}
|
||||
let onEnd: () => void = () => {}
|
||||
let buf = ""
|
||||
input.setEncoding?.("utf8")
|
||||
input.on("data", (chunk: string | Buffer) => {
|
||||
buf += chunk.toString()
|
||||
let nl: number
|
||||
while ((nl = buf.indexOf("\n")) >= 0) {
|
||||
const line = buf.slice(0, nl).trim()
|
||||
buf = buf.slice(nl + 1)
|
||||
if (line) onMsg(line)
|
||||
}
|
||||
})
|
||||
input.on("end", () => onEnd())
|
||||
return {
|
||||
send: (text) => void output.write(text + "\n"),
|
||||
onMessage: (fn) => (onMsg = fn),
|
||||
onClose: (fn) => (onEnd = fn),
|
||||
close: () => onEnd(),
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,560 @@
|
||||
// A terminal's side of the hub (hub.ts): its session shown and used in the web UI too.
|
||||
//
|
||||
// With the service running, the session the terminal has open is shared by itself — within the
|
||||
// device's `remote:` limits — and follows /new and /sessions. /remote shares it by name: through
|
||||
// the service when it runs, otherwise by holding the link here, for as long as this terminal runs.
|
||||
//
|
||||
// Shared, the web UI is a second keyboard on the same session: its prompt runs here as if typed,
|
||||
// its message mid-task joins the task, its stop stops it, and an approval is asked on both sides —
|
||||
// whichever answers first decides, and the other side's card goes away.
|
||||
import type { App } from "../app.ts"
|
||||
import type { AskReply, Asker, AttachmentInfo, Event } from "../bus/index.ts"
|
||||
import { asMode, type Mode } from "../config/schema.ts"
|
||||
import type { ImagePart } from "../provider/types.ts"
|
||||
import { commandSet, sourcesOf } from "../session/commands.ts"
|
||||
import type { PlanReply } from "../tool/plan_exit.ts"
|
||||
import type { QuestionReply, QuestionRequest } from "../tool/question.ts"
|
||||
import { clampMode, effortOf, markRemote, permissionParams, planParams, planReplyFrom, questionParams, questionReplyFrom, replyFrom, STOP, type PromptResult } from "./agent.ts"
|
||||
import { sessionTurnFile, sessionTurns, type TurnLog } from "./turns.ts"
|
||||
import { dropAttachments, preparePrompt } from "./prompt.ts"
|
||||
import { connectHub, Hub } from "./hub.ts"
|
||||
import { addPendingDeleted } from "./pending.ts"
|
||||
import { REMOTE_META, type Store } from "../session/store.ts"
|
||||
import { limitsFromConfig, runLink } from "./link.ts"
|
||||
import { INVALID_PARAMS, RpcError, type Peer } from "./rpc.ts"
|
||||
|
||||
type AskRequest = Parameters<Asker["ask"]>[0]
|
||||
|
||||
/** An approval shown here, which the web UI's answer can take down. */
|
||||
export interface LocalAsk<R = AskReply> {
|
||||
reply: Promise<R>
|
||||
dismiss(r: R): void
|
||||
}
|
||||
|
||||
/** A prompt from the web UI, made into what the TUI sends: the expanded text, what the
|
||||
* transcript shows, the attachments for the model and as chips, its turn id, and a command's model
|
||||
* and mode for this one prompt. */
|
||||
export interface SharedPrompt {
|
||||
prompt: string
|
||||
shown: string
|
||||
extra: (string | ImagePart)[]
|
||||
attachments: AttachmentInfo[]
|
||||
turnId: string
|
||||
model?: string
|
||||
mode?: Mode
|
||||
}
|
||||
|
||||
export interface ShareHooks {
|
||||
/** A prompt from the web UI: run it as if typed here, calling `started` when it stops waiting for
|
||||
* the turn before it. Resolves with the engine's stop reason. */
|
||||
prompt(input: SharedPrompt, started: () => void): Promise<string>
|
||||
/** A command from the web UI that sends no prompt — compact, undo, plan; absent, the
|
||||
* share does it on the app and says `changed`. */
|
||||
action?(name: "compact" | "undo" | "plan", arg: string): Promise<Record<string, unknown>>
|
||||
notice(text: string, level?: "info" | "warn"): void
|
||||
/** /compact, asked from the web UI. Resolves with the summary. */
|
||||
compact(): Promise<string>
|
||||
/** The web UI deleted this session: it is gone here too, and a new one starts. */
|
||||
deleted(): void
|
||||
/** Shared or not, mode, model or effort changed: redraw. */
|
||||
changed(): void
|
||||
}
|
||||
|
||||
let current: Share | undefined
|
||||
|
||||
/** Before a session is resumed in a terminal: the hub lets go of its own copy of it, or says why it
|
||||
* cannot (working on a prompt from the web UI, or open in another terminal). */
|
||||
export async function releaseForResume(id: string): Promise<string | undefined> {
|
||||
return current ? current.release(id) : releaseVia(undefined, id)
|
||||
}
|
||||
|
||||
async function releaseVia(hub: Peer | undefined, id: string): Promise<string | undefined> {
|
||||
const peer = hub ?? (await connectHub())
|
||||
if (!peer) return undefined
|
||||
try {
|
||||
await peer.request("hub/release", { sessionId: id }, 5000)
|
||||
return undefined
|
||||
} catch (e) {
|
||||
return (e as Error).message
|
||||
} finally {
|
||||
if (peer !== hub) peer.close()
|
||||
}
|
||||
}
|
||||
|
||||
/** Delete a stored session here and in the web UI (from Share.deleted): the hub lets go of
|
||||
* its copy first (or says why it cannot — working on a prompt from the web UI, open in another
|
||||
* terminal), the rows and what hangs off them go, and the hub is told so that the instance deletes
|
||||
* the chat. `hub`: a connection to it already open (a terminal's); without one, one is made for
|
||||
* this delete when a hub runs — `lembas sessions delete` reaches it the way a terminal does. With
|
||||
* no hub at all, the word waits in the pending list for the service's next link. */
|
||||
export async function deleteEverywhere(store: Store, id: string, hub?: Peer): Promise<string | undefined> {
|
||||
const peer = hub ?? (await connectHub())
|
||||
try {
|
||||
if (peer) {
|
||||
try {
|
||||
await peer.request("hub/release", { sessionId: id }, 5000)
|
||||
} catch (e) {
|
||||
return (e as Error).message
|
||||
}
|
||||
}
|
||||
store.deleteSession(id)
|
||||
dropAttachments(id)
|
||||
sessionTurnFile(id).drop()
|
||||
if (!peer) return void addPendingDeleted([id])
|
||||
peer.notify("hub/deleted", { sessionId: id })
|
||||
// A connection made for this delete closes next; a request after the notification is answered
|
||||
// only once the hub has read the notification, so the word is not lost with the socket.
|
||||
if (peer !== hub) await peer.request("hub/hello", { pid: process.pid }, 5000).catch(() => {})
|
||||
return undefined
|
||||
} finally {
|
||||
if (peer && peer !== hub) peer.close()
|
||||
}
|
||||
}
|
||||
|
||||
/** The session a terminal leaves, deleted when nothing was ever said in it: every start
|
||||
* makes one, and a TUI opened and closed again left it behind as an empty row in /sessions and,
|
||||
* shared, an empty chat in the web UI. The same rule as a resume's discard (app.ts): no message
|
||||
* rows at all, and the instance told only when it was shown the session (REMOTE_META). Kept when a
|
||||
* running hub will not let go of it — open in another terminal, or the service working in it.
|
||||
* Called after the terminal's own link to the hub is closed, so that link no longer counts. */
|
||||
export async function discardOnExit(store: Store, id: string | undefined): Promise<boolean> {
|
||||
if (!id || !store.session(id) || !store.isEmpty(id)) return false
|
||||
const peer = await connectHub()
|
||||
try {
|
||||
if (peer) {
|
||||
try {
|
||||
await peer.request("hub/release", { sessionId: id }, 5000)
|
||||
} catch {
|
||||
return false
|
||||
}
|
||||
}
|
||||
// Asked again after the wait: a prompt from the web UI could have landed meanwhile.
|
||||
if (!store.isEmpty(id)) return false
|
||||
const shown = store.meta<boolean>(id, REMOTE_META) === true
|
||||
if (!store.deleteSession(id)) return false
|
||||
dropAttachments(id)
|
||||
sessionTurnFile(id).drop()
|
||||
if (shown) {
|
||||
if (!peer) addPendingDeleted([id])
|
||||
else {
|
||||
peer.notify("hub/deleted", { sessionId: id })
|
||||
await peer.request("hub/hello", { pid: process.pid }, 5000).catch(() => {})
|
||||
}
|
||||
}
|
||||
return true
|
||||
} finally {
|
||||
peer?.close()
|
||||
}
|
||||
}
|
||||
|
||||
export class Share {
|
||||
private hub?: Peer
|
||||
/** The hub is the service: every session here is shared by itself. */
|
||||
private service = false
|
||||
/** The link this terminal holds for /remote, when nobody else does. */
|
||||
private own?: { hub: Hub; stop: AbortController }
|
||||
/** The session shared now. */
|
||||
sharedId?: string
|
||||
/** Shared by name (/remote): it stays shared across a dropped hub, and the limits do not decide. */
|
||||
explicit = false
|
||||
private retry?: ReturnType<typeof setInterval>
|
||||
private off: () => void
|
||||
private closed = false
|
||||
/** The shared session's turns, by id: a prompt sent twice runs once — on the file the
|
||||
* service uses too, so a turn run here is not run again there after the session moves. */
|
||||
private turnLog?: { id: string; log: TurnLog<PromptResult> }
|
||||
private get turns(): TurnLog<PromptResult> {
|
||||
const id = this.app.engine.sessionId ?? ""
|
||||
if (this.turnLog?.id !== id) this.turnLog = { id, log: sessionTurns<PromptResult>(id) }
|
||||
return this.turnLog.log
|
||||
}
|
||||
/** The command set last sent to the hub. */
|
||||
private commandsSent?: string
|
||||
|
||||
constructor(
|
||||
private app: App,
|
||||
private hooks: ShareHooks,
|
||||
) {
|
||||
this.off = app.bus.on((e) => this.onEvent(e))
|
||||
current = this
|
||||
}
|
||||
|
||||
/** Find the hub, and keep looking for one (the service may start later). */
|
||||
async start() {
|
||||
await this.connect()
|
||||
this.retry = setInterval(() => void this.reconnect(), 10_000)
|
||||
this.retry.unref?.()
|
||||
}
|
||||
|
||||
/** The status bar's word: shared, and how. */
|
||||
label(): string {
|
||||
if (!this.sharedId) return ""
|
||||
return this.own ? "⇄ web (this terminal)" : "⇄ web"
|
||||
}
|
||||
|
||||
status(): string {
|
||||
if (this.sharedId) return `this session is shared with the web UI ${this.own ? "— the link is held by this terminal, so it lasts as long as the terminal does" : "through the service"}`
|
||||
if (this.hub && this.service) return "not shared: this session is outside what the service shares (remote.roots, trust) — /remote shares it anyway"
|
||||
if (this.hub) return "not shared — /remote shares it"
|
||||
return "not shared, and no service runs here — /remote shares it for as long as this terminal runs"
|
||||
}
|
||||
|
||||
release(id: string) {
|
||||
return releaseVia(this.hub, id)
|
||||
}
|
||||
|
||||
/** A session deleted here (/sessions, /delete): the hub lets go of its copy, and the web UI
|
||||
* deletes the chat it is. Returns why not, when it cannot be (working in the web UI). */
|
||||
async deleted(id: string): Promise<string | undefined> {
|
||||
if (!this.app.store) return undefined
|
||||
return deleteEverywhere(this.app.store, id, this.hub)
|
||||
}
|
||||
|
||||
/** /remote on or off. Returns what to say. */
|
||||
async remote(on: boolean): Promise<string> {
|
||||
if (!on) {
|
||||
const id = this.sharedId
|
||||
this.explicit = false
|
||||
if (!id) return "this session is not shared"
|
||||
this.sharedId = undefined
|
||||
await this.hub?.request("hub/unshare", { sessionId: id }, 5000).catch(() => {})
|
||||
if (this.own) this.dropOwn()
|
||||
this.hooks.changed()
|
||||
return "no longer shared with the web UI"
|
||||
}
|
||||
if (!this.hub) {
|
||||
const why = await this.becomeHub()
|
||||
if (why) return why
|
||||
}
|
||||
await this.share(true)
|
||||
return this.sharedId ? this.status() : "could not share this session"
|
||||
}
|
||||
|
||||
/** An approval of the shared session goes to both sides; the first answer counts. */
|
||||
async ask(req: AskRequest, local: LocalAsk): Promise<AskReply> {
|
||||
const id = this.sharedId
|
||||
const hub = this.hub
|
||||
if (!id || !hub || this.app.engine.sessionId !== id) return local.reply
|
||||
const params = permissionParams(id, req, { patient: true })
|
||||
// A web UI that cannot show it (or a link that drops) never answers: the terminal's card does.
|
||||
const remote = hub.request("hub/ask", { sessionId: id, params }).then(
|
||||
(a) => replyFrom(a, !req.request.alwaysAsk),
|
||||
() => new Promise<never>(() => {}),
|
||||
)
|
||||
const won = await Promise.race([local.reply.then((r) => ({ r, here: true })), remote.then((r) => ({ r, here: false }))])
|
||||
if (won.here) hub.notify("hub/settled", { sessionId: id, toolCallId: params.toolCall.toolCallId })
|
||||
else {
|
||||
local.dismiss(won.r)
|
||||
this.hooks.notice(`answered in the web UI: ${won.r.kind === "deny" ? "denied" : "allowed"}`)
|
||||
}
|
||||
return won.r
|
||||
}
|
||||
|
||||
/** ask_user in the shared session: the card here and in the web UI, the first answer wins. */
|
||||
question(req: QuestionRequest, callId: string | undefined, local: LocalAsk<QuestionReply>): Promise<QuestionReply> {
|
||||
return this.card(local, "hub/question", (id) => questionParams(id, req, callId, { patient: true }), (a) => questionReplyFrom(a, req), (r) => (r.dismissed ? "dismissed" : "answered"))
|
||||
}
|
||||
|
||||
/** plan_submit in the shared session, likewise. */
|
||||
plan(req: { path: string; text: string }, callId: string | undefined, local: LocalAsk<PlanReply>): Promise<PlanReply> {
|
||||
return this.card(local, "hub/plan", (id) => planParams(id, req, callId, { patient: true }), (a) => planReplyFrom(a), (r) => (r.kind === "approve" ? `approved (${r.mode})` : r.kind === "revise" ? "to be revised" : "not decided"))
|
||||
}
|
||||
|
||||
private async card<R>(local: LocalAsk<R>, method: string, params: (sessionId: string) => { toolCallId: string }, read: (answer: unknown) => R, said: (r: R) => string): Promise<R> {
|
||||
const id = this.sharedId
|
||||
const hub = this.hub
|
||||
if (!id || !hub || this.app.engine.sessionId !== id) return local.reply
|
||||
const p = params(id)
|
||||
// A web UI that cannot show it (or a link that drops) never answers: the terminal's card does.
|
||||
const remote = hub.request(method, { sessionId: id, params: p }).then(read, () => new Promise<never>(() => {}))
|
||||
const won = await Promise.race([local.reply.then((r) => ({ r, here: true })), remote.then((r) => ({ r, here: false }))])
|
||||
if (won.here) hub.notify("hub/settled", { sessionId: id, toolCallId: p.toolCallId })
|
||||
else {
|
||||
local.dismiss(won.r)
|
||||
this.hooks.notice(`answered in the web UI: ${said(won.r)}`)
|
||||
}
|
||||
return won.r
|
||||
}
|
||||
|
||||
/** The command set, to the hub when it changed. */
|
||||
private sendCommands(always = false) {
|
||||
const id = this.sharedId
|
||||
if (!id || !this.hub) return
|
||||
let list
|
||||
try {
|
||||
list = commandSet(sourcesOf(this.app))
|
||||
} catch {
|
||||
return
|
||||
}
|
||||
const key = JSON.stringify(list)
|
||||
if (!always && key === this.commandsSent) return
|
||||
this.commandsSent = key
|
||||
this.hub.notify("hub/commands", { sessionId: id, commands: list })
|
||||
}
|
||||
|
||||
close() {
|
||||
this.closed = true
|
||||
if (current === this) current = undefined
|
||||
clearInterval(this.retry)
|
||||
this.off()
|
||||
// Closing the connection is the hub's word that every session of this terminal is gone.
|
||||
this.hub?.close()
|
||||
this.dropOwn()
|
||||
}
|
||||
|
||||
private dropOwn() {
|
||||
if (!this.own) return
|
||||
this.own.stop.abort()
|
||||
this.own.hub.close()
|
||||
this.own = undefined
|
||||
}
|
||||
|
||||
private async connect(): Promise<boolean> {
|
||||
const peer = await connectHub()
|
||||
if (!peer) return false
|
||||
try {
|
||||
const hello = await peer.request<{ service?: boolean }>("hub/hello", { pid: process.pid }, 5000)
|
||||
this.hub = peer
|
||||
this.service = hello?.service === true
|
||||
this.wire(peer)
|
||||
peer.onClose(() => {
|
||||
if (this.hub !== peer) return
|
||||
this.hub = undefined
|
||||
const was = this.sharedId
|
||||
this.sharedId = undefined
|
||||
if (was && !this.closed) this.hooks.notice("the link to the web UI went away — the session goes on here", "warn")
|
||||
this.hooks.changed()
|
||||
})
|
||||
await this.share(this.explicit)
|
||||
return true
|
||||
} catch {
|
||||
peer.close()
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
private async reconnect() {
|
||||
if (this.closed || this.hub) return
|
||||
if (await this.connect()) return
|
||||
// Shared by name, and nobody holds the link any more: hold it here.
|
||||
if (this.explicit) await this.becomeHub()
|
||||
}
|
||||
|
||||
/** Hold the link in this terminal, for /remote while no service runs. Returns why not, if not. */
|
||||
private async becomeHub(): Promise<string | undefined> {
|
||||
const { instances } = await import("../lembas/login.ts")
|
||||
if (!Object.keys(instances()).length) return "not logged in to a LLeMbas instance — /login first"
|
||||
const hub = new Hub({ service: false, limits: limitsFromConfig() })
|
||||
if (!(await hub.listen())) return (await this.connect()) ? undefined : "something holds the hub's socket but does not answer"
|
||||
const stop = new AbortController()
|
||||
this.own = { hub, stop }
|
||||
void runLink({
|
||||
hub,
|
||||
sharedOnly: true,
|
||||
signal: stop.signal,
|
||||
onStatus: (s) => {
|
||||
if (s.state === "stopped" && s.detail !== "stopped") this.hooks.notice(`the link to the web UI stopped: ${s.detail}`, "warn")
|
||||
},
|
||||
}).catch((e) => {
|
||||
this.hooks.notice(`/remote: ${(e as Error).message}`, "warn")
|
||||
this.dropOwn()
|
||||
})
|
||||
return (await this.connect()) ? undefined : "could not reach the hub this terminal started"
|
||||
}
|
||||
|
||||
private async share(explicit: boolean) {
|
||||
const hub = this.hub
|
||||
const id = this.app.engine.sessionId
|
||||
if (!hub || !id) return
|
||||
try {
|
||||
const r = await hub.request<{ shared?: boolean; reason?: string }>(
|
||||
"hub/share",
|
||||
{
|
||||
sessionId: id,
|
||||
cwd: this.app.project.root,
|
||||
// Where it works, which @path mentions are read from: a session started in a
|
||||
// subdirectory is not at its project's root.
|
||||
workdir: this.app.engine.o.toolCtx.cwd,
|
||||
title: this.app.store?.session(id)?.title ?? "",
|
||||
model: this.app.engine.model.ref,
|
||||
// Which connection that model is spoken to through, so the hub can say whether it
|
||||
// is the instance's.
|
||||
modelConnection: this.app.engine.model.connectionName,
|
||||
mode: this.app.engine.mode,
|
||||
effort: this.app.engine.effort ?? "off",
|
||||
busy: Boolean(this.app.turns.current),
|
||||
explicit,
|
||||
},
|
||||
10_000,
|
||||
)
|
||||
if (r?.shared) {
|
||||
// In the web UI's hands from now on: deletable from there even once this terminal has let
|
||||
// go of it, wherever it is (agent.ts, deletableFor).
|
||||
markRemote(this.app.store, id)
|
||||
this.sharedId = id
|
||||
this.explicit = explicit
|
||||
this.sendCommands(true)
|
||||
} else {
|
||||
this.sharedId = undefined
|
||||
if (explicit) this.hooks.notice(`not shared: ${r?.reason ?? "the hub said no"}`, "warn")
|
||||
}
|
||||
} catch (e) {
|
||||
this.sharedId = undefined
|
||||
if (explicit) this.hooks.notice(`not shared: ${(e as Error).message}`, "warn")
|
||||
}
|
||||
this.hooks.changed()
|
||||
}
|
||||
|
||||
/** compact, undo or plan sent from the web UI: the TUI's own way when it has one. */
|
||||
private async action(name: "compact" | "undo" | "plan", arg: string): Promise<Record<string, unknown>> {
|
||||
if (this.hooks.action) return this.hooks.action(name, arg)
|
||||
if (name === "compact") return { summary: await this.hooks.compact() }
|
||||
if (name === "undo") {
|
||||
const r = this.app.turns.undo()
|
||||
this.hooks.changed()
|
||||
return { undone: r ?? null }
|
||||
}
|
||||
this.app.engine.mode = "plan"
|
||||
this.app.bus.emit({ type: "mode", mode: "plan" })
|
||||
this.hooks.changed()
|
||||
return { mode: "plan" }
|
||||
}
|
||||
|
||||
private onEvent(e: Event) {
|
||||
// The empty session a resume left (app.ts): gone from the store already; the web UI's
|
||||
// chat for it goes too, where it was shown one — through the hub, or the pending list.
|
||||
if (e.type === "discarded") {
|
||||
dropAttachments(e.id)
|
||||
sessionTurnFile(e.id).drop()
|
||||
if (!e.shown) return
|
||||
if (this.hub) this.hub.notify("hub/deleted", { sessionId: e.id })
|
||||
else addPendingDeleted([e.id])
|
||||
return
|
||||
}
|
||||
// Every turn here, typed or sent from the web UI, by its id.
|
||||
if (e.type === "task" && e.turnId) e.state === "start" ? this.turns.started(e.turnId) : this.turns.ended(e.turnId)
|
||||
if (e.type === "session") {
|
||||
if (e.id === this.sharedId) return
|
||||
// /new or /sessions: the old one stops being shared, and the new one is shared as any is.
|
||||
const was = this.sharedId
|
||||
this.sharedId = undefined
|
||||
this.explicit = false
|
||||
if (was) void this.hub?.request("hub/unshare", { sessionId: was }, 5000).catch(() => {})
|
||||
void this.share(false)
|
||||
return
|
||||
}
|
||||
if (this.sharedId && this.hub && this.app.engine.sessionId === this.sharedId) {
|
||||
this.hub.notify("hub/event", { sessionId: this.sharedId, event: e })
|
||||
if (e.type === "task" && e.state === "end") this.sendCommands()
|
||||
}
|
||||
}
|
||||
|
||||
private mine(p: any) {
|
||||
if (!this.sharedId || p?.sessionId !== this.sharedId) throw new RpcError(INVALID_PARAMS, "that session is not open in this terminal any more")
|
||||
}
|
||||
|
||||
/** What the web UI asks of the session, done here as if from the keyboard. */
|
||||
private wire(peer: Peer) {
|
||||
peer.handle("session/prompt", async (p): Promise<PromptResult> => {
|
||||
this.mine(p)
|
||||
const turnId = typeof p?.turnId === "string" && p.turnId ? p.turnId : crypto.randomUUID()
|
||||
// A turn id taken already: the original's answer.
|
||||
if (this.turns.has(turnId)) return this.turns.run(turnId, () => Promise.reject(new Error("unreachable")))
|
||||
// The blocks, expanded here as this terminal's keyboard would; `text` from a hub of
|
||||
// an earlier build.
|
||||
const blocks = Array.isArray(p?.prompt) ? p.prompt : [{ type: "text", text: String(p?.text ?? "") }]
|
||||
// The device's limits for a prompt from the web: an @path only inside this session's project
|
||||
// and the remote roots (where any are set).
|
||||
const limits = limitsFromConfig()
|
||||
const roots = limits.roots
|
||||
const prepared = preparePrompt(this.app, this.sharedId!, blocks, { roots: roots.length || !this.explicit ? roots : undefined })
|
||||
return this.turns.run(turnId, async (started) => {
|
||||
const done = (meta: Record<string, unknown>): PromptResult => ({ stopReason: "end_turn", _meta: { lembas: { turnId, ...meta } } })
|
||||
if (prepared.kind === "message") {
|
||||
started()
|
||||
this.hooks.notice(`/${prepared.name} (from the web UI): ${prepared.text}`)
|
||||
return done({ command: prepared.name, message: prepared.text })
|
||||
}
|
||||
if (prepared.kind === "action") {
|
||||
started()
|
||||
return done({ command: prepared.name, ...(await this.action(prepared.name, prepared.arg)) })
|
||||
}
|
||||
const { kind: _, atPaths: __, command, ...input } = prepared
|
||||
// A command's mode from the web never goes past remote.max_mode in a session shared under
|
||||
// the device's limits (not by name), as in a session the service runs.
|
||||
if (input.mode && !this.explicit) input.mode = clampMode(input.mode, limits.maxMode)
|
||||
const reason = await this.hooks.prompt({ ...input, turnId }, started)
|
||||
return { stopReason: STOP[reason] ?? "end_turn", _meta: { lembas: { turnId, ...(command ? { command } : {}) } } }
|
||||
})
|
||||
})
|
||||
// The terminal's turns: `_lembas/session/status` for a shared session asks here.
|
||||
peer.handle("hub/status", (p) => {
|
||||
this.mine(p)
|
||||
return { ...(this.turns.running ? { turnId: this.turns.running } : {}), queued: this.turns.queued(), ...(this.turns.last ? { lastTurnId: this.turns.last } : {}) }
|
||||
})
|
||||
peer.handle("hub/steer", (p) => {
|
||||
this.mine(p)
|
||||
const id = typeof p?.messageId === "string" && p.messageId ? p.messageId : undefined
|
||||
if (!this.app.turns.steer(String(p?.text ?? ""), id)) throw new RpcError(INVALID_PARAMS, "this session is not working on anything; send it as a prompt")
|
||||
return { queued: this.app.engine.queued }
|
||||
})
|
||||
peer.handle("hub/compact", async (p) => {
|
||||
this.mine(p)
|
||||
if (this.app.turns.current) throw new RpcError(INVALID_PARAMS, "this session is working — compact it when the reply is done")
|
||||
try {
|
||||
return { summary: await this.hooks.compact() }
|
||||
} catch (e) {
|
||||
throw new RpcError(INVALID_PARAMS, `compaction failed: ${(e as Error).message}`)
|
||||
}
|
||||
})
|
||||
peer.handle("hub/title", (p) => {
|
||||
this.mine(p)
|
||||
this.app.rename(String(p?.title ?? ""))
|
||||
return { title: String(p?.title ?? "").trim() }
|
||||
})
|
||||
// The web UI deleted the chat this session is: stopped, gone from the store, and the terminal
|
||||
// goes on in a new session (which is shared as any is).
|
||||
peer.handle("hub/delete", async (p) => {
|
||||
this.mine(p)
|
||||
const id = this.sharedId!
|
||||
this.app.cancel()
|
||||
// Stopped first, as the agent's own delete waits for its queue: until then the
|
||||
// cancelled turn could still be writing its last rows — into a session about to be gone, or,
|
||||
// once `deleted` below has moved the terminal on, into the new one. Bounded: a turn that does
|
||||
// not stop in time is deleted under it, as before.
|
||||
for (let i = 0; i < 100 && this.app.turns.current; i++) await Bun.sleep(100)
|
||||
// Moved on meanwhile (/new, /sessions): the session is no longer this terminal's to replace.
|
||||
const still = this.sharedId === id
|
||||
if (still) {
|
||||
this.sharedId = undefined
|
||||
this.explicit = false
|
||||
this.hooks.deleted()
|
||||
}
|
||||
dropAttachments(id)
|
||||
sessionTurnFile(id).drop()
|
||||
return { deleted: this.app.store?.deleteSession(id) ?? false }
|
||||
})
|
||||
peer.on("hub/cancel", (p) => {
|
||||
if (this.sharedId && p?.sessionId === this.sharedId) this.app.cancel()
|
||||
})
|
||||
peer.handle("hub/mode", (p) => {
|
||||
this.mine(p)
|
||||
const mode = asMode(p?.mode)
|
||||
if (!mode) throw new RpcError(INVALID_PARAMS, `unknown mode ${p?.mode}`)
|
||||
this.app.engine.mode = mode
|
||||
this.app.bus.emit({ type: "mode", mode })
|
||||
this.hooks.changed()
|
||||
return {}
|
||||
})
|
||||
peer.handle("hub/configure", (p) => {
|
||||
this.mine(p)
|
||||
// Compared as the ref the model has now: the instance's `<login>/<served>` names the
|
||||
// same model as `deepseek/x`, and switching to it again would reset the effort.
|
||||
if (typeof p?.model === "string" && p.model && this.app.canonicalRef(p.model) !== this.app.engine.model.ref) this.app.switchModel(p.model)
|
||||
const effort = effortOf(p?.effort)
|
||||
if (effort !== undefined) this.app.engine.effort = effort === "off" ? null : effort
|
||||
this.hooks.changed()
|
||||
return { model: this.app.engine.model.ref, effort: this.app.engine.effort ?? "off" }
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,202 @@
|
||||
// Shell integration for a terminal opened from the web UI: the shell marks where each
|
||||
// prompt, command and its output start and how the command ended (OSC 133 A/B/C/D), and says its
|
||||
// working directory after every command (OSC 7). The web UI's terminal reads those marks — the
|
||||
// directory in the panel's header, a green or red mark beside each finished command, jumping
|
||||
// between commands, copying the last one's output — and nothing on the server parses anything.
|
||||
//
|
||||
// The user's own startup files run first, exactly as for `$SHELL -l`, and the marks go in after
|
||||
// them; nothing else in their environment changes. bash gets an rc file in place of its login
|
||||
// files (which it sources), zsh a ZDOTDIR whose files source the user's own and hand ZDOTDIR back,
|
||||
// fish a command run after its config. Any other shell starts as it did before, without marks.
|
||||
// `remote.terminal_integration: false` turns it off.
|
||||
import { mkdirSync, rmSync, writeFileSync } from "node:fs"
|
||||
import { basename, join } from "node:path"
|
||||
import { paths } from "../config/paths.ts"
|
||||
|
||||
export type ShellKind = "bash" | "zsh" | "fish" | "other"
|
||||
|
||||
export function shellKind(shell: string): ShellKind {
|
||||
const b = basename(shell)
|
||||
return b === "bash" || b === "zsh" || b === "fish" ? b : "other"
|
||||
}
|
||||
|
||||
const BASH = `# LLeMbas CLI: shell integration for a terminal opened from the web UI. Generated; deleted when
|
||||
# the terminal closes.
|
||||
#
|
||||
# bash reads this in place of ~/.bashrc (--rcfile), after /etc/bash.bashrc, so it does what a login
|
||||
# shell would: your profile files, in a login shell's order, then logout and ~/.bash_logout as a
|
||||
# login shell has them. /etc/profile reads /etc/bash.bashrc again when PS1 is set (Debian's does),
|
||||
# so it is read with PS1 unset: once is enough.
|
||||
__lembas_ps1=$PS1
|
||||
unset PS1
|
||||
[ -r /etc/profile ] && . /etc/profile
|
||||
PS1=$__lembas_ps1
|
||||
unset __lembas_ps1
|
||||
if [ -r ~/.bash_profile ]; then . ~/.bash_profile
|
||||
elif [ -r ~/.bash_login ]; then . ~/.bash_login
|
||||
elif [ -r ~/.profile ]; then . ~/.profile
|
||||
fi
|
||||
logout() { exit "$@"; }
|
||||
[ -z "$(trap -p EXIT)" ] && trap '[ -r ~/.bash_logout ] && . ~/.bash_logout' EXIT
|
||||
|
||||
# Then the marks: D (how the last command ended), the directory and A before every prompt, B at
|
||||
# its end, C when a command starts to run. PS0 sets __lembas_ran in this shell (an array index is
|
||||
# evaluated here, not in a subshell), so an empty line ends no command. The path in OSC 7 is
|
||||
# percent-encoded byte by byte.
|
||||
__lembas_url() {
|
||||
local LC_ALL=C s=$1 out= c i
|
||||
for ((i = 0; i < \${#s}; i++)); do
|
||||
c=\${s:i:1}
|
||||
case $c in
|
||||
[A-Za-z0-9/._~-]) out+=$c ;;
|
||||
*) printf -v c '%%%02X' "'$c"; out+=$c ;;
|
||||
esac
|
||||
done
|
||||
printf '%s' "$out"
|
||||
}
|
||||
__lembas_prompt() {
|
||||
local s=$?
|
||||
[ -n "$__lembas_ran" ] && printf '\\e]133;D;%s\\a' "$s"
|
||||
__lembas_ran=
|
||||
printf '\\e]7;file://%s%s\\a' "\${HOSTNAME:-localhost}" "$(__lembas_url "$PWD")"
|
||||
printf '\\e]133;A\\a'
|
||||
return $s
|
||||
}
|
||||
# B at the end of the prompt, put back after every hook that builds PS1 afresh (starship, a
|
||||
# powerline prompt): this one runs last.
|
||||
__lembas_mark() {
|
||||
local s=$?
|
||||
case $PS1 in *'\\[\\e]133;B\\a\\]') ;; *) PS1="\${PS1}\\[\\e]133;B\\a\\]" ;; esac
|
||||
return $s
|
||||
}
|
||||
if [[ "$(declare -p PROMPT_COMMAND 2>/dev/null)" == "declare -a"* ]]; then
|
||||
PROMPT_COMMAND=(__lembas_prompt "\${PROMPT_COMMAND[@]}" __lembas_mark)
|
||||
else
|
||||
PROMPT_COMMAND="__lembas_prompt\${PROMPT_COMMAND:+; $PROMPT_COMMAND}; __lembas_mark"
|
||||
fi
|
||||
PS0="\${PS0}\\e]133;C\\a\\\${__lembas_nil[__lembas_ran=1]}"
|
||||
`
|
||||
|
||||
const ZSH_HOOKS = `# Then the marks: D (how the last command ended), the directory and A before every prompt, B at
|
||||
# its end, C when a command starts to run. The path in OSC 7 is percent-encoded byte by byte.
|
||||
__lembas_url() {
|
||||
emulate -L zsh
|
||||
local LC_ALL=C s=$1 out= c i
|
||||
for (( i = 1; i <= \${#s}; i++ )); do
|
||||
c=\${s[i]}
|
||||
if [[ $c == [A-Za-z0-9/._~-] ]]; then out+=$c; else out+=$(printf '%%%02X' "'$c"); fi
|
||||
done
|
||||
print -rn -- $out
|
||||
}
|
||||
__lembas_precmd() {
|
||||
local s=$?
|
||||
[[ -n $__lembas_ran ]] && printf '\\e]133;D;%s\\a' $s
|
||||
__lembas_ran=
|
||||
printf '\\e]7;file://%s%s\\a' "$HOST" "$(__lembas_url "$PWD")"
|
||||
printf '\\e]133;A\\a'
|
||||
return $s
|
||||
}
|
||||
# B at the end of the prompt, put back after any hook that builds PS1 afresh: this one runs last.
|
||||
__lembas_mark() {
|
||||
[[ $PS1 == *$'\\e]133;B\\a%}' ]] || PS1="$PS1%{"$'\\e]133;B\\a'"%}"
|
||||
}
|
||||
__lembas_preexec() { __lembas_ran=1; printf '\\e]133;C\\a' }
|
||||
precmd_functions=(__lembas_precmd $precmd_functions __lembas_mark)
|
||||
preexec_functions=($preexec_functions __lembas_preexec)
|
||||
`
|
||||
|
||||
/** zsh: our ZDOTDIR's .zshenv and .zprofile read the user's own, from the directory they use — the
|
||||
* one ZDOTDIR named when the terminal started, or HOME, or the one their own .zshenv sets (the
|
||||
* XDG way). Our .zshrc hands ZDOTDIR back and drops our variables *before* it reads theirs, so a
|
||||
* .zshrc that execs (tmux) leaves nothing of ours behind, and zsh then reads their .zlogin itself. */
|
||||
const ZSHENV = `# LLeMbas CLI: shell integration (generated; deleted when the terminal closes).
|
||||
__lembas_dir=$ZDOTDIR
|
||||
ZDOTDIR=\${LEMBAS_USER_ZDOTDIR:-$HOME}
|
||||
[[ -r $ZDOTDIR/.zshenv ]] && . $ZDOTDIR/.zshenv
|
||||
# Their .zshenv may have moved their files elsewhere: the rest are read from there.
|
||||
[[ $ZDOTDIR != \${LEMBAS_USER_ZDOTDIR:-$HOME} ]] && export LEMBAS_USER_ZDOTDIR_SET=1
|
||||
export LEMBAS_USER_ZDOTDIR=$ZDOTDIR
|
||||
ZDOTDIR=$__lembas_dir
|
||||
unset __lembas_dir
|
||||
`
|
||||
const ZPROFILE = `# LLeMbas CLI: shell integration (generated; deleted when the terminal closes).
|
||||
if [[ -r $LEMBAS_USER_ZDOTDIR/.zprofile ]]; then
|
||||
__lembas_dir=$ZDOTDIR
|
||||
ZDOTDIR=$LEMBAS_USER_ZDOTDIR
|
||||
. $ZDOTDIR/.zprofile
|
||||
ZDOTDIR=$__lembas_dir
|
||||
unset __lembas_dir
|
||||
fi
|
||||
`
|
||||
const ZSHRC = `# LLeMbas CLI: shell integration (generated; deleted when the terminal closes).
|
||||
__lembas_user=$LEMBAS_USER_ZDOTDIR
|
||||
if [[ -n $LEMBAS_USER_ZDOTDIR_SET ]]; then ZDOTDIR=$__lembas_user; else unset ZDOTDIR; fi
|
||||
unset LEMBAS_USER_ZDOTDIR LEMBAS_USER_ZDOTDIR_SET
|
||||
[[ -r $__lembas_user/.zshrc ]] && . $__lembas_user/.zshrc
|
||||
unset __lembas_user
|
||||
${ZSH_HOOKS}`
|
||||
|
||||
const FISH = `# LLeMbas CLI: shell integration (generated; deleted when the terminal closes). Run after your
|
||||
# own config: D (how the last command ended) after it, the directory and A before every prompt, B
|
||||
# at its end, C when a command starts.
|
||||
function __lembas_preexec --on-event fish_preexec
|
||||
printf '\\e]133;C\\a'
|
||||
end
|
||||
function __lembas_postexec --on-event fish_postexec
|
||||
printf '\\e]133;D;%s\\a' $status
|
||||
end
|
||||
function __lembas_prompt --on-event fish_prompt
|
||||
printf '\\e]7;file://%s%s\\a' (hostname) $PWD
|
||||
printf '\\e]133;A\\a'
|
||||
end
|
||||
if functions -q fish_prompt
|
||||
functions -c fish_prompt __lembas_user_prompt
|
||||
function fish_prompt
|
||||
__lembas_user_prompt
|
||||
printf '\\e]133;B\\a'
|
||||
end
|
||||
end
|
||||
`
|
||||
|
||||
export interface ShellLaunch {
|
||||
kind: ShellKind
|
||||
integration: boolean
|
||||
argv: string[]
|
||||
env: Record<string, string>
|
||||
/** Remove what was written for it. */
|
||||
cleanup(): void
|
||||
}
|
||||
|
||||
/** How to start `shell` for terminal `id`: with the marks when it is bash, zsh or fish and
|
||||
* `integration` is on; as `$SHELL -l` otherwise. */
|
||||
export function shellLaunch(shell: string, id: string, integration: boolean, env: NodeJS.ProcessEnv = process.env): ShellLaunch {
|
||||
const kind = shellKind(shell)
|
||||
const plain: ShellLaunch = { kind, integration: false, argv: [shell, "-l"], env: {}, cleanup: () => {} }
|
||||
if (!integration || kind === "other") return plain
|
||||
const dir = join(paths.state, "terminal", id.replace(/[^\w-]/g, "_"))
|
||||
const cleanup = () => rmSync(dir, { recursive: true, force: true })
|
||||
try {
|
||||
mkdirSync(dir, { recursive: true, mode: 0o700 })
|
||||
if (kind === "bash") {
|
||||
writeFileSync(join(dir, "bashrc"), BASH, { mode: 0o600 })
|
||||
return { kind, integration: true, argv: [shell, "--rcfile", join(dir, "bashrc"), "-i"], env: {}, cleanup }
|
||||
}
|
||||
if (kind === "zsh") {
|
||||
writeFileSync(join(dir, ".zshenv"), ZSHENV, { mode: 0o600 })
|
||||
writeFileSync(join(dir, ".zprofile"), ZPROFILE, { mode: 0o600 })
|
||||
writeFileSync(join(dir, ".zshrc"), ZSHRC, { mode: 0o600 })
|
||||
return {
|
||||
kind,
|
||||
integration: true,
|
||||
argv: [shell, "-l"],
|
||||
env: { ZDOTDIR: dir, LEMBAS_USER_ZDOTDIR: env.ZDOTDIR || env.HOME || "", ...(env.ZDOTDIR ? { LEMBAS_USER_ZDOTDIR_SET: "1" } : {}) },
|
||||
cleanup,
|
||||
}
|
||||
}
|
||||
writeFileSync(join(dir, "integration.fish"), FISH, { mode: 0o600 })
|
||||
return { kind, integration: true, argv: [shell, "-l", "-C", `source '${join(dir, "integration.fish").replace(/'/g, "\\'")}'`], env: {}, cleanup }
|
||||
} catch {
|
||||
cleanup()
|
||||
return plain
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,206 @@
|
||||
// Turn ids: the web UI's prompt carries its own message id, and a prompt with an id this
|
||||
// session has already taken does not run twice. That closes the two races of reattaching after a
|
||||
// dropped link — the instance not knowing whether its prompt arrived, and a terminal's turn matched
|
||||
// to the wrong row by timing: the instance asks (`_lembas/session/status`) and, when in doubt,
|
||||
// simply sends the prompt again.
|
||||
//
|
||||
// Per session, the last 200 ids, each queued, running or done; a repeat of a queued or running one
|
||||
// waits for the original and answers what it answers, a repeat of a finished one answers at once.
|
||||
// The finished ones are kept in one file per session, which the service and a terminal both read
|
||||
// and write — whichever holds the session — so a turn run in a terminal is not run again by the
|
||||
// service after the terminal lets go of it, nor the other way round, nor after a restart.
|
||||
|
||||
import { mkdirSync, readdirSync, readFileSync, renameSync, rmSync, statSync, writeFileSync } from "node:fs"
|
||||
import { dirname, join } from "node:path"
|
||||
import { paths } from "../config/paths.ts"
|
||||
|
||||
export type TurnState = "queued" | "running" | "done"
|
||||
|
||||
interface Entry<R> {
|
||||
state: TurnState
|
||||
result: Promise<R>
|
||||
}
|
||||
|
||||
export const TURNS_KEPT = 200
|
||||
/** A session's file not written for this long goes, the next time a process opens turn files. */
|
||||
const TURN_FILE_DAYS = 30
|
||||
|
||||
/** Where a session's finished turns are kept: `[id, answer | null]` in order, null for a turn this
|
||||
* side did not answer (one typed in a terminal). */
|
||||
export interface TurnStore<R> {
|
||||
load(): [string, R | null][]
|
||||
save(done: [string, R | null][]): void
|
||||
}
|
||||
|
||||
/** The file's entries, those that are entries: a damaged file is an empty one, never a crash. */
|
||||
function readEntries<R>(file: string): [string, R | null][] {
|
||||
try {
|
||||
const v = JSON.parse(readFileSync(file, "utf8"))
|
||||
if (!Array.isArray(v)) return []
|
||||
return v.filter((e): e is [string, R | null] => Array.isArray(e) && e.length === 2 && typeof e[0] === "string" && e[0].length > 0)
|
||||
} catch {
|
||||
return []
|
||||
}
|
||||
}
|
||||
|
||||
/** A JSON file. Written whole to a temporary file and renamed over it, so a reader in the other
|
||||
* process never sees half of one; what that process wrote meanwhile is merged in first. */
|
||||
export function turnFile<R>(file: string): TurnStore<R> & { drop(): void } {
|
||||
return {
|
||||
load: () => readEntries<R>(file),
|
||||
save(done) {
|
||||
try {
|
||||
const merged = new Map<string, R | null>(readEntries<R>(file))
|
||||
for (const [id, v] of done) {
|
||||
// Ours last, so they stay the newest; an answer never replaced by "none known".
|
||||
const had = merged.get(id)
|
||||
merged.delete(id)
|
||||
merged.set(id, v ?? had ?? null)
|
||||
}
|
||||
const kept = [...merged].slice(-TURNS_KEPT)
|
||||
mkdirSync(dirname(file), { recursive: true, mode: 0o700 })
|
||||
const tmp = `${file}.${process.pid}.${Date.now()}.tmp`
|
||||
writeFileSync(tmp, JSON.stringify(kept), { mode: 0o600 })
|
||||
renameSync(tmp, file)
|
||||
} catch {}
|
||||
},
|
||||
drop() {
|
||||
rmSync(file, { force: true })
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
const turnsDir = () => join(paths.state, "turns")
|
||||
let pruned = false
|
||||
|
||||
/** Files of sessions nobody has worked in for a month go (a deleted session's goes at once). */
|
||||
export function pruneTurnFiles(dir = turnsDir(), maxAgeDays = TURN_FILE_DAYS) {
|
||||
let names: string[]
|
||||
try {
|
||||
names = readdirSync(dir)
|
||||
} catch {
|
||||
return
|
||||
}
|
||||
const cutoff = Date.now() - maxAgeDays * 86_400_000
|
||||
for (const n of names) {
|
||||
const f = join(dir, n)
|
||||
try {
|
||||
if (statSync(f).mtimeMs < cutoff || n.endsWith(".tmp")) rmSync(f, { force: true })
|
||||
} catch {}
|
||||
}
|
||||
}
|
||||
|
||||
/** The file of one session's turns. */
|
||||
export function sessionTurnFile<R>(sessionId: string) {
|
||||
return turnFile<R>(join(turnsDir(), `${sessionId.replace(/[^\w.-]/g, "_")}.json`))
|
||||
}
|
||||
|
||||
/** One session's turn log, on its file — the one the service and every terminal use. */
|
||||
export function sessionTurns<R>(sessionId: string): TurnLog<R> {
|
||||
if (!pruned) {
|
||||
pruned = true
|
||||
pruneTurnFiles()
|
||||
}
|
||||
return new TurnLog<R>(sessionTurnFile<R>(sessionId))
|
||||
}
|
||||
|
||||
export class TurnLog<R = unknown> {
|
||||
private entries = new Map<string, Entry<R>>()
|
||||
private values = new Map<string, R | null>()
|
||||
/** The turn running now, and the last one that finished. */
|
||||
running?: string
|
||||
last?: string
|
||||
|
||||
constructor(private store?: TurnStore<R>) {
|
||||
this.sync()
|
||||
}
|
||||
|
||||
/** What the file says now (another process may have written it), for ids not known here. */
|
||||
private sync() {
|
||||
for (const [id, value] of this.store?.load() ?? []) {
|
||||
if (this.entries.has(id)) continue
|
||||
this.entries.set(id, { state: "done", result: Promise.resolve(value as R) })
|
||||
this.values.set(id, value)
|
||||
if (!this.running) this.last = id
|
||||
}
|
||||
}
|
||||
|
||||
has(id: string): boolean {
|
||||
if (!this.entries.has(id)) this.sync()
|
||||
return this.entries.has(id)
|
||||
}
|
||||
|
||||
state(id: string): TurnState | undefined {
|
||||
return this.entries.get(id)?.state
|
||||
}
|
||||
|
||||
/** The ids waiting their turn, oldest first. */
|
||||
queued(): string[] {
|
||||
return [...this.entries].filter(([, e]) => e.state === "queued").map(([id]) => id)
|
||||
}
|
||||
|
||||
/** Run `fn` as turn `id` — or, for an id already taken, answer what that turn answers. `fn` gets
|
||||
* `started`, to call when the turn stops waiting and runs. */
|
||||
run(id: string, fn: (started: () => void) => Promise<R>): Promise<R> {
|
||||
if (!this.entries.has(id)) this.sync()
|
||||
const known = this.entries.get(id)
|
||||
if (known) return known.result
|
||||
const entry: Entry<R> = { state: "queued", result: undefined as unknown as Promise<R> }
|
||||
this.entries.set(id, entry)
|
||||
this.trim()
|
||||
entry.result = fn(() => {
|
||||
entry.state = "running"
|
||||
this.running = id
|
||||
})
|
||||
.then((value) => {
|
||||
this.values.set(id, value)
|
||||
return value
|
||||
})
|
||||
.finally(() => {
|
||||
entry.state = "done"
|
||||
if (this.running === id) this.running = undefined
|
||||
this.last = id
|
||||
this.persist()
|
||||
})
|
||||
// A failed turn is answered as failed to a repeat too, and is not an unhandled rejection here.
|
||||
entry.result.catch(() => {})
|
||||
return entry.result
|
||||
}
|
||||
|
||||
/** A turn this log did not start (one typed in a terminal), followed by its events. */
|
||||
started(id: string) {
|
||||
const e = this.entries.get(id)
|
||||
if (e) e.state = "running"
|
||||
else {
|
||||
this.entries.set(id, { state: "running", result: Promise.resolve(undefined as R) })
|
||||
this.trim()
|
||||
}
|
||||
this.running = id
|
||||
}
|
||||
|
||||
ended(id: string) {
|
||||
const e = this.entries.get(id)
|
||||
if (e) e.state = "done"
|
||||
if (this.running === id) this.running = undefined
|
||||
this.last = id
|
||||
this.persist()
|
||||
}
|
||||
|
||||
/** The oldest finished ones go first; one still queued or running is never forgotten. */
|
||||
private trim() {
|
||||
if (this.entries.size <= TURNS_KEPT) return
|
||||
for (const [id, e] of this.entries) {
|
||||
if (this.entries.size <= TURNS_KEPT) break
|
||||
if (e.state === "done") {
|
||||
this.entries.delete(id)
|
||||
this.values.delete(id)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** The finished turns, to the store: with their answers where this side gave one. */
|
||||
private persist() {
|
||||
if (!this.store) return
|
||||
this.store.save([...this.entries].filter(([, e]) => e.state === "done").map(([id]) => [id, this.values.get(id) ?? null]))
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user