fix(skills): order saves and bind committed recovery sources

Repair orphaned recovery recipes in the source publication transaction, follow converted versions in summary reads, and keep dedicated package uploads from cycling behind a version save's skill lock.

Co-Authored-By: Paperclip <noreply@paperclip.ing>

server/src/__tests__/skill-package-store.test.ts

@@ -5,7 +5,7 @@ import { randomUUID } from "node:crypto";
 import { promises as fs } from "node:fs";
 import os from "node:os";
 import path from "node:path";
-import { and, eq } from "drizzle-orm";
+import { and, eq, sql } from "drizzle-orm";
 import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
 import { companies, companySkills, companySkillVersions, companySkillRepositorySnapshots as snapshots,
   skillPackageDeletions, createDb, withDedicatedDbConnection } from "@paperclipai/db";
@@ -81,6 +81,44 @@ describe.skipIf(!support.supported)("external skill packages", () => {
     return { companyId, id, store, skills, service, resolve };
   }
 
+  it("orders skill and package locks consistently with an edit's dedicated upload connection", async () => {
+    const f = await fixture();
+    const skill = f.skills[0]!.skill;
+    const options = { skill, skipInventoryRefresh: true, fileInventory: files, packageSnapshotId: f.id };
+    let waitingOnSkill!: () => void;
+    const waiting = new Promise<void>(resolve => { waitingOnSkill = resolve; });
+    const debug = db.$client.options.debug;
+    db.$client.options.debug = (_connection, query) => {
+      if (query.includes('from "company_skills"') && query.includes('for update')) waitingOnSkill();
+    };
+    let saving: Promise<unknown> | undefined;
+    try {
+      await withDedicatedDbConnection(db, connection => connection.transaction(async tx => {
+        // updateFile owns this lock before its package is prepared separately.
+        await tx.select({ id: companySkills.id }).from(companySkills).where(eq(companySkills.id, skill.id)).for("update");
+        saving = f.service.createVersion(f.companyId, skill.id, {}, null, options);
+        await waiting;
+        await withDedicatedDbConnection(db, async upload => {
+          // A regression fails promptly and releases the outer lock instead of
+          // leaving a hidden cross-connection wait cycle hanging the suite.
+          await upload.execute(sql`set statement_timeout = '3s'`);
+          expect(await skillPackageStore(upload, storage).prepare(f.companyId, files)).toBe(f.id);
+        });
+        await f.service.createVersion(f.companyId, skill.id, {}, null, { ...options, database: tx });
+      }));
+      await saving;
+      expect(await f.service.listVersions(f.companyId, skill.id)).toHaveLength(3);
+    } finally { db.$client.options.debug = debug; await saving; }
+  }, 15_000);
+
+  it("follows a package published between version metadata and summary reads", async () => {
+    const companyId = await company();
+    const [skill] = await db.insert(companySkills).values({ companyId, key: "summary", slug: "summary", name: "Summary", markdown: files[0]!.content }).returning();
+    const [stale] = await db.insert(companySkillVersions).values({ companyId, companySkillId: skill!.id, revisionNumber: 1, fileInventory: files }).returning();
+    await skillPackageMaintenance(db, storage).migrateVersions({ id: stale!.id, apply: true, clearLegacy: true });
+    expect(await skillPackageStore(db, storage).versionInventory(stale!)).toEqual(files.map(({ path, kind }) => ({ path, kind })));
+  });
+
   it("forks a packaged repository inside an outer transaction with a one-connection pool", async () => {
     const f = await fixture();
     const fork = await db.transaction(tx => companySkillService(tx as unknown as typeof db, { packageDb: db })

server/src/__tests__/skill-sources.test.ts

@@ -1,3 +1,6 @@
+import * as recovery from '../services/skill-package-recovery.js';
+import { skillPackageStore, skillPackagePaths } from '../services/skill-package-store.js';
+import { getStorageService } from '../storage/index.js';
 import { resolveExternalTestDatabaseUrl } from './helpers/external-test-database.js';
 import { applyPendingMigrations } from '@paperclipai/db';
 import { randomUUID } from 'node:crypto';
@@ -312,6 +315,34 @@ describe.skipIf(!support.supported)('skill source persistence', () => {
     expect((await skillSourceService(db).list(companyId)).some(source => source.trackingRef === 'rollback')).toBe(false);
     expect((await db.select().from(companySkillVersions).where(eq(companySkillVersions.companyId, companyId))).every(version => before.some(skill => skill.id === version.companySkillId))).toBe(true);
   });
+  it.each(['skills', 'repository'] as const)('binds %s recovery to the committed retry after an import rollback', async packageMode => {
+    const isolated = randomUUID();
+    await db.insert(companies).values({ id: isolated, name: 'Retry recovery', issuePrefix: `R${isolated.slice(0, 6)}` });
+    const original = { 'one/SKILL.md': md('recover retry'), 'one/helper.md': 'Original helper' };
+    const reader = githubFixture(original, {}, sha);
+    const input = { repositoryUrl: 'https://github.com/acme/skills', trackingRef: 'main', packageMode, commitSha: sha, selectedPaths: ['one/SKILL.md'] };
+    const service = skillSourceService(db);
+    await expect(service.create(isolated, input, { ...context, read: () => reader,
+      audit: async () => { throw new Error('failed import transaction'); } })).rejects.toThrow('failed import transaction');
+    const [abandoned] = await db.select().from(companySkillRepositorySnapshots).where(eq(companySkillRepositorySnapshots.companyId, isolated));
+    const result = await service.create(isolated, input, { ...context, read: () => reader });
+    const [saved] = await db.select().from(companySkillRepositorySnapshots).where(eq(companySkillRepositorySnapshots.id, abandoned!.id));
+    expect(saved!.restore!.sourceId).toBe(result.source.id);
+    expect(saved!.restore!.sourceId).not.toBe(abandoned!.restore!.sourceId);
+    await getStorageService().deleteObject(isolated, saved!.storage!.archiveKey);
+    await fs.rm(skillPackagePaths(isolated, saved!.id).directory, { recursive: true, force: true });
+    const authorize = vi.spyOn(recovery, 'skillPackageRecoveryReader').mockImplementation(async (database, company, recipe) => {
+      const [source] = await database.select().from(companySkillSources).where(and(eq(companySkillSources.companyId, company), eq(companySkillSources.id, recipe.sourceId)));
+      expect(source!.id).toBe(result.source.id);
+      return reader;
+    });
+    try {
+      const restored = await skillPackageStore(db).files(isolated, saved!.id);
+      expect(restored.find(file => file.path.endsWith('SKILL.md'))!.content).toBe(original['one/SKILL.md']);
+      expect(authorize).toHaveBeenCalledOnce();
+    } finally { authorize.mockRestore(); }
+  });
+
   it('retains excluded folders, reviews additions, restores removed paths and preserves pinned versions and copies', async () => {
     const service = skillSourceService(db), skills = companySkillService(db);
     files = { 'keep/SKILL.md': md('keep'), 'keep/assets/data.bin': Buffer.from([0,255]), 'keep/run.sh': '#!/bin/sh\necho original', 'declined/one/SKILL.md': md('one') }; commit = sha;

server/src/services/company-skills.ts

@@ -3738,12 +3738,6 @@ export function companySkillService(db: Db, options: { actor?: AuthorizationActo
     }
     const persist = async (tx: DbOrTransaction) => {
       await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillPackages);
-      if (packageSnapshotId) {
-        const [saved] = await tx.select({ storage: companySkillRepositorySnapshots.storage })
-          .from(companySkillRepositorySnapshots).where(and(eq(companySkillRepositorySnapshots.companyId, companyId),
-            eq(companySkillRepositorySnapshots.id, packageSnapshotId))).for("update");
-        if (!saved?.storage) throw conflict("Skill package changed during publication; retry the operation.");
-      }
       await tx.execute(sql`
         select ${companySkills.id}
         from ${companySkills}
@@ -3751,6 +3745,14 @@ export function companySkillService(db: Db, options: { actor?: AuthorizationActo
           and ${companySkills.companyId} = ${companyId}
         for update
       `);
+      // File edits already own the skill lock when a dedicated connection saves
+      // their package. Never hold a package lock while waiting for that skill.
+      if (packageSnapshotId) {
+        const [saved] = await tx.select({ storage: companySkillRepositorySnapshots.storage })
+          .from(companySkillRepositorySnapshots).where(and(eq(companySkillRepositorySnapshots.companyId, companyId),
+            eq(companySkillRepositorySnapshots.id, packageSnapshotId))).for("update");
+        if (!saved?.storage) throw conflict("Skill package changed during publication; retry the operation.");
+      }
       const [{ nextRevision }] = await tx
         .select({
           nextRevision: sql<number>`coalesce(max(${companySkillVersions.revisionNumber}), 0) + 1`,

server/src/services/skill-package-store.ts

@@ -7,7 +7,7 @@ import { constants, createReadStream, createWriteStream, promises as fs } from "
 import path from "node:path";
 import { pipeline } from "node:stream/promises";
 import { and, eq, getTableColumns, inArray, sql } from "drizzle-orm";
-import { companySkillRepositorySnapshots as snapshots, companySkillVersions, type Db, type SkillPackageFileMapping, type SkillPackageStorageReceipt, type SkillPackageRestoreRecipe } from "@paperclipai/db";
+import { companySkillRepositorySnapshots as snapshots, companySkillVersions, companySkillSources, type Db, type SkillPackageFileMapping, type SkillPackageStorageReceipt, type SkillPackageRestoreRecipe } from "@paperclipai/db";
 import type { CompanySkillVersionFileInventoryEntry } from "@paperclipai/shared";
 import { getStorageService, type StorageService } from "../storage/index.js";
 import { resolvePaperclipInstanceRoot } from "../home-paths.js";
@@ -418,6 +418,18 @@ export function skillPackageStore(db: Db, suppliedStorage?: StorageService) {
     }
     return preparedId;
   }
+  async function bindRestore(companyId: string, id: string, restore: SkillPackageRestoreRecipe) {
+    const manifest = validateSkillPackageManifest(restore.manifest);
+    // Called in the source publication transaction. A failed import must not
+    // leave a later successful import bound to an uncommitted source identity.
+    await db.update(snapshots).set({ restore }).where(and(scope(companyId, id),
+      eq(snapshots.contentHash, manifest.contentHash),
+      sql`${snapshots.storage}->>'manifestSha256' = ${packageSha256(skillPackageManifestBytes(manifest))}`,
+      sql`exists (select 1 from ${companySkillSources} s where s.company_id = ${companyId}
+        and s.id = ${restore.sourceId} and s.removed_at is null)`,
+      sql`not exists (select 1 from ${companySkillSources} s where s.company_id = ${companyId}
+        and s.id::text = ${snapshots.restore}->>'sourceId' and s.removed_at is null)`));
+  }
   async function verifyDurable(companyId: string, id: string, cancellation?: AbortSignal,
     validate?: (root: string, manifest: SkillPackageManifest) => Promise<boolean>): Promise<boolean> {
     const action = async () => {
@@ -513,17 +525,18 @@ export function skillPackageStore(db: Db, suppliedStorage?: StorageService) {
     try { return await read(skillPackagePaths(version.companyId, id).directory); }
     catch { return withIsolatedPackage(version.companyId, id, read, allowRecovery); }
   }
-  async function versionInventory(version: StoredSkillVersion) {
+  async function versionInventory(version: StoredSkillVersion): Promise<Array<{ path: string; kind: CompanySkillVersionFileInventoryEntry["kind"] }>> {
     if (version.packageSnapshotId) {
       const index = version.packageFileMap ?? (await manifestFor(await descriptor(version.companyId, version.packageSnapshotId))).files;
       return index.map(({ path, kind }) => ({ path, kind }));
     }
-    const [row] = await db.select({ files: sql<Array<{ path: string; kind: CompanySkillVersionFileInventoryEntry["kind"] }>>`
+    const [row] = await db.select({ ...skillVersionMetadataColumns, files: sql<Array<{ path: string; kind: CompanySkillVersionFileInventoryEntry["kind"] }>>`
       coalesce((select jsonb_agg(jsonb_build_object('path', f->>'path', 'kind', f->>'kind'))
         from jsonb_array_elements(${companySkillVersions.fileInventory}) f), '[]'::jsonb)` })
       .from(companySkillVersions).where(and(eq(companySkillVersions.companyId, version.companyId), eq(companySkillVersions.id, version.id)));
+    if (row?.packageSnapshotId) return versionInventory(row);
     return row?.files ?? [];
   }
-  return { descriptor, prepare, files, version, prefetch, versionFiles, versionInventory, readVersionFile, withPackage, withIsolatedPackage, verifyDurable,
+  return { descriptor, prepare, bindRestore, files, version, prefetch, versionFiles, versionInventory, readVersionFile, withPackage, withIsolatedPackage, verifyDurable,
     manifest: async (companyId: string, id: string) => manifestFor(await descriptor(companyId, id)) };
 }

server/src/services/skill-sources.ts

@@ -10,7 +10,7 @@ import path from 'node:path';
 import { promises as fs } from 'node:fs';
 import { folderService } from './folders.js';
 import { and, eq, isNull, lt, or, sql } from 'drizzle-orm';
-import { companySkillSources as sources, companySkillSourceEntries as entries, companySkills, companySkillRepositorySnapshots, projectSkillSources, projects, type Db, authUsers } from '@paperclipai/db';
+import { companySkillSources as sources, companySkillSourceEntries as entries, companySkills, companySkillRepositorySnapshots, projectSkillSources, projects, type Db, type SkillPackageRestoreRecipe, authUsers } from '@paperclipai/db';
 import type { CompanySkill, SkillSource, SkillSourceCreateRequest, SkillSourceDiscoveryRequest, SkillSourceSelectionRequest, SkillSourceRefreshResult, SkillSourcePreviewRequest, SkillSourceSettingsRequest, SkillSourceDisconnectRequest, SkillSourceDisconnectResult, SkillSourceBranchesQuery } from '@paperclipai/shared';
 import { normalizeAgentUrlKey } from '@paperclipai/shared';
 import { conflict, notFound, unprocessable } from '../errors.js';
@@ -19,6 +19,7 @@ import { scanGitHubSkillsWithPackageFallback as scanGitHubSkills, previewGitHubS
 import { skillAccessService } from './skill-access.js';
 import { skillSnapshotHash } from './skill-snapshot.js';
 
+type PreparedPackages = Map<string, { id: string; restore: SkillPackageRestoreRecipe }>;
 type Tx = Parameters<Parameters<Db['transaction']>[0]>[0];
 type SourceRow = typeof sources.$inferSelect;
 export type SkillSourceContext = {
@@ -37,23 +38,25 @@ const scope = (companyId: string, id: string) => and(eq(sources.companyId, compa
 export function skillSourceService(db: Db) {
   const skills = companySkillService(db);
   const packages = skillPackageStore(db);
-  async function preparePackages(companyId: string, scan: ScannedSkillSource, selected: string[], sourceId: string, signal?: AbortSignal) {
-    if (!skillPackageStorageEnabled()) return new Map<string, string>();
+  async function preparePackages(companyId: string, scan: ScannedSkillSource, selected: string[], sourceId: string, signal?: AbortSignal): Promise<PreparedPackages> {
+    const prepared: PreparedPackages = new Map();
+    if (!skillPackageStorageEnabled()) return prepared;
     const candidates = scan.skills.filter(candidate => selected.includes(candidate.path) && !candidate.error);
-    const prepared = new Map<string, string>();
     if (!candidates.length) return prepared;
+    const prepare = async (files: ScannedSkillSource['skills'][number]['files'], skillPath: string | null) => {
+      const restore: SkillPackageRestoreRecipe = { version: 1, sourceId, repositoryId: scan.repositoryId,
+        repositoryUrl: scan.repositoryUrl, commitSha: scan.commitSha, skillPath, manifest: skillPackageManifest(files) };
+      const id = await packages.prepare(companyId, files, true, signal, restore);
+      // Deduplication may preserve another version's kind/encoding metadata.
+      // Recovery uses the manifest of the actual stored archive.
+      return { id, restore: { ...restore, manifest: await packages.manifest(companyId, id) } };
+    };
     if (scan.packageMode === 'repository') {
       if (scan.repositoryPackage?.error || !scan.repositoryFiles?.length) throw unprocessable(scan.repositoryPackage?.error ?? 'Repository package has no files.');
-      const id = await packages.prepare(companyId, scan.repositoryFiles, true, signal, {
-        version: 1, sourceId, repositoryId: scan.repositoryId, repositoryUrl: scan.repositoryUrl,
-        commitSha: scan.commitSha, skillPath: null, manifest: skillPackageManifest(scan.repositoryFiles),
-      });
-      for (const candidate of candidates) prepared.set(candidate.path, id);
+      const saved = await prepare(scan.repositoryFiles, null);
+      for (const candidate of candidates) prepared.set(candidate.path, saved);
     } else {
-      for (const candidate of candidates) prepared.set(candidate.path, await packages.prepare(companyId, candidate.files, true, signal, {
-        version: 1, sourceId, repositoryId: scan.repositoryId, repositoryUrl: scan.repositoryUrl,
-        commitSha: scan.commitSha, skillPath: candidate.path, manifest: skillPackageManifest(candidate.files),
-      }));
+      for (const candidate of candidates) prepared.set(candidate.path, await prepare(candidate.files, candidate.path));
     }
     return prepared;
   }
@@ -122,13 +125,13 @@ export function skillSourceService(db: Db) {
     const identity = createHash('sha256').update(`${scan.repositoryId}:${sourceId}${revision ? `:${revision}` : ''}:${skillPath}`).digest('hex').slice(0, 16);
     return `github/${scan.repositoryId}/${identity}/${normalizeAgentUrlKey(name) || 'skill'}`;
   }
-  async function publish(tx: Tx, source: SourceRow, scan: ScannedSkillSource, selectedPaths: string[], excludedFolders: string[], context: SkillSourceContext, selectionReviewed: boolean, prepared: Map<string, string>): Promise<Omit<SkillSourceRefreshResult, 'source'>> {
+  async function publish(tx: Tx, source: SourceRow, scan: ScannedSkillSource, selectedPaths: string[], excludedFolders: string[], context: SkillSourceContext, selectionReviewed: boolean, prepared: PreparedPackages): Promise<Omit<SkillSourceRefreshResult, 'source'>> {
     await tx.execute(sql`SET CONSTRAINTS paperclip_skill_runtime_changed DEFERRED`);
     const mode = scan.packageMode ?? 'skills';
     if (mode === 'repository' && scan.skills.length > 0 && (scan.repositoryPackage?.error || !scan.repositoryFiles?.length)) {
       throw unprocessable(scan.repositoryPackage?.error ?? 'Repository package has no files. Scan it again before importing.');
     }
-    let repositorySnapshotId = mode === 'repository' ? prepared.values().next().value : undefined;
+    let repositorySnapshotId = mode === 'repository' ? prepared.values().next().value?.id : undefined;
     const hasSelected = scan.skills.some(candidate => selectedPaths.includes(candidate.path) && !candidate.error);
     const repositoryHash = mode === 'repository' && hasSelected ? skillSnapshotHash(scan.repositoryFiles!) : null;
     if (repositoryHash && !repositorySnapshotId) {
@@ -184,7 +187,7 @@ export function skillSourceService(db: Db) {
         if (changed) {
           const version = await skills.createVersion(source.companyId, skillId!, { label: `GitHub ${scan.commitSha.slice(0, 8)}` }, context.actor, {
             database: tx, skipInventoryRefresh: true, skill: installed, fileInventory: candidate.files,
-            packageSnapshotId: prepared.get(candidate.path),
+            packageSnapshotId: prepared.get(candidate.path)?.id,
             packageFileMap: repositorySnapshotId && prepared.has(candidate.path) ? candidate.files.map(file => ({ path: file.path, kind: file.kind,
               sourcePath: file.path === 'SKILL.md' ? candidate.path : path.posix.join(path.posix.dirname(candidate.path), file.path) })) : null,
             ...(repositorySnapshotId ? { repositorySnapshotId, repositorySkillPath: candidate.path } : {}),
@@ -212,6 +215,10 @@ export function skillSourceService(db: Db) {
         if (current) await tx.update(companySkills).set({ metadata: { ...current.metadata, skillSourceId: source.id, skillSourcePath: old.path, skillSourceState: 'removed' } }).where(and(eq(companySkills.companyId, source.companyId), eq(companySkills.id, old.skillId)));
       }
     }
+    const publicationStore = skillPackageStore(tx as unknown as Db);
+    for (const { id, restore } of new Map([...prepared.values()].map(saved => [saved.id, saved])).values()) {
+      await publicationStore.bindRestore(source.companyId, id, restore);
+    }
     // HEAD remains the default-branch alias across refreshes and branch renames.
     await tx.update(sources).set({ repositoryId: scan.repositoryId, repositoryUrl: scan.repositoryUrl, fullName: scan.fullName,
       packageMode: mode, repositoryPackage: scan.repositoryPackage ?? null,