diff --git a/doc/observability.md b/doc/observability.md index b4bbd0d60c8..6bcf57b3130 100644 --- a/doc/observability.md +++ b/doc/observability.md @@ -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. | diff --git a/doc/skill-ephemeral-web-servers.md b/doc/skill-ephemeral-web-servers.md index f12c4c2bf5d..e5303e4d25b 100644 --- a/doc/skill-ephemeral-web-servers.md +++ b/doc/skill-ephemeral-web-servers.md @@ -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 diff --git a/server/src/services/execution-preparation-timing.ts b/server/src/services/execution-preparation-timing.ts index 4ba7aa69457..af5a9bebcf4 100644 --- a/server/src/services/execution-preparation-timing.ts +++ b/server/src/services/execution-preparation-timing.ts @@ -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" diff --git a/server/src/services/heartbeat/runtime-selection.test.ts b/server/src/services/heartbeat/runtime-selection.test.ts index 5c027ea3060..1da386ff3e9 100644 --- a/server/src/services/heartbeat/runtime-selection.test.ts +++ b/server/src/services/heartbeat/runtime-selection.test.ts @@ -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"); diff --git a/server/src/services/heartbeat/runtime-selection.ts b/server/src/services/heartbeat/runtime-selection.ts index a8f2239e1cd..a4025cbefe7 100644 --- a/server/src/services/heartbeat/runtime-selection.ts +++ b/server/src/services/heartbeat/runtime-selection.ts @@ -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 diff --git a/server/src/services/native-runtime/runtime-skill-preservation.ts b/server/src/services/native-runtime/runtime-skill-preservation.ts index bc2d9bcc605..04855854af2 100644 --- a/server/src/services/native-runtime/runtime-skill-preservation.ts +++ b/server/src/services/native-runtime/runtime-skill-preservation.ts @@ -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, 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 { + 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`${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);