diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 596b3a21c51..4b44ba74a01 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -3337,6 +3337,29 @@ lease retains its recorded working directory, since optional provider metadata can describe the allocation root. A resume setup failure preserves the retained sandbox and saved work. Plugin task reattachment remains with its task owner. +When restart recovery reconnects a run to its own retained remote Runner +(`reattach_remote_runner`, matching run and lease owner), and the run has its +admitted native input with a runtime context and its saved skill versions, +preparation reuses those exact skill versions instead of downloading them again. +The Runner already holds their files. The server still checks that each saved +version exists and that the agent can still read the skill. It does not +materialize the files or take the company skill cache lock, so a burst of +reattachments after an upgrade does not queue behind other runs' skill downloads +on a server whose cache is empty. The run log records the +`heartbeat.prepare.configuration_skills_adopted` phase and a lifecycle event with +payload `skillPreparation: "reused_admitted"`. Every other start, including a +dead-Runner resume and a run saved by an older release without saved skill +versions or a runtime context, prepares its skills as before. The Runner +controller follows the same rule for the same reconnect: it describes the +Runner's skills by their admitted digests and rehydrates the pinned skill files +on this server only if it must launch a Runner process and stage them again. +The digest checks then run as before. It defers only when every admitted skill +is pinned to its exact version in the run's saved selection, which preparation +has just checked for access. An older run without that selection rehydrates +first, so revoked access still stops the reconnect. The session +configuration fingerprint is unchanged, because it uses each skill's key, +runtime name, and version, not its local path. + After a remote sandbox session passes provider recovery and durable run admission, its local suspension receipt is removed. This applies to built-in sandbox runners and plugin tasks. The running session then owns authority; diff --git a/server/src/__tests__/heartbeat-execution-configuration.test.ts b/server/src/__tests__/heartbeat-execution-configuration.test.ts index f8a2ac0ac56..c907917e3ae 100644 --- a/server/src/__tests__/heartbeat-execution-configuration.test.ts +++ b/server/src/__tests__/heartbeat-execution-configuration.test.ts @@ -9,7 +9,7 @@ import { eq } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; import { agents, authUsers, companies, companyMemberships, companySkills, companySkillTestRuns, createDb, executionWorkspaces, heartbeatRunEvents, heartbeatRuns, issues, projects, plugins, environments } from "@paperclipai/db"; import { companySkillService } from "../services/company-skills.js"; -import { withSkillVersionCacheLock } from "../services/skill-version-cache.js"; +import { skillVersionCachePaths, withSkillVersionCacheLock } from "../services/skill-version-cache.js"; import { registerServerAdapter, unregisterServerAdapter, type AdapterExecutionContext } from "../adapters/index.js"; import { HttpError } from "../errors.js"; import { heartbeatService } from "../services/heartbeat.js"; @@ -333,7 +333,7 @@ describePostgres("heartbeat execution configuration through adapter dispatch", ( } }); - async function configuration(f: Awaited>, config: Record, snapshot: Record = {}, type = "paperclip_runner", explicitTaskDirectory = false, requestedExecutionWorkspaceMode: HeartbeatExecutionConfigurationInput["policy"]["requestedExecutionWorkspaceMode"] = "shared_workspace", selectedEnvironmentForConfig: HeartbeatExecutionConfigurationInput["environment"]["selectedEnvironmentForConfig"] = null, pinnedSkillTest?: HeartbeatExecutionConfigurationInput["task"]["pinnedSkillTest"]) { + async function configuration(f: Awaited>, config: Record, snapshot: Record = {}, type = "paperclip_runner", explicitTaskDirectory = false, requestedExecutionWorkspaceMode: HeartbeatExecutionConfigurationInput["policy"]["requestedExecutionWorkspaceMode"] = "shared_workspace", selectedEnvironmentForConfig: HeartbeatExecutionConfigurationInput["environment"]["selectedEnvironmentForConfig"] = null, pinnedSkillTest?: HeartbeatExecutionConfigurationInput["task"]["pinnedSkillTest"], adoptingLiveRunner = false) { const [agent] = await db.update(agents).set({ adapterType: type, adapterConfig: config }).where(eq(agents.id, f.agentId)).returning(); const [run] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, issueId: f.issueId, scopeKind: "issue", status: "running", contextSnapshot: snapshot }).returning(); const resolvedInstanceSettings = await instanceSettingsService(db).get(); @@ -341,13 +341,16 @@ describePostgres("heartbeat execution configuration through adapter dispatch", ( const environment = selectedEnvironmentForConfig ?? await environmentService(db).ensureLocalEnvironment(); return { run, appendRunEvent, prepare: () => prepareHeartbeatExecutionConfiguration(db, { run, agent, - task: { issueId: f.issueId, context: { ...snapshot, issueId: f.issueId }, responsibleUserId: null, pinnedSkillTest, + task: { issueId: f.issueId, context: { ...snapshot, issueId: f.issueId }, responsibleUserId: null, pinnedSkillTest, adoptingLiveRunner, acceptedPlanContinuationWake: false, acceptedPlanWakeRoutingDecision: null, taskProjectContext: null, routineEnvContext: { routineId: null, env: null, responsibleUserId: null } }, config: { sourceWorkspaceConfig: config, aiBinding: undefined }, policy: { trustPreset: { kind: "standard", preset: "standard", boundary: null, sourcePresets: {} }, useIsolatedTaskDirectory: false, explicitTaskDirectory, requestedExecutionWorkspaceMode }, environment: { selectedEnvironmentId: environment.id, selectedEnvironmentForConfig: environment, - resolvedInstanceSettings, persistedNativeExecutionInput: null }, + resolvedInstanceSettings, persistedNativeExecutionInput: adoptingLiveRunner + // Preparation reads only whether the admitted input carries its runtime context. + ? { schema: "paperclip.native-execution-input.v7", runtimeContext: {} } as unknown as HeartbeatExecutionConfigurationInput["environment"]["persistedNativeExecutionInput"] + : null }, services: { secretsSvc: secretService(db), companySkills: companySkillService(db), appendRunEvent, finalizeAiConnectionBusyDeferral: vi.fn() }, onFailureSecretsResolved: () => {}, onManagedAiRuntimeAcquired: () => {}, onIdentityRedactorPrepared: () => {}, }) }; @@ -673,6 +676,62 @@ describePostgres("heartbeat execution configuration through adapter dispatch", ( if (result.prepared) expect(result.runtimeSkillEntries.some(entry => entry.key === (kind === "selected native" ? selected.key : "paperclipai/paperclip/paperclip"))).toBe(true); }); + it("reconnects to a live remote Runner with its admitted skills without re-downloading them or waiting on the cache", async () => { + const f = await seed(), skills = companySkillService(db); + const selected = await skills.createLocalSkill(f.companyId, { name: "Admitted skill", markdown: "# Admitted\nOriginal bytes.\n" }); + const config = { provider: "codex", model: "gpt-6.1-sol", homeMode: "isolated", paperclipSkillSync: { desiredSkills: [selected.key] } }; + const admitted = await configuration(f, config); + const first = await admitted.prepare(); + if (!first.prepared) throw new Error("Admission did not finish"); + const [saved] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, admitted.run.id)); + const versionId = (saved.contextSnapshot!.skillVersions as Record)[selected.key]!; + expect(versionId).toBeTruthy(); + // A newly deployed server starts with an empty skill cache. + const cached = skillVersionCachePaths(f.companyId, selected.id, versionId).directory; + await rm(cached, { recursive: true, force: true }); + const fingerprint = (entries: typeof first.runtimeSkillEntries) => buildEffectiveRunSessionConfigMetadata({ + adapterType: "paperclip_runner", effectiveAdapterConfig: {}, agentRuntimeConfig: {}, issueOverrides: null, + workspaceConfig: {}, environment: null, environmentEnv: null, projectEnv: null, routineEnv: null, runtimeSkills: entries, + }).then(metadata => metadata.fingerprint); + + let release!: () => void; + let acquired!: () => void; + const ready = new Promise(resolve => { acquired = resolve; }); + const barrier = new Promise(resolve => { release = resolve; }); + const holder = withSkillVersionCacheLock(db, f.companyId, async () => { acquired(); await barrier; }); + await ready; + const adopting = await configuration(f, config, saved.contextSnapshot!, "paperclip_runner", false, "shared_workspace", null, undefined, true); + const normal = await configuration(f, config, saved.contextSnapshot!, "paperclip_runner"); + let normalSettled = false; + const normalPending = normal.prepare().finally(() => { normalSettled = true; }); + const adoptingPending = adopting.prepare(); + let timer: ReturnType | undefined; + try { + const result = await Promise.race([adoptingPending, new Promise(resolve => { timer = setTimeout(() => resolve(null), 5000); })]); + expect(result, "A live Runner reattach must not queue behind skill downloads").not.toBeNull(); + if (!result?.prepared) throw new Error("Reattach preparation did not finish"); + expect(result.runtimeSkillEntries).toEqual([expect.objectContaining({ key: selected.key, versionId, sourceStatus: "available", + runtimeName: first.runtimeSkillEntries[0]!.runtimeName })]); + expect(await fingerprint(result.runtimeSkillEntries)).toBe(await fingerprint(first.runtimeSkillEntries)); + expect(await access(cached).then(() => true, () => false)).toBe(false); + expect(adopting.appendRunEvent).toHaveBeenCalledWith(expect.anything(), expect.objectContaining({ + payload: { skillPreparation: "reused_admitted", skills: 1 } })); + // An ordinary start of the same saved selection still waits for the cache and materializes. + expect(normalSettled).toBe(false); + } finally { + if (timer) clearTimeout(timer); + release(); + await holder; + await adoptingPending.catch(() => undefined); + } + const restored = await normalPending; + if (!restored.prepared) throw new Error("Normal preparation did not finish"); + expect(await readFile(path.join(restored.runtimeSkillEntries[0]!.source, "SKILL.md"), "utf8")).toContain("Original bytes."); + expect(await fingerprint(restored.runtimeSkillEntries)).toBe(await fingerprint(first.runtimeSkillEntries)); + expect(normal.appendRunEvent).not.toHaveBeenCalledWith(expect.anything(), expect.objectContaining({ + payload: expect.objectContaining({ skillPreparation: "reused_admitted" }) })); + }); + it("applies project environment and issue overrides and replaces caller-supplied runtime fields", async () => { const f = await seed(); const { finished, input } = await execute(f); diff --git a/server/src/__tests__/managed-skill-content.test.ts b/server/src/__tests__/managed-skill-content.test.ts index 946c929b497..1430bd763c3 100644 --- a/server/src/__tests__/managed-skill-content.test.ts +++ b/server/src/__tests__/managed-skill-content.test.ts @@ -6,7 +6,7 @@ import { skillPackageStore } from "../services/skill-package-store.js"; import { removeRuntimeAssetTree } from "../services/native-runtime/runtime-asset-retention.js"; import { eq } from "drizzle-orm"; import { writePaperclipSkillSyncPreference } from "@paperclipai/adapter-utils/server-utils"; -import { materializeSelectedSkills, rehydratePinnedSkillAssets } from "../services/native-runtime/runtime-context.js"; +import { materializeSelectedSkills, prepareRuntimeContextSource, rehydratePinnedSkillAssets } from "../services/native-runtime/runtime-context.js"; import { nativeRuntimeContextFixture } from "../services/native-runtime/runtime-context.test-fixture.js"; import { randomUUID } from "node:crypto"; import { execFile } from "node:child_process"; @@ -371,6 +371,64 @@ describe.skipIf(!support.supported)("managed skills use committed versions", () process.env.PAPERCLIP_HOME = originalHome; }); + it("reconnects to a live remote Runner without rehydrating its pinned skills until a launch stages them", async () => { + const companyId = await company(); + const service = companySkillService(db); + const skill = await service.createLocalSkill(companyId, { name: "Reconnect pinned", markdown: "# reconnect bytes" }); + const [agent] = await db.insert(agents).values({ companyId, name: "Runner" }).returning(); + const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId: agent!.id, status: "running", + contextSnapshot: { skillVersions: { [skill.key]: skill.currentVersionId }, skillSelectionSnapshot: true } }).returning(); + const entries = await service.withRuntimeSkillEntries(companyId, { onlyKeys: new Set([skill.key]), snapshotLatest: true, + versionSelections: new Map([[skill.key, skill.currentVersionId!]]) }, async entries => entries); + const selected = await materializeSelectedSkills(writePaperclipSkillSyncPreference({}, [{ key: skill.key, versionId: skill.currentVersionId }]), entries, false); + const context = { ...nativeRuntimeContextFixture(), skills: selected }; + const binding = { companyId, agentId: agent!.id, runId: run!.id }; + // A run saved before skill selections were recorded has no proof that preparation checked access. + const [older] = await db.insert(heartbeatRuns).values({ companyId, agentId: agent!.id, status: "running", contextSnapshot: {} }).returning(); + const originalHome = process.env.PAPERCLIP_HOME!; + // A newly deployed server: its skill cache is empty. + process.env.PAPERCLIP_HOME = path.join(root, `reconnect-${randomUUID()}`); + let release!: () => void; + let acquired!: () => void; + const ready = new Promise(resolve => { acquired = resolve; }); + const barrier = new Promise(resolve => { release = resolve; }); + const holder = withSkillVersionCacheLock(db, companyId, async () => { acquired(); await barrier; }); + await ready; + const settledWithin = async (work: Promise, ms: number) => { + let timer: ReturnType | undefined; + try { return await Promise.race([work.then(value => ({ value })), new Promise(resolve => { timer = setTimeout(() => resolve(null), ms); })]); } + finally { if (timer) clearTimeout(timer); } + }; + let staging: Promise | undefined; + let fresh: Promise | undefined; + let unpinned: Promise | undefined; + try { + const reconnect = await settledWithin(prepareRuntimeContextSource(db, binding, context, { deferUntilStaged: true }), 2_000); + expect(reconnect, "A live Runner reconnect must not wait on the skill cache").not.toBeNull(); + expect(reconnect!.value.described).toBe(context); + staging = reconnect!.value.forStaging(); + fresh = prepareRuntimeContextSource(db, binding, context, { deferUntilStaged: false }); + // Staging a relaunch, and every other start, still rehydrate under the cache lock. + expect(await settledWithin(staging, 200)).toBeNull(); + expect(await settledWithin(fresh, 200)).toBeNull(); + unpinned = prepareRuntimeContextSource(db, { ...binding, runId: older!.id }, context, { deferUntilStaged: true }); + expect(await settledWithin(unpinned, 200), "An older reconnect still rehydrates and checks access first").toBeNull(); + } finally { + release(); + await holder; + } + const restaged = await staging as Awaited>; + expect(restaged.skills[0]!.bundle.digest).toBe(selected[0]!.bundle.digest); + expect(await fs.readFile(path.join(restaged.skills[0]!.bundle.rootPath, "SKILL.md"), "utf8")).toBe("# reconnect bytes"); + const started = await fresh as Awaited>; + expect(started.described.skills[0]!.bundle.digest).toBe(selected[0]!.bundle.digest); + expect((await unpinned as Awaited>).described.skills[0]!.bundle.rootPath) + .not.toBe(selected[0]!.bundle.rootPath); + await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }).where(eq(heartbeatRuns.id, run!.id)); + await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }).where(eq(heartbeatRuns.id, older!.id)); + process.env.PAPERCLIP_HOME = originalHome; + }); + it("converts a legacy managed installation without changing its identity or identical version", async () => { const companyId = await company(); const directory = path.join(root, `legacy-${randomUUID()}`); diff --git a/server/src/services/company-skills.ts b/server/src/services/company-skills.ts index 0be3726541d..ea5253a103a 100644 --- a/server/src/services/company-skills.ts +++ b/server/src/services/company-skills.ts @@ -409,6 +409,13 @@ type RuntimeSkillEntryOptions = { includeLegacy?: boolean; onlyKeys?: Set; snapshotLatest?: boolean; + /** + * Resolve exact saved versions for a turn whose Runner already holds their + * files. Access and version existence are still checked; this server neither + * materializes the bytes nor takes the company skill cache lock, so `source` + * names the cache location only and must not be read. + */ + admittedVersionsOnly?: boolean; }; type SkillActor = { @@ -6006,6 +6013,7 @@ export function companySkillService(db: Db, options: { actor?: AuthorizationActo detail: "The selected skill version no longer exists.", }; } + if (options.admittedVersionsOnly) return { status: "available", source: versionPath }; // A failed snapshot materialization must surface as a "missing" entry // with the real cause — a silent drop makes the skill vanish from the // runtime while the library still shows it installed. @@ -6125,14 +6133,20 @@ export function companySkillService(db: Db, options: { actor?: AuthorizationActo if (!preparedSelection) { await measureRunPhase("skills.inventory", () => ensureSkillInventoryCurrent(companyId)); } - return withSkillPackageReadScope(db, () => withSkillVersionCacheLock(db, companyId, async () => { + const resolve = async (selected: RuntimeSkillEntryOptions) => { // Recheck scope on the full row read: a public-to-project change between // the probe and consumption must fail closed. - const runtimeOptions = preparedSelection ? { ...options, onlyKeys: requested } : options; + const runtimeOptions = preparedSelection ? { ...selected, onlyKeys: requested } : selected; const entries = await listRuntimeSkillEntriesLocked(companyId, runtimeOptions, Boolean(preparedSelection && installed.every(skill => skill.accessSourceId === null))); return measureRunPhase("skills.consume", () => consume(entries)); - })); + }; + // Saved versions only: nothing is published to or pinned in the cache, so + // a reattaching turn does not queue behind other runs' materialization. + if (options.admittedVersionsOnly) { + return withSkillPackageReadScope(db, () => resolve({ ...options, snapshotLatest: false })); + } + return withSkillPackageReadScope(db, () => withSkillVersionCacheLock(db, companyId, () => resolve(options))); } async function listRuntimeSkillEntriesLocked( diff --git a/server/src/services/execution-preparation-timing.ts b/server/src/services/execution-preparation-timing.ts index 3cc34e3f532..b8291d08dda 100644 --- a/server/src/services/execution-preparation-timing.ts +++ b/server/src/services/execution-preparation-timing.ts @@ -13,7 +13,7 @@ type PreparationPhase = "environment_selection" | "execution_configuration" | "s | "workspace" | "github_environment" | "github_launchers" | "run_output" | "workspace_validation" | "runtime_services" | "instructions" | "runtime_selection" | "configuration_mentions" | "configuration_github_identity" | "configuration_secrets" - | "configuration_ai_runtime" | "configuration_skills" | "configuration_connector_assignments" + | "configuration_ai_runtime" | "configuration_skills" | "configuration_skills_adopted" | "configuration_connector_assignments" | "configuration_connector_skills" | "configuration_connector_delivery" | "configuration_connection_instructions" | "configuration_agent_identity" | "workspace_source_resolution" | "workspace_verification" diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 070b371be96..9ce4709d5c4 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -49,6 +49,7 @@ import { AgentDirectoryPreparationDeferral, deferRunForBusyAgentDirectory, prepa import { preparePredecessorRetirementFallback } from "./heartbeat/predecessor-retirement-fallback.js"; import { prepareHeartbeatRuntimeServiceIntents, startHeartbeatRuntimeServices, recordHeartbeatAdapterRuntimeServices } from "./heartbeat/runtime-services.js"; import { bindHeartbeatRunnerHomeMode, prepareHeartbeatExecutionConfiguration } from "./heartbeat/execution-configuration.js"; +import { adoptsLiveRemoteRunner } from "./heartbeat/live-runner-adoption.js"; import { createRunLogWriterAuthority } from "./heartbeat/run-log-authority.js"; import { watchShutdownHandoff } from "./heartbeat/shutdown-handoff.js"; import { createHeartbeatRunOutput } from "./heartbeat/run-output.js"; @@ -2826,7 +2827,8 @@ export function heartbeatService( let nativeChatExpectedCwd = selectedNativeChatExpectedCwd; const executionConfiguration = await timePreparation("execution_configuration", () => prepareHeartbeatExecutionConfiguration(db, { run, agent, - task: { issueId, context, responsibleUserId, acceptedPlanContinuationWake, acceptedPlanWakeRoutingDecision, taskProjectContext, routineEnvContext, pinnedSkillTest }, + task: { issueId, context, responsibleUserId, acceptedPlanContinuationWake, acceptedPlanWakeRoutingDecision, taskProjectContext, routineEnvContext, pinnedSkillTest, + adoptingLiveRunner: adoptsLiveRemoteRunner({ run, restartRecovery: runOptions.nativeRestartRecovery, leaseOwner: runOptions.nativeLeaseOwner }) }, config: { sourceWorkspaceConfig, aiBinding }, policy: { trustPreset, useIsolatedTaskDirectory, explicitTaskDirectory, requestedExecutionWorkspaceMode }, environment: { selectedEnvironmentId, selectedEnvironmentForConfig, resolvedInstanceSettings, persistedNativeExecutionInput }, diff --git a/server/src/services/heartbeat/execution-configuration.ts b/server/src/services/heartbeat/execution-configuration.ts index 262a68fcdfd..f373c688133 100644 --- a/server/src/services/heartbeat/execution-configuration.ts +++ b/server/src/services/heartbeat/execution-configuration.ts @@ -70,6 +70,8 @@ export interface HeartbeatExecutionConfigurationInput { routineEnvContext: Awaited["getRoutineEnvForExecutionIssue"]>>; /** Resolved from the durable Studio test row by task-context, never wake input. */ pinnedSkillTest?: { key: string; versionId: string } | null; + /** Restart recovery is reconnecting to this run's live remote Runner. */ + adoptingLiveRunner?: boolean; }; config: { sourceWorkspaceConfig: Record; @@ -290,6 +292,10 @@ export async function prepareHeartbeatExecutionConfiguration(db: Db, input: Hear const skillsPrepareStartedAtMs = Date.now(); let skillsPrepareEndedAtMs = skillsPrepareStartedAtMs; let skillDeliveryConfig = effectiveResolvedConfig; + // A live remote Runner already holds the files of its admitted skill versions. + // Reuse that exact selection instead of re-downloading it on this server. + const reuseAdmittedSkills = input.task.adoptingLiveRunner === true && hasStoredVersions + && persistedNativeExecutionInput != null && "runtimeContext" in persistedNativeExecutionInput; const prepareSkills = async () => { try { @@ -302,7 +308,7 @@ export async function prepareHeartbeatExecutionConfiguration(db: Db, input: Hear // pins and the Beta skills preference. Recovery keeps the admitted map. if (!hasStoredVersions && pinnedSkillTest) versionSelections.set(pinnedSkillTest.key, pinnedSkillTest.versionId); return await withSkillActor(skillActor, () => companySkills.withRuntimeSkillEntries(agent.companyId, { - versionSelections, snapshotLatest: true, + versionSelections, ...(reuseAdmittedSkills ? { admittedVersionsOnly: true } : { snapshotLatest: true }), ...(hasStoredVersions ? { onlyKeys: new Set(Object.keys(storedVersions)) } : { selectionConfig: effectiveResolvedConfig, includeLegacy: agent.adapterType !== "paperclip_runner" }), @@ -361,7 +367,8 @@ export async function prepareHeartbeatExecutionConfiguration(db: Db, input: Hear throw error; } }; - const skillsPreparation = (preconfiguredHome ? Promise.resolve([]) : time("configuration_skills", prepareSkills)) + const skillsPreparation = (preconfiguredHome ? Promise.resolve([]) + : time(reuseAdmittedSkills ? "configuration_skills_adopted" : "configuration_skills", prepareSkills)) .finally(() => { skillsPrepareEndedAtMs = Date.now(); }); const assignmentsPreparation = preconfiguredHome ? Promise.resolve([]) : time("configuration_connector_assignments", () => resolveConnectorAssignments(db, { companyId: agent.companyId, agentId: agent.id, runId: run.id, issueId: typeof context.issueId === "string" ? context.issueId : undefined })); const [preparedSkills, preparedAssignments, preparedInstructions] = await Promise.allSettled([ @@ -373,6 +380,11 @@ export async function prepareHeartbeatExecutionConfiguration(db: Db, input: Hear if (preparedAssignments.status === "rejected") throw preparedAssignments.reason; if (preparedInstructions.status === "rejected") throw preparedInstructions.reason; const runtimeSkillEntries = preparedSkills.value; + if (reuseAdmittedSkills) { + await appendRunEvent(run, { eventType: "lifecycle", stream: "system", level: "info", + message: "Reconnecting to the running agent with the skills it already has", + payload: { skillPreparation: "reused_admitted", skills: runtimeSkillEntries.length } }); + } nativeRunnerPreparationSpans.push({ name: "skills.prepare", parentName: "task.prepare", diff --git a/server/src/services/heartbeat/live-runner-adoption.test.ts b/server/src/services/heartbeat/live-runner-adoption.test.ts new file mode 100644 index 00000000000..996a9efe105 --- /dev/null +++ b/server/src/services/heartbeat/live-runner-adoption.test.ts @@ -0,0 +1,26 @@ +import { describe, expect, it } from "vitest"; +import type { NativeRestartRecoveryClaim } from "../native-runtime/native-restart-recovery.js"; +import { adoptsLiveRemoteRunner } from "./live-runner-adoption.js"; + +const run = { id: "run-1", runtimeMode: "native" as const }; +const base = { runId: "run-1", leaseOwner: "owner-1", controllerGeneration: 2, providerAttempt: 1, + restartKind: "hard" as const, recoveryRequestId: null }; +const reattach = { ...base, kind: "reattach_remote_runner", remote: { providerLeaseId: "lease-1", remoteCwd: "/w" } } as NativeRestartRecoveryClaim; + +describe("adoptsLiveRemoteRunner", () => { + it("adopts only this run's retained remote Runner under the claim's own owner", () => { + expect(adoptsLiveRemoteRunner({ run, restartRecovery: reattach, leaseOwner: "owner-1" })).toBe(true); + }); + + it.each([ + ["a normal turn start", { run, restartRecovery: undefined, leaseOwner: undefined }], + ["a scheduled resume without a restart claim", { run, restartRecovery: undefined, leaseOwner: "owner-1" }], + ["a dead Runner resume", { run, restartRecovery: { ...base, kind: "resume_dead_runner" } as NativeRestartRecoveryClaim, leaseOwner: "owner-1" }], + ["an incomplete bootstrap", { run, restartRecovery: { ...base, kind: "bootstrap_incomplete" } as NativeRestartRecoveryClaim, leaseOwner: "owner-1" }], + ["another run's claim", { run: { ...run, id: "run-2" }, restartRecovery: reattach, leaseOwner: "owner-1" }], + ["a different lease owner", { run, restartRecovery: reattach, leaseOwner: "owner-2" }], + ["a legacy run", { run: { ...run, runtimeMode: "legacy" as const }, restartRecovery: reattach, leaseOwner: "owner-1" }], + ])("takes the full preparation for %s", (_name, input) => { + expect(adoptsLiveRemoteRunner(input as Parameters[0])).toBe(false); + }); +}); diff --git a/server/src/services/heartbeat/live-runner-adoption.ts b/server/src/services/heartbeat/live-runner-adoption.ts new file mode 100644 index 00000000000..da38e29f24a --- /dev/null +++ b/server/src/services/heartbeat/live-runner-adoption.ts @@ -0,0 +1,20 @@ +import type { heartbeatRuns } from "@paperclipai/db"; +import type { NativeRestartRecoveryClaim } from "../native-runtime/native-restart-recovery.js"; + +/** + * True when restart recovery is reconnecting this exact run to its retained + * remote Runner under the claim's own lease owner. Preparation may then reuse + * what the Runner was admitted with instead of rebuilding it on this server. + */ +export function adoptsLiveRemoteRunner(input: { + run: Pick; + restartRecovery?: NativeRestartRecoveryClaim; + leaseOwner?: string; +}): boolean { + const claim = input.restartRecovery; + return claim?.kind === "reattach_remote_runner" + && claim.runId === input.run.id + && input.leaseOwner !== undefined + && claim.leaseOwner === input.leaseOwner + && input.run.runtimeMode === "native"; +} diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 0cd6a81a2db..80e36428abb 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -25,7 +25,7 @@ import { assertNativeHarnessBackupOwner, durableNativeHarnessStorage, NATIVE_HAR import { restoreNativeProviderArtifacts, retainNativeProviderArtifacts, retainExistingNativeProviderArtifacts, type NativeProviderArtifactReceipt } from "./native-provider-artifacts.js"; import { prepareRemoteOpenCodeArtifacts, type OpenCodeLaunchArtifacts } from "./remote-opencode-artifacts.js"; -import { rehydratePinnedInstructionAsset, rehydratePinnedSkillAssets } from "./runtime-context.js"; +import { prepareRuntimeContextSource, rehydratePinnedInstructionAsset } from "./runtime-context.js"; import { createNativeUsageDelivery } from "./native-usage-delivery.js"; import { createNativeRecoveryUsageDelivery } from "./native-recovery-usage-delivery.js"; import { createNativeTurnAccounting, NATIVE_LOST_TURN_USAGE_EVIDENCE, nativeUsageCostUsd, nativeUsageBiller, normalizeNativeUsage, usesNativeUsageDelivery, isNativeUsageAccountingEvent, isNativeUsageDeliveryEvent } from "./native-usage-accounting.js"; @@ -16409,9 +16409,13 @@ async function createRunnerdBackendWithinSessionClaim( : remoteRunnerFilesystemRoot ? posix.join(remoteRunnerFilesystemRoot, directory.name) : null; - const sourceRuntimeContext = "runtimeContext" in input.execution - ? await rehydratePinnedSkillAssets(input.db, input.execution.binding, input.execution.runtimeContext) + // A reconnect to a live remote Runner stages nothing unless it relaunches, so + // it does not wait on this server's skill cache before reaching the Runner. + const runtimeContextSource = "runtimeContext" in input.execution + ? await prepareRuntimeContextSource(input.db, input.execution.binding, input.execution.runtimeContext, + { deferUntilStaged: input.restartRecovery?.kind === "reattach_remote_runner" && remoteRunnerFilesystemRoot !== null }) : null; + const sourceRuntimeContext = runtimeContextSource?.described ?? null; const remoteRuntimeContext: NativeRuntimeContextSnapshot | null = remoteRunnerFilesystemRoot && sourceRuntimeContext ? { @@ -17220,8 +17224,8 @@ async function createRunnerdBackendWithinSessionClaim( location: directory.location, targetPath: remotePersistencePath(directory), })), - runtimeContext: remoteRunnerFilesystemRoot && sourceRuntimeContext && remoteRuntimeContext - ? { source: sourceRuntimeContext, target: remoteRuntimeContext, filesystemRoot: remoteRunnerFilesystemRoot } + runtimeContext: remoteRunnerFilesystemRoot && runtimeContextSource && remoteRuntimeContext + ? { source: await runtimeContextSource.forStaging(), target: remoteRuntimeContext, filesystemRoot: remoteRunnerFilesystemRoot } : null, caBundleMapping: remoteCaBundleMapping, stageFile: asset => stageRemoteRunnerFile({ target: remoteTarget, runner: remoteCommandRunner, ...asset }), diff --git a/server/src/services/native-runtime/runtime-context.ts b/server/src/services/native-runtime/runtime-context.ts index e69381a677e..74b22dac3b2 100644 --- a/server/src/services/native-runtime/runtime-context.ts +++ b/server/src/services/native-runtime/runtime-context.ts @@ -290,6 +290,36 @@ export async function rehydratePinnedSkillAssets( return { ...context, skills }; } +/** + * The admitted context names every skill by digest, so the Runner's copy is + * described without local files. Pinned skill files are rehydrated on this + * server now, or, when reconnecting to a live remote Runner that already holds + * them, only if a launch must stage them again. + */ +export async function prepareRuntimeContextSource( + db: Db, + binding: { companyId: string; runId: string; agentId: string }, + context: NativeRuntimeContextSnapshot, + options: { deferUntilStaged: boolean }, +) { + // Defer only when preparation already checked access to every admitted skill: + // each one is pinned to its exact version in the run's saved selection. + const pinned = options.deferUntilStaged && context.skills.length > 0 + ? await db.select({ contextSnapshot: heartbeatRuns.contextSnapshot }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, binding.companyId), eq(heartbeatRuns.id, binding.runId), eq(heartbeatRuns.agentId, binding.agentId))) + .then(([run]) => { + const pins = run?.contextSnapshot?.skillVersions; + return Boolean(pins && typeof pins === "object" && !Array.isArray(pins) + && context.skills.every(skill => skill.versionId && (pins as Record)[skill.key] === skill.versionId)); + }) + : options.deferUntilStaged; + let staged = pinned ? null : await rehydratePinnedSkillAssets(db, binding, context); + return { + described: staged ?? context, + forStaging: async () => (staged ??= await rehydratePinnedSkillAssets(db, binding, context)), + }; +} + export async function materializeSelectedSkills(runtimeConfig: Record, entries: PaperclipSkillEntry[], omitLegacy: boolean) { const desiredKeys = resolvePaperclipDesiredSkillNames(runtimeConfig, entries).filter( (key) => !omitLegacy || key !== PAPERCLIP_OPERATIONAL_SKILL_KEY,