diff --git a/doc/agent-files.md b/doc/agent-files.md index 93a4e777649..765384f8fa0 100644 --- a/doc/agent-files.md +++ b/doc/agent-files.md @@ -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 diff --git a/server/src/__tests__/ephemeral-migration.test.ts b/server/src/__tests__/ephemeral-migration.test.ts index a4f816b8c9f..b1f335bc537 100644 --- a/server/src/__tests__/ephemeral-migration.test.ts +++ b/server/src/__tests__/ephemeral-migration.test.ts @@ -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); }); diff --git a/server/src/maintenance/ephemeral-migration-data.ts b/server/src/maintenance/ephemeral-migration-data.ts index f12f073ea6e..17c713b0f12 100644 --- a/server/src/maintenance/ephemeral-migration-data.ts +++ b/server/src/maintenance/ephemeral-migration-data.ts @@ -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 }; diff --git a/server/src/maintenance/ephemeral-migration-service.test.ts b/server/src/maintenance/ephemeral-migration-service.test.ts index 73bbe1672c1..4f11ca250cc 100644 --- a/server/src/maintenance/ephemeral-migration-service.test.ts +++ b/server/src/maintenance/ephemeral-migration-service.test.ts @@ -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; + 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); +}); diff --git a/server/src/maintenance/ephemeral-migration-service.ts b/server/src/maintenance/ephemeral-migration-service.ts index 95c57670fd6..e4cfd957790 100644 --- a/server/src/maintenance/ephemeral-migration-service.ts +++ b/server/src/maintenance/ephemeral-migration-service.ts @@ -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() { diff --git a/server/src/services/agent-instruction-working-copies.ts b/server/src/services/agent-instruction-working-copies.ts index 50e61014072..2d6414c8506 100644 --- a/server/src/services/agent-instruction-working-copies.ts +++ b/server/src/services/agent-instruction-working-copies.ts @@ -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(); const targetKey = (companyId: string, runId: string) => `${companyId}:${runId}`; const copySaves = new WeakMap>>(); +/** 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 { + 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(db: Db, companyId: string, runId: string, save: () => Promise): Promise { const scope = databaseWorkScopeKey(db); let active = copySaves.get(scope);