mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-04 02:32:07 +01:00
fix: share one bounded lifetime across computer file operations
Propagate the admitted operation deadline through readiness, claims, and SSH commands. Keep only a bounded transport-settlement hold, and expire a short file waiter without cancelling shared SSH initialization. Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
2c09c8c16d
commit
9c583bf39d
5 files changed
+188
-28
No files matched your search
@@ -44,6 +44,60 @@ const json = (value: unknown, status = 200, headers?: Record<string, string>) =>
|
||||
headers: { "content-type": "application/json", ...headers },
|
||||
});
|
||||
describe("Boat transport validation", () => {
|
||||
it("uses only the file budget remaining after transport preparation", async () => {
|
||||
let clock = 1_000_000;
|
||||
const time = vi.spyOn(Date, "now").mockImplementation(() => clock);
|
||||
const scoped = { ...record, id: randomUUID(), providerId: `bx_${randomUUID()}` };
|
||||
const execute = vi.fn(async () => ({ exitCode: 0, stdout: "{}", stderr: "", timedOut: false, signal: null, pid: null, startedAt: "" }));
|
||||
sshFactory.mockReturnValue({ execute });
|
||||
const backend = boatBackend(async () => "fixture", vi.fn(async () => {
|
||||
clock += 90_000;
|
||||
return json({ hostKey: "ssh-ed25519 AAAA", sshEndpoint: "fixture.invalid:2222" });
|
||||
}));
|
||||
try {
|
||||
await backend.remote(scoped, { action: "read", path: "AGENTS.md" }, { deadlineMs: 1_120_000 });
|
||||
expect(execute).toHaveBeenCalledWith(expect.objectContaining({ timeoutMs: 30_000 }));
|
||||
clock = 1_120_000;
|
||||
await expect(backend.remote(scoped, { action: "write" }, { deadlineMs: clock })).rejects.toThrow("timed out");
|
||||
expect(execute).toHaveBeenCalledOnce();
|
||||
} finally { time.mockRestore(); sshFactory.mockReset(); }
|
||||
});
|
||||
it("expires only the file waiter while shared SSH initialization remains usable", async () => {
|
||||
vi.useFakeTimers();
|
||||
const scoped = { ...record, id: randomUUID(), providerId: `bx_${randomUUID()}` };
|
||||
let initialized!: () => void;
|
||||
const execute = vi.fn(async () => ({ exitCode: 0, stdout: "{}", stderr: "", timedOut: false, signal: null, pid: null, startedAt: "" }));
|
||||
sshFactory.mockReturnValue({ execute });
|
||||
const fetcher = vi.fn(() => new Promise<Response>(resolve => {
|
||||
initialized = () => resolve(json({ hostKey: "ssh-ed25519 AAAA", sshEndpoint: "fixture.invalid:2222" }));
|
||||
}));
|
||||
const backend = boatBackend(async () => "fixture", fetcher);
|
||||
try {
|
||||
const short = backend.remote(scoped, { action: "read" }, { deadlineMs: Date.now() + 20 });
|
||||
const failure = expect(short).rejects.toThrow("timed out");
|
||||
const other = backend.remote(scoped, { action: "owned-port" });
|
||||
await vi.advanceTimersByTimeAsync(21);
|
||||
await failure;
|
||||
expect(execute).not.toHaveBeenCalled();
|
||||
initialized();
|
||||
await other;
|
||||
await backend.remote(scoped, { action: "read" }, { deadlineMs: Date.now() + 1000 });
|
||||
expect(fetcher).toHaveBeenCalledOnce();
|
||||
expect(execute).toHaveBeenCalledTimes(2);
|
||||
} finally { vi.useRealTimers(); sshFactory.mockReset(); }
|
||||
});
|
||||
it("does not resume after the file readiness budget has expired", async () => {
|
||||
let clock = 1_000_000;
|
||||
const time = vi.spyOn(Date, "now").mockImplementation(() => clock);
|
||||
const fetcher = vi.fn(async () => {
|
||||
clock += 120_000;
|
||||
return json({ sandbox: { id: "bx_test", state: "archived", snapshots: true, stop: null } });
|
||||
});
|
||||
try {
|
||||
await expect(boatBackend(async () => "fixture", fetcher).ready(record, { deadlineMs: 1_120_000 })).rejects.toThrow("timed out");
|
||||
expect(fetcher).toHaveBeenCalledOnce();
|
||||
} finally { time.mockRestore(); }
|
||||
});
|
||||
it("reads outgoing and listening TCP ports across IPv4 and IPv6", () => {
|
||||
const root = mkdtempSync(join(tmpdir(), "boat-port-inventory-"));
|
||||
try {
|
||||
|
||||
@@ -4,7 +4,7 @@ import {
|
||||
shellQuote,
|
||||
} from "@paperclipai/adapter-utils/ssh";
|
||||
import type { CommandManagedRuntimeRunner } from "@paperclipai/adapter-utils/command-managed-runtime";
|
||||
import type { ComputerBackend } from "../application/ports.js";
|
||||
import type { ComputerBackend, ComputerOperationOptions } from "../application/ports.js";
|
||||
import {
|
||||
ComputerError,
|
||||
segment,
|
||||
@@ -195,11 +195,17 @@ export function boatBackend(
|
||||
resolveKey: (record: ComputerRecord) => Promise<string>,
|
||||
fetcher: typeof fetch = fetch,
|
||||
): ComputerBackend {
|
||||
function remaining(options: ComputerOperationOptions | undefined, limit: number) {
|
||||
const ms = options?.deadlineMs === undefined ? limit : Math.min(limit, options.deadlineMs - Date.now());
|
||||
if (ms <= 0) throw new ComputerError("provider_error", "Computer file operation timed out");
|
||||
return ms;
|
||||
}
|
||||
async function api(
|
||||
record: ComputerRecord,
|
||||
method: string,
|
||||
suffix = "",
|
||||
body?: unknown,
|
||||
options?: ComputerOperationOptions,
|
||||
): Promise<any> {
|
||||
const response = await fetcher(
|
||||
`${apiOrigin}/sandboxes/${segment(record.providerId)}${suffix}`,
|
||||
@@ -210,7 +216,7 @@ export function boatBackend(
|
||||
"content-type": "application/json",
|
||||
},
|
||||
...(body === undefined ? {} : { body: JSON.stringify(body) }),
|
||||
signal: AbortSignal.timeout(90_000),
|
||||
signal: AbortSignal.timeout(remaining(options, 90_000)),
|
||||
},
|
||||
);
|
||||
if (!response.ok)
|
||||
@@ -223,8 +229,8 @@ export function boatBackend(
|
||||
throw new ComputerError("provider_error", "Invalid Boat response");
|
||||
return data;
|
||||
}
|
||||
async function inspect(record: ComputerRecord) {
|
||||
const data = await api(record, "GET");
|
||||
async function inspect(record: ComputerRecord, options?: ComputerOperationOptions) {
|
||||
const data = await api(record, "GET", "", undefined, options);
|
||||
const sandbox = data.sandbox ?? data;
|
||||
if (sandbox.id !== record.providerId || typeof sandbox.state !== "string")
|
||||
throw new ComputerError(
|
||||
@@ -344,21 +350,39 @@ export function boatBackend(
|
||||
const execute = admission;
|
||||
return { ...raw, execute: (input) => execute(raw, input, options?.control === true) };
|
||||
}
|
||||
async function fileTransport(
|
||||
record: ComputerRecord,
|
||||
control: boolean,
|
||||
options?: ComputerOperationOptions,
|
||||
) {
|
||||
remaining(options, 120_000);
|
||||
const pending = runner(record, { control });
|
||||
if (options?.deadlineMs === undefined) return pending;
|
||||
// Initialization is shared with unrelated owners. Expire this waiter only;
|
||||
// the cached promise retains its normal success/eviction behavior.
|
||||
return new Promise<CommandManagedRuntimeRunner>((resolve, reject) => {
|
||||
const timer = setTimeout(() => reject(new ComputerError(
|
||||
"provider_error", "Computer file operation timed out",
|
||||
)), remaining(options, 120_000));
|
||||
pending.then(resolve, reject).finally(() => clearTimeout(timer));
|
||||
});
|
||||
}
|
||||
async function execute(
|
||||
record: ComputerRecord,
|
||||
code: string,
|
||||
payload: unknown,
|
||||
control = true,
|
||||
options?: ComputerOperationOptions,
|
||||
) {
|
||||
const result = await (
|
||||
await runner(record, { control })
|
||||
await fileTransport(record, control, options)
|
||||
).execute({
|
||||
command: "python3",
|
||||
args: ["-c", code],
|
||||
stdin: JSON.stringify(payload),
|
||||
timeoutMs: 120_000,
|
||||
timeoutMs: remaining(options, 120_000),
|
||||
});
|
||||
if (result.exitCode !== 0)
|
||||
if (result.exitCode !== 0 || result.timedOut)
|
||||
throw new ComputerError("provider_error", "Computer operation failed");
|
||||
let value: any;
|
||||
try {
|
||||
@@ -422,8 +446,8 @@ export function boatBackend(
|
||||
return {
|
||||
inspect,
|
||||
runner,
|
||||
async ready(record) {
|
||||
const state = await inspect(record);
|
||||
async ready(record, options) {
|
||||
const state = await inspect(record, options);
|
||||
if (!state.snapshots)
|
||||
throw new ComputerError("invalid", "Boat snapshots must be enabled");
|
||||
if (state.stop && ["pending", "failing"].includes(state.stop.status))
|
||||
@@ -432,19 +456,19 @@ export function boatBackend(
|
||||
"Boat is still saving a previous stop",
|
||||
);
|
||||
if (!["ready", "idle", "running"].includes(state.state)) {
|
||||
await api(record, "POST", "/resume", { ttlSeconds: 300 });
|
||||
await api(record, "POST", "/resume", { ttlSeconds: 300 }, options);
|
||||
transportCache.delete(
|
||||
`${record.id}:${record.ledger.secretRef.secretId}:${record.ledger.secretRef.version ?? "latest"}`,
|
||||
);
|
||||
for (let i = 0; i < 90; i++) {
|
||||
const status = await inspect(record);
|
||||
const status = await inspect(record, options);
|
||||
if (["ready", "idle", "running"].includes(status.state)) return;
|
||||
await new Promise((r) => setTimeout(r, 1000));
|
||||
await new Promise((r) => setTimeout(r, remaining(options, 1000)));
|
||||
}
|
||||
throw new ComputerError("provider_error", "Boat did not become ready");
|
||||
}
|
||||
},
|
||||
async claim(record) {
|
||||
async claim(record, options) {
|
||||
await execute(
|
||||
record,
|
||||
String.raw`
|
||||
@@ -462,6 +486,8 @@ print('{}')
|
||||
companyId: record.companyId,
|
||||
computerId: record.id,
|
||||
},
|
||||
true,
|
||||
options,
|
||||
);
|
||||
const retiredOwners = record.ledger.owners
|
||||
.filter((owner) => owner.kind === "runner" && owner.phase === "retired")
|
||||
@@ -472,7 +498,7 @@ print('{}')
|
||||
await execute(record, processProgram, {
|
||||
action: "cleanup-retired",
|
||||
owners: retiredOwners,
|
||||
});
|
||||
}, true, options);
|
||||
}
|
||||
},
|
||||
async runnerPorts(record) {
|
||||
@@ -566,10 +592,10 @@ print('{}')
|
||||
url.pathname = path;
|
||||
return { url: url.toString(), secretHeaders: { Cookie: cookie } };
|
||||
},
|
||||
async remote(record, input) {
|
||||
async remote(record, input, options) {
|
||||
// Git setup and file I/O may wait on locks or transfer substantial data.
|
||||
// Keep lifecycle capacity reserved; only the short owned-port probe uses it.
|
||||
return execute(record, remoteProgram, input, input.action === "owned-port");
|
||||
return execute(record, remoteProgram, input, input.action === "owned-port", options);
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user