From f30005440b8f3f62c0bedb7b8abec9a8879211c5 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Thu, 12 Nov 2026 03:58:01 -0600 Subject: [PATCH] fix(recovery): reconnect a live remote Runner with its admitted skills instead of re-downloading them (#21064) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - Agent turns run in a remote Runner sandbox. A server upgrade detaches each running turn, and the new server must reconnect to its Runner (restart recovery). > - A reconnect runs the full turn preparation before it contacts the Runner. On a new server, skill preparation downloads every saved skill version again, under one lock for the whole company. > - After an upgrade, many reconnects wait behind each other's downloads. The Runner stops answering after about 2 minutes, so a late reconnect loses the turn. > - This pull request lets a reconnect to the same run's live remote Runner use the exact saved skill versions, without downloading them and without the lock. > - The benefit is that skill preparation no longer adds to the reconnect delay, and it does not grow with the number of reconnects. Normal turn starts do not change. ## Linked Issues or Issue Description **What happened?** After an automatic staging upgrade, runs that the new server reached late failed. The measured time from detach to reconnect was 114–152 s for the failed runs. Turn preparation took 30–130 s, and skill preparation alone took 46 s on one run. **Expected behavior** A reconnect to a live Runner reaches it quickly. Its speed must not depend on how many other runs prepare skills at the same time. **Steps to reproduce** 1. Run a remote Runner turn for each of many tasks in one company. 2. Upgrade the server while those turns run, so the new server starts with an empty skill cache. 3. See that each reconnect waits in `heartbeat.prepare.configuration_skills` for the company skill cache lock and downloads the skills again. **Paperclip version or commit** master at 4228bcf89c **Additional context** Root cause: `withRuntimeSkillEntries` holds the company-wide skill cache lock (a Postgres advisory lock) while it materializes each saved skill version. A new server has an empty skill cache, so each reconnect downloads its skills again, one run at a time. The Runner already has these files, and the reconnect path does not read them. ## What Changed - `company-skills.ts`: new `admittedVersionsOnly` option. It still checks that each saved version exists and that the agent can read the skill. It does not materialize the files or take the company skill cache lock. - `heartbeat/live-runner-adoption.ts`: `adoptsLiveRemoteRunner` is true only for a `reattach_remote_runner` claim for this run, with the claim's own lease owner, on a native run. - `heartbeat/execution-configuration.ts`: uses the option only when all of these are true: the reconnect gate is true, the run has its saved native input with a runtime context, and the run has its saved skill versions. In that case it times the phase as `configuration_skills_adopted` and adds one lifecycle run event (`skillPreparation: "reused_admitted"`). - `heartbeat.ts`: passes the gate to execution configuration. - `native-runtime/runtime-context.ts` and `native-session-executor.ts`: on the same reconnect to a remote Runner filesystem, the Runner controller describes the Runner's skills by their admitted digests. It rehydrates the pinned skill files (which takes the same lock) only if it must launch a Runner process and stage them. All other starts rehydrate as before. - `doc/execution-semantics.md`: describes the new behavior. Not in this PR (by design): a separate "dispatch-only" reconnect path, and a new startup scheduling lane. The native executor needs about 40 values from preparation (runner environment, MCP token, managed AI credentials, the reattached execution target, run output). Connector and instruction preparation also feed the session fingerprint. That cut is not mechanical, so this PR removes the largest measured single cost only. Other `heartbeat.prepare.*` costs that stay on the reconnect path (use post-upgrade timings to choose the next cut): - `configuration_secrets`, `configuration_connector_assignments`, `configuration_connection_instructions`, `configuration_connector_skills`/`_delivery`, `configuration_agent_identity`, `configuration_ai_runtime` - `environment_selection`, `workspace` (retained lease resume and realization), `session`, `task_context`, `run_identity` - `github_environment`, `github_launchers`, `run_output`, `runtime_services`, `instructions`, `runtime_selection` (including `runtime_selection.skill_assets` and `runtime_selection.workspace_sync`) - The wait from detach to "run started" (queue and controller-preparing waits) is outside `heartbeat.prepare.*`. ## Verification - New test `reconnects to a live remote Runner with its admitted skills without re-downloading them or waiting on the cache` (`server/src/__tests__/heartbeat-execution-configuration.test.ts`): - It admits a run, deletes its skill cache (like a new server), and holds the company cache lock. - The reconnect preparation finishes while the lock is held. It returns the same key, runtime name and version. The session fingerprint is identical. The cache is not created again. - Normal preparation of the same saved selection still waits for the lock and materializes the files ("normal start unchanged"). - Red on master source: `A live Runner reattach must not queue behind skill downloads: expected null not to be null`. Green with this change. - New test in `managed-skill-content.test.ts`: with the cache lock held and an empty cache, `prepareRuntimeContextSource` for a reconnect returns at once with the admitted context. Its staging step and a normal start still wait for the lock and then rehydrate the same digests. - New `server/src/services/heartbeat/live-runner-adoption.test.ts`: the gate is false for a normal start, a scheduled resume, a dead-Runner resume, an incomplete bootstrap, another run's claim, a different lease owner, and a legacy run. - `npx vitest run src/__tests__/heartbeat-execution-configuration.test.ts src/__tests__/skill-version-cache.test.ts src/services/native-runtime/native-runner-restart-recovery.integration.test.ts src/services/heartbeat/live-runner-adoption.test.ts`: 289 passed, 14 skipped. After the controller change: restart-recovery, diskless restart, runtime skill preservation, managed skill content, session executor and execution configuration suites: 1686 passed, 14 skipped. - `tsc --noEmit -p server/tsconfig.json`: pass. - `pnpm verify` asks for broader coverage (exit 2). Required CI supplies it. After deploy, a reconnect run shows `heartbeat.prepare.configuration_skills_adopted` and the `reused_admitted` event, and no `skills.cache.acquire` or `skills.version.materialize` phases. ## Risks - Low. The option applies only to a same-run reconnect of a retained remote Runner that has a saved runtime context and saved skill versions. Every other start keeps the current path. - Access revocation and deleted versions still fail the reconnect, as before. A failure to download the files on this server no longer fails a live reconnect, because the Runner already holds them. - Nothing new is saved, so rows from older releases, two live servers, and a deploy during a reconnect behave as before. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, used through Claude Code with tool use (code search, editing, shell, tests). ## 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 - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 --- doc/execution-semantics.md | 23 +++++++ .../heartbeat-execution-configuration.test.ts | 67 +++++++++++++++++-- .../__tests__/managed-skill-content.test.ts | 60 ++++++++++++++++- server/src/services/company-skills.ts | 20 +++++- .../services/execution-preparation-timing.ts | 2 +- server/src/services/heartbeat.ts | 4 +- .../heartbeat/execution-configuration.ts | 16 ++++- .../heartbeat/live-runner-adoption.test.ts | 26 +++++++ .../heartbeat/live-runner-adoption.ts | 20 ++++++ .../native-runtime/native-session-executor.ts | 14 ++-- .../native-runtime/runtime-context.ts | 30 +++++++++ 11 files changed, 265 insertions(+), 17 deletions(-) create mode 100644 server/src/services/heartbeat/live-runner-adoption.test.ts create mode 100644 server/src/services/heartbeat/live-runner-adoption.ts 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,