Merge branch 'codex/boat-environment-core' into codex/boat-environment

* codex/boat-environment-core:
  fix: preserve authoritative warm computers during orphan lease sweep
  fix: share Boat upload capacity before command admission

server/src/services/heartbeat/recovery.ts

@@ -1,3 +1,4 @@
+import { COMPUTER_RETIREMENT_GRACE_MS } from "../../modules/computers/domain/ledger.js";
 import { preserveWorkspaceRestoreRecoveryMetadataSql } from "../legacy-workspace-restore-recovery.js";
 import { pendingNativeWorkspaceFinalizationCondition } from "../native-runtime/native-workspace-finalization-state.js";
 import { CHAT_COMPLETION_WAKE_REASON } from "../chat-completion-delivery.js";
@@ -30,6 +31,7 @@ import {
 import type { Db } from "@paperclipai/db";
 import {
   agents,
+  computers,
   environmentLeases,
   heartbeatRuns,
   issues,
@@ -1413,7 +1415,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
     const cutoff = new Date(Date.now() - opts.backoffMs);
 
     const rows = await db
-      .select({ lease: environmentLeases })
+      .select({ lease: environmentLeases, runAgentId: heartbeatRuns.agentId })
       .from(environmentLeases)
       .leftJoin(
         heartbeatRuns,
@@ -1435,7 +1437,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
       .limit(ORPHANED_ACTIVE_LEASE_SWEEP_PAGE_SIZE);
 
     let recovered = 0;
-    for (const { lease } of rows) {
+    for (const { lease, runAgentId } of rows) {
       // A provider resource id names one physical sandbox. A different lease
       // row can still hold that same resource in a live status, so this sweep
       // must not tear down a sandbox that a different lease still owns.
@@ -1470,19 +1472,59 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
       // pending_cleanup sweep can accept the same row in this same tick. A
       // fresh timestamp here would push the row inside that sweep's own
       // backoff window and delay the teardown by one full tick.
-      const flipped = await db
+      const sameLease = and(
+          eq(environmentLeases.id, lease.id),
+          eq(environmentLeases.status, "active"),
+          // A completed heartbeat may retain or replace its bookkeeping while
+          // this sweep waits for the computer serialization boundary.
+          eq(environmentLeases.companyId, lease.companyId),
+          lease.environmentId ? eq(environmentLeases.environmentId, lease.environmentId) : isNull(environmentLeases.environmentId),
+          lease.heartbeatRunId ? eq(environmentLeases.heartbeatRunId, lease.heartbeatRunId) : isNull(environmentLeases.heartbeatRunId),
+          lease.metadata === null ? isNull(environmentLeases.metadata) : eq(environmentLeases.metadata, lease.metadata),
+          lease.provider === null ? isNull(environmentLeases.provider) : eq(environmentLeases.provider, lease.provider),
+          lease.providerLeaseId === null ? isNull(environmentLeases.providerLeaseId) : eq(environmentLeases.providerLeaseId, lease.providerLeaseId),
+        );
+      const claim = (query: Pick<Db, "update">) => query
         .update(environmentLeases)
-        .set({
-          status: "pending_cleanup",
-          failureReason: "orphaned_active_lease_recovered",
-        })
-        .where(
-          and(
-            eq(environmentLeases.id, lease.id),
-            eq(environmentLeases.status, "active"),
-          ),
-        )
+        .set({ status: "pending_cleanup", failureReason: "orphaned_active_lease_recovered" })
+        .where(sameLease)
         .returning({ id: environmentLeases.id });
+      const binding = parseObject(lease.metadata?.computerOwner);
+      const flipped = lease.provider === "boat" && lease.metadata?.driver === "computer"
+        ? await db.transaction(async (tx) => {
+          const [computer] = typeof binding.computerId === "string" && lease.environmentId
+            ? await tx.select().from(computers).where(and(
+              eq(sql`${computers.id}::text`, binding.computerId),
+              eq(computers.companyId, lease.companyId),
+              eq(computers.environmentId, lease.environmentId),
+              eq(computers.provider, "boat"),
+            )).for("update") : [];
+          const owner = Array.isArray(computer?.ledger.owners)
+            ? computer.ledger.owners.map(parseObject).find((value) =>
+              value.id === binding.ownerId && value.id === lease.providerLeaseId &&
+              Number.isSafeInteger(binding.generation) && Number(binding.generation) > 0 &&
+              value.generation === binding.generation && value.kind === "runner" &&
+              value.runId === lease.heartbeatRunId && runAgentId !== null &&
+              value.agentId === runAgentId && value.agentId === lease.metadata?.agentId)
+            : undefined;
+          const process = parseObject(owner?.process);
+          const processBound = [process.bootId, process.unitName, process.nonce]
+            .every((value) => typeof value === "string" && value.length > 0) &&
+            Number.isSafeInteger(process.launchGeneration) && Number(process.launchGeneration) > 0;
+          const deadline = typeof owner?.deadline === "string" ? Date.parse(owner.deadline) : NaN;
+          if (owner?.phase === "warm" && processBound && Number.isFinite(deadline) &&
+              deadline + COMPUTER_RETIREMENT_GRACE_MS > Date.now()) {
+            // Native completion retains the durable owner before heartbeat's
+            // later instruction cleanup can mark the lease retained. A terminal
+            // run is not permission to kill this bounded warm process. Keep it
+            // out of the page until a later sweep; expired/mismatched owners
+            // still follow the ordinary orphan cleanup path.
+            await tx.update(environmentLeases).set({ updatedAt: new Date() })
+              .where(sameLease);
+            return [];
+          }
+          return claim(tx);
+        }) : await claim(db);
       if (flipped.length > 0) recovered += 1;
     }
 

server/src/services/native-runtime/native-session-executor.test.ts

@@ -1,3 +1,4 @@
+import { createBoatTransportAdmission } from "../../modules/computers/adapters/boat.js";
 import { closeWarmNativeSession, compareAndSealRetiredComputerCheckpoint, retiredComputerCheckpointIsSettled, retiredComputerCheckpointUnsettledReason, retireUnlaunchedComputerAttempt, nativeSessionBootstrapTimeoutMs } from "./native-session-executor.js";
 import { prepareHeartbeatGitHubLaunchers } from "../heartbeat-github-launchers.js";
 import { RemoteProviderPackVerificationError } from "./remote-provider-pack-cache.js";
@@ -2717,6 +2718,62 @@ describe("remote provider checkpoint restores", () => {
     } finally { await rm(root, { recursive: true, force: true }); }
   });
 
+  it("shares Boat upload capacity across concurrent packs before command deadlines begin", async () => {
+    const root = await mkdtemp(join(tmpdir(), "paperclip-boat-upload-pressure-"));
+    const sourcePath = join(root, "source");
+    const bytes = randomBytes(24 * 1024 * 1024);
+    await writeFile(sourcePath, bytes);
+    const admission = createBoatTransportAdmission();
+    const result = { exitCode: 0, signal: null, timedOut: false, stdout: "", stderr: "", pid: null, startedAt: "" };
+    let active = 0;
+    let peak = 0;
+    let entered!: () => void;
+    const firstBatch = new Promise<void>(resolve => { entered = resolve; });
+    let wake: (() => void) | undefined;
+    const budgets: number[] = [];
+    const raw = { execute: vi.fn(async (request: { command: string; args?: string[]; stdin?: string; timeoutMs?: number }) => {
+      if (!request.stdin) {
+        execFileSync(request.command, request.args ?? []);
+        return result;
+      }
+      active++;
+      peak = Math.max(peak, active);
+      if (active === 4) entered();
+      wake?.(); wake = undefined;
+      budgets.push(request.timeoutMs!);
+      try {
+        await new Promise(resolve => setTimeout(resolve, Math.min(100_000, request.timeoutMs!)));
+        if (request.timeoutMs! < 100_000) return { ...result, exitCode: null, timedOut: true };
+        execFileSync(request.command, request.args ?? [], { input: request.stdin });
+        return result;
+      } finally { active--; }
+    }) };
+    const runner = {
+      withStdinUploadSlot: admission.withStdinUploadSlot,
+      execute: (request: Parameters<typeof admission>[1]) => admission(raw, request, true),
+    };
+    vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] });
+    try {
+      let settled = false;
+      const transfer = Promise.allSettled(["one", "two"].map(name => stageRemoteRunnerFile({
+        target: { kind: "remote", transport: "computer" } as never,
+        runner: runner as never, sourcePath, targetPath: join(root, name), mode: 0o700,
+      }))).then(results => { settled = true; wake?.(); wake = undefined; return results; });
+      await firstBatch;
+      await expect(admission(raw, { command: "true", timeoutMs: 1000 }, true)).resolves.toMatchObject({ exitCode: 0 });
+      while (!settled) {
+        if (!active) await new Promise<void>(resolve => { wake = resolve; });
+        if (!settled) await vi.advanceTimersByTimeAsync(100_000);
+      }
+      const results = await transfer;
+      expect(results.map(result => result.status)).toEqual(["fulfilled", "fulfilled"]);
+      expect(peak).toBe(4);
+      expect(budgets).toHaveLength(12);
+      expect(budgets.every(budget => budget > 99_999)).toBe(true);
+      for (const name of ["one", "two"]) expect((await readFile(join(root, name))).equals(bytes)).toBe(true);
+    } finally { vi.useRealTimers(); await rm(root, { recursive: true, force: true }); }
+  });
+
   it("shares eight upload slots across concurrent artifact transfers", async () => {
     const root = await mkdtemp(join(tmpdir(), "paperclip-runner-upload-slots-"));
     const sourcePath = join(root, "source");

server/src/services/native-runtime/native-session-executor.ts

@@ -10545,7 +10545,7 @@ export async function stageRemoteRunnerFile(input: {
   }
   const remoteUpload = `${input.targetPath}.upload-${randomUUID()}`;
   try {
-    await uploadRemoteRunnerFileChunks({ sourcePath: input.sourcePath, remotePath: remoteUpload, run });
+    await uploadRemoteRunnerFileChunks({ sourcePath: input.sourcePath, remotePath: remoteUpload, run, runner });
     await run(`chmod ${input.mode.toString(8)} ${quote(remoteUpload)} && mv -f ${quote(remoteUpload)} ${quote(input.targetPath)}`);
   } finally {
     await run(`rm -f ${quote(remoteUpload)}`).catch(() => undefined);
@@ -10572,6 +10572,7 @@ async function withRemoteArtifactUploadSlot(operation: () => Promise<void>): Pro
 }
 
 async function uploadRemoteRunnerFileChunks(input: {
+  runner: CommandManagedRuntimeRunner;
   sourcePath: string;
   remotePath: string;
   run: (script: string, stdin?: string) => Promise<void>;
@@ -10597,9 +10598,12 @@ async function uploadRemoteRunnerFileChunks(input: {
       // Disjoint fixed offsets permit bounded parallel upload and idempotent
       // writes. The final digest rejects missing, duplicated, or partial bytes.
       const offset = chunkIndex++;
-      pending.push(withRemoteArtifactUploadSlot(() => run(
+      const upload = () => run(
         `base64 -d | dd of=${quote(remotePath)} bs=4194304 seek=${offset} conv=notrunc 2>/dev/null`,
-        chunk.toString("base64"))).catch((error) => { uploadError ??= error; }));
+        chunk.toString("base64"));
+      pending.push(withRemoteArtifactUploadSlot(() => input.runner.withStdinUploadSlot
+        ? input.runner.withStdinUploadSlot(upload, { timeoutMs: 180_000 })
+        : upload()).catch((error) => { uploadError ??= error; }));
       if (pending.length === REMOTE_ARTIFACT_UPLOAD_CONCURRENCY) {
         await Promise.all(pending);
         pending.length = 0;
@@ -10708,7 +10712,7 @@ export async function stageRemoteRunnerDirectory(input: {
         (error) => error ? reject(error) : resolve());
     });
     await uploadRemoteRunnerFileChunks({
-      sourcePath: archivePath, remotePath: remoteArchive, run, onProgress: input.onProgress,
+      sourcePath: archivePath, remotePath: remoteArchive, run, runner: input.runner, onProgress: input.onProgress,
     });
     await run(`umask 077 && mkdir -p ${quote(input.targetPath)} && ` +
       `tar -xzf ${quote(remoteArchive)} -C ${quote(input.targetPath)} && chmod ${input.mode.toString(8)} ${quote(input.targetPath)}`);

server/src/services/native-runtime/remote-provider-pack-cache.test.ts

@@ -120,6 +120,19 @@ describe("persistent computer provider pack cache", () => {
     expect(() => computerProviderPackCachePath("/agents/../a", digest)).toThrow("identity_invalid");
   });
 
+  it("preserves transport-shared upload backpressure through the ownership guard", async () => {
+    const input = await fixture();
+    const withStdinUploadSlot = vi.fn(async (operation: () => Promise<void>) => operation());
+    await expect(prepareComputerProviderPackCache({ ...input,
+      runner: { ...input.runner, withStdinUploadSlot } as CommandManagedRuntimeRunner,
+      stage: async (root, guarded) => {
+        expect(guarded.withStdinUploadSlot).toBe(withStdinUploadSlot);
+        await guarded.withStdinUploadSlot!(() => input.stage(root, guarded), { timeoutMs: 180_000 });
+      },
+    })).resolves.toBe("published");
+    expect(withStdinUploadSlot).toHaveBeenCalledOnce();
+  });
+
   it("verifies a shared hit before the compression/upload callback and excludes its ownership marker", async () => {
     const input = await fixture();
     const stage = vi.fn(input.stage); const verify = vi.fn(input.verify);

server/src/services/native-runtime/remote-provider-pack-cache.ts

@@ -298,6 +298,7 @@ export async function prepareComputerProviderPackCache(input: {
   }
   const stagingRoot = posix.join(posix.dirname(input.cacheRoot), ".uploads", session, posix.basename(input.cacheRoot), identity.attemptId, "pack");
   const guardedRunner: CommandManagedRuntimeRunner = {
+    withStdinUploadSlot: input.runner.withStdinUploadSlot,
     execute: (command) => input.runner.execute({ ...command, command: "python3", args: [...args("guard"), command.command, ...(command.args ?? [])] }),
   };
   try {