diff --git a/doc/DEVELOPING.md b/doc/DEVELOPING.md index 727e3469f31..dfec311ad64 100644 --- a/doc/DEVELOPING.md +++ b/doc/DEVELOPING.md @@ -2196,6 +2196,14 @@ with an active or unsettled run whose pins are unknown is skipped. Optional (default `60`). Separate `__version_access__` records track use; existing caches without records receive a full observation period before expiration. +Finished interrupted runs release their pins only after controller ownership, +native finalization, and environment cleanup have settled. Older native runs +without a pin map can supply their complete selection from the saved execution +input. The recovery parser validates its digests, and its company, run, agent, +and issue binding must match. Captured version IDs remain protected; unversioned +skills use separate runtime asset bundles. Missing or invalid evidence still +blocks cleanup. This fallback does not write pins or change recovery records. + Preparation, durable run pins, audits, and eviction share a company-scoped PostgreSQL lock across processes. Eligible directories are atomically renamed to unique managed `.trash-` names. Deletion runs outside the preparation diff --git a/server/src/__tests__/skill-version-cache.test.ts b/server/src/__tests__/skill-version-cache.test.ts index 67e19206e32..725006cc4a5 100644 --- a/server/src/__tests__/skill-version-cache.test.ts +++ b/server/src/__tests__/skill-version-cache.test.ts @@ -4,12 +4,14 @@ import os from "node:os"; import path from "node:path"; import { eq } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; -import { agents, heartbeatRuns, issues, nativeRunFinalizations, companies, companySkills, companySkillVersions, companySkillRepositorySnapshots, createDb } from "@paperclipai/db"; +import { agents, heartbeatRuns, issues, nativeRunFinalizations, environmentLeases, companies, companySkills, companySkillVersions, companySkillRepositorySnapshots, createDb } from "@paperclipai/db"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; import { companySkillService } from "../services/company-skills.js"; import { skillVersionCachePaths, touchSkillVersionCache, withSkillVersionCacheLock } from "../services/skill-version-cache.js"; import { skillVersionRetentionService } from "../services/skill-version-retention.js"; +import { nativeRuntimeContextFixture } from "../services/native-runtime/runtime-context.test-fixture.js"; +import { canonicalNativeRuntimeContextDigest, parseNativeExecutionInput } from "../vendor/paperclip-runner/index.js"; async function sweepSkillVersionCache(db: ReturnType, options: { companyId: string; now?: number; dryRun?: boolean }) { const result = await skillVersionRetentionService(db).pruneCompany(options.companyId, { dryRun: options.dryRun, now: options.now === undefined ? undefined : new Date(options.now) }); return { ...result, evicted: result.removed }; @@ -136,6 +138,88 @@ describePostgres("immutable skill version cache", () => { expect((await sweepSkillVersionCache(db, { companyId: f.companyId })).evicted).toBe(1); }); + it("releases settled interrupted runs but preserves unfinished execution and cleanup", async () => { + const f = await fixture(); + await f.resolve(f.versions[0]!.id); + await touchSkillVersionCache(f.companyId, f.skillId, f.versions[0]!.id, Date.now() - 7_200_000); + const run = await activeRun(f.companyId, null); + await db.update(heartbeatRuns).set({ status: "interrupted" }).where(eq(heartbeatRuns.id, run.id)); + expect((await sweepSkillVersionCache(db, { companyId: f.companyId })).skipped.unknown_run_pins).toBe(1); + await db.update(heartbeatRuns).set({ finishedAt: new Date(), controllerLeaseExpiresAt: new Date(Date.now() + 60_000) }) + .where(eq(heartbeatRuns.id, run.id)); + expect((await sweepSkillVersionCache(db, { companyId: f.companyId })).skipped.unknown_run_pins).toBe(1); + await db.update(heartbeatRuns).set({ controllerLeaseExpiresAt: new Date(0) }).where(eq(heartbeatRuns.id, run.id)); + const [lease] = await db.insert(environmentLeases).values({ companyId: f.companyId, heartbeatRunId: run.id, + status: "pending_cleanup", cleanupStatus: "failed" }).returning(); + expect((await sweepSkillVersionCache(db, { companyId: f.companyId })).skipped.unknown_run_pins).toBe(1); + await db.update(environmentLeases).set({ status: "released", cleanupStatus: "success", releasedAt: new Date() }) + .where(eq(environmentLeases.id, lease!.id)); + expect((await sweepSkillVersionCache(db, { companyId: f.companyId })).evicted).toBe(1); + }); + + async function legacyNativeRun(companyId: string, skills: Array<{ key: string; versionId: string | null }> = []) { + const run = await activeRun(companyId, null); + const [issue] = await db.insert(issues).values({ companyId, title: "Saved native selection" }).returning(); + const runtimeContext = nativeRuntimeContextFixture(); + runtimeContext.skills = skills.map((skill, index) => ({ ...skill, runtimeName: `skill-${index}`, bundle: runtimeContext.instructions.bundle })); + runtimeContext.aggregateDigest = canonicalNativeRuntimeContextDigest(runtimeContext); + const execution = parseNativeExecutionInput({ + schema: "paperclip.native-execution-input.v3", + binding: { companyId, runId: run.id, agentId: run.agentId, issueId: issue!.id, executionWorkspaceId: run.id }, + task: { identifier: "CACHE-1", title: "Saved selection", description: null, prompt: "Read saved skills", workMode: "standard" }, + workspace: { cwd: "/tmp", repoUrl: null, repoRef: null, branchName: null }, + session: { normalizedSessionId: null, driverKind: "codex_app_server", protocolVersion: 1, lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null } }, + provider: { kind: "codex", model: null }, + completionContract: { id: "contract", sha256: "saved", schemaVersion: "paperclip.completion-contract.v1", + contract: { revision: "1", objective: "Read saved skills", criteria: [{ id: "read", requirement: "Read skills" }] } }, + interactionResponses: [], credentialBindings: [], executionMode: "default", planningContext: null, runtimeContext, + }); + await db.update(heartbeatRuns).set({ runtimeMode: "native", nativeIssueId: issue!.id, status: "failed", finishedAt: new Date(), + runnerProfileJson: { nativeExecutionInput: execution } }).where(eq(heartbeatRuns.id, run.id)); + await db.insert(nativeRunFinalizations).values({ companyId, issueId: issue!.id, runId: run.id, + phase: "terminal_failure", recoveryState: "blocked" }); + return { run, execution }; + } + + it.each([false, true])("uses the legacy native manifest without changing unresolved recovery (bundled skill: %s)", async bundled => { + const f = await fixture(); + const old = await f.resolve(f.versions[0]!.id); + await touchSkillVersionCache(f.companyId, f.skillId, f.versions[0]!.id, Date.now() - 7_200_000); + const { run } = await legacyNativeRun(f.companyId, bundled ? [{ key: "paperclipai/paperclip/slack", versionId: null }] : []); + expect((await sweepSkillVersionCache(db, { companyId: f.companyId })).evicted).toBe(1); + expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, run.id)))[0]) + .toMatchObject({ phase: "terminal_failure", recoveryState: "blocked" }); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)))[0]!.contextSnapshot).toBeNull(); + vi.stubGlobal("fetch", vi.fn(() => Promise.reject(new Error("Upstream unavailable")))); + expect(await f.resolve(f.versions[0]!.id)).toMatchObject({ source: old.source }); + }); + + it("preserves exact legacy manifest pins", async () => { + const f = await fixture(); + await f.resolve(f.versions[0]!.id); + await touchSkillVersionCache(f.companyId, f.skillId, f.versions[0]!.id, Date.now() - 7_200_000); + await legacyNativeRun(f.companyId, [{ key: f.key, versionId: f.versions[0]!.id }]); + expect((await sweepSkillVersionCache(db, { companyId: f.companyId })).skipped.protected_version).toBe(1); + }); + + it.each(["companyId", "runId", "agentId", "issueId", "digest", "missing", "invalid_version", "invalid_pins", "legacy_adapter"])( + "fails closed for unusable native selection evidence (%s)", async invalid => { + const f = await fixture(); + await f.resolve(f.versions[0]!.id); + await touchSkillVersionCache(f.companyId, f.skillId, f.versions[0]!.id, Date.now() - 7_200_000); + const { run, execution } = await legacyNativeRun(f.companyId, + invalid === "invalid_version" ? [{ key: f.key, versionId: "not-a-version" }] : []); + if (["companyId", "runId", "agentId", "issueId"].includes(invalid)) Object.assign(execution.binding, { [invalid]: randomUUID() }); + if (invalid === "digest" && "runtimeContext" in execution) execution.runtimeContext.aggregateDigest = "f".repeat(64); + await db.update(heartbeatRuns).set({ + runnerProfileJson: invalid === "missing" ? {} : { nativeExecutionInput: execution }, + ...(invalid === "invalid_pins" ? { contextSnapshot: { skillVersions: null } } : {}), + ...(invalid === "legacy_adapter" ? { runtimeMode: "legacy" } : {}), + }).where(eq(heartbeatRuns.id, run.id)); + expect((await sweepSkillVersionCache(db, { companyId: f.companyId })).skipped.unknown_run_pins).toBe(1); + }, + ); + it("serializes eviction behind preparation and durable pinning across database clients", async () => { const f = await fixture(); await f.resolve(f.versions[0]!.id); diff --git a/server/src/services/skill-version-cache.ts b/server/src/services/skill-version-cache.ts index 50ee27f8468..2f3ec15c5ad 100644 --- a/server/src/services/skill-version-cache.ts +++ b/server/src/services/skill-version-cache.ts @@ -9,6 +9,7 @@ import { resolvePaperclipInstanceRoot } from "../home-paths.js"; import { assertSkillSnapshotPath } from "./skill-snapshot.js"; import { repositorySkillFrontmatter } from "./skill-repository-package.js"; import type { CompanySkillVersionFileInventoryEntry } from "@paperclipai/shared"; +import { parseNativeExecutionInput } from "../vendor/paperclip-runner/index.js"; export const SKILL_VERSION_CACHE_IDLE_MS = 60 * 60 * 1000; const locks = new AsyncLocalStorage>(); @@ -91,9 +92,16 @@ export async function lastUsedAt(filename: string): Promise { // A terminal run can still own provider execution or recovery. Only fully settled // runs release their pins; resume preparation reacquires the same cache lock. export async function protectedVersions(db: Db, companyId: string): Promise | null> { - const rows = await db.select({ pins: sql`${heartbeatRuns.contextSnapshot}->'skillVersions'` }) + const rows = await db.select({ + id: heartbeatRuns.id, agentId: heartbeatRuns.agentId, nativeIssueId: heartbeatRuns.nativeIssueId, + pins: sql`${heartbeatRuns.contextSnapshot}->'skillVersions'`, + // Avoid loading large execution inputs for runs with the modern pin map. + legacyInput: sql`case when ${heartbeatRuns.runtimeMode} = 'native' + and not coalesce(${heartbeatRuns.contextSnapshot} ? 'skillVersions', false) + then ${heartbeatRuns.runnerProfileJson}->'nativeExecutionInput' end`, + }) .from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, companyId), sql`( - ${heartbeatRuns.status} not in ('succeeded', 'failed', 'timed_out', 'cancelled') + ${heartbeatRuns.status} not in ('succeeded', 'failed', 'timed_out', 'cancelled', 'interrupted') or ${heartbeatRuns.finishedAt} is null or ${heartbeatRuns.controllerLeaseExpiresAt} > now() or exists (select 1 from ${nativeRunFinalizations} f @@ -107,7 +115,27 @@ export async function protectedVersions(db: Db, companyId: string): Promise(); - for (const { pins } of rows) { + for (const { pins, legacyInput, id, agentId, nativeIssueId } of rows) { + if (legacyInput != null) { + // Older native runs captured the complete selection in their immutable + // execution input before contextSnapshot.skillVersions was introduced. + // Use the recovery parser (including digest validation) and the exact run + // binding. Never repair a malformed modern pin map with a different source. + try { + const input = parseNativeExecutionInput(legacyInput); + if (!("runtimeContext" in input) || input.binding.companyId !== companyId + || input.binding.runId !== id || input.binding.agentId !== agentId + || input.binding.issueId !== nativeIssueId) return null; + for (const skill of input.runtimeContext.skills) { + // Unversioned skills have their own captured asset bundle; they do + // not reference a directory under __versions__. + if (skill.versionId === null) continue; + if (!uuid.test(skill.versionId)) return null; + versions.add(skill.versionId); + } + continue; + } catch { return null; } + } // Legacy/incomplete preparation is not proof that no skill is in use. if (!pins || typeof pins !== "object" || Array.isArray(pins)) return null; for (const version of Object.values(pins)) { diff --git a/server/src/services/skill-version-retention.ts b/server/src/services/skill-version-retention.ts index 4dc46985d00..ab507fd8ee2 100644 --- a/server/src/services/skill-version-retention.ts +++ b/server/src/services/skill-version-retention.ts @@ -33,9 +33,6 @@ export const SKILL_VERSION_RETENTION_DEFAULT_MAX_REMOVALS = 50; export const SKILL_VERSION_RETENTION_INTERVAL_MS = 60 * 60 * 1000; export const SKILL_VERSION_RETENTION_STARTUP_DELAY_MS = 60 * 1000; -/** A run in one of these states can still read (or re-read) its captured skill versions. */ -export const SKILL_VERSION_LIVE_RUN_STATUSES = ["queued", "scheduled_retry", "running", "interrupted"] as const; - const VERSIONS_DIR = "__versions__"; const TRASH_PREFIX = ".trash-"; const UUID_PATTERN = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i;