From 18be7ac69676c3faa31c800bc520a88a48428407 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Fri, 6 Nov 2026 20:49:30 -0600 Subject: [PATCH] fix(server): retire idle native sessions before tenant sleep (#18988) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip lets people manage AI agents and their work. > - Native agents can retain a warm session between useful turns. > - That owner must close and save its files before the tenant can sleep. > - A failed close must keep the owner visible until retirement succeeds. > - This PR prepares idle resources under the exact current admission hold, then takes a fresh safety snapshot. > - Busy turns, uncertain closure and changed holds still prevent sleep. > - CI also exposed a chat authorization race. The fix repeats full authorization after the exact delivery commits. ## Linked Issues or Issue Description Refs #18639 for local work and incoming wake boundaries. Related #18829 covers suspension receipts after admission. **What happened?** Idle preflight did not retire warm Native owners. Three failed-close paths could forget an owner without proving that its provider stopped. CI also found a false chat rejection when a delivery changed from processing to processed between two validation reads. **Expected behavior** Retire only idle owners under the exact bounded hold. Wait for closure and file collection. Retain failed owners. Scan safety after preparation. Keep diagnostic reads passive. A newly committed chat delivery must pass the full current authorization checks before attestation succeeds. **Steps to reproduce** Finish a warm Native turn. Start an idle drain and request safety with its owner ID. Delay or reject provider closure, or replace the hold during preparation or scanning. The report must keep the instance awake. For the chat race, commit the exact delivery between the first binding read and its fallback read. **Paperclip version or commit** Current head `7f6f02b611dcf5a9a2b05173a44e9631eaf78058` normally integrates actual master `1aa0c3639714f4976cfa56f95aae6e2b7d1235ac`. The merge preserves every unrelated master path. Fresh exact-head CI and complete review remain required. ## What Changed - Retain uncertain Native owners after failed restart, idle timeout and failed-turn closure. - Prepare Native retirement through the existing heartbeat facade. Reuse provider closure and stopped-file collection. - Track preparation under the exact idle owner and expiry. Reject new activity, changed holds and expired holds before and after awaited work. - Take the final durable and local safety snapshot after preparation. Keep diagnostic inspection passive. - Cover ephemeral workspace retirement and useful next use, supported plugin drain and SDK persistence, busy owners, failed closure and hold replacement. - Keep dedicated recovery storage explicit. Use the same validated options for support and storage. Probe-free helpers retain the original implementation bodies and preserve default callers. - Complete the Native fixture through real retirement and budget evaluation. Keep its rejection, file, credential, accounting and strict NONE assertions. - Repeat the complete chat authorization once if the fallback sees the exact batch now processed. Preserve owner locks, principal and binding checks, denial handling and the original retry deadline. ## Verification - The support-mode regression reproduces one failure with four passes under the old policy, then passes all five controls. It retains prior allocation and cleanup cases. The explicit mode does not probe embedded PostgreSQL or use an admin alias. - The deterministic chat race baseline reproduces two authorized failures with all four current-authority denial controls passing. Both Slack and Discord use real PostgreSQL and the real authorization function. The negative setup commits before the attestation transaction; it does not claim to revoke access through a row lock. - The product fix passes normal own App declaration emission with no public declaration changes, plus both current normal owning compiler programs (399 and 611 roots). The same six controls pass after the fix. The full original file plus these controls passes all 30 tests with no skips. - The original controlled Native continuation passes: one test, no failures or skips, 1.718 seconds in its body and 23.138 seconds in its launcher. The original 40-second case and 180-second outer bounds are unchanged. It verifies held closure, strict NONE, an ordinary authorized comment, saved session/files, useful output, encrypted credentials, accepted results and settled accounting. All ten observed processes and groups close; worker connections and listeners reach zero. A separate post-owner audit finds zero sessions; the successful owned database is normally removed. - Earlier preparation, HTTP, lifecycle, workspace and plugin controls remain source-specific evidence. They are not a claim that every final-tree test ran locally. - The original CI assertion log is retained. The controlled race proves a possible cause; the original CI SQL schedule was not captured. Two other jobs lost their self-hosted runners and provided no test verdict. Earlier asset timeouts, readiness failures and unexplained later recovery remain separately recorded. No timeout, failure predicate or authority check was weakened. Focused public commands: ```sh pnpm exec vitest run server/src/__tests__/idle-sleep-preparation.test.ts server/src/__tests__/instance-settings-routes.test.ts --maxWorkers=1 --no-file-parallelism pnpm exec vitest run server/src/__tests__/heartbeat-reviewed-chat-binding.integration.test.ts --maxWorkers=1 --no-file-parallelism ``` Local database qualification uses fresh owned test databases on existing PostgreSQL, bounded pools and one worker. The local allocation adapter changes only database allocation for this test module; it runs normal migrations and client cleanup. It is not part of the published source. No new cluster is created. Local acceptance measured the `0a69efa6` integration with the same committed authorization bytes. The later normal master merge adds startup-timeout classification and recovery changes. Those paths are outside the controlled healthy-backend witness; retirement and authorization hunks are preserved. Exact-head remote CI at `7f6f02b611dcf5a9a2b05173a44e9631eaf78058` has completed with 56 successful contexts and four skips. The complete current Greptile summary is5/5 with no new actionable findings; all review threads are resolved. Ponytail’s current turn is explicitly paused without a current-head verdict; earlier actual findings are fixed and accounted for. No whole final-tree local repository build/test run, live-provider restore or deployed suspension is claimed. Actual master subsequently advanced to `d5c4f7dd02d408a261cedd9a1926b9900a4aedc4`. Its local normal-merge projection with this reviewed head preserves all20 owned paths and every unrelated master blob and mode. Current Runner TypeScript, App declarations and their full owning compiler programs pass. Focused Native cleanup and warm-continuation composition checks are being completed before merge; the new upstream suspension-retry behavior is not inferred from the older controlled-backend witness. The remote PR head remains unchanged. ## Risks - Privileged idle preparation can close idle warm Native sessions. It requires the exact current hold. - Failed closure retains its owner and can keep the tenant awake until normal retry succeeds. - A changed hold cannot authorize sleep. Started cleanup remains tracked until it settles. - The chat repair can perform one extra full authorization read. It keeps the existing deadline and does not grant authority from the fallback query. - Unsupported resources and active work retain their existing blockers. This PR does not certify every transport, plugin or deployed host. ## Model Used OpenAI `gpt-6.1-sol` produced the delegated changes and verification. OpenAI Codex, based on GPT-6, performed managing review and integration. Both used repository tools and code execution. The coordinator's exact backend snapshot and context-window limits were not exposed. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green at exact head `7f6f02b611dcf5a9a2b05173a44e9631eaf78058` - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups at the exact current head - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip Co-authored-by: Codex --- doc/DEVELOPING.md | 56 ++++ ...-reviewed-chat-binding.integration.test.ts | 44 +++ .../helpers/native-restart-fixture.ts | 254 +++++++++++++++++ .../src/__tests__/helpers/native-restart.ts | 256 +----------------- .../helpers/recovery-postgres-fixture.ts | 43 +++ .../__tests__/helpers/recovery-postgres.ts | 29 +- .../__tests__/idle-completed-history.test.ts | 6 +- .../__tests__/idle-sleep-preparation.test.ts | 125 +++++++++ .../instance-settings-routes.test.ts | 36 ++- .../src/__tests__/plugin-idle-manager.test.ts | 48 ++++ .../__tests__/recovery-postgres-mode.test.ts | 112 ++++++++ .../workspace-runtime-lifecycle.test.ts | 112 ++++++++ server/src/routes/instance-settings.ts | 4 + server/src/routes/openapi.ts | 2 +- server/src/services/heartbeat.ts | 2 + .../src/services/heartbeat/run-preparation.ts | 24 +- server/src/services/idle-sleep-safety.ts | 29 +- ...dle-sleep-continuation.integration.test.ts | 211 +++++++++++++++ .../native-session-executor.test.ts | 200 +++++++++++++- .../native-runtime/native-session-executor.ts | 13 +- 20 files changed, 1301 insertions(+), 305 deletions(-) create mode 100644 server/src/__tests__/helpers/native-restart-fixture.ts create mode 100644 server/src/__tests__/helpers/recovery-postgres-fixture.ts create mode 100644 server/src/__tests__/idle-sleep-preparation.test.ts create mode 100644 server/src/__tests__/recovery-postgres-mode.test.ts create mode 100644 server/src/__tests__/workspace-runtime-lifecycle.test.ts create mode 100644 server/src/services/native-runtime/native-idle-sleep-continuation.integration.test.ts 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);