mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-08 11:23:40 +01:00
fix(runtime): bound checkpoint data and preserve durable failed sessions (#18959)
## Thinking Path > - Paperclip manages AI agents and their work. > - Native Runner checkpoints preserve controller state across server replacements. > - Enrollment imports legacy controller state into those durable checkpoints. > - The importer passed extra source fields into each encrypted record, duplicating the whole source and exhausting memory. > - Some failed runs already had durable checkpoints but no session snapshot. Re-importing them incorrectly required resumable state. > - This PR stores only each record's binding and contents, and preserves matching failed checkpoints without changing their recovery status. ## Linked Issues or Issue Description Refs #18957 and #18943. **What happened?** A 538-record migrated checkpoint stored 1,108,851,510 bytes of encrypted material. Every record included the full 1.4 MB legacy source. Large writes disconnected and later reads exhausted memory. **Expected behavior** Each encrypted record contains only its company, issue, run, part and contents. **Steps to reproduce** Run offline controller migration with a retained controller that has many commands. Decrypt a saved envelope and inspect its fields. The old writer also includes the entire offline migration input. **Paperclip version or commit** Reproduced on `a1f3637c96dfadaf3a15ae9c83aa8227f4fed20e`. ## What Changed - Serialize only company, issue, run, part and contents in encrypted checkpoint envelopes. - Preserve an already-durable failed checkpoint when its run, runner and session match and the legacy session snapshot is absent. Keep its failure, recovery blockers, owner and pointers unchanged. - Report the number of failed checkpoints preserved during maintenance. - Test exact encrypted contents, idempotent preservation after removing the old home, and rejection of successful or foreign-session sources. - Document the migration findings and verification boundaries. ## Verification - The oversized-envelope regression failed before the serializer fix. - The already-durable failed-source regression failed before the preservation fix. - Four focused test files: 44 tests passed. The migration suite passed all 5 tests again after adding the foreign-session control. - Server typecheck passed. Greptile reviewed `c4f02a064c` at 5/5 with no findings. The exact-head PR workflow passed, including typecheck, build and all required tests. - Live staging repair preserved all 16 affected checkpoint sets: 2,982 records, 6,400,144,676 bytes reduced to 15,419,528 bytes. Each record matched its legacy source and passed re-encryption verification. Original encrypted rows and source files remain on the retained volume. - Full hosted enrollment and active-agent deployment proof remain pending. Deployment will follow the normal staging release pipeline. ## Risks The serializer affects normal and migrated checkpoint writes. The decoder already accepts the same five fields and continues to accept older envelopes. Preservation applies only to a failed run with an exact, validated durable checkpoint and no session snapshot; it grants no new recovery authority. This change does not rewrite existing database records automatically. Existing failures and missing provider files remain failures until their owning recovery path resolves them. ## Model Used OpenAI Codex, GPT-6 family, with reasoning, tool use and code execution. The exact deployment model ID and context-window size are not exposed in this session. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge
This commit is contained in:
1 parent
81b01bcc16
commit
55d7df2bad
5 files changed
+87
-3
No files matched your search
@@ -142,3 +142,29 @@ forces a conflict in a later batch, verifies that earlier batches roll back,
|
||||
then imports and decrypts the complete original checkpoint. The precise
|
||||
server-side reason for the connection closure is not yet established; the
|
||||
unbatched failures and batched comparison remain in the verification record.
|
||||
|
||||
## Checkpoint envelope expansion found during the live import
|
||||
|
||||
The encryption boundary spread the caller's entire input into every record.
|
||||
For offline imports that input includes the complete legacy controller source.
|
||||
One 538-record checkpoint stored 1,108,851,510 bytes of encrypted material,
|
||||
although the intended records total about 2.2 MB after encryption. This explains
|
||||
the oversized writes and subsequent memory exhaustion; the earlier SQL-shape
|
||||
probe passed a narrow binding and therefore did not reproduce the actual owner
|
||||
payload. The previous write failures and killed attempts remain in the evidence.
|
||||
|
||||
The checkpoint writer now selects company, issue, run, part and contents explicitly.
|
||||
The real database migration test decrypts each imported envelope and requires
|
||||
exactly those fields, with the original record content. It failed before the fix.
|
||||
Existing envelopes remain readable. For the staging imports from this attempt,
|
||||
a guarded offline repair backs up the encrypted rows on the retained volume,
|
||||
checks every record against its original legacy source, and rewrites only the
|
||||
envelope. Run, session, owner and task metadata remain unchanged.
|
||||
|
||||
The complete 807-source audit also found five failed runs with existing durable
|
||||
checkpoints and no provider-session snapshot. Their original controller state
|
||||
is already preserved. Enrollment now verifies the durable run, Runner and
|
||||
session binding and counts these failed states separately. It does not create
|
||||
a session snapshot or recovery authority, clear a blocker, or change their
|
||||
checkpoint rows, task pointers or failed status. A successful run with a missing
|
||||
snapshot still blocks enrollment.
|
||||
@@ -44,6 +44,7 @@ export async function migrateEphemeralServerData(db: Db, input: {
|
||||
counts.nativeControllersUnchanged = controllers.unchanged;
|
||||
counts.nativeControllerSourcesAlreadyAbsent = controllers.sourceAlreadyAbsent;
|
||||
counts.nativeControllersArchivedFailed = controllers.archivedFailed;
|
||||
counts.nativeControllersPreservedFailed = controllers.preservedFailed;
|
||||
phase("native_harness_backups");
|
||||
counts.nativeHarnessBackups = await migrateNativeHarnessBackups(db, { storage, writersStopped: true, signal });
|
||||
phase("workspaces");
|
||||
|
||||
@@ -14,12 +14,13 @@ 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 { createLocalEncryptedCipher } from "../../secrets/local-encrypted-provider.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.each(["import", "archive", "success", "batch"])("offline native controller migration (%s)", scenario => {
|
||||
describeEmbeddedPostgres.each(["import", "archive", "success", "batch", "durable_failed"])("offline native controller migration (%s)", scenario => {
|
||||
const environment = nativeRestartFixture();
|
||||
afterEach(() => vi.unstubAllEnvs());
|
||||
async function fixture(commandCount = 1) {
|
||||
@@ -155,11 +156,57 @@ describeEmbeddedPostgres.each(["import", "archive", "success", "batch"])("offlin
|
||||
await f.db.delete(nativeRunnerCheckpoints).where(eq(nativeRunnerCheckpoints.runId, f.runId));
|
||||
expect(await migrateNativeControllers(f.db, { writersStopped: true, signal: AbortSignal.timeout(30_000) }))
|
||||
.toMatchObject({ imported: 1 });
|
||||
const rows = await f.db.select().from(nativeRunnerCheckpoints).where(eq(nativeRunnerCheckpoints.runId, f.runId));
|
||||
const cipher = createLocalEncryptedCipher("read");
|
||||
for (const row of rows) expect(JSON.parse(cipher.decrypt(row.material))).toEqual({
|
||||
...binding, part: row.part, contents: checkpoint.records.get(row.part),
|
||||
});
|
||||
const restored = await loadNativeRunnerCheckpoint(f.db, binding);
|
||||
expect(restored?.records).toEqual(checkpoint.records);
|
||||
expect(await readFile(f.file, "utf8")).toBe(f.source);
|
||||
} finally { f.db.$client.options.debug = previousDebug; }
|
||||
}, 60_000);
|
||||
if (scenario === "durable_failed") it("preserves a failed durable checkpoint without inventing a missing session snapshot", async () => {
|
||||
const f = await fixture();
|
||||
const binding = { companyId: f.companyId, issueId: f.issueId, runId: f.runId };
|
||||
const checkpoint = captureLegacyDurablePrpCheckpoint(f.source, { runId: f.runId,
|
||||
runnerInstanceId: f.native.runnerInstanceId, normalizedSessionId: f.native.normalizedSessionId },
|
||||
{ version: "migration-fixture", digest: `sha256:${"a".repeat(64)}` });
|
||||
await f.db.transaction(tx => insertMigratedNativeRunnerCheckpoint(tx, binding, checkpoint));
|
||||
await f.db.update(heartbeatRuns).set({ processPid: 12345, runnerProfileJson: { ...f.profile, sessionCheckpoint: null } })
|
||||
.where(eq(heartbeatRuns.id, f.runId));
|
||||
const options = { writersStopped: true as const, signal: AbortSignal.timeout(30_000) };
|
||||
await f.db.update(heartbeatRuns).set({ status: "succeeded" }).where(eq(heartbeatRuns.id, f.runId));
|
||||
await expect(migrateNativeControllers(f.db, options)).rejects.toThrow("stale_supervisor_snapshot");
|
||||
await f.db.update(heartbeatRuns).set({ status: "failed" }).where(eq(heartbeatRuns.id, f.runId));
|
||||
const [run] = await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId));
|
||||
const [owner] = await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
const rows = await f.db.select().from(nativeRunnerCheckpoints).where(eq(nativeRunnerCheckpoints.runId, f.runId));
|
||||
const foreign = JSON.parse(f.source);
|
||||
foreign.identity.normalizedSessionId = randomUUID();
|
||||
const foreignCheckpoint = captureLegacyDurablePrpCheckpoint(JSON.stringify(foreign), { runId: f.runId,
|
||||
runnerInstanceId: f.native.runnerInstanceId, normalizedSessionId: foreign.identity.normalizedSessionId },
|
||||
{ version: "migration-fixture", digest: `sha256:${"a".repeat(64)}` });
|
||||
await f.db.delete(nativeRunnerCheckpoints).where(eq(nativeRunnerCheckpoints.runId, f.runId));
|
||||
await f.db.transaction(tx => insertMigratedNativeRunnerCheckpoint(tx, binding, foreignCheckpoint));
|
||||
expect(await hasLocalNativeControllerObligations(f.db)).toBe(true);
|
||||
await expect(migrateNativeControllers(f.db, options)).rejects.toThrow("stale_supervisor_snapshot");
|
||||
await f.db.delete(nativeRunnerCheckpoints).where(eq(nativeRunnerCheckpoints.runId, f.runId));
|
||||
await f.db.insert(nativeRunnerCheckpoints).values(rows);
|
||||
expect(await migrateNativeControllers(f.db, options)).toMatchObject({ preservedFailed: 1, imported: 0 });
|
||||
expect(await hasLocalNativeControllerObligations(f.db)).toBe(false);
|
||||
expect(await migrateNativeControllers(f.db, options)).toMatchObject({ preservedFailed: 1, imported: 0 });
|
||||
expect(await f.db.select().from(nativeRunnerCheckpoints).where(eq(nativeRunnerCheckpoints.runId, f.runId))).toEqual(rows);
|
||||
expect((await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0]).toEqual(run);
|
||||
expect((await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0]).toEqual(owner);
|
||||
expect((await f.db.select().from(agentTaskSessions).where(eq(agentTaskSessions.id, f.session.id)))[0].lastRunId).toBe(f.runId);
|
||||
expect((await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.queuedId)))[0].status).toBe("queued");
|
||||
await rm(f.paperclipHome, { recursive: true, force: true });
|
||||
expect((await loadNativeRunnerCheckpoint(f.db, binding))?.records).toEqual(checkpoint.records);
|
||||
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 === "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"));
|
||||
|
||||
@@ -68,6 +68,13 @@ async function localPaths(source: NonNullable<Awaited<ReturnType<typeof findNati
|
||||
return { ...descriptor, execution, controlFile: join(control, "control-plane-state.json") };
|
||||
}
|
||||
|
||||
function preservedFailedCheckpoint(source: NonNullable<Awaited<ReturnType<typeof findNativeSessionResumeRun>>>,
|
||||
sessionId: string, checkpoint: Awaited<ReturnType<typeof loadNativeRunnerCheckpoint>>) {
|
||||
return source.status === "failed" && source.runnerProfileJson.sessionCheckpoint == null
|
||||
&& checkpoint?.runId === source.id && checkpoint.recoveryEvidence.runnerInstanceId === source.runnerInstanceId
|
||||
&& checkpoint.recoveryEvidence.normalizedSessionId === sessionId;
|
||||
}
|
||||
|
||||
export async function hasLocalNativeControllerObligations(db: Db, storage?: StorageService | null): Promise<boolean> {
|
||||
const roots = await readdir(nativeHarnessStateBase()).catch((error: NodeJS.ErrnoException) => {
|
||||
if (error.code === "ENOENT") return [] as string[]; throw error;
|
||||
@@ -77,6 +84,8 @@ export async function hasLocalNativeControllerObligations(db: Db, storage?: Stor
|
||||
const binding = { companyId: selected.companyId, issueId: selected.issueId, runId: source.id };
|
||||
if (readNativeControllerRecovery(source.runnerProfileJson.nativeControllerRecovery, source.runnerProfileJson.sessionCheckpoint)
|
||||
&& await hasNativeRunnerCheckpoint(db, binding)) continue;
|
||||
if (source.status === "failed" && source.runnerProfileJson.sessionCheckpoint == null
|
||||
&& preservedFailedCheckpoint(source, selected.sessionId, await loadNativeRunnerCheckpoint(db, binding))) continue;
|
||||
const paths = await localPaths(source);
|
||||
const present = await lstat(paths.controlFile).catch((error: NodeJS.ErrnoException) => {
|
||||
if (error.code === "ENOENT") return undefined; throw error;
|
||||
@@ -96,11 +105,12 @@ export async function hasLocalNativeControllerObligations(db: Db, storage?: Stor
|
||||
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, archivedFailed: 0 };
|
||||
const counts = { inspected: 0, imported: 0, unchanged: 0, sourceAlreadyAbsent: 0, archivedFailed: 0, preservedFailed: 0 };
|
||||
for await (const { selected, source } of retainedSources(db, input.signal)) {
|
||||
counts.inspected++;
|
||||
const binding = { companyId: selected.companyId, issueId: selected.issueId, runId: source.id };
|
||||
const checkpoint = await loadNativeRunnerCheckpoint(db, binding);
|
||||
if (preservedFailedCheckpoint(source, selected.sessionId, checkpoint)) { counts.preservedFailed++; continue; }
|
||||
const metadata = readNativeControllerRecovery(source.runnerProfileJson.nativeControllerRecovery, source.runnerProfileJson.sessionCheckpoint);
|
||||
const snapshot = source.runnerProfileJson.sessionCheckpoint as { identity?: Record<string, unknown> } | null;
|
||||
if (checkpoint && metadata?.checkpoint?.runId === source.id
|
||||
|
||||
@@ -14,7 +14,7 @@ function encryptParts(binding: Binding, records: ReadonlyMap<string, string>) {
|
||||
const cipher = createLocalEncryptedCipher("write");
|
||||
return [...records].map(([part, contents]) => ({
|
||||
companyId: binding.companyId, issueId: binding.issueId, runId: binding.runId, part,
|
||||
material: cipher.encrypt(JSON.stringify({ ...binding, part, contents })),
|
||||
material: cipher.encrypt(JSON.stringify({ companyId: binding.companyId, issueId: binding.issueId, runId: binding.runId, part, contents })),
|
||||
}));
|
||||
}
|
||||
|
||||
|
||||
Reference in new issue
Block a user