mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-03 01:27:09 +01:00
fix(server): close ephemeral enrollment safety gaps
Refuse unresolved local working copies, expire every due trace page through its owner, and enforce the export deadline after success. All 20 focused checks pass, including real PostgreSQL/S3 and the rebuilt executable. Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
dc450aa4bb
commit
6d31d1265d
6 files changed
+79
-6
No files matched your search
+5
-3
@@ -378,8 +378,9 @@ false until adoption completes.
|
||||
|
||||
The process uses the existing agent-file, run-evidence and skill maintenance
|
||||
owners, follows all pages, verifies stored agent bytes and skill versions, and
|
||||
checks whole-database eligibility. Active execution, missing local bytes, local
|
||||
workspaces and external instructions block enrollment. Those conditions must be
|
||||
checks whole-database eligibility. The trace owner expires all due retention
|
||||
pages before the final check. Active execution, unresolved local working copies,
|
||||
missing local bytes, local workspaces and external instructions block enrollment. Those conditions must be
|
||||
resolved by their owning migration paths before detaching the volume.
|
||||
|
||||
`GET /health` is a public maintenance liveness check. `GET /status` and
|
||||
@@ -388,7 +389,8 @@ only the attempt, stamped Core commit, phase, counts and safe error code. Export
|
||||
is available only after all checks succeed; it also returns the original decision
|
||||
signing key for transfer into the stack vault. The process never generates a
|
||||
replacement key. Every response disables caching. Never log or retain the export
|
||||
body outside the vault. A 25-minute deadline blocks export; SIGTERM stops the
|
||||
body outside the vault. A 25-minute deadline blocks export even after migration
|
||||
has succeeded; SIGTERM stops the
|
||||
process, with a ten-second exit deadline.
|
||||
|
||||
Cloud must verify the attempt and image commit, preserve the exact key, and
|
||||
|
||||
@@ -10,7 +10,7 @@ import { randomBytes, randomUUID } from "node:crypto";
|
||||
import { S3Client, CreateBucketCommand, DeleteBucketCommand, DeleteObjectsCommand, ListObjectsV2Command, PutObjectCommand } from "@aws-sdk/client-s3";
|
||||
import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { eq } from "drizzle-orm";
|
||||
import { agents, agentFileRoots, companies, createDb, closeFileStorageDb, heartbeatRuns } from "@paperclipai/db";
|
||||
import { activityLog, agents, agentFileRoots, agentInstructionWorkingCopies, companies, createDb, closeFileStorageDb, heartbeatRuns, providerTraceRecords } from "@paperclipai/db";
|
||||
import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
|
||||
import { migrateEphemeralServerData } from "../maintenance/ephemeral-migration-data.js";
|
||||
import { agentFileStore } from "../services/agent-file-store.js";
|
||||
@@ -42,6 +42,7 @@ describe.skipIf(!process.env.PAPERCLIP_TEST_S3_ENDPOINT)("existing-volume enroll
|
||||
await db.insert(companies).values({ id: companyId, name: "Enrollment fixture", issuePrefix: "MIG" });
|
||||
});
|
||||
afterEach(async () => {
|
||||
await db.delete(activityLog).where(eq(activityLog.companyId, companyId));
|
||||
await db.delete(heartbeatRuns).where(eq(heartbeatRuns.companyId, companyId));
|
||||
await db.delete(agents).where(eq(agents.companyId, companyId));
|
||||
await db.delete(companies).where(eq(companies.id, companyId));
|
||||
@@ -143,4 +144,35 @@ describe.skipIf(!process.env.PAPERCLIP_TEST_S3_ENDPOINT)("existing-volume enroll
|
||||
await expect(migrate()).rejects.toThrow();
|
||||
expect(await db.select().from(agentFileRoots).where(eq(agentFileRoots.agentId, row.id))).toEqual([]);
|
||||
});
|
||||
|
||||
it("retains unsaved terminal-run working copies until their owner records settlement", async () => {
|
||||
const row = await agent();
|
||||
const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId: row.id, status: "succeeded" }).returning();
|
||||
const privateRoot = path.join(home, "private-copy");
|
||||
await materializeInstructionBytes(privateRoot, "AGENTS.md", Buffer.from("Unsaved work"));
|
||||
await db.insert(agentInstructionWorkingCopies).values({ runId: run!.id, companyId, agentId: row.id,
|
||||
responsibleUserId: "fixture", entryFile: "AGENTS.md", baseHash: "fixture", localRoot: privateRoot,
|
||||
executionRoot: privateRoot, location: "local", state: "pending_collection" });
|
||||
await expect(migrate()).rejects.toThrow("local_instruction_copies_require_settlement");
|
||||
expect(await fs.readFile(path.join(privateRoot, "AGENTS.md"), "utf8")).toBe("Unsaved work");
|
||||
expect(await db.select().from(agentFileRoots).where(eq(agentFileRoots.agentId, row.id))).toEqual([]);
|
||||
// Simulate only the owning service's completed collection receipt. The
|
||||
// enrollment coordinator must not fabricate that save or erase the copy.
|
||||
await materializeInstructionBytes(row.root, "AGENTS.md", Buffer.from("Unsaved work"));
|
||||
await db.update(agentInstructionWorkingCopies).set({ state: "saved", processStoppedAt: new Date() })
|
||||
.where(eq(agentInstructionWorkingCopies.runId, run!.id));
|
||||
expect((await migrate()).eligibility.eligible).toBe(true);
|
||||
});
|
||||
|
||||
it("expires every page of local traces through their owner before final eligibility", async () => {
|
||||
const row = await agent();
|
||||
const runs = await db.insert(heartbeatRuns).values(Array.from({ length: 101 }, () => ({
|
||||
companyId, agentId: row.id, status: "succeeded",
|
||||
}))).returning();
|
||||
await db.insert(providerTraceRecords).values(runs.map(run => ({ companyId, runId: run.id, provider: "codex",
|
||||
traceRef: `${randomUUID()}.ndjson`, status: "complete", requestedBy: "fixture", expiresAt: new Date(0) })));
|
||||
expect(await migrate()).toMatchObject({ eligibility: { eligible: true }, counts: { expiredTraces: 101 } });
|
||||
expect((await db.select().from(providerTraceRecords).where(eq(providerTraceRecords.companyId, companyId)))
|
||||
.every(trace => trace.status === "expired" && trace.deletedAt !== null)).toBe(true);
|
||||
}, 120_000);
|
||||
});
|
||||
@@ -6,6 +6,8 @@ import { readEphemeralServerDataEligibility } from "../services/ephemeral-server
|
||||
import { migrateRunEvidence } from "../services/run-evidence-migration.js";
|
||||
import { skillInstallationMaintenance } from "../services/skill-installation-maintenance.js";
|
||||
import { skillPackageMaintenance } from "../services/skill-package-maintenance.js";
|
||||
import { hasUnsettledLocalInstructionCopies } from "../services/agent-instruction-working-copies.js";
|
||||
import { providerTraceStore } from "../services/provider-trace-store.js";
|
||||
|
||||
/** Runs only in the maintenance process after Cloud has confirmed all legacy
|
||||
* deployments stopped. Each owner retains its own migration/publication rules;
|
||||
@@ -18,6 +20,7 @@ export async function migrateEphemeralServerData(db: Db, input: {
|
||||
const [active] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns)
|
||||
.where(inArray(heartbeatRuns.status, ["running", "queued"])).limit(1);
|
||||
if (active) throw new Error("active_execution_requires_drain");
|
||||
if (await hasUnsettledLocalInstructionCopies(db)) throw new Error("local_instruction_copies_require_settlement");
|
||||
const before = await readEphemeralServerDataEligibility(db);
|
||||
const unsupported = before.blockers.filter(blocker => ![
|
||||
"agent_files_require_migration", "run_logs_require_migration", "provider_traces_require_migration", "skills_require_migration",
|
||||
@@ -90,6 +93,15 @@ export async function migrateEphemeralServerData(db: Db, input: {
|
||||
after = page.after;
|
||||
}
|
||||
phase("eligibility");
|
||||
// A trace can expire after the app stops or while other owners migrate.
|
||||
// Its owner tombstones and cleans all expired pages before final eligibility.
|
||||
const traces = providerTraceStore(db), expiryAt = new Date();
|
||||
for (;;) {
|
||||
signal.throwIfAborted();
|
||||
const removed = await traces.cleanupExpired(expiryAt, signal);
|
||||
counts.expiredTraces = (counts.expiredTraces ?? 0) + removed.length;
|
||||
if (removed.length < 100) break;
|
||||
}
|
||||
const eligibility = await readEphemeralServerDataEligibility(db);
|
||||
if (!eligibility.eligible) throw new Error("final_data_eligibility_failed");
|
||||
return { counts, eligibility };
|
||||
|
||||
@@ -60,3 +60,16 @@ it("publishes a blocked deadline immediately and never exports a late completion
|
||||
finish(result()); await service!.finished;
|
||||
expect((await request("/export", "POST")).status).toBe(409);
|
||||
});
|
||||
|
||||
it("expires successful export authority at the original maintenance deadline", async () => {
|
||||
let expired!: Promise<void>;
|
||||
const { request } = await start(async signal => {
|
||||
expired = new Promise(resolve => signal.addEventListener("abort", () => resolve(), { once: true }));
|
||||
return result();
|
||||
}, 100);
|
||||
await service!.finished;
|
||||
expect(service!.receipt().state).toBe("ready");
|
||||
await expired;
|
||||
expect(service!.receipt()).toMatchObject({ state: "blocked", failure: "maintenance_deadline_exceeded" });
|
||||
expect((await request("/export", "POST")).status).toBe(409);
|
||||
});
|
||||
@@ -62,7 +62,12 @@ export async function startEphemeralMigrationService(input: {
|
||||
const code = error instanceof Error ? error.message : "owner_migration_failed";
|
||||
failure = /^[a-z_]{1,80}$/.test(code) ? code : "owner_migration_failed";
|
||||
state = "blocked";
|
||||
} finally { clearTimeout(deadline); completedAt = new Date().toISOString(); }
|
||||
} finally {
|
||||
// Successful export authority expires at the original deadline too.
|
||||
// A failed migration has no export authority to retain.
|
||||
if (state !== "ready") clearTimeout(deadline);
|
||||
completedAt = new Date().toISOString();
|
||||
}
|
||||
})();
|
||||
return { server, finished, receipt,
|
||||
async close() {
|
||||
|
||||
@@ -11,7 +11,7 @@ import { execFile as execFileCallback } from "node:child_process";
|
||||
import { promisify } from "node:util";
|
||||
import fs from "node:fs/promises";
|
||||
import path from "node:path";
|
||||
import { and, asc, desc, eq, gt, inArray, lte, or, isNull, sql, type SQL } from "drizzle-orm";
|
||||
import { and, asc, desc, eq, gt, inArray, notInArray, lte, or, isNull, sql, type SQL } from "drizzle-orm";
|
||||
import { agents, heartbeatRuns, agentInstructionWorkingCopies as copies, databaseWorkScopeKey, type Db } from "@paperclipai/db";
|
||||
import {
|
||||
prepareAdapterExecutionTargetRuntime,
|
||||
@@ -41,6 +41,15 @@ const liveTargets = new Map<string, AdapterExecutionTarget>();
|
||||
const targetKey = (companyId: string, runId: string) => `${companyId}:${runId}`;
|
||||
const copySaves = new WeakMap<object, Map<string, Promise<void>>>();
|
||||
|
||||
/** A terminal run does not establish that its private local edits were saved.
|
||||
* Enrollment may discard these paths only after this owner's save/resolution
|
||||
* or explicit ownership transfer. Unavailable/conflicted copies fail closed. */
|
||||
export async function hasUnsettledLocalInstructionCopies(db: Db): Promise<boolean> {
|
||||
return (await db.select({ runId: copies.runId }).from(copies).where(and(
|
||||
eq(copies.location, "local"), notInArray(copies.state, ["saved", "unchanged", "resolved", "superseded"]),
|
||||
)).limit(1)).length > 0;
|
||||
}
|
||||
|
||||
function serializeCopySave<T>(db: Db, companyId: string, runId: string, save: () => Promise<T>): Promise<T> {
|
||||
const scope = databaseWorkScopeKey(db);
|
||||
let active = copySaves.get(scope);
|
||||
|
||||
Reference in new issue
Block a user