mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-09 07:10:34 +01:00
fix(runtime): save dynamic skill bundles before native admission (#19729)
## Thinking Path > - Paperclip manages agents, tasks, and their saved execution state. > - A replacement server must restore each admitted skill from shared storage. > - Installed company skills already have immutable versions. > - Connector skills add assigned resources and tool definitions for each run. > - Those dynamic bytes keep a null version ID and differ from the static skill. > - Native admission saved their input before it saved a durable package receipt. > - This change stages the exact packages first and publishes the input and receipt together. > - A new connector turn can then survive an empty server home. ## Linked Issues or Issue Description Refs #19675 for the existing runtime asset preservation owner. **What happened?** A new native run with an assigned Slack connector creates a concrete local skill bundle after the old runtime assets were migrated. Admission saves its execution input without a preservation receipt. The new run creates a local durability dependency again. **Expected behavior** With S3 configured, the exact dynamic bundle must be durable before the execution input becomes visible. Its null version ID and admitted bytes must remain unchanged. Local storage must keep its volume-backed behavior. **Steps to reproduce** 1. Configure S3 and a native agent with an assigned Slack resource. 2. Prepare a new native task through the normal runtime selection owner. 3. Inspect its runtime skill receipt or restore its skills with an empty server home. The previous owner saved no receipt. **Paperclip version or commit** Reproduced on master `d1ee25d0f7`. The fix is tested on `4a3396d9b9`. **Deployment mode** S3-configured server. Local tests use PostgreSQL 18.1 and a private disk transport with the S3 receipt contract. ## What Changed - Extend the existing runtime asset owner with package staging and transaction-bound receipt validation. - Stage exact unversioned skill packages before native admission. Keep storage I/O outside the run transaction. - Publish the frozen input and receipt together under the existing run, cancellation, and package collection locks. - Reject changed input bindings and retired package metadata. - Reuse valid same-run receipts without reading the previous host's bundle path or requiring a new object read. - Keep existing online preservation and local storage behavior. - Add a bounded preparation span and document its work. ## Verification - Genuine unchanged-production RED: the actual Slack overlay and real admission transaction save no preservation receipt. - Final rebased head: 68 tests pass in three suites in 25.36 seconds. These include all runtime-selection, runtime-preservation, and connector controls. - The new positive checks that no execution input or coordinator is visible during package readback. It restores exact skill and tool bytes, modes, digests, and null version IDs in a cold home. - Guards cover an uncertain storage failure, cancellation during upload, a changed binding, retired package metadata, and local mode. An existing receipt remains usable after removing the original bundle and during a storage outage. - The canonical PostgreSQL wrapper uses one owned disposable PostgreSQL 18.1 container. Each suite's database is cleaned up. Real package archives, streams, and readback run locally; no external provider or model is called. - Strict server TypeScript, module boundaries, changed-line private-data checks, and `git diff --check` pass. - [Final-head CI](https://github.com/paperclipai/paperclip/actions/runs/42337473062) passes all required build, typecheck, test, and Runner gates. Fresh Greptile is 5/5 with no open review threads. ## Risks New S3-backed admission now requires its dynamic packages to be saved before execution can proceed. A storage failure can therefore stop preparation. Existing resource cleanup keeps its authority. Failed or cancelled staging retains original files and the package owner's pending receipts. Same-run recovery retains its original capture provenance and cache behavior. The tests replace the external S3 transport; they do not qualify live S3 permissions or provider execution. No schema, API, workflow, company skill version, or provider protocol changes are included. ## Model Used OpenAI Codex (GPT-6 family; the exact runtime model ID and context capacity are not exposed in this session), with reasoning, repository editing, code execution, and tool use. ## 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
efc0b359da
commit
9efee7ee16
6 files changed
+205
-23
No files matched your search
@@ -359,7 +359,7 @@ attempt or owning phase:
|
||||
| `authorization.base`, `.responsible_user`, `.authorizing_user`, `.task_privacy`, `.project_privacy` | The existing base decision, responsible/authorizing-user intersections, and applicable app-record privacy decisions beneath `workspace.access.task_authorization` or `workspace.access.authorization`. These details are scoped to workspace authorization. |
|
||||
| `workspace.notification`, `workspace.write` | Notification statements and record writes. |
|
||||
| `workspace.materialize.*`, `workspace.checkout.*` | Private source-directory setup and identity verification, existing checkout lookup, Git credential resolution, clone, optional pristine-object cache lookup/publication (`cache_lookup`, `cache_publish`), requested-ref checkout, and atomic publication. The enclosing `heartbeat.prepare.workspace_source_materialize` includes the complete source-materialization path. |
|
||||
| `runtime_selection.*` | Completion contracts, session/recovery reads, profiles, interaction materialization, backend discovery, bootstrap, run/coordinator writes, transaction entry/settlement, and workspace synchronization. |
|
||||
| `runtime_selection.*` | Completion contracts, session/recovery reads, profiles, interaction materialization, backend discovery, dynamic skill package staging (`skill_assets`), bootstrap, run/coordinator writes, transaction entry/settlement, and workspace synchronization. |
|
||||
| `runtime_context.instructions`, `.skills`, `.mcp` | The three concurrent runtime-context branches. Their durations overlap. |
|
||||
| `runtime_context.asset_*` | Source reads, manifest hashing, cache lookup, cold materialization, and exact file verification on cold and warm paths. |
|
||||
| `skills.inventory`, `.wait`, `.bundled`, `.releases`, `.folders`, `.local_sources` | Company skill reconciliation, waiting for an existing company refresh, and its bundled imports, release verification, folder, and local-source work beneath runtime skill preparation. Inventory children overlap their parent. |
|
||||
|
||||
@@ -40,6 +40,16 @@ 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.
|
||||
|
||||
Dynamic connector skills include run-specific resources and tool definitions.
|
||||
They retain null version IDs because their bytes differ from the static company
|
||||
skill. With S3 configured, native runtime selection stages and verifies their exact
|
||||
packages before its admission transaction. That transaction checks the frozen input
|
||||
and cancellation state, locks the package rows against collection, and publishes
|
||||
the input and its preservation receipt together. Storage failure leaves the input
|
||||
unpublished. Existing receipts retain their capture provenance on same-run recovery;
|
||||
hot cache reads can continue during a storage outage. Local storage keeps its
|
||||
existing volume-backed behavior.
|
||||
|
||||
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
|
||||
|
||||
@@ -75,7 +75,7 @@ export type ExecutionPhase =
|
||||
| "runtime_selection.resume" | "runtime_selection.recovery_profile" | "runtime_selection.profile" | "runtime_selection.interactions"
|
||||
| "runtime_selection.provider_discovery" | "runtime_selection.bootstrap" | "runtime_selection.run_write" | "runtime_selection.finalization_write"
|
||||
| "runtime_selection.transaction" | "runtime_selection.transaction.entry" | "runtime_selection.transaction.commit" | "runtime_selection.transaction.rollback"
|
||||
| "runtime_selection.row_lock" | "runtime_selection.workspace_sync"
|
||||
| "runtime_selection.row_lock" | "runtime_selection.workspace_sync" | "runtime_selection.skill_assets"
|
||||
| "runtime_context.instructions" | "runtime_context.skills" | "runtime_context.mcp"
|
||||
| "runtime_context.asset_read" | "runtime_context.asset_manifest" | "runtime_context.asset_cache" | "runtime_context.asset_materialize" | "runtime_context.asset_verify"
|
||||
| "workspace_sync.snapshot_git" | "workspace_sync.snapshot_baseline" | "workspace_sync.pack" | "workspace_sync.seed_verify" | "workspace_sync.stage_workspace" | "workspace_sync.restore"
|
||||
|
||||
@@ -1,16 +1,24 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { mkdir, mkdtemp, readFile, rm, writeFile } from "node:fs/promises";
|
||||
import { mkdir, mkdtemp, readFile, rm, stat, writeFile } from "node:fs/promises";
|
||||
import { execFile as execFileCallback } from "node:child_process";
|
||||
import { promisify } from "node:util";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { activityLog, agents, authUsers, companyMemberships, principalPermissionGrants, environments, environmentLeases, companies, completionContracts, createDb, applyPendingMigrations, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, executionWorkspaces, heartbeatRuns, issues, nativeRunFinalizations, workspaceOperations, type Db } from "@paperclipai/db";
|
||||
import { activityLog, agents, authUsers, companyMemberships, companySkillRepositorySnapshots, principalPermissionGrants, environments, environmentLeases, companies, completionContracts, createDb, applyPendingMigrations, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, executionWorkspaces, heartbeatRuns, issues, nativeRunFinalizations, workspaceOperations, type Db } from "@paperclipai/db";
|
||||
import { eq, sql } from "drizzle-orm";
|
||||
import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js";
|
||||
import { resolveExternalTestDatabaseUrl } from "../../__tests__/helpers/external-test-database.js";
|
||||
import { CHAT_CONTROL_RECOVERY_ADMISSION_KEY } from "../chat-control-recovery-stop.js";
|
||||
import { buildNativeRuntimeContext } from "../native-runtime/runtime-context.js";
|
||||
import { buildNativeRuntimeContext, materializeSelectedSkills, rehydratePinnedSkillAssets } from "../native-runtime/runtime-context.js";
|
||||
import { applyConnectorSkills } from "../connector-runtime.js";
|
||||
import * as storageModule from "../../storage/index.js";
|
||||
import { createStorageService } from "../../storage/service.js";
|
||||
import { createLocalDiskStorageProvider } from "../../storage/local-disk-provider.js";
|
||||
import { readRuntimeSkillPreservation } from "../native-runtime/runtime-skill-preservation-reference.js";
|
||||
import { hasLocalRuntimeSkillObligations } from "../native-runtime/runtime-skill-preservation.js";
|
||||
import * as runtimeSkillPreservation from "../native-runtime/runtime-skill-preservation.js";
|
||||
import { removeRuntimeAssetTree } from "../native-runtime/runtime-asset-retention.js";
|
||||
import { nativeRuntimeContextFixture } from "../native-runtime/runtime-context.test-fixture.js";
|
||||
import { prepareNativeWorkspaceSync } from "../native-runtime/native-workspace-sync.js";
|
||||
import { resolveHeartbeatNativeRuntimeMode } from "../native-runtime/runtime-mode.js";
|
||||
@@ -87,6 +95,113 @@ describe.skipIf(!support.supported)("heartbeat runtime selection boundary", () =
|
||||
const coordinators = (id: string) => db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, id));
|
||||
const sandbox = { kind: "remote" as const, transport: "sandbox" as const, remoteCwd: "/workspace", reusableLeaseConfigured: false };
|
||||
|
||||
async function connectorFixture(durable = true) {
|
||||
const f = await fixture(), { input } = f;
|
||||
const home = await mkdtemp(join(tmpdir(), "selection-connector-assets-"));
|
||||
vi.stubEnv("PAPERCLIP_HOME", home);
|
||||
const config = await applyConnectorSkills({}, [], [{ key: "slack", label: "Slack", skillKey: "paperclipai/paperclip/slack",
|
||||
resources: [{ id: "synthetic-channel", label: "Fixture channel", connectionId: randomUUID() }],
|
||||
tools: [{ name: "fixture_read", description: "Read synthetic messages", inputSchema: { type: "object" } }] }]);
|
||||
input.config.runtimeConfig = config;
|
||||
input.config.runtimeSkillEntries = config.paperclipRuntimeSkills;
|
||||
vi.mocked(buildNativeRuntimeContext).mockImplementation(async () => {
|
||||
const context = nativeRuntimeContextFixture();
|
||||
context.instructions.bundle.rootPath = instructionRoot;
|
||||
context.skills = await materializeSelectedSkills(config, config.paperclipRuntimeSkills, true);
|
||||
return { ...context, aggregateDigest: canonicalNativeRuntimeContextDigest(context) };
|
||||
});
|
||||
const provider = createLocalDiskStorageProvider(join(home, "objects"));
|
||||
const storage = createStorageService(durable ? { ...provider, id: "s3" } : provider);
|
||||
const storageSpy = vi.spyOn(storageModule, "getStorageService").mockReturnValue(storage);
|
||||
return { ...f, home, storage, cleanup: async () => { storageSpy.mockRestore(); await removeRuntimeAssetTree(home); } };
|
||||
}
|
||||
|
||||
it("publishes a dynamic connector skill receipt with admission and rehydrates its exact bytes on a cold host", async () => {
|
||||
const { input, run, agent, home, storage, cleanup } = await connectorFixture();
|
||||
let admitting = true;
|
||||
const getObject = storage.getObject.bind(storage);
|
||||
const readback = vi.spyOn(storage, "getObject").mockImplementation(async (...args) => {
|
||||
if (admitting) {
|
||||
expect((await read(run.id)).runnerProfileJson ?? {}).not.toHaveProperty("nativeExecutionInput");
|
||||
expect(await coordinators(run.id)).toEqual([]);
|
||||
}
|
||||
return getObject(...args);
|
||||
});
|
||||
try {
|
||||
const selected = await selectHeartbeatRuntime(db, input);
|
||||
admitting = false;
|
||||
if (!selected.selected || !selected.nativeExecution || !("runtimeContext" in selected.nativeExecution)) throw new Error("Expected native selection");
|
||||
const saved = await read(run.id);
|
||||
expect(readRuntimeSkillPreservation(saved)).not.toBeNull();
|
||||
expect(await hasLocalRuntimeSkillObligations(db)).toBe(false);
|
||||
const original = selected.nativeExecution.runtimeContext.skills[0]!;
|
||||
expect(original.versionId).toBeNull();
|
||||
const files = await Promise.all(["SKILL.md", "TOOLS.json"].map(async name => ({ name, bytes: await readFile(join(original.bundle.rootPath, name)) })));
|
||||
vi.stubEnv("PAPERCLIP_HOME", join(home, "cold"));
|
||||
const recovered = await rehydratePinnedSkillAssets(db, { companyId: agent.companyId, agentId: agent.id, runId: run.id }, selected.nativeExecution.runtimeContext);
|
||||
expect(recovered.skills[0]!.versionId).toBeNull();
|
||||
expect(recovered.skills[0]!.bundle.digest).toBe(original.bundle.digest);
|
||||
expect(recovered.skills[0]!.bundle.rootPath).not.toBe(original.bundle.rootPath);
|
||||
for (const file of files) {
|
||||
expect(await readFile(join(recovered.skills[0]!.bundle.rootPath, file.name))).toEqual(file.bytes);
|
||||
expect((await stat(join(recovered.skills[0]!.bundle.rootPath, file.name))).mode & 0o777).toBe(0o444);
|
||||
}
|
||||
readback.mockRestore();
|
||||
await removeRuntimeAssetTree(original.bundle.rootPath);
|
||||
const outage = vi.spyOn(storage, "getObject").mockRejectedValue(new Error("Synthetic storage outage"));
|
||||
try {
|
||||
const staged = await runtimeSkillPreservation.stageNativeRuntimeSkillAssets(db, { companyId: agent.companyId, agentId: agent.id,
|
||||
runId: run.id, issueId: selected.nativeExecution.binding.issueId, execution: selected.nativeExecution }, storage);
|
||||
expect(staged?.receipt).toEqual(readRuntimeSkillPreservation(saved));
|
||||
expect(outage).not.toHaveBeenCalled();
|
||||
} finally { outage.mockRestore(); }
|
||||
expect(await coordinators(run.id)).toMatchObject([{ phase: "observed", leaseOwner: null }]);
|
||||
} finally { readback.mockRestore(); await cleanup(); }
|
||||
});
|
||||
|
||||
it.each(["storage failure", "cancelled", "changed binding", "retired package"])("keeps dynamic skill admission unpublished after %s", async failure => {
|
||||
const { input, run, agent, storage, cleanup } = await connectorFixture();
|
||||
const stage = vi.spyOn(runtimeSkillPreservation, "stageNativeRuntimeSkillAssets");
|
||||
const putFile = storage.putFile.bind(storage);
|
||||
const put = ["storage failure", "cancelled"].includes(failure) ? vi.spyOn(storage, "putFile").mockImplementation(async data => {
|
||||
const result = await putFile(data);
|
||||
if (failure === "storage failure") throw new Error("Synthetic storage failure");
|
||||
await db.update(heartbeatRuns).set({ status: "cancelled" }).where(eq(heartbeatRuns.id, run.id));
|
||||
return result;
|
||||
}) : null;
|
||||
input.beforeNativeRuntimeSelection = async () => {
|
||||
if (failure === "changed binding") {
|
||||
const execution = structuredClone(stage.mock.calls[0]![1].execution);
|
||||
execution.binding.agentId = randomUUID();
|
||||
await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeExecutionInput: execution } }).where(eq(heartbeatRuns.id, run.id));
|
||||
}
|
||||
if (failure === "retired package") await db.update(companySkillRepositorySnapshots).set({ storage: null })
|
||||
.where(eq(companySkillRepositorySnapshots.companyId, agent.companyId));
|
||||
};
|
||||
try {
|
||||
if (failure === "cancelled") expect(await selectHeartbeatRuntime(db, input)).toEqual({ selected: false });
|
||||
else await expect(selectHeartbeatRuntime(db, input)).rejects.toThrow(failure === "storage failure" ? "Synthetic storage failure"
|
||||
: failure === "changed binding" ? "binding_changed" : "package_not_durable");
|
||||
const saved = await read(run.id);
|
||||
expect(saved.runnerProfileJson ?? {}).not.toHaveProperty("nativeRuntimeSkillPreservation");
|
||||
if (failure !== "changed binding") expect(saved.runnerProfileJson ?? {}).not.toHaveProperty("nativeExecutionInput");
|
||||
else expect((saved.runnerProfileJson!.nativeExecutionInput as { binding: { agentId: string } }).binding.agentId).not.toBe(agent.id);
|
||||
expect(await coordinators(run.id)).toEqual([]);
|
||||
expect(prepareNativeWorkspaceSync).not.toHaveBeenCalled();
|
||||
} finally { put?.mockRestore(); stage.mockRestore(); await cleanup(); }
|
||||
});
|
||||
|
||||
it("keeps volume-backed local connector admission independent of object storage", async () => {
|
||||
const { input, run, storage, cleanup } = await connectorFixture(false);
|
||||
const put = vi.spyOn(storage, "putFile").mockRejectedValue(new Error("Local mode must not upload"));
|
||||
try {
|
||||
expect((await selectHeartbeatRuntime(db, input)).selected).toBe(true);
|
||||
expect((await read(run.id)).runnerProfileJson).not.toHaveProperty("nativeRuntimeSkillPreservation");
|
||||
expect(put).not.toHaveBeenCalled();
|
||||
expect(await hasLocalRuntimeSkillObligations(db)).toBe(true);
|
||||
} finally { put.mockRestore(); await cleanup(); }
|
||||
});
|
||||
|
||||
it("preserves preparation failure and skips runtime selection", async () => {
|
||||
const { input, run } = await fixture();
|
||||
const initiating = new Error("Synthetic preparation failure");
|
||||
|
||||
@@ -84,6 +84,7 @@ import type { issueService } from "../issues.js";
|
||||
import { buildPlanReviewContext } from "../plan-review-context.js";
|
||||
import { normalizePaperclipWakePayload, selectPaperclipTaskMarkdown } from "@paperclipai/adapter-utils/server-utils";
|
||||
import type { ProviderResourceDisposition } from "../environment-runtime.js";
|
||||
import { stageNativeRuntimeSkillAssets, bindStagedNativeRuntimeSkillAssets } from "../native-runtime/runtime-skill-preservation.js";
|
||||
|
||||
type Run = typeof heartbeatRuns.$inferSelect;
|
||||
type Agent = typeof agents.$inferSelect;
|
||||
@@ -929,6 +930,9 @@ export async function selectHeartbeatRuntime(db: Db, input: HeartbeatRuntimeSele
|
||||
? "destroy"
|
||||
: undefined;
|
||||
onProviderResourceDisposition(providerResourceDispositionForRun);
|
||||
const stagedRuntimeSkills = await measureRunPhase("runtime_selection.skill_assets", () => stageNativeRuntimeSkillAssets(db, {
|
||||
companyId: agent.companyId, agentId: agent.id, runId: run.id, issueId: issueRef.id, execution: nativeExecution!,
|
||||
}));
|
||||
await beforeNativeRuntimeSelection?.(run.id);
|
||||
const nativeSelected = await measureRunTransaction(db, "runtime_selection", async (tx) => {
|
||||
const lockedRun = await measureRunPhase("runtime_selection.row_lock", () => tx
|
||||
@@ -958,6 +962,10 @@ export async function selectHeartbeatRuntime(db: Db, input: HeartbeatRuntimeSele
|
||||
throw new Error("native_runtime_mode_conflict");
|
||||
}
|
||||
const lockedProfile = parseObject(lockedRun.runnerProfileJson);
|
||||
const runtimeSkillPreservation = await bindStagedNativeRuntimeSkillAssets(tx as unknown as Db, {
|
||||
companyId: lockedRun.companyId, agentId: lockedRun.agentId, runId: lockedRun.id, issueId: issueRef.id,
|
||||
execution: (lockedProfile.nativeExecutionInput ?? nativeExecution) as NativeExecutionInput, staged: stagedRuntimeSkills,
|
||||
});
|
||||
const persistedNativeSessionId =
|
||||
await measureRunPhase("runtime_selection.bootstrap", () => prepareNativeSessionBootstrapPersistence(tx, {
|
||||
run: lockedRun,
|
||||
@@ -982,6 +990,7 @@ export async function selectHeartbeatRuntime(db: Db, input: HeartbeatRuntimeSele
|
||||
runnerProfileJson: {
|
||||
...nativeRuntimeResolution.profile,
|
||||
...lockedProfile,
|
||||
...(runtimeSkillPreservation ? { nativeRuntimeSkillPreservation: runtimeSkillPreservation } : {}),
|
||||
...(nativeExecution?.provider.kind === "cursor_cloud" ? { executionPlacement: resolveProviderApiPlacement({ agent, run }) } : {}),
|
||||
nativePreparationFencingVersion: !lockedRun.runnerInstanceId && !lockedRun.nativeSessionId
|
||||
? NATIVE_PREPARATION_FENCING_VERSION
|
||||
|
||||
@@ -6,6 +6,7 @@ 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 { getStorageService } from "../../storage/index.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";
|
||||
@@ -14,8 +15,8 @@ import { logActivity, publishActivity, type ActivityPublication } from "../activ
|
||||
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";
|
||||
type PreservedRuntimeSkill, type RuntimeSkillPreservation } from "./runtime-skill-preservation-reference.js";
|
||||
import type { NativeExecutionInput, 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(),
|
||||
@@ -73,6 +74,67 @@ async function readSource(asset: NativeRuntimeAssetReference) {
|
||||
return { manifestText, files };
|
||||
}
|
||||
|
||||
async function captureSkills(db: Db, companyId: string, admitted: ReturnType<typeof admittedRuntimeSkills>, storage?: StorageService, signal?: AbortSignal) {
|
||||
const store = skillPackageStore(db, storage), assets: PreservedRuntimeSkill[] = [];
|
||||
for (const skill of admitted.skills) {
|
||||
signal?.throwIfAborted();
|
||||
const captured = await withRuntimeAssetUse(skill.bundle.digest, () => readSource(skill.bundle), true);
|
||||
const packageSnapshotId = await store.prepare(companyId, captured.files, true, signal);
|
||||
if (!await store.verifyDurable(companyId, packageSnapshotId, 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) });
|
||||
}
|
||||
return assets;
|
||||
}
|
||||
|
||||
async function lockPackages(db: Db, companyId: string, assets: PreservedRuntimeSkill[]) {
|
||||
for (const asset of [...assets].sort((a, b) => a.packageSnapshotId.localeCompare(b.packageSnapshotId))) {
|
||||
const [row] = await db.select().from(snapshots).where(and(eq(snapshots.companyId, 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");
|
||||
}
|
||||
}
|
||||
|
||||
type Admission = { companyId: string; runId: string; agentId: string; issueId: string; execution: NativeExecutionInput };
|
||||
export type StagedRuntimeSkillAssets = { inputDigest: string; receipt: RuntimeSkillPreservation };
|
||||
const admissionSource = (input: Admission) => ({ id: input.runId, companyId: input.companyId, agentId: input.agentId,
|
||||
nativeIssueId: input.issueId, runnerProfileJson: { nativeExecutionInput: input.execution } });
|
||||
|
||||
export async function stageNativeRuntimeSkillAssets(db: Db, input: Admission & { signal?: AbortSignal }, storage?: StorageService): Promise<StagedRuntimeSkillAssets | null> {
|
||||
if (!("runtimeContext" in input.execution)) return null;
|
||||
const source = admissionSource(input), admitted = admittedRuntimeSkills(source);
|
||||
if (!admitted.skills.length) return null;
|
||||
const shared = storage ?? getStorageService();
|
||||
if (shared.provider !== "s3") return null;
|
||||
const [current] = await db.select({ receipt: sql<unknown>`${heartbeatRuns.runnerProfileJson}->${RUNTIME_SKILL_PRESERVATION_KEY}` }).from(heartbeatRuns).where(and(
|
||||
eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.agentId, input.agentId), eq(heartbeatRuns.id, input.runId)));
|
||||
if (!current) throw new Error("runtime_skill_source_unavailable");
|
||||
const previous = current.receipt;
|
||||
let receipt = previous == null ? null : readRuntimeSkillPreservation({ ...source,
|
||||
runnerProfileJson: { ...source.runnerProfileJson, [RUNTIME_SKILL_PRESERVATION_KEY]: previous } });
|
||||
if (previous != null && !receipt) throw new Error("runtime_skill_preservation_receipt_invalid");
|
||||
if (receipt) {
|
||||
const store = skillPackageStore(db, shared);
|
||||
for (const asset of receipt.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) throw new Error("runtime_skill_package_not_durable");
|
||||
}
|
||||
} else receipt = { schema: RUNTIME_SKILL_PRESERVATION_SCHEMA, inputDigest: admitted.inputDigest,
|
||||
bindingDigest: admitted.bindingDigest, assets: await captureSkills(db, input.companyId, admitted, shared, input.signal) };
|
||||
return { inputDigest: admitted.inputDigest, receipt };
|
||||
}
|
||||
|
||||
export async function bindStagedNativeRuntimeSkillAssets(db: Db, input: Admission & { staged: StagedRuntimeSkillAssets | null }) {
|
||||
if (!input.staged) return null;
|
||||
const source = admissionSource(input), admitted = admittedRuntimeSkills(source);
|
||||
const receipt = readRuntimeSkillPreservation({ ...source, runnerProfileJson: {
|
||||
...source.runnerProfileJson, [RUNTIME_SKILL_PRESERVATION_KEY]: input.staged.receipt } });
|
||||
if (!receipt || input.staged.inputDigest !== admitted.inputDigest) throw new Error("runtime_skill_preservation_binding_changed");
|
||||
await lockPackages(db, input.companyId, receipt.assets);
|
||||
return receipt;
|
||||
}
|
||||
|
||||
/** 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) {
|
||||
@@ -92,28 +154,14 @@ export async function preserveRuntimeSkillAssets(db: Db, input: { companyId: str
|
||||
}
|
||||
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) });
|
||||
}
|
||||
const assets = await captureSkills(db, input.companyId, admitted, storage, input.signal);
|
||||
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 lockPackages(tx as unknown as Db, input.companyId, assets);
|
||||
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);
|
||||
|
||||
Reference in new issue
Block a user