From a3eecf1fc9fa8c1a88bb68adb9b9fc0328a252b3 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Fri, 6 Nov 2026 18:23:53 -0600 Subject: [PATCH] Preserve admitted runtime skill bundles during server replacement (#19675) ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Native runs keep the exact skill files admitted for execution and recovery. > - Older inputs can name immutable local bundles without a skill version ID. > - Installed skill heads alone do not preserve those admitted bundles. > - Server replacement needs a durable source for each retained bundle. > - This pull request copies authenticated bundles through the existing package owner and restores their exact projections. > - The benefit is preservation during online migration without stopping active work or settling recovery warnings. ## Linked Issues or Issue Description Refs #19656. Related: #19660. **What happened?** Older native execution inputs can contain skill bundles with a null version ID. Recovery skips those bundles. Startup installation checks can pass even when those retained inputs still require the old filesystem. **Expected behavior** Preserve each retained admitted bundle in object storage. Restore its original bytes, modes and digests on an empty replacement host. Keep its null pin and all run, controller and recovery state. Block startup and enrollment while a concrete local asset obligation remains. **Steps to reproduce** Retain a native run with an admitted local skill bundle and a null version ID. Convert its installed skill head to durable storage. Replace the server home and attempt recovery. The old bundle path remains in the recovered context. ## What Changed - Add a bounded `runtime-assets` phase to the existing skill package maintenance CLI. It copies exact admitted files under the runtime asset publication lock, verifies the durable package, and appends a run-bound receipt after source revalidation. - Keep original files, null pins, task intent, warnings and controller leases. Support an active run through copy-only preservation. Do not start a turn or claim provider ownership. - Verify and reuse a healthy canonical local bundle before storage reads. Restore missing or damaged bundles through the existing package and native asset materializers. Use the authenticated original manifest to preserve exact read and execute modes. Accept a later contract revision only when the run binding and exact skill selection remain unchanged. - Check retained recovery obligations and current task sessions during both startup and migration eligibility. Ignore ordinary admission that has no completed execution input. - Protect receipt-backed packages from orphan collection. Serialize receipt publication and deletion on the package row, then check references in a fresh statement after the lock. - Add real PostgreSQL and filesystem regressions. Replace a delete-method proxy with an actual upload-renewal row-lock race. Publish mutation activity only after commit. ## Verification - Baseline production startup gate failed the new concrete-asset regression as expected. The fixed gate passed. The actual before-fix preservation owner also failed recovery from a healthy local bundle during a simulated storage outage. A verified local-first path fixes that regression. - Eight preservation tests passed with real PostgreSQL, streaming storage, archive extraction and native materialization. They cover cold recovery, exact modes and binary bytes, active ownership, changed sources and bindings, contract revisions, missing or corrupt receipts, and publication/collection races. - Six affected package collection/deletion controls and three existing startup controls passed. The existing versioned recovery consumer passed a sequential retry. - One fixture attempt failed on its cleanup foreign-key order; the fixture was corrected. One existing suite hit its unchanged 30-second database startup hook before test execution. The unchanged single consumer passed on retry. These failures remain recorded. - Source, changed-test and CLI TypeScript checks passed. Rebase preserved both the local instruction-copy and runtime-asset guards. Current-head changed-test checks passed. Final PostgreSQL rerun passed all eleven preservation/startup cases. Current CLI check passed. Required CI remains pending at the trusted-runner gate. - Current canonical bundled creation uses package preparation plus atomic content/head publication. Native preparation requests immutable version snapshots before delivery. Converted installations and new canonical bundled creation therefore do not recreate the historical null-version asset dependency. - Fresh Greptile review on exact head `54c7dd048c9016c1af9fc68213d5ae053ac16499` passed at 5/5 with zero unresolved threads. - Module boundaries, whitespace, and credential/internal-reference scans passed. No live source writes, provider actions or migration apply were performed. ## Risks Missing, changed or unsafe local files remain blockers. This command does not repair a previously published package whose objects later disappear or become corrupt. It also refuses malformed preservation receipts. It retains the original local files and prior evidence and keeps removal blocked. Automatic repair requires a separate canonical package-owner recovery operation; direct database edits are not an accepted recovery path. Failed publication keeps the original files and any unreferenced package for normal collection. The readiness check is explicit startup or enrollment work, not a background scan. Its cost grows with retained admitted asset obligations. Older readers must be upgraded before the old volume is removed. This change preserves asset bytes; it does not settle business recovery or qualify unrelated volume dependencies. ## Model Used OpenAI GPT-6 (Codex). The runtime did not expose a more specific backend model ID or context-window size. Used reasoning, code editing, and tool execution. ## 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 --------- Co-authored-by: Paperclip --- doc/skill-ephemeral-web-servers.md | 24 ++ server/scripts/migrate-skill-packages.ts | 6 +- .../runtime-skill-preservation.test.ts | 244 ++++++++++++++++++ .../src/__tests__/skill-package-store.test.ts | 27 +- .../src/services/ephemeral-server-contract.ts | 2 + .../native-runtime/runtime-context.ts | 15 +- .../runtime-skill-preservation-reference.ts | 75 ++++++ .../runtime-skill-preservation.ts | 213 +++++++++++++++ .../src/services/skill-package-maintenance.ts | 16 +- 9 files changed, 605 insertions(+), 17 deletions(-) create mode 100644 server/src/__tests__/runtime-skill-preservation.test.ts create mode 100644 server/src/services/native-runtime/runtime-skill-preservation-reference.ts create mode 100644 server/src/services/native-runtime/runtime-skill-preservation.ts diff --git a/doc/skill-ephemeral-web-servers.md b/doc/skill-ephemeral-web-servers.md index 28194dca51b..f12c4c2bf5d 100644 --- a/doc/skill-ephemeral-web-servers.md +++ b/doc/skill-ephemeral-web-servers.md @@ -40,6 +40,30 @@ Native recovery resolves the version, reconstructs its v1 projection and require the admitted digest, manifest digest, file count and byte count to match. Adjacent compatible releases must preserve that projection contract. +Older native inputs can contain immutable skill bundles without a version ID. +Retained recovery obligations and current task sessions keep those bundles as +local durability dependencies until the `runtime-assets` maintenance phase copies +their authenticated files into the existing package store. The receipt binds the +original run input, skill selection, native manifest and package content hash. +Recovery restores the original bytes and read/execute modes, verifies the admitted +digests, and preserves the null version ID. Package collection protects these +receipts under the same row lock used by receipt publication. + +This conversion may run online, including for an active run: it only copies an +immutable bundle under its publication/collection lock and appends an input-bound +receipt. It does not claim provider or controller ownership, settle recovery, +start a turn, change task intent or remove local data. Missing or changed sources +remain blockers. Ordinary new admission without a completed execution input is +not itself a local asset obligation. Both migration and replacement startup check +the concrete retained asset obligations independently of installation eligibility. + +Run on the host retaining the original runtime asset files, and follow its cursor: + +```sh +pnpm --filter @paperclipai/server exec tsx scripts/migrate-skill-packages.ts --phase=runtime-assets --company= --limit=10 +pnpm --filter @paperclipai/server exec tsx scripts/migrate-skill-packages.ts --phase=runtime-assets --company= --limit=10 --apply +``` + Studio tests deliver the selected saved version through the same native Runner path, even without a permanent agent assignment. Their `skill_test` mode remains explicit in the execution input. The durable test row replaces ordinary agent diff --git a/server/scripts/migrate-skill-packages.ts b/server/scripts/migrate-skill-packages.ts index a0fb8edaac3..7f761cf5afc 100644 --- a/server/scripts/migrate-skill-packages.ts +++ b/server/scripts/migrate-skill-packages.ts @@ -5,6 +5,7 @@ import { createStorageProviderFromConfig } from "../src/storage/provider-registr import { createStorageService } from "../src/storage/service.js"; import { skillPackageMaintenance } from "../src/services/skill-package-maintenance.js"; import { skillInstallationMaintenance } from "../src/services/skill-installation-maintenance.js"; +import { migrateRuntimeSkillAssets } from "../src/services/native-runtime/runtime-skill-preservation.js"; import { isCloudManagedInstance } from "../src/services/cloud-instance.js"; import { cloudRuntimeConfigService } from "../src/services/cloud-runtime-config.js"; import { setCloudRuntimeConfigSnapshot } from "../src/services/cloud-runtime-config-snapshot.js"; @@ -29,9 +30,10 @@ try { const installations = skillInstallationMaintenance(db, storage); const actions = { snapshots: service.migrateSnapshots, versions: service.migrateVersions, orphans: service.collectOrphans, deletions: service.deleteObjects, - installations: installations.installations, eligibility: installations.eligibility, verification: installations.verifyVersions }; + installations: installations.installations, eligibility: installations.eligibility, verification: installations.verifyVersions, + "runtime-assets": (options: Parameters[1]) => migrateRuntimeSkillAssets(db, options, storage) }; const action = actions[phase as keyof typeof actions]; - if (!action) throw new Error("--phase must be snapshots, versions, orphans, deletions, installations, eligibility, or verification"); + if (!action) throw new Error("--phase must be snapshots, versions, orphans, deletions, installations, eligibility, verification, or runtime-assets"); const apply = process.argv.includes("--apply"); const result = await action({ apply, writersStopped: process.argv.includes("--writers-stopped"), clearLegacy: process.argv.includes("--clear-legacy"), limit, after: value("--after"), companyId: value("--company") }); diff --git a/server/src/__tests__/runtime-skill-preservation.test.ts b/server/src/__tests__/runtime-skill-preservation.test.ts new file mode 100644 index 00000000000..d0546257902 --- /dev/null +++ b/server/src/__tests__/runtime-skill-preservation.test.ts @@ -0,0 +1,244 @@ +import { randomUUID } from "node:crypto"; +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { and, eq, sql } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; +import { activityLog, agents, companies, heartbeatRuns, issues, companySkillRepositorySnapshots as snapshots, + nativeRunFinalizations, createDb } from "@paperclipai/db"; +import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import * as storageModule from "../storage/index.js"; +import { createStorageService } from "../storage/service.js"; +import { createLocalDiskStorageProvider } from "../storage/local-disk-provider.js"; +import { materializeAsset, rehydratePinnedSkillAssets } from "../services/native-runtime/runtime-context.js"; +import { nativeRuntimeContextFixture } from "../services/native-runtime/runtime-context.test-fixture.js"; +import { canonicalNativeRuntimeContextDigest, parseNativeExecutionInput } from "../vendor/paperclip-runner/index.js"; +import { preserveRuntimeSkillAssets, hasLocalRuntimeSkillObligations, migrateRuntimeSkillAssets } from "../services/native-runtime/runtime-skill-preservation.js"; +import { readEphemeralServerDataEligibility } from "../services/ephemeral-server-contract.js"; +import { skillPackageMaintenance } from "../services/skill-package-maintenance.js"; +import { removeRuntimeAssetTree } from "../services/native-runtime/runtime-asset-retention.js"; + +describe("admitted unversioned runtime skill preservation", () => { + let database: Awaited>; + let db: ReturnType; + let home: string; + let storage: storageModule.StorageService; + const previousHome = process.env.PAPERCLIP_HOME; + beforeAll(async () => { + database = await startEmbeddedPostgresTestDatabase("runtime-skill-preservation-"); + db = createDb(database.connectionString, { maxConnections: 6 }); + home = await fs.mkdtemp(path.join(os.tmpdir(), "runtime-skill-preservation-")); + process.env.PAPERCLIP_HOME = home; + // Real streaming storage/archives on disk with the same durable receipt + // contract; no external provider credentials or HTTP mocks. + storage = createStorageService({ ...createLocalDiskStorageProvider(path.join(home, "objects")), id: "s3" }); + vi.spyOn(storageModule, "getStorageService").mockReturnValue(storage); + }, 90_000); + afterEach(async () => { + process.env.PAPERCLIP_HOME = home; + await db.delete(activityLog); + await db.delete(nativeRunFinalizations); + await db.delete(heartbeatRuns); + await db.delete(issues); + await db.delete(agents); + await db.delete(snapshots); + await db.delete(companies); + }); + afterAll(async () => { + vi.restoreAllMocks(); + if (previousHome === undefined) delete process.env.PAPERCLIP_HOME; else process.env.PAPERCLIP_HOME = previousHome; + await db?.$client.end({ timeout: 1 }); + await database?.cleanup(); + if (home) await removeRuntimeAssetTree(home); + }); + async function fixture() { + const companyId = randomUUID(), agentId = randomUUID(), runId = randomUUID(), issueId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Skill preservation", issuePrefix: `R${companyId.slice(0, 6)}` }); + await db.insert(agents).values({ id: agentId, companyId, name: "Recovery source" }); + await db.insert(issues).values({ id: issueId, companyId, title: "Preserve admitted skill" }); + const bundle = await materializeAsset([ + { path: "SKILL.md", content: Buffer.from("# original admitted skill\n"), mode: 0o400 }, + { path: "scripts/run.sh", content: Buffer.from("#!/bin/sh\necho preserve\n"), mode: 0o500 }, + { path: "binary", content: Buffer.from([0, 255, 7]), mode: 0o444 }, + ]); + const context = nativeRuntimeContextFixture(); + context.skills = [{ key: "legacy", runtimeName: "legacy", versionId: null, bundle }]; + context.aggregateDigest = canonicalNativeRuntimeContextDigest(context); + const execution = parseNativeExecutionInput({ schema: "paperclip.native-execution-input.v3", + binding: { companyId, runId, agentId, issueId, executionWorkspaceId: runId }, + task: { identifier: "PRESERVE-1", title: "Saved selection", description: null, prompt: "Read saved skills", workMode: "standard" }, + workspace: { cwd: "/tmp", repoUrl: null, repoRef: null, branchName: null }, + session: { normalizedSessionId: null, driverKind: "codex_app_server", protocolVersion: 1, lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null } }, + provider: { kind: "codex", model: null }, + completionContract: { id: "contract", sha256: "saved", schemaVersion: "paperclip.completion-contract.v1", + contract: { revision: "1", objective: "Read saved skills", criteria: [{ id: "read", requirement: "Read skills" }] } }, + interactionResponses: [], credentialBindings: [], executionMode: "default", planningContext: null, runtimeContext: context }); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, nativeIssueId: issueId, + runtimeMode: "native", status: "failed", finishedAt: new Date(), + processPid: 1234, processStartedAt: new Date(0), runnerProfileJson: { nativeExecutionInput: execution, retainedWarning: "unresolved recovery" } }); + await db.insert(nativeRunFinalizations).values({ companyId, runId, issueId, phase: "terminal_failure", recoveryState: "blocked" }); + return { companyId, agentId, runId, issueId, context, bundle, execution }; + } + it("blocks startup on a concrete local asset but ignores an unfinished ordinary admission", async () => { + const f = await fixture(); + const [ordinary] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, + runtimeMode: "native", status: "running", contextSnapshot: {} }).returning(); + expect((await readEphemeralServerDataEligibility(db, { purpose: "startup" })).blockers) + .toContain("runtime_skill_assets_require_migration"); + await preserveRuntimeSkillAssets(db, f, storage); + expect((await readEphemeralServerDataEligibility(db, { purpose: "startup" })).blockers) + .not.toContain("runtime_skill_assets_require_migration"); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, ordinary!.id)))[0]!.contextSnapshot).toEqual({}); + }); + it("preserves exact bytes, modes and null pins for cold recovery without settling warnings or starting a turn", async () => { + const f = await fixture(); + const before = (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0]!; + expect(await preserveRuntimeSkillAssets(db, f, storage)).toEqual({ preserved: 1 }); + const saved = (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0]!; + expect(saved).toEqual({ ...before, runnerProfileJson: { ...before.runnerProfileJson, + nativeRuntimeSkillPreservation: saved.runnerProfileJson!.nativeRuntimeSkillPreservation } }); + expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0]) + .toMatchObject({ phase: "terminal_failure", recoveryState: "blocked", leaseOwner: null }); + expect(await hasLocalRuntimeSkillObligations(db)).toBe(false); + process.env.PAPERCLIP_HOME = path.join(home, `cold-${randomUUID()}`); + const recovered = await rehydratePinnedSkillAssets(db, f, f.context); + expect(recovered.aggregateDigest).toBe(f.context.aggregateDigest); + expect(recovered.skills[0]!.versionId).toBeNull(); + const root = recovered.skills[0]!.bundle.rootPath; + expect(root).not.toBe(f.bundle.rootPath); + expect({ ...recovered.skills[0]!.bundle, rootPath: f.bundle.rootPath }).toEqual(f.bundle); + expect(await fs.readFile(path.join(root, "binary"))).toEqual(Buffer.from([0, 255, 7])); + expect((await fs.stat(path.join(root, "SKILL.md"))).mode & 0o777).toBe(0o400); + expect((await fs.stat(path.join(root, "scripts/run.sh"))).mode & 0o777).toBe(0o500); + expect(await fs.readFile(path.join(f.bundle.rootPath, "SKILL.md"), "utf8")).toBe("# original admitted skill\n"); + expect(await db.select().from(heartbeatRuns)).toHaveLength(1); + }); + it("reuses an authenticated local bundle during a storage outage and restores a damaged cache from its package", async () => { + const f = await fixture(); + await preserveRuntimeSkillAssets(db, f, storage); + const get = vi.spyOn(storage, "getObject").mockRejectedValue(new Error("storage offline")); + try { + expect((await rehydratePinnedSkillAssets(db, f, f.context)).skills[0]!.bundle).toEqual(f.bundle); + expect(get).not.toHaveBeenCalled(); + } finally { get.mockRestore(); } + const file = path.join(f.bundle.rootPath, "SKILL.md"); + await fs.chmod(file, 0o600); await fs.writeFile(file, "damaged"); await fs.chmod(file, 0o400); + const read = vi.spyOn(storage, "getObject"); + try { + expect((await rehydratePinnedSkillAssets(db, f, f.context)).skills[0]!.bundle).toEqual(f.bundle); + expect(read).toHaveBeenCalled(); + expect(await fs.readFile(file, "utf8")).toBe("# original admitted skill\n"); + } finally { read.mockRestore(); } + }); + it("rejects changed source bytes and changed admitted input before receipt publication", async () => { + const f = await fixture(); + const file = path.join(f.bundle.rootPath, "SKILL.md"); + await fs.chmod(file, 0o600); await fs.writeFile(file, "changed"); await fs.chmod(file, 0o400); + await expect(preserveRuntimeSkillAssets(db, f, storage)).rejects.toThrow("source_changed"); + expect((await db.select().from(snapshots))).toHaveLength(0); + const g = await fixture(); + const put = storage.putFile.bind(storage); + const spy = vi.spyOn(storage, "putFile").mockImplementation(async input => { + const result = await put(input); + await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeExecutionInput: { ...g.execution, + task: { ...g.execution.task, prompt: "new admitted intent" } } } }).where(eq(heartbeatRuns.id, g.runId)); + return result; + }); + try { await expect(preserveRuntimeSkillAssets(db, g, storage)).rejects.toThrow("binding_changed"); } + finally { spy.mockRestore(); } + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, g.runId)))[0]!.runnerProfileJson) + .not.toHaveProperty("nativeRuntimeSkillPreservation"); + }); + it("fails closed for missing/corrupt durable receipts and refuses another company's run", async () => { + const f = await fixture(); + await expect(preserveRuntimeSkillAssets(db, { ...f, companyId: randomUUID() }, storage)).rejects.toThrow("source_unavailable"); + await preserveRuntimeSkillAssets(db, f, storage); + const [snapshot] = await db.select().from(snapshots).where(eq(snapshots.companyId, f.companyId)); + await db.update(snapshots).set({ storage: null }).where(eq(snapshots.id, snapshot!.id)); + expect(await hasLocalRuntimeSkillObligations(db)).toBe(true); + await db.update(snapshots).set({ storage: { ...snapshot!.storage!, durable: false } }).where(eq(snapshots.id, snapshot!.id)); + expect(await hasLocalRuntimeSkillObligations(db)).toBe(true); + await expect(rehydratePinnedSkillAssets(db, f, f.context)).rejects.toThrow("package_not_durable"); + await db.update(snapshots).set({ storage: snapshot!.storage }).where(eq(snapshots.id, snapshot!.id)); + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + const receipt = run!.runnerProfileJson!.nativeRuntimeSkillPreservation as { assets: Array<{ manifestText: string }> }; + receipt.assets[0]!.manifestText += " "; + await db.update(heartbeatRuns).set({ runnerProfileJson: { ...run!.runnerProfileJson, nativeRuntimeSkillPreservation: receipt } }) + .where(eq(heartbeatRuns.id, f.runId)); + expect(await hasLocalRuntimeSkillObligations(db)).toBe(true); + await expect(rehydratePinnedSkillAssets(db, f, f.context)).rejects.toThrow("manifest_digest_mismatch"); + }); + it("copies a running owner's immutable asset without taking its lease, changing its state, or requiring completion", async () => { + const f = await fixture(); + const controllerBootId = randomUUID(), controllerLeaseExpiresAt = new Date(Date.now() + 600_000); + await db.update(heartbeatRuns).set({ status: "running", finishedAt: null, controllerBootId, controllerLeaseExpiresAt }) + .where(eq(heartbeatRuns.id, f.runId)); + await db.update(nativeRunFinalizations).set({ leaseOwner: "original-live-owner" }).where(eq(nativeRunFinalizations.runId, f.runId)); + const preview = await migrateRuntimeSkillAssets(db, { companyId: f.companyId }); + expect(preview.results).toEqual([{ runId: f.runId, status: "candidate" }]); + expect(await hasLocalRuntimeSkillObligations(db)).toBe(true); + expect((await migrateRuntimeSkillAssets(db, { companyId: f.companyId, apply: true }, storage)).results) + .toEqual([{ runId: f.runId, status: "preserved", preserved: 1 }]); + expect(await hasLocalRuntimeSkillObligations(db)).toBe(false); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0]) + .toMatchObject({ status: "running", finishedAt: null, controllerBootId, controllerLeaseExpiresAt, processPid: 1234 }); + expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0]) + .toMatchObject({ phase: "terminal_failure", recoveryState: "blocked", leaseOwner: "original-live-owner" }); + await removeRuntimeAssetTree(f.bundle.rootPath); + expect(await preserveRuntimeSkillAssets(db, f, storage)).toEqual({ preserved: 1 }); + }); + it("retains the original receipt through a contract revision but rejects a changed admitted asset or binding", async () => { + const f = await fixture(); + await preserveRuntimeSkillAssets(db, f, storage); + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + const profile = run!.runnerProfileJson!; + await db.update(heartbeatRuns).set({ runnerProfileJson: { ...profile, + nativeExecutionInput: { ...f.execution, task: { ...f.execution.task, prompt: "Revised task contract" } } } }) + .where(eq(heartbeatRuns.id, f.runId)); + expect(await hasLocalRuntimeSkillObligations(db)).toBe(false); + process.env.PAPERCLIP_HOME = path.join(home, `revision-${randomUUID()}`); + expect((await rehydratePinnedSkillAssets(db, f, f.context)).skills[0]!.bundle.digest).toBe(f.bundle.digest); + const changed = structuredClone(f.context); + changed.skills[0]!.runtimeName = "different-admitted-selection"; + changed.aggregateDigest = canonicalNativeRuntimeContextDigest(changed); + await db.update(heartbeatRuns).set({ runnerProfileJson: { ...profile, + nativeExecutionInput: { ...f.execution, runtimeContext: changed } } }).where(eq(heartbeatRuns.id, f.runId)); + expect(await hasLocalRuntimeSkillObligations(db)).toBe(true); + await expect(rehydratePinnedSkillAssets(db, f, changed)).rejects.toThrow("preservation_required"); + await db.update(heartbeatRuns).set({ runnerProfileJson: { ...profile, + nativeExecutionInput: { ...f.execution, binding: { ...f.execution.binding, executionWorkspaceId: randomUUID() } } } }) + .where(eq(heartbeatRuns.id, f.runId)); + expect(await hasLocalRuntimeSkillObligations(db)).toBe(true); + }); + it("keeps referenced packages through orphan collection and observes a receipt committed while deletion waits", async () => { + const f = await fixture(); + await preserveRuntimeSkillAssets(db, f, storage); + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + const receipt = run!.runnerProfileJson!.nativeRuntimeSkillPreservation; + const [snapshot] = await db.select().from(snapshots).where(eq(snapshots.companyId, f.companyId)); + await db.update(snapshots).set({ createdAt: new Date(0) }).where(eq(snapshots.id, snapshot!.id)); + expect((await skillPackageMaintenance(db, storage).collectOrphans({ companyId: f.companyId, apply: true })).removed).toBe(0); + await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeExecutionInput: f.execution } }).where(eq(heartbeatRuns.id, f.runId)); + let unlock!: () => void, locked!: () => void, waiting!: () => void; + const release = new Promise(resolve => { unlock = resolve; }); + const acquired = new Promise(resolve => { locked = resolve; }); + const deleting = new Promise(resolve => { waiting = resolve; }); + const publisher = db.transaction(async tx => { + await tx.select().from(snapshots).where(eq(snapshots.id, snapshot!.id)).for("update"); + await tx.update(heartbeatRuns).set({ runnerProfileJson: { nativeExecutionInput: f.execution, + nativeRuntimeSkillPreservation: receipt } }).where(eq(heartbeatRuns.id, f.runId)); + locked(); await release; + }); + await acquired; + const debug = db.$client.options.debug; + db.$client.options.debug = (_connection, query) => { if (query.includes('from "company_skill_repository_snapshots"') && query.includes("for update")) waiting(); }; + try { + const collection = skillPackageMaintenance(db, storage).collectOrphans({ companyId: f.companyId, apply: true }); + await deleting; unlock(); await publisher; + expect((await collection).removed).toBe(0); + } finally { unlock(); db.$client.options.debug = debug; } + expect(await hasLocalRuntimeSkillObligations(db)).toBe(false); + expect(await db.select().from(snapshots).where(and(eq(snapshots.companyId, f.companyId), eq(snapshots.id, snapshot!.id)))) + .toHaveLength(1); + }); +}); diff --git a/server/src/__tests__/skill-package-store.test.ts b/server/src/__tests__/skill-package-store.test.ts index d06071c862c..3c11db094f6 100644 --- a/server/src/__tests__/skill-package-store.test.ts +++ b/server/src/__tests__/skill-package-store.test.ts @@ -510,14 +510,25 @@ describe.skipIf(!support.supported)("external skill packages", () => { const upload = { token: randomUUID(), provider: "local_disk" as const, archiveKey: `${companyId}/old/archive`, manifestKey: `${companyId}/old/manifest`, expiresAt: new Date(0).toISOString() }; const [row] = await db.insert(snapshots).values({ companyId, contentHash: skillSnapshotHash(files), createdAt: new Date(0), upload }).returning(); const renewed = { ...upload, token: randomUUID(), expiresAt: new Date(Date.now() + 600_000).toISOString() }; - const racingDb = new Proxy(db, { get(target, key) { - if (key !== "delete") return Reflect.get(target, key); - return (table: typeof snapshots) => ({ where: (predicate: Parameters["where"]>[0]) => ({ returning: async () => { - await db.update(snapshots).set({ upload: renewed }).where(eq(snapshots.id, row!.id)); - return db.delete(table).where(predicate).returning({ id: snapshots.id }); - } }) }); - } }); - expect((await skillPackageMaintenance(racingDb, storage).collectOrphans({ companyId, apply: true })).removed).toBe(0); + let unlock!: () => void, locked!: () => void, waiting!: () => void; + const release = new Promise(resolve => { unlock = resolve; }); + const acquired = new Promise(resolve => { locked = resolve; }); + const deleting = new Promise(resolve => { waiting = resolve; }); + const renewal = withDedicatedDbConnection(db, connection => connection.transaction(async tx => { + await tx.select().from(snapshots).where(eq(snapshots.id, row!.id)).for("update"); + await tx.update(snapshots).set({ upload: renewed }).where(eq(snapshots.id, row!.id)); + locked(); await release; + })); + await acquired; + const debug = db.$client.options.debug; + db.$client.options.debug = (_connection, query) => { + if (query.includes('from "company_skill_repository_snapshots"') && query.includes("for update")) waiting(); + }; + try { + const collection = skillPackageMaintenance(db, storage).collectOrphans({ companyId, apply: true }); + await deleting; unlock(); await renewal; + expect((await collection).removed).toBe(0); + } finally { unlock(); db.$client.options.debug = debug; } const [saved] = await db.select().from(snapshots).where(eq(snapshots.id, row!.id)); expect(saved!.upload).toEqual(renewed); }); diff --git a/server/src/services/ephemeral-server-contract.ts b/server/src/services/ephemeral-server-contract.ts index ca9209eadc9..f5b84347b78 100644 --- a/server/src/services/ephemeral-server-contract.ts +++ b/server/src/services/ephemeral-server-contract.ts @@ -7,6 +7,7 @@ import { skillInstallationMaintenance } from "./skill-installation-maintenance.j import { hasLocalNativeHarnessObligations } from "./native-runtime/native-harness-backup-migration.js"; import { hasLocalNativeControllerObligations } from "./native-runtime/native-controller-migration.js"; import { hasUnsettledLocalInstructionCopies } from "./agent-instruction-copy-preservation.js"; +import { hasLocalRuntimeSkillObligations } from "./native-runtime/runtime-skill-preservation.js"; export const ephemeralServerEnabled = () => process.env.PAPERCLIP_EPHEMERAL_SERVER === "true"; @@ -70,6 +71,7 @@ export async function readEphemeralServerDataEligibility(db: Db, options: { purp if (localTrace) blockers.push("provider_traces_require_migration"); if (await hasLocalNativeHarnessObligations(db)) blockers.push("native_harness_backups_require_migration"); if (await hasLocalNativeControllerObligations(db)) blockers.push("native_controllers_require_migration"); + if (await hasLocalRuntimeSkillObligations(db)) blockers.push("runtime_skill_assets_require_migration"); const skills = skillInstallationMaintenance(db); let after: string | undefined; for (;;) { diff --git a/server/src/services/native-runtime/runtime-context.ts b/server/src/services/native-runtime/runtime-context.ts index 77e7c3d991f..4426ce58990 100644 --- a/server/src/services/native-runtime/runtime-context.ts +++ b/server/src/services/native-runtime/runtime-context.ts @@ -34,6 +34,7 @@ import { agentInstructionsService, agentInstructionsBundleMode } from "../agent- import { agentInstructionRevisionService } from "../agent-instruction-revisions.js"; import { toolAccessService } from "../tool-access.js"; import { filterResolvedGitHubConnectionsForRun } from "../git-credentials.js"; +import { rehydratePreservedRuntimeSkill } from "./runtime-skill-preservation.js"; const MAX_ASSET_FILES = 10_000; const MAX_ASSET_BYTES = 64 * 1024 * 1024; @@ -203,8 +204,10 @@ export async function rehydratePinnedSkillAssets( binding: { companyId: string; runId: string; agentId: string }, context: NativeRuntimeContextSnapshot, ): Promise { - if (!context.skills.some(skill => skill.versionId)) return context; + if (!context.skills.length) return context; const [run] = await db.select({ responsibleUserId: heartbeatRuns.responsibleUserId, + id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId, + nativeIssueId: heartbeatRuns.nativeIssueId, runnerProfileJson: heartbeatRuns.runnerProfileJson, contextSnapshot: heartbeatRuns.contextSnapshot }).from(heartbeatRuns).where(and( eq(heartbeatRuns.companyId, binding.companyId), eq(heartbeatRuns.id, binding.runId), eq(heartbeatRuns.agentId, binding.agentId))); if (!run) throw new Error("Skill recovery has no matching admitted run"); @@ -215,7 +218,15 @@ export async function rehydratePinnedSkillAssets( const pins = run.contextSnapshot?.skillVersions; const skills: NativeRuntimeContextSnapshot["skills"] = []; for (const skill of context.skills) { - if (!skill.versionId) { skills.push(skill); continue; } + if (!skill.versionId) { + // Newly admitted local assets remain usable during an ordinary rolling + // upgrade. A migrated asset uses its exact durable receipt on cold hosts. + const preserved = run.runnerProfileJson?.nativeRuntimeSkillPreservation; + if (!preserved) { skills.push(skill); continue; } + const bundle = await rehydratePreservedRuntimeSkill(db, run, skill, materializeSkillAsset); + skills.push({ ...skill, bundle }); + continue; + } if (pins !== undefined && (!pins || typeof pins !== "object" || Array.isArray(pins) || (pins as Record)[skill.key] !== skill.versionId)) { throw new Error("Skill recovery differs from the admitted version selection"); diff --git a/server/src/services/native-runtime/runtime-skill-preservation-reference.ts b/server/src/services/native-runtime/runtime-skill-preservation-reference.ts new file mode 100644 index 00000000000..a09a0fd5042 --- /dev/null +++ b/server/src/services/native-runtime/runtime-skill-preservation-reference.ts @@ -0,0 +1,75 @@ +import { sql } from "drizzle-orm"; +import { heartbeatRuns, companySkillRepositorySnapshots } from "@paperclipai/db"; +import { nativeSha256 } from "./canonical.js"; +import { parseNativeExecutionInput, type NativeRuntimeContextSnapshot } from "../../vendor/paperclip-runner/index.js"; + +export const RUNTIME_SKILL_PRESERVATION_KEY = "nativeRuntimeSkillPreservation"; +export const RUNTIME_SKILL_PRESERVATION_SCHEMA = "paperclip.runtime-skill-preservation.v1"; +export type PreservedRuntimeSkill = { + key: string; runtimeName: string; bundle: NativeRuntimeContextSnapshot["skills"][number]["bundle"]; + manifestText: string; packageSnapshotId: string; packageContentHash: string; +}; +export type RuntimeSkillPreservation = { + schema: typeof RUNTIME_SKILL_PRESERVATION_SCHEMA; inputDigest: string; bindingDigest: string; assets: PreservedRuntimeSkill[]; +}; +export function admittedRuntimeSkills(row: { id: string; companyId: string; agentId: string; nativeIssueId: string | null; + runnerProfileJson: Record | null }) { + const input = parseNativeExecutionInput(row.runnerProfileJson?.nativeExecutionInput); + if (input.binding.companyId !== row.companyId || input.binding.runId !== row.id + || input.binding.agentId !== row.agentId || input.binding.issueId !== row.nativeIssueId || !("runtimeContext" in input)) { + throw new Error("runtime_skill_preservation_binding_changed"); + } + return { input, skills: input.runtimeContext.skills.filter(skill => skill.versionId === null), + inputDigest: nativeSha256(row.runnerProfileJson?.nativeExecutionInput), bindingDigest: nativeSha256(input.binding) }; +} +export function sameRuntimeSkill(a: Pick, + b: Pick) { + return a.key === b.key && a.runtimeName === b.runtimeName && a.bundle.digest === b.bundle.digest + && a.bundle.manifestDigest === b.bundle.manifestDigest && a.bundle.fileCount === b.bundle.fileCount + && a.bundle.totalBytes === b.bundle.totalBytes; +} +export function readRuntimeSkillPreservation(row: Parameters[0]): RuntimeSkillPreservation | null { + const admitted = admittedRuntimeSkills(row); + const receipt = row.runnerProfileJson?.[RUNTIME_SKILL_PRESERVATION_KEY] as RuntimeSkillPreservation | undefined; + // Explicit continuation may revise a completion contract while preserving the + // exact admitted asset set. Retain the original capture digest as provenance; + // bind readers to the current run scope and unchanged immutable selection. + if (receipt?.schema !== RUNTIME_SKILL_PRESERVATION_SCHEMA || !/^[a-f0-9]{64}$/.test(receipt.inputDigest) + || receipt.bindingDigest !== admitted.bindingDigest + || !Array.isArray(receipt.assets) || receipt.assets.length !== admitted.skills.length) return null; + const ids = new Set(); + for (let i = 0; i < admitted.skills.length; i++) { + const asset = receipt.assets[i]; + if (!asset || !sameRuntimeSkill(asset, admitted.skills[i]!) || typeof asset.manifestText !== "string" + || !/^[a-f0-9-]{36}$/i.test(asset.packageSnapshotId) || !/^[a-f0-9]{64}$/.test(asset.packageContentHash) + || ids.has(asset.key)) return null; + ids.add(asset.key); + } + return receipt; +} + +/** Ordinary admission has no input yet. Only a concrete admitted local asset is + * a durability obligation; missing transient version pins alone are not one. */ +export function hasUnversionedRuntimeSkillInputSql() { + const skills = sql`${heartbeatRuns.runnerProfileJson} #> '{nativeExecutionInput,runtimeContext,skills}'`; + return sql`${heartbeatRuns.runtimeMode} = 'native' + and ${heartbeatRuns.runnerProfileJson} #> '{nativeExecutionInput,runtimeContext}' is not null + and (jsonb_typeof(${skills}) is distinct from 'array' or exists ( + select 1 from jsonb_array_elements(case when jsonb_typeof(${skills}) = 'array' then ${skills} else '[]'::jsonb end) s + where s->>'versionId' is null or s->>'versionId' !~* '^[a-f0-9]{8}(-[a-f0-9]{4}){3}-[a-f0-9]{12}$'))`; +} + +/** JSON receipts are references just like version FKs. Unknown receipt shapes + * retain the company's packages rather than allowing destructive inference. */ +export function runtimeSkillPackageUnreferencedSql() { + return sql`not exists (select 1 from ${heartbeatRuns} r where r.company_id = ${companySkillRepositorySnapshots.companyId} + and r.runner_profile_json ? 'nativeRuntimeSkillPreservation' + and (r.runner_profile_json->'nativeRuntimeSkillPreservation'->>'schema' is distinct from ${RUNTIME_SKILL_PRESERVATION_SCHEMA} + or coalesce(r.runner_profile_json->'nativeRuntimeSkillPreservation'->>'inputDigest', '') !~ '^[a-f0-9]{64}$' + or jsonb_typeof(r.runner_profile_json->'nativeRuntimeSkillPreservation'->'assets') is distinct from 'array' + or exists (select 1 from jsonb_array_elements(case + when jsonb_typeof(r.runner_profile_json->'nativeRuntimeSkillPreservation'->'assets') = 'array' + then r.runner_profile_json->'nativeRuntimeSkillPreservation'->'assets' else '[]'::jsonb end) a + where coalesce(a->>'packageSnapshotId', '') !~* '^[a-f0-9]{8}(-[a-f0-9]{4}){3}-[a-f0-9]{12}$' + or a->>'packageSnapshotId' = ${companySkillRepositorySnapshots.id}::text)))`; +} diff --git a/server/src/services/native-runtime/runtime-skill-preservation.ts b/server/src/services/native-runtime/runtime-skill-preservation.ts new file mode 100644 index 00000000000..bc2d9bcc605 --- /dev/null +++ b/server/src/services/native-runtime/runtime-skill-preservation.ts @@ -0,0 +1,213 @@ +import { constants } from "node:fs"; +import fs from "node:fs/promises"; +import path from "node:path"; +import { createHash } from "node:crypto"; +import { and, asc, eq, gt, or, sql } from "drizzle-orm"; +import { heartbeatRuns, companySkillRepositorySnapshots as snapshots, type Db } from "@paperclipai/db"; +import { resolvePaperclipInstanceRoot } from "../../home-paths.js"; +import type { StorageService } from "../../storage/types.js"; +import { skillPackageStore, withSkillPackageReadScope } from "../skill-package-store.js"; +import { assertSkillSnapshotPath, snapshotFile, skillSnapshotHash } from "../skill-snapshot.js"; +import { ensureSkillCacheDirectory, skillRunNeedsFilesSql } from "../skill-version-cache.js"; +import { verifyFileManifest } from "../file-manifest-cache.js"; +import { logActivity, publishActivity, type ActivityPublication } from "../activity-log.js"; +import { withRuntimeAssetUse } from "./runtime-asset-retention.js"; +import { admittedRuntimeSkills, hasUnversionedRuntimeSkillInputSql, readRuntimeSkillPreservation, + RUNTIME_SKILL_PRESERVATION_KEY, RUNTIME_SKILL_PRESERVATION_SCHEMA, sameRuntimeSkill, + type PreservedRuntimeSkill } from "./runtime-skill-preservation-reference.js"; +import type { NativeRuntimeAssetReference, NativeRuntimeContextSnapshot } from "../../vendor/paperclip-runner/index.js"; + +const hash = (value: string | Buffer) => createHash("sha256").update(value).digest("hex"); +const retainedRuntimeSkillsSql = () => and(hasUnversionedRuntimeSkillInputSql(), or(skillRunNeedsFilesSql(), + sql`exists (select 1 from agent_task_sessions s where s.company_id = ${heartbeatRuns.companyId} + and s.agent_id = ${heartbeatRuns.agentId} and s.last_run_id = ${heartbeatRuns.id})`)); +type ManifestFile = { path: string; sha256: string; mode: number; size: number }; +function manifestFiles(asset: NativeRuntimeAssetReference, text: string): ManifestFile[] { + if (Buffer.byteLength(text) > 4 * 1024 * 1024 || hash(text) !== asset.manifestDigest) throw new Error("runtime_skill_manifest_digest_mismatch"); + const manifest = JSON.parse(text); + if (manifest.schema !== "paperclip.runtime-asset-manifest.v1" || manifest.digest !== asset.digest + || manifest.fileCount !== asset.fileCount || manifest.totalBytes !== asset.totalBytes + || !Array.isArray(manifest.files) || manifest.files.length !== asset.fileCount + || manifest.files.length > 10_000 || asset.totalBytes > 64 * 1024 * 1024) throw new Error("runtime_skill_manifest_invalid"); + const files: ManifestFile[] = manifest.files; + const seen = new Set(); + for (const file of files) { + assertSkillSnapshotPath(file.path); + if (seen.has(file.path) || !/^[a-f0-9]{64}$/.test(file.sha256) || !Number.isSafeInteger(file.size) || file.size < 0 + || !Number.isInteger(file.mode) || file.mode < 0 || (file.mode & ~0o555) !== 0) throw new Error("runtime_skill_manifest_invalid"); + seen.add(file.path); + } + if (hash(JSON.stringify(files)) !== asset.digest || files.reduce((n, f) => n + f.size, 0) !== asset.totalBytes) throw new Error("runtime_skill_manifest_invalid"); + return files; +} +async function readLocalManifest(asset: NativeRuntimeAssetReference) { + const root = path.join(resolvePaperclipInstanceRoot(), "runtime-context-assets"); + const expected = path.join(root, "bundles", asset.digest); + if (asset.rootPath !== expected) throw new Error("runtime_skill_source_path_mismatch"); + await ensureSkillCacheDirectory(expected); + await ensureSkillCacheDirectory(path.join(root, "manifests")); + const handle = await fs.open(path.join(root, "manifests", `${asset.digest}.json`), constants.O_RDONLY | constants.O_NOFOLLOW); + let text: string; + try { + const stat = await handle.stat(); + if (!stat.isFile() || stat.nlink !== 1 || stat.size > 4 * 1024 * 1024) throw new Error("runtime_skill_manifest_invalid"); + text = await handle.readFile("utf8"); + } finally { await handle.close(); } + const entries = manifestFiles(asset, text); + return { expected, manifestText: text, entries }; +} +async function readSource(asset: NativeRuntimeAssetReference) { + const { expected, manifestText, entries } = await readLocalManifest(asset); + if (!await verifyFileManifest(expected, entries.map(f => ({ ...f, executable: Boolean(f.mode & 0o111) })))) throw new Error("runtime_skill_source_changed"); + const files = []; + for (const entry of entries) { + const handle = await fs.open(path.join(expected, entry.path), constants.O_RDONLY | constants.O_NOFOLLOW); + try { + const stat = await handle.stat(); + if (!stat.isFile() || stat.nlink !== 1 || stat.size !== entry.size || (stat.mode & 0o777) !== entry.mode) throw new Error("runtime_skill_source_changed"); + const bytes = await handle.readFile(); + if (bytes.length !== entry.size || hash(bytes) !== entry.sha256) throw new Error("runtime_skill_source_changed"); + files.push(snapshotFile(entry.path, entry.path === "SKILL.md" ? "skill" : "other", bytes, Boolean(entry.mode & 0o111))); + } finally { await handle.close(); } + } + return { manifestText, files }; +} + +/** Copy immutable admitted assets, never resolve a run's business/recovery state + * or infer that its provider stopped. Local source bytes and null pins survive. */ +export async function preserveRuntimeSkillAssets(db: Db, input: { companyId: string; runId: string; signal?: AbortSignal }, storage?: StorageService) { + const where = and(eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.id, input.runId)); + const [source] = await db.select().from(heartbeatRuns).where(where); + if (!source || source.runtimeMode !== "native") throw new Error("runtime_skill_source_unavailable"); + const admitted = admittedRuntimeSkills(source); + const store = skillPackageStore(db, storage); + const existing = readRuntimeSkillPreservation(source); + if (source.runnerProfileJson?.[RUNTIME_SKILL_PRESERVATION_KEY] != null && !existing) throw new Error("runtime_skill_preservation_receipt_invalid"); + if (existing) { + for (const asset of existing.assets) { + manifestFiles(asset.bundle, asset.manifestText); + const row = await store.descriptor(input.companyId, asset.packageSnapshotId); + if (row.storage?.provider !== "s3" || row.storage.durable !== true || row.contentHash !== asset.packageContentHash + || !await store.verifyDurable(input.companyId, asset.packageSnapshotId, input.signal)) throw new Error("runtime_skill_package_not_durable"); + } + return { preserved: existing.assets.length }; + } + const assets: PreservedRuntimeSkill[] = []; + for (const skill of admitted.skills) { + input.signal?.throwIfAborted(); + const captured = await withRuntimeAssetUse(skill.bundle.digest, () => readSource(skill.bundle), true); + const packageSnapshotId = await store.prepare(input.companyId, captured.files, true, input.signal); + if (!await store.verifyDurable(input.companyId, packageSnapshotId, input.signal)) throw new Error("runtime_skill_package_not_durable"); + assets.push({ key: skill.key, runtimeName: skill.runtimeName, bundle: skill.bundle, manifestText: captured.manifestText, + packageSnapshotId, packageContentHash: skillSnapshotHash(captured.files) }); + } + if (!assets.length) return { preserved: 0 }; + const publications: ActivityPublication[] = []; + await db.transaction(async tx => { + const [current] = await tx.select().from(heartbeatRuns).where(where).for("update"); + if (!current || admittedRuntimeSkills(current).inputDigest !== admitted.inputDigest || current.status !== source.status + || current.finishedAt?.getTime() !== source.finishedAt?.getTime()) throw new Error("runtime_skill_preservation_binding_changed"); + // Collector takes these same row locks before a fresh reference check. If + // deletion won, publication fails and leaves every local asset untouched. + for (const asset of [...assets].sort((a, b) => a.packageSnapshotId.localeCompare(b.packageSnapshotId))) { + const [row] = await tx.select().from(snapshots).where(and(eq(snapshots.companyId, input.companyId), eq(snapshots.id, asset.packageSnapshotId))).for("update"); + if (!row?.storage || row.storage.provider !== "s3" || row.storage.durable !== true + || row.contentHash !== asset.packageContentHash) throw new Error("runtime_skill_package_not_durable"); + } + await tx.update(heartbeatRuns).set({ runnerProfileJson: { ...current.runnerProfileJson, + [RUNTIME_SKILL_PRESERVATION_KEY]: { schema: RUNTIME_SKILL_PRESERVATION_SCHEMA, inputDigest: admitted.inputDigest, + bindingDigest: admitted.bindingDigest, assets } } }).where(where); + await logActivity(tx as unknown as Db, { companyId: input.companyId, actorType: "system", actorId: "runtime-skill-preservation", + action: "native_runtime_skill_assets_preserved", entityType: "heartbeat_run", entityId: input.runId, + agentId: source.agentId, runId: input.runId, details: { inputDigest: admitted.inputDigest, assets: assets.length } }, publications); + }); + for (const publication of publications) publishActivity(publication); + return { preserved: assets.length }; +} + +export async function hasLocalRuntimeSkillObligations(db: Db): Promise { + let after: string | undefined; + const checkedPackages = new Map(); + for (;;) { + const rows = await db.select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId, + nativeIssueId: heartbeatRuns.nativeIssueId, runnerProfileJson: sql>`jsonb_build_object( + 'nativeExecutionInput', ${heartbeatRuns.runnerProfileJson}->'nativeExecutionInput', + 'nativeRuntimeSkillPreservation', ${heartbeatRuns.runnerProfileJson}->'nativeRuntimeSkillPreservation')` }) + .from(heartbeatRuns).where(and(retainedRuntimeSkillsSql(), after ? gt(heartbeatRuns.id, after) : undefined)) + .orderBy(asc(heartbeatRuns.id)).limit(100); + for (const row of rows) { + let receipt; + try { receipt = readRuntimeSkillPreservation(row); } catch { return true; } + if (!receipt) return true; + for (const asset of receipt.assets) { + try { manifestFiles(asset.bundle, asset.manifestText); } catch { return true; } + const key = `${row.companyId}:${asset.packageSnapshotId}`; + let snapshot = checkedPackages.get(key); + if (!snapshot) { + [snapshot] = await db.select({ hash: snapshots.contentHash, storage: snapshots.storage }).from(snapshots) + .where(and(eq(snapshots.companyId, row.companyId), eq(snapshots.id, asset.packageSnapshotId))); + if (snapshot) checkedPackages.set(key, snapshot); + } + if (snapshot?.storage?.provider !== "s3" || snapshot.storage.durable !== true || snapshot.hash !== asset.packageContentHash) return true; + } + } + if (rows.length < 100) return false; + after = rows.at(-1)!.id; + } +} + +/** Existing bounded skill maintenance owns this optional online conversion; + * there is no scanner, new execution admission or controller takeover. */ +export async function migrateRuntimeSkillAssets(db: Db, options: { companyId?: string; after?: string; + limit?: number; apply?: boolean; signal?: AbortSignal } = {}, storage?: StorageService) { + const rows = await db.select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId }).from(heartbeatRuns) + .where(and(retainedRuntimeSkillsSql(), options.companyId ? eq(heartbeatRuns.companyId, options.companyId) : undefined, + options.after ? gt(heartbeatRuns.id, options.after) : undefined)).orderBy(asc(heartbeatRuns.id)) + .limit(Math.max(1, Math.min(50, options.limit ?? 10))); + const results: Array<{ runId: string; status: "candidate" | "preserved"; preserved?: number }> = []; + await withSkillPackageReadScope(db, async () => { + for (const row of rows) { + options.signal?.throwIfAborted(); + if (!options.apply) { results.push({ runId: row.id, status: "candidate" }); continue; } + const result = await preserveRuntimeSkillAssets(db, { companyId: row.companyId, runId: row.id, signal: options.signal }, storage); + results.push({ runId: row.id, status: "preserved", preserved: result.preserved }); + } + }); + return { inspected: rows.length, after: rows.at(-1)?.id ?? null, results }; +} + +export async function rehydratePreservedRuntimeSkill(db: Db, row: Parameters[0], + skill: NativeRuntimeContextSnapshot["skills"][number], materialize: (root: string) => Promise) { + const receipt = readRuntimeSkillPreservation(row); + const asset = receipt?.assets.find(entry => sameRuntimeSkill(entry, skill)); + if (!asset) throw new Error("runtime_skill_preservation_required"); + const files = manifestFiles(asset.bundle, asset.manifestText); + const store = skillPackageStore(db); + const descriptor = await store.descriptor(row.companyId, asset.packageSnapshotId); + if (descriptor.storage?.provider !== "s3" || descriptor.storage.durable !== true + || descriptor.contentHash !== asset.packageContentHash) throw new Error("runtime_skill_package_not_durable"); + // An intact disposable cache remains useful during a storage outage. Only + // use the current host's canonical cache, never the old host's recorded path. + const local = { ...skill.bundle, rootPath: path.join(resolvePaperclipInstanceRoot(), "runtime-context-assets", "bundles", skill.bundle.digest) }; + const healthy = await withRuntimeAssetUse(local.digest, async () => { + try { + const { expected, entries } = await readLocalManifest(local); + return await verifyFileManifest(expected, entries.map(file => ({ ...file, executable: Boolean(file.mode & 0o111) }))); + } catch (error) { + if (["ENOENT", "ENOTDIR"].includes((error as NodeJS.ErrnoException).code ?? "") + || error instanceof SyntaxError || (error instanceof Error && error.message.startsWith("runtime_skill_manifest_"))) return false; + throw error; + } + }); + if (healthy) return local; + return store.withIsolatedPackage(row.companyId, asset.packageSnapshotId, async (root, manifest) => { + if (manifest.files.length !== files.length || files.some(file => !manifest.files.some(entry => entry.path === file.path + && entry.sha256 === file.sha256 && entry.size === file.size && entry.executable === Boolean(file.mode & 0o111)))) throw new Error("runtime_skill_package_mismatch"); + // Package storage preserves executable bits; the authenticated native + // manifest additionally restores the exact original read/execute mode. + for (const file of files) await fs.chmod(path.join(root, file.path), file.mode); + const bundle = await materialize(root); + if (!sameRuntimeSkill({ ...skill, bundle }, asset)) throw new Error("runtime_skill_projection_mismatch"); + return bundle; + }, false); +} diff --git a/server/src/services/skill-package-maintenance.ts b/server/src/services/skill-package-maintenance.ts index 8f33396cfbe..7bc012f49bb 100644 --- a/server/src/services/skill-package-maintenance.ts +++ b/server/src/services/skill-package-maintenance.ts @@ -6,6 +6,7 @@ import { getStorageService, type StorageService } from "../storage/index.js"; import { skillPackageStore, skillVersionFileMap, skillVersionMetadataColumns } from "./skill-package-store.js"; import { packageSha256 } from "./skill-package-files.js"; import { skillFileBytes, skillSnapshotHash } from "./skill-snapshot.js"; +import { runtimeSkillPackageUnreferencedSql } from "./native-runtime/runtime-skill-preservation-reference.js"; type PageOptions = { id?: string; signal?: AbortSignal; companyId?: string; after?: string; limit?: number; apply?: boolean; clearLegacy?: boolean }; const pageSize = (options: PageOptions) => Math.min(50, Math.max(1, options.limit ?? 10)); @@ -92,7 +93,8 @@ export function skillPackageMaintenance(db: Db, storage?: StorageService) { } async function collectOrphans(options: PageOptions = {}) { - const unused = sql`not exists (select 1 from ${versions} where ${versions.repositorySnapshotId} = ${snapshots.id} or ${versions.packageSnapshotId} = ${snapshots.id})`; + const unused = sql`not exists (select 1 from ${versions} where ${versions.repositorySnapshotId} = ${snapshots.id} or ${versions.packageSnapshotId} = ${snapshots.id}) + and ${runtimeSkillPackageUnreferencedSql()}`; const expired = sql`(${snapshots.upload} is null or (${snapshots.upload}->>'expiresAt')::timestamptz < now())`; const rows = await db.select({ id: snapshots.id }).from(snapshots).where(and( options.companyId ? eq(snapshots.companyId, options.companyId) : undefined, @@ -103,10 +105,14 @@ export function skillPackageMaintenance(db: Db, storage?: StorageService) { )).orderBy(asc(snapshots.id)).limit(pageSize(options)); let removed = 0; if (options.apply) for (const row of rows) { - // FK locking wins against a concurrent publication. The trigger persists - // deletion receipts in the same transaction, including company cascades. - const deleted = await db.delete(snapshots).where(and(eq(snapshots.id, row.id), expired, unused)).returning({ id: snapshots.id }); - removed += deleted.length; + removed += await db.transaction(async tx => { + // JSON preservation references have no FK. Acquire the publisher's + // package row lock, then inspect references in a NEW READ COMMITTED + // statement so a just-committed receipt cannot be missed after waiting. + await tx.select({ id: snapshots.id }).from(snapshots).where(eq(snapshots.id, row.id)).for("update"); + const deleted = await tx.delete(snapshots).where(and(eq(snapshots.id, row.id), expired, unused)).returning({ id: snapshots.id }); + return deleted.length; + }); } return { inspected: rows.length, removed, after: rows.at(-1)?.id ?? null }; }