mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-10 12:37:55 +01:00
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 <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip authored and GitHub committed
1 parent
6d1ecfc420
commit
a3eecf1fc9
9 files changed
+605
-17
No files matched your search
@@ -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<NativeRuntimeContextSnapshot> {
|
||||
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<string, unknown>)[skill.key] !== skill.versionId)) {
|
||||
throw new Error("Skill recovery differs from the admitted version selection");
|
||||
|
||||
@@ -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<string, unknown> | 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<PreservedRuntimeSkill, "key" | "runtimeName" | "bundle">,
|
||||
b: Pick<PreservedRuntimeSkill, "key" | "runtimeName" | "bundle">) {
|
||||
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<typeof admittedRuntimeSkills>[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<string>();
|
||||
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)))`;
|
||||
}
|
||||
@@ -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<string>();
|
||||
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<boolean> {
|
||||
let after: string | undefined;
|
||||
const checkedPackages = new Map<string, { hash: string; storage: typeof snapshots.$inferSelect.storage }>();
|
||||
for (;;) {
|
||||
const rows = await db.select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId,
|
||||
nativeIssueId: heartbeatRuns.nativeIssueId, runnerProfileJson: sql<Record<string, unknown>>`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<typeof admittedRuntimeSkills>[0],
|
||||
skill: NativeRuntimeContextSnapshot["skills"][number], materialize: (root: string) => Promise<NativeRuntimeAssetReference>) {
|
||||
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);
|
||||
}
|
||||
Reference in new issue
Block a user