Integrate bounded Boat SSH admission with cleanup capacity

Co-Authored-By: Paperclip <noreply@paperclip.ing>

server/src/modules/computers/adapters/boat.test.ts

@@ -16,7 +16,7 @@ import { join } from "node:path";
 import { spawnSync } from "node:child_process";
 import { remoteProgram } from "./remote-program.js";
 import { afterEach, describe, it, expect, vi } from "vitest";
-import { boatBackend, desktopReadinessProgram, processProgram } from "./boat.js";
+import { boatBackend, createBoatTransportAdmission, desktopReadinessProgram, processProgram } from "./boat.js";
 import { buildGitAuthInvocation } from "../../../services/git-credentials.js";
 import type { ComputerRecord } from "../domain/ledger.js";
 const record: ComputerRecord = {
@@ -480,3 +480,55 @@ describe("Boat desktop input readiness", () => {
     expect(existsSync(f.healthy)).toBe(true);
   });
 });
+
+
+describe("Boat SSH admission", () => {
+  const result = { exitCode: 0, signal: null, timedOut: false, stdout: "", stderr: "", pid: null, startedAt: "" };
+  it("reserves lifecycle and identity capacity while bulk and long legacy commands are active", async () => {
+    const admission = createBoatTransportAdmission();
+    const completions: Array<() => void> = [];
+    const runner = { execute: vi.fn(async () => new Promise<typeof result>((resolve) => completions.push(() => resolve(result)))) };
+    const input = { command: "legacy", timeoutMs: 60_000 };
+    const jobs = Array.from({ length: 8 }, () => admission(runner, { ...input, stdin: "x".repeat(1024 * 1024 + 1) }, true));
+    jobs.push(admission(runner, input, false), admission(runner, input, false));
+    await Promise.resolve();
+    expect(runner.execute).toHaveBeenCalledTimes(6);
+    jobs.push(admission(runner, { command: "retire", timeoutMs: 10_000 }, true));
+    jobs.push(admission(runner, { command: "identity", timeoutMs: 10_000 }, true));
+    await Promise.resolve();
+    expect(runner.execute).toHaveBeenCalledTimes(8);
+    while (completions.length) {
+      completions.splice(0).forEach((finish) => finish());
+      await new Promise((resolve) => setTimeout(resolve, 0));
+    }
+    await Promise.all(jobs);
+    expect(runner.execute).toHaveBeenCalledTimes(12);
+  });
+  it("expires queued work without starting it and preserves the original remaining timeout", async () => {
+    vi.useFakeTimers();
+    try {
+      const admission = createBoatTransportAdmission();
+      const completions: Array<() => void> = [];
+      const runner = { execute: vi.fn(async (_input: { command: string; timeoutMs?: number }) => new Promise<typeof result>((resolve) => completions.push(() => resolve(result)))) };
+      const jobs = Array.from({ length: 6 }, () => admission(runner, { command: "long", timeoutMs: 60_000 }, false));
+      await vi.advanceTimersByTimeAsync(0);
+      const expired = admission(runner, { command: "expired", timeoutMs: 100 }, false);
+      const queued = admission(runner, { command: "queued", timeoutMs: 1000 }, false);
+      await vi.advanceTimersByTimeAsync(200);
+      expect(await expired).toMatchObject({ timedOut: true, pid: null });
+      expect(runner.execute).toHaveBeenCalledTimes(6);
+      completions.shift()!();
+      await vi.advanceTimersByTimeAsync(0);
+      expect(runner.execute).toHaveBeenLastCalledWith(expect.objectContaining({ command: "queued", timeoutMs: 800 }));
+      completions.splice(0).forEach((finish) => finish());
+      await Promise.all([...jobs, queued]);
+    } finally { vi.useRealTimers(); }
+  });
+  it("releases all capacity after failed or cancelled transport commands without replay", async () => {
+    const admission = createBoatTransportAdmission();
+    const runner = { execute: vi.fn(async () => { throw new Error("transport cancelled"); }) };
+    const settled = await Promise.allSettled(Array.from({ length: 12 }, () => admission(runner, { command: "once", timeoutMs: 1000 }, false)));
+    expect(settled.every((entry) => entry.status === "rejected")).toBe(true);
+    expect(runner.execute).toHaveBeenCalledTimes(12);
+  });
+});

server/src/modules/computers/adapters/boat.ts

@@ -15,6 +15,66 @@ import { remoteProgram } from "./remote-program.js";
 
 const apiOrigin = "https://boat.dev/api/v1";
 const transportCache = new Map<string, Promise<CommandManagedRuntimeRunner>>();
+type CommandInput = Parameters<CommandManagedRuntimeRunner["execute"]>[0];
+
+/** Reserve SSH admission for lifecycle/identity calls while bulk and legacy commands run. */
+export function createBoatTransportAdmission() {
+  let active = 0;
+  let ordinary = 0;
+  let bulk = 0;
+  const queue: Array<{ control: boolean; bulk: boolean; start: () => void }> = [];
+  function drain() {
+    while (active < 8) {
+      const eligible = (entry: (typeof queue)[number]) =>
+        entry.control || (ordinary < 6 && (!entry.bulk || bulk < 4));
+      let index = queue.findIndex((entry) => entry.control && eligible(entry));
+      if (index < 0) index = queue.findIndex(eligible);
+      if (index < 0) break;
+      queue.splice(index, 1)[0]!.start();
+    }
+  }
+  return async function execute(
+    runner: CommandManagedRuntimeRunner,
+    input: CommandInput,
+    control: boolean,
+  ) {
+    const startedAt = new Date().toISOString();
+    const deadline = Date.now() + (input.timeoutMs ?? 15_000);
+    const isBulk = Buffer.byteLength(input.stdin ?? "") > 1024 * 1024;
+    // Even process-control callers must use the ordinary budget for large uploads.
+    const isControl = control && !isBulk;
+    if (queue.length >= 128)
+      throw new ComputerError("conflict", "Computer command admission is busy");
+    return new Promise<Awaited<ReturnType<CommandManagedRuntimeRunner["execute"]>>>((resolve, reject) => {
+      const timedOut = () => resolve({ exitCode: null, signal: null, timedOut: true,
+        stdout: "", stderr: "Computer command timed out waiting for SSH admission", pid: null, startedAt });
+      const entry = { control: isControl, bulk: isBulk, start: () => {
+        clearTimeout(timer);
+        const remaining = deadline - Date.now();
+        if (remaining <= 0) { timedOut(); return; }
+        active++;
+        if (!isControl) ordinary++;
+        if (isBulk) bulk++;
+        void Promise.resolve().then(() => runner.execute({ ...input, timeoutMs: remaining }))
+          .then(resolve, reject).finally(() => {
+            active--;
+            if (!isControl) ordinary--;
+            if (isBulk) bulk--;
+            drain();
+          });
+      } };
+      const timer = setTimeout(() => {
+        const index = queue.indexOf(entry);
+        if (index >= 0) queue.splice(index, 1);
+        timedOut();
+        drain();
+      }, Math.max(1, deadline - Date.now()));
+      queue.push(entry);
+      drain();
+    });
+  };
+}
+const transportAdmission = new Map<string, ReturnType<typeof createBoatTransportAdmission>>();
 const desktopCache = new Map<
   string,
   { viewerUrl: string; expiresAt: string }
@@ -166,7 +226,7 @@ export function boatBackend(
       stop: sandbox.stop ?? null,
     };
   }
-  async function runner(
+  async function rawRunner(
     record: ComputerRecord,
   ): Promise<CommandManagedRuntimeRunner> {
     const key = `${record.id}:${record.ledger.secretRef.secretId}:${record.ledger.secretRef.version ?? "latest"}`;
@@ -263,13 +323,23 @@ export function boatBackend(
     }
     return pending;
   }
+  async function runner(record: ComputerRecord, options?: { control?: boolean }): Promise<CommandManagedRuntimeRunner> {
+    const raw = await rawRunner(record);
+    let admission = transportAdmission.get(record.providerId);
+    if (!admission) {
+      admission = createBoatTransportAdmission();
+      transportAdmission.set(record.providerId, admission);
+    }
+    const execute = admission;
+    return { ...raw, execute: (input) => execute(raw, input, options?.control === true) };
+  }
   async function execute(
     record: ComputerRecord,
     code: string,
     payload: unknown,
   ) {
     const result = await (
-      await runner(record)
+      await runner(record, { control: true })
     ).execute({
       command: "python3",
       args: ["-c", code],

server/src/modules/computers/application/ports.ts

@@ -31,7 +31,7 @@ export interface ComputerBackend {
   ready(record: ComputerRecord): Promise<void>;
   claim(record: ComputerRecord): Promise<void>;
   advance(record: ComputerRecord, owner: Owner): Promise<void>;
-  runner(record: ComputerRecord): Promise<CommandManagedRuntimeRunner>;
+  runner(record: ComputerRecord, options?: { control?: boolean }): Promise<CommandManagedRuntimeRunner>;
   launch(
     record: ComputerRecord,
     owner: Owner,

server/src/modules/computers/application/service.ts

@@ -170,6 +170,7 @@ export function createComputerService(
     if (!placement)
       throw new ComputerError("conflict", "Computer placement is missing");
     const raw = await backend.runner(record);
+    const control = await backend.runner(record, { control: true });
     let pinnedProcess = owner.process ? structuredClone(owner.process) : null;
     async function processScope() {
       if (!pinnedProcess) return scoped({ ...record, owner: ownerRef });
@@ -203,7 +204,7 @@ export function createComputerService(
           throw new ComputerError("conflict", "Computer owner is retired");
         // Start a command unit under the same owner slice while holding the remote tombstone lock.
         // The lock is released only after systemd knows the unit, so retirement cannot miss a delayed launch.
-        return raw.execute({
+        return (processScoped ? control : raw).execute({
           command: "python3",
           args: [
             "-c",