diff --git a/doc/DEVELOPING.md b/doc/DEVELOPING.md index 14a7cd31419..a4c931f945a 100644 --- a/doc/DEVELOPING.md +++ b/doc/DEVELOPING.md @@ -3557,6 +3557,29 @@ or cancelled result must retire that session and collect its managed files befor heartbeat stops the reusable sandbox. A later retry can resume the same sandbox without inheriting the stopped provider transport. +A rejected warm-session close retains its owner, even when no managed instruction +copy exists. Idle expiry, failed-turn release, and restart must not report that +owner as quiet or launch a replacement over it. The next retirement attempt uses +the same owner and must finish provider closure and stopped-file collection. +The existing explicit-reset exception still requires its typed retirement proof. + +`heartbeatService(...).prepareIdleWarmNativeSessionsForSleep({ canRetire })` +delegates to the native owner's retirement path for idle owners. Its caller owns +the admission hold and supplies a synchronous +current-authority check. The check runs before each retirement and after its +awaited effects. A started cleanup settles if the hold expires, but no further +owner retires under that expired hold. Busy turns remain active and do not inherit +a close-on-release request from an abandoned sleep attempt. Counts describe the +attempt; the caller must still validate its hold and final local-work proof. +Diagnostic-only inspection must never invoke this resource preparation. + +The privileged `task-drain?idleSleepSafety=1` preflight invokes this operation only +for the exact current, unexpired idle owner with no active admitted work. It tracks +the actual cleanup promise, rejects concurrent work or a changed hold, and then +takes a fresh durable/local safety snapshot. Busy or failed retirement keeps the +instance awake. Plain drain reads and `idleSleepDiagnostics=1` remain passive with +respect to native retirement. + An idle warm owner proves only the exact prior run in the same company, session, workspace and environment. Mutable configuration compatibility is checked separately: changed settings retire the owned provider before a replacement can @@ -5017,6 +5040,39 @@ database without loading home `.env` files or repairing worktree config. See secret handling, initialization order and migration interfaces. +### Dedicated native recovery test database + +The native idle continuation fixture reads `PAPERCLIP_RECOVERY_TEST_DATABASE_URL` +and passes it through `nativeRestartFixture({ dedicatedDatabaseUrl })`. The URL +must name a dedicated PostgreSQL database ending in `_test`, without database-name +or connection-option overrides. This explicit option applies pending migrations +and closes registered clients at cleanup; it does not create or drop that +database. Its provisioning owner remains responsible for disposal. The environment +variable alone does not opt other `startRecoveryPostgres()` callers into shared +storage: they retain their existing fresh-database allocation and cleanup. + +The native idle-sleep continuation integration uses a controlled local provider +with the real heartbeat queue, authorized comment wake, native session +checkpoint, files, retained secret encryption and fresh resolution, and owned +idle preflight. Credential resolution through the secret owner does not prove +production backend credential injection. It uses an explicit two-connection +main pool, alongside the production auxiliary pools, and verifies the driver's +actual setting. It does not +exercise the remote Runner harness backup transport or a live provider. + +The workspace runtime lifecycle test uses the actual local runtime owner and an +ephemeral HTTP child. Normal run release must remove the original PID, listener, +and service registry entry before the same owner starts another useful service +from the retained workspace files. This is physical local resource retirement +and next-use coverage; it does not establish global tenant quiet. + +The plugin idle manager lifecycle test uses a real SDK worker process and a +controlled host state handler that persists worker requests to a temporary file. +It checks the actual manifest generation, refusal of work under the owned hold, +and a second useful write after release. This boundary does not exercise the +production plugin database loader or plugin jobs. Unsupported manifests, active +manual resources, and missing completion receipts remain sleep blockers. + ### Read resilience during deploys Read surfaces use `useQueryView` and `QueryErrorState`. Transient refetch failures keep loaded data visible. An initial outage keeps a quiet placeholder until recovery. Authorization, missing resources and validation failures still show readable errors. Exhausted first reads on a reachable server offer Retry, including task filters and hierarchy reads. Health probes and the socket own app-wide recovery; a successful component load may be an older in-flight response or a local no-op. Sensitive manual reads use `useResilientFetch`; browser viewer URLs and pairing previews remain in component memory and never enter a shared cache. diff --git a/server/src/__tests__/heartbeat-reviewed-chat-binding.integration.test.ts b/server/src/__tests__/heartbeat-reviewed-chat-binding.integration.test.ts index 7f1e4dd2295..be5b2e57a46 100644 --- a/server/src/__tests__/heartbeat-reviewed-chat-binding.integration.test.ts +++ b/server/src/__tests__/heartbeat-reviewed-chat-binding.integration.test.ts @@ -35,6 +35,7 @@ import { listAuthorizedChatAttachments, resolveExternalChatResponseWaitAuthorization, } from "../services/native-runtime/chat-attachment-reuse.js"; +import * as chatAttachmentReuse from "../services/native-runtime/chat-attachment-reuse.js"; import { decideIssueReviewPathRecovery } from "../services/recovery/review-path-recovery.js"; import { resolveCurrentWakeCommentsBinding } from "../services/native-runtime/current-wake-comments.js"; import { commitNativeStatusDecision } from "../services/native-runtime/status-decision-committer.js"; @@ -651,6 +652,49 @@ describe.each(["slack", "discord"] as const)("reviewed %s execution binding", (p } }); + it.each(["authorized", "principal_denied", "binding_denied"] as const)( + "rechecks full authorization when inbound processing commits between reads: %s", + async (outcome) => { + await db.update(chatDeliveries) + .set({ state: "processing", processedAt: null }) + .where(eq(chatDeliveries.id, deliveryId)); + const authorize = chatAttachmentReuse.authorizeChatConversationForBoundRun; + const observed = vi.spyOn(chatAttachmentReuse, "authorizeChatConversationForBoundRun") + .mockImplementationOnce(async (...args) => { + try { + return await authorize(...args); + } catch (error) { + expect(error).toMatchObject({ message: "paperclip_runner_chat_attachment_binding_denied" }); + await db.update(chatDeliveries) + .set({ state: "processed", processedAt: new Date() }) + .where(eq(chatDeliveries.id, deliveryId)); + throw error; + } + }); + try { + if (outcome === "principal_denied") { + await db.update(companyMemberships).set({ status: "suspended" }) + .where(eq(companyMemberships.principalId, userId)); + } + if (outcome === "binding_denied") { + await db.update(chatConversations).set({ issueId: otherIssueId }) + .where(eq(chatConversations.id, conversationId)); + } + await expect(attest()).resolves.toBe(outcome === "authorized"); + if (outcome === "authorized") expect(observed).toHaveBeenCalledTimes(2); + } finally { + observed.mockRestore(); + await db.update(companyMemberships).set({ status: "active" }) + .where(eq(companyMemberships.principalId, userId)); + await db.update(chatConversations).set({ issueId }) + .where(eq(chatConversations.id, conversationId)); + await db.update(chatDeliveries) + .set({ state: "processed", processedAt: new Date() }) + .where(eq(chatDeliveries.id, deliveryId)); + } + }, + ); + it("waits for exact inbound delivery processing to commit before attesting", async () => { await db .update(chatDeliveries) diff --git a/server/src/__tests__/helpers/native-restart-fixture.ts b/server/src/__tests__/helpers/native-restart-fixture.ts new file mode 100644 index 00000000000..2607aba738b --- /dev/null +++ b/server/src/__tests__/helpers/native-restart-fixture.ts @@ -0,0 +1,254 @@ +import { startRecoveryPostgres } from "./recovery-postgres-fixture.js"; +import { randomUUID } from "node:crypto"; +import { existsSync } from "node:fs"; +import { mkdtemp, rm } from "node:fs/promises"; +import { createServer, type Server } from "node:http"; +import { tmpdir } from "node:os"; +import { resolve } from "node:path"; +import { eq } from "drizzle-orm"; +import { agents, companies, closeRegisteredClients, createDb, heartbeatRuns, issues, nativeRunFinalizations, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS } from "@paperclipai/db"; +import { afterAll, afterEach, beforeAll, it } from "vitest"; +import { defaultCapabilityRunnerdBinary } from "../../vendor/paperclip-runner/index.js"; +import { startEmbeddedPostgresTestDatabase } from "./embedded-postgres.js"; +import { registerRunnerPrpAuthority, runnerPrpWebSocketInternals, setupRunnerPrpWebSocketServer } from "../../realtime/runner-prp-ws.js"; +import { readProcessStartedAt } from "../../services/hot-restart.js"; +import { prepareNativeHeartbeatRun } from "../../services/native-runtime/prepare-native-run.js"; +import { createIsolatedTestHome } from "./isolated-test-home.js"; +import { type NativeControllerIdentity } from "../../services/native-runtime/native-restart-recovery.js"; +import { waitForPendingRunFailureReports } from "../../services/run-failure-report.js"; +import { registerLocalRuntimeApiServer } from "../../runtime-api.js"; + +export const fakeCodexAppServer = resolve( + import.meta.dirname, + "../../../../packages/paperclip-runner/runner/target/debug/fake-codex-app-server", +); +const binariesAvailable = + existsSync(defaultCapabilityRunnerdBinary()) && existsSync(fakeCodexAppServer); +export const realProcessIt: typeof it.skip = binariesAvailable ? it : it.skip; + +async function closeServer(server: Server | null): Promise { + if (!server) return; + server.closeAllConnections(); + if (!server.listening) return; + await new Promise((resolveClose) => server.close(() => resolveClose())); +} + +export function processAlive(pid: number | null): boolean { + if (!pid) return false; + try { + process.kill(pid, 0); + return true; + } catch { + return false; + } +} + +export async function waitForCondition( + description: string, + condition: () => boolean | Promise, + timeoutMs = 10_000, +): Promise { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (await condition()) return; + await new Promise((resolveWait) => setTimeout(resolveWait, 25)); + } + throw new Error(`Timed out waiting for ${description}`); +} + +export async function stopOwnedProcessGroup( + pid: number | null, + expectedStateRoot: string, +): Promise { + if (!pid || !processAlive(pid)) return; + const command = await import("node:child_process").then( + ({ execFileSync }) => + execFileSync("ps", ["-p", String(pid), "-o", "command="], { + encoding: "utf8", + }).trim(), + ); + if (!command.includes(expectedStateRoot)) { + throw new Error(`Refusing to signal process ${pid}; ownership changed`); + } + if (process.platform !== "win32") process.kill(-pid, "SIGKILL"); + else process.kill(pid, "SIGKILL"); + await waitForCondition(`owned process ${pid} to exit`, () => !processAlive(pid)); +} + +export function nativeRestartFixture(options: { dedicatedDatabaseUrl?: string } = {}) { + let temporary: Awaited>; + let runtimeRoot: string; + let paperclipHome: string; + let server: Server | null = null; + let apiUrl: string; + const originalPaperclipHome = process.env.PAPERCLIP_HOME; + let isolatedHome: ReturnType; + + const companyId = randomUUID(); + const agentId = randomUUID(); + let successor: NativeControllerIdentity; + + beforeAll(async () => { + temporary = await startRecoveryPostgres("native-runner-restart-recovery-", options); + runtimeRoot = await mkdtemp(resolve(tmpdir(), "native-restart-runtime-")); + isolatedHome = createIsolatedTestHome(); + paperclipHome = isolatedHome.home; + process.env.PAPERCLIP_HOME = paperclipHome; + const controllerStartedAt = await readProcessStartedAt(process.pid); + successor = { + bootId: randomUUID(), + pid: process.pid, + processStartedAt: controllerStartedAt + ? new Date(controllerStartedAt) + : new Date(), + }; + server = createServer(); + registerLocalRuntimeApiServer(server); + await new Promise((resolveListen) => + server!.listen(0, "127.0.0.1", resolveListen), + ); + const address = server.address(); + if (!address || typeof address === "string") { + throw new Error("Expected restart recovery TCP listener"); + } + apiUrl = `http://127.0.0.1:${address.port}`; + setupRunnerPrpWebSocketServer(server, { apiUrl }); + + const db = createDb(temporary.connectionString); + try { + await db.insert(companies).values({ + id: companyId, + name: "Native restart recovery", + issuePrefix: options.dedicatedDatabaseUrl || process.env.PAPERCLIP_RECOVERY_TEST_DATABASE_URL + ? `NRR${companyId.replaceAll("-", "").slice(0, 12).toUpperCase()}` : "NRR", + requireBoardApprovalForNewAgents: false, + }); + await db.insert(agents).values({ + id: agentId, + companyId, + name: "Native restart runner", + role: "engineer", + status: "active", + adapterType: "paperclip_runner", + adapterConfig: { provider: "codex" }, + runtimeConfig: {}, + permissions: {}, + }); + } finally { await db.$client.end(); } + }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS); + + afterEach(async () => { + // Each case creates its own clients. With the long waits in other files, + // idle reaping no longer keeps their total below Postgres's connection cap. + await waitForPendingRunFailureReports(); + if (temporary) await closeRegisteredClients(temporary.connectionString, { scope: "database" }); + }); + + afterAll(async () => { + runnerPrpWebSocketInternals.resetForTests(); + await closeServer(server); + await temporary?.cleanup(); + if (runtimeRoot) await rm(runtimeRoot, { recursive: true, force: true }); + isolatedHome?.cleanup(); + if (originalPaperclipHome === undefined) delete process.env.PAPERCLIP_HOME; + else process.env.PAPERCLIP_HOME = originalPaperclipHome; + }); + + async function seedRun( + label: string, + existingDb?: ReturnType, + ) { + const db = existingDb ?? createDb(temporary.connectionString); + const issueId = randomUUID(); + const runId = randomUUID(); + await db.insert(issues).values({ + id: issueId, + companyId, + identifier: `NRR-${label}`, + title: `Recover ${label}`, + status: "in_progress", + priority: "medium", + workMode: "standard", + assigneeAgentId: agentId, + }); + const [run] = await db + .insert(heartbeatRuns) + .values({ + id: runId, + companyId, + agentId, + status: "running", + runtimeMode: "native", + nativeIssueId: issueId, + invocationSource: "assignment", + triggerDetail: "system", + contextSnapshot: { issueId }, + }) + .returning(); + if (!run) throw new Error("Failed to seed native restart run"); + await db + .update(issues) + .set({ executionRunId: runId }) + .where(eq(issues.id, issueId)); + const native = await prepareNativeHeartbeatRun({ + db, + run, + issue: { + id: issueId, + title: `Recover ${label}`, + description: null, + reviewPolicy: null, + }, + environmentLeaseId: `lease-${label}`, + }); + await db.insert(nativeRunFinalizations).values({ + runId, + companyId, + issueId, + phase: "observed", + }); + const [preparedRun] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)); + return { db, issueId, runId, native, preparedProfile: preparedRun.runnerProfileJson }; + } + + function transportOptions( + fixture: Awaited>, + stateDirectory: string, + ) { + return { + runnerBinary: defaultCapabilityRunnerdBinary(), + codexCommand: fakeCodexAppServer, + codexArgs: [ + "--state-file", + resolve(stateDirectory, "fake-codex-state.json"), + "--call-log", + resolve(stateDirectory, "fake-codex-calls.log"), + ], + stateDirectory, + lifecyclePolicy: { mode: "warm" as const, idleTimeoutMs: 60_000 }, + prpIdentity: { + runnerInstanceId: fixture.native.runnerInstanceId, + environmentLeaseId: fixture.native.environmentLeaseId, + runId: fixture.runId, + normalizedSessionId: fixture.native.normalizedSessionId, + turnId: fixture.native.turnId, + itemId: fixture.native.itemId, + }, + controlPlaneRegistration: (authority: Parameters[0]["authority"]) => + registerRunnerPrpAuthority({ + companyId, + runId: fixture.runId, + authority, + }), + }; + } + + return { seedRun, transportOptions, get current() { return { temporary, runtimeRoot, paperclipHome, apiUrl, companyId, agentId, successor, server }; } }; +} + +export function remotePreparedWorkspace() { + return { schema: "paperclip.native-workspace-sync/v3", state: "prepared", + descriptorSha256: "a".repeat(64), baselineSha256: "b".repeat(64), finalHostSha256: null, + workspaceId: randomUUID(), leaseId: randomUUID(), providerLeaseId: randomUUID(), + remoteCwd: "/workspace", resourceDisposition: "destroy" }; +} diff --git a/server/src/__tests__/helpers/native-restart.ts b/server/src/__tests__/helpers/native-restart.ts index 8d642f4ca12..a648ea04d31 100644 --- a/server/src/__tests__/helpers/native-restart.ts +++ b/server/src/__tests__/helpers/native-restart.ts @@ -1,255 +1,5 @@ -import { recoveryPostgresSupport, startRecoveryPostgres } from "./recovery-postgres.js"; -import { randomUUID } from "node:crypto"; -import { existsSync } from "node:fs"; -import { mkdtemp, rm } from "node:fs/promises"; -import { createServer, type Server } from "node:http"; -import { tmpdir } from "node:os"; -import { resolve } from "node:path"; -import { eq } from "drizzle-orm"; -import { agents, companies, closeRegisteredClients, createDb, heartbeatRuns, issues, nativeRunFinalizations, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS } from "@paperclipai/db"; -import { afterAll, afterEach, beforeAll, describe, it } from "vitest"; -import { defaultCapabilityRunnerdBinary } from "../../vendor/paperclip-runner/index.js"; -import { startEmbeddedPostgresTestDatabase } from "./embedded-postgres.js"; -import { registerRunnerPrpAuthority, runnerPrpWebSocketInternals, setupRunnerPrpWebSocketServer } from "../../realtime/runner-prp-ws.js"; -import { readProcessStartedAt } from "../../services/hot-restart.js"; -import { prepareNativeHeartbeatRun } from "../../services/native-runtime/prepare-native-run.js"; -import { createIsolatedTestHome } from "./isolated-test-home.js"; -import { type NativeControllerIdentity } from "../../services/native-runtime/native-restart-recovery.js"; -import { waitForPendingRunFailureReports } from "../../services/run-failure-report.js"; -import { registerLocalRuntimeApiServer } from "../../runtime-api.js"; +import { describe } from "vitest"; +import { recoveryPostgresSupport } from "./recovery-postgres.js"; +export * from "./native-restart-fixture.js"; -// An explicitly supplied disposable database also permits qualification on hosts -// where native Postgres cannot allocate semaphores. Never fall back silently. export const describeEmbeddedPostgres: typeof describe.skip = recoveryPostgresSupport.supported ? describe : describe.skip; - -export const fakeCodexAppServer = resolve( - import.meta.dirname, - "../../../../packages/paperclip-runner/runner/target/debug/fake-codex-app-server", -); -const binariesAvailable = - existsSync(defaultCapabilityRunnerdBinary()) && existsSync(fakeCodexAppServer); -export const realProcessIt: typeof it.skip = binariesAvailable ? it : it.skip; - -async function closeServer(server: Server | null): Promise { - if (!server) return; - server.closeAllConnections(); - if (!server.listening) return; - await new Promise((resolveClose) => server.close(() => resolveClose())); -} - -export function processAlive(pid: number | null): boolean { - if (!pid) return false; - try { - process.kill(pid, 0); - return true; - } catch { - return false; - } -} - -export async function waitForCondition( - description: string, - condition: () => boolean | Promise, - timeoutMs = 10_000, -): Promise { - const deadline = Date.now() + timeoutMs; - while (Date.now() < deadline) { - if (await condition()) return; - await new Promise((resolveWait) => setTimeout(resolveWait, 25)); - } - throw new Error(`Timed out waiting for ${description}`); -} - -export async function stopOwnedProcessGroup( - pid: number | null, - expectedStateRoot: string, -): Promise { - if (!pid || !processAlive(pid)) return; - const command = await import("node:child_process").then( - ({ execFileSync }) => - execFileSync("ps", ["-p", String(pid), "-o", "command="], { - encoding: "utf8", - }).trim(), - ); - if (!command.includes(expectedStateRoot)) { - throw new Error(`Refusing to signal process ${pid}; ownership changed`); - } - if (process.platform !== "win32") process.kill(-pid, "SIGKILL"); - else process.kill(pid, "SIGKILL"); - await waitForCondition(`owned process ${pid} to exit`, () => !processAlive(pid)); -} - -export function nativeRestartFixture() { - let temporary: Awaited>; - let runtimeRoot: string; - let paperclipHome: string; - let server: Server | null = null; - let apiUrl: string; - const originalPaperclipHome = process.env.PAPERCLIP_HOME; - let isolatedHome: ReturnType; - - const companyId = randomUUID(); - const agentId = randomUUID(); - let successor: NativeControllerIdentity; - - beforeAll(async () => { - temporary = await startRecoveryPostgres("native-runner-restart-recovery-"); - runtimeRoot = await mkdtemp(resolve(tmpdir(), "native-restart-runtime-")); - isolatedHome = createIsolatedTestHome(); - paperclipHome = isolatedHome.home; - process.env.PAPERCLIP_HOME = paperclipHome; - const controllerStartedAt = await readProcessStartedAt(process.pid); - successor = { - bootId: randomUUID(), - pid: process.pid, - processStartedAt: controllerStartedAt - ? new Date(controllerStartedAt) - : new Date(), - }; - server = createServer(); - registerLocalRuntimeApiServer(server); - await new Promise((resolveListen) => - server!.listen(0, "127.0.0.1", resolveListen), - ); - const address = server.address(); - if (!address || typeof address === "string") { - throw new Error("Expected restart recovery TCP listener"); - } - apiUrl = `http://127.0.0.1:${address.port}`; - setupRunnerPrpWebSocketServer(server, { apiUrl }); - - const db = createDb(temporary.connectionString); - await db.insert(companies).values({ - id: companyId, - name: "Native restart recovery", - issuePrefix: "NRR", - requireBoardApprovalForNewAgents: false, - }); - await db.insert(agents).values({ - id: agentId, - companyId, - name: "Native restart runner", - role: "engineer", - status: "active", - adapterType: "paperclip_runner", - adapterConfig: { provider: "codex" }, - runtimeConfig: {}, - permissions: {}, - }); - }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS); - - afterEach(async () => { - // Each case creates its own clients. With the long waits in other files, - // idle reaping no longer keeps their total below Postgres's connection cap. - await waitForPendingRunFailureReports(); - if (temporary) await closeRegisteredClients(temporary.connectionString, { scope: "database" }); - }); - - afterAll(async () => { - runnerPrpWebSocketInternals.resetForTests(); - await closeServer(server); - await temporary?.cleanup(); - if (runtimeRoot) await rm(runtimeRoot, { recursive: true, force: true }); - isolatedHome?.cleanup(); - if (originalPaperclipHome === undefined) delete process.env.PAPERCLIP_HOME; - else process.env.PAPERCLIP_HOME = originalPaperclipHome; - }); - - async function seedRun( - label: string, - existingDb?: ReturnType, - ) { - const db = existingDb ?? createDb(temporary.connectionString); - const issueId = randomUUID(); - const runId = randomUUID(); - await db.insert(issues).values({ - id: issueId, - companyId, - identifier: `NRR-${label}`, - title: `Recover ${label}`, - status: "in_progress", - priority: "medium", - workMode: "standard", - assigneeAgentId: agentId, - }); - const [run] = await db - .insert(heartbeatRuns) - .values({ - id: runId, - companyId, - agentId, - status: "running", - runtimeMode: "native", - nativeIssueId: issueId, - invocationSource: "assignment", - triggerDetail: "system", - contextSnapshot: { issueId }, - }) - .returning(); - if (!run) throw new Error("Failed to seed native restart run"); - await db - .update(issues) - .set({ executionRunId: runId }) - .where(eq(issues.id, issueId)); - const native = await prepareNativeHeartbeatRun({ - db, - run, - issue: { - id: issueId, - title: `Recover ${label}`, - description: null, - reviewPolicy: null, - }, - environmentLeaseId: `lease-${label}`, - }); - await db.insert(nativeRunFinalizations).values({ - runId, - companyId, - issueId, - phase: "observed", - }); - const [preparedRun] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)); - return { db, issueId, runId, native, preparedProfile: preparedRun.runnerProfileJson }; - } - - function transportOptions( - fixture: Awaited>, - stateDirectory: string, - ) { - return { - runnerBinary: defaultCapabilityRunnerdBinary(), - codexCommand: fakeCodexAppServer, - codexArgs: [ - "--state-file", - resolve(stateDirectory, "fake-codex-state.json"), - "--call-log", - resolve(stateDirectory, "fake-codex-calls.log"), - ], - stateDirectory, - lifecyclePolicy: { mode: "warm" as const, idleTimeoutMs: 60_000 }, - prpIdentity: { - runnerInstanceId: fixture.native.runnerInstanceId, - environmentLeaseId: fixture.native.environmentLeaseId, - runId: fixture.runId, - normalizedSessionId: fixture.native.normalizedSessionId, - turnId: fixture.native.turnId, - itemId: fixture.native.itemId, - }, - controlPlaneRegistration: (authority: Parameters[0]["authority"]) => - registerRunnerPrpAuthority({ - companyId, - runId: fixture.runId, - authority, - }), - }; - } - - return { seedRun, transportOptions, get current() { return { temporary, runtimeRoot, paperclipHome, apiUrl, companyId, agentId, successor, server }; } }; -} - -export function remotePreparedWorkspace() { - return { schema: "paperclip.native-workspace-sync/v3", state: "prepared", - descriptorSha256: "a".repeat(64), baselineSha256: "b".repeat(64), finalHostSha256: null, - workspaceId: randomUUID(), leaseId: randomUUID(), providerLeaseId: randomUUID(), - remoteCwd: "/workspace", resourceDisposition: "destroy" }; -} diff --git a/server/src/__tests__/helpers/recovery-postgres-fixture.ts b/server/src/__tests__/helpers/recovery-postgres-fixture.ts new file mode 100644 index 00000000000..9695dc37be9 --- /dev/null +++ b/server/src/__tests__/helpers/recovery-postgres-fixture.ts @@ -0,0 +1,43 @@ +import { resolveExternalTestDatabaseUrl } from "./external-test-database.js"; +import { randomUUID } from "node:crypto"; +import { applyPendingMigrations, closeRegisteredClients, createDb } from "@paperclipai/db"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase, type EmbeddedPostgresTestSupport } from "./embedded-postgres.js"; + +type RecoveryPostgresOptions = { dedicatedDatabaseUrl?: string }; + +export async function getRecoveryPostgresSupport(options: RecoveryPostgresOptions = {}): Promise { + if (options.dedicatedDatabaseUrl !== undefined) { + if (!resolveExternalTestDatabaseUrl("dedicated recovery database", options.dedicatedDatabaseUrl)) { + throw new Error("Dedicated recovery database URL is required"); + } + return { supported: true }; + } + return process.env.PAPERCLIP_TEST_DATABASE_URL ? { supported: true } : getEmbeddedPostgresTestSupport(); +} + +export async function startRecoveryPostgres(prefix: string, options: RecoveryPostgresOptions = {}) { + const dedicated = options.dedicatedDatabaseUrl; + if (dedicated !== undefined) { + const connectionString = resolveExternalTestDatabaseUrl("dedicated recovery database", dedicated); + if (!connectionString) throw new Error("Dedicated recovery database URL is required"); + await applyPendingMigrations(connectionString); + return { connectionString, cleanup: () => closeRegisteredClients(connectionString, { scope: "database" }) }; + } + const external = process.env.PAPERCLIP_TEST_DATABASE_URL; + if (!external) return startEmbeddedPostgresTestDatabase(prefix); + const databaseName = `recovery_${randomUUID().replaceAll("-", "")}`; + const admin = createDb(external); + try { await admin.$client.unsafe(`create database "${databaseName}"`); } + finally { await admin.$client.end(); } + const url = new URL(external); url.pathname = `/${databaseName}`; + const connectionString = url.toString(); + const cleanup = async () => { + await closeRegisteredClients(connectionString, { scope: "database" }); + const owner = createDb(external); + try { await owner.$client.unsafe(`drop database "${databaseName}"`); } + finally { await owner.$client.end(); } + }; + try { await applyPendingMigrations(connectionString); } + catch (error) { await cleanup().catch(() => undefined); throw error; } + return { connectionString, cleanup }; +} diff --git a/server/src/__tests__/helpers/recovery-postgres.ts b/server/src/__tests__/helpers/recovery-postgres.ts index 696aab4c823..5e9f2cd0173 100644 --- a/server/src/__tests__/helpers/recovery-postgres.ts +++ b/server/src/__tests__/helpers/recovery-postgres.ts @@ -1,27 +1,4 @@ -import { randomUUID } from "node:crypto"; -import { applyPendingMigrations, closeRegisteredClients, createDb } from "@paperclipai/db"; -import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./embedded-postgres.js"; +import { getRecoveryPostgresSupport } from "./recovery-postgres-fixture.js"; +export { getRecoveryPostgresSupport, startRecoveryPostgres } from "./recovery-postgres-fixture.js"; -export const recoveryPostgresSupport = process.env.PAPERCLIP_TEST_DATABASE_URL - ? { supported: true } : await getEmbeddedPostgresTestSupport(); - -/** Optional external host, always a fresh owned database; never reuse its data. */ -export async function startRecoveryPostgres(prefix: string) { - const external = process.env.PAPERCLIP_TEST_DATABASE_URL; - if (!external) return startEmbeddedPostgresTestDatabase(prefix); - const databaseName = `recovery_${randomUUID().replaceAll("-", "")}`; - const admin = createDb(external); - try { await admin.$client.unsafe(`create database "${databaseName}"`); } - finally { await admin.$client.end(); } - const url = new URL(external); url.pathname = `/${databaseName}`; - const connectionString = url.toString(); - const cleanup = async () => { - await closeRegisteredClients(connectionString, { scope: "database" }); - const owner = createDb(external); - try { await owner.$client.unsafe(`drop database "${databaseName}"`); } - finally { await owner.$client.end(); } - }; - try { await applyPendingMigrations(connectionString); } - catch (error) { await cleanup().catch(() => undefined); throw error; } - return { connectionString, cleanup }; -} +export const recoveryPostgresSupport = await getRecoveryPostgresSupport(); diff --git a/server/src/__tests__/idle-completed-history.test.ts b/server/src/__tests__/idle-completed-history.test.ts index a24c9719086..89e221d89b0 100644 --- a/server/src/__tests__/idle-completed-history.test.ts +++ b/server/src/__tests__/idle-completed-history.test.ts @@ -3,7 +3,7 @@ import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import path from "node:path"; import net from "node:net"; -import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; import { eq, sql } from "drizzle-orm"; import * as tables from "@paperclipai/db"; import { isRuntimeExposureAppPort } from "@paperclipai/shared"; @@ -30,8 +30,9 @@ const DELETE_METHOD = "DELETE" as never; let db: ReturnType; const now = new Date(); const ownerId = randomUUID(); + let holdExpiresAt: Date; const hold = () => ({ ownerId, draining: true, startedAt: new Date(now.getTime() - 1000), - expiresAt: new Date(Date.now() + 60_000), activeRuns: 0, pendingWakes: 0 }); + expiresAt: holdExpiresAt, activeRuns: 0, pendingWakes: 0 }); const read = () => readIdleSleepSafety(db, hold, Date.now, ownerId, async () => "none"); const blocked = async (reason: string) => { const check = completedHistoryChecks.find(check => check.reason === reason); @@ -47,6 +48,7 @@ const DELETE_METHOD = "DELETE" as never; const [row] = await db.execute<{ blocked: boolean }>(sql`SELECT EXISTS (${sql.raw(query)} AND a.kind = ${kind}) AS blocked`); return row.blocked; }; + beforeEach(() => { holdExpiresAt = new Date(Date.now() + 60_000); }); beforeAll(async () => { if (!externalDatabaseUrl) database = await startEmbeddedPostgresTestDatabase("idle-history-"); db = tables.createDb(externalDatabaseUrl ?? database!.connectionString, { maxConnections: 4 }); diff --git a/server/src/__tests__/idle-sleep-preparation.test.ts b/server/src/__tests__/idle-sleep-preparation.test.ts new file mode 100644 index 00000000000..29332846097 --- /dev/null +++ b/server/src/__tests__/idle-sleep-preparation.test.ts @@ -0,0 +1,125 @@ +import { beginIdleTrackedWork, idleWorkSnapshot } from "../services/task-admission.js"; +import type { Db } from "@paperclipai/db"; +import { describe, expect, it, vi } from "vitest"; +import { readIdleSleepSafety, type IdleSleepDrainStatus } from "../services/idle-sleep-safety.js"; + +const now = Date.UTC(2020, 0, 1); +const ownerId = "d0b833f4-4098-42de-8420-1907f3aa4895"; +const owned = (): IdleSleepDrainStatus => ({ draining: true, ownerId, startedAt: new Date(now - 1000), + expiresAt: new Date(now + 60_000), activeRuns: 0, pendingWakes: 0 }); +const unknown = { version: 1, backgroundWork: "unknown" }; +const none = { version: 1, backgroundWork: "none" }; +const emptyLocal = async () => "none" as const; + +describe("explicit idle resource preparation", () => { + const database = () => { + const transaction = vi.fn(async (run: (tx: unknown) => Promise) => run({ execute: async () => [{ blocked: false }] })); + return { db: { transaction } as unknown as Db, transaction }; + }; + const inspect = (prepareLocalWork: NonNullable[9]>["prepareLocalWork"], + getStatus = owned, clock = () => now, local = emptyLocal) => + readIdleSleepSafety(database().db, getStatus, clock, ownerId, local, undefined, undefined, false, false, { prepareLocalWork }); + + it("awaits tracked retirement before capturing the fresh durable and local proof", async () => { + const { db, transaction } = database(); + let warm = true; + const prepare = vi.fn(async ({ canPrepare }: { canPrepare: () => boolean }) => { + expect(canPrepare()).toBe(true); + expect(idleWorkSnapshot().active).toBe(1); + expect(transaction).not.toHaveBeenCalled(); + warm = false; + return "none" as const; + }); + const local = vi.fn(async () => warm ? "present" as const : "none" as const); + expect(await readIdleSleepSafety(db, owned, () => now, ownerId, local, undefined, undefined, false, false, + { prepareLocalWork: prepare })).toEqual(none); + expect(prepare).toHaveBeenCalledOnce(); + expect(transaction).toHaveBeenCalledOnce(); + expect(local).toHaveBeenCalledTimes(2); + expect(idleWorkSnapshot().active).toBe(0); + }); + + it.each([undefined, "stale-owner"])("does not retire resources for requested owner %s", async requestedOwner => { + const { db, transaction } = database(); + const prepare = vi.fn(); + expect(await readIdleSleepSafety(db, owned, () => now, requestedOwner, emptyLocal, undefined, undefined, false, false, + { prepareLocalWork: prepare })).toEqual(unknown); + expect(prepare).not.toHaveBeenCalled(); + expect(transaction).not.toHaveBeenCalled(); + }); + + it("does not retire resources while other admitted work is active", async () => { + const done = beginIdleTrackedWork(); + const prepare = vi.fn(); + try { expect(await inspect(prepare)).toEqual(unknown); } + finally { done(); } + expect(prepare).not.toHaveBeenCalled(); + }); + + it.each(["present", "unknown"] as const)("retains preparation result %s without a durable scan", async state => { + const { db, transaction } = database(); + expect(await readIdleSleepSafety(db, owned, () => now, ownerId, emptyLocal, undefined, undefined, false, false, + { prepareLocalWork: async () => state })).toEqual({ version: 1, backgroundWork: state }); + expect(transaction).not.toHaveBeenCalled(); + expect(idleWorkSnapshot().active).toBe(0); + }); + + it.each(["expired", "replaced", "renewed", "concurrent", "threw"])("rejects a preparation whose boundary is %s", async kind => { + let clock = now; + let status = owned(); + expect(await inspect(async ({ canPrepare }) => { + expect(canPrepare()).toBe(true); + if (kind === "expired") clock += 60_000; + if (kind === "replaced") status = { ...status, ownerId: "replacement" }; + if (kind === "renewed") status = { ...status, expiresAt: new Date(now + 120_000) }; + if (kind === "concurrent") { const finish = beginIdleTrackedWork(); finish(); } + if (kind === "threw") throw new Error("private provider detail"); + expect(canPrepare()).toBe(false); + return "none"; + }, () => status, () => clock)).toEqual(unknown); + expect(idleWorkSnapshot().active).toBe(0); + }); + + it("keeps preparation tracked after expiry until the real owner promise settles", async () => { + let settle!: () => void; + const pending = new Promise(resolve => { settle = resolve; }); + let clock = now; + const prepare = vi.fn(async () => { await pending; return "none" as const; }); + const scan = inspect(prepare, owned, () => clock); + await vi.waitFor(() => expect(prepare).toHaveBeenCalledOnce()); + clock += 60_000; + expect(idleWorkSnapshot().active).toBe(1); + expect(await inspect(vi.fn())).toEqual(unknown); + settle(); + expect(await scan).toEqual(unknown); + expect(idleWorkSnapshot().active).toBe(0); + }); + + it.each(["local renewal", "local unbounded", "transaction renewal", "transaction unbounded"])( + "rejects an expiry replacement after preparation during %s inspection", async kind => { + const { db, transaction } = database(); + let status = owned(); + const replaceExpiry = () => { status = { ...status, + expiresAt: kind.endsWith("unbounded") ? null : new Date(now + 120_000) }; }; + if (kind.startsWith("transaction")) transaction.mockImplementationOnce(async run => { + const result = await run({ execute: async () => [{ blocked: false }] }); + replaceExpiry(); + return result; + }); + const local = async () => { + if (kind.startsWith("local")) replaceExpiry(); + return "none" as const; + }; + expect(await readIdleSleepSafety(db, () => status, () => now, ownerId, local, undefined, undefined, false, false, + { prepareLocalWork: async () => "none" })).toEqual(unknown); + expect(idleWorkSnapshot().active).toBe(0); + }, + ); + + it("leaves diagnostic inspection free of resource retirement", async () => { + const prepare = vi.fn(); + await readIdleSleepSafety(database().db, owned, () => now, ownerId, emptyLocal, undefined, undefined, false, false, + { prepareLocalWork: prepare, collectDiagnostics: true }); + expect(prepare).not.toHaveBeenCalled(); + }); +}); diff --git a/server/src/__tests__/instance-settings-routes.test.ts b/server/src/__tests__/instance-settings-routes.test.ts index f0705810ebd..cf4803eda84 100644 --- a/server/src/__tests__/instance-settings-routes.test.ts +++ b/server/src/__tests__/instance-settings-routes.test.ts @@ -20,6 +20,7 @@ const mockHeartbeatService = vi.hoisted(() => ({ applyTaskDrain: vi.fn(), stopTaskDrain: vi.fn(), getTaskDrainStatus: vi.fn(), + prepareIdleWarmNativeSessionsForSleep: vi.fn(), readUpgradeReadiness: vi.fn(), resumeQueuedRunsAfterDrain: vi.fn(), })); @@ -1079,7 +1080,7 @@ describe("instance settings routes", () => { await request(createApp(adminActor)).post("/api/instance/task-drain").send({ purpose: "idle", ttlMs: 60_000 }).expect(200); expect(mockHeartbeatService.computeTaskDrain).toHaveBeenCalledWith({ purpose: "idle", ttlMs: 60_000 }); await request(createApp(adminActor)).get(`/api/instance/task-drain?idleSleepSafety=1&ownerId=${ownerId}`).expect(200); - expect(mockReadIdleSleepSafety).toHaveBeenLastCalledWith(mockDb, expect.any(Function), Date.now, ownerId, expect.any(Function), undefined, undefined, false, false, { incomingWake: undefined, deadlineCapability: undefined }); + expect(mockReadIdleSleepSafety).toHaveBeenLastCalledWith(mockDb, expect.any(Function), Date.now, ownerId, expect.any(Function), undefined, undefined, false, false, { incomingWake: undefined, deadlineCapability: undefined, prepareLocalWork: expect.any(Function) }); }); it.each([{}, { ttlMs: null }, { ttlMs: 1 }, { ttlMs: 300_001 }])("rejects an unbounded or invalid idle TTL: %j", async body => { @@ -1106,10 +1107,25 @@ describe("instance settings routes", () => { const res = await request(createApp(adminActor)).get("/api/instance/task-drain?idleSleepSafety=1"); expect(res.status).toBe(200); expect(res.body).toEqual({ ...idleStatus, idleSleepSafety: report }); - expect(mockReadIdleSleepSafety).toHaveBeenCalledWith(mockDb, expect.any(Function), Date.now, undefined, expect.any(Function), undefined, undefined, false, false, { incomingWake: undefined, deadlineCapability: undefined }); + expect(mockReadIdleSleepSafety).toHaveBeenCalledWith(mockDb, expect.any(Function), Date.now, undefined, expect.any(Function), undefined, undefined, false, false, { incomingWake: undefined, deadlineCapability: undefined, prepareLocalWork: expect.any(Function) }); expect(mockReadIdleSleepSafety.mock.calls[0][1]()).toEqual(idleStatus); }); + it.each([ + [{ closed: 1, busy: 0, failed: 0 }, "none"], + [{ closed: 0, busy: 1, failed: 0 }, "present"], + [{ closed: 0, busy: 0, failed: 1 }, "present"], + ] as const)("binds native retirement to the safety owner's current-admission predicate: %j", async (result, state) => { + const canPrepare = () => true; + mockHeartbeatService.getTaskDrainStatus.mockReturnValue(idleStatus); + mockHeartbeatService.prepareIdleWarmNativeSessionsForSleep.mockResolvedValue(result); + mockReadIdleSleepSafety.mockImplementation(async (_db, _status, _now, _owner, _local, _plugins, _backup, _routines, _monitors, options) => + ({ version: 1, backgroundWork: await options.prepareLocalWork({ canPrepare }) })); + const response = await request(createApp(adminActor)).get("/api/instance/task-drain?idleSleepSafety=1&ownerId=fixture-owner").expect(200); + expect(mockHeartbeatService.prepareIdleWarmNativeSessionsForSleep).toHaveBeenCalledWith({ canRetire: canPrepare }); + expect(response.body.idleSleepSafety.backgroundWork).toBe(state); + }); + it("collects fixed-key diagnostics only through the privileged safety report", async () => { mockHeartbeatService.getTaskDrainStatus.mockReturnValue(idleStatus); mockReadIdleSleepSafety.mockResolvedValue({ version: 1, backgroundWork: "present", @@ -1120,7 +1136,7 @@ describe("instance settings routes", () => { .get("/api/instance/task-drain?idleSleepSafety=1&idleSleepDiagnostics=1&ownerId=fixture-owner").expect(200); expect(response.body.idleSleepSafety.diagnostics.checks).toEqual([{ reason: "agent_api_keys", state: "present" }]); expect(mockReadIdleSleepSafety).toHaveBeenCalledWith(mockDb, expect.any(Function), Date.now, - "fixture-owner", expect.any(Function), undefined, undefined, false, false, { incomingWake: undefined, deadlineCapability: undefined, collectDiagnostics: true }); + "fixture-owner", expect.any(Function), undefined, undefined, false, false, { incomingWake: undefined, deadlineCapability: undefined, prepareLocalWork: expect.any(Function), collectDiagnostics: true }); mockReadIdleSleepSafety.mockClear(); for (const actor of [nonAdminActor, { type: "agent", agentId: "agent-1", companyId: "company-1", source: "api_key" }]) { await request(createApp(actor)) @@ -1142,7 +1158,7 @@ describe("instance settings routes", () => { await request(createApp({ ...adminActor, source: "cloud_control" })) .get(`/api/instance/task-drain?idleSleepSafety=1&scheduledWake=${version}&ownerId=fixture-owner`).expect(200); expect(mockReadIdleSleepSafety).toHaveBeenCalledWith(mockDb, expect.any(Function), Date.now, - "fixture-owner", expect.any(Function), undefined, undefined, true, version === "2", { incomingWake: undefined, deadlineCapability: undefined }); + "fixture-owner", expect.any(Function), undefined, undefined, true, version === "2", { incomingWake: undefined, deadlineCapability: undefined, prepareLocalWork: expect.any(Function) }); }); describe("signed incoming capability handoff", () => { @@ -1183,7 +1199,7 @@ describe("instance settings routes", () => { expect(response.body.idleSleepSafety).toEqual({ version: 3, backgroundWork: "none" }); expect(mockReadIdleSleepSafety).toHaveBeenCalledExactlyOnceWith(mockDb, expect.any(Function), Date.now, ownerId, expect.any(Function), undefined, undefined, false, false, - { incomingWake: catalog, deadlineCapability: undefined }); + { incomingWake: catalog, deadlineCapability: undefined, prepareLocalWork: expect.any(Function) }); const options = mockReadIdleSleepSafety.mock.calls[0][9]; expect(options.deadlineCapability).toBeUndefined(); expect(Object.isFrozen(options.incomingWake)).toBe(true); @@ -1198,7 +1214,7 @@ describe("instance settings routes", () => { .set("x-paperclip-cloud-control", assertion()).expect(200); expect(mockReadIdleSleepSafety).toHaveBeenCalledExactlyOnceWith(mockDb, expect.any(Function), Date.now, ownerId, expect.any(Function), undefined, undefined, true, true, - { incomingWake: catalog, deadlineCapability: catalog }); + { incomingWake: catalog, deadlineCapability: catalog, prepareLocalWork: expect.any(Function) }); const options = mockReadIdleSleepSafety.mock.calls[0][9]; expect(options.incomingWake).toBe(options.deadlineCapability); expect(Object.isFrozen(options.deadlineCapability)).toBe(true); @@ -1226,7 +1242,7 @@ describe("instance settings routes", () => { it("leaves legacy signed reads without a catalog conservative", async () => { await request(createControlApp({ type: "none" })).get(path) .set("x-paperclip-cloud-control", assertion({ idleCapabilities: undefined })).expect(200); - expect(mockReadIdleSleepSafety.mock.calls[0][9]).toEqual({ incomingWake: undefined, deadlineCapability: undefined }); + expect(mockReadIdleSleepSafety.mock.calls[0][9]).toEqual({ incomingWake: undefined, deadlineCapability: undefined, prepareLocalWork: expect.any(Function) }); }); it("cannot install capabilities from tenant, actor, query or body values", async () => { @@ -1234,7 +1250,7 @@ describe("instance settings routes", () => { const actor = { ...adminActor, source, idleCapabilities: catalog, incomingWake: catalog }; await request(createControlApp(actor)).get(`${path}&incomingWake=${encodeURIComponent(JSON.stringify(catalog))}`) .send({ idleCapabilities: catalog }).expect(200); - expect(mockReadIdleSleepSafety.mock.lastCall?.[9]).toEqual({ incomingWake: undefined, deadlineCapability: undefined }); + expect(mockReadIdleSleepSafety.mock.lastCall?.[9]).toEqual({ incomingWake: undefined, deadlineCapability: undefined, prepareLocalWork: expect.any(Function) }); } mockReadIdleSleepSafety.mockClear(); await request(createControlApp(nonAdminActor)).get(path).send({ idleCapabilities: catalog }).expect(403); @@ -1274,7 +1290,7 @@ describe("instance settings routes", () => { expect(mockReadIdleSleepSafety).not.toHaveBeenCalled(); await request(app).get("/api/instance/task-drain?idleSleepSafety=1&ownerId=fixture-owner").expect(200); expect(mockReadIdleSleepSafety).toHaveBeenCalledWith(mockDb, expect.any(Function), Date.now, - "fixture-owner", expect.any(Function), undefined, backup, false, false, { incomingWake: undefined, deadlineCapability: undefined }); + "fixture-owner", expect.any(Function), undefined, backup, false, false, { incomingWake: undefined, deadlineCapability: undefined, prepareLocalWork: expect.any(Function) }); mockReadIdleSleepSafety.mockClear(); await request(createApp(nonAdminActor, undefined, backup)).get("/api/instance/task-drain?idleSleepSafety=1").expect(403); expect(mockReadIdleSleepSafety).not.toHaveBeenCalled(); @@ -1296,7 +1312,7 @@ describe("instance settings routes", () => { const control = createApp({ ...adminActor, source: "cloud_control" }, undefined, backup); await request(control).get("/api/instance/task-drain?idleSleepSafety=1&ownerId=fixture-owner").expect(200); expect(mockReadIdleSleepSafety).toHaveBeenCalledWith(mockDb, expect.any(Function), Date.now, - "fixture-owner", expect.any(Function), undefined, backup, false, false, { incomingWake: undefined, deadlineCapability: undefined }); + "fixture-owner", expect.any(Function), undefined, backup, false, false, { incomingWake: undefined, deadlineCapability: undefined, prepareLocalWork: expect.any(Function) }); } finally { delete process.env.PAPERCLIP_CLOUD_TENANT_SERVER_TOKEN; } diff --git a/server/src/__tests__/plugin-idle-manager.test.ts b/server/src/__tests__/plugin-idle-manager.test.ts index c32c5133ff3..abeaec2b1ea 100644 --- a/server/src/__tests__/plugin-idle-manager.test.ts +++ b/server/src/__tests__/plugin-idle-manager.test.ts @@ -1,4 +1,7 @@ import { pluginIdleManifestGeneration } from "../services/plugin-idle-catalog.js"; +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; import { fileURLToPath } from "node:url"; import { createRequire } from "node:module"; import { setTimeout as delay } from "node:timers/promises"; @@ -49,6 +52,51 @@ describe("plugin manager idle receipts", () => { } finally { stopTaskDrain(); await manager.stopAll(); } }); + it("drains a supported worker and persists its next useful operation under the same manifest", async () => { + const directory = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-plugin-idle-")); + const stateFile = path.join(directory, "writes.json"); + const writes: unknown[] = []; + const manager = createPluginWorkerManager(); + try { + const handle = await manager.startWorker("test.idle", { + entrypointPath: fileURLToPath(new URL("./fixtures/plugin-worker-idle.cjs", import.meta.url)), + execArgv: ["--import", createRequire(import.meta.url).resolve("tsx")], + manifest, config: {}, instanceInfo: { instanceId: "fixture", hostVersion: "1.0.0" }, apiVersion: 1, + env: { IDLE_TEST_WRITE: "1" }, + hostHandlers: { + "events.subscribe": async () => undefined, + "state.set": async (params) => { + writes.push(params); + await fs.writeFile(stateFile, JSON.stringify(writes)); + }, + }, + }); + expect(await manager.call("test.idle", "health", {})).toMatchObject({ status: "ok" }); + const firstWrites = JSON.parse(await fs.readFile(stateFile, "utf8")); + expect(firstWrites).toHaveLength(1); + expect(firstWrites[0]).toMatchObject({ scopeKind: "instance", stateKey: "fixture", value: { value: 1 } }); + const first = await manager.inspectIdleSleep!(hold()); + expect(first.backgroundWork).toBe("none"); + expect(first.catalog).toEqual([{ pluginId: "test.idle", manifestGeneration: pluginIdleManifestGeneration(manifest) }]); + await expect(manager.call("test.idle", "health", {})).rejects.toThrow("held"); + expect(JSON.parse(await fs.readFile(stateFile, "utf8"))).toEqual(firstWrites); + stopTaskDrain(); + manager.releaseIdleSleep!(); + expect(await manager.call("test.idle", "health", {})).toMatchObject({ status: "ok" }); + expect(manager.getWorker("test.idle")).toBe(handle); + const resumedWrites = JSON.parse(await fs.readFile(stateFile, "utf8")); + expect(resumedWrites).toEqual([firstWrites[0], firstWrites[0]]); + const resumed = await manager.inspectIdleSleep!(hold()); + expect(resumed.backgroundWork).toBe("none"); + expect(resumed.catalog).toEqual(first.catalog); + expect(resumed.catalogGeneration).toBe(first.catalogGeneration); + } finally { + stopTaskDrain(); + await manager.stopAll(); + await fs.rm(directory, { recursive: true, force: true }); + } + }); + it("retains timed-out worker operations and admits work after the owned hold releases", async () => { const handle = worker(); await handle.start(); diff --git a/server/src/__tests__/recovery-postgres-mode.test.ts b/server/src/__tests__/recovery-postgres-mode.test.ts new file mode 100644 index 00000000000..c1c90c34574 --- /dev/null +++ b/server/src/__tests__/recovery-postgres-mode.test.ts @@ -0,0 +1,112 @@ +import { afterEach, beforeEach, expect, it, vi } from "vitest"; + +const mocks = vi.hoisted(() => ({ + migrations: vi.fn(async (_url: string) => {}), + closeRegistered: vi.fn(async (_url: string, _options: { scope: string }) => {}), + unsafe: vi.fn(async (_query: string) => []), + end: vi.fn(async () => {}), + createDb: vi.fn(), + support: vi.fn(async (): Promise<{ supported: true } | { supported: false; reason: string }> => ({ supported: true })), + embedded: vi.fn(), +})); +vi.mock("@paperclipai/db", () => ({ + applyPendingMigrations: mocks.migrations, closeRegisteredClients: mocks.closeRegistered, createDb: mocks.createDb, +})); +vi.mock("./helpers/embedded-postgres.js", () => ({ + getEmbeddedPostgresTestSupport: mocks.support, startEmbeddedPostgresTestDatabase: mocks.embedded, +})); + +const recoveryUrl = "postgresql://paperclip@127.0.0.1:55448/recovery_modes_test"; +beforeEach(() => { + vi.resetAllMocks(); + vi.resetModules(); + vi.stubEnv("PAPERCLIP_TEST_DATABASE_URL", undefined); + vi.stubEnv("PAPERCLIP_RECOVERY_TEST_DATABASE_URL", recoveryUrl); + mocks.createDb.mockReturnValue({ $client: { unsafe: mocks.unsafe, end: mocks.end } }); +}); +afterEach(() => { vi.unstubAllEnvs(); }); + +it("keeps separate fixture allocations when a recovery URL belongs to another explicitly opted-in caller", async () => { + const firstCleanup = vi.fn(async () => {}), secondCleanup = vi.fn(async () => {}); + mocks.embedded + .mockResolvedValueOnce({ connectionString: "postgresql://paperclip@127.0.0.1:55448/first_owned_test", cleanup: firstCleanup }) + .mockResolvedValueOnce({ connectionString: "postgresql://paperclip@127.0.0.1:55448/second_owned_test", cleanup: secondCleanup }); + const { startRecoveryPostgres } = await import("./helpers/recovery-postgres.js"); + const supportChecks = mocks.support.mock.calls.length; + const first = await startRecoveryPostgres("first-fixture-"); + const second = await startRecoveryPostgres("second-fixture-"); + expect(first.connectionString).not.toBe(second.connectionString); + expect(first.connectionString).not.toBe(recoveryUrl); + expect(second.connectionString).not.toBe(recoveryUrl); + expect(mocks.embedded.mock.calls).toEqual([["first-fixture-"], ["second-fixture-"]]); + expect(mocks.migrations).not.toHaveBeenCalled(); + expect(mocks.createDb).not.toHaveBeenCalled(); + expect(mocks.support).toHaveBeenCalledTimes(supportChecks); + await first.cleanup(); await second.cleanup(); + expect(firstCleanup).toHaveBeenCalledOnce(); expect(secondCleanup).toHaveBeenCalledOnce(); + expect(mocks.closeRegistered).not.toHaveBeenCalled(); +}); + +it("uses the exact dedicated database only when the owning caller passes its explicit option", async () => { + vi.stubEnv("PAPERCLIP_TEST_DATABASE_URL", "postgresql://paperclip@127.0.0.1:55448/postgres"); + const { startRecoveryPostgres } = await import("./helpers/recovery-postgres.js"); + const fixture = await startRecoveryPostgres("owned-fixture-", { dedicatedDatabaseUrl: recoveryUrl }); + expect(fixture.connectionString).toBe(recoveryUrl); + expect(mocks.migrations).toHaveBeenCalledExactlyOnceWith(recoveryUrl); + expect(mocks.embedded).not.toHaveBeenCalled(); expect(mocks.createDb).not.toHaveBeenCalled(); + expect(mocks.support).not.toHaveBeenCalled(); + await fixture.cleanup(); + expect(mocks.closeRegistered).toHaveBeenCalledExactlyOnceWith(recoveryUrl, { scope: "database" }); +}); + +it("retains fresh external database allocation and matching cleanup for existing non-dedicated callers", async () => { + const admin = "postgresql://paperclip@127.0.0.1:55448/postgres"; + vi.stubEnv("PAPERCLIP_TEST_DATABASE_URL", admin); + const { startRecoveryPostgres } = await import("./helpers/recovery-postgres.js"); + const first = await startRecoveryPostgres("first-fixture-"); + const second = await startRecoveryPostgres("second-fixture-"); + expect(first.connectionString).not.toBe(second.connectionString); + expect(first.connectionString).not.toBe(recoveryUrl); + expect(second.connectionString).not.toBe(recoveryUrl); + const firstName = new URL(first.connectionString).pathname.slice(1); + const secondName = new URL(second.connectionString).pathname.slice(1); + expect(mocks.unsafe.mock.calls.map(([query]) => query)).toEqual([ + `create database "${firstName}"`, `create database "${secondName}"`, + ]); + expect(mocks.migrations.mock.calls.map(([url]) => url)).toEqual([first.connectionString, second.connectionString]); + await first.cleanup(); await second.cleanup(); + expect(mocks.closeRegistered.mock.calls).toEqual([ + [first.connectionString, { scope: "database" }], [second.connectionString, { scope: "database" }], + ]); + expect(mocks.unsafe.mock.calls.map(([query]) => query)).toEqual([ + `create database "${firstName}"`, `create database "${secondName}"`, + `drop database "${firstName}"`, `drop database "${secondName}"`, + ]); + expect(mocks.createDb.mock.calls).toEqual([[admin], [admin], [admin], [admin]]); + expect(mocks.end).toHaveBeenCalledTimes(4); + expect(mocks.embedded).not.toHaveBeenCalled(); expect(mocks.support).not.toHaveBeenCalled(); +}); + +it("keeps default recovery support unavailable on an unsupported platform despite another owner's ambient dedicated URL", async () => { + mocks.support.mockResolvedValue({ supported: false, reason: "Unsupported synthetic platform" }); + const { recoveryPostgresSupport } = await import("./helpers/recovery-postgres.js"); + expect(recoveryPostgresSupport).toEqual({ supported: false, reason: "Unsupported synthetic platform" }); + expect(mocks.support).toHaveBeenCalledOnce(); + expect(mocks.embedded).not.toHaveBeenCalled(); expect(mocks.createDb).not.toHaveBeenCalled(); + expect(mocks.migrations).not.toHaveBeenCalled(); +}); + +it("supports the exact explicit dedicated mode without probing unsupported embedded PostgreSQL", async () => { + mocks.support.mockResolvedValue({ supported: false, reason: "Unsupported synthetic platform" }); + const { getRecoveryPostgresSupport, startRecoveryPostgres } = await import("./helpers/recovery-postgres-fixture.js"); + const selectedUrl = "postgresql://paperclip@127.0.0.1:55448/explicit_recovery_modes_test"; + const options = { dedicatedDatabaseUrl: selectedUrl }; + expect(await getRecoveryPostgresSupport(options)).toEqual({ supported: true }); + const fixture = await startRecoveryPostgres("explicit-dedicated-", options); + expect(fixture.connectionString).toBe(selectedUrl); + expect(mocks.migrations).toHaveBeenCalledExactlyOnceWith(selectedUrl); + expect(mocks.support).not.toHaveBeenCalled(); + expect(mocks.embedded).not.toHaveBeenCalled(); expect(mocks.createDb).not.toHaveBeenCalled(); + await fixture.cleanup(); + expect(mocks.closeRegistered).toHaveBeenCalledExactlyOnceWith(selectedUrl, { scope: "database" }); +}); diff --git a/server/src/__tests__/workspace-runtime-lifecycle.test.ts b/server/src/__tests__/workspace-runtime-lifecycle.test.ts new file mode 100644 index 00000000000..0f4e8cad268 --- /dev/null +++ b/server/src/__tests__/workspace-runtime-lifecycle.test.ts @@ -0,0 +1,112 @@ +import { randomUUID } from "node:crypto"; +import fs from "node:fs/promises"; +import net from "node:net"; +import os from "node:os"; +import path from "node:path"; +import { expect, it } from "vitest"; +import { + ensureRuntimeServicesForRun, + releaseRuntimeServicesForRun, + stopRuntimeServicesForExecutionWorkspace, + type RealizedExecutionWorkspace, +} from "../services/workspace-runtime.js"; +import { isPidAlive, listLocalServiceRegistryRecords } from "../services/local-service-supervisor.js"; + +async function expectListenerClosed(port: number) { + const error = await new Promise((resolve, reject) => { + const socket = net.connect({ host: "127.0.0.1", port }); + socket.setTimeout(1_000, () => { + socket.destroy(); + reject(new Error("Retired listener did not reject a connection")); + }); + socket.once("connect", () => { + socket.destroy(); + reject(new Error("Retired listener still accepts connections")); + }); + socket.once("error", resolve); + }); + expect(error.code).toBe("ECONNREFUSED"); +} + +it("retires an ephemeral runtime and preserves useful files for the same owner's next use", async () => { + const root = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-runtime-lifecycle-")); + const previousHome = process.env.PAPERCLIP_HOME; + const previousInstance = process.env.PAPERCLIP_INSTANCE_ID; + process.env.PAPERCLIP_HOME = path.join(root, "home"); + process.env.PAPERCLIP_INSTANCE_ID = randomUUID(); + const cwd = path.join(root, "workspace"); + await fs.mkdir(cwd); + await fs.writeFile(path.join(cwd, "retained.txt"), "useful workspace result"); + const script = path.join(cwd, "service.cjs"); + await fs.writeFile(script, ` +const fs = require("node:fs"); +const http = require("node:http"); +const count = fs.existsSync("starts.txt") ? Number(fs.readFileSync("starts.txt", "utf8")) + 1 : 1; +fs.writeFileSync("starts.txt", String(count)); +http.createServer((req, res) => { + res.setHeader("content-type", "application/json"); + res.end(JSON.stringify({ pid: process.pid, count, retained: fs.readFileSync("retained.txt", "utf8") })); +}).listen(Number(process.env.PORT), "127.0.0.1"); +`); + const workspace: RealizedExecutionWorkspace = { + baseCwd: cwd, cwd, source: "project_primary", projectId: randomUUID(), workspaceId: randomUUID(), + repoUrl: null, repoRef: "HEAD", strategy: "project_primary", branchName: null, + branchCreatedByRuntime: false, worktreePath: null, warnings: [], created: false, + }; + const input = { + agent: { id: randomUUID(), companyId: randomUUID(), name: "Lifecycle worker" }, + issue: null, + executionWorkspaceId: randomUUID(), + workspace, + adapterEnv: {}, + config: { workspaceRuntime: { services: [{ + name: "web", command: `${JSON.stringify(process.execPath)} ${JSON.stringify(script)}`, + port: { type: "auto" }, + readiness: { type: "http", urlTemplate: "http://127.0.0.1:{{port}}", timeoutSec: 10, intervalMs: 100 }, + expose: { type: "url", urlTemplate: "http://127.0.0.1:{{port}}" }, + lifecycle: "ephemeral", reuseScope: "execution_workspace", stopPolicy: { type: "on_run_finish" }, + }] } }, + }; + const runs = [randomUUID(), randomUUID()]; + try { + const [first] = await ensureRuntimeServicesForRun({ ...input, runId: runs[0]! }); + expect(first).toMatchObject({ reused: false, ownerAgentId: input.agent.id, companyId: input.agent.companyId }); + const firstResult = await (await fetch(first!.url!)).json() as { pid: number; count: number; retained: string }; + expect(firstResult).toMatchObject({ count: 1, retained: "useful workspace result" }); + expect(isPidAlive(firstResult.pid)).toBe(true); + const firstRecord = (await listLocalServiceRegistryRecords()).find(record => record.runtimeServiceId === first!.id); + expect(firstRecord).toBeDefined(); + await releaseRuntimeServicesForRun(runs[0]!); + expect(isPidAlive(firstResult.pid)).toBe(false); + expect(isPidAlive(firstRecord!.pid)).toBe(false); + await expectListenerClosed(first!.port!); + expect((await listLocalServiceRegistryRecords()).filter(record => record.runtimeServiceId === first!.id)).toEqual([]); + expect(await fs.readFile(path.join(cwd, "retained.txt"), "utf8")).toBe("useful workspace result"); + + const [second] = await ensureRuntimeServicesForRun({ ...input, runId: runs[1]! }); + expect(second).toMatchObject({ reused: false, ownerAgentId: input.agent.id, companyId: input.agent.companyId }); + expect(second!.id).not.toBe(first!.id); + const secondResult = await (await fetch(second!.url!)).json() as { pid: number; count: number; retained: string }; + expect(secondResult).toMatchObject({ count: 2, retained: "useful workspace result" }); + expect(isPidAlive(secondResult.pid)).toBe(true); + expect(await fs.readFile(path.join(cwd, "starts.txt"), "utf8")).toBe("2"); + await releaseRuntimeServicesForRun(runs[1]!); + expect(isPidAlive(secondResult.pid)).toBe(false); + await expectListenerClosed(second!.port!); + expect((await listLocalServiceRegistryRecords()).filter(record => record.cwd === cwd)).toEqual([]); + } finally { + try { + try { + for (const run of runs) await releaseRuntimeServicesForRun(run); + } finally { + await stopRuntimeServicesForExecutionWorkspace({ executionWorkspaceId: input.executionWorkspaceId, workspaceCwd: cwd }); + } + } finally { + if (previousHome === undefined) delete process.env.PAPERCLIP_HOME; + else process.env.PAPERCLIP_HOME = previousHome; + if (previousInstance === undefined) delete process.env.PAPERCLIP_INSTANCE_ID; + else process.env.PAPERCLIP_INSTANCE_ID = previousInstance; + await fs.rm(root, { recursive: true, force: true }); + } + } +}); diff --git a/server/src/routes/instance-settings.ts b/server/src/routes/instance-settings.ts index 1e0ce20ab6f..99f6572dc55 100644 --- a/server/src/routes/instance-settings.ts +++ b/server/src/routes/instance-settings.ts @@ -334,6 +334,10 @@ export function instanceSettingsRoutes(db: Db, pluginWorkers?: PluginWorkerManag deadlineCapability: domainDeadlines ? capabilities : undefined }), pluginWorkers?.inspectIdleSleep?.bind(pluginWorkers), prepareIdleDatabaseBackup, scheduledRoutines, scheduledIssueMonitors, { deadlineOwners, incomingWake: capabilities, deadlineCapability: domainDeadlines ? capabilities : undefined, + prepareLocalWork: async ({ canPrepare }) => { + const result = await heartbeat.prepareIdleWarmNativeSessionsForSleep({ canRetire: canPrepare }); + return result.busy || result.failed ? "present" : "none"; + }, ...(req.query.idleSleepDiagnostics === "1" ? { collectDiagnostics: true } : {}) }); res.json({ ...heartbeat.getTaskDrainStatus(), idleSleepSafety }); return; diff --git a/server/src/routes/openapi.ts b/server/src/routes/openapi.ts index 98d52aeff80..00b9cf811ad 100644 --- a/server/src/routes/openapi.ts +++ b/server/src/routes/openapi.ts @@ -7112,7 +7112,7 @@ registry.registerPath({ tags: ["instance"], summary: "Get process task-drain status, with optional bounded upgrade-readiness or idle-safety reports for instance operators", - description: "With idleSleepSafety=1, scheduledWake=domain-deadlines-v1 requires signed Cloud authority with idleCapabilities.deadlineCatalog=domain_deadlines_v1 and returns a mandatory deadlineProof even when no wake is scheduled. Cloud must acknowledge any returned durable alarm and revalidate the report before stopping the instance.", + description: "With idleSleepSafety=1, the exact current idle owner can retire idle native sessions before a fresh safety inspection. Busy or failed retirement prevents sleep. Native retirement is skipped for diagnostic inspection. scheduledWake=domain-deadlines-v1 requires signed Cloud authority with idleCapabilities.deadlineCatalog=domain_deadlines_v1 and returns a mandatory deadlineProof even when no wake is scheduled. Cloud must acknowledge any returned durable alarm and revalidate the report before stopping the instance.", request: { query: z.object({ upgradeReadiness: z.literal("1").optional().describe("Instance operators can request a bounded aggregate report of current remote-runner recovery eligibility. This read does not hold admission or detach runners."), ownerId: z.string().uuid().optional(), idleSleepSafety: z.literal("1").optional().describe("Instance admins can request a conservative durable-work report while admission and ingress are held. Unknown or present work must prevent automatic sleep."), scheduledWake: z.enum(["1", "2", "domain-deadlines-v1"]).optional().describe("Cloud control callers may request routine deadlines (1), routines and issue monitors (2), or the domain deadline catalog (domain-deadlines-v1). The domain catalog requires signed Cloud authority with idleCapabilities.deadlineCatalog=domain_deadlines_v1 and returns a deadlineProof even when no wake is scheduled. Cloud must acknowledge any returned durable alarm and revalidate this report before stopping the instance.") }) }, responses: { 200: r.ok(), 401: r.unauthorized, 403: r.forbidden }, }); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index dc73e163043..1d9d40bb4be 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -373,6 +373,7 @@ import { import { dispatchNativeSessionResumptions, + prepareIdleWarmNativeSessionsForSleep, NativeCancellationPendingRecoveryError, NativeControllerDetachedForRestartError, prepareNativeWorkspaceSync, @@ -4359,6 +4360,7 @@ export function heartbeatService( startTaskDrain, stopTaskDrain, getTaskDrainStatus, + prepareIdleWarmNativeSessionsForSleep, readUpgradeReadiness: () => readUpgradeReadiness(db, { activeRunIds: activeRunExecutions, counters: getTaskDrainStatus }), computeTaskDrain, applyTaskDrain, diff --git a/server/src/services/heartbeat/run-preparation.ts b/server/src/services/heartbeat/run-preparation.ts index 95f6ad89e4e..273ab370bd3 100644 --- a/server/src/services/heartbeat/run-preparation.ts +++ b/server/src/services/heartbeat/run-preparation.ts @@ -2115,7 +2115,7 @@ export async function attestReviewedExternalChatRun(input: { admittedIds.some((id, index) => id !== suppliedIds[index]) ) return { denied: "context_mismatch" }; - try { + const authorize = async (): Promise => { const answer = admittedContext.source === "issue.interaction.respond" ? await resolveExternalChatQuestionResponse( @@ -2145,7 +2145,11 @@ export async function attestReviewedExternalChatRun(input: { input.contextSnapshot[EXTERNAL_CHAT_QUESTION_RESPONSE_KEY] = answer.marker; return answer ? { answeredAtMs: answer.answeredAtMs } : true; + }; + try { + return await authorize(); } catch (error) { + let authorizationError = error; if ( error instanceof Error && error.message === "paperclip_runner_chat_attachment_binding_denied" @@ -2190,13 +2194,25 @@ export async function attestReviewedExternalChatRun(input: { ) { return "pending_delivery"; } + if ( + admittedIds.every((id) => + links.some((link) => link.commentId === id), + ) && + links.every((link) => link.state === "processed") + ) { + try { + return await authorize(); + } catch (retryError) { + authorizationError = retryError; + } + } } const denied = - error instanceof Error - ? CHAT_ATTACHMENT_DENIAL_REASONS.get(error.message) + authorizationError instanceof Error + ? CHAT_ATTACHMENT_DENIAL_REASONS.get(authorizationError.message) : undefined; if (denied) return { denied }; - throw error; + throw authorizationError; } }, ); diff --git a/server/src/services/idle-sleep-safety.ts b/server/src/services/idle-sleep-safety.ts index 3171632469a..3feba49b8e6 100644 --- a/server/src/services/idle-sleep-safety.ts +++ b/server/src/services/idle-sleep-safety.ts @@ -5,7 +5,7 @@ import type { IdlePluginCatalogEntry } from "./plugin-idle-catalog.js"; import type { ApiWakeAdmission } from "./idle-api-wake-admission.js"; import { readIdleIncomingWakeProof, incomingWakeWorkQuery, type IdleIncomingWakeProof } from "./idle-incoming-wake.js"; import type { VerifiedIdleCapabilities, IdleIncomingWakeCapability } from "./idle-incoming-wake-capability.js"; -import { idleWorkSnapshot } from "./task-admission.js"; +import { beginIdleTrackedWork, idleWorkSnapshot } from "./task-admission.js"; import { readIdleLocalWork, type IdleLocalWork } from "./idle-local-work.js"; import { readIdleRoutineSchedule, type ScheduledWake } from "./idle-scheduled-work.js"; import { combineIdleSchedules, readIdleIssueSchedule } from "./idle-issue-schedule.js"; @@ -280,7 +280,8 @@ function sameQuietHold(before: IdleSleepDrainStatus, after: IdleSleepDrainStatus return before.ownerId === after.ownerId && before.draining && after.draining && before.activeRuns === 0 && before.pendingWakes === 0 && after.activeRuns === 0 && after.pendingWakes === 0 && before.startedAt !== null && after.startedAt?.getTime() === before.startedAt.getTime() - && (after.expiresAt === null || after.expiresAt.getTime() > now); + && before.expiresAt !== null && before.expiresAt.getTime() > now + && after.expiresAt?.getTime() === before.expiresAt.getTime(); } /** Inventory persisted work after the task-drain admission hold lands. @@ -290,6 +291,7 @@ export interface IdleSleepSafetyOptions { incomingWake?: VerifiedIdleCapabilities; deadlineCapability?: VerifiedIdleCapabilities; collectDiagnostics?: true; + prepareLocalWork?: (input: { canPrepare: () => boolean }) => Promise; deadlineOwners?: IdleDeadlineOwners; } @@ -325,7 +327,7 @@ export async function readIdleSleepSafety( ...(diagnostics ? { diagnostics } : {}) }; }; const before = getDrainStatus(); - const localBefore = idleWorkSnapshot(); + let localBefore = idleWorkSnapshot(); const owned = () => Boolean(ownerId && ownerId === before.ownerId && before.expiresAt !== null); if (!sameQuietHold(before, before, now()) || (diagnostic && !owned())) { diagnostic?.record("inspection.hold", "unknown"); @@ -335,6 +337,27 @@ export async function readIdleSleepSafety( diagnostic?.record("inspection.hold", "none"); let failureReason = "local.inspection"; try { + if (options.prepareLocalWork && !diagnostic) { + if (!owned() || localBefore.active !== 0) return report("unknown"); + const finish = beginIdleTrackedWork(); + const preparation = idleWorkSnapshot(); + const canPrepare = () => { + const current = getDrainStatus(); + const work = idleWorkSnapshot(); + return sameQuietHold(before, current, now()) + && work.active === 1 && work.generation === preparation.generation; + }; + let prepared: IdleLocalWork; + let current: boolean; + try { + if (!canPrepare()) return report("unknown"); + prepared = await options.prepareLocalWork({ canPrepare }); + current = canPrepare(); + } finally { finish(); } + if (!current) return report("unknown"); + if (prepared !== "none") return report(prepared); + localBefore = idleWorkSnapshot(); + } const localFirst = await inspectLocalWork(diagnostic?.record); diagnostic?.record("local.inspection", localFirst); if (localFirst !== "none" && !diagnostic) return report(localFirst); diff --git a/server/src/services/native-runtime/native-idle-sleep-continuation.integration.test.ts b/server/src/services/native-runtime/native-idle-sleep-continuation.integration.test.ts new file mode 100644 index 00000000000..1e64b0bc626 --- /dev/null +++ b/server/src/services/native-runtime/native-idle-sleep-continuation.integration.test.ts @@ -0,0 +1,211 @@ +import { randomUUID } from "node:crypto"; +import { mkdir, readFile, writeFile } from "node:fs/promises"; +import { join } from "node:path"; +import express from "express"; +import request from "supertest"; +import { and, eq } from "drizzle-orm"; +import { agents, agentTaskSessions, agentWakeupRequests, authUsers, companies, companyMemberships, companySecretVersions, createDb, heartbeatRuns, + issues, issueWorkProducts, nativeRunResults } from "@paperclipai/db"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import type { NativeExecutionInput, NativeSession, NativeSessionBackend, PrpEvent } from "../../vendor/paperclip-runner/index.js"; +import { CONTROL_PLANE_CONFORMANCE_RESULT, CONTROL_PLANE_CONFORMANCE_TERMINAL } from "../../vendor/paperclip-runner/testing.js"; +import { nativeRestartFixture } from "../../__tests__/helpers/native-restart-fixture.js"; +import { resolveExternalTestDatabaseUrl } from "../../__tests__/helpers/external-test-database.js"; +import { getRecoveryPostgresSupport } from "../../__tests__/helpers/recovery-postgres-fixture.js"; +import { heartbeatService } from "../heartbeat.js"; +import { flushBudgetEvaluationWork } from "../budgets.js"; +import { instanceSettingsService } from "../instance-settings.js"; +import { secretService } from "../secrets.js"; +import { issueRoutes } from "../../routes/issues.js"; +import { errorHandler } from "../../middleware/index.js"; +import { readIdleSleepSafety } from "../idle-sleep-safety.js"; +import { closeIdleWarmNativeSessionsForRestart, readNativeSessionIdleWork } from "./native-session-executor.js"; + +const dedicatedDatabaseUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_RECOVERY_TEST_DATABASE_URL"); +const recoveryOptions = { dedicatedDatabaseUrl }; +const recoverySupport = await getRecoveryPostgresSupport(recoveryOptions); +const describeRecoveryPostgres = recoverySupport.supported ? describe : describe.skip; + +describeRecoveryPostgres("native idle sleep cold continuation", () => { + const fixture = nativeRestartFixture(recoveryOptions); + afterEach(async () => { + await closeIdleWarmNativeSessionsForRestart(); + vi.unstubAllEnvs(); + }); + + it("retires a completed warm owner and resumes saved work through an ordinary authorized comment", async () => { + const { companyId, agentId, runtimeRoot, temporary } = fixture.current; + vi.stubEnv("DATABASE_URL", temporary.connectionString); + vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", "89".repeat(32)); + const db = createDb(temporary.connectionString, { maxConnections: 2 }); + expect(db.$client.options.max).toBe(2); + if (dedicatedDatabaseUrl) expect(db.$client.options.database).toBe(new URL(dedicatedDatabaseUrl).pathname.slice(1)); + const settings = instanceSettingsService(db), previousSettings = await settings.getExperimental(); + const userId = randomUUID(), issueId = randomUUID(), productId = randomUUID(); + const cwd = join(runtimeRoot, "ordinary-task"); + await mkdir(cwd, { recursive: true }); + await db.insert(authUsers).values({ id: userId, name: "Task owner", email: `${userId}@example.invalid`, createdAt: new Date(), updatedAt: new Date() }); + await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: userId, membershipRole: "owner" }); + await db.update(companies).set({ defaultResponsibleUserId: userId }).where(eq(companies.id, companyId)); + await db.insert(issues).values({ id: issueId, companyId, title: "Continue saved work", status: "todo", assigneeAgentId: agentId }); + await db.insert(issueWorkProducts).values({ id: productId, companyId, issueId, type: "artifact", provider: "paperclip", + title: "Saved and continued work", status: "ready_for_review", reviewState: "approved" }); + const secrets = secretService(db), credential = `synthetic-${randomUUID()}`; + const secret = await secrets.create(companyId, { name: "Idle continuation credential", provider: "local_encrypted", value: credential }); + const binding = { SLEEP_CREDENTIAL: { type: "secret_ref", secretId: secret.id, version: "latest" } }; + await secrets.syncEnvBindingsForTarget(companyId, { targetType: "agent", targetId: agentId }, binding); + await db.update(agents).set({ status: "idle", adapterConfig: { provider: "codex", model: "gpt-6.1-sol", cwd, + lifecycleMode: "warm", idleTimeoutMs: 300_000, env: binding }, + runtimeConfig: { heartbeat: { enabled: false, maxConcurrentRuns: 1 } } }).where(eq(agents.id, agentId)); + const opened: NativeSession[] = [], recovered: string[] = [], closed: string[] = [], turns: string[] = []; + const executions: NativeExecutionInput[] = []; + const closeStarted: string[] = []; + let finishClose: (() => void) | undefined; + let closeGate: Promise | undefined; + const backendFactory = (execution: NativeExecutionInput): NativeSessionBackend => { + executions.push(execution); + const makeSession = (): NativeSession => { + const identity = { companyId, agentId, issueId, runId: execution.binding.runId, sessionId: execution.session.normalizedSessionId! }; + const turnId = `turn-${identity.runId}`; + let start!: () => void; + const started = new Promise(resolve => { start = resolve; }); + const result = structuredClone(CONTROL_PLANE_CONFORMANCE_RESULT), evidenceRef = `work_product:${productId}`; + result.completionClaim.contractRevision = execution.completionContract.contract.revision; + result.completionClaim.criteria = execution.completionContract.contract.criteria.map(criterion => ({ + ...result.completionClaim.criteria[0]!, criterionId: criterion.id, evidenceRefs: [evidenceRef] })); + result.evidence = [{ kind: "work_product", ref: evidenceRef }]; + result.verification[0]!.artifactRef = evidenceRef; + const session: NativeSession = { + identity: () => identity, + async capabilities() { return { resume: true, typedEvents: true, steering: false, interruption: true, structuredResult: true }; }, + async *events() { + await started; + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, identity.runId)); + yield { schema: "paperclip.prp.event.v1", sourceEventId: `${identity.runId}:usage`, sourceSeq: 1, + sourceInstanceId: run.runnerInstanceId!, sourceKind: "runner", runId: identity.runId, + normalizedSessionId: identity.sessionId, turnId, eventType: "item.completed", schemaVersion: 1, + priority: 0, emittedAt: new Date().toISOString(), payload: { kind: "usage", usage: { + runDelta: { inputTokens: 1, outputTokens: 1 }, runDeltaComplete: true } } } satisfies PrpEvent; + yield { schema: "paperclip.prp.event.v1", sourceEventId: `${identity.runId}:terminal`, sourceSeq: 2, + sourceInstanceId: run.runnerInstanceId!, sourceKind: "runner", runId: identity.runId, + normalizedSessionId: identity.sessionId, turnId, eventType: "turn.completed", schemaVersion: 1, + priority: 0, emittedAt: new Date().toISOString(), payload: {} } satisfies PrpEvent; + }, + async startTurn() { + const resolved = await secretService(db).resolveEnvBindings(companyId, binding); + expect(resolved.env.SLEEP_CREDENTIAL).toBe(credential); + if (turns.length === 0) { + await writeFile(join(execution.workspace.cwd!, "work.txt"), "first useful work\n"); + } else { + expect(await readFile(join(execution.workspace.cwd!, "work.txt"), "utf8")).toBe("first useful work\n"); + await writeFile(join(execution.workspace.cwd!, "continued.txt"), "continued first useful work\n"); + } + turns.push(identity.runId); + start(); + return { turnId }; + }, + async result() { await started; return { result, terminal: CONTROL_PLANE_CONFORMANCE_TERMINAL, turnId }; }, + async snapshot() { return { backendKind: "mock", sessionId: "saved-provider-session", identity, + providerSessionId: "saved-provider-session", activeTurnId: null, pendingRuntimeRequests: [], lineage: [], cursor: "1" }; }, + async close() { closeStarted.push(identity.runId); await closeGate; closed.push(identity.runId); }, + }; + opened.push(session); + return session; + }; + return { + async descriptor() { return { kind: "mock", name: "Controlled local provider", version: "1", + capabilities: { resume: true, typedEvents: true, steering: false, interruption: true, structuredResult: true }, + runtimeContextCapabilities: { instructions: "native", skills: "native", mcp: "native" } }; }, + async openSession() { return makeSession(); }, + async recoverSession(checkpoint) { + expect(checkpoint.providerSessionId).toBe("saved-provider-session"); + recovered.push(checkpoint.providerSessionId!); + return { recovered: true, session: makeSession() }; + }, + }; + }; + const heartbeat = heartbeatService(db, { nativeSessionBackendFactory: backendFactory }); + try { + await settings.updateExperimental({ enableNativeRunner: true }); + await heartbeat.invoke(agentId, "on_demand", { issueId }, "manual", { actorType: "user", actorId: userId }); + await heartbeat.drainActiveRunExecutions(); + await heartbeat.reconcileCostAccounting(); + await flushBudgetEvaluationWork(db, { companyId }); + expect(turns).toHaveLength(1); + expect(opened).toHaveLength(1); + const [first] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, turns[0])); + expect(first.status).toBe("succeeded"); + expect(await db.select().from(nativeRunResults).where(and(eq(nativeRunResults.runId, first.id), eq(nativeRunResults.schemaStatus, "accepted")))).toHaveLength(1); + expect(JSON.stringify(first.runnerProfileJson)).not.toContain(credential); + expect(readNativeSessionIdleWork()).toBe("present"); + const sessionsBefore = await db.select().from(agentTaskSessions).where(and(eq(agentTaskSessions.companyId, companyId), eq(agentTaskSessions.agentId, agentId))); + expect(sessionsBefore).toHaveLength(1); + expect(JSON.stringify(sessionsBefore)).toContain("saved-provider-session"); + const [encryptedBefore] = await db.select().from(companySecretVersions).where(eq(companySecretVersions.secretId, secret.id)); + expect(encryptedBefore.material).toMatchObject({ scheme: "local_encrypted_v1", ciphertext: expect.any(String) }); + expect(JSON.stringify(encryptedBefore.material)).not.toContain(credential); + closeGate = new Promise(resolve => { finishClose = resolve; }); + const hold = heartbeat.startTaskDrain({ purpose: "idle", ttlMs: 60_000 }); + const inspecting = readIdleSleepSafety(db, heartbeat.getTaskDrainStatus, Date.now, hold.ownerId, + async () => readNativeSessionIdleWork(), undefined, undefined, false, false, + { prepareLocalWork: async ({ canPrepare }) => { + const result = await heartbeat.prepareIdleWarmNativeSessionsForSleep({ canRetire: canPrepare }); + return result.busy || result.failed ? "present" : "none"; + } }); + let inspectionSettled = false; + void inspecting.then(() => { inspectionSettled = true; }); + try { + await vi.waitFor(() => expect(closeStarted).toEqual([turns[0]])); + expect(readNativeSessionIdleWork()).toBe("present"); + expect(closed).toEqual([]); + expect(inspectionSettled).toBe(false); + finishClose!(); + const prepared = await inspecting; + const diagnostic = prepared.backgroundWork === "none" ? undefined : await readIdleSleepSafety(db, heartbeat.getTaskDrainStatus, Date.now, hold.ownerId, + async () => readNativeSessionIdleWork(), undefined, undefined, false, false, { collectDiagnostics: true }); + expect(prepared.backgroundWork, JSON.stringify(diagnostic?.diagnostics)).toBe("none"); + } finally { finishClose!(); await inspecting; heartbeat.stopTaskDrain(); } + expect(closed).toEqual([turns[0]]); + expect(readNativeSessionIdleWork()).toBe("none"); + expect(await db.select().from(agentTaskSessions).where(eq(agentTaskSessions.id, sessionsBefore[0].id))).toEqual(sessionsBefore); + const app = express(); + app.use(express.json()); + app.use((req, _res, next) => { req.actor = { type: "board", userId, source: "local_implicit" }; next(); }); + app.use("/api", issueRoutes(db, {} as Parameters[1])); + app.use(errorHandler); + heartbeat.startTaskDrain({ ttlMs: 60_000 }); + try { + await request(app).post(`/api/issues/${issueId}/comments`).send({ body: "Continue the saved work and record the continuation.", reopen: true, resume: true }).expect(201); + await vi.waitFor(async () => { + const wakes = await db.select().from(agentWakeupRequests).where(and(eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.reason, "issue_reopened_via_comment"))); + expect(wakes).toHaveLength(1); + expect(wakes[0].requestedByActorId).toBe(userId); + expect(heartbeat.getTaskDrainStatus().pendingWakes).toBe(0); + }); + expect(turns).toHaveLength(1); + } finally { heartbeat.stopTaskDrain(); } + await heartbeat.resumeQueuedRuns(); + await heartbeat.drainActiveRunExecutions(); + await heartbeat.reconcileCostAccounting(); + await flushBudgetEvaluationWork(db, { companyId }); + expect(turns).toHaveLength(2); + expect(new Set(turns).size).toBe(2); + expect(opened).toHaveLength(2); + expect(opened[1]).not.toBe(opened[0]); + expect(recovered).toEqual(["saved-provider-session"]); + expect(executions[1].session.normalizedSessionId).toBe(executions[0].session.normalizedSessionId); + const [continued] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, turns[1])); + expect(continued.status).toBe("succeeded"); + expect(await db.select().from(nativeRunResults).where(and(eq(nativeRunResults.runId, continued.id), eq(nativeRunResults.schemaStatus, "accepted")))).toHaveLength(1); + expect(await readFile(join(executions[1].workspace.cwd!, "continued.txt"), "utf8")).toBe("continued first useful work\n"); + expect((await db.select().from(companySecretVersions).where(eq(companySecretVersions.id, encryptedBefore.id)))[0].material).toEqual(encryptedBefore.material); + } finally { + finishClose?.(); + heartbeat.stopTaskDrain(); + await heartbeat.drainActiveRunExecutions(); + await closeIdleWarmNativeSessionsForRestart(); + await settings.updateExperimental(previousSettings); + await db.$client.end(); + } + }, 40_000); +}); diff --git a/server/src/services/native-runtime/native-session-executor.test.ts b/server/src/services/native-runtime/native-session-executor.test.ts index 571db37a555..9a775f099fc 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -10371,7 +10371,7 @@ describe("native warm session supervision", () => { } as NativeExecutionInputV1; let canonicalNote = "original"; const checkpointWarm = vi.fn(async () => { canonicalNote = "edited during failed turn"; return true; }); - const close = vi.fn(async () => { throw new Error("provider shutdown failed"); }); + const close = vi.fn<() => Promise>(async () => { throw new Error("provider shutdown failed"); }); const collectStopped = vi.fn(async () => undefined); state.execute.mockReset().mockImplementationOnce(async options => { await options.onSession?.({ close }); @@ -10389,6 +10389,10 @@ describe("native warm session supervision", () => { // Failed close is not proof that the process stopped. The preceding warm // checkpoint preserves the edits without calling stopped-only collection. expect(collectStopped).not.toHaveBeenCalled(); + expect(readNativeSessionIdleWork()).toBe("present"); + close.mockResolvedValue(undefined); + expect(await closeIdleWarmNativeSessionsForRestart()).toEqual({ closed: 1, busy: 0, failed: 0 }); + expect(collectStopped).toHaveBeenCalledOnce(); }); it.each(["failed", "cancelled"] as const)("retires a %s session even if its warm checkpoint rejects", async (runTerminalState) => { @@ -10668,7 +10672,10 @@ describe("native warm session supervision", () => { } } finally { releaseClose(); - await closeWarmNativeSessionsForEnvironment({ environmentId: identity, reason: "test cleanup" }); + close.mockResolvedValue(undefined); + expect(await closeWarmNativeSessionsForEnvironment({ environmentId: identity, reason: "test cleanup" })) + .toEqual({ closed: changed && !closeFails ? 0 : 1, busy: 0, failed: 0 }); + expect(collectStopped).toHaveBeenCalledOnce(); } }); @@ -11351,6 +11358,187 @@ describe("native warm session supervision", () => { } finally { finishClose(); await closeIdleWarmNativeSessionsForRestart(); } }); + it.each(["controller restart", "idle timeout", "failed turn"])("retains a warm owner without an instruction working copy after rejected retirement: %s", async retirement => { + await closeIdleWarmNativeSessionsForRestart(); + const failure = new Error("provider retirement unconfirmed"); + const close = vi.fn(async () => undefined).mockRejectedValue(failure); + const warmExecution = { ...execution, binding: { ...execution.binding, + runId: "run-retirement-retry", executionWorkspaceId: "workspace-retirement-retry" }, + session: { ...execution.session, normalizedSessionId: "session-retirement-retry", + lifecyclePolicy: { mode: "warm" as const, idleTimeoutMs: retirement === "idle timeout" ? 10 : 60_000 } } } as NativeExecutionInputV1; + state.execute.mockReset().mockImplementationOnce(async options => { + await options.onSession?.({ close }); + return { result: { summary: "Completed" }, terminal: { runTerminalState: retirement === "failed turn" ? "failed" : "succeeded" }, + turnId: "turn-retirement-retry", normalizedSessionId: warmExecution.session.normalizedSessionId, + providerSessionId: "provider-retirement-retry", driverKind: "test", driverVersion: "1", + nativeEventCount: 1, highestContiguousSourceSeq: 1, usage: null }; + }); + try { + await executePaperclipNativeSession({ db: leaseDb(warmExecution), execution: warmExecution, runnerInstanceId: "runner" }); + expect(readNativeSessionIdleWork()).toBe("present"); + if (retirement === "controller restart") { + expect(await closeIdleWarmNativeSessionsForRestart()).toEqual({ closed: 0, busy: 0, failed: 1 }); + } else { + await vi.waitFor(() => expect(close).toHaveBeenCalledOnce()); + } + expect(readNativeSessionIdleWork()).toBe("present"); + const next = { ...warmExecution, binding: { ...warmExecution.binding, runId: "run-retirement-retry-next" } }; + await expect(executePaperclipNativeSession({ db: leaseDb(next), execution: next, runnerInstanceId: "runner" })) + .rejects.toBe(failure); + expect(state.execute).toHaveBeenCalledOnce(); + expect(readNativeSessionIdleWork()).toBe("present"); + close.mockResolvedValue(undefined); + expect(await closeIdleWarmNativeSessionsForRestart()).toEqual({ closed: 1, busy: 0, failed: 0 }); + expect(close).toHaveBeenCalledTimes(3); + expect(readNativeSessionIdleWork()).toBe("none"); + } finally { + close.mockResolvedValue(undefined); + await closeIdleWarmNativeSessionsForRestart(); + } + }); + + describe("native idle sleep retirement", () => { + async function prepareSleep(input: { canRetire: () => boolean }) { + const { heartbeatService } = await import("../heartbeat.js"); + return heartbeatService(leaseDb()).prepareIdleWarmNativeSessionsForSleep(input); + } + async function start(name: string, close = vi.fn<() => Promise>(async () => undefined), onTurn?: () => Promise, + instructionWorkingCopy?: Parameters[0]["instructionWorkingCopy"]) { + const current = { ...execution, binding: { ...execution.binding, + runId: name, agentId: `agent-${name}`, executionWorkspaceId: name }, session: { ...execution.session, + normalizedSessionId: name, lifecyclePolicy: { mode: "warm" as const, idleTimeoutMs: 60_000 } } } as NativeExecutionInputV1; + const result = { result: { summary: "Completed" }, terminal: { runTerminalState: "succeeded" }, + turnId: name, normalizedSessionId: name, providerSessionId: name, driverKind: "test", driverVersion: "1", + nativeEventCount: 1, highestContiguousSourceSeq: 1, usage: null }; + state.execute.mockReset().mockImplementationOnce(async options => { + await options.onSession?.({ close }); + await onTurn?.(); + return result; + }); + await executePaperclipNativeSession({ db: leaseDb(current), execution: current, runnerInstanceId: name, instructionWorkingCopy }); + return { close, current, result }; + } + afterEach(async () => { await closeIdleWarmNativeSessionsForRestart(); }); + + it("does no work when no warm owner exists", async () => { + await closeIdleWarmNativeSessionsForRestart(); + expect(await prepareSleep({ canRetire: () => true })) + .toEqual({ closed: 0, busy: 0, failed: 0 }); + expect(readNativeSessionIdleWork()).toBe("none"); + }); + + it("leaves an idle owner intact without current drain authority", async () => { + const f = await start("sleep-stale-hold"); + expect(await prepareSleep({ canRetire: () => false })) + .toEqual({ closed: 0, busy: 0, failed: 0 }); + expect(f.close).not.toHaveBeenCalled(); + expect(readNativeSessionIdleWork()).toBe("present"); + expect(await prepareSleep({ canRetire: () => true })) + .toEqual({ closed: 1, busy: 0, failed: 0 }); + expect(f.close).toHaveBeenCalledExactlyOnceWith({ reason: "tenant idle sleep" }); + expect(readNativeSessionIdleWork()).toBe("none"); + }); + + it("keeps task-drain reads and local diagnostics passive", async () => { + const f = await start("sleep-passive-read"); + const { heartbeatService } = await import("../heartbeat.js"); + const heartbeat = heartbeatService(leaseDb()); + heartbeat.getTaskDrainStatus(); + expect(readNativeSessionIdleWork()).toBe("present"); + await readIdleLocalWork({ diagnostic: vi.fn() }); + heartbeat.getTaskDrainStatus(); + expect(f.close).not.toHaveBeenCalled(); + expect(await heartbeat.prepareIdleWarmNativeSessionsForSleep({ canRetire: () => true })) + .toEqual({ closed: 1, busy: 0, failed: 0 }); + expect(readNativeSessionIdleWork()).toBe("none"); + }); + + it("waits for a started checkpoint and stops retiring more owners after authority expires", async () => { + let finish!: () => void; + const checkpoint = new Promise(resolve => { finish = resolve; }); + const first = await start("sleep-expiring-first", vi.fn(() => checkpoint)); + const second = await start("sleep-expiring-second"); + let current = true; + const preparation = prepareSleep({ canRetire: () => current }); + try { + await vi.waitFor(() => expect(first.close).toHaveBeenCalledOnce()); + expect(readNativeSessionIdleWork()).toBe("present"); + current = false; + finish(); + expect(await preparation).toEqual({ closed: 1, busy: 0, failed: 0 }); + expect(second.close).not.toHaveBeenCalled(); + expect(readNativeSessionIdleWork()).toBe("present"); + expect(await prepareSleep({ canRetire: () => true })) + .toEqual({ closed: 1, busy: 0, failed: 0 }); + expect(readNativeSessionIdleWork()).toBe("none"); + } finally { finish(); await preparation; } + }); + + it("preserves busy turns without closing them on release after an abandoned sleep attempt", async () => { + const close = vi.fn(async () => undefined); + await start("sleep-busy-turn", close, async () => { + expect(await prepareSleep({ canRetire: () => true })) + .toEqual({ closed: 0, busy: 1, failed: 0 }); + expect(close).not.toHaveBeenCalled(); + }); + expect(close).not.toHaveBeenCalled(); + expect(readNativeSessionIdleWork()).toBe("present"); + expect(await prepareSleep({ canRetire: () => true })) + .toEqual({ closed: 1, busy: 0, failed: 0 }); + }); + + it("collects managed edits after the provider checkpoint before clearing the owner", async () => { + const root = await mkdtemp(join(tmpdir(), "native-sleep-retirement-")); + const saved = join(root, "saved-instructions.md"); + const providerCheckpoint = join(root, "provider-checkpoint.json"); + let stopped = false; + const close = vi.fn(async () => { + await writeFile(providerCheckpoint, JSON.stringify({ session: "sleep-files", resume: "retained" })); + stopped = true; + }); + const collectStopped = vi.fn(async () => { + expect(stopped).toBe(true); + expect(readNativeSessionIdleWork()).toBe("present"); + await writeFile(saved, "Edits collected from stopped provider"); + }); + try { + await start("sleep-files", close, undefined, { + runId: "sleep-files", root, preparationKey: "sleep-files", hasChanges: async () => false, collectStopped, + }); + expect(await prepareSleep({ canRetire: () => true })) + .toEqual({ closed: 1, busy: 0, failed: 0 }); + expect(await readFile(saved, "utf8")).toBe("Edits collected from stopped provider"); + expect(JSON.parse(await readFile(providerCheckpoint, "utf8"))).toEqual({ session: "sleep-files", resume: "retained" }); + expect(collectStopped).toHaveBeenCalledOnce(); + expect(readNativeSessionIdleWork()).toBe("none"); + } finally { + await closeIdleWarmNativeSessionsForRestart(); + await rm(root, { recursive: true, force: true }); + } + }); + + it("retains a rejected sleep retirement and admits the next ordinary turn after a successful retry", async () => { + const failure = new Error("sleep checkpoint unavailable"); + const close = vi.fn(async () => undefined).mockRejectedValueOnce(failure); + const f = await start("sleep-retry", close); + expect(await prepareSleep({ canRetire: () => true })) + .toEqual({ closed: 0, busy: 0, failed: 1 }); + expect(readNativeSessionIdleWork()).toBe("present"); + expect(await prepareSleep({ canRetire: () => true })) + .toEqual({ closed: 1, busy: 0, failed: 0 }); + const next = { ...f.current, binding: { ...f.current.binding, runId: "sleep-retry-next" } }; + state.execute.mockImplementationOnce(async options => { + expect(options.existingSession).toBeUndefined(); + return { ...f.result, result: { summary: "Continued after sleep retirement" } }; + }); + expect(await executePaperclipNativeSession({ db: leaseDb(next), execution: next, runnerInstanceId: "sleep-next" })) + .toMatchObject({ exitCode: 0, summary: "Continued after sleep retirement", sessionId: "sleep-retry", + resultJson: { nativeResult: { summary: "Continued after sleep retirement" } } }); + expect(state.execute).toHaveBeenCalledTimes(2); + expect(readNativeSessionIdleWork()).toBe("none"); + }); + }); + it.each(["environment deletion", "controller restart"])("closes an idle warm session before %s", async (shutdownKind) => { const close = vi.fn(async () => undefined); const warmExecution = { @@ -11629,7 +11817,13 @@ describe("native warm session supervision", () => { if (checkpointFails) await expect(running).rejects.toBe(checkpointError); else await running; } - await expect(closeIdleWarmNativeSessionsForRestart()).resolves.toEqual({ closed: 0, busy: 0, failed: 0 }); + if (checkpointFails) { + expect(readNativeSessionIdleWork()).toBe("present"); + close.mockResolvedValue(undefined); + await expect(closeIdleWarmNativeSessionsForRestart()).resolves.toEqual({ closed: 1, busy: 0, failed: 0 }); + } else { + await expect(closeIdleWarmNativeSessionsForRestart()).resolves.toEqual({ closed: 0, busy: 0, failed: 0 }); + } }); it("preserves the active turn when a warm checkpoint resumes the same run", async () => { diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 1374af9cd1e..bab52898a4a 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -856,15 +856,23 @@ export async function closeIdleWarmNativeSessionsForRestart(): Promise<{ }); } +export async function prepareIdleWarmNativeSessionsForSleep(input: { + canRetire: () => boolean; +}): Promise<{ closed: number; busy: number; failed: number }> { + return closeIdleWarmNativeSessions({ reason: "tenant idle sleep", canRetire: input.canRetire }); +} + async function closeIdleWarmNativeSessions(input: { environmentId?: string; reason: string; closeBusyOnRelease?: boolean; + canRetire?: () => boolean; }): Promise<{ closed: number; busy: number; failed: number }> { let closed = 0; let busy = 0; let failed = 0; for (const [sessionId, entry] of [...warmNativeSessions]) { + if (input.canRetire && !input.canRetire()) break; if (input.environmentId !== undefined && entry.environmentId !== input.environmentId) { continue; } @@ -890,7 +898,6 @@ async function closeIdleWarmNativeSessions(input: { forgetWarmNativeSession(sessionId, entry); closed += 1; } catch { - if (!entry.instructionWorkingCopy?.runId) forgetWarmNativeSession(sessionId, entry); failed += 1; } finally { entry.busy = false; @@ -898,6 +905,7 @@ async function closeIdleWarmNativeSessions(input: { closingWarmNativeSessions.delete(sessionId); } } + if (input.canRetire && !input.canRetire()) break; } return { closed, busy, failed }; } @@ -7391,7 +7399,6 @@ async function releaseWarmNativeSession( await closeWarmNativeSession(entry, entry.closeOnReleaseReason); forgetWarmNativeSession(sessionId, entry); } catch (error) { - if (!entry.instructionWorkingCopy?.runId) forgetWarmNativeSession(sessionId, entry); if (requireCleanClose) throw error; } finally { entry.busy = false; } return; @@ -7418,7 +7425,7 @@ function armWarmNativeSessionIdleTimer( current.closeOnReleaseReason = "warm native session idle timeout"; const closing = closeWarmNativeSession(current, current.closeOnReleaseReason) .then(() => { forgetWarmNativeSession(sessionId, current); }) - .catch(() => { if (!current.instructionWorkingCopy?.runId) forgetWarmNativeSession(sessionId, current); }) + .catch(() => {}) .finally(() => { current.busy = false; if (closingWarmNativeSessions.get(sessionId) === closing) closingWarmNativeSessions.delete(sessionId);