mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-15 03:51:22 +02:00
Merge bounded Boat file operations into acceptance branch
Co-Authored-By: Paperclip <noreply@paperclip.ing> * commit '47b4bc1d4ea61e4b92b9d6a3bb36067e4b06490e': fix: enforce file operation deadlines inside the remote process group fix: preserve automatic workspace fallback for missing Boat files fix: share one bounded lifetime across computer file operations
This commit is contained in:
commit
b375f46557
7 files changed
+316
-43
No files matched your search
@@ -248,6 +248,71 @@ describeEmbeddedPostgres("workspace file resources", () => {
|
||||
} finally { factory.mockRestore(); await fs.rm(workspace.root, { recursive: true, force: true }); }
|
||||
});
|
||||
|
||||
it.each(["project", "discovery"])("falls through a missing persistent file to the existing automatic %s candidate", async (fallback) => {
|
||||
const workspace = await makeWorkspace();
|
||||
const graph = await seedGraph(db, { projectRoot: workspace.projectRoot, targetProjectRoot: workspace.targetProjectRoot, executionRoot: workspace.executionRoot });
|
||||
const environmentId = crypto.randomUUID(), placementId = crypto.randomUUID();
|
||||
await db.update(executionWorkspaces).set({ providerType: "computer", cwd: workspace.executionRoot,
|
||||
metadata: { fileAuthority: { kind: "remote-persistent", environmentId, placementId } } })
|
||||
.where(eq(executionWorkspaces.sourceIssueId, graph.issueId));
|
||||
const root = fallback === "project" ? workspace.projectRoot : workspace.targetProjectRoot;
|
||||
const workspaceId = fallback === "project" ? graph.projectWorkspaceId : graph.targetProjectWorkspaceId;
|
||||
await fs.mkdir(path.join(root, "docs"));
|
||||
await fs.writeFile(path.join(root, "docs", "guide.md"), "legitimate project fallback");
|
||||
// A controller path with the same spelling is never the placement's authority.
|
||||
await fs.mkdir(path.join(workspace.executionRoot, "docs"));
|
||||
await fs.writeFile(path.join(workspace.executionRoot, "docs", "guide.md"), "stale controller copy");
|
||||
const missing = Object.assign(new Error("Remote file missing"), { code: "not_found" });
|
||||
const stat = vi.fn().mockRejectedValue(missing);
|
||||
const listPage = vi.fn().mockRejectedValue(missing);
|
||||
const readBytes = vi.fn();
|
||||
const workspaceFiles = vi.fn(async () => ({ root: workspace.executionRoot, stat, listPage, readBytes }));
|
||||
const factory = vi.spyOn(computerModule, "computerService").mockReturnValue({ workspaceFiles } as never);
|
||||
try {
|
||||
const service = workspaceFileResourceService(db);
|
||||
const input = { path: "docs/guide.md" };
|
||||
expect(await service.resolve(graph.issueId, input)).toMatchObject({ workspaceId });
|
||||
expect(await service.readContent(graph.issueId, input)).toMatchObject({ resource: { workspaceId }, content: { data: "legitimate project fallback" } });
|
||||
expect(await service.prepareDownload(graph.issueId, input)).toMatchObject({ resource: { workspaceId }, realPath: path.join(root, "docs", "guide.md") });
|
||||
const availability = await service.availability(graph.issueId, { queries: [input] });
|
||||
expect(availability.results[0]).toMatchObject({ openable: true, resource: { workspaceId } });
|
||||
expect(await service.list(graph.issueId, { path: "docs/" })).toMatchObject({ items: [expect.objectContaining({ workspaceId, title: "guide.md" })] });
|
||||
expect(readBytes).not.toHaveBeenCalled();
|
||||
expect(workspaceFiles).toHaveBeenCalledWith({ companyId: graph.companyId, environmentId, placementId });
|
||||
for (const method of [service.resolve, service.readContent, service.prepareDownload]) {
|
||||
await expect(method(graph.issueId, { ...input, workspace: "execution" })).rejects.toMatchObject({ status: 404 });
|
||||
}
|
||||
await expect(service.list(graph.issueId, { path: "docs/", workspace: "execution" })).rejects.toMatchObject({ status: 404 });
|
||||
expect((await service.availability(graph.issueId, { queries: [{ ...input, workspace: "execution" }] })).results[0])
|
||||
.toMatchObject({ openable: false, unavailableReason: "not_found", resource: null });
|
||||
} finally { factory.mockRestore(); await fs.rm(workspace.root, { recursive: true, force: true }); }
|
||||
});
|
||||
|
||||
it.each([["forbidden", 403], ["provider_error", 502]] as const)("does not fall back from a persistent %s error", async (code, status) => {
|
||||
const workspace = await makeWorkspace();
|
||||
const graph = await seedGraph(db, { projectRoot: workspace.projectRoot, executionRoot: workspace.executionRoot });
|
||||
await db.update(executionWorkspaces).set({ providerType: "computer", cwd: workspace.executionRoot,
|
||||
metadata: { fileAuthority: { kind: "remote-persistent", environmentId: crypto.randomUUID(), placementId: crypto.randomUUID() } } })
|
||||
.where(eq(executionWorkspaces.sourceIssueId, graph.issueId));
|
||||
await fs.mkdir(path.join(workspace.projectRoot, "docs"));
|
||||
await fs.writeFile(path.join(workspace.projectRoot, "docs", "guide.md"), "must not be returned");
|
||||
const failure = Object.assign(new Error("Remote access failed"), { code });
|
||||
const workspaceFiles = vi.fn(async () => ({ stat: vi.fn().mockRejectedValue(failure), listPage: vi.fn().mockRejectedValue(failure) }));
|
||||
const factory = vi.spyOn(computerModule, "computerService").mockReturnValue({ workspaceFiles } as never);
|
||||
try {
|
||||
const service = workspaceFileResourceService(db);
|
||||
for (const method of [service.resolve, service.readContent, service.prepareDownload]) {
|
||||
await expect(method(graph.issueId, { path: "docs/guide.md" })).rejects.toMatchObject({ status });
|
||||
}
|
||||
await expect(service.list(graph.issueId, { path: "docs/" })).rejects.toMatchObject({ status });
|
||||
if (code === "forbidden") {
|
||||
expect((await service.availability(graph.issueId, { queries: [{ path: "docs/guide.md" }] })).results[0]).toMatchObject({ openable: false, resource: null });
|
||||
} else {
|
||||
await expect(service.availability(graph.issueId, { queries: [{ path: "docs/guide.md" }] })).rejects.toMatchObject({ status });
|
||||
}
|
||||
} finally { factory.mockRestore(); await fs.rm(workspace.root, { recursive: true, force: true }); }
|
||||
});
|
||||
|
||||
it("withholds private task files and private cross-project targets", async () => {
|
||||
const { projectRoot, executionRoot, targetProjectRoot } = await makeWorkspace();
|
||||
const graph = await seedGraph(db, { projectRoot, executionRoot, targetProjectRoot });
|
||||
|
||||
@@ -44,6 +44,88 @@ 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({
|
||||
command: "timeout", args: expect.arrayContaining(["--signal=KILL", "30s", "python3"]), 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("kills the remote Python group at the file deadline, independent of SSH closure", async () => {
|
||||
const temp = realpathSync(mkdtempSync(join(tmpdir(), "boat-file-deadline-")));
|
||||
const lateWrite = join(temp, "late-write");
|
||||
let pid: number | undefined;
|
||||
const scoped = { ...record, id: randomUUID(), providerId: `bx_${randomUUID()}` };
|
||||
const execute = vi.fn(async (input: { command: string; args: string[]; stdin?: string }) => {
|
||||
expect(input.command).toBe("timeout");
|
||||
const args = [...input.args];
|
||||
args[args.length - 1] = `import os,time;print(os.getpid(),flush=True);time.sleep(2);open(${JSON.stringify(lateWrite)},'w').write('late')`;
|
||||
const result = spawnSync(input.command, args, { encoding: "utf8", input: input.stdin });
|
||||
pid = Number(result.stdout.trim());
|
||||
expect(pid).toBeGreaterThan(0);
|
||||
return { exitCode: result.status, stdout: result.stdout, stderr: result.stderr, timedOut: false, signal: result.signal, pid: null, startedAt: "" };
|
||||
});
|
||||
sshFactory.mockReturnValue({ execute });
|
||||
const backend = boatBackend(async () => "fixture", vi.fn(async () => json({ hostKey: "ssh-ed25519 AAAA", sshEndpoint: "fixture.invalid:2222" })));
|
||||
try {
|
||||
await expect(backend.remote(scoped, { action: "read" }, { deadlineMs: Date.now() + 300 })).rejects.toThrow("Computer operation failed");
|
||||
await vi.waitFor(() => expect(() => process.kill(pid!, 0)).toThrow(), { timeout: 2000 });
|
||||
expect(existsSync(lateWrite)).toBe(false);
|
||||
expect(execute).toHaveBeenCalledOnce();
|
||||
} finally {
|
||||
sshFactory.mockReset();
|
||||
rmSync(temp, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
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(
|
||||
@@ -334,7 +340,7 @@ export function boatBackend(
|
||||
}
|
||||
return pending;
|
||||
}
|
||||
async function runner(record: ComputerRecord, options?: { control?: boolean }): Promise<CommandManagedRuntimeRunner> {
|
||||
async function runner(record: ComputerRecord, options?: { control?: boolean; remoteTimeout?: boolean }): Promise<CommandManagedRuntimeRunner> {
|
||||
const raw = await rawRunner(record);
|
||||
let admission = transportAdmission.get(record.providerId);
|
||||
if (!admission) {
|
||||
@@ -342,23 +348,51 @@ export function boatBackend(
|
||||
transportAdmission.set(record.providerId, admission);
|
||||
}
|
||||
const execute = admission;
|
||||
return { ...raw, execute: (input) => execute(raw, input, options?.control === true) };
|
||||
const bounded = options?.remoteTimeout ? {
|
||||
...raw,
|
||||
execute: (input: CommandInput) => raw.execute({
|
||||
...input,
|
||||
// Admission has already deducted queue time. Bound the remote process
|
||||
// group as well as the local SSH child; socket closure is not stop proof.
|
||||
command: "timeout",
|
||||
args: ["--signal=KILL", `${input.timeoutMs! / 1000}s`, input.command, ...(input.args ?? [])],
|
||||
}),
|
||||
} : raw;
|
||||
return { ...raw, execute: (input) => execute(bounded, input, options?.control === true) };
|
||||
}
|
||||
async function fileTransport(
|
||||
record: ComputerRecord,
|
||||
control: boolean,
|
||||
options?: ComputerOperationOptions,
|
||||
) {
|
||||
remaining(options, 120_000);
|
||||
const pending = runner(record, { control, remoteTimeout: options?.deadlineMs !== undefined });
|
||||
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 +456,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 +466,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 +496,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 +508,7 @@ print('{}')
|
||||
await execute(record, processProgram, {
|
||||
action: "cleanup-retired",
|
||||
owners: retiredOwners,
|
||||
});
|
||||
}, true, options);
|
||||
}
|
||||
},
|
||||
async runnerPorts(record) {
|
||||
@@ -566,10 +602,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);
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
@@ -23,14 +23,16 @@ export interface ComputerRepository {
|
||||
agentId: string,
|
||||
): Promise<"active" | "terminal" | "missing">;
|
||||
}
|
||||
export type ComputerOperationOptions = { deadlineMs?: number };
|
||||
|
||||
export interface ComputerBackend {
|
||||
inspect(record: ComputerRecord): Promise<{
|
||||
state: string;
|
||||
snapshots: boolean;
|
||||
stop: null | { id: string; status: string };
|
||||
}>;
|
||||
ready(record: ComputerRecord): Promise<void>;
|
||||
claim(record: ComputerRecord): Promise<void>;
|
||||
ready(record: ComputerRecord, options?: ComputerOperationOptions): Promise<void>;
|
||||
claim(record: ComputerRecord, options?: ComputerOperationOptions): Promise<void>;
|
||||
advance(record: ComputerRecord, owner: Owner): Promise<void>;
|
||||
runnerPorts(record: ComputerRecord): Promise<RunnerPortAvailability>;
|
||||
runner(record: ComputerRecord, options?: { control?: boolean }): Promise<CommandManagedRuntimeRunner>;
|
||||
@@ -66,6 +68,6 @@ export interface ComputerBackend {
|
||||
path: string,
|
||||
): Promise<{ url: string; secretHeaders: Record<string, string> }>;
|
||||
preview(record: ComputerRecord, port: number): Promise<{ url: string }>;
|
||||
remote(record: ComputerRecord, input: Record<string, unknown>): Promise<any>;
|
||||
remote(record: ComputerRecord, input: Record<string, unknown>, options?: ComputerOperationOptions): Promise<any>;
|
||||
}
|
||||
export type { Ledger };
|
||||
@@ -512,6 +512,66 @@ describe("computer ownership", () => {
|
||||
).resolves.toHaveProperty("owner");
|
||||
expect(f.backend.stop).toHaveBeenCalledOnce();
|
||||
});
|
||||
it("shares one file budget across readiness, claim, and command without early retirement", async () => {
|
||||
const f = fixture();
|
||||
await f.attach();
|
||||
const files = await f.service.files({ ...f.scope, agentId: "agent" });
|
||||
const expectedDeadline = Date.parse("2026-10-10T12:02:00Z");
|
||||
vi.mocked(f.backend.ready).mockImplementation(async (_record, options) => {
|
||||
expect(options?.deadlineMs).toBe(expectedDeadline);
|
||||
f.advance(90_000);
|
||||
});
|
||||
vi.mocked(f.backend.claim).mockImplementation(async (_record, options) => {
|
||||
expect(options?.deadlineMs).toBe(expectedDeadline);
|
||||
f.advance(20_000);
|
||||
});
|
||||
vi.mocked(f.backend.remote).mockImplementation(async (_record, _input, options) => {
|
||||
expect(options?.deadlineMs).toBe(expectedDeadline);
|
||||
f.advance(9999);
|
||||
await f.service.reconcile();
|
||||
expect(f.backend.stop).not.toHaveBeenCalled();
|
||||
f.advance(1);
|
||||
return {};
|
||||
});
|
||||
await files.stat("AGENTS.md");
|
||||
f.advance(5001);
|
||||
await f.service.reconcile();
|
||||
expect(f.backend.stop).toHaveBeenCalledOnce();
|
||||
});
|
||||
it("does not launch file work after slow readiness consumes its budget", async () => {
|
||||
const f = fixture();
|
||||
await f.attach();
|
||||
const files = await f.service.files({ ...f.scope, agentId: "agent" });
|
||||
vi.mocked(f.backend.claim).mockClear();
|
||||
vi.mocked(f.backend.ready).mockImplementation(async () => { f.advance(120_000); });
|
||||
await expect(files.stat("AGENTS.md")).rejects.toThrow("timed out");
|
||||
expect(f.backend.claim).not.toHaveBeenCalled();
|
||||
expect(f.backend.remote).not.toHaveBeenCalled();
|
||||
f.advance(5001);
|
||||
await f.service.reconcile();
|
||||
expect(f.backend.stop).toHaveBeenCalledOnce();
|
||||
});
|
||||
it("bounds an abandoned file hold and fences a late readiness continuation", async () => {
|
||||
const f = fixture();
|
||||
await f.attach();
|
||||
const files = await f.service.files({ ...f.scope, agentId: "agent" });
|
||||
let finish!: () => void;
|
||||
let entered!: () => void;
|
||||
const readyEntered = new Promise<void>(resolve => { entered = resolve; });
|
||||
vi.mocked(f.backend.claim).mockClear();
|
||||
vi.mocked(f.backend.ready).mockImplementation(() => new Promise<void>(resolve => { finish = resolve; entered(); }));
|
||||
const operation = files.stat("AGENTS.md");
|
||||
const rejected = expect(operation).rejects.toThrow("timed out");
|
||||
await readyEntered;
|
||||
f.advance(125_001);
|
||||
await f.service.reconcile();
|
||||
expect(f.backend.stop).toHaveBeenCalledOnce();
|
||||
finish();
|
||||
await rejected;
|
||||
expect(f.backend.claim).not.toHaveBeenCalled();
|
||||
expect(f.backend.remote).not.toHaveBeenCalled();
|
||||
expect((await f.repository.get(f.scope)).ledger.owners.every(owner => owner.phase === "retired")).toBe(true);
|
||||
});
|
||||
it("keeps a short bounded file batch hold between editor operations", async () => {
|
||||
const f = fixture();
|
||||
await f.attach();
|
||||
|
||||
@@ -26,6 +26,8 @@ class ComputerSessionBusyError extends ComputerError {
|
||||
|
||||
// This is shutdown time only; admission and ordinary work stop at the warm deadline.
|
||||
const RETIREMENT_GRACE_MS = 30_000;
|
||||
const FILE_OPERATION_BUDGET_MS = 120_000;
|
||||
const FILE_OPERATION_SETTLEMENT_MS = 5000;
|
||||
type Scope = { companyId: string; environmentId: string };
|
||||
export function createComputerService(
|
||||
repository: ComputerRepository,
|
||||
@@ -697,26 +699,39 @@ finally:
|
||||
async function withFiles<T>(
|
||||
input: Scope,
|
||||
root: string,
|
||||
fn: (record: ComputerRecord) => Promise<T>,
|
||||
fn: (record: ComputerRecord, deadlineMs: number) => Promise<T>,
|
||||
): Promise<T> {
|
||||
const data = await admitRecord(input, (record) => {
|
||||
const deadlineMs = now().getTime() + FILE_OPERATION_BUDGET_MS;
|
||||
const owner: Owner = {
|
||||
id: randomUUID(),
|
||||
generation: 1,
|
||||
kind: "file-operation",
|
||||
phase: "active",
|
||||
port: 0,
|
||||
deadline: new Date(now().getTime() + 120_000).toISOString(),
|
||||
// Transport cancellation has a bounded settlement window; it is not
|
||||
// additional time in which commands may start or continue.
|
||||
deadline: new Date(deadlineMs + FILE_OPERATION_SETTLEMENT_MS).toISOString(),
|
||||
absoluteDeadline: null,
|
||||
process: null,
|
||||
};
|
||||
record.ledger.owners.push(owner);
|
||||
return { record: structuredClone(record), owner };
|
||||
return { record: structuredClone(record), owner, deadlineMs };
|
||||
});
|
||||
const assertLive = async () => {
|
||||
if (now().getTime() >= data.deadlineMs)
|
||||
throw new ComputerError("provider_error", "Computer file operation timed out");
|
||||
const current = await repository.get(input);
|
||||
if (exactOwner(current, ref(data.record, data.owner)).phase !== "active")
|
||||
throw new ComputerError("conflict", "Computer file operation is no longer active");
|
||||
};
|
||||
try {
|
||||
await backend.ready(data.record);
|
||||
await backend.claim(data.record);
|
||||
return await fn(data.record);
|
||||
await assertLive();
|
||||
await backend.ready(data.record, { deadlineMs: data.deadlineMs });
|
||||
await assertLive();
|
||||
await backend.claim(data.record, { deadlineMs: data.deadlineMs });
|
||||
await assertLive();
|
||||
return await fn(data.record, data.deadlineMs);
|
||||
} finally {
|
||||
await repository.update(input, (record) => {
|
||||
const owner = record.ledger.owners.find(
|
||||
@@ -724,15 +739,18 @@ finally:
|
||||
);
|
||||
if (owner?.phase === "active") {
|
||||
owner.phase = "warm";
|
||||
owner.deadline = new Date(now().getTime() + 5000).toISOString();
|
||||
owner.deadline = new Date(Math.min(
|
||||
now().getTime() + FILE_OPERATION_SETTLEMENT_MS,
|
||||
data.deadlineMs + FILE_OPERATION_SETTLEMENT_MS,
|
||||
)).toISOString();
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
function fileAccess(input: Scope, root: string) {
|
||||
const remote = (payload: Record<string, unknown>) =>
|
||||
withFiles(input, root, (record) =>
|
||||
backend.remote(record, { ...payload, root }),
|
||||
withFiles(input, root, (record, deadlineMs) =>
|
||||
backend.remote(record, { ...payload, root }, { deadlineMs }),
|
||||
);
|
||||
const readBytes = async (
|
||||
path: string,
|
||||
|
||||
@@ -1385,14 +1385,14 @@ export function workspaceFileResourceService(db: Db, actor?: AuthorizationActor,
|
||||
const hasExplicitTarget = Boolean(explicitTarget?.candidate);
|
||||
let lastNotFound: unknown = null;
|
||||
for (const candidate of candidates) {
|
||||
if (candidate.placement) return availabilityResult(item.query, await statPersistentCandidate(candidate, item.normalizedPath, item.directory));
|
||||
if (candidate.remote) {
|
||||
if (candidate.remote && !candidate.placement) {
|
||||
if (hasExplicitTarget || item.query.workspace !== "auto") {
|
||||
return availabilityResult(item.query, remoteResource(candidate, item.normalizedPath.relativePath));
|
||||
}
|
||||
continue;
|
||||
}
|
||||
try {
|
||||
if (candidate.placement) return availabilityResult(item.query, await statPersistentCandidate(candidate, item.normalizedPath, item.directory));
|
||||
const resource = item.directory
|
||||
? (await statLocalDirectory(candidate, item.normalizedPath)).resource
|
||||
: (await statLocalCandidate(candidate, item.normalizedPath)).resource;
|
||||
@@ -1454,12 +1454,12 @@ export function workspaceFileResourceService(db: Db, actor?: AuthorizationActor,
|
||||
|
||||
let lastNotFound: unknown = null;
|
||||
for (const candidate of candidates) {
|
||||
if (candidate.placement) return statPersistentCandidate(candidate, normalized, isDirectoryRequest);
|
||||
if (candidate.remote) {
|
||||
if (candidate.remote && !candidate.placement) {
|
||||
if (explicitTarget || selector !== "auto") return remoteResource(candidate, normalized.relativePath);
|
||||
continue;
|
||||
}
|
||||
try {
|
||||
if (candidate.placement) return await statPersistentCandidate(candidate, normalized, isDirectoryRequest);
|
||||
return isDirectoryRequest
|
||||
? (await statLocalDirectory(candidate, normalized)).resource
|
||||
: (await statLocalCandidate(candidate, normalized)).resource;
|
||||
@@ -1513,7 +1513,17 @@ export function workspaceFileResourceService(db: Db, actor?: AuthorizationActor,
|
||||
let firstUnavailable: { candidate: WorkspaceCandidate; reason: string } | null = null;
|
||||
let lastNotFound: unknown = null;
|
||||
for (const candidate of candidates) {
|
||||
if (candidate.placement) return persistentList(candidate, { selector, mode, normalizedPath, q, normalizedQuery, limit, offset });
|
||||
if (candidate.placement) {
|
||||
try {
|
||||
return await persistentList(candidate, { selector, mode, normalizedPath, q, normalizedQuery, limit, offset });
|
||||
} catch (error) {
|
||||
if (normalizedPath && !explicitTarget && selector === "auto" && isHttpStatus(error, 404)) {
|
||||
lastNotFound = error;
|
||||
continue;
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
if (candidate.remote) {
|
||||
firstUnavailable ??= { candidate, reason: "remote_workspace" };
|
||||
if (explicitTarget || selector !== "auto") {
|
||||
@@ -1720,8 +1730,7 @@ export function workspaceFileResourceService(db: Db, actor?: AuthorizationActor,
|
||||
|
||||
let lastNotFound: unknown = null;
|
||||
for (const candidate of candidates) {
|
||||
if (candidate.placement) return persistentContent(candidate, normalized);
|
||||
if (candidate.remote) {
|
||||
if (candidate.remote && !candidate.placement) {
|
||||
if (explicitTarget || selector !== "auto") {
|
||||
throw unprocessable("Remote workspaces cannot be previewed by the server", { code: "remote_workspace" });
|
||||
}
|
||||
@@ -1729,6 +1738,7 @@ export function workspaceFileResourceService(db: Db, actor?: AuthorizationActor,
|
||||
}
|
||||
let resolved: LocalResolvedFile;
|
||||
try {
|
||||
if (candidate.placement) return await persistentContent(candidate, normalized);
|
||||
resolved = await statLocalCandidate(candidate, normalized);
|
||||
} catch (error) {
|
||||
if (!explicitTarget && selector === "auto" && isHttpStatus(error, 404)) {
|
||||
@@ -1805,18 +1815,18 @@ export function workspaceFileResourceService(db: Db, actor?: AuthorizationActor,
|
||||
|
||||
let lastNotFound: unknown = null;
|
||||
for (const candidate of candidates) {
|
||||
if (candidate.placement) {
|
||||
const resource = await statPersistentCandidate(candidate, normalized);
|
||||
const { bytes } = await (await placementFiles(candidate)).readBytes(normalized.relativePath);
|
||||
return { resource: { ...resource, byteSize: bytes.length }, bytes };
|
||||
}
|
||||
if (candidate.remote) {
|
||||
if (candidate.remote && !candidate.placement) {
|
||||
if (explicitTarget || selector !== "auto") {
|
||||
throw unprocessable("Remote workspaces cannot be downloaded by the server", { code: "remote_workspace" });
|
||||
}
|
||||
continue;
|
||||
}
|
||||
try {
|
||||
if (candidate.placement) {
|
||||
const resource = await statPersistentCandidate(candidate, normalized);
|
||||
const { bytes } = await (await placementFiles(candidate)).readBytes(normalized.relativePath);
|
||||
return { resource: { ...resource, byteSize: bytes.length }, bytes };
|
||||
}
|
||||
return await statLocalCandidate(candidate, normalized);
|
||||
} catch (error) {
|
||||
if (!explicitTarget && selector === "auto" && isHttpStatus(error, 404)) {
|
||||
|
||||
Reference in new issue
Block a user