mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-03 17:51:58 +01:00
fix(server): preserve sessions and defer busy overlap recovery (#18552)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - A web server replacement must preserve remote execution and later task continuation. > - The active-run handoff now survives deletion of the departing server home. > - The next admission still compared disposable instruction and skill paths as configuration. > - A native receipt could also settle without the old adapter returning its terminal metadata. > - A replacement can also collide with a run row locked by the serving controller. > - This change preserves durable session identity and terminal evidence, and defers locked recovery work so overlap can reach readiness. ## Linked Issues or Issue Description Refs #18543, #18366, #9977, #18118. **What happened?** A real Codex/Daytona journey replaced both control-plane servers and deleted their homes. Reads, saves, saved files and active execution passed. A later useful answer used a new provider session. All four run records lacked the native terminal contract. The whole qualification therefore failed. A later attempt also exposed startup failure: the replacement waited for a row lock held by the serving controller and aborted on its one-second lock timeout. **Expected behavior** An unchanged managed instruction entry and immutable skill version retain the session across server cache relocation. Actual content and configuration changes still reset it. A committed native receipt retains its terminal contract even when the originating adapter never returns. A busy recovery row defers through the existing scheduler while the replacement reaches readiness. **Steps to reproduce** Run the `rolling-web-deployment.runner-codex.daytona.recover-controller` Product E2E cell with separate controller homes and shared PostgreSQL/S3. The focused regressions reproduce the failures without a model call. ## What Changed - Read the managed instruction entry digest from the company-scoped durable file head. Do not restore or read the cache for this check. - Omit disposable instruction paths only when that durable authority exists. Fingerprint selected immutable skill versions in both runtime skill inputs. - Preserve the existing local and external instruction path checks, and unversioned skill identity. - Publish the accepted native terminal contract in the same transaction as the canonical result receipt. - Skip locked recovery rows and schedule the same run for retry. Avoid an audit write that would block on the same row. No ownership is transferred while busy. - Keep replacement readiness independent of an in-progress skill selection; migration and cache eviction still require exact pins. Missing durable skill versions still refuse startup. - Project only the instruction digest in SQL instead of loading the entire agent-file manifest per admission. - Add real PostgreSQL lock contention tests for the issue, run and coordinator, including the heartbeat startup caller and later claim. - Limit native test teardown to its owned database on a shared test host. The old behavior also fails on clean master. - Add cache-loss, scope, changed-content, changed-version and finalization regressions. Update the deployment qualification record. ## Verification - Both focused regressions failed before the source changes. - All 249 tests pass across `heartbeat-workspace-session`, `agent-file-durability`, and `paperclip-control-plane-port`. - Server typecheck and exact-revision build pass. - Assembled admission ran 59 passing cases but failed during native-suite teardown. The teardown failure reproduced on clean master. All 16 native resumption cases and teardown pass after the database-scope fix. - All three row-lock regressions fail with the pre-fix lock timeout and pass with deferred recovery. The full recovery suite passes 74 cases, including the startup caller; eight Docker-specific cases were not enabled. - The real attempt at `d2165dec5d` failed when B aborted on the run lock. Its credential and bucket were removed; all three provider sandboxes were independently confirmed absent. The next real attempt at `aae8a53d01` failed a different readiness gate while a run was preparing skill pins. That race now has a failing-before regression; all 33 readiness, durable-file and managed-skill tests pass after the fix. Its owned resources are cleaned. The `a15a12651c` live attempt passed both web replacements: 765 reads, 78 saves and 78 verified saves with zero HTTP probe failures. The first three turns preserved the provider session and native terminal contract. The fourth continuation failed because a safely suspended remote Runner was treated as live/idle; that distinct recovery gap is being handled in a follow-up. All three owned sandboxes were independently confirmed absent. No claim of full Railway qualification yet. ## Risks - The new fingerprint shape may reset a session once when upgrading from older fingerprint metadata. It is stable across subsequent cache replacements. - A versioned skill is identified by its immutable selected version; materialization and admission still enforce availability. Unversioned skills retain their existing path boundary. - A native terminal describes the accepted execution result. Workspace and issue settlement remain separate authorities. - This does not yet prove the ten-deploy staging campaign, natural database autosuspension or the production canary. ## Model Used OpenAI GPT-6 Codex, with reasoning, repository editing, shell execution and live integration-test tools. The exact runtime model variant and context-window size are not exposed in this session. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
7751fc3107
commit
c0550e7010
15 files changed
+239
-56
No files matched your search
@@ -11,7 +11,7 @@ import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.j
|
||||
import { agentFileStore, adoptAgentFiles, fileHash } from "../services/agent-file-store.js";
|
||||
import { invalidateAgentFileProjection, persistDurableAgentFiles } from "../services/agent-file-durability.js";
|
||||
import { materializeInstructionBytes } from "../services/agent-instruction-files.js";
|
||||
import { resolveManagedInstructionsRoot } from "../services/agent-instructions.js";
|
||||
import { agentInstructionsService, resolveManagedInstructionsRoot } from "../services/agent-instructions.js";
|
||||
import { AGENT_FILE_OBJECT_GRACE_MS, collectAgentFileObjects, migrateAgentFiles } from "../services/agent-file-maintenance.js";
|
||||
import { getStorageService } from "../storage/index.js";
|
||||
|
||||
@@ -89,6 +89,28 @@ describe("durable agent file authority", () => {
|
||||
root = resolveManagedInstructionsRoot({ id: agentId, companyId, name: "File owner", adapterConfig: {} });
|
||||
}
|
||||
|
||||
it("reads instruction compatibility from the committed head after deleting the entire home", async () => {
|
||||
await materializeInstructionBytes(root, "AGENTS.md", Buffer.from("Original"));
|
||||
await db.update(agents).set({ adapterConfig: { instructionsBundleMode: "managed", instructionsRootPath: root,
|
||||
instructionsFilePath: path.join(root, "AGENTS.md"), instructionsEntryFile: "AGENTS.md" } }).where(eq(agents.id, agentId));
|
||||
const [agent] = await db.select().from(agents).where(eq(agents.id, agentId));
|
||||
const instructions = agentInstructionsService(db);
|
||||
expect(await instructions.getManagedEntryFingerprint(agent)).toBeNull();
|
||||
await store().read(companyId, agentId, "AGENTS.md", actor());
|
||||
const expected = { entryFile: "AGENTS.md", contentHash: `sha256:${fileHash(Buffer.from("Original"))}` };
|
||||
expect(await instructions.getManagedEntryFingerprint(agent)).toEqual(expected);
|
||||
await replaceFilesystem();
|
||||
expect(await instructions.getManagedEntryFingerprint(agent)).toEqual(expected);
|
||||
await expect(fs.stat(root)).rejects.toMatchObject({ code: "ENOENT" });
|
||||
expect(await instructions.getManagedEntryFingerprint({ ...agent, companyId: randomUUID() })).toBeNull();
|
||||
expect(await instructions.getManagedEntryFingerprint({ ...agent, adapterConfig: {
|
||||
...agent.adapterConfig, instructionsBundleMode: "external",
|
||||
} })).toBeNull();
|
||||
expect(await instructions.getManagedEntryFingerprint({ ...agent, adapterConfig: {
|
||||
...agent.adapterConfig, instructionsEntryFile: "missing.md",
|
||||
} })).toEqual({ entryFile: "missing.md", contentHash: null });
|
||||
});
|
||||
|
||||
it("imports legacy bytes, modes and empty entries, then restores without any old filesystem", async () => {
|
||||
const binary = Buffer.alloc(2 * 1024 * 1024, 0x81);
|
||||
await materializeInstructionBytes(root, "AGENTS.md", Buffer.from("Original"));
|
||||
|
||||
@@ -1,9 +1,9 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { beforeAll, afterAll, afterEach, expect, it, vi } from "vitest";
|
||||
import { eq } from "drizzle-orm";
|
||||
import { agentFileRoots, agents, companies, executionWorkspaces, heartbeatRuns, providerTraceRecords, createDb } from "@paperclipai/db";
|
||||
import { agentFileRoots, agents, companies, companySkills, companySkillVersions, executionWorkspaces, heartbeatRuns, providerTraceRecords, createDb } from "@paperclipai/db";
|
||||
import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
|
||||
import { assertEphemeralServerData } from "../services/ephemeral-server-contract.js";
|
||||
import { assertEphemeralServerData, readEphemeralServerDataEligibility } from "../services/ephemeral-server-contract.js";
|
||||
import { initializeDurableAgentFiles } from "../services/agent-file-durability.js";
|
||||
vi.mock("../storage/index.js", () => ({ getStorageService: () => ({ provider: "s3" }) }));
|
||||
let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
@@ -40,3 +40,28 @@ it("refuses volume-dependent rows before readiness and accepts their durable rep
|
||||
})).rejects.toThrow("abort hire");
|
||||
expect(await db.select().from(agentFileRoots)).toHaveLength(1);
|
||||
});
|
||||
|
||||
it("keeps replacement readiness independent of a serving controller's unfinished skill selection", async () => {
|
||||
vi.stubEnv("PAPERCLIP_EPHEMERAL_SERVER", "true"); vi.stubEnv("PAPERCLIP_DURABLE_AGENT_FILES", "true");
|
||||
const companyId = randomUUID(), agentId = randomUUID(), skillId = randomUUID(), versionId = randomUUID();
|
||||
await db.insert(companies).values({ id: companyId, name: "Overlapping skill preparation", issuePrefix: "PINS" });
|
||||
await db.insert(agents).values({ id: agentId, companyId, name: "Preparing agent" });
|
||||
await db.transaction(tx => initializeDurableAgentFiles(tx, { id: agentId, companyId }));
|
||||
await db.insert(companySkills).values({ id: skillId, companyId, key: "overlap", slug: "overlap", name: "Overlap",
|
||||
sourceType: "managed", markdown: "# Overlap", fileInventory: [{ path: "SKILL.md", kind: "entrypoint" }] });
|
||||
await db.insert(companySkillVersions).values({ id: versionId, companyId, companySkillId: skillId, revisionNumber: 1,
|
||||
fileInventory: [{ path: "SKILL.md", kind: "entrypoint", content: "# Overlap" }] });
|
||||
await db.update(companySkills).set({ currentVersionId: versionId }).where(eq(companySkills.id, skillId));
|
||||
await expect(assertEphemeralServerData(db)).resolves.toBeUndefined();
|
||||
// Admission marks the run running before preparation publishes its skill pins.
|
||||
// That prevents cache eviction/migration, but does not make durable bytes local.
|
||||
const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId, invocationSource: "on_demand",
|
||||
status: "running", contextSnapshot: {} }).returning();
|
||||
await expect(assertEphemeralServerData(db)).resolves.toBeUndefined();
|
||||
expect((await readEphemeralServerDataEligibility(db)).blockers).toContain("skills_require_migration");
|
||||
await db.update(heartbeatRuns).set({ contextSnapshot: { skillVersions: { overlap: versionId } } }).where(eq(heartbeatRuns.id, run.id));
|
||||
await expect(assertEphemeralServerData(db)).resolves.toBeUndefined();
|
||||
expect((await readEphemeralServerDataEligibility(db)).eligible).toBe(true);
|
||||
await db.update(companySkills).set({ currentVersionId: null }).where(eq(companySkills.id, skillId));
|
||||
await expect(assertEphemeralServerData(db)).rejects.toThrow("skills_require_migration");
|
||||
});
|
||||
@@ -2703,6 +2703,31 @@ describe("effective run session config freshness", () => {
|
||||
expect(next.fingerprints.sessionFingerprint.canonicalJson).not.toContain("Version two instructions");
|
||||
});
|
||||
|
||||
it("retains a session across relocated durable instruction and versioned skill caches", async () => {
|
||||
const metadata = (home: string, contentHash = "sha256:entry-v1", versionId = "skill-version-1") => {
|
||||
const runtimeSkills = [{ key: "paperclip", runtimeName: "paperclip", versionId,
|
||||
currentVersionId: versionId, source: `${home}/packages/skill`, sourceStatus: "available" }];
|
||||
return buildSessionConfigMetadata({
|
||||
adapterType: "paperclip_runner",
|
||||
effectiveAdapterConfig: { model: "gpt-5.4-mini", instructionsBundleMode: "managed",
|
||||
instructionsRootPath: `${home}/instructions`, instructionsFilePath: `${home}/instructions/AGENTS.md`,
|
||||
instructionsEntryFile: "AGENTS.md", paperclipRuntimeSkills: runtimeSkills },
|
||||
managedInstructions: { entryFile: "AGENTS.md", contentHash },
|
||||
runtimeSkills,
|
||||
});
|
||||
};
|
||||
const before = await metadata("/retired-controller");
|
||||
const after = await metadata("/replacement-controller");
|
||||
const freshness = (next: SessionConfigMetadata) => resolveTaskSessionConfigFreshness({
|
||||
hasTaskSession: true, configuredModel: "gpt-5.4-mini",
|
||||
taskSessionParams: sessionParamsWithConfigMetadata(before), configMetadata: next,
|
||||
});
|
||||
expect(freshness(after).reset).toBe(false);
|
||||
expect(after.fingerprints.sessionFingerprint.canonicalJson).not.toContain("/replacement-controller");
|
||||
expect(freshness(await metadata("/replacement-controller", "sha256:entry-v2")).changedCategories).toContain("instructions");
|
||||
expect(freshness(await metadata("/replacement-controller", "sha256:entry-v1", "skill-version-2")).changedCategories).toContain("runtimeSkills");
|
||||
});
|
||||
|
||||
it("does not read unbounded legacy instructions paths for config fingerprints", async () => {
|
||||
const root = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-session-fingerprint-"));
|
||||
const instructionsPath = path.join(root, "AGENTS.md");
|
||||
|
||||
@@ -1530,7 +1530,7 @@ describe("goal-less native task-session publication", () => {
|
||||
}
|
||||
}, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS);
|
||||
afterAll(async () => {
|
||||
await closeRegisteredClients(process.env.PAPERCLIP_TEST_DATABASE_URL ?? database!.connectionString);
|
||||
await closeRegisteredClients(process.env.PAPERCLIP_TEST_DATABASE_URL ?? database!.connectionString, { scope: "database" });
|
||||
await database?.cleanup();
|
||||
});
|
||||
|
||||
|
||||
@@ -2,8 +2,8 @@ import { adoptAgentFiles, inspectAgentFile, fileHash, agentFilePath, MAX_AGENT_F
|
||||
import { invalidateAgentFileProjection, persistDurableAgentFiles, restoreDurableAgentFiles } from "./agent-file-durability.js";
|
||||
import fs from "node:fs/promises";
|
||||
import path from "node:path";
|
||||
import { and, eq } from "drizzle-orm";
|
||||
import { agentInstructionHeads, agents, fileStorageDb, type Db } from "@paperclipai/db";
|
||||
import { and, eq, sql } from "drizzle-orm";
|
||||
import { agentFileRoots, agentInstructionHeads, agents, fileStorageDb, type Db } from "@paperclipai/db";
|
||||
import { instructionPath, assertInstructionPathSafe, instructionBytes, readInstructionBytes } from "./agent-instruction-files.js";
|
||||
import { notFound, unprocessable } from "../errors.js";
|
||||
import { resolveHomeAwarePath, resolvePaperclipInstanceRoot } from "../home-paths.js";
|
||||
@@ -506,6 +506,18 @@ export function syncInstructionsBundleConfigFromFilePath(
|
||||
|
||||
export function agentInstructionsService(db?: Db) {
|
||||
if (db) db = fileStorageDb(db);
|
||||
async function getManagedEntryFingerprint(agent: AgentLike): Promise<{ entryFile: string; contentHash: string | null } | null> {
|
||||
if (!db || agentInstructionsBundleMode(agent) !== "managed") return null;
|
||||
const state = deriveBundleState(agent);
|
||||
const [head] = await db.select({ sha256: sql<string | null>`jsonb_path_query_first(
|
||||
${agentFileRoots.entries}, '$[*] ? (@.path == $entry && @.kind == "file")',
|
||||
jsonb_build_object('entry', ${state.entryFile}::text)) ->> 'sha256'` }).from(agentFileRoots)
|
||||
.where(and(eq(agentFileRoots.companyId, agent.companyId), eq(agentFileRoots.agentId, agent.id)));
|
||||
if (!head) return null;
|
||||
// This head is the file authority. A missing or retired local projection
|
||||
// must not make identical instructions appear to change at admission.
|
||||
return { entryFile: state.entryFile, contentHash: head.sha256 ? `sha256:${head.sha256}` : null };
|
||||
}
|
||||
async function assertUnversionedEntry(agent: AgentLike, entryFile: string, connection: Db | Parameters<Parameters<Db["transaction"]>[0]>[0] | undefined = db) {
|
||||
if (!connection) throw unprocessable("Bundle initialization requires the database-backed instructions service");
|
||||
const [head] = await connection.select().from(agentInstructionHeads).where(and(eq(agentInstructionHeads.companyId, agent.companyId),
|
||||
@@ -871,6 +883,7 @@ export function agentInstructionsService(db?: Db) {
|
||||
}
|
||||
|
||||
return {
|
||||
getManagedEntryFingerprint,
|
||||
getBundleLocation: (agent: AgentLike) => withProjection(agent, false, getBundleLocation),
|
||||
getBundle: (agent: AgentLike) => withProjection(agent, false, getBundle),
|
||||
readFile: (agent: AgentLike, relativePath: string) => withProjection(agent, false, current => readFile(current, relativePath)),
|
||||
|
||||
@@ -37,11 +37,11 @@ export function assertEphemeralServerConfig(config: Pick<Config, "databaseUrl" |
|
||||
* volume data is adopted by the owning maintenance commands before removal. */
|
||||
export async function assertEphemeralServerData(db: Db) {
|
||||
if (!ephemeralServerEnabled()) return;
|
||||
const { blockers } = await readEphemeralServerDataEligibility(db);
|
||||
const { blockers } = await readEphemeralServerDataEligibility(db, { purpose: "startup" });
|
||||
if (blockers.length) throw new Error(`Volume-free server data is not eligible: ${blockers.join(", ")}`);
|
||||
}
|
||||
|
||||
export async function readEphemeralServerDataEligibility(db: Db) {
|
||||
export async function readEphemeralServerDataEligibility(db: Db, options: { purpose?: "migration" | "startup" } = {}) {
|
||||
const blockers: string[] = [];
|
||||
const [missingFiles] = await db.select({ id: agents.id }).from(agents).leftJoin(agentFileRoots,
|
||||
and(eq(agentFileRoots.companyId, agents.companyId), eq(agentFileRoots.agentId, agents.id)))
|
||||
@@ -67,7 +67,7 @@ export async function readEphemeralServerDataEligibility(db: Db) {
|
||||
const skills = skillInstallationMaintenance(db);
|
||||
let after: string | undefined;
|
||||
for (;;) {
|
||||
const page = await skills.eligibility({ after, limit: 50 });
|
||||
const page = await skills.eligibility({ after, limit: 50, purpose: options.purpose });
|
||||
if (page.results.some(skill => skill.status !== "version_backed")) { blockers.push("skills_require_migration"); break; }
|
||||
if (page.inspected < 50 || !page.after) break;
|
||||
after = page.after;
|
||||
|
||||
@@ -1012,6 +1012,9 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
|
||||
if (disposition.kind === "awaiting_evidence" && disposition.retryAt) {
|
||||
scheduleNativeSessionResumeDispatch(disposition.runId, disposition.retryAt, "restart");
|
||||
}
|
||||
// No ownership decision was made for a locked row. Writing an event
|
||||
// would wait on that same row and defeat the non-blocking startup scan.
|
||||
if (disposition.kind === "awaiting_evidence" && disposition.reason === "recovery_rows_busy") continue;
|
||||
const run = await getRun(disposition.runId);
|
||||
if (run) {
|
||||
const isClaim =
|
||||
|
||||
@@ -5,6 +5,7 @@ import { getServerAdapter } from "../../adapters/index.js";
|
||||
import type { prepareManagedAiRuntime } from "../ai-connection-runtime.js";
|
||||
import { aiConnectionSessionCompatibilityInputs, managedAiSessionIdentityCompatible } from "../ai-connection-session.js";
|
||||
import { createNativeSessionHandoffLoader } from "../native-runtime/native-session-handoff.js";
|
||||
import { agentInstructionsService } from "../agent-instructions.js";
|
||||
import {
|
||||
describeSessionResetReason, shouldResetTaskSessionForWake, isCanonicalSessionIdForAdapter,
|
||||
normalizeResumeParamsForAdapter, normalizeSessionParams, requiresCanonicalSessionIds,
|
||||
@@ -87,6 +88,7 @@ export async function prepareHeartbeatSession(db: Db, input: HeartbeatSessionPre
|
||||
agentIdentityKeyId: agentIdentity?.keyId,
|
||||
adapterType: agent.adapterType,
|
||||
effectiveAdapterConfig: runtimeConfig,
|
||||
managedInstructions: await agentInstructionsService(db).getManagedEntryFingerprint({ ...agent, adapterConfig: runtimeConfig }),
|
||||
managedAiHome: managedAiRuntime?.home,
|
||||
agentRuntimeConfig: agent.runtimeConfig,
|
||||
issueOverrides: issueAssigneeOverrides,
|
||||
|
||||
@@ -2838,6 +2838,7 @@ export async function buildEffectiveRunSessionConfigMetadata(input: {
|
||||
adapterType: string;
|
||||
effectiveAdapterConfig: Record<string, unknown>;
|
||||
managedAiHome?: string;
|
||||
managedInstructions?: { entryFile: string; contentHash: string | null } | null;
|
||||
agentRuntimeConfig: unknown;
|
||||
issueOverrides: unknown;
|
||||
workspaceConfig: unknown;
|
||||
@@ -2851,12 +2852,30 @@ export async function buildEffectiveRunSessionConfigMetadata(input: {
|
||||
agentIdentityKeyId?: string;
|
||||
}): Promise<EffectiveRunSessionConfigMetadata> {
|
||||
const secretManifest = input.secretManifest ?? [];
|
||||
const instructions = await resolveInstructionsConfigFingerprintMetadata(
|
||||
input.effectiveAdapterConfig,
|
||||
);
|
||||
const instructions = input.managedInstructions ? {
|
||||
configured: true, bundleMode: "managed", entryFile: input.managedInstructions.entryFile,
|
||||
pathKind: "absolute", readPolicy: "root_bound", readable: input.managedInstructions.contentHash !== null,
|
||||
...(input.managedInstructions.contentHash !== null ? { contentHash: input.managedInstructions.contentHash } : {}),
|
||||
} : await resolveInstructionsConfigFingerprintMetadata(input.effectiveAdapterConfig);
|
||||
const effectiveAdapterConfig = { ...managedAiSessionFingerprintConfig(input.effectiveAdapterConfig, input.managedAiHome) };
|
||||
if (input.managedInstructions) {
|
||||
delete effectiveAdapterConfig.instructionsRootPath;
|
||||
delete effectiveAdapterConfig.instructionsFilePath;
|
||||
}
|
||||
// A selected immutable skill version is its identity; source is merely its
|
||||
// disposable projection. Unversioned/local skills retain their path boundary.
|
||||
const versionedSkills = (value: unknown) => Array.isArray(value) ? value.map(entry => {
|
||||
const skill = parseObject(entry);
|
||||
return readNonEmptyString(skill.versionId)
|
||||
? { key: skill.key, runtimeName: skill.runtimeName, versionId: skill.versionId }
|
||||
: entry;
|
||||
}) : value;
|
||||
if (effectiveAdapterConfig.paperclipRuntimeSkills !== undefined) {
|
||||
effectiveAdapterConfig.paperclipRuntimeSkills = versionedSkills(effectiveAdapterConfig.paperclipRuntimeSkills);
|
||||
}
|
||||
const categoryValues = buildSessionConfigCategoryValues({
|
||||
adapterType: input.adapterType,
|
||||
effectiveAdapterConfig: managedAiSessionFingerprintConfig(input.effectiveAdapterConfig, input.managedAiHome),
|
||||
effectiveAdapterConfig,
|
||||
agentRuntimeConfig: input.agentRuntimeConfig,
|
||||
instructions,
|
||||
issueOverrides: input.issueOverrides,
|
||||
@@ -2866,7 +2885,7 @@ export async function buildEffectiveRunSessionConfigMetadata(input: {
|
||||
projectEnv: input.projectEnv,
|
||||
routineEnv: input.routineEnv,
|
||||
secretManifest,
|
||||
runtimeSkills: input.runtimeSkills,
|
||||
runtimeSkills: versionedSkills(input.runtimeSkills),
|
||||
agentConfigRevision: input.agentConfigRevision ?? null,
|
||||
agentIdentityKeyId: input.agentIdentityKeyId,
|
||||
});
|
||||
|
||||
@@ -560,11 +560,17 @@ export async function claimNativeRestartRecoveries(input: {
|
||||
await tx.execute(
|
||||
sql`select set_config('statement_timeout', '15000', true), set_config('lock_timeout', '1000', true)`,
|
||||
);
|
||||
await tx
|
||||
const busy = () => ({ kind: "awaiting_evidence" as const, runId: candidate.runId,
|
||||
reason: "recovery_rows_busy", retryAt: new Date(now.getTime() + 1_000) });
|
||||
// During overlap the serving controller can be committing this run.
|
||||
// Defer that concrete recovery obligation instead of failing startup
|
||||
// or waiting while holding a lock the serving controller needs.
|
||||
const [issue] = await tx
|
||||
.select({ id: issues.id })
|
||||
.from(issues)
|
||||
.where(eq(issues.id, candidate.issueId))
|
||||
.for("update");
|
||||
.for("update", { skipLocked: true });
|
||||
if (!issue) return busy();
|
||||
const row = await tx
|
||||
.select({
|
||||
run: heartbeatRuns,
|
||||
@@ -586,15 +592,11 @@ export async function claimNativeRestartRecoveries(input: {
|
||||
),
|
||||
)
|
||||
.where(eq(heartbeatRuns.id, candidate.runId))
|
||||
.for("update")
|
||||
.for("update", { skipLocked: true })
|
||||
.limit(1)
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (!row) {
|
||||
return {
|
||||
kind: "blocked",
|
||||
runId: candidate.runId,
|
||||
reason: "recovery_rows_missing",
|
||||
} as const;
|
||||
return busy();
|
||||
}
|
||||
if (isNativeRunnerOwnershipHeld(row.run)) {
|
||||
return {
|
||||
|
||||
@@ -36,6 +36,7 @@ import {
|
||||
EMBEDDED_POSTGRES_TEST_TIMEOUT_MS,
|
||||
} from "@paperclipai/db";
|
||||
import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
|
||||
import { heartbeatService } from "../heartbeat.js";
|
||||
|
||||
import {
|
||||
canonicalNativeRuntimeContextDigest,
|
||||
@@ -322,6 +323,41 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", (
|
||||
};
|
||||
}
|
||||
|
||||
it.each(["issue", "run", "coordinator"] as const)("defers a busy %s row without blocking replacement startup", async kind => {
|
||||
const fixture = await seedRun(`OVERLAP-BUSY-${kind}`);
|
||||
let acquired!: () => void, release!: () => void;
|
||||
const locked = new Promise<void>(resolve => { acquired = resolve; });
|
||||
const released = new Promise<void>(resolve => { release = resolve; });
|
||||
const holder = fixture.db.transaction(async tx => {
|
||||
if (kind === "issue") await tx.select().from(issues).where(eq(issues.id, fixture.issueId)).for("update");
|
||||
else if (kind === "run") await tx.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, fixture.runId)).for("update");
|
||||
else await tx.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, fixture.runId)).for("update");
|
||||
acquired(); await released;
|
||||
});
|
||||
await locked;
|
||||
const now = new Date();
|
||||
const heartbeat = heartbeatService(fixture.db);
|
||||
try {
|
||||
const dispositions = await claimNativeRestartRecoveries({ db: fixture.db, controller: successor,
|
||||
restartKind: "hard", runIds: [fixture.runId], now });
|
||||
expect(dispositions).toEqual([{ kind: "awaiting_evidence", runId: fixture.runId,
|
||||
reason: "recovery_rows_busy", retryAt: new Date(now.getTime() + 1_000) }]);
|
||||
const [coordinator] = await fixture.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, fixture.runId));
|
||||
expect(coordinator.controllerGeneration).toBe(0);
|
||||
let deadline: ReturnType<typeof setTimeout> | undefined;
|
||||
try {
|
||||
const startup = await Promise.race([
|
||||
heartbeat.recoverNativeRunsAfterRestart(now, [fixture.runId]),
|
||||
new Promise<never>((_resolve, reject) => { deadline = setTimeout(() => reject(new Error("startup blocked on the serving controller")), 2_000); }),
|
||||
]);
|
||||
expect(startup).toMatchObject({ claims: [], awaitingEvidenceRunIds: [fixture.runId] });
|
||||
} finally { clearTimeout(deadline); }
|
||||
} finally { await heartbeat.drainActiveRunExecutions(); release(); await holder; }
|
||||
const [next] = await claimNativeRestartRecoveries({ db: fixture.db, controller: successor,
|
||||
restartKind: "hard", runIds: [fixture.runId] });
|
||||
expect(next.kind).toBe("bootstrap_incomplete");
|
||||
});
|
||||
|
||||
it.skipIf(!process.env.PAPERCLIP_TEST_RUNNER_CONTAINER).each(["warm", "hard", "legacy", "changed", "same-server", "diskless", "diskless-per-turn", "diskless-per-turn-read-retry"])("preserves remote warm continuity across completed turns: %s", async (scenario) => {
|
||||
const hardRestart = !["warm", "same-server"].includes(scenario);
|
||||
const diskless = scenario.startsWith("diskless");
|
||||
|
||||
@@ -733,6 +733,10 @@ describe("PaperclipControlPlanePort conformance", () => {
|
||||
.where(eq(nativeRunFinalizations.runId, identity.runId))).resolves.toEqual([
|
||||
expect.objectContaining({ phase: "committed" }),
|
||||
]);
|
||||
// Completion and restart reconciliation must retain the contract even if
|
||||
// the originating heartbeat never returns from its adapter completion.
|
||||
const [completedRun] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, identity.runId));
|
||||
expect(completedRun.resultJson?.nativeTerminal).toEqual(CONTROL_PLANE_CONFORMANCE_TERMINAL);
|
||||
await expect(db.select().from(issues).where(eq(issues.id, identity.issueId))).resolves.toEqual([
|
||||
expect.objectContaining({ status: "in_progress", statusVersion: 1 }),
|
||||
]);
|
||||
|
||||
@@ -568,6 +568,9 @@ export class PaperclipControlPlanePort implements ControlPlanePort {
|
||||
nativePhaseUpdatedAt: new Date(),
|
||||
resultJson: {
|
||||
...record(run.resultJson),
|
||||
// Publish the accepted contract with its durable receipt. Recovery
|
||||
// may settle this run without the original adapter ever returning.
|
||||
nativeTerminal: value.terminal,
|
||||
prpTurnTerminalState: value.terminal.turnTerminalState,
|
||||
prpRunTerminalState: value.terminal.runTerminalState,
|
||||
prpReportedWorkDisposition: value.terminal.reportedWorkDisposition,
|
||||
|
||||
@@ -158,13 +158,16 @@ export function skillInstallationMaintenance(db: Db, storage?: StorageService) {
|
||||
return { inspected: skills.length, after: skills.at(-1)?.id ?? null, results };
|
||||
}
|
||||
|
||||
async function eligibility(options: Page = {}) {
|
||||
async function eligibility(options: Page & { purpose?: "migration" | "startup" } = {}) {
|
||||
const skills = await db.select().from(companySkills).where(scope(options)).orderBy(asc(companySkills.id)).limit(limit(options));
|
||||
const results = [];
|
||||
for (const skill of skills) {
|
||||
const pins = await protectedVersions(db, skill.companyId);
|
||||
// Migration/eviction must wait for exact pins. Replacement readiness only
|
||||
// inspects installation durability: a serving controller may legitimately
|
||||
// be between run admission and publishing its immutable skill selection.
|
||||
const pins = options.purpose === "startup" ? undefined : await protectedVersions(db, skill.companyId);
|
||||
let status = linkedLocal(skill) ? "linked_local" : legacyManaged(skill) ? "legacy_managed" : "version_backed";
|
||||
if (pins.versions === null) status = "unsupported_active_run";
|
||||
if (pins?.versions === null) status = "unsupported_active_run";
|
||||
else if (!linkedLocal(skill) && !skill.currentVersionId) status = "missing_current_version";
|
||||
else if (skill.sourceType === "catalog" && !skill.metadata?.originVersionId) status = "legacy_catalog_origin";
|
||||
else if (status === "version_backed" && skill.currentVersionId) {
|
||||
|
||||
Reference in new issue
Block a user