Reserve Boat SSH capacity for lifecycle and process control

Bound each physical Boat to eight SSH operations with six ordinary slots and four bulk transfers. Preserve command deadlines while queued and always release admission after transport completion or failure.

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",