diff --git a/packages/redrob/src/cli/cmd/acp.ts b/packages/redrob/src/cli/cmd/acp.ts index bdb54a0ab8..9a6b81ee74 100644 --- a/packages/redrob/src/cli/cmd/acp.ts +++ b/packages/redrob/src/cli/cmd/acp.ts @@ -29,11 +29,15 @@ export const AcpCommand = effectCmd({ ACPProfile.mark("cli.acp.handler") process.env.REDROB_CLIENT = "acp" // --attach: this process is only the ACP face; the engine is the shared daemon. Without it the - // engine runs in here, as it always has. + // engine runs in here, as it always has. The attachment is what keeps the daemon from counting + // this session as idle time; it is released when the host closes stdin. + let detach: (() => Promise) | undefined const sdk = args.attach ? yield* Effect.promise(async () => { const { Daemon } = await import("@/daemon/connect") const connection = await Daemon.connectOrStart() + const attachment = await Daemon.attach(connection, process.env.REDROB_APP?.trim() || "acp") + detach = attachment.detach return createRedrobClient({ baseUrl: connection.url, headers: connection.headers }) }) : yield* Effect.gen(function* () { @@ -87,5 +91,6 @@ export const AcpCommand = effectCmd({ process.stdin.on("error", reject) }), ) + if (detach) yield* Effect.promise(() => detach!()) }), }) diff --git a/packages/redrob/src/daemon/connect.ts b/packages/redrob/src/daemon/connect.ts index c8140bb5c4..741005ca09 100644 --- a/packages/redrob/src/daemon/connect.ts +++ b/packages/redrob/src/daemon/connect.ts @@ -104,4 +104,62 @@ export async function connectOrStart(options: StartOptions = {}): Promise Promise +} + +/** + * Tell the daemon this process depends on it, and keep saying so: `POST /api/clients`, then a + * heartbeat every third of the daemon's TTL. The daemon exits once nobody has been attached for + * its idle window, so `redrob acp --attach` must hold one for as long as its host keeps it open -- + * connecting alone leaves the daemon counting idle time under a live session. Mirrors + * `attach()` in `@redrob-labs/sdk/daemon`. + */ +export async function attach( + connection: Connection, + app: string, + options: { readonly pid?: number; readonly fetch?: typeof fetch } = {}, +): Promise { + const fetchImpl = options.fetch ?? fetch + const base = connection.url.replace(/\/+$/, "") + const headers = { ...connection.headers, "content-type": "application/json" } + const body = JSON.stringify({ app, pid: options.pid ?? process.pid }) + const response = await fetchImpl(`${base}/api/clients`, { method: "POST", headers, body }) + if (!response.ok) throw new Error(`The engine refused the attachment (${response.status})`) + const client = (await response.json()) as { id: string } + const list = (await fetchImpl(`${base}/api/clients`, { headers: connection.headers }) + .then((r) => (r.ok ? r.json() : null)) + .catch(() => null)) as { ttlMs?: number } | null + const every = Math.max(5_000, Math.floor((list?.ttlMs ?? 90_000) / 3)) + let stopped = false + const beat = async () => { + if (stopped) return + const r = await fetchImpl(`${base}/api/clients/${client.id}/heartbeat`, { method: "POST", headers }).catch( + () => undefined, + ) + // 404: the daemon restarted (or dropped us); attach again so it keeps counting this app. + if (r && r.status === 404) { + const again = await fetchImpl(`${base}/api/clients`, { method: "POST", headers, body }).catch(() => undefined) + if (again?.ok) client.id = ((await again.json()) as { id: string }).id + } + } + const timer = setInterval(() => void beat(), every) + timer.unref?.() + return { + get id() { + return client.id + }, + detach: async () => { + if (stopped) return + stopped = true + clearInterval(timer) + await fetchImpl(`${base}/api/clients/${client.id}`, { method: "DELETE", headers: connection.headers }).catch( + () => undefined, + ) + }, + } +} + export * as Daemon from "./connect" diff --git a/packages/redrob/test/daemon.test.ts b/packages/redrob/test/daemon.test.ts index cfc7b058ec..f5d24083ae 100644 --- a/packages/redrob/test/daemon.test.ts +++ b/packages/redrob/test/daemon.test.ts @@ -7,6 +7,7 @@ import { afterEach, describe, expect, test } from "bun:test" import fs from "node:fs/promises" import os from "node:os" import path from "node:path" +import { Daemon } from "@/daemon/connect" import * as Registry from "@/daemon/registry" import * as Clients from "@redrob-code/server/clients" @@ -114,3 +115,37 @@ describe("attached apps", () => { expect(Clients.attach({ app: " x".repeat(80) }).app.length).toBe(64) }) }) + +describe("attach", () => { + test("posts the app, heartbeats, re-attaches after a 404 and deletes on detach", async () => { + const calls: string[] = [] + let ids = 0 + const fetchImpl = (async (input: string | URL, init?: RequestInit) => { + const url = String(input) + const method = init?.method ?? "GET" + calls.push(`${method} ${new URL(url).pathname}`) + if (method === "POST" && url.endsWith("/api/clients")) { + expect(JSON.parse(String(init?.body)).app).toBe("cut") + return Response.json({ id: `c${++ids}` }) + } + if (method === "GET") return Response.json({ clients: [], ttlMs: 15_000 }) + if (url.endsWith("/heartbeat")) return new Response("", { status: ids === 1 ? 404 : 200 }) + return new Response("", { status: 204 }) + }) as unknown as typeof fetch + const connection = { url: "http://127.0.0.1:4096/", headers: Registry.headers("s"), record: record() } + const a = await Daemon.attach(connection, "cut", { pid: 7, fetch: fetchImpl }) + expect(a.id).toBe("c1") + await new Promise((r) => setTimeout(r, 5_300)) + // The first heartbeat met a 404 (daemon restarted), so the attachment was re-created. + expect(a.id).toBe("c2") + await a.detach() + await a.detach() + expect(calls).toEqual([ + "POST /api/clients", + "GET /api/clients", + "POST /api/clients/c1/heartbeat", + "POST /api/clients", + "DELETE /api/clients/c2", + ]) + }, 10_000) +})