mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-05 03:51:54 +01:00
fix(server): archive failed legacy controllers without inventing recovery authority
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
5c2764634f
commit
3e82aeaa60
6 files changed
+138
-14
No files matched your search
@@ -27,6 +27,17 @@ files stay intact, canonical database checkpoints take precedence, and missing
|
||||
source files are reported separately rather than fabricated. This migration
|
||||
does not recover provider state that was already lost before enrollment.
|
||||
|
||||
For an already-failed run, a valid legacy controller can lack proof of the
|
||||
original Runner artifact after its old bootstrap tickets were pruned. The
|
||||
migration preserves those exact controller and supervisor bytes as an encrypted
|
||||
`paperclip.legacy-controller-archive.v1` object through the same backup ledger.
|
||||
Its scope is the hash of its manifest and is separate from the provider session
|
||||
scope. Publication requires a terminal-failure coordinator with no live owner.
|
||||
The archive does not become an executable Runner checkpoint. Task pointers,
|
||||
failed results and recovery blockers stay unchanged. A successful retained
|
||||
session without artifact proof still blocks migration. The migration receipt
|
||||
reports failed-session archives separately from imported recovery authorities.
|
||||
|
||||
### Proactive agent chat messages
|
||||
|
||||
Migration `0419_slippery_impossible_man.sql` adds an agent-scoped unique constraint
|
||||
|
||||
@@ -35,10 +35,11 @@ export async function migrateEphemeralServerData(db: Db, input: {
|
||||
const storage = getStorageService();
|
||||
if (storage.provider !== "s3") throw new Error("shared_object_storage_required");
|
||||
phase("native_controllers");
|
||||
const controllers = await migrateNativeControllers(db, { writersStopped: true, signal });
|
||||
const controllers = await migrateNativeControllers(db, { writersStopped: true, signal, storage });
|
||||
counts.nativeControllers = controllers.imported;
|
||||
counts.nativeControllersUnchanged = controllers.unchanged;
|
||||
counts.nativeControllerSourcesAlreadyAbsent = controllers.sourceAlreadyAbsent;
|
||||
counts.nativeControllersArchivedFailed = controllers.archivedFailed;
|
||||
phase("native_harness_backups");
|
||||
counts.nativeHarnessBackups = await migrateNativeHarnessBackups(db, { storage, writersStopped: true, signal });
|
||||
phase("workspaces");
|
||||
|
||||
@@ -0,0 +1,40 @@
|
||||
import { createHash } from "node:crypto";
|
||||
import { mkdtemp, writeFile } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import type { StorageService } from "../../storage/types.js";
|
||||
import { removeNativeHarnessBackup } from "./native-harness-backup-cleanup.js";
|
||||
import { persistNativeHarnessBackup, verifyDurableNativeHarnessBackup } from "./native-harness-backup-storage.js";
|
||||
|
||||
type Source = { companyId: string; issueId: string; runId: string; controller: string; supervisor?: string };
|
||||
const sha256 = (value: string) => createHash("sha256").update(value).digest("hex");
|
||||
|
||||
export function legacyNativeControllerArchive(source: Source) {
|
||||
const manifest = JSON.stringify({ schema: "paperclip.legacy-controller-archive.v1", companyId: source.companyId,
|
||||
issueId: source.issueId, runId: source.runId, reason: "original_runner_artifact_unproven",
|
||||
files: { "control-plane-state.json": sha256(source.controller),
|
||||
...(source.supervisor === undefined ? {} : { "supervisor.json": sha256(source.supervisor) }) } });
|
||||
return { manifest, scopeDigest: sha256(manifest), manifestSha256: `sha256:${sha256(manifest)}` };
|
||||
}
|
||||
|
||||
export async function hasArchivedLegacyNativeController(db: Db, input: Source & { storage: StorageService; signal?: AbortSignal }) {
|
||||
return verifyDurableNativeHarnessBackup({ db, storage: input.storage, companyId: input.companyId,
|
||||
...legacyNativeControllerArchive(input), signal: input.signal });
|
||||
}
|
||||
|
||||
export async function archiveFailedLegacyNativeController(db: Db, input: Source & {
|
||||
storage: StorageService; writersStopped: true; signal: AbortSignal;
|
||||
}) {
|
||||
if (await hasArchivedLegacyNativeController(db, input)) return;
|
||||
const descriptor = legacyNativeControllerArchive(input);
|
||||
const root = await mkdtemp(join(tmpdir(), "pc-legacy-controller-archive-"));
|
||||
try {
|
||||
await writeFile(join(root, "manifest.json"), descriptor.manifest, { mode: 0o600 });
|
||||
await writeFile(join(root, "control-plane-state.json"), input.controller, { mode: 0o600 });
|
||||
if (input.supervisor !== undefined) await writeFile(join(root, "supervisor.json"), input.supervisor, { mode: 0o600 });
|
||||
await persistNativeHarnessBackup({ db, storage: input.storage, root, scopeDigest: descriptor.scopeDigest, signal: input.signal,
|
||||
owner: { kind: "offline_migration", writersStopped: input.writersStopped, terminalFailureOnly: true,
|
||||
companyId: input.companyId, issueId: input.issueId, runId: input.runId } });
|
||||
} finally { await removeNativeHarnessBackup(root); }
|
||||
}
|
||||
@@ -3,7 +3,7 @@ import { mkdir, readFile, rm, writeFile } from "node:fs/promises";
|
||||
import { join } from "node:path";
|
||||
import { eq } from "drizzle-orm";
|
||||
import { afterEach, expect, it, vi } from "vitest";
|
||||
import { agentTaskSessions, heartbeatRuns, issues, nativeRunFinalizations, nativeRunnerCheckpoints } from "@paperclipai/db";
|
||||
import { agentTaskSessions, heartbeatRuns, issues, nativeHarnessBackups, nativeRunFinalizations, nativeRunnerCheckpoints } from "@paperclipai/db";
|
||||
import { DurablePrpControlPlane, parseNativeExecutionInput } from "../../vendor/paperclip-runner/index.js";
|
||||
import { describeEmbeddedPostgres, nativeRestartFixture } from "../../__tests__/helpers/native-restart.js";
|
||||
import { nativeToolContractFingerprintForTarget } from "./native-session-resume.js";
|
||||
@@ -13,11 +13,15 @@ import { resolveNativeControllerRecovery } from "./native-controller-recovery.js
|
||||
import { resolveRemoteProviderPackAuthority } from "./remote-provider-pack-authority.js";
|
||||
import { canonicalProviderPackJson } from "./remote-provider-pack-verification.js";
|
||||
import { nativeSha256 } from "./canonical.js";
|
||||
import { createLocalDiskStorageProvider } from "../../storage/local-disk-provider.js";
|
||||
import { createStorageService } from "../../storage/service.js";
|
||||
import { legacyNativeControllerArchive } from "./native-controller-archive.js";
|
||||
import { restoreDurableNativeHarnessBackup } from "./native-harness-backup-storage.js";
|
||||
|
||||
describeEmbeddedPostgres("offline native controller migration", () => {
|
||||
describeEmbeddedPostgres.each(["import", "archive", "success"])("offline native controller migration (%s)", scenario => {
|
||||
const environment = nativeRestartFixture();
|
||||
afterEach(() => vi.unstubAllEnvs());
|
||||
it("imports an exact settled authority once, preserves queued work, and resumes from an empty home", async () => {
|
||||
async function fixture() {
|
||||
vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", "73".repeat(32));
|
||||
const f = await environment.seedRun(randomUUID());
|
||||
const { companyId, agentId, paperclipHome, runtimeRoot } = environment.current;
|
||||
@@ -71,6 +75,11 @@ describeEmbeddedPostgres("offline native controller migration", () => {
|
||||
await f.db.update(issues).set({ executionRunId: queuedId }).where(eq(issues.id, f.issueId));
|
||||
const [session] = await f.db.insert(agentTaskSessions).values({ companyId, agentId, adapterType: "paperclip_runner",
|
||||
taskKey: f.issueId, lastRunId: f.runId }).returning();
|
||||
return { ...f, companyId, agentId, paperclipHome, runtimeRoot, paths, file, source, profile, session, queuedId, identity, snapshot };
|
||||
}
|
||||
if (scenario === "import") it("imports an exact settled authority once, preserves queued work, and resumes from an empty home", async () => {
|
||||
const f = await fixture();
|
||||
const { companyId, agentId, paperclipHome, runtimeRoot, profile, file, source, session, queuedId, identity, snapshot } = f;
|
||||
expect(await hasLocalNativeControllerObligations(f.db)).toBe(true);
|
||||
await f.db.update(heartbeatRuns).set({ runnerProfileJson: { ...profile,
|
||||
nativeWorkspaceSync: { ...profile.nativeWorkspaceSync, leaseId: randomUUID() } } }).where(eq(heartbeatRuns.id, f.runId));
|
||||
@@ -103,4 +112,45 @@ describeEmbeddedPostgres("offline native controller migration", () => {
|
||||
expect(JSON.parse(await readFile(join(replacement, "control-plane/control-plane-state.json"), "utf8")).identity).toEqual(identity);
|
||||
expect(recovered.snapshot).toEqual(snapshot);
|
||||
});
|
||||
if (scenario === "archive") it("preserves an already-failed legacy controller without inventing executable recovery authority", async () => {
|
||||
const f = await fixture();
|
||||
await rm(join(f.paths.root, "provider-pack-authority.json"));
|
||||
await f.db.update(nativeRunFinalizations).set({ recoveryState: "blocked", failureCode: "native_provider_terminal_failed" })
|
||||
.where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
const [originalRun] = await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId));
|
||||
const [originalOwner] = await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
const storage = createStorageService(createLocalDiskStorageProvider(join(f.runtimeRoot, "durable-objects")));
|
||||
const options = { storage, writersStopped: true as const, signal: AbortSignal.timeout(30_000) };
|
||||
const put = vi.spyOn(storage, "putFile").mockRejectedValue(new Error("storage unavailable"));
|
||||
await expect(migrateNativeControllers(f.db, options)).rejects.toThrow("storage unavailable");
|
||||
put.mockRestore();
|
||||
expect(await hasLocalNativeControllerObligations(f.db, storage)).toBe(true);
|
||||
expect(await migrateNativeControllers(f.db, options)).toMatchObject({ archivedFailed: 1, imported: 0 });
|
||||
expect(await migrateNativeControllers(f.db, options)).toMatchObject({ archivedFailed: 1, imported: 0 });
|
||||
expect(await hasLocalNativeControllerObligations(f.db, storage)).toBe(false);
|
||||
expect(await f.db.select().from(nativeRunnerCheckpoints).where(eq(nativeRunnerCheckpoints.runId, f.runId))).toHaveLength(0);
|
||||
expect((await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0]).toEqual(originalRun);
|
||||
expect((await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0]).toEqual(originalOwner);
|
||||
expect((await f.db.select().from(issues).where(eq(issues.id, f.issueId)))[0].executionRunId).toBe(f.queuedId);
|
||||
expect((await f.db.select().from(agentTaskSessions).where(eq(agentTaskSessions.id, f.session.id)))[0].lastRunId).toBe(f.runId);
|
||||
await rm(f.paperclipHome, { recursive: true, force: true });
|
||||
const descriptor = legacyNativeControllerArchive({ companyId: f.companyId, issueId: f.issueId, runId: f.runId, controller: f.source });
|
||||
const replacement = join(f.runtimeRoot, "archived-inspection");
|
||||
expect(await restoreDurableNativeHarnessBackup({ db: f.db, storage, companyId: f.companyId,
|
||||
scopeDigest: descriptor.scopeDigest, backupRoot: replacement })).toBe(true);
|
||||
expect(await readFile(join(replacement, "current", "control-plane-state.json"), "utf8")).toBe(f.source);
|
||||
const recovery = await resolveNativeControllerRecovery(f.db, { companyId: f.companyId, agentId: f.agentId,
|
||||
issueId: f.issueId, normalizedSessionId: f.native.normalizedSessionId, currentRunId: f.queuedId });
|
||||
expect(recovery.kind).not.toBe("recoverable");
|
||||
});
|
||||
if (scenario === "success") it("does not substitute a failure archive for a successful retained session", async () => {
|
||||
const f = await fixture();
|
||||
await rm(join(f.paths.root, "provider-pack-authority.json"));
|
||||
await f.db.update(heartbeatRuns).set({ status: "succeeded" }).where(eq(heartbeatRuns.id, f.runId));
|
||||
const storage = createStorageService(createLocalDiskStorageProvider(join(f.runtimeRoot, "durable-objects")));
|
||||
await expect(migrateNativeControllers(f.db, { storage, writersStopped: true, signal: AbortSignal.timeout(30_000) }))
|
||||
.rejects.toThrow("native_runner_checkpoint_artifact_unproven");
|
||||
expect(await f.db.select().from(nativeHarnessBackups)).toHaveLength(0);
|
||||
expect(await hasLocalNativeControllerObligations(f.db, storage)).toBe(true);
|
||||
});
|
||||
});
|
||||
@@ -13,6 +13,9 @@ import { nativeHarnessStateBase } from "./native-harness-backup-stamp.js";
|
||||
import { providerPackWorkspaceId, resolveRemoteProviderPackAuthority } from "./remote-provider-pack-authority.js";
|
||||
import { nativeSha256 } from "./canonical.js";
|
||||
import { readNativeWorkspaceSyncReference } from "./native-workspace-sync.js";
|
||||
import type { StorageService } from "../../storage/types.js";
|
||||
import { durableNativeHarnessStorage } from "./native-harness-backup-storage.js";
|
||||
import { archiveFailedLegacyNativeController, hasArchivedLegacyNativeController } from "./native-controller-archive.js";
|
||||
|
||||
type Selected = { id: string; companyId: string; agentId: string; issueId: string; sessionId: string; createdAt: Date };
|
||||
|
||||
@@ -65,7 +68,7 @@ async function localPaths(source: NonNullable<Awaited<ReturnType<typeof findNati
|
||||
return { ...descriptor, execution, controlFile: join(control, "control-plane-state.json") };
|
||||
}
|
||||
|
||||
export async function hasLocalNativeControllerObligations(db: Db): Promise<boolean> {
|
||||
export async function hasLocalNativeControllerObligations(db: Db, storage = durableNativeHarnessStorage()): Promise<boolean> {
|
||||
const roots = await readdir(nativeHarnessStateBase()).catch((error: NodeJS.ErrnoException) => {
|
||||
if (error.code === "ENOENT") return [] as string[]; throw error;
|
||||
});
|
||||
@@ -78,15 +81,21 @@ export async function hasLocalNativeControllerObligations(db: Db): Promise<boole
|
||||
const present = await lstat(paths.controlFile).catch((error: NodeJS.ErrnoException) => {
|
||||
if (error.code === "ENOENT") return undefined; throw error;
|
||||
});
|
||||
if (present && source.status === "failed" && storage) {
|
||||
const controller = await readPrivateSource(paths.controlFile, 256 * 1024 * 1024);
|
||||
const supervisor = paths.execution.session.lifecyclePolicy.mode === "warm"
|
||||
? await readPrivateSource(paths.supervisorPath, 4 * 1024 * 1024) : undefined;
|
||||
if (controller !== undefined && await hasArchivedLegacyNativeController(db, { ...binding, storage, controller, supervisor })) continue;
|
||||
}
|
||||
if (present || await hasNativeRunnerCheckpoint(db, binding)) return true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
export async function migrateNativeControllers(db: Db, input: { writersStopped: true; signal: AbortSignal }) {
|
||||
export async function migrateNativeControllers(db: Db, input: { writersStopped: true; signal: AbortSignal; storage?: StorageService }) {
|
||||
const [active] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(offlineMigrationExecutionWhere()).limit(1);
|
||||
if (!input.writersStopped || active) throw new Error("native_controller_migration_requires_offline_writers");
|
||||
const counts = { inspected: 0, imported: 0, unchanged: 0, sourceAlreadyAbsent: 0 };
|
||||
const counts = { inspected: 0, imported: 0, unchanged: 0, sourceAlreadyAbsent: 0, archivedFailed: 0 };
|
||||
for await (const { selected, source } of retainedSources(db, input.signal)) {
|
||||
counts.inspected++;
|
||||
const binding = { companyId: selected.companyId, issueId: selected.issueId, runId: source.id };
|
||||
@@ -129,10 +138,21 @@ export async function migrateNativeControllers(db: Db, input: { writersStopped:
|
||||
const supervisor = paths.execution.session.lifecyclePolicy.mode === "warm"
|
||||
? await readPrivateSource(paths.supervisorPath, 4 * 1024 * 1024) : undefined;
|
||||
input.signal.throwIfAborted();
|
||||
const result = await importNativeControllerRecovery(db, { ...binding, agentId: selected.agentId,
|
||||
normalizedSessionId: selected.sessionId, expectedConfigDigest: paths.configDigest,
|
||||
supervisor: supervisor === undefined ? undefined : JSON.parse(supervisor),
|
||||
offline: { writersStopped: true, legacyControllerState, legacyRunnerArtifact } });
|
||||
let result: Awaited<ReturnType<typeof importNativeControllerRecovery>>;
|
||||
try {
|
||||
result = await importNativeControllerRecovery(db, { ...binding, agentId: selected.agentId,
|
||||
normalizedSessionId: selected.sessionId, expectedConfigDigest: paths.configDigest,
|
||||
supervisor: supervisor === undefined ? undefined : JSON.parse(supervisor),
|
||||
offline: { writersStopped: true, legacyControllerState, legacyRunnerArtifact } });
|
||||
} catch (error) {
|
||||
const storage = input.storage ?? durableNativeHarnessStorage();
|
||||
if (!(error instanceof Error) || error.message !== "native_runner_checkpoint_artifact_unproven"
|
||||
|| source.status !== "failed" || legacyControllerState === undefined || !storage) throw error;
|
||||
await archiveFailedLegacyNativeController(db, { ...binding, storage, controller: legacyControllerState, supervisor,
|
||||
writersStopped: true, signal: input.signal });
|
||||
counts.archivedFailed++;
|
||||
continue;
|
||||
}
|
||||
if (result.kind !== "imported" && result.kind !== "unchanged") {
|
||||
throw new Error(`native_controller_migration_blocked:${source.id}:${"reason" in result ? result.reason : result.kind}`);
|
||||
}
|
||||
|
||||
@@ -24,7 +24,8 @@ type Reference = Binding & {
|
||||
material: Record<string, unknown>; manifestSha256: string;
|
||||
};
|
||||
type Row = typeof nativeHarnessBackups.$inferSelect;
|
||||
type OfflineOwner = { kind: "offline_migration"; companyId: string; issueId: string; runId: string; writersStopped: true };
|
||||
type OfflineOwner = { kind: "offline_migration"; companyId: string; issueId: string; runId: string;
|
||||
writersStopped: true; terminalFailureOnly?: true };
|
||||
const fail = (reason: string): never => { throw new Error(`native_harness_storage_${reason}`); };
|
||||
const aad = (binding: Binding) => Buffer.from(JSON.stringify([SCHEMA, binding.companyId, binding.scopeDigest, binding.id]));
|
||||
const scope = (binding: Pick<Binding, "companyId" | "scopeDigest">) => and(
|
||||
@@ -99,11 +100,12 @@ async function fence(tx: Parameters<Parameters<Db["transaction"]>[0]>[0], owner:
|
||||
eq(heartbeatRuns.id, owner.runId), eq(heartbeatRuns.companyId, owner.companyId),
|
||||
eq(heartbeatRuns.nativeIssueId, owner.issueId), eq(heartbeatRuns.runtimeMode, "native"))).for("share");
|
||||
const [coordinator] = await tx.select({ leaseOwner: nativeRunFinalizations.leaseOwner,
|
||||
leaseExpiresAt: nativeRunFinalizations.leaseExpiresAt })
|
||||
leaseExpiresAt: nativeRunFinalizations.leaseExpiresAt, phase: nativeRunFinalizations.phase })
|
||||
.from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, owner.runId)).for("share");
|
||||
const [clock] = await tx.execute<{ now: Date }>(sql`select clock_timestamp() as now`);
|
||||
if (!run?.finishedAt || !["succeeded", "failed", "cancelled", "interrupted", "timed_out"].includes(run.status)
|
||||
|| !coordinator || coordinator.leaseOwner && (!coordinator.leaseExpiresAt || coordinator.leaseExpiresAt > new Date(clock.now))) {
|
||||
|| !coordinator || coordinator.leaseOwner && (!coordinator.leaseExpiresAt || coordinator.leaseExpiresAt > new Date(clock.now))
|
||||
|| owner.terminalFailureOnly && (run.status !== "failed" || coordinator.phase !== "terminal_failure")) {
|
||||
fail("offline_owner_unsettled");
|
||||
}
|
||||
return;
|
||||
|
||||
Reference in new issue
Block a user