mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-10 12:37:55 +01:00
Merge remote-tracking branch 'origin/master' into paperclip-work
This commit is contained in:
cryppadotta committed
commit
ee030bdf22
89 files changed
+5635
-127
No files matched your search
@@ -0,0 +1,34 @@
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import type { AiConnectionBinding } from "@paperclipai/shared";
|
||||
import { readAiConnectionConfigurationFailure } from "../services/ai-connection-configuration-failure.js";
|
||||
import { aiConnectionService } from "../services/ai-connections.js";
|
||||
|
||||
describe("AI connection binding mode guard", () => {
|
||||
it("rejects an unknown binding mode before it reads any connection", async () => {
|
||||
// Any database read fails the test: the guard must run first.
|
||||
const select = vi.fn(() => {
|
||||
throw new Error("unexpected database read");
|
||||
});
|
||||
const service = aiConnectionService({ select } as unknown as Db);
|
||||
const binding = {
|
||||
provider: "openai",
|
||||
method: "api_key",
|
||||
mode: "connection",
|
||||
connectionId: "11111111-1111-4111-8111-111111111111",
|
||||
grantId: "22222222-2222-4222-8222-222222222222",
|
||||
} as unknown as AiConnectionBinding;
|
||||
|
||||
const error = await service.select({
|
||||
companyId: "33333333-3333-4333-8333-333333333333",
|
||||
userId: "user-1",
|
||||
agentId: "44444444-4444-4444-8444-444444444444",
|
||||
adapterType: "codex_local",
|
||||
binding,
|
||||
}).catch((caught: unknown) => caught);
|
||||
|
||||
expect(error).toMatchObject({ status: 422 });
|
||||
expect(readAiConnectionConfigurationFailure(error)).toBe("ai_connection_incompatible");
|
||||
expect(select).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
@@ -1020,6 +1020,75 @@ describe("environment routes", () => {
|
||||
expect(mockEnvironmentService.create).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
describe("managed plugin environment rows", () => {
|
||||
const PLUGIN_KEY = "example.remote-host-plugin";
|
||||
const pluginRow = () => ({
|
||||
...createPlatformSandboxEnvironment(),
|
||||
id: "env-plugin-1",
|
||||
name: "Remote Host",
|
||||
driver: "plugin",
|
||||
config: { pluginKey: PLUGIN_KEY, driverKey: "tasks", driverConfig: {} },
|
||||
envVars: {},
|
||||
metadata: { managedByPaperclip: true, managedPluginEnvironment: `${PLUGIN_KEY}:tasks` },
|
||||
});
|
||||
|
||||
it("refuses edits, env-var edits, and deletes, including for instance admins", async () => {
|
||||
mockEnvironmentService.getById.mockResolvedValue(pluginRow());
|
||||
const app = createApp(ownerAdminActor);
|
||||
|
||||
for (const body of [{ name: "Renamed" }, { envVars: { A: "1" } }, { status: "archived" }]) {
|
||||
const res = await request(app).patch("/api/environments/env-plugin-1").send(body);
|
||||
expect(res.status).toBe(403);
|
||||
expect(res.body.details).toMatchObject({ code: "environment_platform_managed" });
|
||||
}
|
||||
const deleted = await request(app).delete("/api/environments/env-plugin-1");
|
||||
expect(deleted.status).toBe(403);
|
||||
expect(deleted.body.details).toMatchObject({ code: "environment_platform_managed" });
|
||||
expect(mockEnvironmentService.update).not.toHaveBeenCalled();
|
||||
expect(mockEnvironmentService.removeIfDeletable).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("keeps the row behind the floor on the plugin marker alone", async () => {
|
||||
mockEnvironmentService.getById.mockResolvedValue({
|
||||
...pluginRow(),
|
||||
metadata: { managedPluginEnvironment: `${PLUGIN_KEY}:tasks` },
|
||||
});
|
||||
const res = await request(createApp(ownerAdminActor)).patch("/api/environments/env-plugin-1").send({ name: "Renamed" });
|
||||
expect(res.status).toBe(403);
|
||||
expect(res.body.details).toMatchObject({ code: "environment_platform_managed" });
|
||||
});
|
||||
|
||||
it("allows the marker clear while no managed-config entry can own the row", async () => {
|
||||
// Nothing provisions marked plugin rows yet, so a marker is stale and
|
||||
// stays recoverable. The managed-config plugin entry PR refuses this
|
||||
// while the document declares the row's entry.
|
||||
const row = pluginRow();
|
||||
mockEnvironmentService.getById.mockResolvedValue(row);
|
||||
mockEnvironmentService.update.mockResolvedValue({ ...row, metadata: {} });
|
||||
|
||||
const res = await request(createApp(ownerAdminActor))
|
||||
.patch("/api/environments/env-plugin-1")
|
||||
.send({ metadata: { managedByPaperclip: null, managedPluginEnvironment: null } });
|
||||
|
||||
expect(res.status).toBe(200);
|
||||
expect(mockEnvironmentService.update).toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("refuses a client that stamps the plugin marker on its own row", async () => {
|
||||
const res = await request(createApp(ownerAdminActor))
|
||||
.post("/api/companies/company-1/environments")
|
||||
.send({
|
||||
name: "Fake remote host",
|
||||
driver: "plugin",
|
||||
config: { pluginKey: PLUGIN_KEY, driverKey: "tasks", driverConfig: {} },
|
||||
metadata: { managedPluginEnvironment: `${PLUGIN_KEY}:tasks` },
|
||||
});
|
||||
expect(res.status).toBe(422);
|
||||
expect(res.body.details).toMatchObject({ code: "environment_platform_marker_reserved" });
|
||||
expect(mockEnvironmentService.create).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
it("rejects tenant patches that stamp platform markers so the row cannot become locked", async () => {
|
||||
const tenantEnvironment = {
|
||||
...createPlatformSandboxEnvironment(),
|
||||
|
||||
@@ -0,0 +1,178 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { eq } from "drizzle-orm";
|
||||
import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import {
|
||||
activityLog,
|
||||
agents,
|
||||
companies,
|
||||
createDb,
|
||||
environments,
|
||||
plugins,
|
||||
type Db,
|
||||
} from "@paperclipai/db";
|
||||
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
|
||||
import { environmentService } from "../services/environments.js";
|
||||
|
||||
const PLUGIN_KEY = "example.remote-host-plugin";
|
||||
const MARKER = `${PLUGIN_KEY}:tasks`;
|
||||
|
||||
const support = await getEmbeddedPostgresTestSupport();
|
||||
(support.supported ? describe : describe.skip)("managed plugin environment rows", () => {
|
||||
let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
let db: Db;
|
||||
let pluginId: string;
|
||||
let running: boolean;
|
||||
const call = vi.fn(async (_id: string, method: string, input: { config: Record<string, unknown> }) =>
|
||||
method === "environmentValidateConfig" ? { ok: true, normalizedConfig: { ...input.config, normalized: true } } : null);
|
||||
const workerManager = { isRunning: vi.fn(() => running), call };
|
||||
|
||||
beforeAll(async () => {
|
||||
database = await startEmbeddedPostgresTestDatabase("paperclip-managed-plugin-environment-rows-");
|
||||
db = createDb(database.connectionString);
|
||||
}, 90_000);
|
||||
afterAll(async () => { await database?.cleanup(); });
|
||||
beforeEach(async () => {
|
||||
running = true;
|
||||
call.mockClear();
|
||||
await db.delete(activityLog);
|
||||
await db.delete(agents);
|
||||
await db.delete(environments);
|
||||
await db.delete(plugins);
|
||||
await db.delete(companies);
|
||||
pluginId = randomUUID();
|
||||
await db.insert(plugins).values({
|
||||
id: pluginId,
|
||||
pluginKey: PLUGIN_KEY,
|
||||
packageName: "@example/remote-host-plugin",
|
||||
version: "1.0.0",
|
||||
status: "ready",
|
||||
manifestJson: {
|
||||
capabilities: ["environment.drivers.register"],
|
||||
environmentDrivers: [{ driverKey: "tasks", requiresAgent: true, supportsTasks: true }],
|
||||
} as never,
|
||||
});
|
||||
});
|
||||
|
||||
const ensure = (overrides: Record<string, unknown> = {}) =>
|
||||
environmentService(db).ensureManagedPluginEnvironment({
|
||||
name: "Remote Host",
|
||||
description: "Runs the agent on a remote host",
|
||||
pluginKey: PLUGIN_KEY,
|
||||
driverKey: "tasks",
|
||||
driverConfig: {},
|
||||
workerManager,
|
||||
...overrides,
|
||||
});
|
||||
const pluginRows = () => db.select().from(environments).where(eq(environments.driver, "plugin"));
|
||||
const company = async () => {
|
||||
const id = randomUUID();
|
||||
await db.insert(companies).values({ id, name: "Managed", issuePrefix: `M${id.slice(0, 7)}` });
|
||||
return id;
|
||||
};
|
||||
|
||||
it("creates the marked row through the plugin validator, then changes nothing", async () => {
|
||||
await company();
|
||||
const first = await ensure();
|
||||
expect(first.action).toBe("added");
|
||||
expect(first.environment).toMatchObject({
|
||||
name: "Remote Host",
|
||||
description: "Runs the agent on a remote host",
|
||||
driver: "plugin",
|
||||
status: "active",
|
||||
config: { pluginKey: PLUGIN_KEY, driverKey: "tasks", driverConfig: { normalized: true } },
|
||||
metadata: { managedByPaperclip: true, managedPluginEnvironment: MARKER },
|
||||
});
|
||||
expect(call).toHaveBeenCalledWith(pluginId, "environmentValidateConfig", { driverKey: "tasks", config: {} });
|
||||
expect(await db.select().from(activityLog)).toHaveLength(1);
|
||||
|
||||
const second = await ensure();
|
||||
const rows = await pluginRows();
|
||||
expect(second.action).toBe("unchanged");
|
||||
expect(rows).toHaveLength(1);
|
||||
expect(rows[0]!.updatedAt.getTime()).toBe(first.environment.updatedAt.getTime());
|
||||
expect(await db.select().from(activityLog)).toHaveLength(1);
|
||||
});
|
||||
|
||||
it("adopts a hand-made row for the same plugin and driver, keeping its agents", async () => {
|
||||
const companyId = await company();
|
||||
const [handMade] = await db.insert(environments).values({
|
||||
name: "Remote Host (manual)",
|
||||
driver: "plugin",
|
||||
status: "active",
|
||||
config: { pluginKey: PLUGIN_KEY, driverKey: "tasks", driverConfig: {} },
|
||||
metadata: { note: "made by hand" },
|
||||
}).returning();
|
||||
const agentId = randomUUID();
|
||||
await db.insert(agents).values({ id: agentId, companyId, name: "Agent", adapterType: "process", defaultEnvironmentId: handMade!.id });
|
||||
|
||||
const result = await ensure();
|
||||
const rows = await pluginRows();
|
||||
expect(result.action).toBe("adopted");
|
||||
expect(rows).toHaveLength(1);
|
||||
expect(rows[0]).toMatchObject({
|
||||
id: handMade!.id,
|
||||
name: "Remote Host",
|
||||
metadata: { note: "made by hand", managedByPaperclip: true, managedPluginEnvironment: MARKER },
|
||||
});
|
||||
const [kept] = await db.select().from(agents).where(eq(agents.id, agentId));
|
||||
expect(kept!.defaultEnvironmentId).toBe(handMade!.id);
|
||||
});
|
||||
|
||||
it("adopts a row again after its markers were cleared", async () => {
|
||||
const first = await ensure();
|
||||
// What the route's marker-clear request stores.
|
||||
await db.update(environments)
|
||||
.set({ metadata: { managedByPaperclip: null, managedPluginEnvironment: null } })
|
||||
.where(eq(environments.id, first.environment.id));
|
||||
|
||||
const again = await ensure();
|
||||
expect(again.action).toBe("adopted");
|
||||
expect(again.environment.id).toBe(first.environment.id);
|
||||
expect(await pluginRows()).toHaveLength(1);
|
||||
});
|
||||
|
||||
it("lets several rows hold a cleared marker", async () => {
|
||||
const config = { pluginKey: PLUGIN_KEY, driverKey: "other", driverConfig: {} };
|
||||
await db.insert(environments).values([
|
||||
{ name: "Cleared one", driver: "plugin", status: "active", config, metadata: { managedPluginEnvironment: false } },
|
||||
{ name: "Cleared two", driver: "plugin", status: "active", config, metadata: { managedPluginEnvironment: false } },
|
||||
{ name: "Cleared three", driver: "plugin", status: "active", config, metadata: { managedPluginEnvironment: null } },
|
||||
]);
|
||||
expect(await pluginRows()).toHaveLength(3);
|
||||
});
|
||||
|
||||
it("archives the active marked row and reactivates it on the next ensure", async () => {
|
||||
await ensure();
|
||||
const archived = await environmentService(db).archiveManagedPluginEnvironment({ pluginKey: PLUGIN_KEY, driverKey: "tasks" });
|
||||
expect(archived?.status).toBe("archived");
|
||||
expect(await environmentService(db).archiveManagedPluginEnvironment({ pluginKey: PLUGIN_KEY, driverKey: "tasks" })).toBeNull();
|
||||
expect((await environmentService(db).findManagedPluginEnvironment({ pluginKey: PLUGIN_KEY, driverKey: "tasks" }))?.status)
|
||||
.toBe("archived");
|
||||
|
||||
const reactivated = await ensure();
|
||||
expect(reactivated.action).toBe("updated");
|
||||
expect(reactivated.environment.status).toBe("active");
|
||||
});
|
||||
|
||||
it("refuses to ensure while the plugin has no running worker", async () => {
|
||||
running = false;
|
||||
await expect(ensure()).rejects.toThrow(/has no running worker/);
|
||||
expect(await pluginRows()).toHaveLength(0);
|
||||
});
|
||||
|
||||
it("refuses to take a name another environment already holds", async () => {
|
||||
await db.insert(environments).values({ name: "Remote Host", driver: "ssh", status: "active", config: {} });
|
||||
await expect(ensure()).rejects.toThrow(/already exists with driver "ssh"/);
|
||||
});
|
||||
|
||||
it("allows at most one row per marker", async () => {
|
||||
await ensure();
|
||||
await expect(db.insert(environments).values({
|
||||
name: "Second remote host",
|
||||
driver: "plugin",
|
||||
status: "active",
|
||||
config: { pluginKey: PLUGIN_KEY, driverKey: "tasks", driverConfig: {} },
|
||||
metadata: { managedPluginEnvironment: MARKER },
|
||||
})).rejects.toThrow();
|
||||
});
|
||||
});
|
||||
@@ -673,6 +673,94 @@ describe("plugin tool and bridge authz", () => {
|
||||
);
|
||||
});
|
||||
|
||||
it("allows tool execution for a run with no project and skips the project lookup", async () => {
|
||||
const executeTool = vi.fn().mockResolvedValue({ content: "ok" });
|
||||
const { app } = await createApp(boardActor(), {}, {
|
||||
// Only the agent and run lookups. A project lookup would get no row and fail.
|
||||
db: createSelectQueueDb([
|
||||
[{ companyId: companyA }],
|
||||
[{ companyId: companyA, agentId: agentA }],
|
||||
]),
|
||||
toolDeps: {
|
||||
toolDispatcher: {
|
||||
listToolsForAgent: vi.fn(),
|
||||
getTool: vi.fn(() => ({ name: "paperclip.example:search" })),
|
||||
executeTool,
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
const res = await request(app)
|
||||
.post("/api/plugins/tools/execute")
|
||||
.send({
|
||||
tool: "paperclip.example:search",
|
||||
parameters: {},
|
||||
runContext: { agentId: agentA, runId: runA, companyId: companyA, projectId: "" },
|
||||
});
|
||||
|
||||
expect(res.status).toBe(200);
|
||||
expect(executeTool).toHaveBeenCalledWith(
|
||||
"paperclip.example:search",
|
||||
{},
|
||||
{ agentId: agentA, runId: runA, companyId: companyA, projectId: "" },
|
||||
);
|
||||
});
|
||||
|
||||
it.each([
|
||||
["missing", {}],
|
||||
["null", { projectId: null }],
|
||||
])("passes an empty projectId to the plugin when runContext.projectId is %s", async (_label, project) => {
|
||||
const executeTool = vi.fn().mockResolvedValue({ content: "ok" });
|
||||
const { app } = await createApp(boardActor(), {}, {
|
||||
db: createSelectQueueDb([
|
||||
[{ companyId: companyA }],
|
||||
[{ companyId: companyA, agentId: agentA }],
|
||||
]),
|
||||
toolDeps: {
|
||||
toolDispatcher: {
|
||||
listToolsForAgent: vi.fn(),
|
||||
getTool: vi.fn(() => ({ name: "paperclip.example:search" })),
|
||||
executeTool,
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
const res = await request(app)
|
||||
.post("/api/plugins/tools/execute")
|
||||
.send({
|
||||
tool: "paperclip.example:search",
|
||||
parameters: {},
|
||||
runContext: { agentId: agentA, runId: runA, companyId: companyA, ...project },
|
||||
});
|
||||
|
||||
expect(res.status).toBe(200);
|
||||
expect(executeTool).toHaveBeenCalledWith(
|
||||
"paperclip.example:search",
|
||||
{},
|
||||
{ agentId: agentA, runId: runA, companyId: companyA, projectId: "" },
|
||||
);
|
||||
});
|
||||
|
||||
it("rejects tool execution when runContext.projectId is not a string", async () => {
|
||||
const executeTool = vi.fn();
|
||||
const { app } = await createApp(boardActor(), {}, {
|
||||
toolDeps: {
|
||||
toolDispatcher: { listToolsForAgent: vi.fn(), getTool: vi.fn(), executeTool },
|
||||
},
|
||||
});
|
||||
|
||||
const res = await request(app)
|
||||
.post("/api/plugins/tools/execute")
|
||||
.send({
|
||||
tool: "paperclip.example:search",
|
||||
parameters: {},
|
||||
runContext: { agentId: agentA, runId: runA, companyId: companyA, projectId: 42 },
|
||||
});
|
||||
|
||||
expect(res.status).toBe(400);
|
||||
expect(executeTool).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it.each([
|
||||
["legacy data", "post", `/api/plugins/${pluginId}/bridge/data`, { key: "health" }],
|
||||
["legacy action", "post", `/api/plugins/${pluginId}/bridge/action`, { key: "sync" }],
|
||||
|
||||
@@ -19,6 +19,7 @@ import {
|
||||
import { conflict, forbidden, unprocessable } from "../errors.js";
|
||||
import { isCloudManagedInstance } from "../services/cloud-instance.js";
|
||||
import { getManagedInstanceConfig, SECRET_LIKE_CONFIG_KEY_PATTERN } from "../services/managed-config.js";
|
||||
import { MANAGED_PLUGIN_ENVIRONMENT_MARKER_KEY } from "../services/managed-plugin-environments.js";
|
||||
import { parseExecutionPolicyBootstrapEnv } from "../services/execution-policy-bootstrap.js";
|
||||
import { isExecutionForcedToKubernetes } from "../services/execution-allowlist.js";
|
||||
import { validate } from "../middleware/validate.js";
|
||||
@@ -69,8 +70,10 @@ function isPlainRecord(value: unknown): value is Record<string, unknown> {
|
||||
|
||||
/**
|
||||
* Whether this environment row was provisioned by the platform: the
|
||||
* managed-config environment provisioner marker, or the legacy managed
|
||||
* Kubernetes wrapper marker (rows created by builds that predate the
|
||||
* managed-config environment provisioner marker, the managed plugin
|
||||
* environment marker (`"<pluginKey>:<driverKey>"`, stamped next to
|
||||
* `managedByPaperclip` on a managed-config plugin entry's row), or the legacy
|
||||
* managed Kubernetes wrapper marker (rows created by builds that predate the
|
||||
* generalized provisioner and have not been adopted yet).
|
||||
*/
|
||||
export function isPlatformProvisionedEnvironment(environment: {
|
||||
@@ -78,6 +81,7 @@ export function isPlatformProvisionedEnvironment(environment: {
|
||||
}): boolean {
|
||||
return (
|
||||
environment.metadata?.managedByPaperclip === true ||
|
||||
typeof environment.metadata?.[MANAGED_PLUGIN_ENVIRONMENT_MARKER_KEY] === "string" ||
|
||||
environment.metadata?.managedKubernetesSandbox === true
|
||||
);
|
||||
}
|
||||
@@ -151,6 +155,7 @@ function isTenantEditableManagedSandbox(environment: {
|
||||
|
||||
const PLATFORM_PROVISIONED_MARKER_KEYS = [
|
||||
"managedByPaperclip",
|
||||
MANAGED_PLUGIN_ENVIRONMENT_MARKER_KEY,
|
||||
"managedKubernetesSandbox",
|
||||
] as const;
|
||||
|
||||
|
||||
@@ -9586,7 +9586,7 @@ registry.registerPath({
|
||||
agentId: z.string(),
|
||||
runId: z.string(),
|
||||
companyId: z.string(),
|
||||
projectId: z.string(),
|
||||
projectId: z.string().nullable().optional(),
|
||||
}),
|
||||
}),
|
||||
),
|
||||
|
||||
@@ -482,8 +482,8 @@ interface PluginToolExecuteRequest {
|
||||
tool: string;
|
||||
/** Parameters matching the tool's declared JSON Schema. */
|
||||
parameters?: unknown;
|
||||
/** Agent run context. */
|
||||
runContext: ToolRunContext;
|
||||
/** Agent run context. projectId may be omitted or null when the run has no project. */
|
||||
runContext: Omit<ToolRunContext, "projectId"> & { projectId?: string | null };
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -862,13 +862,15 @@ export function pluginRoutes(
|
||||
return '"runContext.runId" does not belong to "runContext.agentId"';
|
||||
}
|
||||
|
||||
const [project] = await db
|
||||
.select({ companyId: projects.companyId })
|
||||
.from(projects)
|
||||
.where(eq(projects.id, runContext.projectId))
|
||||
.limit(1);
|
||||
if (!project || project.companyId !== runContext.companyId) {
|
||||
return '"runContext.projectId" does not belong to "runContext.companyId"';
|
||||
if (runContext.projectId) {
|
||||
const [project] = await db
|
||||
.select({ companyId: projects.companyId })
|
||||
.from(projects)
|
||||
.where(eq(projects.id, runContext.projectId))
|
||||
.limit(1);
|
||||
if (!project || project.companyId !== runContext.companyId) {
|
||||
return '"runContext.projectId" does not belong to "runContext.companyId"';
|
||||
}
|
||||
}
|
||||
|
||||
return null;
|
||||
@@ -1060,7 +1062,7 @@ export function pluginRoutes(
|
||||
return;
|
||||
}
|
||||
|
||||
const { tool, parameters, runContext } = body;
|
||||
const { tool, parameters, runContext: requestedRunContext } = body;
|
||||
|
||||
// Validate required fields
|
||||
if (!tool || typeof tool !== "string") {
|
||||
@@ -1068,18 +1070,27 @@ export function pluginRoutes(
|
||||
return;
|
||||
}
|
||||
|
||||
if (!runContext || typeof runContext !== "object") {
|
||||
if (!requestedRunContext || typeof requestedRunContext !== "object") {
|
||||
res.status(400).json({ error: '"runContext" is required and must be an object' });
|
||||
return;
|
||||
}
|
||||
|
||||
if (!runContext.agentId || !runContext.runId || !runContext.companyId || !runContext.projectId) {
|
||||
if (!requestedRunContext.agentId || !requestedRunContext.runId || !requestedRunContext.companyId) {
|
||||
res.status(400).json({
|
||||
error: '"runContext" must include agentId, runId, companyId, and projectId',
|
||||
error: '"runContext" must include agentId, runId, and companyId',
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
const requestedProjectId: unknown = requestedRunContext.projectId;
|
||||
if (requestedProjectId != null && typeof requestedProjectId !== "string") {
|
||||
res.status(400).json({ error: '"runContext.projectId" must be a string when present' });
|
||||
return;
|
||||
}
|
||||
// A run can have no project. Plugins then get an empty projectId, the same
|
||||
// value the tool gateway sends, so both paths give plugins one contract.
|
||||
const runContext: ToolRunContext = { ...requestedRunContext, projectId: requestedProjectId ?? "" };
|
||||
|
||||
assertCompanyAccess(req, runContext.companyId);
|
||||
const scopeError = await validateToolRunContextScope(runContext);
|
||||
if (scopeError) {
|
||||
|
||||
@@ -281,6 +281,10 @@ export function aiConnectionService(db: Db) {
|
||||
binding: AiConnectionBinding;
|
||||
}) {
|
||||
const { companyId, userId, agentId, binding } = input;
|
||||
// The grant-kind checks below cover only these three modes. Callers parse
|
||||
// the binding first, but an unknown mode must never skip those checks.
|
||||
if (binding.mode !== "responsible_user" && binding.mode !== "shared" && binding.mode !== "delegated")
|
||||
throw aiConnectionConfigurationFailure("ai_connection_incompatible", "Select a supported AI connection mode");
|
||||
if (
|
||||
!isAiConnectionCompatible(
|
||||
binding,
|
||||
|
||||
@@ -50,6 +50,13 @@ import {
|
||||
stockHash,
|
||||
type ManagedResourceStockStatus,
|
||||
} from "./managed-resource-drift.js";
|
||||
import {
|
||||
archiveManagedPluginEnvironment as archiveManagedPluginEnvironmentRow,
|
||||
ensureManagedPluginEnvironment as ensureManagedPluginEnvironmentRow,
|
||||
findManagedPluginEnvironment as findManagedPluginEnvironmentRow,
|
||||
type ManagedPluginEnvironmentInput,
|
||||
type ManagedPluginEnvironmentReconcileAction,
|
||||
} from "./managed-plugin-environments.js";
|
||||
|
||||
export async function attachSandboxCreation(db: Db, id: string, scope: { companyId: string; environmentId: string; heartbeatRunId?: string | null },
|
||||
attach: (tx: Db) => Promise<EnvironmentLease>): Promise<EnvironmentLease> {
|
||||
@@ -1062,6 +1069,35 @@ export function environmentService(db: Db) {
|
||||
|
||||
archiveManagedSandboxEnvironment,
|
||||
|
||||
/**
|
||||
* Ensure the instance-level row of one managed-config plugin entry,
|
||||
* validating its driver config through the plugin environment validator.
|
||||
* Adopts an unmarked row for the same plugin and driver (see
|
||||
* `managed-plugin-environments.ts`).
|
||||
*/
|
||||
ensureManagedPluginEnvironment: async (
|
||||
input: ManagedPluginEnvironmentInput,
|
||||
): Promise<{ environment: Environment; action: ManagedPluginEnvironmentReconcileAction }> => {
|
||||
const result = await ensureManagedPluginEnvironmentRow(db, input);
|
||||
return { environment: toEnvironment(result.row), action: result.action };
|
||||
},
|
||||
|
||||
/** Archive the active row of a managed plugin entry whose plugin is not running. */
|
||||
archiveManagedPluginEnvironment: async (
|
||||
input: { pluginKey: string; driverKey: string },
|
||||
): Promise<Environment | null> => {
|
||||
const row = await archiveManagedPluginEnvironmentRow(db, input);
|
||||
return row ? toEnvironment(row) : null;
|
||||
},
|
||||
|
||||
/** The marked row of a managed plugin entry, archived or not. */
|
||||
findManagedPluginEnvironment: async (
|
||||
input: { pluginKey: string; driverKey: string },
|
||||
): Promise<Environment | null> => {
|
||||
const row = await findManagedPluginEnvironmentRow(db, input);
|
||||
return row ? toEnvironment(row) : null;
|
||||
},
|
||||
|
||||
/**
|
||||
* Idempotently ensure a managed Kubernetes sandbox environment exists for
|
||||
* an instance, configured from instance/operator-supplied config. A thin
|
||||
|
||||
@@ -0,0 +1,287 @@
|
||||
/**
|
||||
* Instance-level plugin environment rows declared by the managed-config
|
||||
* `environments` section (`driver: "plugin"` entries, see `managed-config.ts`).
|
||||
*
|
||||
* Each declared entry owns exactly one row, identified by the marker
|
||||
* `metadata.managedPluginEnvironment = "<pluginKey>:<driverKey>"` (unique
|
||||
* index `environments_managed_plugin_environment_idx`) next to
|
||||
* `managedByPaperclip: true`, which puts the row behind the environments-route
|
||||
* write floor on cloud-managed instances.
|
||||
*
|
||||
* - The first ensure adopts an existing unmarked row for the same plugin and
|
||||
* driver instead of adding a second one, so a hand-made row keeps the agents
|
||||
* that point at it.
|
||||
* - Later ensures converge the row to the declaration and write nothing when
|
||||
* it already matches.
|
||||
* - Archiving is only for a declared plugin that is not running; the next
|
||||
* successful ensure reactivates the row. Nothing here archives a row
|
||||
* because its entry left the document.
|
||||
*
|
||||
* The driver config always goes through the plugin environment validator
|
||||
* (`normalizeEnvironmentConfig` + `validatePluginEnvironmentDriverConfig`),
|
||||
* the same path an operator-created plugin environment takes.
|
||||
*/
|
||||
|
||||
import { and, eq, sql } from "drizzle-orm";
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import { companies, environments } from "@paperclipai/db";
|
||||
import type { PluginEnvironmentConfig } from "@paperclipai/shared";
|
||||
import { logActivity, publishActivity, type ActivityPublication } from "./activity-log.js";
|
||||
import { normalizeEnvironmentConfig } from "./environment-config.js";
|
||||
import { validatePluginEnvironmentDriverConfig } from "./plugin-environment-driver.js";
|
||||
import type { PluginWorkerManager } from "./plugin-worker-manager.js";
|
||||
|
||||
export const MANAGED_PLUGIN_ENVIRONMENT_MARKER_KEY = "managedPluginEnvironment";
|
||||
|
||||
export function managedPluginEnvironmentMarker(input: { pluginKey: string; driverKey: string }): string {
|
||||
return `${input.pluginKey}:${input.driverKey}`;
|
||||
}
|
||||
|
||||
export interface ManagedPluginEnvironmentInput {
|
||||
name: string;
|
||||
description?: string;
|
||||
pluginKey: string;
|
||||
driverKey: string;
|
||||
driverConfig?: Record<string, unknown>;
|
||||
/** Runs the plugin's own config validation (`environmentValidateConfig`). */
|
||||
workerManager: Pick<PluginWorkerManager, "call" | "isRunning">;
|
||||
}
|
||||
|
||||
export type ManagedPluginEnvironmentReconcileAction = "added" | "adopted" | "updated" | "unchanged";
|
||||
|
||||
export type EnvironmentRow = typeof environments.$inferSelect;
|
||||
|
||||
export interface ManagedPluginEnvironmentRowResult {
|
||||
row: EnvironmentRow;
|
||||
action: ManagedPluginEnvironmentReconcileAction;
|
||||
}
|
||||
|
||||
function metadataOf(row: Pick<EnvironmentRow, "metadata">): Record<string, unknown> {
|
||||
return (row.metadata ?? {}) as Record<string, unknown>;
|
||||
}
|
||||
|
||||
function stableJson(value: unknown): string {
|
||||
if (Array.isArray(value)) return `[${value.map(stableJson).join(",")}]`;
|
||||
if (value && typeof value === "object") {
|
||||
return `{${Object.keys(value as Record<string, unknown>)
|
||||
.sort()
|
||||
.map((key) => `${JSON.stringify(key)}:${stableJson((value as Record<string, unknown>)[key])}`)
|
||||
.join(",")}}`;
|
||||
}
|
||||
return JSON.stringify(value ?? null);
|
||||
}
|
||||
|
||||
function servesPair(row: EnvironmentRow, input: { pluginKey: string; driverKey: string }): boolean {
|
||||
const config = (row.config ?? {}) as Record<string, unknown>;
|
||||
return config.pluginKey === input.pluginKey && config.driverKey === input.driverKey;
|
||||
}
|
||||
|
||||
async function validatedDriverConfig(db: Db, input: ManagedPluginEnvironmentInput): Promise<PluginEnvironmentConfig> {
|
||||
const normalized = normalizeEnvironmentConfig({
|
||||
driver: "plugin",
|
||||
config: { pluginKey: input.pluginKey, driverKey: input.driverKey, driverConfig: input.driverConfig ?? {} },
|
||||
}) as unknown as PluginEnvironmentConfig;
|
||||
return await validatePluginEnvironmentDriverConfig({
|
||||
db,
|
||||
workerManager: input.workerManager as PluginWorkerManager,
|
||||
config: normalized,
|
||||
});
|
||||
}
|
||||
|
||||
/** Marker value of a row, or null when it has none (absent, or cleared to `null` / `false`). */
|
||||
function markerOf(row: Pick<EnvironmentRow, "metadata">): string | null {
|
||||
const value = metadataOf(row)[MANAGED_PLUGIN_ENVIRONMENT_MARKER_KEY];
|
||||
return typeof value === "string" ? value : null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Record activity for every company inside the caller's transaction, so the
|
||||
* entries commit (or roll back) with the row change. Live events go to
|
||||
* `publications` and are published by the caller after commit.
|
||||
*/
|
||||
async function logForEveryCompany(
|
||||
tx: Db,
|
||||
publications: ActivityPublication[],
|
||||
input: { action: string; environmentId: string; details: Record<string, unknown> },
|
||||
): Promise<void> {
|
||||
const companyIds = await tx.select({ id: companies.id }).from(companies).then((rows) => rows.map((row) => row.id));
|
||||
for (const companyId of companyIds) {
|
||||
await logActivity(tx, {
|
||||
companyId,
|
||||
actorType: "system",
|
||||
actorId: "managed-environment-reconciler",
|
||||
action: input.action,
|
||||
entityType: "environment",
|
||||
entityId: input.environmentId,
|
||||
details: input.details,
|
||||
}, publications);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Ensure the row for one declared plugin entry. Idempotent: a row that
|
||||
* already matches the declaration is returned without a write.
|
||||
*/
|
||||
export async function ensureManagedPluginEnvironment(
|
||||
db: Db,
|
||||
input: ManagedPluginEnvironmentInput,
|
||||
): Promise<ManagedPluginEnvironmentRowResult> {
|
||||
const marker = managedPluginEnvironmentMarker(input);
|
||||
const config = await validatedDriverConfig(db, input);
|
||||
const desiredConfig = { ...config } as Record<string, unknown>;
|
||||
const description = input.description ?? null;
|
||||
|
||||
const publications: ActivityPublication[] = [];
|
||||
const result = await db.transaction(async (tx): Promise<ManagedPluginEnvironmentRowResult> => {
|
||||
const reconciled = await reconcile(tx);
|
||||
if (reconciled.action !== "unchanged") {
|
||||
await logForEveryCompany(tx as unknown as Db, publications, {
|
||||
action: `environment.managed_plugin_${reconciled.action}`,
|
||||
environmentId: reconciled.row.id,
|
||||
details: { pluginKey: input.pluginKey, driverKey: input.driverKey, reconciliationAction: reconciled.action },
|
||||
});
|
||||
}
|
||||
return reconciled;
|
||||
});
|
||||
publications.forEach(publishActivity);
|
||||
return result;
|
||||
|
||||
async function reconcile(tx: Parameters<Parameters<Db["transaction"]>[0]>[0]): Promise<ManagedPluginEnvironmentRowResult> {
|
||||
// Lock every plugin row: concurrent ensures (the boot pass and a worker
|
||||
// recovery, or two builds in a rolling deploy) serialize here, so only one
|
||||
// of them can adopt or insert.
|
||||
const pluginRows = await tx.select().from(environments).where(eq(environments.driver, "plugin")).for("update");
|
||||
let row = pluginRows.find((candidate) => markerOf(candidate) === marker) ?? null;
|
||||
let adopted = false;
|
||||
if (!row) {
|
||||
const candidates = pluginRows
|
||||
// A cleared marker (`null` / `false`, from the marker-clear request)
|
||||
// counts as unmarked, so a cleared row is adopted again.
|
||||
.filter((candidate) => servesPair(candidate, input) && markerOf(candidate) === null)
|
||||
.sort((left, right) =>
|
||||
Number(right.name === input.name) - Number(left.name === input.name)
|
||||
|| Number(right.status === "active") - Number(left.status === "active")
|
||||
|| left.createdAt.getTime() - right.createdAt.getTime());
|
||||
row = candidates[0] ?? null;
|
||||
adopted = row !== null;
|
||||
}
|
||||
|
||||
const nameOwner = await tx
|
||||
.select({ id: environments.id, driver: environments.driver })
|
||||
.from(environments)
|
||||
.where(eq(environments.name, input.name))
|
||||
.then((rows) => rows[0] ?? null);
|
||||
|
||||
if (!row) {
|
||||
if (nameOwner) {
|
||||
throw new Error(
|
||||
`Failed to ensure managed plugin environment: environment "${input.name}" already exists with driver "${nameOwner.driver}"`,
|
||||
);
|
||||
}
|
||||
const now = new Date();
|
||||
const inserted = await tx
|
||||
.insert(environments)
|
||||
.values({
|
||||
name: input.name,
|
||||
description,
|
||||
driver: "plugin",
|
||||
status: "active",
|
||||
config: desiredConfig,
|
||||
envVars: {},
|
||||
metadata: { managedByPaperclip: true, [MANAGED_PLUGIN_ENVIRONMENT_MARKER_KEY]: marker },
|
||||
createdAt: now,
|
||||
updatedAt: now,
|
||||
})
|
||||
.onConflictDoNothing()
|
||||
.returning()
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (inserted) return { row: inserted, action: "added" };
|
||||
// A concurrent pass won the marker index; converge on its row.
|
||||
row = await tx
|
||||
.select()
|
||||
.from(environments)
|
||||
.where(sql`${environments.metadata} ->> ${MANAGED_PLUGIN_ENVIRONMENT_MARKER_KEY} = ${marker}`)
|
||||
.for("update")
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (!row) throw new Error("Failed to ensure managed plugin environment");
|
||||
}
|
||||
|
||||
// Keep the row's own name when another row already holds the declared one.
|
||||
const desiredName = nameOwner && nameOwner.id !== row.id ? row.name : input.name;
|
||||
const currentMetadata = metadataOf(row);
|
||||
const desiredMetadata = {
|
||||
...currentMetadata,
|
||||
managedByPaperclip: true,
|
||||
[MANAGED_PLUGIN_ENVIRONMENT_MARKER_KEY]: marker,
|
||||
};
|
||||
const matches =
|
||||
row.name === desiredName &&
|
||||
(row.description ?? null) === description &&
|
||||
row.status === "active" &&
|
||||
stableJson(row.config) === stableJson(desiredConfig) &&
|
||||
stableJson(currentMetadata) === stableJson(desiredMetadata);
|
||||
if (matches) return { row, action: "unchanged" };
|
||||
|
||||
const updated = await tx
|
||||
.update(environments)
|
||||
.set({
|
||||
name: desiredName,
|
||||
description,
|
||||
status: "active",
|
||||
config: desiredConfig,
|
||||
metadata: desiredMetadata,
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(eq(environments.id, row.id))
|
||||
.returning()
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (!updated) throw new Error("Managed plugin environment changed during reconciliation");
|
||||
return { row: updated, action: adopted ? "adopted" : "updated" };
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Archive the marked row of a declared plugin that is not ready or has no
|
||||
* running worker, so run scheduling stops selecting it. Returns null when
|
||||
* there is no active marked row.
|
||||
*/
|
||||
export async function archiveManagedPluginEnvironment(
|
||||
db: Db,
|
||||
input: { pluginKey: string; driverKey: string },
|
||||
): Promise<EnvironmentRow | null> {
|
||||
const marker = managedPluginEnvironmentMarker(input);
|
||||
const publications: ActivityPublication[] = [];
|
||||
const archived = await db.transaction(async (tx) => {
|
||||
const row = await tx
|
||||
.update(environments)
|
||||
.set({ status: "archived", updatedAt: new Date() })
|
||||
.where(and(
|
||||
eq(environments.driver, "plugin"),
|
||||
eq(environments.status, "active"),
|
||||
sql`${environments.metadata} ->> ${MANAGED_PLUGIN_ENVIRONMENT_MARKER_KEY} = ${marker}`,
|
||||
))
|
||||
.returning()
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (!row) return null;
|
||||
await logForEveryCompany(tx as unknown as Db, publications, {
|
||||
action: "environment.managed_provider_unavailable_archived",
|
||||
environmentId: row.id,
|
||||
details: { pluginKey: input.pluginKey, driverKey: input.driverKey },
|
||||
});
|
||||
return row;
|
||||
});
|
||||
publications.forEach(publishActivity);
|
||||
return archived;
|
||||
}
|
||||
|
||||
/** The marked row for one plugin entry, archived or not. */
|
||||
export async function findManagedPluginEnvironment(
|
||||
db: Db,
|
||||
input: { pluginKey: string; driverKey: string },
|
||||
): Promise<EnvironmentRow | null> {
|
||||
return await db
|
||||
.select()
|
||||
.from(environments)
|
||||
.where(sql`${environments.metadata} ->> ${MANAGED_PLUGIN_ENVIRONMENT_MARKER_KEY} = ${managedPluginEnvironmentMarker(input)}`)
|
||||
.then((rows) => rows[0] ?? null);
|
||||
}
|
||||
@@ -4,7 +4,7 @@ import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator
|
||||
import { registerScopedDeadlineWork } from "../scoped-deadline-work.js";
|
||||
import { notifyExecutionRunWork } from "../execution-control-notifications.js";
|
||||
import { and, asc, eq, gt, inArray, isNull, or, sql } from "drizzle-orm";
|
||||
import { environmentLeases, executionWorkspaces, heartbeatRuns, workspaceRestoreFailureCodesSql, issues, nativeRunFinalizations, type Db } from "@paperclipai/db";
|
||||
import { environmentLeases, executionWorkspaces, heartbeatRuns, workspaceRestoreFailureCodesSql, issues, nativeRunFinalizations, nativeRunResults, type Db } from "@paperclipai/db";
|
||||
import { recordRequiredWorkspaceRestoreFailure } from "../legacy-execution-recovery.js";
|
||||
import { hasRequiredWorkspaceRecovery, nativeRunHasNoWorkspaceAccess } from "../workspace-restore-recovery-state.js";
|
||||
import { logActivity } from "../activity-log.js";
|
||||
@@ -102,7 +102,10 @@ export async function preserveNativeFailedWorkspace(db: Db, selected: Pick<typeo
|
||||
&& (claim.kind !== "reattach_remote_runner" || (claim.remote.kind !== "plugin_task" && claim.remote.providerLeaseId === reference.providerLeaseId
|
||||
&& claim.remote.remoteCwd === reference.remoteCwd)));
|
||||
if ((coordinator?.leaseOwner || (claim && coordinator?.phase === "observed")) && !ownsClaim) throw new NativeWorkspaceFinalizationBusyError();
|
||||
if (!coordinator || (coordinator.resultId && !(options.requireCancelledStop && stoppedAcceptedCopyback(coordinator)))
|
||||
const exhaustedAcceptedCopyback = Boolean(coordinator?.resultId && coordinator.phase === "terminal_failure"
|
||||
&& coordinator.failureCode === "native_workspace_sync_out_retry_exhausted");
|
||||
if (!coordinator || (coordinator.resultId && !exhaustedAcceptedCopyback
|
||||
&& !(options.requireCancelledStop && stoppedAcceptedCopyback(coordinator)))
|
||||
|| !(options.allowPreterminal ? ["observed", "terminal_failure"] : ["terminal_failure"]).includes(coordinator.phase)) return;
|
||||
const [run] = await tx.select().from(heartbeatRuns).where(and(
|
||||
eq(heartbeatRuns.id, selected.id), eq(heartbeatRuns.companyId, selected.companyId),
|
||||
@@ -120,9 +123,23 @@ export async function preserveNativeFailedWorkspace(db: Db, selected: Pick<typeo
|
||||
)).limit(1);
|
||||
if (!source || source.issueId !== issueId || source.metadata?.remoteCwd !== currentReference.remoteCwd
|
||||
|| (source.executionWorkspaceId && source.executionWorkspaceId !== currentReference.workspaceId)) return;
|
||||
if (exhaustedAcceptedCopyback) {
|
||||
const remote = remoteWorkspaceRunReferenceSchema.safeParse(run.runnerProfileJson?.remoteWorkspace);
|
||||
if (run.status !== "failed" || !run.finishedAt || !run.completionContractId
|
||||
|| run.controllerLeaseExpiresAt && run.controllerLeaseExpiresAt.getTime() > Date.now()
|
||||
|| run.runnerProfileJson?.nativeWorkspaceSync != null || !remote.success || remote.data.state !== "prepared"
|
||||
|| !hasConfirmedSandboxStopAndRetain(source)
|
||||
|| !await preparedRemoteWorkspaceSource(tx as unknown as Db, source)) return;
|
||||
const [accepted] = await tx.select({ id: nativeRunResults.id }).from(nativeRunResults).where(and(
|
||||
eq(nativeRunResults.companyId, run.companyId), eq(nativeRunResults.issueId, issueId),
|
||||
eq(nativeRunResults.runId, run.id), eq(nativeRunResults.id, coordinator.resultId!),
|
||||
eq(nativeRunResults.completionContractId, run.completionContractId), eq(nativeRunResults.schemaStatus, "accepted"),
|
||||
)).limit(1);
|
||||
if (!accepted) return;
|
||||
}
|
||||
// A stopped reusable sandbox is retained only with its exact stop receipt
|
||||
// while the workspace is still bound to this writer. Otherwise record nothing.
|
||||
if (source.status !== "active" && reusableLeasePolicies.includes(source.leasePolicy) && (!hasConfirmedSandboxStopAndRetain(source)
|
||||
if (!exhaustedAcceptedCopyback && source.status !== "active" && reusableLeasePolicies.includes(source.leasePolicy) && (!hasConfirmedSandboxStopAndRetain(source)
|
||||
|| !await preparedRemoteWorkspaceSource(tx as unknown as Db, source))) return;
|
||||
// Ownership settlement still applies to workspace-less external sessions.
|
||||
// Their Runner descriptor is not evidence that user files need exporting.
|
||||
|
||||
@@ -0,0 +1,135 @@
|
||||
/** Runs unchanged in the sandbox and on its transferred, private controller copy. */
|
||||
export async function readWorkspaceInventory(root: string, controllerCopy = false, signal?: AbortSignal) {
|
||||
const fs = await import("node:fs/promises"), path = await import("node:path"), crypto = await import("node:crypto");
|
||||
const fail = (): never => { throw new Error("remote_workspace_readonly_capture_unsafe"); };
|
||||
const budget = (): never => { throw new Error("remote_workspace_readonly_capture_budget"); };
|
||||
const check = () => signal?.throwIfAborted();
|
||||
check();
|
||||
if (!path.isAbsolute(root) || await fs.realpath(root) !== root || !(await fs.lstat(root)).isDirectory()) fail();
|
||||
const digest = crypto.createHash("sha256"), exclude = [".paperclip-runtime"];
|
||||
const repositories: string[] = [], configs: string[] = [];
|
||||
let entries = 0, scannedEntries = 0, byteSize = 0, configBytes = 0;
|
||||
const gitRead = async (cwd: string, args: string[]) => {
|
||||
check();
|
||||
const { execFile } = await import("node:child_process");
|
||||
return new Promise<string>((resolve, reject) => execFile("git", ["-c", "core.fsmonitor=false", "-c", "core.hooksPath=/dev/null", "-C", cwd, ...args], {
|
||||
signal, timeout: 10_000, maxBuffer: 64 * 1024 * 1024,
|
||||
env: { PATH: process.env.PATH, GIT_CONFIG_NOSYSTEM: "1", GIT_CONFIG_GLOBAL: "/dev/null", GIT_CONFIG_COUNT: "0",
|
||||
GIT_OPTIONAL_LOCKS: "0", GIT_NO_LAZY_FETCH: "1" },
|
||||
}, (error, result) => error ? reject(new Error("remote_workspace_readonly_capture_unsafe")) : resolve(result)));
|
||||
};
|
||||
const visit = async (relative: string, parentRepository: string | null = null) => {
|
||||
check();
|
||||
const directory = path.join(root, relative), before = await fs.lstat(directory);
|
||||
if (!before.isDirectory() || before.isSymbolicLink() || await fs.realpath(directory) !== directory) fail();
|
||||
const names: string[] = [];
|
||||
for await (const entry of await fs.opendir(directory)) {
|
||||
check(); if (++scannedEntries > 100_000) budget(); names.push(entry.name);
|
||||
}
|
||||
const inGit = relative.split("/").includes(".git");
|
||||
const repository = !inGit && names.includes(".git") ? directory : parentRepository;
|
||||
if (!inGit && names.includes(".git")) {
|
||||
const entry = await fs.lstat(path.join(directory, ".git"));
|
||||
if (!entry.isDirectory() || entry.isSymbolicLink()) fail();
|
||||
repositories.push(directory); if (repositories.length > 64) budget();
|
||||
// The sandbox uses the same tracked-dependency rule as normal capture.
|
||||
// Controller verification never invokes repository-aware Git here.
|
||||
if (!controllerCopy && await fs.realpath((await gitRead(directory, ["rev-parse", "--show-toplevel"])).trim()) !== directory) fail();
|
||||
}
|
||||
for (const name of names.sort()) {
|
||||
check();
|
||||
const next = relative ? `${relative}/${name}` : name;
|
||||
if (next === ".paperclip-runtime") continue;
|
||||
if (!name || name === "." || name === ".." || /[\\\u0000-\u001f\u007f]/u.test(name)) fail();
|
||||
const file = path.join(root, next), stat = await fs.lstat(file);
|
||||
// A copied worktree pointer or alternate could resolve against the
|
||||
// controller's filesystem. Only standalone, self-contained Git metadata
|
||||
// can use this fallback; the sandbox remains retained otherwise.
|
||||
const parts = next.split("/"), git = parts.indexOf(".git");
|
||||
if (name === ".git" && !stat.isDirectory()) fail();
|
||||
// Git follows configuration and administrative links itself. A link
|
||||
// inside this tree must not bypass the controller config-file check.
|
||||
if (git >= 0 && stat.isSymbolicLink()) fail();
|
||||
const administrationPath = parts.slice(git + 1).join("/");
|
||||
const foldedAdministrationPath = administrationPath.toLowerCase();
|
||||
// A case-sensitive sandbox may be copied onto a case-insensitive host.
|
||||
// Reject aliases before Git can interpret them as controller metadata.
|
||||
if (git >= 0 && ((foldedAdministrationPath === "config" && administrationPath !== "config")
|
||||
|| ["commondir", "config.worktree", "objects/info/alternates", "objects/info/http-alternates"]
|
||||
.includes(foldedAdministrationPath))) fail();
|
||||
if (stat.isSocket()) { exclude.push(next); continue; }
|
||||
if (stat.isDirectory()) {
|
||||
if (!controllerCopy && !inGit && name !== ".git") {
|
||||
const marker = await fs.lstat(name === "node_modules" ? path.join(directory, "package.json")
|
||||
: path.join(file, "pyvenv.cfg")).catch(() => null);
|
||||
if (marker?.isFile() && !marker.isSymbolicLink()
|
||||
&& (!repository || !(await gitRead(directory, ["ls-files", "-z", "--", name])))) {
|
||||
exclude.push(next); continue;
|
||||
}
|
||||
}
|
||||
digest.update(JSON.stringify([next, "directory", stat.mode & 0o777])); entries++;
|
||||
await visit(next, repository);
|
||||
} else if (stat.isSymbolicLink()) {
|
||||
const target = await fs.readlink(file), resolved = path.posix.normalize(path.posix.join(path.posix.dirname(next), target));
|
||||
if (!target || path.posix.isAbsolute(target) || /[\\\u0000-\u001f\u007f]/u.test(target)
|
||||
|| resolved === ".." || resolved.startsWith("../")) fail();
|
||||
digest.update(JSON.stringify([next, "symlink", target])); entries++;
|
||||
} else if (stat.isFile()) {
|
||||
if (git >= 0 && parts.slice(git + 1).join("/") === "config") {
|
||||
configs.push(file); configBytes += stat.size;
|
||||
if (stat.size > 65_536 || configBytes > 1024 * 1024) budget();
|
||||
}
|
||||
const original = await fs.lstat(file, { bigint: true });
|
||||
const handle = await fs.open(file, fs.constants.O_RDONLY | fs.constants.O_NOFOLLOW | fs.constants.O_NONBLOCK);
|
||||
const hash = crypto.createHash("sha256"); let size = 0;
|
||||
try {
|
||||
const before = await handle.stat({ bigint: true });
|
||||
const fields = ["dev", "ino", "size", "mode", "mtimeNs", "ctimeNs"] as const;
|
||||
if (!before.isFile() || !original.isFile() || Number(before.dev) !== stat.dev || Number(before.ino) !== stat.ino
|
||||
|| Number(before.size) !== stat.size || Number(before.mode) !== stat.mode
|
||||
|| fields.some(key => original[key] !== before[key]) || await fs.realpath(file) !== file) fail();
|
||||
for await (const bytes of handle.createReadStream({ autoClose: false, signal })) { check(); size += bytes.length; hash.update(bytes); }
|
||||
const after = await handle.stat({ bigint: true }), current = await fs.lstat(file, { bigint: true });
|
||||
if (!current.isFile() || size !== Number(before.size) || await fs.realpath(file) !== file
|
||||
|| fields.some(key => before[key] !== after[key] || after[key] !== current[key])) fail();
|
||||
} finally { await handle.close(); }
|
||||
byteSize += size;
|
||||
if (!Number.isSafeInteger(byteSize)) fail();
|
||||
digest.update(JSON.stringify([next, "file", stat.mode & 0o777, size, hash.digest("hex")])); entries++;
|
||||
} else fail();
|
||||
}
|
||||
const after = await fs.lstat(directory);
|
||||
if (before.dev !== after.dev || before.ino !== after.ino || before.mtimeMs !== after.mtimeMs || before.ctimeMs !== after.ctimeMs) fail();
|
||||
};
|
||||
await visit("");
|
||||
if (Buffer.byteLength(JSON.stringify(exclude)) > 65_536) fail();
|
||||
if (controllerCopy) {
|
||||
// Bound the complete tree and verify every repository layout before the
|
||||
// first controller subprocess. Files named config inside Git metadata are
|
||||
// ordinary data unless they are the exact repository-root configuration.
|
||||
for (const repository of repositories) {
|
||||
check();
|
||||
const directory = path.join(repository, ".git"), head = path.join(directory, "HEAD");
|
||||
if ((await fs.stat(head)).size > 4096 || !/^(ref: refs\/\S+|[0-9a-f]{40}|[0-9a-f]{64})\s*$/u.test(await fs.readFile(head, "utf8"))) fail();
|
||||
if (!(await fs.lstat(path.join(directory, "objects"))).isDirectory()
|
||||
|| !(await fs.lstat(path.join(directory, "refs")).catch(() => fs.lstat(path.join(directory, "reftable")))).isDirectory()) fail();
|
||||
}
|
||||
let settings = 0, refspecs = 0;
|
||||
for (const config of configs) {
|
||||
check();
|
||||
// Parse this file as data. Never discover a repository or follow includes.
|
||||
const keys = (await gitRead(path.dirname(root), ["config", "--file", config, "--no-includes", "--null", "--name-only", "--list"]))
|
||||
.split("\0").filter(Boolean);
|
||||
settings += keys.length; refspecs += keys.filter(key => /^remote\..+\.fetch$|^branch\..+\.merge$/u.test(key)).length;
|
||||
if (settings > 1024 || refspecs > 256) budget();
|
||||
const portableKey = /^(?:core\.(?:repositoryformatversion|filemode|bare|logallrefupdates|ignorecase|precomposeunicode|symlinks|sparsecheckout|sparsecheckoutcone)|index\.sparse|extensions\.(?:objectformat|refstorage|partialclone)|remote\..+\.(?:url|pushurl|fetch|promisor|partialclonefilter)|branch\..+\.(?:remote|merge|rebase)|user\.(?:name|email))$/u;
|
||||
if (!keys.every(key => portableKey.test(key))) fail();
|
||||
}
|
||||
}
|
||||
return { sha256: digest.digest("hex"), byteSize, entries, exclude };
|
||||
}
|
||||
|
||||
export function readonlyWorkspaceInventoryScript() {
|
||||
const source = readWorkspaceInventory.toString().replaceAll("__vite_ssr_dynamic_import__", "import");
|
||||
return `const __name=value=>value;const inspect=${source};inspect(process.argv[1]).then(value=>process.stdout.write(JSON.stringify(value))).catch(()=>{process.stderr.write('remote_workspace_readonly_capture_unsafe');process.exitCode=1});`;
|
||||
}
|
||||
@@ -306,6 +306,10 @@ async function workspaceArchive(input: {
|
||||
};
|
||||
const directories: string[] = [];
|
||||
const excludedPaths = new Set<string>();
|
||||
for (const relative of input.exclude ?? []) {
|
||||
if (!safeRelative(relative)) fail("unsupported_filename");
|
||||
excludedPaths.add(relative);
|
||||
}
|
||||
const excluded = (next: string) => input.exclude?.some(item => next === item || next.startsWith(`${item}/`))
|
||||
|| next === ".paperclip-runtime" || next.startsWith(".paperclip-runtime/");
|
||||
const walk = async (root: string, skipGit = false) => {
|
||||
|
||||
@@ -6,6 +6,7 @@ import path from "node:path";
|
||||
import { pathToFileURL } from "node:url";
|
||||
import { promisify } from "node:util";
|
||||
import { deflateSync } from "node:zlib";
|
||||
import net from "node:net";
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import type { CommandManagedRuntimeRunner } from "@paperclipai/adapter-utils/command-managed-runtime";
|
||||
import {
|
||||
@@ -18,6 +19,8 @@ import {
|
||||
type RemoteWorkspaceTarget,
|
||||
} from "./remote-workspace-transport.js";
|
||||
import { classifyNativeWorkspaceFailure } from "./native-runtime/native-workspace-failure.js";
|
||||
import { restoreRemoteWorkspaceArchive } from "./remote-workspace-snapshot-archive.js";
|
||||
import { readWorkspaceInventory, readonlyWorkspaceInventoryScript } from "./remote-workspace-readonly-inventory.js";
|
||||
|
||||
const exec = promisify(execFile);
|
||||
|
||||
@@ -468,6 +471,58 @@ describe("remote workspace transport safety with real sandbox commands", () => {
|
||||
.rejects.toThrow(/^remote_workspace_operation_timed_out: archive_helper \(configured timeout 300000ms\)$/);
|
||||
});
|
||||
|
||||
it.each(["helper", "archive"])("saves a full sandbox through read-only transfer when the %s cannot write", async failureStage => {
|
||||
await git(workspace, "init", "--initial-branch=main");
|
||||
await fs.writeFile(path.join(workspace, "result.txt"), "committed result\n");
|
||||
await git(workspace, "add", "."); await git(workspace, "commit", "-m", "Saved work");
|
||||
const head = (await git(workspace, "rev-parse", "HEAD")).stdout;
|
||||
await fs.writeFile(path.join(workspace, "result.txt"), "staged result\n");
|
||||
await git(workspace, "add", "result.txt");
|
||||
await fs.writeFile(path.join(workspace, "notes.txt"), "untracked result\n");
|
||||
await fs.symlink("notes.txt", path.join(workspace, "notes-link"));
|
||||
const preload = path.join(scratch, "full-sandbox.cjs");
|
||||
const failing = failureStage === "helper" ? String.raw`/archive\.cjs\.[^/]+\.tmp$/` : String.raw`/workspace\.pcws$/`;
|
||||
await fs.writeFile(preload, String.raw`const fs=require('node:fs/promises'),failing=${failing};
|
||||
const fail=()=>{throw Object.assign(new Error('No space left on device'),{code:'ENOSPC'});};
|
||||
const writeFile=fs.writeFile,open=fs.open;fs.writeFile=async(file,...args)=>{
|
||||
if(failing.test(String(file)))fail();return writeFile(file,...args);};
|
||||
fs.open=async(file,...args)=>{const handle=await open(file,...args);if(failing.test(String(file)))handle.writeFile=async()=>fail();return handle;};
|
||||
require('node:module').syncBuiltinESMExports();`);
|
||||
const { remote, execute } = target();
|
||||
const run = execute.getMockImplementation()!;
|
||||
execute.mockImplementation(input => run({ ...input, env: { ...input.env, NODE_OPTIONS: `--require ${preload}` } }));
|
||||
const transfer = remote.runner!.syncOut!;
|
||||
remote.runner!.syncOut = async (operations, options) => {
|
||||
for (const operation of operations) for (const file of operation.files) {
|
||||
if (file.kind === "directory") {
|
||||
expect(file.followSymlinks).toBe(false);
|
||||
expect(file.exclude).toContain(".paperclip-runtime");
|
||||
await fs.cp(file.sourcePath, file.targetPath, { recursive: true, verbatimSymlinks: true,
|
||||
filter: source => !file.exclude!.some(excluded => {
|
||||
const relative = path.relative(file.sourcePath, source).split(path.sep).join("/");
|
||||
return relative === excluded || relative.startsWith(`${excluded}/`);
|
||||
}) });
|
||||
} else return transfer(operations, options);
|
||||
}
|
||||
return { operations: operations.map(operation => ({ operationId: operation.operationId,
|
||||
filesTransferred: operation.files.length, bytesTransferred: 0 })) };
|
||||
};
|
||||
const saved = await captureRemoteWorkspaceArchive({ target: remote, remoteCwd: workspace, operationId: randomUUID() });
|
||||
try {
|
||||
expect(await fs.readFile(path.join(workspace, "result.txt"), "utf8")).toBe("staged result\n");
|
||||
expect(await fs.readFile(path.join(workspace, "notes.txt"), "utf8")).toBe("untracked result\n");
|
||||
await fs.rm(workspace, { recursive: true });
|
||||
const restored = path.join(scratch, "restored");
|
||||
await restoreRemoteWorkspaceArchive({ archivePath: saved.archivePath, targetDir: restored,
|
||||
expectedManifestDigest: saved.manifestDigest });
|
||||
expect((await git(restored, "rev-parse", "HEAD")).stdout).toBe(head);
|
||||
expect((await git(restored, "show", ":result.txt")).stdout).toBe("staged result\n");
|
||||
expect(await fs.readFile(path.join(restored, "notes.txt"), "utf8")).toBe("untracked result\n");
|
||||
expect(await fs.readlink(path.join(restored, "notes-link"))).toBe("notes.txt");
|
||||
await git(restored, "fsck", "--full");
|
||||
} finally { await saved.cleanup(); }
|
||||
});
|
||||
|
||||
it.each([
|
||||
{ step: "helper install", failing: String.raw`/archive\.cjs\.[^/]+\.tmp$/`, outputMode: "split" as const },
|
||||
{ step: "archive write", failing: String.raw`/workspace\.pcws$/`, outputMode: "split" as const },
|
||||
@@ -489,6 +544,7 @@ describe("remote workspace transport safety with real sandbox commands", () => {
|
||||
const { remote, execute } = target({}, outputMode);
|
||||
const run = execute.getMockImplementation()!;
|
||||
execute.mockImplementation(input => run({ ...input, env: { ...input.env, NODE_OPTIONS: `--require ${preload}` } }));
|
||||
remote.runner!.syncOut = async () => { throw new Error("synthetic transfer unavailable"); };
|
||||
const operationId = randomUUID();
|
||||
await expect(captureRemoteWorkspaceArchive({ target: remote, remoteCwd: workspace, operationId }))
|
||||
.rejects.toThrow(/^remote_workspace_sandbox_disk_full: the sandbox disk is full; free space in the sandbox before saving the workspace$/);
|
||||
@@ -498,6 +554,260 @@ describe("remote workspace transport safety with real sandbox commands", () => {
|
||||
expect(await fs.readFile(path.join(workspace, "result.txt"), "utf8")).toBe("synthetic saved result\n");
|
||||
});
|
||||
|
||||
it("preserves work beside a socket and records its exclusion during full-disk recovery", async () => {
|
||||
await fs.writeFile(path.join(workspace, "saved.txt"), "retained work\n");
|
||||
const socket = path.join(workspace, "ipc"), server = net.createServer();
|
||||
await new Promise<void>((resolve, reject) => { server.once("error", reject); server.listen(socket, resolve); });
|
||||
let saved: Awaited<ReturnType<typeof captureRemoteWorkspaceArchive>> | undefined;
|
||||
try {
|
||||
const { remote, execute } = target();
|
||||
execute.mockResolvedValueOnce({ exitCode: 1, stdout: "", stderr: "remote_workspace_sandbox_disk_full: the sandbox disk is full; free space in the sandbox before saving the workspace",
|
||||
timedOut: false, signal: null, pid: process.pid, startedAt: new Date().toISOString() });
|
||||
remote.runner!.syncOut = async operations => {
|
||||
const mapping = operations[0]!.files[0]!;
|
||||
await fs.cp(mapping.sourcePath, mapping.targetPath, { recursive: true, verbatimSymlinks: true,
|
||||
filter: file => !mapping.exclude!.includes(path.relative(mapping.sourcePath, file)) });
|
||||
return { operations: [] };
|
||||
};
|
||||
saved = await captureRemoteWorkspaceArchive({ target: remote, remoteCwd: workspace, operationId: randomUUID() });
|
||||
const bytes = await fs.readFile(saved.archivePath), length = bytes.readUInt32BE(6);
|
||||
expect(JSON.parse(bytes.subarray(10, 10 + length).toString()).excluded).toContain("ipc");
|
||||
const restored = path.join(scratch, "restored");
|
||||
await restoreRemoteWorkspaceArchive({ archivePath: saved.archivePath, targetDir: restored, expectedManifestDigest: saved.manifestDigest });
|
||||
expect(await fs.readdir(restored)).toEqual(["saved.txt"]);
|
||||
expect((await fs.lstat(socket)).isSocket()).toBe(true);
|
||||
} finally {
|
||||
await saved?.cleanup();
|
||||
await new Promise<void>((resolve, reject) => server.close(error => error ? reject(error) : resolve()));
|
||||
}
|
||||
});
|
||||
|
||||
it.each(["source changed", "copy changed", "extra file"])("rejects a read-only fallback when %s", async fault => {
|
||||
await fs.writeFile(path.join(workspace, "saved.txt"), "retained work\n");
|
||||
const { remote, execute } = target();
|
||||
execute.mockResolvedValueOnce({ exitCode: 1, stdout: "", stderr: "remote_workspace_sandbox_disk_full: the sandbox disk is full; free space in the sandbox before saving the workspace",
|
||||
timedOut: false, signal: null, pid: process.pid, startedAt: new Date().toISOString() });
|
||||
let copiedRoot: string | undefined;
|
||||
remote.runner!.syncOut = async operations => {
|
||||
const file = operations[0]!.files[0]!; copiedRoot = path.dirname(file.targetPath);
|
||||
await fs.cp(file.sourcePath, file.targetPath, { recursive: true, verbatimSymlinks: true });
|
||||
if (fault === "source changed") await fs.writeFile(path.join(workspace, "saved.txt"), "newer work\n");
|
||||
else await fs.writeFile(path.join(file.targetPath, fault === "extra file" ? "extra.txt" : "saved.txt"), "wrong copy\n");
|
||||
return { operations: [] };
|
||||
};
|
||||
await expect(captureRemoteWorkspaceArchive({ target: remote, remoteCwd: workspace, operationId: randomUUID() }))
|
||||
.rejects.toMatchObject({ message: "remote_workspace_sandbox_disk_full: the sandbox disk is full; free space in the sandbox before saving the workspace",
|
||||
cause: { message: "remote_workspace_readonly_capture_failed: stage=verify reason=source_changed transport=provider_error" } });
|
||||
expect(await fs.readFile(path.join(workspace, "saved.txt"), "utf8")).toBe(fault === "source changed" ? "newer work\n" : "retained work\n");
|
||||
expect(copiedRoot).toBeDefined();
|
||||
await expect(fs.access(copiedRoot!)).rejects.toMatchObject({ code: "ENOENT" });
|
||||
});
|
||||
|
||||
it.each(["python", "node"])("recovers a full sandbox without copying an untracked %s environment", async kind => {
|
||||
await git(workspace, "init", "--initial-branch=main");
|
||||
await fs.writeFile(path.join(workspace, "saved.txt"), "retained work\n");
|
||||
await git(workspace, "add", "saved.txt"); await git(workspace, "commit", "-m", "Saved work");
|
||||
const dependency = kind === "python" ? ".venv" : "node_modules";
|
||||
await fs.mkdir(path.join(workspace, dependency, "bin"), { recursive: true });
|
||||
await fs.writeFile(path.join(workspace, kind === "python" ? ".venv/pyvenv.cfg" : "package.json"), kind === "python" ? "home = /usr/bin\n" : "{}\n");
|
||||
await fs.symlink(process.execPath, path.join(workspace, dependency, "bin", "runtime"));
|
||||
const { remote, execute } = target();
|
||||
execute.mockResolvedValueOnce({ exitCode: 1, stdout: "", stderr: "remote_workspace_sandbox_disk_full: the sandbox disk is full; free space in the sandbox before saving the workspace",
|
||||
timedOut: false, signal: null, pid: process.pid, startedAt: new Date().toISOString() });
|
||||
remote.runner!.syncOut = async operations => {
|
||||
const mapping = operations[0]!.files[0]!;
|
||||
expect(mapping.exclude).toContain(dependency);
|
||||
await fs.cp(mapping.sourcePath, mapping.targetPath, { recursive: true, verbatimSymlinks: true,
|
||||
filter: file => !(mapping.exclude ?? []).some(excluded => {
|
||||
const relative = path.relative(mapping.sourcePath, file).split(path.sep).join("/");
|
||||
return relative === excluded || relative.startsWith(`${excluded}/`);
|
||||
}) });
|
||||
return { operations: [] };
|
||||
};
|
||||
const saved = await captureRemoteWorkspaceArchive({ target: remote, remoteCwd: workspace, operationId: randomUUID() });
|
||||
try {
|
||||
const restored = path.join(scratch, "restored");
|
||||
await restoreRemoteWorkspaceArchive({ archivePath: saved.archivePath, targetDir: restored, expectedManifestDigest: saved.manifestDigest });
|
||||
expect(await fs.readFile(path.join(restored, "saved.txt"), "utf8")).toBe("retained work\n");
|
||||
await expect(fs.access(path.join(restored, dependency))).rejects.toMatchObject({ code: "ENOENT" });
|
||||
expect(await fs.readlink(path.join(workspace, dependency, "bin", "runtime"))).toBe(process.execPath);
|
||||
} finally { await saved.cleanup(); }
|
||||
});
|
||||
|
||||
it("never excludes Git history because its metadata contains a virtualenv marker", async () => {
|
||||
await git(workspace, "init", "--initial-branch=main");
|
||||
await fs.writeFile(path.join(workspace, "saved.txt"), "retained work\n");
|
||||
await git(workspace, "add", "saved.txt"); await git(workspace, "commit", "-m", "Saved work");
|
||||
const commit = (await git(workspace, "rev-parse", "HEAD")).stdout;
|
||||
await fs.writeFile(path.join(workspace, ".git", "pyvenv.cfg"), "home = /usr/bin\n");
|
||||
const { remote, execute } = target();
|
||||
execute.mockResolvedValueOnce({ exitCode: 1, stdout: "", stderr: "remote_workspace_sandbox_disk_full: the sandbox disk is full; free space in the sandbox before saving the workspace",
|
||||
timedOut: false, signal: null, pid: process.pid, startedAt: new Date().toISOString() });
|
||||
remote.runner!.syncOut = async operations => {
|
||||
const mapping = operations[0]!.files[0]!;
|
||||
expect(mapping.exclude).not.toContain(".git");
|
||||
await fs.cp(mapping.sourcePath, mapping.targetPath, { recursive: true, verbatimSymlinks: true });
|
||||
return { operations: [] };
|
||||
};
|
||||
const saved = await captureRemoteWorkspaceArchive({ target: remote, remoteCwd: workspace, operationId: randomUUID() });
|
||||
try {
|
||||
const restored = path.join(scratch, "restored");
|
||||
await restoreRemoteWorkspaceArchive({ archivePath: saved.archivePath, targetDir: restored, expectedManifestDigest: saved.manifestDigest });
|
||||
expect((await git(restored, "rev-parse", "HEAD")).stdout).toBe(commit);
|
||||
expect(await fs.readFile(path.join(restored, "saved.txt"), "utf8")).toBe("retained work\n");
|
||||
} finally { await saved.cleanup(); }
|
||||
});
|
||||
|
||||
it("rejects excessive Git roots before parsing controller configuration", async () => {
|
||||
for (let index = 0; index < 65; index++) {
|
||||
const directory = path.join(workspace, `repo-${index}`, ".git"); await fs.mkdir(directory, { recursive: true });
|
||||
await fs.writeFile(path.join(directory, "config"), "");
|
||||
}
|
||||
const bin = path.join(scratch, "bin"), invoked = path.join(scratch, "git-invoked"); await fs.mkdir(bin);
|
||||
await fs.writeFile(path.join(bin, "git"), `#!/bin/sh\nprintf invoked > '${invoked}'\nexit 1\n`, { mode: 0o700 });
|
||||
vi.stubEnv("PATH", `${bin}${path.delimiter}${process.env.PATH}`);
|
||||
try {
|
||||
await expect(readWorkspaceInventory(workspace, true)).rejects.toThrow("remote_workspace_readonly_capture_budget");
|
||||
await expect(fs.access(invoked)).rejects.toMatchObject({ code: "ENOENT" });
|
||||
} finally { vi.unstubAllEnvs(); }
|
||||
});
|
||||
|
||||
it("bounds refspec work before controller normalization", async () => {
|
||||
await git(workspace, "init", "--initial-branch=main");
|
||||
await fs.appendFile(path.join(workspace, ".git", "config"), `[remote "origin"]\n${"fetch = +refs/heads/*:refs/remotes/origin/*\n".repeat(257)}`);
|
||||
await expect(readWorkspaceInventory(workspace, true)).rejects.toThrow("remote_workspace_readonly_capture_budget");
|
||||
});
|
||||
|
||||
it("cancels controller inventory before reading its source", async () => {
|
||||
const reason = new Error("verification deadline");
|
||||
await expect(readWorkspaceInventory(path.join(scratch, "missing"), true, AbortSignal.abort(reason))).rejects.toBe(reason);
|
||||
});
|
||||
|
||||
it("retains tracked environment files during full-disk recovery", async () => {
|
||||
await git(workspace, "init", "--initial-branch=main");
|
||||
await fs.mkdir(path.join(workspace, ".venv"));
|
||||
await fs.writeFile(path.join(workspace, ".venv", "pyvenv.cfg"), "tracked user data\n");
|
||||
await fs.writeFile(path.join(workspace, ".venv", "result.txt"), "must survive\n");
|
||||
await git(workspace, "add", ".venv"); await git(workspace, "commit", "-m", "Tracked environment data");
|
||||
const { remote, execute } = target();
|
||||
execute.mockResolvedValueOnce({ exitCode: 1, stdout: "", stderr: "remote_workspace_sandbox_disk_full: the sandbox disk is full; free space in the sandbox before saving the workspace",
|
||||
timedOut: false, signal: null, pid: process.pid, startedAt: new Date().toISOString() });
|
||||
const saved = await captureRemoteWorkspaceArchive({ target: remote, remoteCwd: workspace, operationId: randomUUID() });
|
||||
try {
|
||||
const restored = path.join(scratch, "restored");
|
||||
await restoreRemoteWorkspaceArchive({ archivePath: saved.archivePath, targetDir: restored, expectedManifestDigest: saved.manifestDigest });
|
||||
expect(await fs.readFile(path.join(restored, ".venv", "result.txt"), "utf8")).toBe("must survive\n");
|
||||
} finally { await saved.cleanup(); }
|
||||
});
|
||||
|
||||
it.skipIf(process.platform === "win32")("rejects a file replaced by a FIFO before inventory open without blocking", async () => {
|
||||
const file = path.join(workspace, "saved.txt"), fifo = path.join(scratch, "fifo"), preload = path.join(scratch, "replace-with-fifo.cjs");
|
||||
await fs.writeFile(file, "retained work\n"); await exec("mkfifo", [fifo]);
|
||||
await fs.writeFile(preload, `const fs=require('node:fs/promises');const lstat=fs.lstat;let replaced=false;
|
||||
fs.lstat=async(file,...rest)=>{const stat=await lstat(file,...rest);
|
||||
if(!replaced&&String(file)===${JSON.stringify(file)}&&stat.isFile()){replaced=true;
|
||||
await fs.rename(file,file+'.retained');await fs.rename(${JSON.stringify(fifo)},file);}return stat;};
|
||||
require('node:module').syncBuiltinESMExports();`);
|
||||
await expect(exec(process.execPath, ["--require", preload, "-e", readonlyWorkspaceInventoryScript(), workspace], { timeout: 3000 }))
|
||||
.rejects.toMatchObject({ code: 1, stderr: "remote_workspace_readonly_capture_unsafe" });
|
||||
expect(await fs.readFile(`${file}.retained`, "utf8")).toBe("retained work\n");
|
||||
});
|
||||
|
||||
it("rejects internal Git config symlinks before controller normalization can follow an include", async () => {
|
||||
await fs.mkdir(path.join(workspace, ".git"));
|
||||
const outside = path.join(scratch, "outside-config");
|
||||
await fs.writeFile(outside, "[user]\nname = outside\n");
|
||||
await fs.writeFile(path.join(workspace, "payload-config"), `[include]\npath = ${outside}\n`);
|
||||
await fs.symlink("../payload-config", path.join(workspace, ".git", "config"));
|
||||
await expect(readWorkspaceInventory(workspace, true)).rejects.toThrow("remote_workspace_readonly_capture_unsafe");
|
||||
expect(await fs.readFile(outside, "utf8")).toBe("[user]\nname = outside\n");
|
||||
});
|
||||
|
||||
it("saves read-only directories and removes only the private controller copy", async () => {
|
||||
const source = path.join(workspace, "readonly"), restored = path.join(scratch, "restored");
|
||||
await fs.mkdir(source); await fs.writeFile(path.join(source, "saved.txt"), "read-only work\n"); await fs.chmod(source, 0o555);
|
||||
const { remote, execute } = target();
|
||||
execute.mockResolvedValueOnce({ exitCode: 1, stdout: "", stderr: "remote_workspace_sandbox_disk_full: the sandbox disk is full; free space in the sandbox before saving the workspace",
|
||||
timedOut: false, signal: null, pid: process.pid, startedAt: new Date().toISOString() });
|
||||
const transfer = remote.runner!.syncOut!; let copy: string | undefined;
|
||||
remote.runner!.syncOut = (operations, options) => { copy = operations[0]!.files[0]!.targetPath; return transfer(operations, options); };
|
||||
let saved: Awaited<ReturnType<typeof captureRemoteWorkspaceArchive>> | undefined;
|
||||
try {
|
||||
saved = await captureRemoteWorkspaceArchive({ target: remote, remoteCwd: workspace, operationId: randomUUID() });
|
||||
expect((await fs.stat(source)).mode & 0o777).toBe(0o555);
|
||||
expect(await fs.readFile(path.join(source, "saved.txt"), "utf8")).toBe("read-only work\n");
|
||||
await expect(fs.access(copy!)).rejects.toMatchObject({ code: "ENOENT" });
|
||||
await restoreRemoteWorkspaceArchive({ archivePath: saved.archivePath, targetDir: restored, expectedManifestDigest: saved.manifestDigest });
|
||||
expect((await fs.stat(path.join(restored, "readonly"))).mode & 0o777).toBe(0o555);
|
||||
expect(await fs.readFile(path.join(restored, "readonly", "saved.txt"), "utf8")).toBe("read-only work\n");
|
||||
} finally {
|
||||
for (const directory of [source, path.join(restored, "readonly"), ...(copy ? [path.join(copy, "readonly")] : [])]) {
|
||||
await fs.chmod(directory, 0o700).catch(() => undefined);
|
||||
}
|
||||
await saved?.cleanup();
|
||||
if (copy) await fs.rm(path.dirname(copy), { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
it("retains initialized submodule workspaces when a full sandbox cannot normalize their Git pointers", async () => {
|
||||
const origin = await repository();
|
||||
await git(workspace, "init", "--initial-branch=main");
|
||||
await fs.writeFile(path.join(workspace, "result.txt"), "retain parent work\n");
|
||||
await git(workspace, "add", "result.txt"); await git(workspace, "commit", "-m", "Parent");
|
||||
await git(workspace, "-c", "protocol.file.allow=always", "submodule", "add", "--", origin, "sub");
|
||||
const before = (await git(workspace, "status", "--porcelain")).stdout;
|
||||
const { remote, execute } = target();
|
||||
execute.mockResolvedValueOnce({ exitCode: 1, stdout: "", stderr: "remote_workspace_sandbox_disk_full: the sandbox disk is full; free space in the sandbox before saving the workspace",
|
||||
timedOut: false, signal: null, pid: process.pid, startedAt: new Date().toISOString() });
|
||||
const syncOut = vi.fn(remote.runner!.syncOut!); remote.runner!.syncOut = syncOut;
|
||||
await expect(captureRemoteWorkspaceArchive({ target: remote, remoteCwd: workspace, operationId: randomUUID() }))
|
||||
.rejects.toThrow(/^remote_workspace_sandbox_disk_full:/);
|
||||
expect(syncOut).not.toHaveBeenCalled();
|
||||
expect((await git(workspace, "status", "--porcelain")).stdout).toBe(before);
|
||||
expect(await fs.readFile(path.join(workspace, "sub", "source.txt"), "utf8")).toBe("initial synthetic source\n");
|
||||
});
|
||||
|
||||
it.each(["Config", "objects/info/Alternates"])("rejects case aliases of Git administration paths before controller Git reads: %s", async relative => {
|
||||
await git(workspace, "init", "--initial-branch=main");
|
||||
const outside = path.join(scratch, "outside"); await fs.mkdir(outside);
|
||||
if (relative === "Config") {
|
||||
await fs.rename(path.join(workspace, ".git", "config"), path.join(workspace, ".git", "Config"));
|
||||
await fs.writeFile(path.join(workspace, ".git", "Config"), `[include]\npath = ${path.join(outside, "config")}\n`);
|
||||
} else await fs.writeFile(path.join(workspace, ".git", relative), `${outside}\n`);
|
||||
await expect(readWorkspaceInventory(workspace, true)).rejects.toThrow("remote_workspace_readonly_capture_unsafe");
|
||||
expect(await fs.readdir(outside)).toEqual([]);
|
||||
});
|
||||
|
||||
it("rejects nonportable Git config before Git can read a controller attributes file", async () => {
|
||||
await git(workspace, "init", "--initial-branch=main");
|
||||
await git(workspace, "config", "core.attributesFile", path.join(scratch, "outside-attributes"));
|
||||
await expect(readWorkspaceInventory(workspace, true)).rejects.toThrow("remote_workspace_readonly_capture_unsafe");
|
||||
});
|
||||
|
||||
it.each(["worktree pointer", "alternate store", "config include", "core worktree"])("retains a full sandbox with a %s instead of resolving controller paths", async fault => {
|
||||
const outside = path.join(scratch, "outside"); await fs.mkdir(outside);
|
||||
await fs.writeFile(path.join(outside, "sentinel"), "must stay outside\n");
|
||||
await fs.writeFile(path.join(workspace, "saved.txt"), "retained work\n");
|
||||
if (fault === "worktree pointer") await fs.writeFile(path.join(workspace, ".git"), `gitdir: ${outside}\n`);
|
||||
else {
|
||||
await git(workspace, "init", "--initial-branch=main");
|
||||
if (fault === "alternate store") await fs.writeFile(path.join(workspace, ".git", "objects", "info", "alternates"), `${outside}\n`);
|
||||
if (fault === "config include") await git(workspace, "config", "include.path", path.join(outside, "sentinel"));
|
||||
if (fault === "core worktree") await git(workspace, "config", "core.worktree", outside);
|
||||
}
|
||||
const { remote, execute } = target();
|
||||
execute.mockResolvedValueOnce({ exitCode: 1, stdout: "", stderr: "remote_workspace_sandbox_disk_full: the sandbox disk is full; free space in the sandbox before saving the workspace",
|
||||
timedOut: false, signal: null, pid: process.pid, startedAt: new Date().toISOString() });
|
||||
const transfer = remote.runner!.syncOut!;
|
||||
const syncOut = vi.fn(transfer); remote.runner!.syncOut = syncOut;
|
||||
await expect(captureRemoteWorkspaceArchive({ target: remote, remoteCwd: workspace, operationId: randomUUID() }))
|
||||
.rejects.toThrow(/^remote_workspace_sandbox_disk_full:/);
|
||||
if (["worktree pointer", "alternate store"].includes(fault)) expect(syncOut).not.toHaveBeenCalled();
|
||||
expect(await fs.readFile(path.join(workspace, "saved.txt"), "utf8")).toBe("retained work\n");
|
||||
expect(await fs.readdir(outside)).toEqual(["sentinel"]);
|
||||
expect(await fs.readFile(path.join(outside, "sentinel"), "utf8")).toBe("must stay outside\n");
|
||||
});
|
||||
|
||||
it("keeps the full-disk diagnostic exact", async () => {
|
||||
const { remote, execute } = target();
|
||||
execute.mockResolvedValueOnce({ exitCode: 1, stdout: "", timedOut: false, signal: null, pid: process.pid,
|
||||
@@ -507,13 +817,14 @@ describe("remote workspace transport safety with real sandbox commands", () => {
|
||||
.rejects.toThrow(/^remote_workspace_operation_failed: operation=[a-z_]+ exit=1$/);
|
||||
});
|
||||
|
||||
it.each(["split", "one-shot"] as const)("preserves a Git child disk failure with %s output without its private text", async outputMode => {
|
||||
it.each(["split", "one-shot"] as const)("preserves a Git child disk failure with %s output when read-only download is unavailable", async outputMode => {
|
||||
await git(workspace, "init", "--initial-branch=main");
|
||||
await fs.writeFile(path.join(workspace, "saved.txt"), "retained source bytes\n");
|
||||
const bin = path.join(scratch, "controlled-bin"); await fs.mkdir(bin);
|
||||
const gitWrapper = path.join(bin, "git");
|
||||
await fs.writeFile(gitWrapper, "#!/bin/sh\nprintf '%s' 'No space left on device: SYNTHETIC_PRIVATE_GIT_PATH' >&2\nexit 2\n", { mode: 0o700 });
|
||||
const { remote } = target({ PATH: `${bin}${path.delimiter}${process.env.PATH}` }, outputMode);
|
||||
remote.runner!.syncOut = async () => { throw new Error("synthetic transfer unavailable"); };
|
||||
await expect(captureRemoteWorkspaceArchive({ target: remote, remoteCwd: workspace, operationId: randomUUID() }))
|
||||
.rejects.toThrow(/^remote_workspace_sandbox_disk_full: the sandbox disk is full; free space in the sandbox before saving the workspace$/);
|
||||
expect(await fs.readFile(path.join(workspace, "saved.txt"), "utf8")).toBe("retained source bytes\n");
|
||||
@@ -567,6 +878,7 @@ describe("remote workspace transport safety with real sandbox commands", () => {
|
||||
const diskFull = "remote_workspace_sandbox_disk_full: the sandbox disk is full; free space in the sandbox before saving the workspace";
|
||||
execute.mockResolvedValueOnce({ exitCode: 1, stdout: diskFull, stderr: "remote_workspace_archive_fs_enoent",
|
||||
timedOut: false, signal: null, pid: process.pid, startedAt: new Date().toISOString() });
|
||||
remote.runner!.syncOut = async () => { throw new Error("synthetic transfer unavailable"); };
|
||||
await expect(captureRemoteWorkspaceArchive({ target: remote, remoteCwd: workspace, operationId: randomUUID() }))
|
||||
.rejects.toThrow(diskFull);
|
||||
});
|
||||
|
||||
@@ -1,11 +1,13 @@
|
||||
import { createHash, randomUUID } from "node:crypto";
|
||||
import { createReadStream } from "node:fs";
|
||||
import { mkdtemp, rm, stat } from "node:fs/promises";
|
||||
import { mkdtemp, realpath, rm, stat } from "node:fs/promises";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import type { AdapterExecutionTarget } from "@paperclipai/adapter-utils/execution-target";
|
||||
import type { RemoteWorkspaceSource } from "@paperclipai/shared";
|
||||
import { remoteWorkspaceArchiveScript, remoteWorkspaceCaptureFailureMessage, remoteWorkspaceFailureDetail } from "./remote-workspace-snapshot-archive.js";
|
||||
import { captureRemoteWorkspaceDirectory, remoteWorkspaceArchiveScript, remoteWorkspaceCaptureFailureMessage, remoteWorkspaceFailureDetail } from "./remote-workspace-snapshot-archive.js";
|
||||
import { readWorkspaceInventory, readonlyWorkspaceInventoryScript } from "./remote-workspace-readonly-inventory.js";
|
||||
import { removeTaskDirectoryTree } from "./isolated-task-directory-snapshot.js";
|
||||
import type { GitRemoteAuthProvider } from "./git-credentials.js";
|
||||
import { classifyWorkspaceRestoreFailure } from "@paperclipai/adapter-utils/workspace-restore-merge";
|
||||
import { MAX_SANDBOX_SYNC_TIMEOUT_MS, type SandboxSyncOperation } from "@paperclipai/adapter-utils/sandbox-managed-runtime";
|
||||
@@ -398,9 +400,96 @@ export async function cleanupRemoteWorkspaceArchive(input: {
|
||||
await command(input.target, "cleanup", ["-e", script, input.remoteCwd, input.operationId], undefined, 10_000);
|
||||
}
|
||||
|
||||
/**
|
||||
* Wall-clock milliseconds for each save phase, filled in as each phase ends so a
|
||||
* failed save still reports the phases it finished. It holds only durations and
|
||||
* archive counts, never paths or provider text.
|
||||
*/
|
||||
export type RemoteWorkspaceSavePhaseTiming = {
|
||||
/** Archive helper install plus in-sandbox capture. */
|
||||
captureMs?: number;
|
||||
/** Provider download of the captured archive. */
|
||||
transferMs?: number;
|
||||
/** Server-side re-hash of the downloaded archive. */
|
||||
verifyMs?: number;
|
||||
byteSize?: number;
|
||||
entries?: number;
|
||||
};
|
||||
|
||||
const elapsedSince = (started: number) => Math.max(0, Math.round(performance.now() - started));
|
||||
|
||||
export async function captureRemoteWorkspaceArchive(input: {
|
||||
target: RemoteWorkspaceTarget; remoteCwd: string; operationId: string; timing?: RemoteWorkspaceSavePhaseTiming;
|
||||
}) {
|
||||
try { return await captureSandboxWorkspaceArchive(input); }
|
||||
catch (error) {
|
||||
if (!(error instanceof Error) || error.message !== REMOTE_WORKSPACE_SANDBOX_DISK_FULL) throw error;
|
||||
// Disk exhaustion cannot be repaired by writing another archive to the
|
||||
// same sandbox. Transfer a verified standalone source without writing to
|
||||
// it, then normalize and seal it on the controller. Failed verification
|
||||
// retains the original disk-full failure and never permits source cleanup.
|
||||
try { return await captureReadOnlyWorkspaceArchive(input); }
|
||||
catch (cause) { throw new Error(error.message, { cause }); }
|
||||
}
|
||||
}
|
||||
|
||||
async function captureReadOnlyWorkspaceArchive(input: {
|
||||
target: RemoteWorkspaceTarget; remoteCwd: string; operationId: string;
|
||||
}) {
|
||||
if (!/^[a-f0-9-]{36}$/.test(input.operationId)) throw new Error("remote_workspace_operation_identity_invalid");
|
||||
const inspect = async (timeoutMs = 300_000) => {
|
||||
const value = JSON.parse(await command(input.target, "capture", ["-e", readonlyWorkspaceInventoryScript(), input.remoteCwd], undefined, timeoutMs)) as Awaited<ReturnType<typeof readWorkspaceInventory>>;
|
||||
if (!/^[a-f0-9]{64}$/.test(value.sha256) || !Number.isSafeInteger(value.byteSize) || value.byteSize < 0
|
||||
|| !Number.isSafeInteger(value.entries) || value.entries < 0 || !Array.isArray(value.exclude)
|
||||
|| value.exclude.length > 4096 || !value.exclude.every(item => typeof item === "string" && item.length > 0
|
||||
&& !/[\\\u0000-\u001f\u007f]/u.test(item) && !path.posix.isAbsolute(item)
|
||||
&& item.split("/").every(part => part !== "" && part !== "." && part !== ".."))
|
||||
|| !value.exclude.includes(".paperclip-runtime")) throw new Error("remote_workspace_readonly_capture_unsafe");
|
||||
return value;
|
||||
};
|
||||
let root: string | undefined;
|
||||
let stage: "inspect" | "transfer" | "verify" | "archive" = "inspect";
|
||||
try {
|
||||
const before = await inspect();
|
||||
root = await mkdtemp(path.join(os.tmpdir(), "paperclip-workspace-readonly-"));
|
||||
root = await realpath(root);
|
||||
const copy = path.join(root, "source"), archivePath = path.join(root, "workspace.pcws");
|
||||
stage = "transfer";
|
||||
await transferArchive(input.target, "syncOut", [{ operationId: input.operationId,
|
||||
files: [{ kind: "directory", sourcePath: input.remoteCwd, targetPath: copy,
|
||||
followSymlinks: false, preserveSourceEntries: true, exclude: before.exclude }] }], Math.max(1, before.byteSize));
|
||||
stage = "verify";
|
||||
const verificationStarted = performance.now();
|
||||
const verificationSignal = AbortSignal.timeout(120_000);
|
||||
const remaining = () => { verificationSignal.throwIfAborted(); return Math.max(1, Math.ceil(120_000 - (performance.now() - verificationStarted))); };
|
||||
const received = await readWorkspaceInventory(copy, true, verificationSignal);
|
||||
const same = (observed: typeof before) => observed.sha256 === before.sha256
|
||||
&& observed.byteSize === before.byteSize && observed.entries === before.entries;
|
||||
if (!same(received) || !same(await inspect(remaining()))) throw new Error("remote_workspace_readonly_capture_changed");
|
||||
stage = "archive";
|
||||
const receipt = await captureRemoteWorkspaceDirectory({ cwd: copy, archivePath, exclude: before.exclude, signal: verificationSignal });
|
||||
stage = "verify";
|
||||
if (!same(await inspect(remaining()))) throw new Error("remote_workspace_readonly_capture_changed");
|
||||
verificationSignal.throwIfAborted();
|
||||
// Only the sealed portable archive remains a candidate for durable upload.
|
||||
await removeTaskDirectoryTree(copy);
|
||||
const savedRoot = root;
|
||||
return { ...receipt, archivePath, cleanup: () => removeTaskDirectoryTree(savedRoot) };
|
||||
} catch (error) {
|
||||
if (root) await removeTaskDirectoryTree(root);
|
||||
const reason = error instanceof Error && error.message === "remote_workspace_readonly_capture_unsafe" ? "unsafe_source"
|
||||
: error instanceof Error && error.message === "remote_workspace_readonly_capture_changed" ? "source_changed"
|
||||
: error instanceof Error && error.message === "remote_workspace_readonly_capture_budget" ? "budget_exceeded" : "operation_failed";
|
||||
const { label, code } = sandboxProviderFailureLabel(error);
|
||||
throw new Error(`remote_workspace_readonly_capture_failed: stage=${stage} reason=${reason} transport=${label}${code ? ` code=${code}` : ""}`);
|
||||
}
|
||||
}
|
||||
|
||||
async function captureSandboxWorkspaceArchive(input: {
|
||||
target: RemoteWorkspaceTarget; remoteCwd: string; operationId: string; timing?: RemoteWorkspaceSavePhaseTiming;
|
||||
}) {
|
||||
const timing = input.timing ?? {};
|
||||
const captureStarted = performance.now();
|
||||
const staged = await installArchiveHelper(input.target, input.remoteCwd, input.operationId);
|
||||
// Publish the archive and receipt as one immutable directory. Concurrent
|
||||
// invocations own separate attempts; no invocation unlinks another's archive.
|
||||
@@ -444,14 +533,20 @@ export async function captureRemoteWorkspaceArchive(input: {
|
||||
};
|
||||
if (!["legacy", "captured"].includes(layout) || !/^[a-f0-9]{64}$/.test(receipt.manifestDigest) || !/^[a-f0-9]{64}$/.test(receipt.sha256)
|
||||
|| !Number.isSafeInteger(receipt.byteSize) || receipt.byteSize <= 0) throw new Error("remote_workspace_capture_receipt_invalid");
|
||||
timing.captureMs = elapsedSince(captureStarted); timing.byteSize = receipt.byteSize;
|
||||
if (Number.isSafeInteger(receipt.entries) && receipt.entries >= 0) timing.entries = receipt.entries;
|
||||
const root = await mkdtemp(path.join(os.tmpdir(), "paperclip-workspace-upload-"));
|
||||
const archivePath = path.join(root, "workspace.pcws");
|
||||
try {
|
||||
const transferStarted = performance.now();
|
||||
await transferArchive(input.target, "syncOut", [{ operationId: input.operationId,
|
||||
files: [{ kind: "file", sourcePath: layout === "captured" ? path.posix.join(staged.directory, "captured", "workspace.pcws") : staged.archive, targetPath: archivePath }],
|
||||
}], receipt.byteSize);
|
||||
timing.transferMs = elapsedSince(transferStarted);
|
||||
const verifyStarted = performance.now();
|
||||
const hash = createHash("sha256"); let bytes = 0;
|
||||
for await (const chunk of createReadStream(archivePath)) { hash.update(chunk); bytes += chunk.length; }
|
||||
timing.verifyMs = elapsedSince(verifyStarted);
|
||||
if (hash.digest("hex") !== receipt.sha256 || bytes !== receipt.byteSize) throw new Error("remote_workspace_capture_transfer_mismatch");
|
||||
return { ...receipt, archivePath, cleanup: () => rm(root, { recursive: true, force: true }) };
|
||||
} catch (error) {
|
||||
|
||||
@@ -1105,6 +1105,52 @@ suite("remote workspace lifecycle with real Postgres and independent filesystem
|
||||
expect(operation?.issueId).toBe(secondIssue);
|
||||
}, 60_000);
|
||||
|
||||
it("durably saves a full sandbox and restores its work after losing the original allocation", async () => {
|
||||
const reserved = await reserve({ kind: "empty" }), remote = await target("full-sandbox");
|
||||
const prepared = await prepareRemoteWorkspace({ db, storage, workspace: reserved, runId: ids.run, lease: await lease(), target: remote });
|
||||
await fs.writeFile(path.join(prepared.remoteRoot, "saved.txt"), "completed work before restart\n");
|
||||
const execute = remote.runner!.execute.bind(remote.runner);
|
||||
let fullDiskFailures = 0;
|
||||
vi.spyOn(remote.runner!, "execute").mockImplementation(input => {
|
||||
if (input.args?.some(arg => arg.endsWith("/archive.cjs"))) {
|
||||
fullDiskFailures++;
|
||||
return Promise.resolve({ exitCode: 1, stdout: "", stderr: "remote_workspace_sandbox_disk_full: the sandbox disk is full; free space in the sandbox before saving the workspace",
|
||||
timedOut: false, signal: null, pid: process.pid, startedAt: new Date().toISOString() });
|
||||
}
|
||||
return execute(input);
|
||||
});
|
||||
const syncOut = remote.runner!.syncOut!;
|
||||
remote.runner!.syncOut = async (operations, options) => {
|
||||
if (!operations.some(operation => operation.files.some(file => file.kind === "directory"))) return syncOut(operations, options);
|
||||
for (const operation of operations) for (const mapping of operation.files) {
|
||||
await fs.cp(mapping.sourcePath, mapping.targetPath, { recursive: true, verbatimSymlinks: true,
|
||||
filter: source => !(mapping.exclude ?? []).some(excluded => {
|
||||
const relative = path.relative(mapping.sourcePath, source).split(path.sep).join("/");
|
||||
return relative === excluded || relative.startsWith(`${excluded}/`);
|
||||
}) });
|
||||
}
|
||||
return { operations: [] };
|
||||
};
|
||||
const saved = await saveRemoteWorkspace({ db, storage, workspace: prepared.workspace, runId: ids.run,
|
||||
lease: prepared.lease, target: remote, boundary: "final", assertOwned: async () => undefined });
|
||||
expect(fullDiskFailures).toBe(1);
|
||||
expect(readRemoteWorkspaceAuthority((await workspace(reserved.id)).metadata)).toMatchObject({
|
||||
snapshotId: saved.snapshotId, dirty: false, writerRunId: null,
|
||||
});
|
||||
const [snapshot] = await db.select().from(executionWorkspaceSnapshots).where(eq(executionWorkspaceSnapshots.id, saved.snapshotId));
|
||||
expect(snapshot).toMatchObject({ status: "committed" });
|
||||
await fs.rm(remote.remoteCwd, { recursive: true, force: true });
|
||||
const load = createSavedWorkspaceLoader(db, storage);
|
||||
for (let reconnect = 0; reconnect < 2; reconnect++) {
|
||||
const restored = await load({ companyId: ids.company, workspaceId: reserved.id, snapshotId: saved.snapshotId });
|
||||
try { expect(await fs.readFile(path.join(restored.root, "saved.txt"), "utf8")).toBe("completed work before restart\n"); }
|
||||
finally { await restored.cleanup(); }
|
||||
}
|
||||
const operations = await db.select().from(workspaceOperations).where(and(eq(workspaceOperations.heartbeatRunId, ids.run),
|
||||
eq(workspaceOperations.phase, "remote_workspace_snapshot")));
|
||||
expect(operations.at(-1)).toMatchObject({ status: "succeeded" });
|
||||
}, 60_000);
|
||||
|
||||
it("retains the writer and previous snapshot when object storage fails", async () => {
|
||||
const reserved = await reserve({ kind: "empty" }), first = await target("outage-sandbox");
|
||||
const prepared = await prepareRemoteWorkspace({ db, storage, workspace: reserved, runId: ids.run, lease: await lease(), target: first });
|
||||
@@ -1115,9 +1161,26 @@ suite("remote workspace lifecycle with real Postgres and independent filesystem
|
||||
expect(readRemoteWorkspaceAuthority((await workspace(reserved.id)).metadata)).toMatchObject({ dirty: true, writerRunId: ids.run,
|
||||
snapshotId: readRemoteWorkspaceAuthority(prepared.workspace.metadata)!.snapshotId });
|
||||
expect((await fs.readdir(path.join(prepared.remoteRoot, ".paperclip-runtime", "workspace-snapshots"))).length).toBe(1);
|
||||
const saved = await saveRemoteWorkspace({ db, storage, workspace: await workspace(reserved.id), runId: ids.run,
|
||||
// Each settled save records where its time went: the failed upload keeps the
|
||||
// phases it finished, and the retry of the same boundary records every phase.
|
||||
const finalTiming = async () => (await db.select().from(workspaceOperations).where(and(
|
||||
eq(workspaceOperations.executionWorkspaceId, reserved.id), eq(workspaceOperations.phase, "remote_workspace_snapshot"),
|
||||
sql`${workspaceOperations.metadata}->>'boundary' = 'final'`))).map(save => save.metadata?.saveTiming);
|
||||
const duration = expect.any(Number), archive = { byteSize: expect.any(Number), entries: expect.any(Number) };
|
||||
expect(await finalTiming()).toEqual([{ schema: "paperclip.remote-workspace-save-timing.v1", outcome: "failed", totalMs: duration,
|
||||
captureMs: duration, transferMs: duration, verifyMs: duration, ...archive, at: expect.any(String) }]);
|
||||
const beforeSuccess = await workspace(reserved.id);
|
||||
const saved = await saveRemoteWorkspace({ db, storage, workspace: beforeSuccess, runId: ids.run,
|
||||
lease: prepared.lease, target: first, assertOwned: async () => undefined });
|
||||
expect(saved.generation).toBe(2);
|
||||
const [succeededTiming] = await finalTiming();
|
||||
// A lost-reply replay of the committed save keeps the measured record.
|
||||
await expect(saveRemoteWorkspace({ db, storage, workspace: beforeSuccess, runId: ids.run,
|
||||
lease: prepared.lease, target: first, assertOwned: async () => undefined })).resolves.toMatchObject({ snapshotId: saved.snapshotId });
|
||||
expect(await finalTiming()).toEqual([succeededTiming]);
|
||||
expect(await finalTiming()).toEqual([{ schema: "paperclip.remote-workspace-save-timing.v1", outcome: "succeeded", totalMs: duration,
|
||||
captureMs: duration, transferMs: duration, verifyMs: duration, uploadMs: duration, publishMs: duration, ...archive,
|
||||
at: expect.any(String) }]);
|
||||
expect(await fs.readdir(path.join(prepared.remoteRoot, ".paperclip-runtime", "workspace-snapshots"))).toEqual([]);
|
||||
}, 60_000);
|
||||
|
||||
@@ -1224,9 +1287,12 @@ suite("remote workspace lifecycle with real Postgres and independent filesystem
|
||||
.rejects.toThrow("remote_workspace_snapshot_operation_binding_changed");
|
||||
expect(upload).not.toHaveBeenCalled();
|
||||
const [held] = await db.select().from(workspaceOperations).where(eq(workspaceOperations.id, poisoned!.id));
|
||||
const { saveWork, ...originalMetadata } = held!.metadata!;
|
||||
const { saveWork, saveTiming, ...originalMetadata } = held!.metadata!;
|
||||
expect(originalMetadata).toEqual(poisoned!.metadata);
|
||||
expect(saveWork).toMatchObject({ activeIds: [], unknown: true, quiescedAt: expect.any(String) });
|
||||
// The refused invocation records only its own failed duration; it never captured or uploaded.
|
||||
expect(saveTiming).toEqual({ schema: "paperclip.remote-workspace-save-timing.v1", outcome: "failed",
|
||||
totalMs: expect.any(Number), at: expect.any(String) });
|
||||
expect(Number.isFinite(Date.parse((saveWork as { quiescedAt: string }).quiescedAt))).toBe(true);
|
||||
expect(await db.select().from(executionWorkspaceSnapshots).where(eq(executionWorkspaceSnapshots.sourceOperationId, poisoned!.id)))
|
||||
.toHaveLength(0);
|
||||
|
||||
@@ -28,6 +28,7 @@ import {
|
||||
captureRemoteWorkspaceArchive, cleanupRemoteWorkspaceArchive, prepareRemoteWorkspaceDirectory, prepareRemoteRepository, remoteWorkspaceRepositoryPath,
|
||||
RemoteWorkspaceOperationTimeoutError, RemoteWorkspacePreparationTimeoutError, remoteWorkspaceFailureCause,
|
||||
requireRemoteWorkspaceTarget, restoreRemoteWorkspaceDirectory, RemoteWorkspaceRestoreError, type RemoteWorkspaceRestoreStep,
|
||||
type RemoteWorkspaceSavePhaseTiming,
|
||||
} from "./remote-workspace-transport.js";
|
||||
import {
|
||||
beginWorkspaceSnapshot, getWorkspaceSnapshot, loadWorkspaceSnapshotArchive, uploadWorkspaceSnapshot,
|
||||
@@ -46,6 +47,16 @@ type Workspace = Pick<typeof executionWorkspaces.$inferSelect, "id" | "companyId
|
||||
type Transaction = Parameters<Parameters<Db["transaction"]>[0]>[0];
|
||||
const object = (value: unknown): Record<string, unknown> => value && typeof value === "object" && !Array.isArray(value) ? value as Record<string, unknown> : {};
|
||||
const SAVE_WORK_SCHEMA = "paperclip.remote-workspace-save-work.v1";
|
||||
const SAVE_TIMING_SCHEMA = "paperclip.remote-workspace-save-timing.v1";
|
||||
|
||||
/** Per-phase durations of the latest settled save invocation, kept in the operation's local metadata. */
|
||||
type RemoteWorkspaceSaveTiming = RemoteWorkspaceSavePhaseTiming & {
|
||||
/** Encrypt, hash, and store the archive (zero work when a prior upload is reused). */
|
||||
uploadMs?: number;
|
||||
/** Read the archive manifest and commit the saved authority. */
|
||||
publishMs?: number;
|
||||
reusedUpload?: boolean;
|
||||
};
|
||||
const RETIRED_SAVE_SCHEMA = "paperclip.remote-workspace-retired-save.v1";
|
||||
type SaveWork = { schema: typeof SAVE_WORK_SCHEMA; activeIds: string[]; unknown: boolean; quiescedAt: string | null };
|
||||
function readSaveWork(metadata: Record<string, unknown> | null | undefined): SaveWork | null {
|
||||
@@ -610,9 +621,15 @@ export function remoteWorkspaceSaveFailure(error: unknown): { code: string; caus
|
||||
return cause ? { code, cause } : { code };
|
||||
}
|
||||
|
||||
function saveTimingRecord(timing: RemoteWorkspaceSaveTiming & { startedAt: number }, outcome: "succeeded" | "failed", at: Date) {
|
||||
const { startedAt, ...phases } = timing;
|
||||
return { schema: SAVE_TIMING_SCHEMA, outcome, totalMs: Math.max(0, Math.round(performance.now() - startedAt)),
|
||||
...phases, at: at.toISOString() };
|
||||
}
|
||||
|
||||
async function settleSaveInvocation(input: { db: Db; workspace: Workspace; runId: string; epoch: number;
|
||||
location: NonNullable<RemoteWorkspaceAuthority["location"]>; operationId: string; invocationId: string;
|
||||
failure?: { code: string; cause?: string } }) {
|
||||
failure?: { code: string; cause?: string }; timing?: RemoteWorkspaceSaveTiming & { startedAt: number } }) {
|
||||
return input.db.transaction(async tx => {
|
||||
const workspace = await currentWorkspace(tx, input.workspace.companyId, input.workspace.id);
|
||||
const [operation] = await tx.select().from(workspaceOperations).where(and(eq(workspaceOperations.id, input.operationId),
|
||||
@@ -629,6 +646,7 @@ async function settleSaveInvocation(input: { db: Db; workspace: Workspace; runId
|
||||
const failed = Boolean(failure) && operation.status === "running" && !activeIds.length && !work.unknown;
|
||||
await tx.update(workspaceOperations).set({ metadata: { ...operation.metadata,
|
||||
saveWork: { ...work, activeIds, quiescedAt: activeIds.length ? null : now.toISOString() },
|
||||
...(input.timing ? { saveTiming: saveTimingRecord(input.timing, input.failure ? "failed" : "succeeded", now) } : {}),
|
||||
...(failed && failure ? { lastFailure: { ...failure, at: now.toISOString() } } : {}) },
|
||||
...(failed && failure ? { status: "failed", exitCode: 1, finishedAt: now,
|
||||
stderrExcerpt: `${[failure.code, failure.cause].filter(Boolean).join("\n")}\n` } : {}),
|
||||
@@ -644,6 +662,7 @@ export async function saveRemoteWorkspace(input: {
|
||||
assertOwned: () => Promise<void>; storage?: StorageService; boundary?: string;
|
||||
repositories?: NonNullable<RemoteWorkspaceSource["repositories"]>;
|
||||
}) {
|
||||
const saveStarted = performance.now();
|
||||
const target = requireRemoteWorkspaceTarget(input.target), boundary = input.boundary ?? "final";
|
||||
const expected = readRemoteWorkspaceAuthority(input.workspace.metadata);
|
||||
if (!expected?.location || expected.location.providerLeaseId !== input.lease.providerLeaseId || expected.location.leaseId !== input.lease.id) throw new Error("remote_workspace_save_binding_mismatch");
|
||||
@@ -696,21 +715,31 @@ export async function saveRemoteWorkspace(input: {
|
||||
let captured: Awaited<ReturnType<typeof captureRemoteWorkspaceArchive>> | null = null;
|
||||
let publicationScratch: string | undefined;
|
||||
let failure: { code: string; cause?: string } | undefined;
|
||||
const timing: RemoteWorkspaceSaveTiming & { startedAt: number } = { startedAt: saveStarted };
|
||||
// A replay of an already-committed save measures no phase; it keeps the
|
||||
// original save's timing instead of overwriting it with an empty record.
|
||||
let replayedCommittedSave = false;
|
||||
try {
|
||||
const snapshot = await beginWorkspaceSnapshot(input.db, { companyId: input.workspace.companyId, workspaceId: input.workspace.id,
|
||||
operationId: operation.id, runId: input.runId, parentSnapshotId: operation.metadata?.parentSnapshotId as string | null,
|
||||
generation: operation.metadata?.generation as number, epoch: expected.epoch });
|
||||
if (operation.status === "succeeded" && ["committed", "superseded"].includes(snapshot.status)) {
|
||||
replayedCommittedSave = true;
|
||||
return { snapshotId: snapshot.id, generation: snapshot.generation, sha256: snapshot.sha256!, savedAt: snapshot.updatedAt.toISOString() };
|
||||
}
|
||||
const existingUpload = snapshot.status === "uploaded" ? await getWorkspaceSnapshot(input.db, { companyId: input.workspace.companyId,
|
||||
workspaceId: input.workspace.id, snapshotId: snapshot.id }) : null;
|
||||
captured = existingUpload ? null : await captureRemoteWorkspaceArchive({ target, remoteCwd: expected.location.remoteCwd, operationId: operation.id });
|
||||
if (existingUpload) timing.reusedUpload = true;
|
||||
captured = existingUpload ? null : await captureRemoteWorkspaceArchive({ target, remoteCwd: expected.location.remoteCwd,
|
||||
operationId: operation.id, timing });
|
||||
await input.assertOwned();
|
||||
const uploadStarted = performance.now();
|
||||
const saved = await uploadWorkspaceSnapshot({ db: input.db, storage: input.storage ?? getStorageService(), snapshotId: snapshot.id,
|
||||
companyId: input.workspace.companyId, workspaceId: input.workspace.id, archivePath: captured?.archivePath ?? "",
|
||||
manifestDigest: existingUpload?.manifestDigest ?? captured!.manifestDigest });
|
||||
timing.uploadMs = Math.max(0, Math.round(performance.now() - uploadStarted));
|
||||
await input.assertOwned();
|
||||
const publishStarted = performance.now();
|
||||
let publicationArchive = captured?.archivePath;
|
||||
if (!publicationArchive) {
|
||||
publicationScratch = await mkdtemp(path.join(os.tmpdir(), "pc-workspace-publication-"));
|
||||
@@ -756,6 +785,7 @@ export async function saveRemoteWorkspace(input: {
|
||||
deleteAfter: new Date(at.getTime() + 7 * 86400_000), updatedAt: at }).where(and(eq(executionWorkspaceSnapshots.id, snapshot.parentSnapshotId),
|
||||
eq(executionWorkspaceSnapshots.companyId, workspace.companyId), eq(executionWorkspaceSnapshots.executionWorkspaceId, workspace.id)));
|
||||
});
|
||||
timing.publishMs = Math.max(0, Math.round(performance.now() - publishStarted));
|
||||
return { snapshotId: saved.id, generation: saved.generation, sha256: saved.sha256, savedAt: at.toISOString() };
|
||||
} catch (error) {
|
||||
failure = remoteWorkspaceSaveFailure(error);
|
||||
@@ -768,7 +798,8 @@ export async function saveRemoteWorkspace(input: {
|
||||
catch { logger.warn({ runId: input.runId, operationId: operation.id }, "Remote workspace local archive cleanup failed"); }
|
||||
try {
|
||||
if (await settleSaveInvocation({ db: input.db, workspace: input.workspace, runId: input.runId, epoch: expected.epoch,
|
||||
location: expected.location, operationId: operation.id, invocationId, failure })) {
|
||||
location: expected.location, operationId: operation.id, invocationId, failure,
|
||||
timing: replayedCommittedSave ? undefined : timing })) {
|
||||
await cleanupSavedExports({ db: input.db, companyId: input.workspace.companyId, workspaceId: input.workspace.id,
|
||||
location: expected.location, target, assertOwned: input.assertOwned });
|
||||
}
|
||||
|
||||
@@ -33,7 +33,7 @@ import { workspaceOperationService } from "./workspace-operations.js";
|
||||
import { instanceSettingsService } from "./instance-settings.js";
|
||||
import { projectTaskInterruption, taskInterruptionPresentation } from "./task-interruption-presentation.js";
|
||||
import * as nativeWorkspaceSync from "./native-runtime/native-workspace-sync.js";
|
||||
import { registerCancelledWorkspacePreservationWork } from "./native-runtime/native-failed-workspace-recovery.js";
|
||||
import { registerCancelledWorkspacePreservationWork, preserveTaskRemoteWorkspaceWriter, preserveOrphanedNativeWorkspace } from "./native-runtime/native-failed-workspace-recovery.js";
|
||||
import { GitShallowMetadataError } from "@paperclipai/adapter-utils/git-workspace-sync";
|
||||
import { readSystemRecovery } from "./system-recovery-state.js";
|
||||
import { adoptSystemRecovery, updateSystemRecovery } from "./system-recovery.js";
|
||||
@@ -974,6 +974,126 @@ describe("automatic retained workspace recovery", () => {
|
||||
return { ...f, ...(sourceLayout === "published" ? { remote: remoteCwd } : {}), source, workspaceId, resultId, git, commit };
|
||||
}
|
||||
|
||||
async function exhaustedAcceptedRemoteWriter(leasePolicy: "ephemeral" | "reuse_by_environment" = "ephemeral") {
|
||||
const f = await acceptedRemoteExport();
|
||||
await db.update(issueRecoveryActions).set({ status: "cancelled", cause: "native_workspace_sync_out_failed",
|
||||
evidence: { runId: f.runId, automaticRecovery: { replay: "blocked", actionOutcome: "unknown" } },
|
||||
}).where(eq(issueRecoveryActions.id, f.actionId));
|
||||
await db.update(heartbeatRuns).set({ errorCode: "native_workspace_sync_out_retry_exhausted" })
|
||||
.where(eq(heartbeatRuns.id, f.runId));
|
||||
const [lease] = await db.update(environmentLeases).set({ leasePolicy })
|
||||
.where(eq(environmentLeases.id, f.lease.id)).returning();
|
||||
const stopping = (await prepareStoppedSandboxForRecovery(db, lease))!;
|
||||
const stopped = (await settleStopOnlyCleanup(db, stopping, {
|
||||
attemptId: String(stopping.metadata!.pendingCleanupAttemptId),
|
||||
receipt: { providerLeaseId: f.providerLeaseId, state: "stopped" },
|
||||
}))!;
|
||||
const source = await run(f.runId);
|
||||
const [preparation] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: source.agentId,
|
||||
issueId: f.issueId, runtimeMode: "legacy", status: "failed", executionStage: "preparing",
|
||||
startedAt: new Date(), finishedAt: new Date(), errorCode: "setup_failed",
|
||||
error: "The workspace still has an unfinished writer or save operation.",
|
||||
contextSnapshot: { issueId: f.issueId, executionWorkspaceId: f.workspaceId },
|
||||
runnerProfileJson: { adapterDispatch: { adapterType: "paperclip_runner" } },
|
||||
resultJson: { executionRecovery: { kind: "bootstrap", providerWorkStarted: false } },
|
||||
}).returning();
|
||||
return { ...f, stopped, preparation };
|
||||
}
|
||||
|
||||
it.each(["bound writer", "orphaned lease", "reusable writer"] as const)("retains an exhausted accepted remote copy-back through its %s", async entry => {
|
||||
const f = await exhaustedAcceptedRemoteWriter(entry === "reusable writer" ? "reuse_by_environment" : "ephemeral");
|
||||
const [result] = await db.select().from(nativeRunResults).where(eq(nativeRunResults.id, f.resultId));
|
||||
const [task] = await db.select().from(issues).where(eq(issues.id, f.issueId));
|
||||
const historical = await action(f.actionId);
|
||||
if (entry === "orphaned lease") expect(await preserveOrphanedNativeWorkspace(db, f.stopped)).toBe(true);
|
||||
else await preserveTaskRemoteWorkspaceWriter(db, f.companyId, f.issueId);
|
||||
expect((await run(f.runId)).resultJson).toMatchObject({ workspaceRestoreFailure: "restore_failed",
|
||||
workspaceRestoreRecovery: { leaseIds: [f.lease.id] } });
|
||||
expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0])
|
||||
.toMatchObject({ phase: "terminal_failure", resultId: result.id, leaseOwner: null });
|
||||
expect((await db.select().from(nativeRunResults).where(eq(nativeRunResults.id, f.resultId)))[0]).toEqual(result);
|
||||
expect(await action(f.actionId)).toEqual(historical);
|
||||
expect(await run(f.preparation.id)).toEqual(f.preparation);
|
||||
expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0]).toEqual(task);
|
||||
expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, f.companyId))).toHaveLength(2);
|
||||
expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId))).toHaveLength(0);
|
||||
expect(await readFile(path.join(f.remote, "saved.txt"), "utf8")).toBe("finished work\n");
|
||||
}, 30_000);
|
||||
|
||||
it.each(["active-finalizer", "retryable", "other-failure", "live-controller", "unfinished-run", "missing-stop",
|
||||
"rejected-result", "new-writer", "saved-workspace", "wrong-epoch", "native-sync", "source-binding"] as const)(
|
||||
"keeps accepted remote copy-back held when its %s proof changes", async gate => {
|
||||
const f = await exhaustedAcceptedRemoteWriter();
|
||||
const source = await run(f.runId);
|
||||
if (gate === "active-finalizer") await db.update(nativeRunFinalizations).set({ leaseOwner: "live-finalizer" })
|
||||
.where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
if (gate === "retryable") await db.update(nativeRunFinalizations).set({ phase: "retryable_failure" })
|
||||
.where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
if (gate === "other-failure") await db.update(nativeRunFinalizations).set({ failureCode: "native_workspace_sync_out_unsafe_archive" })
|
||||
.where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
if (gate === "live-controller") await db.update(heartbeatRuns).set({ controllerLeaseExpiresAt: new Date(Date.now() + 60_000) })
|
||||
.where(eq(heartbeatRuns.id, f.runId));
|
||||
if (gate === "unfinished-run") await db.update(heartbeatRuns).set({ finishedAt: null }).where(eq(heartbeatRuns.id, f.runId));
|
||||
if (gate === "missing-stop") await db.update(environmentLeases).set({ metadata: { ...f.stopped.metadata,
|
||||
remoteExecutionTermination: null } }).where(eq(environmentLeases.id, f.lease.id));
|
||||
if (gate === "rejected-result") await db.update(nativeRunResults).set({ schemaStatus: "rejected" })
|
||||
.where(eq(nativeRunResults.id, f.resultId));
|
||||
if (gate === "new-writer" || gate === "saved-workspace") {
|
||||
const [workspace] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, f.workspaceId));
|
||||
const authority = readRemoteWorkspaceAuthority(workspace.metadata)!;
|
||||
await db.update(executionWorkspaces).set({ metadata: { ...workspace.metadata, remoteWorkspace: { ...authority,
|
||||
writerRunId: gate === "new-writer" ? randomUUID() : null, dirty: gate === "new-writer" } } })
|
||||
.where(eq(executionWorkspaces.id, f.workspaceId));
|
||||
}
|
||||
if (gate === "wrong-epoch") await db.update(heartbeatRuns).set({ runnerProfileJson: { ...source.runnerProfileJson,
|
||||
remoteWorkspace: { ...(source.runnerProfileJson!.remoteWorkspace as object), epoch: 99 } } }).where(eq(heartbeatRuns.id, f.runId));
|
||||
if (gate === "native-sync") await db.update(heartbeatRuns).set({ runnerProfileJson: { ...source.runnerProfileJson,
|
||||
nativeWorkspaceSync: { schema: "paperclip.native-workspace-sync/v3" } } }).where(eq(heartbeatRuns.id, f.runId));
|
||||
if (gate === "source-binding") await db.update(environmentLeases).set({ metadata: { ...f.stopped.metadata, remoteCwd: "/other" } })
|
||||
.where(eq(environmentLeases.id, f.lease.id));
|
||||
const [lease] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id));
|
||||
const [result] = await db.select().from(nativeRunResults).where(eq(nativeRunResults.id, f.resultId));
|
||||
await preserveTaskRemoteWorkspaceWriter(db, f.companyId, f.issueId);
|
||||
expect((await run(f.runId)).resultJson).not.toHaveProperty("workspaceRestoreRecovery");
|
||||
expect((await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id)))[0]).toEqual(lease);
|
||||
expect((await db.select().from(nativeRunResults).where(eq(nativeRunResults.id, f.resultId)))[0]).toEqual(result);
|
||||
expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.companyId, f.companyId))).toHaveLength(1);
|
||||
expect(await readFile(path.join(f.remote, "saved.txt"), "utf8")).toBe("finished work\n");
|
||||
}, 30_000);
|
||||
|
||||
it("repairs an older accepted remote writer before one HTTP Retry continues a newer failed preparation", async () => {
|
||||
const f = await exhaustedAcceptedRemoteWriter();
|
||||
const [result] = await db.select().from(nativeRunResults).where(eq(nativeRunResults.id, f.resultId));
|
||||
await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.preparation.agentId, status: "running" });
|
||||
const receipt = (await request(receiptOnlyRecoveryApp()).post(`/api/issues/${f.issueId}/retry`)
|
||||
.send({ observedRunId: f.preparation.id, idempotencyKey: randomUUID() }).expect(202)).body;
|
||||
expect((await run(f.runId)).resultJson).toHaveProperty("workspaceRestoreRecovery");
|
||||
const records = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.companyId, f.companyId));
|
||||
const preservation = records.find(record => record.id !== f.actionId && record.evidence.runId === f.runId
|
||||
&& record.evidence.workspaceRestoreRecovery)!;
|
||||
expect(preservation).toBeDefined();
|
||||
const input = { ...f.input, actionId: preservation.id };
|
||||
await adoptSystemRecovery(db, input);
|
||||
await recoverRetainedWorkspace(input);
|
||||
expect(readWorkspaceRecovery((await action(preservation.id)).evidence)?.status).toBe("recovered");
|
||||
const heartbeat = heartbeatService(db);
|
||||
await Promise.all([heartbeat.resumeTaskRetry(f.companyId, receipt.requestId),
|
||||
heartbeat.resumeTaskRetry(f.companyId, receipt.requestId)]);
|
||||
const successors = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId),
|
||||
sql`${heartbeatRuns.contextSnapshot}->>'taskRetryRequestId' = ${receipt.requestId}`));
|
||||
expect(successors).toHaveLength(1);
|
||||
expect(successors[0]).toMatchObject({ status: "queued", contextSnapshot: { previousRunId: f.preparation.id } });
|
||||
const recovered = await run(f.runId), reference = recovered.runnerProfileJson!.remoteWorkspace as { snapshotId: string };
|
||||
const saved = await createSavedWorkspaceLoader(db)({ companyId: f.companyId, workspaceId: f.workspaceId,
|
||||
snapshotId: reference.snapshotId });
|
||||
try { expect(await readFile(path.join(saved.root, "saved.txt"), "utf8")).toBe("finished work\n"); }
|
||||
finally { await saved.cleanup(); }
|
||||
expect((await db.select().from(nativeRunResults).where(eq(nativeRunResults.id, f.resultId)))[0]).toEqual(result);
|
||||
expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0])
|
||||
.toMatchObject({ phase: "terminal_failure", resultId: result.id });
|
||||
expect(await run(f.preparation.id)).toEqual(f.preparation);
|
||||
}, 30_000);
|
||||
|
||||
async function pressuredToolCache(pressure: "transfer" | "sandbox-write" = "transfer") {
|
||||
const f = await acceptedRemoteExport();
|
||||
const runner = path.join(f.remote, ".paperclip-runtime", "paperclip-runner");
|
||||
@@ -998,6 +1118,7 @@ describe("automatic retained workspace recovery", () => {
|
||||
} }).where(eq(heartbeatRuns.id, f.runId));
|
||||
const target = targets.get(f.lease.id)!, syncOut = target.runner!.syncOut!, execute = target.runner!.execute;
|
||||
let transfers = 0, sandboxWriteFailures = 0;
|
||||
const readonlyExports: Array<Parameters<typeof syncOut>[0][number]["files"][number]> = [];
|
||||
if (pressure === "sandbox-write") target.runner!.execute = async input => {
|
||||
const fault = `(()=>{const fs=require('node:fs/promises'),sync=require('node:fs');
|
||||
const full=()=>${JSON.stringify([path.join(pack, "node_modules"), path.join(pack, "provider-assets")])}.some(p=>sync.existsSync(p));
|
||||
@@ -1015,13 +1136,21 @@ describe("automatic retained workspace recovery", () => {
|
||||
return response;
|
||||
};
|
||||
target.runner!.syncOut = async operations => {
|
||||
const mappings = operations.flatMap(operation => operation.files);
|
||||
if (mappings.some(mapping => mapping.kind === "directory")) {
|
||||
// These regressions isolate journaled cache reclamation after BOTH
|
||||
// export paths fail. Directory-export recovery has its own real-provider
|
||||
// coverage; this boundary explicitly cannot perform that alternative.
|
||||
readonlyExports.push(...mappings);
|
||||
throw new Error("cache-reclaim fixture directory export unavailable");
|
||||
}
|
||||
transfers++;
|
||||
if (pressure === "transfer" && (await Promise.all(["node_modules", "provider-assets"].map(name =>
|
||||
lstat(path.join(pack, name)).then(() => true, () => false)))).some(Boolean))
|
||||
throw new Error(REMOTE_WORKSPACE_SANDBOX_DISK_FULL);
|
||||
return syncOut(operations);
|
||||
};
|
||||
return { ...f, runner, pack, sessions, digest, transfers: () => transfers,
|
||||
return { ...f, runner, pack, sessions, digest, transfers: () => transfers, readonlyExports,
|
||||
sandboxWriteFailures: () => sandboxWriteFailures };
|
||||
}
|
||||
|
||||
@@ -1069,6 +1198,9 @@ describe("automatic retained workspace recovery", () => {
|
||||
await recoverRetainedWorkspace(f.input);
|
||||
expect(readWorkspaceRecovery((await action(f.actionId)).evidence)?.status).toBe("recovered");
|
||||
expect(f.transfers()).toBe(2);
|
||||
expect(f.readonlyExports).toEqual([{ kind: "directory", sourcePath: await realpath(f.remote),
|
||||
targetPath: expect.stringMatching(/[/]source$/), followSymlinks: false, preserveSourceEntries: true,
|
||||
exclude: [".paperclip-runtime"] }]);
|
||||
await expect(lstat(path.join(f.pack, "node_modules"))).rejects.toMatchObject({ code: "ENOENT" });
|
||||
await expect(lstat(path.join(f.pack, "provider-assets"))).rejects.toMatchObject({ code: "ENOENT" });
|
||||
expect(JSON.parse(await readFile(path.join(f.pack, "provider-pack.json"), "utf8")).digest).toBe(f.digest);
|
||||
@@ -1144,6 +1276,9 @@ describe("automatic retained workspace recovery", () => {
|
||||
const f = await pressuredToolCache(), target = targets.get(f.lease.id)!, syncOut = target.runner!.syncOut!;
|
||||
let transfers = 0;
|
||||
target.runner!.syncOut = async operations => {
|
||||
// Fail the second sealed-archive transfer, after cache cleanup, rather
|
||||
// than consuming this fault on the earlier read-only directory fallback.
|
||||
if (operations.some(operation => operation.files.some(mapping => mapping.kind === "directory"))) return syncOut(operations);
|
||||
if (++transfers === 2) throw new Error("controlled later transfer failure");
|
||||
return syncOut(operations);
|
||||
};
|
||||
|
||||
Reference in new issue
Block a user