mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-11 05:11:49 +01:00
fix(workspaces): refresh saved cohort source proofs on resume (#20429)
## Thinking Path > - Paperclip preserves agent work when a control-plane server is replaced. > - Online workspace imports save exact cohort, operation and snapshot identities for restart. > - A cohort proof also includes the complete current companion source scope. > - An independent companion import removes that companion from the local scope. > - The saved cohort then fails resume even when every saved member is unchanged. > - This pull request rechecks the current scope under fresh locks and refreshes its proof for all members in one transaction. > - The benefit is restartable online migration while independent work progresses. ## Linked Issues or Issue Description Refs: #19873, #20394, #20418. Related PR search found the original cohort owner, recovery-proof refresh and separate child-verification reuse. This fix addresses cohort scope freshness. **What happened?** An interrupted cohort rejects resume with `workspace_cohort_membership_changed` after a nonmember companion is migrated. The saved source proof includes companion workspace and task IDs. The exact saved members and their fingerprints can remain unchanged. **Expected behavior** The same saved cohort must resume after fresh admission. It must keep all member, operation and snapshot IDs. Changed members, active owners, changed journals and lost source locks must still refuse publication. **Steps to reproduce** 1. Start a two-member cohort and interrupt the second upload after the first archive is saved. 2. Import an archived nonmember companion through the existing canonical migration owner. 3. Resume the same two-member cohort. The original implementation refuses before it acquires the source fence. **Paperclip version or commit** Reproduced on master `3a5572c9e5`. **Deployment mode** Online administrative workspace migration with PostgreSQL and authenticated snapshot storage. ## What Changed - Admit a changed aggregate scope proof only when every saved member has a running journal and its current fingerprint matches both saved fingerprints. - Keep partial reservations on the original strict aggregate-proof path. Keep exact cohort and operation membership checks. - Run the existing source, lifecycle, recovery and finalizer admission before refresh. Lock member rows, inspect the current scope again, and compare every journal in one transaction. - Change only `cohortMigration.sourceProof`. Preserve all other metadata, identities, archive and recovery records. Abort the complete refresh on cancellation, a changed journal or a lost fence. - Extend the existing real PostgreSQL fixture with partial upload, canonical companion migration, refusal controls, transaction rollback, cold archive restore and replay. - Document the proof lifetime and resume boundary. ## Verification - Canonical Node24 worktree preflight passed. Plugin SDK build prerequisites passed. - Genuine RED on unchanged product: the real partial-upload and companion-import sequence fails with `workspace_cohort_membership_changed` (1 failed; 22 intentionally not selected; 10.44 seconds). - The identical positive passes after the fix (1 passed; 22 intentionally not selected; 9.87 seconds). - The complete focused cohort file passes: 23 tests in 33.33 seconds. Existing controls include changed member fingerprints, foreign and completed journals, active operations, peer cancellation, actual source-fence connection loss and outer-commit failure. - The new sequence also checks a held source lock, a held finalizer lock, cancellation before refresh, and an injected second-member update failure. The failed update rolls back every proof change. Both saved archives restore tracked and untracked work, and replay uses the same cohort. - After one final read-only preflight assertion, that complete new sequence passes again in 14.52 seconds with 22 existing cases intentionally not selected. Server TypeScript passes on the final files. Module boundaries and `git diff --check` pass. - Commands: `pnpm exec vitest run server/src/services/remote-workspace-cohort-migration.test.ts`; final case selection: `-t 'same uploaded cohort'`; `pnpm --filter @paperclipai/server exec tsc --noEmit`; `node scripts/check-module-boundaries.mjs`. - Initial local setup attempts remain nonpasses: sandbox PostgreSQL could not bind localhost, then the new archived companion fixture used a noncanonical repository folder. The corrected fixture uses the existing managed repository path. Neither setup failure is counted as the product RED. - Fresh [Greptile review](https://github.com/paperclipai/paperclip/pull/20429#issuecomment-6755862789) is 5/5 on exact head `6c640c4667b136e5416c1375f491729cf4bb9b49`, with zero inline threads. Required final-head CI is running. Hosted acceptance is not claimed by this PR. ## Risks - The old aggregate proof contains only a hash. Refresh therefore requires durable fingerprints for every saved member and repeats all current admission guards. A partially reserved cohort cannot use this path. - A stale proof must never hide a changed member or a new writer. Current source inspection, P/I/L/F locks, locked member rows, complete journal compare-and-swap and the existing capture and publication checks remain authoritative. - The all-member update must stay atomic. A database fault or cancellation rolls it back. The original cohort, operation and snapshot IDs remain fixed. - The change adds one bounded transaction only when a fully saved cohort's scope proof changes. It adds no recurring work, schema, provider, archive format, concurrency or deployment changes. ## Model Used OpenAI Codex, GPT-6 family, with repository tools and code execution. The runtime does not expose a more specific deployment identifier or context-window size. ## 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 - [ ] 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 Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
3a5572c9e5
commit
3040448da5
3 files changed
+140
-6
No files matched your search
@@ -145,6 +145,18 @@ Receipt validation adds bounded per-scope database reads to startup eligibility.
|
||||
It does not add a recurring scan or network work. Qualification must measure the
|
||||
selected cohort's startup latency on the serving revision before enrollment.
|
||||
|
||||
### Workspace cohort admission proofs
|
||||
|
||||
A saved cohort keeps its exact members, operations and snapshots on resume.
|
||||
Independent companion imports can change the current source-scope proof without
|
||||
changing those members. When every member has a running journal and its current
|
||||
fingerprint matches both saved fingerprints, the importer can reprove the current
|
||||
scope under fresh source, lifecycle and finalizer locks. After current source and
|
||||
recovery checks, one transaction compares every journal and refreshes only the
|
||||
cohort proof for all members. Cancellation or a changed journal rolls back the
|
||||
whole refresh. Partial reservations still require the original aggregate proof.
|
||||
Archives, recovery obligations and publication checks retain their existing owners.
|
||||
|
||||
### Original local Git sources for archived workspaces
|
||||
|
||||
When an archived runtime branch has an exact successful deletion receipt, its
|
||||
|
||||
@@ -6,7 +6,7 @@ import path from "node:path";
|
||||
import { promisify } from "node:util";
|
||||
import { eq, inArray, sql } from "drizzle-orm";
|
||||
import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { activityLog, agents, companies, createDb, environmentLeases, environments, heartbeatRuns, nativeRunFinalizations, executionWorkspaceSnapshots, executionWorkspaces, issues, projects, workspaceOperations, type Db } from "@paperclipai/db";
|
||||
import { activityLog, agents, companies, createDb, environmentLeases, environments, heartbeatRuns, nativeRunFinalizations, executionWorkspaceSnapshots, executionWorkspaces, issues, projects, projectWorkspaces, workspaceOperations, type Db } from "@paperclipai/db";
|
||||
import { readRemoteWorkspaceAuthority } from "@paperclipai/shared";
|
||||
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../__tests__/helpers/embedded-postgres.js";
|
||||
import { resolveManagedProjectWorkspaceDir } from "../home-paths.js";
|
||||
@@ -46,11 +46,11 @@ describe.skipIf(!support.supported)("bounded workspace cohort migration", () =>
|
||||
await db.execute(sql`drop trigger if exists cohort_commit_fault on workspace_operations`); await db.execute(sql`drop function if exists cohort_commit_fault()`);
|
||||
await db.execute(sql`drop trigger if exists cohort_reserve_fault on workspace_operations`); await db.execute(sql`drop function if exists cohort_reserve_fault()`);
|
||||
await db.delete(nativeRunFinalizations); await db.delete(environmentLeases); await db.delete(environments); await db.delete(heartbeatRuns);
|
||||
await db.delete(issues); await db.delete(agents); await db.delete(executionWorkspaces); await db.delete(projects); await db.delete(companies);
|
||||
await db.delete(issues); await db.delete(agents); await db.delete(executionWorkspaces); await db.delete(projectWorkspaces); await db.delete(projects); await db.delete(companies);
|
||||
vi.unstubAllEnvs(); await fs.rm(home, { recursive: true, force: true });
|
||||
});
|
||||
async function fixture() {
|
||||
const companyId = randomUUID(), projectId = randomUUID(), root = resolveManagedProjectWorkspaceDir({ companyId, projectId });
|
||||
async function fixture(repoName?: string) {
|
||||
const companyId = randomUUID(), projectId = randomUUID(), root = resolveManagedProjectWorkspaceDir({ companyId, projectId, repoName });
|
||||
await db.insert(companies).values({ id: companyId, name: "Cohort", issuePrefix: `C${companyId.slice(0, 6)}` });
|
||||
await db.insert(projects).values({ id: projectId, companyId, name: "Shared Git" });
|
||||
await fs.mkdir(root, { recursive: true }); await fs.writeFile(path.join(root, "saved.txt"), "committed base\n");
|
||||
@@ -200,6 +200,88 @@ describe.skipIf(!support.supported)("bounded workspace cohort migration", () =>
|
||||
await assertLocal(f);
|
||||
expect(await db.select().from(workspaceOperations)).toHaveLength(mode === "foreign" ? 3 : 2);
|
||||
});
|
||||
it("resumes the same uploaded cohort after an archived companion is independently migrated", async () => {
|
||||
const f = await fixture("cohort"), held = await recoverable(f), cwd = path.join(f.root, ".paperclip", "worktrees", "retired"), closedAt = new Date();
|
||||
await git(f.root, "worktree", "add", "-qb", "retired", cwd);
|
||||
await git(f.root, "worktree", "remove", cwd);
|
||||
const receipt = await git(f.root, "branch", "-d", "retired");
|
||||
const [projectWorkspace] = await db.insert(projectWorkspaces).values({ companyId: f.companyId, projectId: f.projectId,
|
||||
name: "Original repository", cwd: f.root, repoUrl: f.parent.repoUrl }).returning();
|
||||
const [issue] = await db.insert(issues).values({ companyId: f.companyId, projectId: f.projectId, title: "Retired companion" }).returning();
|
||||
const [companion] = await db.insert(executionWorkspaces).values({ companyId: f.companyId, projectId: f.projectId,
|
||||
projectWorkspaceId: projectWorkspace.id, sourceIssueId: issue.id, name: "retired", mode: "isolated_workspace",
|
||||
strategyType: "git_worktree", providerType: "git_worktree", cwd, providerRef: cwd, repoUrl: f.parent.repoUrl,
|
||||
branchName: "retired", status: "archived", closedAt, metadata: { createdByRuntime: true, gitBranchOwnershipVersion: 1 } }).returning();
|
||||
await db.insert(workspaceOperations).values({ companyId: f.companyId, executionWorkspaceId: companion.id,
|
||||
phase: "worktree_cleanup", status: "succeeded", command: "git branch -d retired", stdoutExcerpt: receipt,
|
||||
startedAt: new Date(), finishedAt: new Date() });
|
||||
const put = storage.putFile.bind(storage); let uploads = 0;
|
||||
const upload = vi.spyOn(storage, "putFile").mockImplementation(async input => {
|
||||
if (++uploads === 2) throw new Error("interrupted_cohort_upload");
|
||||
return put(input);
|
||||
});
|
||||
await expect(run(f, { concurrency: 1 })).rejects.toThrow("interrupted_cohort_upload");
|
||||
upload.mockRestore();
|
||||
const saved = (await db.select().from(workspaceOperations)).filter(row => row.phase === "remote_workspace_migration");
|
||||
expect(saved).toHaveLength(2);
|
||||
const cohort = saved[0].metadata!.cohortMigration as { id: string; sourceProof: string; members: unknown[] };
|
||||
const snapshots = await db.select().from(executionWorkspaceSnapshots);
|
||||
expect(snapshots.some(row => row.status === "uploaded")).toBe(true);
|
||||
await migrateRemoteWorkspace({ db, storage, companyId: f.companyId, workspaceId: companion.id, apply: true,
|
||||
signal: AbortSignal.timeout(50_000) });
|
||||
const noStorage = new Proxy(storage, { get() { throw new Error("unexpected_storage_operation"); } });
|
||||
await db.transaction(async tx => {
|
||||
await expect(run(f, { db: tx as unknown as Db, storage: noStorage, apply: false }))
|
||||
.resolves.toMatchObject({ applied: false, eligible: true, cohortId: cohort.id });
|
||||
}, { accessMode: "read only" });
|
||||
const other = createDb(database.connectionString, { maxConnections: 1 });
|
||||
try {
|
||||
await withWorkspaceSourceLocks(other, { companyId: f.companyId, projectIds: [f.projectId] }, async () => {
|
||||
await expect(run(f, { storage: noStorage })).rejects.toThrow("workspace_source_busy");
|
||||
}, true);
|
||||
await other.transaction(async tx => {
|
||||
expect(await tryLockNativeWorkspaceFinalization(tx as unknown as Db, { companyId: f.companyId, runId: held.runId })).toBe(true);
|
||||
await expect(run(f, { storage: noStorage })).rejects.toThrow("remote_copyback_owned");
|
||||
});
|
||||
} finally { await other.$client.end({ timeout: 1 }); }
|
||||
const inspect = onlineWorkspaceSource.inspectOnlineWorkspaceSource, cancelled = new AbortController(), calls = new Map<string, number>();
|
||||
const probe = vi.spyOn(onlineWorkspaceSource, "inspectOnlineWorkspaceSource").mockImplementation(async (...args) => {
|
||||
const result = await inspect(...args), count = (calls.get(args[1].id) ?? 0) + 1; calls.set(args[1].id, count);
|
||||
if (count === 5) cancelled.abort(new Error("cancel_cohort_proof_refresh"));
|
||||
return result;
|
||||
});
|
||||
await expect(run(f, { storage: noStorage, signal: cancelled.signal })).rejects.toThrow("cancel_cohort_proof_refresh");
|
||||
probe.mockRestore();
|
||||
const last = [...saved].sort((a, b) => a.executionWorkspaceId!.localeCompare(b.executionWorkspaceId!)).at(-1)!;
|
||||
await db.execute(sql.raw(`create function cohort_reserve_fault() returns trigger language plpgsql as $$ begin if new.id='${last.id}'::uuid then raise exception 'cohort_proof_refresh_fault'; end if; return new; end $$`));
|
||||
await db.execute(sql`create trigger cohort_reserve_fault before update on workspace_operations for each row execute function cohort_reserve_fault()`);
|
||||
await expect(run(f, { storage: noStorage })).rejects.toThrow();
|
||||
await db.execute(sql`drop trigger cohort_reserve_fault on workspace_operations`);
|
||||
await db.execute(sql`drop function cohort_reserve_fault()`);
|
||||
expect((await db.select().from(workspaceOperations)).filter(row => saved.some(old => old.id === row.id))).toEqual(saved);
|
||||
await assertLocal(f);
|
||||
await expect(run(f)).resolves.toMatchObject({ applied: true, cohortId: cohort.id });
|
||||
const current = (await db.select().from(workspaceOperations)).filter(row => saved.some(old => old.id === row.id));
|
||||
expect(current).toHaveLength(2);
|
||||
for (const operation of current) {
|
||||
const refreshed = operation.metadata!.cohortMigration as typeof cohort;
|
||||
expect(refreshed).toMatchObject({ id: cohort.id, members: cohort.members });
|
||||
expect(refreshed.sourceProof).not.toBe(cohort.sourceProof);
|
||||
expect(operation.status).toBe("succeeded");
|
||||
}
|
||||
expect((await db.select().from(executionWorkspaceSnapshots)).filter(row => snapshots.some(old => old.id === row.id))).toHaveLength(snapshots.length);
|
||||
await expect(run(f)).resolves.toMatchObject({ applied: true, replayed: true, cohortId: cohort.id });
|
||||
for (const workspace of f.leaves) {
|
||||
const [row] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, workspace.id));
|
||||
const ref = (await getWorkspaceSnapshot(db, { companyId: f.companyId, workspaceId: row.id,
|
||||
snapshotId: readRemoteWorkspaceAuthority(row.metadata)!.snapshotId! }))!;
|
||||
const archivePath = path.join(home, `${workspace.name}-resumed.pcws`), targetDir = path.join(home, `${workspace.name}-resumed`);
|
||||
await loadWorkspaceSnapshotArchive({ storage, reference: ref, targetPath: archivePath });
|
||||
await restoreRemoteWorkspaceArchive({ archivePath, targetDir, expectedManifestDigest: ref.manifestDigest });
|
||||
expect(await fs.readFile(path.join(targetDir, "saved.txt"), "utf8")).toBe(`${workspace.name} dirty tracked work\n`);
|
||||
expect(await fs.readFile(path.join(targetDir, "untracked.txt"), "utf8")).toBe(`${workspace.name} untracked work\n`);
|
||||
}
|
||||
}, 60_000);
|
||||
it("resumes the exact membership after reservation stops halfway", async () => {
|
||||
const f = await fixture(), second = [...f.leaves].sort((a, b) => a.id.localeCompare(b.id))[1];
|
||||
await db.execute(sql.raw(`create function cohort_reserve_fault() returns trigger language plpgsql as $$ begin if new.execution_workspace_id='${second.id}'::uuid then raise exception 'cohort_reservation_fault'; end if; return new; end $$`));
|
||||
|
||||
@@ -741,6 +741,7 @@ export async function migrateRemoteWorkspaceCohort(input: {
|
||||
const sourceProof = createHash("sha256").update(canonical({ scope: scopeIdentity(members[0].onlineSource),
|
||||
members: initial.map(workspace => ({ workspaceId: workspace.id, fingerprint: fingerprint(workspace) })) })).digest("hex");
|
||||
const pending = [];
|
||||
let unchangedMembers = true;
|
||||
for (const workspace of initial) {
|
||||
const marker = object(workspace.metadata?.remoteWorkspaceMigration);
|
||||
if (marker.schema !== SCHEMA || typeof marker.operationId !== "string") continue;
|
||||
@@ -749,11 +750,14 @@ export async function migrateRemoteWorkspaceCohort(input: {
|
||||
eq(workspaceOperations.id, marker.operationId), eq(workspaceOperations.phase, "remote_workspace_migration")));
|
||||
if (!operation?.metadata?.cohortMigration) throw new Error("workspace_cohort_foreign_journal");
|
||||
if (operation.status !== "running" || operation.finishedAt !== null) throw new Error("workspace_cohort_journal_changed");
|
||||
unchangedMembers &&= marker.sourceFingerprint === fingerprint(workspace)
|
||||
&& operation.metadata.sourceFingerprint === marker.sourceFingerprint;
|
||||
pending.push(cohortSchema.parse(operation.metadata.cohortMigration));
|
||||
}
|
||||
const cohort: WorkspaceMigrationCohort = pending[0] ?? { schema: "paperclip.remote-workspace-cohort.v1", id: randomUUID(), companyId,
|
||||
let cohort: WorkspaceMigrationCohort = pending[0] ?? { schema: "paperclip.remote-workspace-cohort.v1", id: randomUUID(), companyId,
|
||||
sourceProof, members: workspaceIds.map(workspaceId => ({ workspaceId, operationId: randomUUID() })) };
|
||||
if (cohort.companyId !== companyId || cohort.sourceProof !== sourceProof
|
||||
const refreshSourceProof = cohort.sourceProof !== sourceProof;
|
||||
if (cohort.companyId !== companyId || refreshSourceProof && (pending.length !== workspaceIds.length || !unchangedMembers)
|
||||
|| canonical(cohort.members.map(member => member.workspaceId)) !== canonical(workspaceIds)
|
||||
|| new Set(cohort.members.map(member => member.operationId)).size !== workspaceIds.length
|
||||
|| pending.some(marker => canonical(marker) !== canonical(cohort))) throw new Error("workspace_cohort_membership_changed");
|
||||
@@ -785,6 +789,42 @@ export async function migrateRemoteWorkspaceCohort(input: {
|
||||
return reservation.value;
|
||||
});
|
||||
await lockMigrationRecovery(lock, plans, cohort);
|
||||
if (refreshSourceProof) {
|
||||
const refreshedCohort = { ...cohort, sourceProof };
|
||||
const operations = await input.db.transaction(async tx => {
|
||||
await lockMigrationParents(tx, companyId, null, plans.flatMap(plan => [plan.reserved.workspace.sourceIssueId!,
|
||||
...plan.reserved.legacyRecoverySources.map(source => source.issueId),
|
||||
...plan.onlineSource!.companions.flatMap(proof => proof.sources.map(source => source.issueId))]));
|
||||
const locked = await tx.select().from(executionWorkspaces).where(and(eq(executionWorkspaces.companyId, companyId),
|
||||
inArray(executionWorkspaces.id, workspaceIds))).orderBy(executionWorkspaces.id).for("update");
|
||||
if (locked.length !== plans.length || plans.some(plan => {
|
||||
const workspace = locked.find(row => row.id === plan.input.workspaceId)!;
|
||||
return fingerprint(workspace) !== plan.reserved.sourceFingerprint
|
||||
|| canonical(workspace.metadata?.remoteWorkspaceMigration) !== canonical(plan.reserved.workspace.metadata?.remoteWorkspaceMigration);
|
||||
})) throw new Error("workspace_cohort_source_changed");
|
||||
await preflight(tx);
|
||||
const updated = [];
|
||||
for (const plan of plans) {
|
||||
signal.throwIfAborted();
|
||||
const operation = plan.reserved.operation!;
|
||||
const [current] = await tx.update(workspaceOperations).set({ metadata: { ...operation.metadata,
|
||||
cohortMigration: refreshedCohort }, updatedAt: new Date() }).where(and(
|
||||
eq(workspaceOperations.companyId, companyId), eq(workspaceOperations.executionWorkspaceId, plan.input.workspaceId),
|
||||
eq(workspaceOperations.id, operation.id), eq(workspaceOperations.phase, "remote_workspace_migration"),
|
||||
eq(workspaceOperations.status, "running"), sql`${workspaceOperations.finishedAt} is null`,
|
||||
sql`${workspaceOperations.issueId} is not distinct from ${operation.issueId}::uuid`,
|
||||
sql`${workspaceOperations.cwd} is not distinct from ${operation.cwd}`,
|
||||
eq(workspaceOperations.metadata, operation.metadata!))).returning();
|
||||
if (!current) throw new Error("workspace_cohort_journal_changed");
|
||||
updated.push(current);
|
||||
}
|
||||
await lock.execute(sql`select 1`);
|
||||
signal.throwIfAborted();
|
||||
return updated;
|
||||
});
|
||||
for (const [index, plan] of plans.entries()) plan.reserved.operation = operations[index];
|
||||
cohort = refreshedCohort;
|
||||
}
|
||||
const captures: Array<CapturedMigration | undefined> = Array(plans.length);
|
||||
const childVerifier = createMigrationChildVerifier(plans[0].input, plans[0].onlineSource, lock);
|
||||
let next = 0, failure: unknown;
|
||||
|
||||
Reference in new issue
Block a user